Tối ưu hóa giám sát log với AWS CloudWatch Subscription Filters: Từ lý thuyết đến thực tiễn

Trong thế giới của các ứng dụng hiện đại, việc giám sát và phân tích log theo thời gian thực (real-time) là cực kỳ quan trọng. Log không chỉ giúp chúng ta gỡ lỗi mà còn là chìa khóa để hiểu hành vi người dùng, theo dõi hiệu suất và phản ứng nhanh chóng với các sự cố. Hôm nay, chúng ta sẽ khám phá một công cụ mạnh mẽ của AWS: CloudWatch Subscription Filters.

Subscription Filters là gì?

Hãy tưởng tượng bạn có một dòng suối dữ liệu log khổng lồ chảy vào hồ chứa CloudWatch. Subscription Filters giống như những chiếc lưới thông minh mà bạn đặt trên dòng suối đó. Thay vì lấy toàn bộ lượng nước, chiếc lưới này chỉ chọn những giọt nước (log entries) có đặc điểm phù hợp với "mẫu lọc" (filter pattern) mà bạn đã định nghĩa, rồi chuyển chúng đến một hồ chứa khác để xử lý.

Công cụ này cho phép chúng ta stream log theo thời gian thực đến các dịch vụ như AWS Lambda, Kinesis Data Streams, hoặc Kinesis Data Firehose. Điều này mở ra khả năng xử lý, phân tích và phản hồi tự động đối với log data ngay khi chúng được tạo ra.

Các loại Destination và ứng dụng thực tế

Tùy vào nhu cầu, bạn có thể lựa chọn điểm đến phù hợp cho log của mình:

• AWS Lambda: Thích hợp cho các tác vụ xử lý nhỏ và linh hoạt. Khi một log entry khớp với bộ lọc, Lambda sẽ được kích hoạt để thực hiện một hành động cụ thể, chẳng hạn như gửi thông báo, phân tích dữ liệu log, hoặc đẩy log sang một hệ thống bên ngoài.

• Kinesis Data Streams/Firehose: Đây là lựa chọn lý tưởng cho việc xử lý log với quy mô lớn. Kinesis giúp bạn thu thập và chuyển log đến các kho dữ liệu lớn (data lake) như Amazon S3 hoặc dịch vụ phân tích như Amazon Elasticsearch Service (OpenSearch).

Sức mạnh của Filter Patterns

"Lưới lọc" của bạn có hiệu quả đến đâu phụ thuộc vào Filter Pattern. Bạn có thể sử dụng các mẫu lọc đơn giản hoặc phức tạp để chỉ chọn những log cần thiết:

• Lọc cơ bản: Bạn có thể lọc theo một từ khóa ("ERROR", "WARN") hoặc một cụm từ cụ thể ("duration:").

• Kết hợp từ khóa: Sử dụng ? để kết hợp các từ khóa với logic OR (?ERROR ?WARN) hoặc khoảng trắng để sử dụng logic AND (ERROR TIMEOUT).

• Lọc chuyên sâu (Structured logs): Đối với log có cấu trúc (như JSON), bạn có thể lọc dựa trên các trường cụ thể, ví dụ: "{ $.level = \"ERROR\" }".

Bài toán thực tế: Theo dõi Slow Queries của PostgreSQL

Giả sử bạn đang chạy một ứng dụng trên cơ sở dữ liệu Amazon RDS for PostgreSQL. Bạn muốn giám sát các truy vấn có thời gian thực thi quá lâu (slow queries) để xử lý kịp thời, tránh ảnh hưởng đến trải nghiệm người dùng.

Kiến trúc giải pháp:

1. RDS PostgreSQL sẽ ghi các slow queries vào CloudWatch Logs.

2. Một Subscription Filter sẽ lắng nghe log group này và chỉ chọn những dòng log chứa từ khóa duration:.

3. Bộ lọc sẽ kích hoạt một Lambda Function đã được cấu hình sẵn.

4. Lambda này sẽ xử lý log và đẩy dữ liệu slow query đó đến Grafana Loki để trực quan hóa và cảnh báo.

Cách triển khai:

1. Viết code cho Lambda: 

import base64
import gzip
import json
import logging
import os
import requests

# Configure logging
logging.basicConfig(level=logging.INFO)

LOKI_URL = os.environ.get(
    "LOKI_URL", "https://logs-prod-030.grafana.net/loki/api/v1/push"
)
LOKI_API_KEY = os.environ.get("LOKI_API_KEY")


