Hey folks — been tuning my agent event pipeline and wanted to share a pattern that’s cut my SIEM ingestion costs and improved reliability.

I kept seeing timeouts and high egress costs when my agents sent individual JSON events directly to the SIEM’s HTTP Event Collector (HEC). The solution? Batch, compress, and add a lightweight retry layer. This is especially useful for agent runtimes that generate verbose operational events (think heartbeats, policy evaluations, dependency scans).

Here’s the core Python script I run as a sidecar container. It collects events from a local socket or file, batches them for 10 seconds or 500KB, then posts gzipped JSON to Splunk (but adapts easily to Elastic or Chronicle).

```python
import gzip
import json
import time
import threading
from queue import Queue
from datetime import datetime
import requests

class SIEMBatcher:
def __init__(self, siem_url, hec_token, batch_max_size=500000, batch_max_interval=10):
self.siem_url = siem_url
self.headers = {'Authorization': f'Splunk {hec_token}', 'Content-Type': 'application/json'}
self.batch_max_size = batch_max_size
self.batch_max_interval = batch_max_interval
self.event_queue = Queue()
self.current_batch = []
self.current_batch_size = 0
self.lock = threading.Lock()

def add_event(self, event):
"""Add a single event dict to the batch."""
event_str = json.dumps(event)
with self.lock:
self.current_batch.append(event)
self.current_batch_size += len(event_str)
if self.current_batch_size >= self.batch_max_size:
self._flush()

def _flush(self):
"""Compress and send the current batch."""
if not self.current_batch:
return
batch_data = json.dumps(self.current_batch)
compressed = gzip.compress(batch_data.encode('utf-8'))

# Retry logic for transient failures
for _ in range(3):
try:
resp = requests.post(self.siem_url, headers=self.headers, data=compressed, timeout=30)
if resp.status_code == 200:
break
else:
time.sleep(2)
except requests.exceptions.RequestException:
time.sleep(2)

with self.lock:
self.current_batch = []
self.current_batch_size = 0

def start_periodic_flush(self):
"""Background thread to flush based on time interval."""
def flush_loop():
while True:
time.sleep(self.batch_max_interval)
self._flush()
thread = threading.Thread(target=flush_loop, daemon=True)
thread.start()
```

Key takeaways:
* **Reduced connections** — from thousands per hour to a few dozen.
* **Compression** — cuts payload size by ~70-80% for text-based logs.
* **Simple retry** — handles SIEM hiccups without losing the whole batch.
* **Thread-safe** — agents can submit events from multiple threads.

I’ve been running this with a Rust agent that emits security events (file integrity, process lineage). The batcher normalizes timestamps and adds an `agent_id` field before queuing.

Would love to hear:
* How you’re handling event batching for other SIEMs like Chronicle.
* If you’ve built similar sidecars for agent runtimes in production.
* Any pitfalls with clock skew or event ordering I might have missed.

--sasha