无服务器使得在创纪录的时间内启动和运行 API 变得非常容易。但是如何在分布式架构中实现高水平的可观察性?在这篇文章中,我们将研究使用适用于 Python 的 AWS Lambda Powertools 为 FastAPI 应用程序检测日志记录、指标和跟踪功能。

本指南假定您知道如何使用 AWS SAM CLI 构建和部署 SAM 应用程序。如果您需要对此进行复习,请参阅我之前的博客文章。

0\。目录

1.什么是可观察性?

2.示例 FastAPI 应用程序

3.检查基线

4. AWS Lambda Powertools Python

5.记录

6.指标

7.追踪

8.结论

1\。什么是可观察性?

云计算和无服务器计算的兴起对我们构建、扩展和管理应用程序的方式产生了重大影响。开发人员现在可以以无与伦比的敏捷性和速度独立构建和部署应用程序。随着分布式应用程序变得越来越普遍,我们必须注意不要失去系统和服务的可观察性。不久之后,我们将达到一个单一的请求通过任意数量的函数、容器、队列和其他服务遍历我们的架构的地步。

通过在日志记录和指标方面实施最佳实践,以及能够在整个系统中跟踪请求,我们可以更好地了解我们的应用程序是如何执行的。可观察性的目标是检测我们的应用程序,以便当我们的分布式系统某处发生问题时,我们可以快速定位问题的根本原因。良好的可观察性还可以深入了解我们的应用程序和服务的使用方式,这些宝贵的见解可以指导未来的业务决策。

可观察性可以通过多种方式实现,在本文中,您将学习如何使用 AWS Lambda Powertools for Python 库检测在 Lambda 函数中运行的 FastAPI 应用程序。我们将专注于在我们的应用程序中实现三种不同的功能。

  1. 结构化日志记录 我们将上下文信息添加到日志中,例如请求和相关 ID、Lambda 和 FastAPI 上下文、服务名称、异常信息等。

  2. Metrics 以便我们可以跟踪我们的应用程序是如何使用的以及它是如何执行的。

  3. 跟踪以便我们可以在请求通过我们的系统时对其进行跟踪。

2\。示例 FastAPI 应用程序

