A high-throughput, fault-tolerant streaming data pipeline designed to ingest, buffer, normalize, and store raw telecom Call Detail Records (CDRs) using pure Python (framework-less asyncio), Kafka (Redpanda), and ClickHouse.

Built for at-least-once delivery, backpressure protection, deterministic deduplication, and operational observability.

1. Quickstart

Prerequisites

Docker & Docker Compose v2+
Python 3.12+ with python -m pip install -r requirements.txt (for local tests and the traffic verifier)

1. Start the Pipeline

docker compose up --build

No .env file is required. To override defaults, copy .env.example to .env and edit it.

This single command brings up:

Redpanda (Kafka-compatible message broker on port 9092)
Infrastructure Initializer (init: creates/verifies topics and the canonical app/worker/schema.sql table definition)
ClickHouse (Analytics database on port 8123 HTTP / 9000 Native with auto-initialized schema)
Ingestion API (Port 8000)
Processing Worker (Port 8001)

The initializer waits for broker/database health; API and worker wait for init to complete successfully. Existing incompatible schemas or topic settings fail initialization without deleting data.

2. Verify Pipeline Behaviors

In a separate terminal, run the automated traffic generator:

python scripts/generate_traffic.py

Illustrative expected verification summary (not evidence that the current revision passed):

=== Assessment Verification Summary ===
Records/requests sent:         1570 (expected 1,570)
All HTTP statuses:             202:1540, 400:30
Run-scoped ClickHouse rows:    1500 (expected 1,500 unique valid+burst rows)
/stats received delta:         1540
/stats processed delta:        1520
/stats DLQ delta:              50 (expected 50)
/stats in-flight buffer:       0

Scenario results:
  [PASS] 1. Valid records         HTTP 202=1000/1000; ClickHouse=1000/1000
  [PASS] 2. Missing caller_num    HTTP=400:20; DLQ missing_required_field=20/20
  [PASS] 3. Malformed JSON        HTTP 400=10/10; structured JSON errors=10/10
  [PASS] 4. Invalid RAT           HTTP 202=10/10; DLQ invalid_rat_value=10/10
  [PASS] 5. Invalid epoch         HTTP 202=10/10; DLQ invalid_epoch=10/10
  [PASS] 6. Duplicate raw_id      duplicate POSTs accepted=20/20; unique rows after FINAL=20/20
  [PASS] 7. 500-record burst      duration=<5s; 202=500/500; 429=0; network_errors=0; ClickHouse=500/500
  [PASS] Actual DLQ envelopes     verified=True; records=50/50; reasons={'missing_required_field': 20, 'malformed_json': 10, 'invalid_rat_value': 10, 'invalid_epoch': 10}
  [PASS] Pipeline metrics         received=1570; acknowledged=1540; processed=1520; DLQ=50 expected
  [PASS] Normalized storage       all submitted identities and field transformations match

OVERALL: PASS
============================================================

3. Load Testing (k6)

The default k6 assessment profile sends exactly 500 requests and requires all acknowledgements within five seconds. Set PROFILE=stress for the separate sustained-load profile. k6 checks API acceptance; the Python verifier checks durable ClickHouse and DLQ outcomes. Set P95_MS only when you want an additional latency SLA; it is disabled by default because the assessment does not define one.

Option A: Running with Docker (Recommended)

Running k6 via Docker avoids Windows host outbound socket limits:

Windows (PowerShell / Docker Desktop):

docker run --rm -v "${PWD}/scripts:/scripts" grafana/k6 run -e API_URL="http://host.docker.internal:8000/api/v1/cdr/raw" /scripts/load_test.js

Linux / macOS (Host Network):

docker run --rm -i --network=host grafana/k6 run - <scripts/load_test.js

Option B: Running with Local k6

If k6 is installed on your machine:

Windows (PowerShell):

k6 run scripts/load_test.js
# Or using explicit path:
& "C:\Program Files\k6\k6.exe" run scripts/load_test.js

Linux / macOS:
```
k6 run scripts/load_test.js
```

Tip

Windows Ephemeral Port Note: Under rapid high-concurrency load (>20,000 requests in <60 seconds), Windows may exhaust dynamic outbound ports held in TIME_WAIT, throwing connectex: Only one usage of each socket address is normally permitted. Running k6 inside Docker (Option A) avoids this host-level limitation.

2. Architecture & Data Flow

