Лучшие практики наблюдаемости при запуске FastAPI в лямбде

Serverless позволяет невероятно легко запустить API в рекордно короткие сроки. Но как добиться высокого уровня наблюдаемости в распределенной архитектуре? В этом посте мы рассмотрим инструментальные возможности протоколирования, метрики и трассировки для FastAPI-приложения с помощью AWS Lambda Powertools for Python.

Это руководство предполагает, что вы знаете, как создать и развернуть SAM-приложение с помощью AWS SAM CLI. Если вам нужно освежить этот вопрос, пожалуйста, обратитесь к моей предыдущей статье в блоге.

0. Оглавление

1. Что такое наблюдаемость?
2. Образец приложения FastAPI
3. Проверка базовой линии
4. AWS Lambda Powertools Python
5. Ведение логов
6. Метрики
7. Трассировка
8. Заключение

1. Что такое наблюдаемость?

Развитие облачных и бессерверных вычислений оказало большое влияние на то, как мы создаем, масштабируем и управляем приложениями. Теперь разработчики могут самостоятельно создавать и развертывать приложения с непревзойденной гибкостью и скоростью. Поскольку распределенные приложения становятся все более распространенными, мы должны позаботиться о том, чтобы не потерять наблюдаемость в наших системах и сервисах. Вскоре мы придем к тому, что один запрос будет проходить через любое количество функций, контейнеров, очередей и других сервисов.

Внедряя лучшие практики ведения журналов и метрик, а также имея возможность проследить запрос через всю систему, мы можем получить лучшее понимание того, как работают наши приложения. Цель наблюдаемости — инструментировать наши приложения, чтобы при возникновении проблемы где-то в распределенной системе мы могли быстро найти ее первопричину. Хорошая наблюдаемость также позволяет понять, как используются наши приложения и сервисы, а это ценная информация, которая может помочь в принятии будущих бизнес-решений.

Наблюдаемость может быть достигнута несколькими способами, и в этой статье вы узнаете, как инструментировать приложение FastAPI, запущенное внутри функции Lambda, используя библиотеку AWS Lambda Powertools for Python. Мы сосредоточимся на реализации трех различных возможностей в нашем приложении.

  1. Структурированное логирование, при котором мы добавляем в журналы контекстную информацию, такую как идентификаторы запроса и корреляции, контекст Lambda и FastAPI, имя сервиса, информацию об исключениях и многое другое.
  2. Метрики, чтобы мы могли отслеживать, как используется наше приложение и как оно работает.
  3. Трассировка, чтобы мы могли отслеживать запросы по мере их прохождения через наши системы.

2. Пример приложения FastAPI

В этом примере мы будем использовать простое FastAPI-приложение, Pets API, которое позволяет добавлять, удалять, перечислять, получать и обновлять домашних животных, хранящихся в DynamoDB. Мы будем использовать REST API Gateway для работы в качестве прокси перед приложением.

Как показано в AWS Docs, HTTP API не поддерживают (на момент написания статьи) трассировку с помощью AWS X-ray, поэтому мы должны использовать REST API вместо этого.

Файлы примера

Необходимые файлы для этого примера доступны на GitHub или здесь, в моем блоге.

3. Проверка базовой линии

Прежде чем мы добавим в приложение более продвинутые функции, давайте развернем API в его текущем состоянии, чтобы установить базовую линию. При тестировании API я использовал этот скрипт Python. Сценарий использует boto3 для получения URL API из результатов стека CloudFormation. Затем он запускает последовательность запросов к API, затрагивая все конечные точки. Вы также можете отправлять запросы с помощью cURL, через Swagger UI, доступный на https://API_GATEWAY_URL/STAGE_NAME/docs, или с помощью чего-то вроде Postman. Выбор за вами.

После выполнения нескольких запросов перейдите в консоль CloudWatch и зайдите в раздел Logs Insights. Вы должны увидеть группу журналов под названием /aws/lambda/FUNCTION_NAME в выпадающем списке Select log group(s).

Полученные записи журнала не содержат практически никакой полезной информации, а все потому, что сейчас мы используем в коде только простые операторы 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. Ведение журнала

Давайте начнем с добавления структурированного протоколирования в приложение. Для этого мы будем использовать утилиту Logger из библиотеки Lambda Powertools.

Переход от операторов печати

В файле dynamo.py добавьте from .utils import logger и измените все утверждения print(...) на logger.info(...). После нескольких запросов опыт работы в CloudWatch должен был стать немного лучше.

