如何使用 aws_lambda_powertools 记录器过滤/禁用特定的 api 端点日志,如健康检查/ping api

问题描述 投票:0回答:1

我试图遵循这个通用日志记录类 filter() 方法,但该方法在 aws_lambda_powertools 日志记录中不存在,它会抛出错误。我这样做是为了丢弃 aws 云监视日志中的

INFO:     127.0.0.1:51927 - "GET /ping HTTP/1.1" 200 OK
行,因为它每 20 秒通过我们无法禁用的 terraform 健康检查触发一次。

我试过的代码:

import os
from aws_lambda_powertools import Logger, logging

class EndpointFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        return record.getMessage().find("/ping") == -1

def get_aws_powertool_logger():
    date_format = "%Y-%m-%dT%H:%M:%S.%f%z"

    service = os.path.basename(os.getcwd())
    print("---- logger---- ", service)
    # logger: Logger = Logger(service=service, datefmt=date_format, level="DEBUG")
    logger = logging.getLogger("uvicorn.access").addFilter(EndpointFilter())

    return logger

我得到的错误:

  class EndpointFilter(logging.Filter):
AttributeError: module 'aws_lambda_powertools.logging' has no attribute 'Filter'

参考:https://github.com/encode/starlette/issues/864

非常感谢任何帮助。提前致谢。

python logging amazon-cloudwatchlogs health-check aws-lambda-powertools
1个回答
0
投票

我设法解决了问题中的错误,但日志仍在终端上打印并记录在 cloudwatch 日志中。

import os
from aws_lambda_powertools import Logger
import logging
import typing as t

class EndpointFilter(logging.Filter):
    def __init__(
        self,
        path: str,
        *args: t.Any,
        **kwargs: t.Any,
    ):
        super().__init__(*args, **kwargs)
        self._path = path

    def filter(self, record: logging.LogRecord) -> bool:
        return record.getMessage().find(self._path) == -1

def get_aws_powertool_logger():
    date_format = "%Y-%m-%dT%H:%M:%S.%f%z"
    service = os.path.basename(os.getcwd())
    logger: Logger = Logger(service=service, datefmt=date_format, level="INFO")
    logger.addFilter(EndpointFilter("/ping"))
    return logger
© www.soinside.com 2019 - 2024. All rights reserved.