Async Log Batching for SOC Pipeline Throughput

Asynchronous log batching is the buffering layer that lets a Security Operations Center survive bursty telemetry without dropping the events its detection logic depends on. It sits inside the broader Log Ingestion & Parsing Workflows pipeline, between raw collection and downstream parsing, and replaces synchronous per-event forwarding with configurable time- or size-bound aggregation. By decoupling collection from dispatch, async batching absorbs network and parser latency, collapses thousands of tiny requests into a handful of efficient payloads, and gives the pipeline a deterministic memory profile under load. Done correctly it stabilizes correlation latency and protects alert fidelity; done naively — with an unbounded queue and no backpressure — it becomes the exact component that OOM-kills the collector during the incident you most need it for.

Problem Framing

Consider a forwarder collecting firewall and authentication telemetry from 400 hosts. At steady state it sees roughly 3,000 events/second, well within the SIEM’s ingestion quota. During a credential-stuffing wave the authentication subsystem emits a burst — 80,000 events in under a minute from a handful of sources. A synchronous collector that POSTs each event individually now opens thousands of concurrent sockets, the SIEM endpoint starts returning HTTP 429, blocked sends pile up, and resident memory climbs until the kernel OOM-killer reaps the process. The events generated during the kill window — the ones describing the attack — are lost, and the correlation rule that should have fired on the brute-force sequence never sees a complete event stream.

Async batching resolves this by changing three things at once: it makes I/O non-blocking so the collector never stalls waiting on a slow endpoint, it bounds memory with an explicit queue ceiling and backpressure, and it amortizes egress cost by flushing aggregated payloads on a time-or-size trigger. The unit of work shifts from “one event, one request” to “one batch, one request,” which is what keeps both the host and the SIEM endpoint inside their operating envelopes. This page builds that batcher in Python and wires it into the validation, error-handling, and rate-limiting controls the rest of the pipeline expects.

Prerequisites & Environment

The reference implementation targets Python 3.11+ and uses only the standard library for the core batcher (asyncio, dataclasses, json, logging, time, uuid). Production deployments add two third-party libraries for real network egress and strict validation:

pip install "httpx>=0.27"      # async HTTP client for SIEM/HEC dispatch
pip install "pydantic>=2.7"    # strict schema validation at the batch boundary

Infrastructure assumptions:

• A downstream sink that accepts batched payloads — Splunk HEC, Elastic _bulk, an OTLP collector, or a Kafka topic. Each enforces a maximum request size (commonly 1–10 MB) and a per-token ingest rate; align your batch ceiling to the smaller of the two.
• A reachable dead-letter sink (an on-disk spool, a separate Kafka topic, or an S3 prefix) for events that fail validation or exhaust retries.
• Newline-delimited JSON (NDJSON) as the wire format, since both Splunk HEC and the Elastic bulk API consume it directly and it streams without holding the whole batch in a single JSON array.

Throughout, treat the batcher as a long-lived asyncio task that shares an event loop with your collectors — the same loop model used by the deeper async log collectors built with asyncio on the ingestion edge.

Architecture Overview

The batcher is a single coroutine fed by a bounded asyncio.Queue. Producers (collectors) enqueue LogEvent objects with non-blocking put_nowait; a background flush loop drains the queue into an in-memory buffer and dispatches whenever the size or time threshold trips. Validation runs at the flush boundary so malformed events never reach the sink, a token bucket throttles egress, and any event that fails validation or dispatch is routed to a dead-letter path.

The design has four invariants worth stating explicitly, because every failure mode below is a violation of one of them:

1. The queue is always bounded. Memory is capped by maxsize; there is no path that lets the buffer grow without limit.
2. Producers never block the event loop. Enqueue is non-blocking and signals backpressure by returning False rather than awaiting indefinitely.
3. A flush is atomic with respect to the buffer. The buffer is dispatched and cleared together, so a crash mid-dispatch cannot double-count or silently drop accounted events.
4. Nothing leaves un-validated, and nothing failed disappears. Every event either reaches the sink, the retry path, or the dead-letter sink — never the void.

Step-by-Step Implementation

Step 1 — Model the event and the metrics

Strongly typed events make the validation and serialization boundaries explicit. A small metrics record gives observability for free when the batcher stops.

import asyncio
import json
import logging
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any, Dict, List, Optional

@dataclass
class LogEvent:
    event_id: str
    timestamp: float
    source: str
    payload: Dict[str, Any]
    severity: str = "INFO"

@dataclass
class BatchMetrics:
    total_flushed: int = 0
    validation_errors: int = 0
    permanent_failures: int = 0
    transient_retries: int = 0
    dropped_backpressure: int = 0

Step 2 — Configure structured logging

