Pipeline Overview

The Consumer Pipeline is the core processing engine — a single-process batch consumer that transforms raw Kafka messages into enriched, deduplicated, ML-classified events stored in PostgreSQL.

---

Pipeline at a Glance

Property	Value
Process	Single Python process (services/pipeline/main.py)
Concurrency	Batch-oriented (no threads, GIL-friendly)
Batch Size	100 messages (default 5000 in code, overridden to 100 in .env)
Batch Timeout	0.5 seconds
Throughput	~2,100 events/sec sustained
Scaling	Horizontal (add replicas matching Kafka partitions)
State	Drain3 state persisted to /app/state/drain3_state.json

---

Architecture

%%{init: {'theme':'default','themeVariables':{'primaryColor':'#1a73e8'}}}%%
flowchart TD
  subgraph CONSUMER["Consumer Pipeline"]
    KAFKA[Kafka Consumer<br/>group: aiops-pipeline<br/>topics: raw_*]
    BATCH[Batch Accumulator<br/>100 msgs / 0.5s]

    subgraph STAGES["Processing Stages"]
      F1[Noise Filter]
      S1[1. Parse<br/>Drain3]
      S2[2. Normalize<br/>Pydantic + PII]
      S3[3. Dedup<br/>Redis Lua]
      S4[4. Features<br/>Polars]
      S5[5. Classify<br/>RoBERTa/Rule]
      S6[6. Store<br/>PostgreSQL executemany]
    end

    DLQ[DLQ Producer<br/>Kafka + PostgreSQL]
    STATS[Stats Collector]
  end

  KAFKA --> BATCH --> F1
  F1 -->|Pass| S1
  F1 -.->|Noise| SKIP
  S1 -->|Valid| S2
  S1 -.->|Invalid| DLQ
  S2 --> S3
  S3 -->|Unique| S4
  S3 -.->|Duplicate| SKIP
  S4 --> S5
  S5 --> S6
  S6 --> PG[(PostgreSQL)]
  S6 -->|Notify| REDIS[Redis Pub/Sub]
  DLQ -.->|Reject| KAFKA_DLQ[(Kafka DLQ)]
  DLQ -.->|Persist| PG_DLQ[(dlq_events)]
  STATS -.->|Metrics| LOG[Structured Logs]

---

Stage Summary

#	Stage	File	Input	Output	Key Tech
0	Noise Filter	main.py	Raw Kafka msg	Filtered msg	Heuristics
1	Parse	parser.py	Message text	template_id, template_text	Drain3
2	Normalize	normalizer.py	Raw + template	CanonicalEvent	Pydantic v2, Regex
3	Dedup	dedup.py	CanonicalEvent	Unique events	Redis Lua (SHA-256)
4	Features	feature_engineering.py	Event	Enriched event	Polars
5	Classify	ml_classifier.py	Event + template	Priority P1–P4	RoBERTa + Cache
6	Store	storage.py	Enriched batch	PG rows	executemany INSERT

---

Batch Processing Flow

sequenceDiagram
  participant CONS as PipelineRunner
  participant KAFKA as Kafka
  participant STAGES as Stages
  participant REDIS as Redis
  participant PG as PostgreSQL
  participant DLQ as DLQ

  loop Forever
    CONS->>KAFKA: poll_batch(100, 0.5s)
    KAFKA-->>CONS: Messages
    CONS->>CONS: Filter noise
    par Process each message
      CONS->>STAGES: Parse → Normalize → Dedup → Features → Classify
    end
    CONS->>STAGES: Batch write (executemany)
    STAGES->>PG: INSERT ... ON CONFLICT
    PG-->>STAGES: OK
    STAGES->>REDIS: PUBLISH events:new
    CONS->>KAFKA: commit offsets
    CONS->>LOG: Log batch stats
  end

---

Graceful Shutdown

# Signal handling
_SHUTDOWN_REQUESTED = False

def _handle_shutdown(signum, frame):
    global _SHUTDOWN_REQUESTED
    _SHUTDOWN_REQUESTED = True

signal.signal(signal.SIGTERM, _handle_shutdown)
signal.signal(signal.SIGINT, _handle_shutdown)

Shutdown Sequence: 1. Signal received → _SHUTDOWN_REQUESTED = True 2. Finish current batch (process + write) 3. Commit Kafka offsets 4. Save Drain3 state (parser.shutdown()) 5. Close PostgreSQL connection 6. Exit

---

Statistics (Logged per Batch)

{
  "timestamp": "2026-01-15T10:30:00Z",
  "level": "info",
  "event": "batch_processed",
  "received": 100,
  "written": 96,
  "duplicates": 3,
  "noise_filtered": 1,
  "ml_cache_hits": 84,
  "ml_predictions": 10,
  "ml_fallbacks": 2,
  "templates_known": 3847,
  "processing_time_ms": 125
}

Counters (in PipelineRunner._stats): - received_total, written_total, dlq_total - noise_filtered_total, duplicate_total - ml_cache_hits_total, ml_predictions_total, ml_fallback_total - batches_total, processing_errors_total

---

Configuration

Parameter	Env Var	Default	Description
Batch Size	PIPELINE_BATCH_SIZE	5000 (100 in .env)	Messages per poll
Batch Timeout	PIPELINE_BATCH_TIMEOUT_SECONDS	0.5	Max wait for batch
Dedup Window	DEDUP_WINDOW_SECONDS	300	Redis key TTL
Drain3 Max Clusters	DRAIN3_MAX_CLUSTERS	100000	Template limit
Drain3 State Path	DRAIN3_STATE_PATH	/app/state/drain3_state.json	Persistence
ML Min Confidence	ML_MIN_CONFIDENCE	0.80	Cache threshold
Kafka Group	KAFKA_CONSUMER_GROUP	aiops-pipeline	Consumer group
Kafka Topics	KAFKA_TOPIC_RAW_*	raw_logs, etc.	Input topics

---

Scaling

Replicas	Partitions Required	Throughput
1	1	~2,100 eps
2	2	~4,200 eps
3	3	~6,300 eps
6	6	~12,600 eps

Rule: replicas ≤ partitions for each input topic. Increase partitions in kafka-init for production.

# Scale pipeline
docker compose up -d --scale consumer-pipeline=3

---

Monitoring

Health

# Check if running
docker compose ps consumer-pipeline

# View logs
docker compose logs -f consumer-pipeline

Key Metrics (from logs)

Metric	Healthy Range
written / received	> 90%
duplicates / received	5-20% (depends on source)
ml_cache_hits / (ml_predictions + ml_cache_hits)	> 95%
ml_fallbacks / ml_predictions	< 5%
processing_time_ms	< 2000 ms

Drain3 State

# Check state file size
docker compose exec consumer-pipeline ls -lh /app/state/drain3_state.json
# ~300 KB typical

Kafka Lag

docker compose exec kafka-broker \
  kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group aiops-pipeline \
  --describe