Building Async Log Collectors with asyncio — LogParsing

Building Async Log Collectors with asyncio

A synchronous SOC log collector looks healthy at steady state but silently drops events the moment a traffic spike makes its blocking I/O and unbounded buffers collide with the host’s memory ceiling. This page builds a non-blocking, memory-bounded collector with asyncio, as one technique inside async log batching and the wider Log Ingestion & Parsing Workflows pipeline.

Root-Cause Context

The failure is not bandwidth — it is the collector’s inability to yield control during an I/O wait. A synchronous forwarder built on blocking requests or raw socket calls, feeding an unbounded Python list, behaves predictably right up to the point where a downstream SIEM throttles or a network partition occurs. Then sends stop completing, the producer keeps appending, and resident memory climbs linearly with the pending payload until the kernel OOM-killer reaps the process. Every event generated during that kill window is lost.

In a SOC this surfaces as correlation gaps rather than an obvious crash. A credential-stuffing wave mapped to MITRE ATT&CK T1110 (Brute Force), or a lateral-movement sequence using T1078 (Valid Accounts), can emit 50,000 authentication events in 90 seconds from a handful of sources. If the collector only persists 32,000 of them, the rule that should have fired on the brute-force sequence never sees a complete stream, so the alert either suppresses entirely or escalates as a noisy false positive.

asyncio removes the thread-contention root cause by multiplexing every I/O-bound task onto a single event loop: a coroutine awaiting a slow endpoint suspends and lets others run instead of pinning a thread. Pairing that cooperative model with an explicit maxsize on every asyncio.Queue converts the unbounded-memory failure into deterministic backpressure — when the queue fills, the producer blocks (or sheds) instead of allocating without limit.

Memory-bounded async log collector data flow on a single asyncio event loop<br>Pull-based JSON log sources feed a semaphore-bounded aiohttp fetcher that enqueues records with put_nowait into a bounded raw asyncio.Queue; when that queue is full it applies backpressure and sheds a labelled ERR_QUEUE_001 record. A Pydantic validator drains the raw queue, dead-letters failures as ERR_SCHEMA_001 or ERR_SCHEMA_002, and pushes valid records into a second bounded queue. A token-bucket-gated batch dispatcher drains the valid queue and POSTs size-triggered batches to the SIEM sink, where an HTTP 429 emits ERR_DISPATCH_001. Every queue carries an explicit maxsize, so peak memory is bounded by construction.

Async collector — bounded data flow on one event loop<br>Every queue has an explicit maxsize, so peak memory is bounded by construction — bursts shed, they do not grow.

put_nowait<br>await put

full → backpressure<br>full → backpressure

queue full<br>invalid<br>HTTP 429

Log sources<br>pull · JSON<br>over HTTP

Async fetcher<br>aiohttp · Semaphore(8)<br>non-blocking I/O

raw Queue<br>maxsize 10k

Validator<br>Pydantic model<br>off the hot path

valid Queue<br>maxsize 10k

Batch dispatcher<br>token-bucket gated<br>batch_size 500

SIEM sink<br>Splunk HEC ·<br>Elastic _bulk

ERR_QUEUE_001<br>backpressure shed<br>+ shed metric

Dead-letter queue<br>ERR_SCHEMA_001/002<br>replay after fix

ERR_DISPATCH_001<br>SIEM 429 throttle<br>lower token rate

Prerequisites

Python 3.11+ for asyncio.TaskGroup and modern timeout semantics.

Two third-party libraries beyond the standard library:

Copypip install "aiohttp>=3.9" # non-blocking HTTP fetch + connection pooling<br>pip install "pydantic>=2.7" # strict, typed validation at the queue boundary

A pull-based log source that returns JSON over HTTP (cloud audit API, vendor webhook spool, or an internal collector endpoint), and a downstream sink that accepts batched POSTs (Splunk HEC, Elastic _bulk, or an OTLP gateway).

A reachable dead-letter sink — an on-disk spool, a Kafka topic, or an object-store prefix — for events that fail validation or exhaust retries.

Production-Ready Implementation

The collector is one self-contained module wiring four bounded stages onto a single event loop: a semaphore-limited fetcher, a Pydantic validator that routes failures to a dead-letter queue, a token-bucket egress limiter, and a size-triggered batch dispatcher. Every queue carries an explicit maxsize, so memory is bounded by construction rather than by hope.

Copyfrom __future__ import annotations

import asyncio<br>import logging<br>import time<br>from datetime import datetime<br>from typing import Any, Optional

import aiohttp<br>from aiohttp import ClientTimeout<br>from pydantic import BaseModel, Field, ValidationError

logger = logging.getLogger("soc.async_collector")

# --- Typed event model ------------------------------------------------------

class SOCLogRecord(BaseModel):<br>timestamp: datetime<br>source_ip: str<br>event_type: str<br>severity: int = Field(ge=0, le=10)<br>raw_payload: Optional[dict[str, Any]] = None

# --- Token-bucket egress...