Search code examples
python-3.xunit-testingloggingaws-lambdaamazon-cloudwatch

Lambda CloudWatch Log : Logger doesn't log, but print works


Problem

I cannot use my logger to log to the console. Is this because logging requires writing the contents of the log to a file, which Lambda cannot do, because it's a function invocation ? Can we not use the CloudWatch Log as the "file".

Clarifications

  • Lambda has the permission to write into CloudWatch via the basic AWSLambdaBasicExecutionRole policy

I am using terms such as "file", function invocation loosely. Please critique freely.

Code I tested on

import logging
import os
import json
import boto3
from collections import defaultdict
from datetime import datetime, timedelta

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()

IP_COUNT_TABLE = os.getenv("AWS_IP_COUNT_TABLE", "ip-tracking-list")
DENIED_IP_TABLE = os.getenv("AWS_DENIED_IP_TABLE", "ip-denied-list")
PROCESSED_EVENTS_TABLE = os.getenv("AWS_PROCESSED_EVENTS_TABLE", "cloudwatch-my-logs")
HTTP_REQUEST_HEADER_TO_CHECK = os.getenv("HTTP_REQUEST_HEADER_TO_CHECK", "some.endpoint.com")

# Initialize DynamoDB client
dynamodb = boto3.resource('dynamodb', region_name='us-west-1')

def lambda_handler(event, context):

    ip_count_table = dynamodb.Table(IP_COUNT_TABLE)
    denied_ip_table = dynamodb.Table(DENIED_IP_TABLE)
    processed_events_table = dynamodb.Table(PROCESSED_EVENTS_TABLE)

    ip_count = defaultdict(int)

    try:
        logger.info(f"Processing event: {event}")
        print(f"Processing event: {event}")
        for record in event.get('logEvents', []):
            event_id = record.get('id', '')
            message = record.get('message', {})
            if not event_id:
                logger.error("No timestamp found in log event")
                continue

            if check_event_processed(str(event_id), processed_events_table):
                continue

            if 'httpRequest' in message and 'headers' in message['httpRequest']:
                for header in message['httpRequest']['headers']:
                    if header['name'].lower() == 'host' and header['value'] == HTTP_REQUEST_HEADER_TO_CHECK:
                        ip_address = message['httpRequest']['clientIp']
                        timestamp = convert_to_iso_format(message['timestamp'])
                        ip_count[ip_address] += 1
                        break

            mark_event_as_processed(str(event_id), processed_events_table)

        if len(ip_count) == 0:
            logger.info(f"No IP addresses to process")
            return {
                'statusCode': 200,
                'body': json.dumps('No IP addresses to process')
            }

        # Batch update IP counts and check the deny list
        for ip, count in ip_count.items():
            update_ip_count(ip, timestamp, ip_count_table, count)

        check_and_add_to_denied_list(ip_count_table, denied_ip_table)

    except Exception as e:
        logger.error(f"Error processing event record: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps('Error processing logs')
        }

    return {
        'statusCode': 200,
        'body': json.dumps('IP addresses processed')
    }

Goal

Trying to understand why this doesn't work and how to properly set up my AWS Lambda for unit tests. Printing a statement is a sanity check tool.


Solution

  • As per the docs here, you should be able to use the logger library. As logging for INFO level is disabled by default you should change it. However, you should use setLevel of logger to achieve it and not basicConfig. The doc above has the basic example in it.