Analyzing log data with CloudWatch Logs Insights#

Keywords: AWS, Amazon, CloudWatch, CW, Log, Logs, Insight, Insights

How do we learn CloudWatch Logs Insights#

最好的学习方法永远是实践. 所以我们准备了一个脚本可以轻松的创建 Log Group / Stream 并且往里面每秒打一个自定义的 Log.

我们还有个函数可以轻松地清除这些资源. 一旦我们 run 了 delete_log_group API, 指定的 Log Group 以及下面的所有 Log Stream 的数据就都会被删除. 我们可以立刻开始一个新的实验.

Query Language Quick Start#

CloudWatch Logs Insights (下面简称 Insights) 的查询语言是一个类 SQL 的语言. 它的语法包含这么几个部分:

选择, 定义你想要选择哪些字段. 这里包括类似于 SQL 中的 SELECT 的用来选择字段的 fields 关键字, 以及用来创建图表的 stats 关键字.
筛选, 定义你想要筛选哪些数据. 这里包括类似于 SQL 中的 WHERE 的用来筛选数据的 filter 关键字.
提取数据, 由于 Log 数据并不像 SQL 中的都是严格结构化的数据, 所以你有时候需要从 Log 中提取结构化的数据. 这里的关键字是 parse.
和 SQL 一样, Insights 也有 limit 关键字可以限制返回的数据条数, 以及 sort 关键字可以按照指定的字段排序.

当然以上几个只是核心功能, Insights 还有一些针对 Log 的高阶语法. 我们先不急着展开讲, 之后我们会慢慢学到的.

下面这条是一个最基础的例子, 类似于 SQL 中的 SELECT * FROM TABLE LIMIT 10. 其中 @timestamp 是 log 的时间戳, @message 是 log 的内容``. @logStream 和 @log 分别是 Stream 和 Group. 带 @ 的都是 CloudWatch log 中的特殊字段:

fields @timestamp, @message, @logStream, @log
| sort @timestamp desc
| limit 20

如果你的 Log Message 的结构是 {"server_id": "container-1", "processing_time": 500}. 那么你可以直接用 JSON Dot Notation 来选择字段:

fields @timestamp, @message, server_id, processing_time
| sort @timestamp desc
| limit 20

当然在 Filter 中的条件也可以使用 JSON Dot Notation:

fields @timestamp, @message, server_id, processing_time
| filter server_id = "container-1"
| sort @timestamp desc
| limit 20

这里有个小知识点, 如果你要对 timestamp 进行 filter 时, 它不支持 human readable format, 你需要自行将其转化为 millisecond 的 timestamp. 而且你转换的时候一定要注意时区, 否则你的结果可能会出现偏差:

fields @timestamp, @message, server_id, processing_time
| filter @timestamp <= 1699797262424
| sort @timestamp desc
| limit 20

如果你的 filter 的条件有多个, 你可以用逻辑运算符 and, or, not 来连接它们. 就跟在 SQL 中的 WHERE col1 = value1 and col2 = value2 一样.

Pattern#

Pattern 是一个很强大的函数, 它可以对你的 Log 进行采样, 然后分析出来有哪些 Pattern. 例如我们的测试数据中有两种不同模式的 JSON:

{"server_id": "container-1", "status": "succeeded"}
{"server_id": "container-1", "processing_time": 2000}

那么 pattern 这个函数就可以自动分析出这两种 pattern 的 regex:

fields @timestamp, @message
| pattern @message

Playbook#

这里我们提供了几个 Python 模块, 用于方便地创建 fake data, 以及测试不同的 query.

recipe.py 一些能使得代码更精炼的模块.

# -*- coding: utf-8 -*-

"""
A helper module to work with CloudWatch Logs Group, Stream and put log events.

Usage:

.. code-block:: python

    from recipe import (
        get_log_group,
        create_log_group,
        delete_log_group,
        get_log_stream,
        create_log_stream,
        delete_log_stream,
        Event,
        BaseJsonMessage,
        put_log_events,
        get_ts_in_second,
        get_ts_in_millisecond,
        QueryStatusEnum,
        wait_logs_insights_query_to_succeed,
        run_query,
        reformat_query_results,
    )
"""

import typing as T
import time
import json
import enum
import dataclasses
from datetime import datetime, timezone


def get_log_group(
    logs_client,
    group_name: str,
) -> T.Optional[dict]:
    """
    Get a log group details by name, if it doesn't exist, return None.

    :return: A dict with the log group details, or None if it doesn't exist.
    """
    # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/describe_log_groups.html
    res = logs_client.describe_log_groups(
        logGroupNamePrefix=group_name,
    )
    groups = [
        dct
        for dct in res.get("logGroups", [])
        if dct.get("logGroupName", "unknown-log-group-name") == group_name
    ]
    if len(groups):
        return groups[0]
    else:
        return None


def create_log_group(
    logs_client,
    group_name: str,
) -> bool:
    """
    Create a log group, if it already exists, do nothing.

    :return: True if the log group was created, False if it already existed.
    """
    group = get_log_group(logs_client, group_name)
    if group is None:
        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/create_log_group.html
        logs_client.create_log_group(
            logGroupName=group_name,
        )
        return True
    else:
        return False


def delete_log_group(
    logs_client,
    group_name: str,
) -> bool:
    """
    Delete a log group, if it doesn't exist, do nothing.

    :return: True if the log group was deleted, False if it didn't exist.
    """
    group = get_log_group(logs_client, group_name)
    if group is None:
        return False
    else:
        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/delete_log_group.html
        logs_client.delete_log_group(
            logGroupName=group_name,
        )
        return True


def get_log_stream(
    logs_client,
    group_name: str,
    stream_name: str,
) -> T.Optional[dict]:
    """
    Get a log stream details by name, if it doesn't exist, return None.

    :return: A dict with the log stream details, or None if it doesn't exist.
    """
    # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/describe_log_streams.html
    res = logs_client.describe_log_streams(
        logGroupName=group_name,
        logStreamNamePrefix=stream_name,
    )
    streams = [
        dct
        for dct in res.get("logStreams", [])
        if dct.get("logStreamName", "unknown-log-stream-name") == stream_name
    ]
    if len(streams):
        return streams[0]
    else:
        return None


def create_log_stream(
    logs_client,
    group_name: str,
    stream_name: str,
) -> bool:
    """
    Create a log stream, if it already exists, do nothing.

    :return: True if the log stream was created, False if it already existed.
    """
    stream = get_log_stream(logs_client, group_name, stream_name)
    if stream is None:
        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/create_log_stream.html
        logs_client.create_log_stream(
            logGroupName=group_name,
            logStreamName=stream_name,
        )
        return True
    else:
        return False


def delete_log_stream(
    logs_client,
    group_name: str,
    stream_name: str,
) -> bool:
    """
    Delete a log stream, if it doesn't exist, do nothing.

    :return: True if the log stream was deleted, False if it didn't exist.
    """
    stream = get_log_stream(logs_client, group_name, stream_name)
    if stream is None:
        return False
    else:
        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/delete_log_stream.html
        logs_client.delete_log_stream(
            logGroupName=group_name,
            logStreamName=stream_name,
        )
        return True


EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)


def get_now_ts() -> int:
    """
    The put log events API expects a timestamp in milliseconds since epoch.
    """
    return int(
        (datetime.utcnow().replace(tzinfo=timezone.utc) - EPOCH).total_seconds() * 1000
    )


@dataclasses.dataclass
class Event:
    """
    Log event data model.
    """

    message: str = dataclasses.field()
    timestamp: int = dataclasses.field(default_factory=get_now_ts)


@dataclasses.dataclass
class BaseJsonMessage:
    """
    Base class for json encoded log message.
    """

    def to_json(self) -> str:
        """
        Convert the object to a json string.

        You can override this method to customize the json serialization.
        """
        return json.dumps(dataclasses.asdict(self))

    @classmethod
    def from_json(cls, json_str: str):
        """
        You can override this module to customize the json deserialization.
        """
        dct = json.loads(json_str)
        return cls(**dct)


def put_log_events(
    logs_client,
    group_name: str,
    stream_name: str,
    events: T.List[Event],
) -> T.Optional[dict]:
    """
    Put a list of events into a log stream.

    Ref: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/put_log_events.html

    :param logs_client: The boto3 logs client.
    :param group_name: The log group name.
    :param stream_name: The log stream name.
    :param events: A list of :class:`Event` objects.

    :return: A dict with the response from the put_log_events call.
    """
    if len(events) == 0:
        return None
    res = logs_client.put_log_events(
        logGroupName=group_name,
        logStreamName=stream_name,
        logEvents=[dataclasses.asdict(event) for event in events],
    )
    return res


def get_ts(dt: datetime) -> float:
    """
    Convert a datetime object to a timestamp in seconds since epoch.

    It assumes the datetime object is in UTC if it doesn't have a timezone.
    """
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    else:
        dt = dt.astimezone(timezone.utc)
    return (dt - EPOCH).total_seconds()