SOC pipelines need machine-parseable operational logs so the batcher’s own behavior is queryable in the SIEM. Emit JSON, not free text.

class StructuredJSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "timestamp": self.formatTime(record, self.datefmt),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "function": record.funcName,
            "line": record.lineno,
        }
        if hasattr(record, "context"):
            entry.update(record.context)  # type: ignore[attr-defined]
        return json.dumps(entry)

logger = logging.getLogger("soc_async_batcher")
logger.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(StructuredJSONFormatter())
logger.addHandler(_handler)

Step 3 — Bound the queue and enqueue with backpressure

The queue’s maxsize is the single most important reliability parameter on the page. When it is full, the producer is told False immediately so it can decide policy — shed low-priority debug telemetry, throttle the source, or spill to disk — rather than blocking the loop or growing memory.

class AsyncLogBatcher:
    def __init__(
        self,
        max_queue_size: int = 5000,
        max_batch_size: int = 1000,
        flush_interval: float = 2.0,
        rate_limit_per_sec: float = 10.0,
    ) -> None:
        self.queue: asyncio.Queue[LogEvent] = asyncio.Queue(maxsize=max_queue_size)
        self.max_batch_size = max_batch_size
        self.flush_interval = flush_interval
        self.metrics = BatchMetrics()
        self._running = False
        self._flush_task: Optional[asyncio.Task] = None
        # Token bucket for egress shaping.
        self._tokens = float(rate_limit_per_sec)
        self._max_tokens = float(rate_limit_per_sec)
        self._last_refill = time.monotonic()
        self._rate_lock = asyncio.Lock()

    async def enqueue(self, event: LogEvent) -> bool:
        """Non-blocking enqueue. Returns False when the queue is saturated."""
        try:
            self.queue.put_nowait(event)
            return True
        except asyncio.QueueFull:
            self.metrics.dropped_backpressure += 1
            logger.warning(
                "Queue saturated; applying backpressure.",
                extra={"context": {"queue_size": self.queue.qsize(),
                                   "event_id": event.event_id}},
            )
            return False

Step 4 — Implement the token bucket for egress shaping

The token bucket caps how often batches leave the process so a flush storm cannot trip the sink’s ingest quota. Refilling is computed from elapsed monotonic time, so it is robust to wall-clock jumps and needs no background timer. This is the same primitive covered in depth under rate limiting strategies, applied here per batcher instance.

    async def _acquire_token(self) -> bool:
        async with self._rate_lock:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(self._max_tokens,
                               self._tokens + elapsed * self._max_tokens)
            self._last_refill = now
            if self._tokens >= 1.0:
                self._tokens -= 1.0
                return True
            return False

Step 5 — Validate, dispatch, and categorize failures

Each flush validates events, serializes survivors to NDJSON, and dispatches under a rate-limit token. Failures are sorted into transient (retryable) and permanent (dead-lettered) before any logging decision is made.

    def _validate_schema(self, event: LogEvent) -> bool:
        required = ("source", "timestamp", "payload")
        return all(getattr(event, f, None) is not None for f in required)

    @staticmethod
    def _categorize_error(error: Exception) -> str:
        if isinstance(error, (json.JSONDecodeError, TypeError, AttributeError)):
            return "PERMANENT"
        if isinstance(error, (ConnectionError, TimeoutError, asyncio.TimeoutError)):
            return "TRANSIENT"
        return "UNKNOWN"

    async def _dispatch_batch(self, batch: List[LogEvent]) -> List[LogEvent]:
        """Dispatch a batch. Returns events that must be retried on the next flush."""
        if not await self._acquire_token():
            logger.info("Rate limit reached; deferring batch.",
                        extra={"context": {"batch_size": len(batch)}})
            return batch  # defer the whole batch; no token available

        valid: List[LogEvent] = []
        for event in batch:
            if self._validate_schema(event):
                valid.append(event)
            else:
                self.metrics.validation_errors += 1
                logger.error(
                    "Schema validation failed; routing to DLQ.",
                    extra={"context": {"event_id": event.event_id,
                                       "source": event.source,
                                       "error_code": "ERR_VALIDATION_001"}},
                )
        if not valid:
            return []

        payload = "\n".join(json.dumps(asdict(e)) for e in valid)
        try:
            # Production: await client.post(sink_url, content=payload, ...)
            await asyncio.sleep(0.01)  # placeholder for async network I/O
            self.metrics.total_flushed += len(valid)
            logger.info(
                "Batch dispatched.",
                extra={"context": {"batch_size": len(valid),
                                   "payload_bytes": len(payload),
                                   "total_flushed": self.metrics.total_flushed}},
            )
            return []
        except Exception as exc:  # noqa: BLE001 - categorized below
            if self._categorize_error(exc) == "TRANSIENT":
                self.metrics.transient_retries += 1
                logger.warning("Transient dispatch failure; will retry.",
                               extra={"context": {"error": str(exc),
                                                  "error_code": "ERR_DISPATCH_002"}})
                return valid  # re-enqueue logically for next flush
            self.metrics.permanent_failures += len(valid)
            logger.error("Permanent dispatch failure; dead-lettering batch.",
                         extra={"context": {"error": str(exc),
                                            "error_code": "ERR_DISPATCH_003"}})
            return []