Здесь мы можем увидеть силу структурированного протоколирования. CloudWatch определил такие поля, как level, location, message, service и так далее. Теперь мы можем использовать эти поля для фильтрации запросов. Если мы хотим видеть только журналы ошибок, мы можем добавить фильтр к запросу:

fields @timestamp, message
| sort @timestamp desc
| limit 200
| filter level = "ERROR"
Войти в полноэкранный режим Выйти из полноэкранного режима

Что это за значение service_undefined в поле service, спросите вы? Правильно, мы забыли одну вещь в шаблоне 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.

Контекст лямбда

Powertools позволяет легко добавлять информацию из контекста Lambda в журналы с помощью декоратора inject_lambda_context, как показано здесь:

@logger.inject_lambda_context
def handler(event, context):
    ...
Вход в полноэкранный режим Выход из полноэкранного режима

Но у нас нет функции-обработчика, не так ли? У нас есть объект Mangum, обертывающий приложение FastAPI. К счастью, объект 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?

Некоторую информацию о контексте Lambda, такую как имя функции, конфигурация памяти, а также о том, требовал ли вызов холодного старта или нет. Отлично!

Идентификатор корреляции

Идентификаторы корреляции могут использоваться для уникальной идентификации запроса при его прохождении через несколько сервисов. Это ключевой компонент для корреляции журналов различных служб в распределенных системах.

В нашем примере API принимает необязательный заголовок X-Correlation-Id, а если он отсутствует, то используется идентификатор запроса из контекста Lambda. Идентификатор корреляции может быть добавлен к запросам к последующим сервисам (если таковые имеются), чтобы получить полное представление о потоке запросов. Утилита Logger включает две вспомогательные функции для работы с идентификаторами корреляции, 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() и передан последующему сервису. Перейдите в CloudWatch и посмотрите на новое блестящее поле correlation_id.

Контекст 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 на этот раз.

Посмотрите на эту красоту. Теперь у нас есть фильтруемые поля для метода, маршрута и фактического пути. Хотите получить все журналы, которые произошли, когда кто-то взял определенного питомца с ID 123456? Просто добавьте фильтр к запросу, например | filter fastapi.method = "GET" и 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 и посмотрите, как было зарегистрировано исключение.

Похоже, что нам не хватает много информации, такой как ID корреляции, маршрут, сервис и все остальные поля, которые мы добавили ранее. Давайте исправим это, добавив пользовательский обработчик исключений.

Создание обработчика исключений, который перехватывает все исключения, похоже, не полностью поддерживается. Я попробовал просто создать обработчик исключений для исключений типа Exception, и, похоже, обработчик нормально заходит в обработчик, и логирование работает. Однако он не перехватывает исключение полностью, поэтому исключение все равно распространяется и в конце концов перехватывается в цикле событий Mangum. Оно также не попадает в промежуточное ПО, которое устанавливает идентификатор корреляции в ответе.

Проблема, похоже, решается добавлением starlette.exceptions.ExceptionMiddleware, хотя я не уверен, что это может иметь какие-либо непредвиденные побочные эффекты. Используйте на свой страх и риск! 🙂

Дополнительная информация на FastAPI GitHub Issue и Starlette GitHub Issue:

В 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.

Теперь у нас есть много релевантной информации в журналах, когда возникает необработанное исключение, что позволяет легко найти причину сбоя запроса, если у вас под рукой есть идентификатор корреляции.

Журналы теперь намного полезнее, и нам не пришлось добавлять так много кода в приложение. Большая часть кода находится в промежуточном ПО, обработчиках исключений или в пользовательском классе APIRoute. Изменения в фактической бизнес-логике были минимальными, что показывает, что Lambda Powertools можно легко добавить в существующие приложения для улучшения практики протоколирования.

6. Метрики

Давайте рассмотрим следующую основную утилиту в Lambda Powertools, утилиту Metrics. Эта утилита позволяет легко передавать метрики в CloudWatch, позаботившись обо всем необходимом. Она работает асинхронно, используя Amazon CloudWatch Embedded Metrics Format, записывая метрики в stdout. Он также агрегирует все метрики от каждого вызова, чтобы сэкономить на количестве обращений к CloudWatch.

Здесь необходимо знать некоторую терминологию. Метрики CloudWatch группируются в контейнеры, называемые пространствами имен. Если приложение состоит из нескольких сервисов, вы можете, например, использовать одно и то же пространство имен для всех них, чтобы сгруппировать все метрики для этого приложения. Метрики также могут иметь измерения, которые представляют собой пары ключ-значение, добавляемые в качестве метаданных к метрикам, что позволяет фильтровать и агрегировать метрики в зависимости от значений измерений.