def get_ts_in_second(dt: datetime) -> int:
    """
    Convert a datetime object to a timestamp in seconds since epoch.
    """
    return int(get_ts(dt))


def get_ts_in_millisecond(dt: datetime) -> int:
    """
    Convert a datetime object to a timestamp in milliseconds since epoch.
    """
    return int(get_ts(dt) * 1000)


class QueryStatusEnum(str, enum.Enum):
    """
    Enum for the query status.

    Ref: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/get_query_results.html
    """

    Scheduled = "Scheduled"
    Running = "Running"
    Complete = "Complete"
    Failed = "Failed"
    Cancelled = "Cancelled"
    Timeout = "Timeout"
    Unknown = "Unknown"


def wait_logs_insights_query_to_succeed(
    logs_client,
    query_id: str,
    delta: int = 1,
    timeout: int = 30,
) -> dict:
    """
    Wait a given athena query to reach ``Complete`` status. If failed,
    raise ``RuntimeError`` immediately. If timeout, raise ``TimeoutError``.

    Ref: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/get_query_results.html

    :param logs_client: The boto3 cloudwatch logs client.
    :param query_id: The query id from the response of ``start_query`` API call.
    :param delta: The time interval in seconds between each query status check.
    :param timeout: The maximum time in seconds to wait for the query to succeed.
    """
    elapsed = 0
    for _ in range(999999):
        res = logs_client.get_query_results(queryId=query_id)
        status = res["status"]
        if status == QueryStatusEnum.Complete.value:
            return res
        elif status in [
            QueryStatusEnum.Failed.value,
            QueryStatusEnum.Cancelled.value,
            QueryStatusEnum.Timeout.value,
        ]:
            raise RuntimeError(f"query {query_id} reached status: {status}")
        else:
            time.sleep(delta)
        elapsed += delta
        if elapsed > timeout:
            raise TimeoutError(f"logs insights query timeout in {timeout} seconds!")


def strip_out_limit_clause(query: str) -> str:
    """
    Strip out the limit clause from a query string.
    """
    lines = query.splitlines()
    return "\n".join([line for line in lines if not line.startswith("| limit")])


def run_query(
    logs_client,
    start_datetime: datetime,
    end_datetime: datetime,
    query: str,
    log_group_name: T.Optional[str] = None,
    log_group_name_list: T.Optional[T.List[str]] = None,
    log_group_id_list: T.Optional[T.List[str]] = None,
    limit: int = 1000,
    delta: int = 1,
    timeout: int = 30,
) -> T.Tuple[str, dict]:
    """
    Run a logs insights query and wait for the query to succeed. It is a more
    human friendly wrapper of the ``start_query`` and ``get_query_results`` API.

    Ref: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/start_query.html

    :param logs_client: The boto3 cloudwatch logs client.
    :param start_datetime: python datetime object for start time,
        if timezone is not set, it assumes UTC.
    :param end_datetime: python datetime object for end time,
        if timezone is not set, it assumes UTC.
    :param query: The query string. don't use ``| limit abc`` in your query,
        use the ``limit`` parameter instead.
    :param log_group_name: see ``start_query`` API.
    :param log_group_name_list: see ``start_query`` API.
    :param log_group_id_list: see ``start_query`` API.
    :param limit: see ``start_query`` API.
    :param delta: The time interval in seconds between each query status check.
    :param timeout: The maximum time in seconds to wait for the query to succeed.
    """
    start_ts = get_ts_in_second(start_datetime)
    end_ts = get_ts_in_second(end_datetime)
    kwargs = dict(
        startTime=start_ts,
        endTime=end_ts,
        queryString=query,
        limit=limit,
    )
    if log_group_name is not None:
        kwargs["logGroupName"] = log_group_name
    elif log_group_name_list:
        kwargs["logGroupNames"] = log_group_name_list
    elif log_group_id_list:
        kwargs["logGroupIds"] = log_group_id_list
    else:  # it will raise error in API call
        pass
    res = logs_client.start_query(**kwargs)
    query_id = res["queryId"]
    res = wait_logs_insights_query_to_succeed(logs_client, query_id, delta, timeout)
    return query_id, res


def reformat_query_results(response: dict) -> T.List[dict]:
    """
    Convert the response from ``get_query_results`` API call to a more Pythonic
    format.

    :param response: the response from ``get_query_results`` API call.
    """
    return [
        {dct["field"]: dct["value"] for dct in result}
        for result in response.get("results", [])
    ]

