# Telecom Raw CDR Ingestion & Analytics Pipeline 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 ```bash 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: ```bash 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)**: ```powershell 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)**: ```bash docker run --rm -i --network=host grafana/k6 run - [!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 ```mermaid 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_id`s, 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) 1. **Count Threshold**: Accumulated in-memory buffer reaches `BATCH_SIZE` (default: **1,000 records**). 2. **Time Threshold**: Elapsed time since last flush reaches `BATCH_FLUSH_INTERVAL_SECONDS` (default: **2 seconds**). ### Graceful Shutdown (`SIGTERM` / `SIGINT`) On container stop or deployment: 1. Worker stops polling new messages from Kafka. 2. Any remaining records buffered in memory are flushed to ClickHouse. 3. Kafka offsets for the flushed batch are committed. 4. 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: 1. **Deterministic `cdr_id` Generation**: Derived using **UUIDv5** with the DNS namespace and `raw_id`: ```python import uuid cdr_id = uuid.uuid5(uuid.NAMESPACE_DNS, raw_id) ``` Identical `raw_id`s always generate the exact same `cdr_id`. 2. **`ReplacingMergeTree(processed_at)` Engine**: ```sql 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 1. **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. 2. **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. 3. **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: ```powershell # 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`: ```json { "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: ```sql -- 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:` 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`) | | `API_MAX_BODY_BYTES` | `65536` | API | Maximum request body; oversized requests return 413 | | `API_READ_TIMEOUT_SECONDS` | `10` | API | Request read deadline; timeout returns 408 | | `KAFKA_DELIVERY_TIMEOUT_SECONDS` | `10` | API / worker | Broker acknowledgement deadline | | `SHUTDOWN_TIMEOUT_SECONDS` | `45` | API / Linux worker | Graceful shutdown budget | | `KAFKA_AUTO_OFFSET_RESET` | `earliest` | Worker | Initial position when no committed offset exists | | `KAFKA_PORT` | `9092` | Compose | External broker listener and published port | | `KAFKA_ADVERTISED_HOST` | `localhost` | Compose | Broker address advertised to external clients | | `REDPANDA_ADMIN_PORT` | `9644` | Compose | Published Redpanda administration port | | `CLICKHOUSE_NATIVE_PORT` | `9000` | Compose | Published native ClickHouse port | | `CLICKHOUSE_USER` | `default` | Database / clients | Database user | | `CLICKHOUSE_PASSWORD` | empty | Database / clients | Database password | ## 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.