EMFを利用してALBのログからちょっと細かいカスタムメトリクスを生成する

import json
import logging
import boto3
import re

import urllib.parse
import gzip
import io

import time
import datetime
from urllib.parse import urlparse
import os

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3_client = boto3.client('s3')

ALB_LOG_PATTERN = re.compile(
    r'([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*)[:-]([0-9]*) ([-.0-9]*) ([-.0-9]*) ([-.0-9]*) (|[-0-9]*) (-|[-0-9]*) ([-0-9]*) ([-0-9]*) \"([^ ]*) (.*) (- |[^ ]*)\" \"([^\"]*)\" ([A-Z0-9-_]+) ([A-Za-z0-9.-]*) ([^ ]*) \"([^\"]*)\" \"([^\"]*)\" \"([^\"]*)\" ([-.0-9]*) ([^ ]*) \"([^\"]*)\" \"([^\"]*)\" \"([^ ]*)\" \"([^\\s]+?)\" \"([^\\s]+)\" \"([^ ]*)\" \"([^ ]*)\" ?([^ ]*)? ?( .*)?'
)

API_DEFINITIONS = [
    # api.example.com patterns
    # Revise it as per the requirements.
    ("GET",  r"^/api/customer/channel/[^/]+/customer-channel/[^/]+$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/summary$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/summary"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/coupon/[^/]+/_use$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/coupon/{coupon_code}/_use"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/coupon/[^/]+/_list$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/coupon/{coupon_code}/_list"),
    ("GET",  r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/coupon/[^/]+$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/coupon/{coupon_code}"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/point/_use$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/point/_use"),
    ("POST", r"^/api/customer$", "/api/customer"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/_enable-registration-completed$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/_enable-registration-completed"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/point/_add$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/point/_add"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/coupon-config/_list$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/coupon-config/_list"),
    ("POST", r"^/api/customer/channel/[^/]+/customer-channel/[^/]+/coupon-config/[^/]+/_exchange$", "/api/customer/channel/{channel_code}/customer-channel/{customer_channel_code}/coupon-config/{coupon_config_code}/_exchange"),
]

COMPILED_API_DEFINITIONS = [
    (method, re.compile(pattern), normalized) for method, pattern, normalized in API_DEFINITIONS
]

def lambda_handler(event, context):
    for record in event['Records']:
        try:
            s3_event = json.loads(record['body'])            
            if 'Records' not in s3_event:
                logger.info("Not an S3 event record, skipping.")
                continue
            for s3_record in s3_event['Records']:
                process_s3_record(s3_record)
        except Exception as e:
            logger.error(f"Error processing SQS record: {e}")
            raise e
    return {'statusCode': 200, 'body': json.dumps('Processing complete')}

def process_s3_record(s3_record):
    bucket = s3_record['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(s3_record['s3']['object']['key'], encoding='utf-8')
    logger.info(f"Processing file: s3://{bucket}/{key}")
    try:
        response = s3_client.get_object(Bucket=bucket, Key=key)
        content = response['Body'].read()
        with gzip.GzipFile(fileobj=io.BytesIO(content)) as gz:
            log_content = gz.read().decode('utf-8')
        for line in log_content.splitlines():
            parse_and_log_metrics(line)
    except Exception as e:
        logger.error(f"Failed to process file {key} from bucket {bucket}. Error: {e}")
        raise e

def identify_normalized_uri(http_method, path):
    for method, regex, normalized in COMPILED_API_DEFINITIONS:
        if http_method == method and regex.match(path):
            return normalized
    return None

def parse_and_log_metrics(log_line):
    match = ALB_LOG_PATTERN.match(log_line)
    if not match:
        return

    # timestamp
    dt = datetime.datetime.fromisoformat(match.group(2).replace('Z', '+00:00'))

    # ELB status code
    elb_status_code = match.group(11)

    # HTTP method
    http_method = match.group(15)

    # uri
    parsed = urlparse(match.group(16))
    normalized_uri = identify_normalized_uri(http_method, parsed.path)
    if normalized_uri is None:
        return None

    # target processing time
    target_processing_time = float(match.group(9))

    ### This value is set to -1 if the load balancer can't dispatch the request to a target. This can happen if the target closes the connection before the idle timeout or if the client sends a malformed request.
    ### This value can also be set to -1 if the registered target does not respond before the idle timeout.
    ### https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-access-logs.html#access-log-entry-format
    if target_processing_time == -1:
        target_processing_time = 0.0

    emf_payload = {
        "_aws": {
            "Timestamp": int(dt.timestamp() * 1000),
            "CloudWatchMetrics": [
                {
                    "Namespace": "alb-emf-test",
                    "Dimensions": [["ElbStatusCode", "HttpMethod", "Uri"]],
                    "Metrics": [
                        {"Name": "TargetProcessingTime", "Unit": "Seconds", "StorageResolution": 60},
                        {"Name": "RequestCount", "Unit": "Count", "StorageResolution": 60},
                    ]
                }
            ]
        },
        "ElbStatusCode": elb_status_code,
        "HttpMethod": http_method,
        "Uri": normalized_uri,
        "TargetProcessingTime": target_processing_time,
        "RequestCount": 1
    }
    print(json.dumps(emf_payload))