Step 6 — Run the flush loop on size or time

The loop waits briefly for the next event so an idle pipeline still flushes on the time trigger, then dispatches whenever the buffer reaches max_batch_size or flush_interval elapses. Deferred or transiently failed events are carried into the next buffer rather than dropped.

    async def _flush_loop(self) -> None:
        buffer: List[LogEvent] = []
        last_flush = time.monotonic()
        while self._running:
            try:
                event = await asyncio.wait_for(self.queue.get(), timeout=0.5)
                buffer.append(event)
            except asyncio.TimeoutError:
                pass

            elapsed = time.monotonic() - last_flush
            if buffer and (len(buffer) >= self.max_batch_size
                           or elapsed >= self.flush_interval):
                buffer = await self._dispatch_batch(buffer)
                last_flush = time.monotonic()

    async def start(self) -> None:
        if self._running:
            return
        self._running = True
        self._flush_task = asyncio.create_task(self._flush_loop())
        logger.info("AsyncLogBatcher started.",
                    extra={"context": {"max_queue": self.queue.maxsize}})

    async def stop(self) -> None:
        self._running = False
        if self._flush_task:
            self._flush_task.cancel()
            try:
                await self._flush_task
            except asyncio.CancelledError:
                pass
        remaining: List[LogEvent] = []
        while not self.queue.empty():
            remaining.append(self.queue.get_nowait())
        if remaining:
            await self._dispatch_batch(remaining)
        logger.info("AsyncLogBatcher drained and stopped.",
                    extra={"context": asdict(self.metrics)})

Step 7 — Drive it end to end

A short driver exercises backpressure (a small queue), the size trigger (max_batch_size=5), and graceful drain on shutdown.

async def main() -> None:
    batcher = AsyncLogBatcher(max_queue_size=2000, max_batch_size=5,
                              flush_interval=1.0, rate_limit_per_sec=5.0)
    await batcher.start()
    for i in range(15):
        event = LogEvent(
            event_id=str(uuid.uuid4()),
            timestamp=time.time(),
            source="firewall-01",
            payload={"src_ip": f"192.168.1.{i}", "dst_port": 443, "action": "ALLOW"},
        )
        if not await batcher.enqueue(event):
            logger.warning("Event shed under backpressure.",
                           extra={"context": {"event_id": event.event_id}})
        await asyncio.sleep(0.1)
    await asyncio.sleep(2.0)   # let the loop drain on the time trigger
    await batcher.stop()

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

Schema & Validation Integration

The _validate_schema stub above keeps the batcher self-contained, but in production the flush boundary is where the pipeline’s schema validation pipeline enforces structure before events touch the sink. Replace the stub with a Pydantic model whose fields map to the site’s Elastic Common Schema (ECS) normalization target — @timestamp, event.action, source.ip, event.severity — so that what leaves the batcher is already index-ready:

from pydantic import BaseModel, IPvAnyAddress, field_validator

class EcsEvent(BaseModel):
    timestamp: float
    source_ip: IPvAnyAddress           # ECS source.ip
    event_action: str                  # ECS event.action
    severity: str = "INFO"             # ECS event.severity (keyword)

    @field_validator("severity")
    @classmethod
    def _known_severity(cls, v: str) -> str:
        allowed = {"INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL"}
        if v.upper() not in allowed:
            raise ValueError(f"unknown severity: {v}")
        return v.upper()

Validating at the batch boundary, rather than per event at ingest, is deliberate: it lets the CPU-bound validation work amortize across the batch and keeps the hot enqueue path allocation-light. Events that parse but fail field constraints are the ones most likely to be early signals of schema drift from a vendor — feed those into the error categorization framework so the drift is classified and surfaced rather than silently quarantined.

Error Handling & DLQ Routing

Every event in the batcher has exactly three terminal destinations: the sink, the retry buffer, or the dead-letter queue (DLQ). The _categorize_error split decides which: transient faults (connection resets, timeouts) are returned from _dispatch_batch and re-buffered for the next flush with exponential backoff; permanent faults (decode errors, schema violations) are dead-lettered with a stable error code so analysts can triage them in bulk. Codes follow the established ERR_CATEGORY_NNN convention:

The non-negotiable rule for SOC pipelines is that backpressure must degrade gracefully, never catastrophically. When the queue saturates, shedding clearly-labeled low-priority telemetry (ERR_BACKPRESSURE_004) while preserving security-relevant streams is correct behavior; an unbounded queue that defers the decision until the OOM-killer makes it is not. Persist DLQ records with their original payload and error code so a corrected schema or a recovered sink lets you replay them without data loss.

Performance Tuning

Three knobs govern the throughput/latency/memory triangle, and they must be tuned against the downstream sink rather than in isolation:

• max_batch_size / flush_interval set the latency floor and the request efficiency. Larger batches and longer intervals improve compression ratio and reduce request overhead but add end-to-end correlation latency. For interactive detection, keep flush_interval at 1–2 s; for archival lanes, 10–30 s batches are fine. Whichever fires first wins, so size acts as a ceiling during spikes and time guarantees liveness when idle.
• max_queue_size is your memory ceiling. Worst-case resident memory for the buffer is roughly max_queue_size × avg_event_bytes; pick it from the host’s headroom, not a round number. A 5,000-event queue of ~1 KB events costs ~5 MB — cheap insurance against a spike.
• rate_limit_per_sec must sit at or below the sink’s documented ingest quota. Set it too high and you trade local stability for ERR_DISPATCH_002 retries; too low and the queue backs up into avoidable backpressure.

For genuinely high-volume lanes, run several batcher instances partitioned by log source on the same event loop — the single-threaded asyncio model multiplexes their I/O without lock contention, and per-source instances give each stream its own quota and backpressure policy. Keep serialization allocation-light: NDJSON with a generator expression (as in _dispatch_batch) avoids materializing a second large list, and you can layer gzip on the payload when the sink accepts compressed bodies. CSV-heavy sources benefit from normalizing to dicts upstream first, per the CSV ingestion patterns guidance, so the batcher only ever serializes uniform records.

Verification & Observability

Confirm correct operation along three axes — conservation, boundedness, and shaping:

• Event conservation. After a run, total_flushed + validation_errors + permanent_failures + dropped_backpressure must equal the number of events enqueued plus the number shed. A mismatch means an event leaked out of the three-destinations invariant. Assert it directly in a test:

async def test_no_event_is_lost():
    b = AsyncLogBatcher(max_queue_size=1000, max_batch_size=10,
                        flush_interval=0.2, rate_limit_per_sec=1000)
    await b.start()
    enqueued = shed = 0
    for i in range(500):
        ok = await b.enqueue(LogEvent(str(i), time.time(), "test",
                                      {"src_ip": "10.0.0.1"}))
        enqueued += 1
        shed += 0 if ok else 1
    await asyncio.sleep(0.5)
    await b.stop()
    m = b.metrics
    accounted = m.total_flushed + m.validation_errors + m.permanent_failures
    assert accounted + m.dropped_backpressure == enqueued
    assert m.dropped_backpressure == shed

• Boundedness. Sample queue.qsize() during a synthetic spike; it must plateau at max_queue_size, never climb past it. Resident memory should track the same plateau.
• Shaping. Count dispatch log lines per second — they must not exceed rate_limit_per_sec. The structured Batch dispatched. lines carry payload_bytes and total_flushed, so you can chart egress volume and cumulative throughput straight from the SIEM that consumes the batcher’s own logs.

Troubleshooting

• Memory climbs steadily and the process is OOM-killed. The queue is effectively unbounded — either maxsize was set to 0 (unlimited) or producers are awaiting queue.put() instead of using non-blocking put_nowait. Set a real max_queue_size and shed on QueueFull.
• Batches arrive late or never under low volume. The flush loop is blocking on queue.get() with no timeout, so the time trigger never evaluates while the queue is empty. Keep the asyncio.wait_for(..., timeout=0.5) guard so the loop wakes to check flush_interval.
• The sink returns HTTP 429 in bursts. Egress is unshaped or rate_limit_per_sec exceeds the sink quota. Lower the limit, confirm the token bucket is actually consulted before every dispatch, and enable payload compression to fit more events per permitted request.
• total_flushed is lower than expected and DLQ is filling with ERR_VALIDATION_001. A vendor changed its format — schema drift. Diff a dead-lettered payload against the ECS model, update the validator, and replay the DLQ.
• Events arrive out of order at the SIEM. Multiple batcher instances are writing to the same partition without a key. Partition by log source and rely on each event’s own timestamp for correlation ordering rather than arrival order.

FAQ

Related

• Log Ingestion & Parsing Workflows — parent architecture
• Building async log collectors with asyncio
• Rate Limiting Strategies
• Schema Validation Pipelines
• Error Categorization Frameworks