Fetch AWS load balancer recent logs for today to STDOUT

fetch_lb_logs.py

#!/usr/bin/env python3

import os
import sys
from time import strftime

import gzip
import io

import boto3.session

COUNT = 10  # Number of recent logs to fetch
BUCKET_NAME = os.environ["BUCKET_NAME"]  # Set this to the bucket used for LB logs
PREFIX = os.environ["PREFIX"]  # Set this to the prefix set in the load balancing log attribute

DATE_PART = strftime("%Y/%m/%d")

session = boto3.session.Session()
region_name = session.region_name
account = session.client("sts").get_caller_identity()["Account"]
s3 = session.client('s3')

today_prefix = f"{PREFIX}/AWSLogs/{account}/elasticloadbalancing/{region_name}/{DATE_PART}"


def fetch_recent_logs(bucket_name, prefix, count=COUNT):
    print(f"Fetching from {bucket_name}/{prefix}", file=sys.stderr)
    paginator = s3.get_paginator("list_objects_v2")
    contents = []
    iter = paginator.paginate(Bucket=bucket_name, Prefix=prefix)
    for page in iter:
        contents.extend(page["Contents"])

    if len(contents) == 0:
        print("No objects found", file=sys.stderr)

    objects = sorted(contents, key=lambda obj: obj['LastModified'], reverse=True)
    recent_objects = objects[:count]

    for obj in reversed(recent_objects):
        response = s3.get_object(Bucket=bucket_name, Key=obj['Key'])
        with gzip.GzipFile(fileobj=io.BytesIO(response['Body'].read())) as gz:
            for line in gz:
                print(line.decode('utf-8').strip())


fetch_recent_logs(BUCKET_NAME, today_prefix)