flowchart LR
    NE([Network Element]) -->|POST /api/v1/cdr/raw| API[Ingestion API :8000]
    API -->|Bounded Pending Deliveries| BP[Backpressure Gate]
    BP -->|Key: raw_id| Kafka[(Redpanda: cdr.raw)]

    Kafka -->|Consume| Worker[Processing Worker :8001]

    subgraph WorkerLogic [Worker Pipeline]
        direction TB
        V[Validate & Clean] --> N[Normalize Schema]
        N --> B[Micro-Batch Buffer]
        B --> OC[Commit Offsets after ClickHouse ACK]
    end
    Worker --- WorkerLogic

    WorkerLogic -->|Batch Insert| CH[(ClickHouse: cdr_records)]
    WorkerLogic -.->|Malformed / Errors| DLQ[(Redpanda: cdr.raw.dlq)]
    WorkerLogic --> PromW[Worker /metrics]

    API --> PromA[API /metrics]
    API -->|Query Internal Stats| Worker
    API --> Stats[Public GET /stats]

Component Roles

Ingestion API: A framework-less HTTP/1.1 server built with standard library asyncio.start_server. Bounds outstanding Kafka publications for backpressure (429 Too Many Requests), validates JSON syntax and field types, and publishes to Kafka using raw_id as the partition key.
Message Broker (Redpanda/Kafka):
- cdr.raw: 3 partitions, LZ4 compression, 24-hour retention.
- cdr.raw.dlq: 1 partition, 7-day retention.
Processing Worker: Consumes cdr.raw, normalizes subscriber telephone numbers to E.164 (phonenumbers), maps radio access technology (RAT), derives deterministic UUIDv5 cdr_ids, micro-batches into ClickHouse, and manages Kafka offset commits.
ClickHouse: Columnar storage using ReplacingMergeTree(processed_at) partitioned by month and ordered by subscriber metadata and timestamp.

3. Micro-Batching Strategy

Inserting row-by-row into ClickHouse causes MergeTree merge storms and performance degradation. The processing worker implements micro-batching:

Flush Conditions (Whichever Occurs First)

Count Threshold: Accumulated in-memory buffer reaches BATCH_SIZE (default: 1,000 records).
Time Threshold: Elapsed time since last flush reaches BATCH_FLUSH_INTERVAL_SECONDS (default: 2 seconds).

Graceful Shutdown (`SIGTERM` / `SIGINT`)

On container stop or deployment:

Worker stops polling new messages from Kafka.
Any remaining records buffered in memory are flushed to ClickHouse.
Kafka offsets for the flushed batch are committed.
Worker exits with code 0 only when storage and commits succeed. A destination/commit failure exits nonzero and retains uncommitted source offsets for replay.

The Linux containers enforce a 45-second shutdown budget and a 60-second Compose grace period. On forced termination, uncommitted records replay on restart. The API stops admission and drains broker acknowledgements before exit. The worker deadline uses POSIX signals; Windows-host worker execution does not enforce that alarm.

4. Offset Management & At-Least-Once Delivery

Kafka consumer offsets advance only after ClickHouse acknowledges insertion or the DLQ acknowledges delivery. Both automatic offset storage and automatic commits are disabled. Commits specify the next processed offset for each affected partition.

Failure Scenario Handling

Failure Scenario	Resolution Mechanism
ClickHouse Write Timeout	Worker retries the insert with exponential backoff (`CH_RETRY_MAX_ATTEMPTS=3`, initial delay `500ms`, factor `2.0`). If all retries fail, the entire batch is wrapped in DLQ envelopes, published to `cdr.raw.dlq` with reason `normalization_error`, and offsets advance only after all DLQ acknowledgements. A failed DLQ delivery stops the worker without committing that batch.
Worker Crash After Write, Before Commit	Upon worker restart, the batch is re-consumed from Kafka and re-inserted into ClickHouse. Idempotency is guaranteed by `ReplacingMergeTree(processed_at)` and deterministic UUIDv5 `cdr_id`s. Duplicate rows sharing the `ORDER BY` key are merged.
Partial Batch Failure	Do not assume a cross-partition transaction. Retry the batch with stable identities. Advance source offsets only after the entire batch has a confirmed destination; inspect every partition commit result and stop on errors. Partial commit or DLQ success can cause duplicates on replay, but never justifies skipping undelivered records.

5. Deduplication Strategy

To guarantee idempotency during network retries and crash replays:

Deterministic cdr_id Generation: Derived using UUIDv5 with the DNS namespace and raw_id:
```
import uuid
cdr_id = uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)
```
Identical raw_ids always generate the exact same cdr_id.
ReplacingMergeTree(processed_at) Engine:
```
ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn, cdr_id)
```
The deterministic cdr_id is part of the replacement key so exact replays are deduplicated while distinct raw_id values cannot collapse merely because their session metadata matches. Queries can use FINAL for an immediate deduplicated view.

Delivery Safety Contracts & Assessment Conflict Resolution

API Validation Boundary:
- The assessment specification contained conflicting requirements: Scenario 2 expected 20 records with missing caller_num in the DLQ, while the API specification required strict input validation rejecting missing required fields or invalid types with HTTP 400 Bad Request.
- Resolution: The Ingestion API enforces strict wire validation (rejecting missing required fields, non-string/non-integer types, and malformed JSON with structured 400 errors). To verify the worker's DLQ routing for malformed and missing-field records without compromising the API gate, generate_traffic.py directly injects 20 missing-field records and 10 malformed JSON records into Kafka, producing 50 DLQ records for the resolved scenarios. These are in addition to 1,570 HTTP requests; the 500-record burst contributes 500 additional unique stored rows. Missing Kafka evidence makes verification fail, never fall back to counters.
Broker-Confirmed Acceptance:
- The Ingestion API returns HTTP 202 Accepted only after Kafka acknowledges delivery. If Kafka delivery fails or times out (default 10s), the API returns HTTP 503 Service Unavailable. Timeout does not prove non-delivery: retry the identical raw_id and payload. Reusing an identity for a changed event is unsupported.
At-Least-Once Offsets & DLQ Delivery:
- The worker commits Kafka offsets per-partition only after ClickHouse acknowledges batch insertion, or after DLQ acknowledges message persistence. When an invalid record is routed to DLQ, its offset is committed immediately if no uncommitted records precede it on that partition, or deferred and committed alongside the preceding records when the batch flushes, preserving micro-batching without row-by-row inserts.

Assessment Data Reset / Rebuild Procedure

The canonical schema is app/worker/schema.sql. Initialization refuses any non-canonical sorting key; truncating rows does not change that key. There is no automatic migration or reset.

Only for disposable local assessment data, explicitly run the following commands with the same Compose project and environment options used at startup:

# Destructive: deletes this Compose project's Kafka and ClickHouse volumes.
docker compose down --volumes
docker compose up --build -d

If started with -p NAME --env-file PATH, use those options for both commands. Do not use this procedure for data that must be preserved. No reset has been performed by the static-review follow-up.

6. Observability & Monitoring

Prometheus Metrics

Ingestion API: http://localhost:8000/metrics
- cdr_api_received_total: Total CDRs received by the Ingestion API endpoint (including rejected and failed requests).
- cdr_api_acknowledged_total: Broker-confirmed publications observed by successful API handlers.
Processing Worker: http://localhost:8001/metrics
- cdr_processed_total{service_type="...", radio_access_type="..."}
- cdr_dlq_total{reason="..."}
- cdr_kafka_to_clickhouse_latency_seconds (Histogram)
- cdr_batch_flush_duration_seconds (Histogram)
- cdr_batch_size_at_flush (Histogram)
- cdr_in_flight_buffer_size (Gauge)
- cdr_kafka_consumer_lag{partition="..."} (Gauge)

Counters measure physical acknowledged processing/delivery operations, including replay. FINAL queries measure logical records. Kafka latency uses the message timestamp; lag is high watermark minus committed next offset and refreshes every five seconds. A worker restart resets its in-memory counters, so run the verifier against a quiet stack without restarts.

Unified `/stats` Endpoint

Exposed publicly by the Ingestion API at http://localhost:8000/stats:

{
  "pipeline": {
    "total_received": 1570,
    "total_acknowledged": 1540,
    "total_processed": 1520,
    "total_dlq": 50,
    "in_flight_buffer": 0,
    "last_flush_at": "2026-09-19T10:00:00Z"
  },
  "breakdown": {
    "by_service_type": { "4G_DATA": 1020, "5G_DATA": 500 },
    "by_radio_access_type": { "LTE_4G": 1020, "5G_NR": 500 },
    "dlq_by_reason": {
      "missing_required_field": 20,
      "invalid_rat_value": 10,
      "invalid_epoch": 10,
      "malformed_json": 10
    }
  }
}

Analytical Queries (ClickHouse)

Connect via HTTP http://localhost:8123 or CLI:

-- 1. Total count by service type
SELECT service_type, count() AS total
FROM telecom.cdr_records FINAL
GROUP BY service_type;