[应用架构](https://res.cloudinary.com/practicaldev/image/fetch/s--LHGTLmaB--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https://www.eliasbrange .dev/posts/observability-with-fastapi-aws-lambda-powertools/architecture.png)

在此示例中,我们将使用一个简单的 FastAPI 应用程序,这是一个允许您添加、删除、列出、获取和更新存储在 DynamoDB 中的宠物的宠物 API。我们将使用 REST API 网关作为应用程序前面的代理。

如AWS Docs所示,HTTP API(在撰写本文时)不支持使用 AWS X-ray 进行跟踪,这就是我们必须改用 REST API 的原因。

示例文件

本示例中所需的文件可在GitHub或我的博客](https://www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/#sample-fastapi-application)上的[上找到。

3\。检查基线

在向应用程序添加更多高级功能之前,让我们以当前状态部署 API 以建立基线。在测试 API 时,我使用了这个Python 脚本。该脚本使用boto3从 CloudFormation 堆栈输出中获取 API URL。然后它向 API 运行一系列请求,触及所有端点。您还可以通过https://API_GATEWAY_URL/STAGE_NAME/docs上的 Swagger UI 或 Postman 等使用 cURL 发出请求。这是你的选择。

发出几个请求后,导航到 CloudWatch 控制台并转到_Logs Insights_。您应该在 Select log group(s) 下拉列表中看到一个名为/aws/lambda/FUNCTION_NAME的日志组。

[CloudWatch 日志没有任何花里胡哨](https://res.cloudinary.com/practicaldev/image/fetch/s--SCdFPLMh--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:/ /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/logs-baseline.png)

生成的日志记录几乎不包含任何有用的信息,这是因为现在我们只在代码中使用了简单的print(...)语句。让我们改变一下,好吗?

4\。 AWS Lambda Powertools Python

Lambda Powertools Python是一个库,它完全包含实用程序,可在 AWS Lambda 函数的可观察性方面轻松采用最佳实践。我们将专注于三个核心实用程序,它们是:Logger、Metrics 和 Tracer。

首先,在requirements.txt中,添加库。

mangum
fastapi
boto3
pydantic
+ aws-lambda-powertools

进入全屏模式 退出全屏模式

在应用程序代码中,您需要导入并初始化这三个实用程序。在新文件example/src/app/utils.py中,添加以下内容:

from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit  # noqa: F401


logger: Logger = Logger()
metrics: Metrics = Metrics()
tracer: Tracer = Tracer()

进入全屏模式 退出全屏模式

5\。日志记录

让我们首先向应用程序添加一些结构化日志记录。为此,我们将使用 Lambda Powertools 库中的Logger实用程序。

从打印报表升级

在dynamo.py文件中,添加from .utils import logger并将所有print(...)语句更改为logger.info(...)。经过几次请求,CloudWatch 中的体验应该会好一些。

[带有一些花里胡哨的 CloudWatch 日志](https://res.cloudinary.com/practicaldev/image/fetch/s--12NVgYzk--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:// /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/logs-basic-info.png)

在这里,我们可以看到结构化日志记录的威力。 CloudWatch 已识别出level、location、message、service等字段。我们现在可以使用这些字段来过滤查询。如果我们只想查看错误日志,我们可以在查询中添加一个过滤器:

fields @timestamp, message
| sort @timestamp desc
| limit 200
| filter level = "ERROR"

进入全屏模式 退出全屏模式

你问的service字段中的service_undefined值是多少?没错,我们在 SAM 模板中忘记了一件事。我们可以通过在函数中设置环境变量来控制 Lambda Powertools 使用的服务名称。如果您有很多不同的服务,那么在您对生产中的问题进行故障排除时,能够轻松地按服务名称过滤日志将是至关重要的。

Function:
  Type: AWS::Serverless::Function
  Properties:
    ...
    Environment:
      Variables:
+       POWERTOOLS_SERVICE_NAME: FastAPIPowerToolsExample
+       POWERTOOLS_METRICS_NAMESPACE: FastAPINamespace
        TABLE_NAME: !Ref Table
    ...

进入全屏模式 退出全屏模式

那没那么难吧?只需几行代码,我们现在在 FastAPI 应用程序中就有了结构良好的日志。让我们探索 Lambda Powertools Logger 中的更多功能。

Lambda 上下文

Powertools 使用inject_lambda_context装饰器可以轻松地将 Lambda 上下文中的信息添加到日志中,如下所示:

@logger.inject_lambda_context
def handler(event, context):
    ...

进入全屏模式 退出全屏模式

但是我们没有处理程序_function_,对吗?我们有一个包装 FastAPI 应用程序的Mangum对象。幸运的是,Mangum对象充当处理函数,因此我们可以在example/src/app/__init__.py中添加以下内容:

from .utils import logger  # add this import

...

handler = Mangum(app)
# add this line
handler = logger.inject_lambda_context(handler, clear_state=True)

进入全屏模式 退出全屏模式

在这里,我们设置参数clear_state=True来清除每次调用的状态。如果您想确保日志不被先前调用的状态污染,这很有用。

所以,多了一行代码。该行在 CloudWatch 控制台中为我们提供了什么?

[CloudWatch 日志与 Lambda 上下文](https://res.cloudinary.com/practicaldev/image/fetch/s--uENDEpzG--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https://www .eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/logs-lambda-context.png)

有关 Lambda 上下文的一些信息,例如函数名称、内存配置以及调用是否需要冷启动。好的!

相关 ID

当请求遍历多个服务时,相关 ID 可用于唯一标识请求。这是能够关联来自分布式系统中不同服务的日志的关键组件。

在我们的示例中,API 将接受一个可选的X-Correlation-Id标头,如果它不存在,它将使用来自 Lambda 上下文的请求 ID。然后可以将相关 ID 添加到对下游服务(如果有)的请求中,以便能够获得请求流的完整视图。Logger实用程序包括两个帮助函数来处理相关 ID,logger.set_correlation_id("ID")和logger.get_correlation_id()。

由于我们想在每个请求中提取相关 ID,并在每个响应中返回它,我们将使用FastAPI 中间件来实现它。

在example/src/app/__init__.py中,添加以下内容:

from fastapi import FastAPI, HTTPException, Request
...

@app.middleware("http")
async def add_correlation_id(request: Request, call_next):
    # Get correlation id from X-Correlation-Id header
    corr_id = request.headers.get("x-correlation-id")
    if not corr_id:
        # If empty, use request id from aws context
        corr_id = request.scope["aws.context"].aws_request_id

    # Add correlation id to logs
    logger.set_correlation_id(corr_id)

    response = await call_next(request)

    # Return correlation header in response
    response.headers["X-Correlation-Id"] = corr_id
    return response

进入全屏模式 退出全屏模式

用几个请求点击 API。如果您在请求中提供X-Correlation-Id标头,您应该在响应中看到相同的X-Correlation-Id。如果您不提供,则会为您生成一个。如果应用程序向下游服务发出进一步请求,则可以使用logger.get_correlation_id()检索相关 ID 并将其传递给下游服务。导航到 CloudWatch 并查看闪亮的新correlation_id字段。

[相关 ID 为的 CloudWatch 日志](https://res.cloudinary.com/practicaldev/image/fetch/s--2OzP-Px2--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:// /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/logs-correlation-id.png)

FastAPI 上下文

让我们将有关 FastAPI 请求上下文的一些信息添加到日志中。在我的头顶上,如果能够按请求路径、请求路由和请求方法过滤日志,那就太好了。既然我们在上面创建的中间件中有Request对象,我们不能在那里添加这个功能吗?我们不能。这是因为在 FastAPI 应用程序中进行任何路由之前执行中间件。因此,如果我们想要添加应用程序中声明的路由,例如/pets/{pet_id},我们需要使用自定义 APIRoute 类。

添加一个新文件example/src/app/router.py,并添加以下内容:

from fastapi import Request, Response
from fastapi.routing import APIRoute
from typing import Callable
from .utils import logger


class LoggerRouteHandler(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def route_handler(request: Request) -> Response:
            # Add fastapi context to logs
            ctx = {
                "path": request.url.path,
                "route": self.path,
                "method": request.method,
            }
            logger.append_keys(fastapi=ctx)
            logger.info("Received request")

            return await original_route_handler(request)

        return route_handler

进入全屏模式 退出全屏模式

在这里,我们向logger添加了一些字段,例如未来的每个logger.info(...)调用都包含这些字段。如果您不完全了解自定义APIRoute的工作原理,请阅读FastAPI 文档。他解释得比我能解释的要好得多。

现在,要使用自定义的APIRoute类,我们需要在example/src/app/__init__.py中添加以下内容:

...
from .router import LoggerRouteHandler  # add this import


app = FastAPI()
# add this line
app.router.route_class = LoggerRouteHandler

...

进入全屏模式 退出全屏模式

让我们看看这次我们在 CloudWatch 中有什么。

[具有 FastAPI 上下文的 CloudWatch 日志](https://res.cloudinary.com/practicaldev/image/fetch/s--be-Pgm0M--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:// /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/logs-fastapi-context.png)

看那美女。我们现在有方法、路由和实际路径的可过滤字段。想要获取当有人获取 ID 为123456的特定宠物时发生的所有日志?只需在查询中添加一个过滤器,例如| filter fastapi.method = "GET" and fastapi.path = "pets/123456"。想要查看所有对DELETE /pets/{pet_id}路由的调用的所有日志吗?你知道该做什么。

例外

Logger实用程序还可以通过使用logger.exception(...)记录异常来帮助我们更轻松地理解和调试错误。为了说明 Lambda Powertools 如何在此处为我们提供帮助,让我们首先添加一个端点来注入错误。将以下代码添加到example/src/app/__init__.py:

@app.get("/fail")
def fail():
    some_dict = {}
    return some_dict["missing_key"]

进入全屏模式 退出全屏模式

调用端点,您应该会收到500 Internal Server Error。前往 CloudWatch 并查看异常是如何记录的。

[没有任何可过滤键的 CloudWatch 日志异常](https://res.cloudinary.com/practicaldev/image/fetch/s--UllUQXd6--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https: //www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/logs-bare-exception.png)

看起来我们在这里遗漏了很多信息,例如相关 ID、路由、服务以及我们之前添加的所有其他字段。让我们通过添加自定义异常处理程序来解决这个问题。

创建一个捕获所有异常的异常处理程序似乎并不完全受支持。我已经尝试为Exception类型的异常创建一个异常处理程序,它似乎可以很好地进入处理程序并且日志记录工作正常。但是,它并没有完全捕捉到异常,所以异常仍然被传播并最终在 Mangum 事件循环中被捕捉到。它也不会进入在响应中设置相关 ID 的中间件。

这个问题似乎可以通过添加starlette.exceptions.ExceptionMiddleware来解决,但我不确定这是否会产生任何无法预料的副作用。使用风险自负! :)

有关FastAPI GitHub 问题和Starlette GitHub 问题的更多信息:

在example/src/app/__init__.py中,添加以下内容:

from fastapi import FastAPI, HTTPException
# add the following imports
from fastapi.responses import JSONResponse
from starlette.exceptions import ExceptionMiddleware

...

app = FastAPI()
app.router.route_class = LoggerRouteHandler
# add the below code
# +++
app.add_middleware(ExceptionMiddleware, handlers=app.exception_handlers)


@app.exception_handler(Exception)
async def unhandled_exception_handler(request, err):
    logger.exception("Unhandled exception")
    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

# +++

进入全屏模式 退出全屏模式

再次调用/fail端点并转到 CloudWatch。

[带有异常和可过滤键的 CloudWatch 日志](https://res.cloudinary.com/practicaldev/image/fetch/s--CdBpGHSe--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:// /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/logs-exception-with-fields.png)

现在,每当发生未处理的异常时,我们都会在日志中获得大量相关信息,这使得如果您手头有相关 ID,则可以轻松找到导致请求失败的原因。

日志现在更有用了,我们不需要向应用程序添加那么多代码。大多数代码驻留在中间件、异常处理程序或自定义APIRoute类中。对实际业务逻辑的更改很小,这表明 Lambda Powertools 可以轻松添加到现有应用程序中以改进日志记录实践。

6\。指标

让我们探索 Lambda Powertools 中的下一个核心实用程序,即Metrics实用程序。此实用程序可让您通过处理所有必要的样板文件轻松地将指标推送到 CloudWatch。它使用Amazon CloudWatch Embedded Metrics Format异步工作,将指标记录到标准输出。它还汇总每次调用的所有指标,以节省对 CloudWatch 的调用次数。

这里有一些术语需要注意。 CloudWatch 指标分组在称为命名空间的_容器_中。如果一个应用程序包含多个服务,例如,您可以为所有服务使用相同的命名空间来对该应用程序的所有指标进行分组。指标也可以有维度,它们是作为元数据添加到指标的键值对,允许您根据维度值过滤和聚合指标。

Metrics实用程序的默认配置使用两个环境变量POWERTOOLS_METRICS_NAMESPACE和POWERTOOLS_SERVICE_NAME,其中前者指定指标命名空间,后者为所有指标添加维度service=POWERTOOLS_SERVICE_NAME。

仪表

如果到目前为止您按照指南进行操作,您应该已经在example/src/app/utils.py中有一个metrics对象。我们使用了默认配置,它从环境变量中设置命名空间和服务维度。您还可以明确指定这些,如下所示:

from aws_lambda_powertools import Metrics

metrics = Metrics() # Sets metric namespace and service dimension via environment variables
metrics = Metrics(namespace="Namespace", service="ServiceA") # Sets metric namespace to "Namespace" and service dimension to "ServiceA"

进入全屏模式 退出全屏模式

如果您想为所有指标添加另一个维度,例如环境,您可以使用set_default_dimensions方法执行此操作:

from aws_lambda_powertools import Metrics

metrics = Metrics()
metrics.set_default_dimensions(environment="development")

进入全屏模式 退出全屏模式

现在,要在应用程序中启用该功能,请将以下内容添加到example/src/app/__init__.py:

...
# Add metrics to the following import
from .utils import logger, metrics

...

handler = Mangum(app)
handler = logger.inject_lambda_context(handler, clear_state=True)
# Add the following (last to properly flush metrics):
handler = metrics.log_metrics(handler, capture_cold_start_metric=True)

进入全屏模式 退出全屏模式

确保log_metricsdecorator 最后添加到处理函数。

添加指标

让我们将以下指标添加到应用程序中:

  • 创建宠物的次数

  • 发生未处理异常的次数

创建的宠物指标

要添加一个计算宠物创建次数的指标,我们可以将以下内容添加到POST /pets路由中。我们在调用 DynamoDB 之后添加它,以便仅在成功创建宠物时记录一个指标。

@app.post("/pets", status_code=201, response_model=models.PetResponse)
def post_pet(payload: models.CreatePayload):
    res = dynamo.create_pet(kind=payload.kind, name=payload.name)
    # Add the following line
    metrics.add_metric(name="CreatedPets", unit=MetricUnit.Count, value=1)
    return res

进入全屏模式 退出全屏模式

未处理的异常度量

要计算未处理异常的数量,我们可以在异常处理程序中添加以下内容:

@app.exception_handler(Exception)
async def validation_exception_handler(request, err):
    # add the following line
    metrics.add_metric(name="UnhandledExceptions", unit=MetricUnit.Count, value=1)
    logger.exception("Unhandled exception")
    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

进入全屏模式 退出全屏模式

使用不同尺寸

使用 CloudWatch EMF 时,文档中的所有指标必须具有相同的维度。因此,对add_metric的所有调用都将生成具有相同维度、service以及您添加的任何其他默认维度的指标。要为特定指标添加不同的维度,我们需要使用single_metric。例如,我们可能想要一个名为RequestCount的指标,它同时具有服务和路由维度。

首先,在example/src/app/utils.py中,更新以下导入以包含single_metric:

from aws_lambda_powertools import Logger, Metrics, Tracer
# Add single_metric to the following import
from aws_lambda_powertools.metrics import MetricUnit, single_metric  # noqa: F401

进入全屏模式 退出全屏模式

然后,在example/src/app/router.py中,添加以下代码:

from fastapi import Request, Response
from fastapi.routing import APIRoute
from typing import Callable
# Add single_metric to the following import
from .utils import metrics, MetricUnit, single_metric


class LoggerRouteHandler(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def route_handler(request: Request) -> Response:
            # Add fastapi context to logs
            ctx = {
                "path": request.url.path,
                "route": self.path,
                "method": request.method,
            }
            logger.append_keys(fastapi=ctx)
            logger.info("Received request")

            # Add the following:
            # +++
            # Add count metric with method + route as dimension
            with single_metric(name="RequestCount", unit=MetricUnit.Count, value=1) as metric:
                metric.add_dimension(name="route", value=f"{request.method} {self.path}")
            # +++

            return await original_route_handler(request)

        return route_handler

进入全屏模式 退出全屏模式

CloudWatch 放映时间

让我们在不同的路由以及/fail端点上多次访问 API。然后,导航到 CloudWatch 并单击左侧的 All metrics。然后,单击您指定的_Namespace_。您应该看到以下内容。

[按维度划分的 CloudWatch 指标](https://res.cloudinary.com/practicaldev/image/fetch/s--NsTX_xkV--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https://www. eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/metrics-namespace.png)

首先,点击function_name, service。在这里,您应该会看到一个指标,该指标计算函数经历冷启动的次数,当您添加capture_cold_start_metric=True参数时,该指标由 Lambda Powertools 指标实用程序自动添加。

[CloudWatch 指标,冷启动次数为](https://res.cloudinary.com/practicaldev/image/fetch/s--sBMYdsEc--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:// /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/metrics-cold-starts.png)

接下来,点击命名空间下的service。在这里,您将能够看到已创建的宠物数量,以及未处理的异常数量。

[CloudWatch 指标异常和创建宠物](https://res.cloudinary.com/practicaldev/image/fetch/s--qTlp1heT--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:/ /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/metrics-exceptions-and-created-pets.png)

最后,查看路线、服务。在这里,您应该有描述每个路由的请求总量的指标。

[带有路由调用的 CloudWatch 指标](https://res.cloudinary.com/practicaldev/image/fetch/s--3MHJ0p-1--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:/ /www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/metrics-routes.png)

这展示了在使用 Lambda Powertools 处理繁重的工作时,向 FastAPI Lambda 应用程序添加指标是多么容易。

7\。追踪

现在,让我们继续讨论最终的核心实用程序Tracer。该实用程序可以轻松检测您的 Lambda 函数以捕获跟踪并将它们发送到 AWS X-Ray。 AWS X-Ray 是一个分布式跟踪系统,可以帮助您分析和调试分布式应用程序,并提供一种在请求通过您的不同服务时跟踪它们的方法。

Powertools 还具有 X-Ray 支持的自动修补模块的便利性,例如boto3。每当您使用boto3从您的 Lambda 函数中调用 DynamoDB 或其他 AWS 服务时,这将自动在您的跟踪中创建分段,同样适用于其他受支持的requests等模块。

仪表

首先,我们需要在 Lambda 函数和 API Gateway 上启用跟踪。在 SAM 模板中,添加以下内容:

Function:
  Type: AWS::Serverless::Function
  Properties:
+   Tracing: Active
    MemorySize: 128
...

RestApi:
  Type: AWS::Serverless::Api
  Properties:
    StageName: Prod
+   TracingEnabled: true

进入全屏模式 退出全屏模式

然后,与其他实用程序一样,我们需要装饰我们的处理程序。将以下内容添加到example/src/app/__init__.py:

...
# Add tracer to the following import
from .utils import tracer, logger, metrics, MetricUnit

...

handler = Mangum(app)

# Add the following lines
# +++
handler.__name__ = "handler"  # tracer requires __name__ to be set
handler = tracer.capture_lambda_handler(handler)
# +++

handler = logger.inject_lambda_context(handler, clear_state=True)
handler = metrics.log_metrics(handler, capture_cold_start_metric=True)

进入全屏模式 退出全屏模式

如果您想测量特定的方法调用并将其添加到您的跟踪中,您可以使用capture_method装饰器来装饰函数。出于说明目的,将以下内容添加到example/src/app/dynamo.py:

...
# Add the following import
from .utils import tracer

...

@tracer.capture_method  # add this
def create_pet(kind: str, name: str) -> dict:
...

@tracer.capture_method  # and this
def get_pet(pet_id: str) -> dict:
...

@tracer.capture_method  # this as well
def update_pet(pet_id: str, kind: str = None, name: str = None):
...

@tracer.capture_method  # and this
def list_pets(next_token: str = None) -> dict:
...

@tracer.capture_method # and finally this
def delete_pet(pet_id: str):
...

进入全屏模式 退出全屏模式

部署 API,并向其发送几个请求。稍微混合一下并点击/fail端点几次,并尝试通过尝试访问不存在的宠物或创建具有无效负载的宠物来生成一些4xx响应。转到 CloudWatch 控制台并单击左侧的Service Map。

[AWS X-Ray 服务地图](https://res.cloudinary.com/practicaldev/image/fetch/s--mpy9nV9O--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:// www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/tracing-service-map.png)

在这里,您可以看到 AWS X-Ray 生成了我们整个架构的服务地图。我们可以看到客户端向 API Gateway 发出了请求,API Gateway 将请求转发给 Lambda 服务,Lambda 服务又调用 Lambda 函数,然后向 DynamoDB 表发出进一步的请求。随着系统的发展,服务地图将成为获得整个系统全貌的宝贵工具。对于地图中的每个节点,还可以查看成功请求的百分比、平均延迟和其他有用的指标。

您还可以深入了解各个跟踪。您可以在下面看到创建宠物请求的时间表。

[AWS X-Ray 分段时间线](https://res.cloudinary.com/practicaldev/image/fetch/s--ArymGGG5--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https:// www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/tracing-segments-timeline.png)

高级功能

tracer 实用程序中有很多高级功能,例如添加元数据和注释、记录响应和捕获异常。

例如,当我们筹集PetNotFoundError时,它将在相关的细分详细信息中可见。

[AWS X-Ray 异常详情](https://res.cloudinary.com/practicaldev/image/fetch/s--UNnc36-r--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https: //www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/tracing-exception.png)

我们还可以将关联 ID 添加为注释,以允许我们通过关联 ID 过滤和查询跟踪。在add_correlation_id中间件中,添加tracer.put_annotation(key="correlation_id", value=corr_id):

@app.middleware("http")
async def add_correlation_id(request: Request, call_next):
    # Get correlation id from X-Correlation-Id header
    corr_id = request.headers.get("x-correlation-id")
    if not corr_id:
        # If empty, use request id from aws context
        corr_id = request.scope["aws.context"].aws_request_id

    # Add correlation id to logs
    logger.set_correlation_id(corr_id)

    # Add correlation id to traces
    tracer.put_annotation(key="correlation_id", value=corr_id)

    response = await call_next(request)

    # Return correlation header in response
    response.headers["X-Correlation-Id"] = corr_id
    return response

进入全屏模式 退出全屏模式

现在,通过深入到##handler段,您应该会在correlation_id注释中看到相关 ID。

[具有相关 ID 注释的 AWS X-Ray 段](https://res.cloudinary.com/practicaldev/image/fetch/s--RqZ-CMZ---/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto% 2Cw_880/https://www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/tracing-corr-id-segment.png)

如果要查找特定 ID 的跟踪,可以使用控制台中的查询功能。

[AWS X-Ray 通过相关 ID 查询](https://res.cloudinary.com/practicaldev/image/fetch/s--kChDs116--/c_limit%2Cf_auto%2Cfl_progressive%2Cq_auto%2Cw_880/https: //www.eliasbrange.dev/posts/observability-with-fastapi-aws-lambda-powertools/tracing-corr-id-query.png)

8\。结论

目前为止就这样了。希望您对如何使用 Lambda Powertools 库在 FastAPI 应用程序(以及一般的 Lambda 函数)的可观察性方面实施最佳实践有所了解。 AWS CloudWatch 和 AWS X-Ray 是帮助您分析和监控无服务器应用程序的绝佳工具。 Lambda Powertools 让开始使用这些服务变得异常容易,让您可以专注于应用程序逻辑,同时将指标、日志和跟踪的实现留给库。

# aws # serverless
Logo
云原生

云原生社区为您提供最前沿的新闻资讯和知识内容

更多推荐

cover

本地Docker部署Navidrome音乐服务器与远程访问听歌详细教程

avatar 云原生
cover

【docker系列】docker删除指定容器

avatar 云原生
cover

【Docker系列】制作基础镜像

avatar 云原生