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".
Lambda
has the permission to write into CloudWatch
via the basic AWSLambdaBasicExecutionRole
policyI am using terms such as "file", function invocation loosely. Please critique freely.
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')
}
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.
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.