По умолчанию утилита Metrics использует две переменные окружения POWERTOOLS_METRICS_NAMESPACE и POWERTOOLS_SERVICE_NAME, где первая определяет пространство имен метрик, а вторая добавляет измерение service=POWERTOOLS_SERVICE_NAME ко всем метрикам.

Инструментирование

Если вы следовали этому руководству до сих пор, у вас уже должен быть объект metrics в example/src/app/utils.py. Там мы использовали конфигурацию по умолчанию, которая устанавливает пространство имен и размерность сервиса из переменных окружения. Вы также можете задать их явно, например, так:

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_metrics добавлен последним в функцию-обработчик.

Добавление метрик

Давайте добавим в приложение следующие метрики:

  • Количество раз, когда создается питомец
  • Количество случаев возникновения необработанного исключения

Метрика созданных питомцев

Чтобы добавить метрику, которая подсчитывает, сколько раз был создан питомец, мы можем добавить следующее в маршрут 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

Давайте несколько раз обратимся к API по разным маршрутам, а также к конечной точке /fail. Затем перейдите в CloudWatch и нажмите на Все метрики с левой стороны. Затем нажмите на указанное вами пространство имен. Вы должны увидеть следующее.

Сначала щелкните на function_name, service. Здесь вы должны увидеть метрику, подсчитывающую количество раз, когда функция подвергалась холодному запуску, которая автоматически добавляется утилитой метрик Lambda Powertools, когда вы добавляете параметр capture_cold_start_metric=True.

Далее нажмите на сервис в пространстве имен. Здесь вы сможете увидеть количество созданных питомцев, а также количество необработанных исключений.

Наконец, проверьте route, service. Здесь у вас должны быть метрики, описывающие общее количество запросов к каждому маршруту.

Это показывает, как легко добавить метрики в приложение FastAPI Lambda при использовании Lambda Powertools.

7. Трассировка

Теперь перейдем к последней основной утилите, Tracer. Эта утилита позволяет легко инструментировать вашу функцию Lambda для захвата трассировок и отправки их в AWS X-Ray. AWS X-Ray — это распределенная система трассировки, которая может помочь вам анализировать и отлаживать распределенные приложения, а также предоставляет возможность отслеживать запросы по мере их прохождения через различные сервисы.

В Powertools также есть возможность автоматического подключения модулей, поддерживаемых X-Ray, таких как boto3. Это позволит автоматически создавать сегменты в ваших трассировках всякий раз, когда вы используете boto3 для вызова DynamoDB или другого сервиса AWS из вашей функции Lambda, и аналогично для других поддерживаемых модулей, таких как 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 сгенерировал карту сервисов всей нашей архитектуры. Мы видим, что клиент сделал запрос к шлюзу API, который перенаправил запрос к службе Lambda, которая, в свою очередь, вызвала функцию Lambda, которая затем сделала дальнейшие запросы к таблице DynamoDB. По мере роста системы карта сервисов станет бесценным инструментом для получения полной картины всей системы. Для каждого узла в карте можно также увидеть процент успешных запросов, среднюю задержку и другие полезные метрики.

Вы также можете детализировать отдельные трассы. Ниже показана временная шкала для запроса на создание домашнего животного.

Расширенные возможности

В утилите трассировки есть множество дополнительных функций, таких как добавление метаданных и аннотаций, запись ответов и перехват исключений.

Например, когда мы поднимаем PetNotFoundError, это будет видно в деталях соответствующего сегмента.

Мы также можем добавить идентификатор корреляции в качестве аннотации, чтобы позволить нам фильтровать и запрашивать трассы по идентификатору корреляции. В промежуточном ПО 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, вы можете воспользоваться функцией запроса в консоли.

8. Заключение

На этом пока все. Надеюсь, вы узнали кое-что о том, как можно использовать библиотеку Lambda Powertools для реализации лучших практик, когда речь идет о наблюдаемости для FastAPI-приложений (и Lambda-функций в целом). AWS CloudWatch и AWS X-Ray — это замечательные инструменты, которые помогут вам анализировать и контролировать бессерверные приложения. Lambda Powertools позволяет невероятно легко начать использовать эти сервисы, позволяя вам сосредоточиться на логике приложения, оставив реализацию метрик, журналов и трассировок библиотеке.

Оставьте комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *