Issue
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 intoCloudWatch
via the basicAWSLambdaBasicExecutionRole
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.
Answered By - Bhavesh Parvatkar
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.