-- 2. Average session duration by Radio Access Type
SELECT radio_access_type, avg(duration_seconds) AS avg_duration
FROM telecom.cdr_records FINAL
GROUP BY radio_access_type;

-- 3. Top cells by total traffic volume
SELECT cell_id, sum(total_bytes) AS total_traffic
FROM telecom.cdr_records FINAL
GROUP BY cell_id
ORDER BY total_traffic DESC
LIMIT 10;

7. Configuration Reference

Compose reads optional .env overrides. Python scripts read exported environment variables, not .env automatically. Within Compose, ClickHouse always listens at clickhouse:8123; CLICKHOUSE_PORT controls only the published host port. For standalone Python clients it specifies their destination port. With a custom external ClickHouse, use explicit Compose service overrides for its internal address/port.

When running the verifier on the host, use --kafka-brokers localhost:<KAFKA_PORT> and the matching --api-url / --clickhouse-url. Run it without unrelated producers so metric deltas remain meaningful.

Variable	Default	Component	Description
`API_HOST`	`0.0.0.0`	API	HTTP host interface to bind
`API_PORT`	`8000`	API	Ingestion API port
`API_QUEUE_MAX_SIZE`	`10000`	API	Outstanding publications, held until delivery completes
`WORKER_STATS_URL`	`http://worker:8001/internal/stats`	API	Internal worker stats endpoint for aggregation
`WORKER_INSTANCE`	`worker-1`	Worker	Worker instance identifier for DLQ tracing
`WORKER_STATS_HOST`	`0.0.0.0`	Worker	Worker internal HTTP host
`WORKER_STATS_PORT`	`8001`	Worker	Worker internal HTTP port
`KAFKA_BOOTSTRAP_SERVERS`	`localhost:9092`	All	Kafka / Redpanda broker address
`KAFKA_RAW_TOPIC`	`cdr.raw`	All	Raw CDR topic name (3 partitions)
`KAFKA_DLQ_TOPIC`	`cdr.raw.dlq`	Worker	Dead Letter Queue topic name
`KAFKA_CONSUMER_GROUP`	`cdr-processing-group`	Worker	Kafka consumer group ID
`CLICKHOUSE_HOST`	`localhost`	Worker	ClickHouse host address
`CLICKHOUSE_PORT`	`8123`	Compose / standalone client	Published host HTTP port / standalone destination port
`CLICKHOUSE_DATABASE`	`telecom`	Worker	Target database name
`CLICKHOUSE_TABLE`	`cdr_records`	Worker	Target table name
`BATCH_SIZE`	`1000`	Worker	Number of records before triggering a micro-batch flush
`BATCH_FLUSH_INTERVAL_SECONDS`	`2.0`	Worker	Maximum seconds before triggering a micro-batch flush
`CH_RETRY_MAX_ATTEMPTS`	`3`	Worker	Maximum ClickHouse insert retry attempts
`CH_RETRY_INITIAL_DELAY_MS`	`500`	Worker	Initial retry backoff delay in milliseconds
`CH_RETRY_BACKOFF_FACTOR`	`2.0`	Worker	Exponential retry multiplier
`LOG_LEVEL`	`INFO`	All	Logging level (`DEBUG`, `INFO`, `WARN`, `ERROR`)

8. Verification status

The current worktree passes all 64 pytest tests, resolved Compose configuration validation, and strict OpenSpec validation. On 2026-09-20, an isolated pp-cdr-verify-0920 Compose project with project-owned volumes and host ports 18000/18001/19092/18123 passed the full end-to-end verifier: 1,570 HTTP requests produced 1,540 acknowledgements and 30 structured rejections, 1,500 logical FINAL ClickHouse rows, 50 run-correlated DLQ envelopes, and matching metric deltas. The acknowledged 500-record burst completed in 1.672 seconds with 500 HTTP 202 responses and no 429 or network errors.

The same isolated run verified broker recovery (503 while Redpanda was stopped, then 202 after restart), worker restart/replay (the stable identity remained one FINAL row), idempotent canonical bootstrap, and incompatible-schema refusal without mutation. The disposable stack and volumes were removed after verification. Never reset an existing deployment to make a test pass.

Description

a high-throughput, fault-tolerant data pipeline that ingests raw, vendor-specific telecom Call Detail Records (CDRs) via an HTTP API, buffers them in Kafka, processes and normalizes them with a dedicated worker, stores them into ClickHouse, and exposes real-time monitoring of ingested CDR counts. ]

Readme 235 KiB

README.md

Telecom Raw CDR Ingestion & Analytics Pipeline