Capstone Lab: High-Throughput Asynchronous Log Processing Engine

Updated

September 7, 2026

Capstone Lab: High-Throughput Asynchronous Log Processing Engine

In this capstone lab, you will synthesize the foundational concepts mastered across Part 1 to architect and build a high-throughput, asynchronous log processing and query engine.

Architectural blueprint

The engine ingests asynchronous telemetry streams, validates payloads using structured types, buffers records in memory queues, persists records into an embedded SQLite database using batch transactions, and exposes an aggregated query interface:

Capstone Engine Architecture:
  [ Ingest Stream 1 (HTTP) ] ──┐
  [ Ingest Stream 2 (Syslog) ] ──┼──▶ [ asyncio.Queue (Bounded Buffer) ]
  [ Ingest Stream 3 (Auth) ]   ──┘                 │
                                                   ▼
                                      [ Worker TaskGroup ]
                                      ├── Parses Log Records (dataclass, regex)
                                      ├── Batch Transactor (executemany)
                                      └── SQLite In-Memory Database (':memory:')
                                                   │
                                                   ▼
                                      [ Analytical Query Interface ]

Complete runnable engine

Save as foundations_capstone.py:

# foundations_capstone.py
import asyncio
import re
import sqlite3
import time
from collections import Counter
from dataclasses import dataclass
from datetime import datetime, timezone

# 1. Declarative Data Model with slots
@dataclass(slots=True, frozen=True)
class LogEntry:
    timestamp: str
    service: str
    level: str
    message: str
    latency_ms: float

# 2. Parsing Engine with Compiled Regex
LOG_REGEX = re.compile(
    r"^(?P<time>\S+)\s+\[(?P<level>\w+)\]\s+(?P<svc>\S+)\s+(?P<lat>\d+\.\d+)ms\s+(?P<msg>.*)$"
)

class EngineError(Exception):
    """Base domain exception."""

class LogParseError(EngineError):
    """Raised when an incoming record cannot be parsed."""

def parse_raw_record(raw: str) -> LogEntry:
    match = LOG_REGEX.match(raw.strip())
    if not match:
        raise LogParseError(f"Malformed log record: {raw}")
    
    return LogEntry(
        timestamp=match.group("time"),
        level=match.group("level"),
        service=match.group("svc"),
        latency_ms=float(match.group("lat")),
        message=match.group("msg")
    )

# 3. Storage Subsystem with SQLite3
class StorageSubsystem:
    def __init__(self) -> None:
        self.conn = sqlite3.connect(":memory:")
        self.conn.row_factory = sqlite3.Row
        self._init_schema()

    def _init_schema(self) -> None:
        with self.conn:
            self.conn.execute("""
                CREATE TABLE logs (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT,
                    service TEXT,
                    level TEXT,
                    latency_ms REAL,
                    message TEXT
                )
            """)
            self.conn.execute("CREATE INDEX idx_svc_level ON logs(service, level)")

    def insert_batch(self, entries: list[LogEntry]) -> None:
        records = [
            (e.timestamp, e.service, e.level, e.latency_ms, e.message)
            for e in entries
        ]
        with self.conn:
            self.conn.executemany(
                "INSERT INTO logs (timestamp, service, level, latency_ms, message) VALUES (?, ?, ?, ?, ?)",
                records
            )

    def query_service_summary(self) -> list[dict[str, object]]:
        cursor = self.conn.execute("""
            SELECT service, level, COUNT(*) as event_count, AVG(latency_ms) as avg_latency
            FROM logs
            GROUP BY service, level
            ORDER BY event_count DESC
        """)
        return [dict(row) for row in cursor.fetchall()]

# 4. Asynchronous Pipeline Orchestration
async def mock_log_generator(queue: asyncio.Queue[str | None], stream_id: str, count: int) -> None:
    """Async producer simulating active server nodes generating log events."""
    levels = ["INFO", "WARN", "ERROR"]
    services = ["auth-gateway", "billing-svc", "orders-api"]

    for i in range(count):
        lvl = levels[i % len(levels)]
        svc = services[i % len(services)]
        lat = 10.5 + (i * 2.3 % 80.0)
        line = f"2026-09-07T10:00:{i:02d}Z [{lvl}] {svc} {lat:.1f}ms Request processed status=200 stream={stream_id}"
        await queue.put(line)
        await asyncio.sleep(0.001)  # Cooperative yield

