在 CloudWatch 仪表板中显示不同数量的用户调用 API Gateway

Question

如何获得在给定时间范围内使用我的服务的不同用户数量？用户数量必须显示在 CloudWatch 控制面板中。

我将 Cognito 与用于用户身份验证的托管 UI、HTTP API Gateway 以及用于授权的 Lambda 集成一起使用，并且 API Gateway 请求由另一个 Lambda 函数处理。

在 API Gateway 的 CloudWatch 访问日志中，我可以记录用户名。我知道我可以在 CloudWatch Insights 中使用 stats count(*) by username 来计算每个用户向 API Gateway 发送了多少请求，但我不知道如何获取不同用户的列表。 count_distinct 不起作用，因为它只会近似用户，因为该字段可以具有高基数。

最后，我希望在我的 CloudWatch 仪表板中有一个数字小部件，它将显示在所选时间范围内使用该服务的不同用户数量。

Answer 1

使用custom CloudWatch dashboard widgets，我决定构建一个执行日志洞察查询并将结果呈现为自定义小部件的 Lambda 函数。

import os

import boto3
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.data_classes import (
    CloudWatchDashboardCustomWidgetEvent,
    event_source,
)
from aws_lambda_powertools.utilities.typing import LambdaContext

LOG_GROUP_NAME = os.environ["LOG_GROUP_NAME"]

logger = Logger()
cloud_watch_logs = boto3.client("logs")

DOCS = """
## User Widget
A script to get the number of unique users accessing the API in a given time range.
"""

CSS = """
<style>
.container {
    align-content: center;
    align-items: center;
    display: flex;
    flex-direction: row;
    justify-content: center;
    width: 100%;
}

.value {
    font-size: 45px;
}
</style>"""


def get_unique_api_users(start_time: int, end_time: int) -> int:
    start_query_response = cloud_watch_logs.start_query(
        logGroupName=LOG_GROUP_NAME,
        startTime=start_time,
        endTime=end_time,
        queryString='filter ispresent(user) and user != "-" | stats count(*) as userCount by user',
        limit=10000,
    )

    response = None
    while response == None or response["status"] != "Complete":
        response = cloud_watch_logs.get_query_results(
            queryId=start_query_response["queryId"]
        )

    return len(response["results"])


@logger.inject_lambda_context(log_event=False)
@event_source(data_class=CloudWatchDashboardCustomWidgetEvent)
def lambda_handler(event: CloudWatchDashboardCustomWidgetEvent, context: LambdaContext):
    if event.describe:
        return DOCS

    start_time = event.widget_context.time_range.start
    end_time = event.widget_context.time_range.end
    if event.widget_context.time_range.zoom_start:
        start_time = event.widget_context.time_range.zoom_start
        end_time = event.widget_context.time_range.zoom_end

    return f"""
{CSS}
<div class="container">
    <div class="value">
        ? {get_unique_api_users(start_time=start_time, end_time=end_time)}
    </div>
</div>"""

通过这种方法，我们可以确保获得准确数量的 API 用户。不利的一面是，我们查询的日志越多，拥有的用户越多，获取用户数量所需的时间就越长。此外，每次我们刷新小部件时，都会调用一个 Lambda 函数，计入我们在该区域的并发执行限制，并且每次调用都要花钱，尽管可以说这只是很少的钱。