def lambda_handler(event, context):
    """
    Lambda function to process CloudWatch logs and push to Grafana Loki.
    Compatible with Python 3.13 runtime.
    """
    if not LOKI_API_KEY:
        logging.error("LOKI_API_KEY environment variable is not set.")
        return {"statusCode": 400, "body": "Loki API key not configured."}

    try:
        # Extract and decompress log data
        encoded_data = event["awslogs"]["data"]
        decoded_data = base64.b64decode(encoded_data)
        decompressed_data = gzip.decompress(decoded_data)
        log_data = json.loads(decompressed_data)
    except Exception as e:
        logging.exception("Failed to decode CloudWatch logs.")
        return {"statusCode": 400, "body": f"Invalid log data: {e}"}

    loki_streams = []

    for log_event in log_data.get("logEvents", []):
        message = log_event.get("message", "")
        timestamp = log_event.get("timestamp", 0) * 1_000_000  # ms → ns

        labels = {
            "source": "aurora-postgresql",
            "log_group": log_data.get("logGroup", ""),
            "log_stream": log_data.get("logStream", ""),
        }

        loki_streams.append(
            {"stream": labels, "values": [[str(timestamp), message]]}
        )

    if not loki_streams:
        logging.info("No relevant log events to send to Loki.")
        return {"statusCode": 200, "body": "No events processed."}

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {LOKI_API_KEY}",
    }

    try:
        response = requests.post(LOKI_URL, json={"streams": loki_streams}, headers=headers)
        response.raise_for_status()
        logging.info(
            "Successfully pushed %d log lines to Loki. Response: %s",
            len(loki_streams),
            response.text,
        )
        return {"statusCode": 200, "body": "Logs successfully pushed to Loki."}
    except requests.exceptions.RequestException as e:
        logging.exception("Failed to push logs to Loki.")
        return {"statusCode": 500, "body": f"Failed to push logs: {e}"}

# Add Lambda permission
aws lambda add-permission \
  --function-name "slow-query-processor" \
  --statement-id "allow-cloudwatch-logs" \
  --action "lambda:InvokeFunction" \
  --principal "logs.amazonaws.com" \
  --source-arn "arn:aws:logs:region:account:log-group:/aws/rds/cluster/my-db/postgresql:*"

2. Tạo Subscription Filter: Bạn sử dụng AWS CLI để tạo bộ lọc với mẫu chính xác.

aws logs put-subscription-filter \
  --log-group-name "/aws/rds/cluster/dev-liveapp-main-db-aurora/postgresql" \
  --filter-name "slow-query-filter" \
  --filter-pattern "?\"duration:\"" \
  --destination-arn "arn:aws:lambda:region:account:function:slow-query-processor"

# Test với sample logs
aws logs test-metric-filter \
  --filter-pattern "?\"duration:\"" \
  --log-event-messages \
    "2025-09-15 09:42:32 UTC:10.0.0.166(54240):liveappuser@liveappdb:[13297]:LOG:  duration: 11043.387 ms" \
    "2025-09-15 09:42:33 UTC:10.0.0.166(54240):liveappuser@liveappdb:[13298]:LOG:  connection received"

Lưu ý rằng chúng ta sử dụng ?\"duration:\" thay vì chỉ "duration:" để đảm bảo mẫu lọc khớp chính xác với log format của PostgreSQL.

3. Kiểm tra và tối ưu: Trước khi đưa vào production, hãy sử dụng lệnh aws logs test-metric-filter với một vài log mẫu để đảm bảo rằng bộ lọc của bạn hoạt động như mong đợi.

Kết luận

AWS CloudWatch Subscription Filters là một công cụ cực kỳ hữu ích, giúp chúng ta vượt qua giới hạn của việc giám sát log truyền thống. Thay vì phải "săn lùng" lỗi sau khi chúng xảy ra, chúng ta có thể xây dựng các hệ thống phản ứng theo thời gian thực, tự động phát hiện và xử lý vấn đề ngay từ khi chúng bắt đầu. Bằng cách sử dụng các filter patterns thông minh, bạn không chỉ tiết kiệm chi phí mà còn có thể xây dựng một hệ thống giám sát hiệu quả và đáng tin cậy.

Bạn đã từng sử dụng Subscription Filters cho trường hợp nào chưa? Hãy chia sẻ kinh nghiệm của bạn ở phần bình luận bên dưới nhé!