(benchmark): Add a script to download and parse logs

Author: alex9smith

.gitignore

@@ -0,0 +1 @@
+benchmark/data/*

.python-version

@@ -0,0 +1 @@
+3.13.2

README.md

@@ -16,6 +16,7 @@ Each lambda implementation has the same functionality:
 - NPM
 - ESBuild availble globally
 - OpenTofu (see `.opentofu-version`)
+- Python
 
 ## Deploying
 
@@ -37,3 +38,15 @@ cd infra
 tofu init
 tofu apply
 ```
+
+## Analysing the data
+
+Install the Python requirements and run the script to download the report data from Cloudwatch:
+
+```bash
+cd benchmark/scripts
+pip install -r requirements.txt
+python download_log_data.py
+```
+
+This will save a CSV of the results in `benchmark/data`

TODO.md

@@ -14,6 +14,7 @@
 
 ## Benchmark
 
+- Filter out reports from invocations that failed
 - Plan how to benchmark the lambdas
 - Where to put data for analysis
 - Initial analysis - see what's there

benchmark/scripts/download_log_data.py

@@ -0,0 +1,127 @@
+from typing import List, TypedDict, Optional
+from datetime import datetime, timedelta, timezone
+from time import sleep, time
+from csv import DictWriter
+import re
+import boto3
+
+LIVE_LAMBDAS = ["typescript"]
+
+
+class Row(TypedDict):
+    timestamp: str
+    message: str
+
+
+class InvocationSummary(TypedDict):
+    name: str
+    execution_time_ms: float
+    cold_start: bool
+    init_duration_ms: Optional[float]
+
+
+def submit_logs_query(client, lambda_name: str) -> str:
+    finish = datetime.now(tz=timezone.utc)
+    start = datetime.now(tz=timezone.utc) - timedelta(hours=1)
+
+    response = client.start_query(
+        logGroupName=f"/aws/lambda/{lambda_name}",
+        startTime=int(start.timestamp()),
+        endTime=int(finish.timestamp()),
+        queryString="fields @timestamp, @message",
+        limit=10000,
+    )
+
+    return response["queryId"]
+
+
+def poll_for_query_result(client, query_id: str):
+    results = client.get_query_results(queryId=query_id)
+    while results["status"] not in [
+        "Complete",
+        "Failed",
+        "Cancelled",
+        "Timeout",
+        "Unknown",
+    ]:
+        print(f"Query ID {query_id} not ready. Waiting")
+        sleep(10)
+        results = client.get_query_results(queryId=query_id)
+
+    return results.get("results", [])
+
+
+def parse_result_row(row) -> Row:
+    result = Row(timestamp="", message="")
+    for field in row:
+        if field["field"] == "@timestamp":
+            result["timestamp"] = field["value"]
+        elif field["field"] == "@message":
+            result["message"] = field["value"]
+
+    return result
+
+
+def process_report_rows(rows: List[Row], name: str) -> List[InvocationSummary]:
+    result = []
+    for row in rows:
+        if row["message"].startswith("REPORT"):
+            result.append(calc_report_summary(row["message"], name))
+
+    return result
+
+
+def calc_report_summary(message: str, name: str) -> InvocationSummary:
+    cold_start = False
+    init_duration_ms = None
+    if init_match := re.search(
+        r"Init Duration: (\d*\.?\d+) ms", message, re.IGNORECASE
+    ):
+        init_duration_ms = float(init_match.group(1))
+        cold_start = True
+
+    if exec_match := re.search(
+        r"Billed Duration: (\d*\.?\d+) ms", message, re.IGNORECASE
+    ):
+        return InvocationSummary(
+            name=name,
+            execution_time_ms=float(exec_match.group(1)),
+            cold_start=cold_start,
+            init_duration_ms=init_duration_ms,
+        )
+
+    print(f"Malformed message: {message}")
+    raise ValueError("Malformed message")
+
+
+def write_to_csv(data):
+    with open("../data/parsed_cloudwatch_logs.csv", "w") as f:
+        writer = DictWriter(
+            f,
+            fieldnames=["name", "execution_time_ms", "cold_start", "init_duration_ms"],
+        )
+        writer.writeheader()
+        writer.writerows(data)
+
+
+if __name__ == "__main__":
+    logs_client = boto3.client("logs")
+    print("Submitting queries")
+    queries = [
+        (submit_logs_query(logs_client, f"lambda_benchmark_{name}_lambda"), name)
+        for name in LIVE_LAMBDAS
+    ]
+    sleep(1)
+
+    raw_results = {}
+    for query_id, name in queries:
+        print(f"Querying results for {name}")
+        raw_results[name] = poll_for_query_result(logs_client, query_id)
+
+    parsed: List[InvocationSummary] = []
+    for name, results in raw_results.items():
+        rows = [parse_result_row(row) for row in results]
+        for report in process_report_rows(rows, name):
+            parsed.append(report)
+
+    write_to_csv(parsed)

benchmark/scripts/requirements.txt

@@ -0,0 +1,7 @@
+boto3==1.38.6
+botocore==1.38.6
+jmespath==1.0.1
+python-dateutil==2.9.0.post0
+s3transfer==0.12.0
+six==1.17.0
+urllib3==2.4.0

infra/dynamodb.tf

@@ -2,6 +2,7 @@ resource "aws_dynamodb_table" "table" {
   name         = var.dynamodb_table_name
   billing_mode = "PAY_PER_REQUEST"
   hash_key     = "eventId"
+  tags         = var.default_tags
 
   attribute {
     name = "eventId"