AWS Lambda for parsing ALB logs stored in S3 after CloudWatch Alert triggered

# AWS clients
s3 = boto3.client('s3')
logs = boto3.client('logs')
sqs = boto3.client('sqs')
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Config
BUCKET = 'BUCKET_WITH_ALB_LOGS'
BASE_PREFIX = 'BUCKET_WITH_ALB_LOGS/AWSLogs/AWS_ACCOUND_ID/elasticloadbalancing/REGION/'
LOG_GROUP = 'CLOUDWATCH_LOG_GROUP_NAME'
MAX_FILES_TO_PROCESS = 6
SQS_QUEUE_URL = 'SQS_URL'
RETRY_DELAY_SECONDS = 60 # Time where this lambda will invoke again

def ensure_log_group_and_stream_exists(log_group, log_stream):
"""Ensure the log group and log stream exist, creating them if necessary."""
try:
response = logs.describe_log_groups(logGroupNamePrefix=log_group)
log_groups = [lg for lg in response['logGroups'] if lg['logGroupName'] == log_group]
if not log_groups:
logger.info(f"Log group {log_group} not found, creating it...")
logs.create_log_group(logGroupName=log_group)
logger.info(f"Log group {log_group} created successfully")
else:
logger.info(f"Log group {log_group} already exists")
except Exception as e:
logger.error(f"Error checking/creating log group {log_group}: {str(e)}")
raise

try:
response = logs.describe_log_streams(
logGroupName=log_group,
logStreamNamePrefix=log_stream,
limit=1
)
streams = [s for s in response['logStreams'] if s['logStreamName'] == log_stream]
if not streams:
logger.info(f"Log stream {log_stream} not found, creating it...")
logs.create_log_stream(logGroupName=log_group, logStreamName=log_stream)
logger.info(f"Log stream {log_stream} created successfully")
else:
logger.info(f"Log stream {log_stream} already exists")
except Exception as e:
logger.error(f"Error checking/creating log stream {log_stream}: {str(e)}")
raise

def schedule_retry(event):
"""Schedule a retry by sending the event to SQS with a delay."""
logger.info(f"Scheduling retry after {RETRY_DELAY_SECONDS} seconds via SQS")
new_event = event.copy()
new_event['is_retry'] = True
try:
sqs.send_message(
QueueUrl=SQS_QUEUE_URL,
MessageBody=json.dumps(new_event),
DelaySeconds=RETRY_DELAY_SECONDS
)
logger.info("Successfully scheduled retry via SQS")
except Exception as e:
logger.error(f"Failed to schedule retry via SQS: {str(e)}")
raise

def lambda_handler(event, context):
logger.info(f"Event: {json.dumps(event)}")

# Handle SQS event structure
if 'Records' in event and event['Records']:
# Extract the body from the SQS message and parse it
sqs_body = event['Records'][0]['body']
try:
event = json.loads(sqs_body)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse SQS message body: {str(e)}")
return {'status': 'error', 'message': f"Invalid SQS message body: {str(e)}"}

# Now process the event (either direct or from SQS)
is_retry = event.get('is_retry', False)
logger.info(f"Is retry: {is_retry}")

alarm_data = event.get('alarmData', {})
state_data = alarm_data.get('state', {})
metric_data = alarm_data.get('configuration', {}).get('metrics', [{}])[0]

try:
timestamp_str = state_data.get('timestamp', '')
if not timestamp_str:
logger.error("No timestamp provided in alarm state data")
return {'status': 'error', 'message': 'Missing timestamp in alarm state data'}
alarm_time = datetime.strptime(timestamp_str.split('.')[0], '%Y-%m-%dT%H:%M:%S')
logger.info(f"Alarm triggered at: {alarm_time}")

metric_name = metric_data.get('metricStat', {}).get('metric', {}).get('name', '')
if '4XX' in metric_name:
error_pattern = '4'
elif '5XX' in metric_name:
error_pattern = '5'
elif '3XX' in metric_name:
error_pattern = '3'
else:
logger.error(f"Unknown metric name: {metric_name}")
return {'status': 'error', 'message': 'Unknown metric name'}

logger.info(f"Filtering for status codes starting with: {error_pattern}")
log_stream = metric_name
logger.info(f"Using log stream name: {log_stream}")

date_prefix = alarm_time.strftime('%Y/%m/%d/')
s3_prefix = f"{BASE_PREFIX}{date_prefix}"
logger.info(f"S3 prefix: {s3_prefix}")

s3_objects = []
continuation_token = None
while True:
list_kwargs = {'Bucket': BUCKET, 'Prefix': s3_prefix}
if continuation_token:
list_kwargs['ContinuationToken'] = continuation_token
response = s3.list_objects_v2(**list_kwargs)
s3_objects.extend(response.get('Contents', []))
if not response.get('IsTruncated', False):
break
continuation_token = response.get('NextContinuationToken')

if not s3_objects:
logger.warning(f"No log files found under prefix {s3_prefix}")
return {'status': 'success', 'processed': 0}

log_files = []
for obj in s3_objects:
key = obj['Key']
try:
file_time_str = key.split('_')[-3]
file_time = datetime.strptime(file_time_str, '%Y%m%dT%H%MZ')
log_files.append({'key': key, 'file_time': file_time})
except Exception as e:
logger.error(f"Error parsing timestamp for {key}: {str(e)}")
continue

log_files.sort(key=lambda x: x['file_time'], reverse=True)
logger.info(f"Found {len(log_files)} log files")