pythonloggingamazon-cloudwatchlogshealth-checkaws-lambda-powertools

How to use aws_lambda_powertools logger to filter/disable specific api endpoint logs like health check /ping api


I tried to follow this general logging class filter() method but the method doesn't exist in aws_lambda_powertools logging and it throws an error. I am doing this to discard INFO: 127.0.0.1:51927 - "GET /ping HTTP/1.1" 200 OK rows in the aws cloud watch log as it gets fired every 20 sec through terraform health check which we can't disable.

Code I tried:

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

Error I am getting:

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

Refrence: https://github.com/encode/starlette/issues/864

Any help is highly appreciated. Thanks in advance.


Solution

  • I managed to fix the error in the question but still logs are getting printed on terminal and getting recorded in cloudwatch logs.

    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