shared.py 该测试所用到的一些常用的变量值.

# -*- coding: utf-8 -*-

from boto_session_manager import BotoSesManager
from aws_console_url.api import AWSConsole

bsm = BotoSesManager(profile_name="awshsh_app_dev_us_east_1")
aws = AWSConsole(aws_account_id=bsm.aws_account_id, aws_region=bsm.aws_region, bsm=bsm)
logs_client = bsm.cloudwatchlogs_client

group_name = "learn_aws_cloudwatch/Analyzing-log-data-with-CloudWatch-Logs-Insights"
stream_name_1 = "container-1"
stream_name_2 = "container-2"

data_faker.py 用于创建测试数据的脚本.

# -*- coding: utf-8 -*-

import typing as T
import time
import random
import dataclasses

from recipe import (
    create_log_group,
    delete_log_group,
    create_log_stream,
    Event,
    BaseJsonMessage,
    put_log_events,
)
from shared import bsm, logs_client, aws, group_name, stream_name_1, stream_name_2


def set_up():
    """
    Set up cloudwatch logs resource for this example.
    """
    create_log_group(logs_client, group_name)
    create_log_stream(logs_client, group_name, stream_name_1)
    create_log_stream(logs_client, group_name, stream_name_2)
    print(aws.cloudwatch.get_log_group(group_name))


@dataclasses.dataclass
class StatusMessage(BaseJsonMessage):
    server_id: str = dataclasses.field()
    status: str = dataclasses.field()


@dataclasses.dataclass
class ProcessingTimeMessage(BaseJsonMessage):
    server_id: str = dataclasses.field()
    processing_time: int = dataclasses.field()


server_id_list = [stream_name_1, stream_name_2]


def rand_event() -> T.List[T.Union[ProcessingTimeMessage, StatusMessage]]:
    """
    70% chance it succeeds, 30% chance it fails. When succeeded, it will generate
    two messages, one for status and one for processing time. When failed, it will
    generate one failed message for status.
    """
    server_id = random.choice(server_id_list)
    stream_name = server_id
    if random.randint(1, 100) <= 70:
        messages = [
            StatusMessage(
                server_id=server_id,
                status="succeeded",
            ),
            ProcessingTimeMessage(
                server_id=server_id,
                processing_time=random.randint(1000, 10000),
            ),
        ]
    else:
        messages = [
            StatusMessage(
                server_id=server_id,
                status="failed",
            )
        ]
    put_log_events(
        bsm.cloudwatchlogs_client,
        group_name,
        stream_name,
        events=[Event(message=message.to_json()) for message in messages],
    )
    return messages


def run_data_faker():
    """
    Run :func:`rand_event` every 1 second.

    Ref: https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/cloudwatch_limits_cwl.html

    The maximum batch size of a PutLogEvents request is 1MB.

    **800** transactions per second per account per Region, except for the following Regions where the quota is 1500 transactions per second per account per Region: US East (N. Virginia), US West (Oregon), and Europe (Ireland). You can request an increase to the per-second throttling quota by using the Service Quotas service.
    """
    ith = 0
    while True:
        ith += 1
        print(f"ith: {ith} sec")
        time.sleep(1)
        messages = rand_event()
        for message in messages:
            print(f"  {message}")


def clean_up():
    """
    Clearn up cloudwatch logs resource for this example.
    """
    delete_log_group(logs_client, group_name)


if __name__ == "__main__":
    set_up()
    run_data_faker()
    # clean_up()

run_query.py 用于测试 logs insights query 的脚本.

# -*- coding: utf-8 -*-

from datetime import datetime, timedelta, timezone

from rich import print as rprint

from recipe import run_query, reformat_query_results
from shared import logs_client, group_name

now = datetime.utcnow().replace(tzinfo=timezone.utc)
five_minutes_ago = now - timedelta(minutes=5)

query = """
fields @timestamp, @message, @logStream, @log
| sort @timestamp desc
""".strip()

query_id, res = run_query(
    logs_client,
    log_group_name=group_name,
    start_datetime=five_minutes_ago,
    end_datetime=now,
    query=query,
    limit=20,
)
print(f"query_id = {query_id}")
# res = logs_client.get_query_results(queryId="a1b2c3d4") # use this for a known query id
records = reformat_query_results(res)
print("records =")
rprint(records)
print(f"total records = {len(records)}")