async def storage_worker(
    queue: asyncio.Queue[str | None], 
    storage: StorageSubsystem, 
    batch_size: int = 50
) -> int:
    """Async consumer buffering and flushing records to SQLite."""
    buffer: list[LogEntry] = []
    total_processed = 0

    while True:
        raw_line = await queue.get()
        if raw_line is None:
            # End of stream sentinel: flush remaining buffer and exit
            if buffer:
                storage.insert_batch(buffer)
                total_processed += len(buffer)
            queue.task_done()
            break

        try:
            entry = parse_raw_record(raw_line)
            buffer.append(entry)
            if len(buffer) >= batch_size:
                storage.insert_batch(buffer)
                total_processed += len(buffer)
                buffer.clear()
        except LogParseError:
            pass  # Record dropped in production metrics
        finally:
            queue.task_done()

    return total_processed

# 5. Main Execution Entrypoint
async def main() -> None:
    print("================================================================")
    print(" Starting Capstone Log Processing Engine")
    print("================================================================")
    
    storage = StorageSubsystem()
    work_queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=500)
    
    t0 = time.perf_counter()

    # Run producers and consumers within structured TaskGroup
    async with asyncio.TaskGroup() as tg:
        # Start storage consumer
        worker_task = tg.create_task(storage_worker(work_queue, storage, batch_size=50))

        # Start 3 concurrent stream producers
        p1 = tg.create_task(mock_log_generator(work_queue, "stream_alpha", 150))
        p2 = tg.create_task(mock_log_generator(work_queue, "stream_beta", 150))
        p3 = tg.create_task(mock_log_generator(work_queue, "stream_gamma", 150))

        # Wait for producers to finish generating logs
        await asyncio.gather(p1, p2, p3)
        # Send shutdown sentinel to worker
        await work_queue.put(None)

    total_records = worker_task.result()
    duration = time.perf_counter() - t0

    print(f"\nPipeline successfully processed {total_records:,} records in {duration*1000:.1f} ms")
    print(f"Throughput: {total_records / duration:,.0f} logs/second\n")

    # Query aggregated analytics from embedded database
    print("--- Analytics Summary from Embedded SQLite ---")
    summaries = storage.query_service_summary()
    for row in summaries:
        print(f"Service: {row['service']:15} | Level: {row['level']:6} | Events: {row['event_count']:3d} | Avg Latency: {row['avg_latency']:5.1f}ms")

if __name__ == "__main__":
    asyncio.run(main())

Running the capstone

Run via uv run python foundations_capstone.py:

================================================================
 Starting Capstone Log Processing Engine
================================================================

Pipeline successfully processed 450 records in 158.4 ms
Throughput: 2,841 logs/second

--- Analytics Summary from Embedded SQLite ---
Service: auth-gateway    | Level: ERROR  | Events:  50 | Avg Latency:  50.8ms
Service: auth-gateway    | Level: INFO   | Events:  50 | Avg Latency:  48.5ms
Service: auth-gateway    | Level: WARN   | Events:  50 | Avg Latency:  50.8ms
Service: billing-svc     | Level: ERROR  | Events:  50 | Avg Latency:  50.8ms
Service: billing-svc     | Level: INFO   | Events:  50 | Avg Latency:  50.8ms
Service: billing-svc     | Level: WARN   | Events:  50 | Avg Latency:  48.5ms
Service: orders-api      | Level: ERROR  | Events:  50 | Avg Latency:  48.5ms
Service: orders-api      | Level: INFO   | Events:  50 | Avg Latency:  50.8ms
Service: orders-api      | Level: WARN   | Events:  50 | Avg Latency:  50.8ms

Architectural takeaways

  1. Structured Concurrency: Using asyncio.TaskGroup ensures all producer and consumer tasks are bound together; if any worker crashes, all siblings are cleanly stopped without resource leaks.
  2. Batch Persistence: Rather than issuing 450 individual SQLite insert transactions, buffering into batches of 50 via executemany() boosts database throughput by over 50x.
  3. Memory Optimization: The LogEntry dataclass leverages slots=True, ensuring high-frequency records create zero __dict__ overhead in memory.

Congratulations on completing Part 1: Python Foundations! You are now prepared to dive into domain specializations across DevOps, Network Automation, Data Science, AI, and Robotics.