# Telecom CDR Pipeline High-throughput, fault-tolerant telecom Call Detail Record ingestion and analytics pipeline built with FastAPI, Redpanda/Kafka, ClickHouse, and Python. ## Architecture ```text Network Element │ ▼ FastAPI Ingestion API │ ├── Validation │ ├── Bounded Async Queue │ └── Kafka Producer │ ▼ Redpanda / cdr.raw │ ▼ Processing Worker │ ├── Decode + Validate ├── Normalize ├── Micro-Batch │ ├── valid ──────► ClickHouse │ └── invalid ────► cdr.raw.dlq ``` The main services are: * **FastAPI API** — validates incoming CDRs, applies backpressure, and publishes accepted records to Kafka. * **Redpanda** — Kafka-compatible durable buffer between ingestion and processing. * **Processing Worker** — validates, normalizes, batches, and persists CDRs. * **ClickHouse** — analytical storage for normalized records. * **DLQ** — stores malformed or unprocessable records with structured failure information. ## Features * Bounded asynchronous ingestion queue * `202`, `400`, `429`, and `503` ingestion responses * Long-lived asynchronous Kafka producer * Kafka delivery acknowledgement before `202 Accepted` * Kafka consumer with manual offset management * Worker-side validation and normalization * E.164 phone-number normalization using `phonenumbers` * RAT mapping for `3G`, `LTE_4G`, and `5G_NR` * Deterministic UUIDv5 `cdr_id` generation * Structured Dead Letter Queue * Configurable ClickHouse micro-batching * Flush on 1,000 records or 2-second timeout * ClickHouse retries with exponential backoff * At-least-once delivery semantics * Replay-safe ClickHouse storage using `ReplacingMergeTree` * Graceful worker shutdown with final buffer flush * Separate Prometheus metrics endpoints for API and worker * Kafka consumer lag monitoring * Aggregated `/stats` endpoint * Health-based Docker Compose startup ordering * Automated Kafka topic provisioning * Automated and idempotent ClickHouse initialization * Unit, integration, functional, and load testing --- ## Project Structure Generated directories such as `__pycache__` and `*.egg-info` are omitted below. ```text telecom-cdr-pipeline/ ├── Dockerfile.api ├── Dockerfile.worker ├── README.md ├── docker-compose.yml ├── pyproject.toml │ ├── infrastructure/ │ └── clickhouse/ │ └── init.sql │ ├── scripts/ │ ├── functional_traffic.py │ ├── load_test.js │ ├── locustfile.py │ └── sample-valid-cdr.json │ ├── src/ │ └── core/ │ ├── __init__.py │ ├── config.py │ │ │ ├── api/ │ │ ├── __init__.py │ │ ├── kafka_publisher.py │ │ ├── main.py │ │ ├── metrics.py │ │ ├── publisher.py │ │ ├── routes.py │ │ └── stats.py │ │ │ ├── domain/ │ │ ├── __init__.py │ │ ├── errors.py │ │ ├── models.py │ │ ├── normalization.py │ │ └── validation.py │ │ │ └── worker/ │ ├── __init__.py │ ├── batch.py │ ├── clickhouse.py │ ├── consumer.py │ ├── dlq.py │ ├── main.py │ ├── metrics.py │ ├── observability.py │ ├── processor.py │ └── stats.py │ └── tests/ ├── integration/ │ └── test_ingestion_api.py │ └── unit/ ├── test_batch.py ├── test_dlq.py ├── test_normalization.py ├── test_project_setup.py ├── test_validation.py ├── test_worker_processor.py └── test_worker_stats.py ``` --- ## Quickstart ### 1. Configure Environment Create your local environment file from `.env.example`: ```bash cp .env.example .env ``` ### 2. Start the Complete Pipeline ```bash docker compose up --build -d ``` Check service status: ```bash docker compose ps -a ``` Expected state: ```text telecom-redpanda healthy telecom-clickhouse healthy telecom-api healthy telecom-worker healthy telecom-redpanda-init exited (0) ``` ### 3. Verify Health API: ```bash curl http://localhost:8000/health ``` Worker: ```bash curl http://localhost:8001/health ``` ### 4. Send a Sample CDR ```bash curl -X POST \ http://localhost:8000/api/v1/cdr/raw \ -H "Content-Type: application/json" \ --data-binary "@scripts/sample-valid-cdr.json" ``` Expected response: ```json { "status": "accepted", "raw_id": "raw-test-001" } ``` --- ## API Endpoints ### Ingestion API ```text POST /api/v1/cdr/raw ``` Possible responses: | Status | Meaning | | ------ | ------------------------------------------- | | `202` | Payload validated and acknowledged by Kafka | | `400` | Invalid JSON or invalid CDR schema | | `429` | Bounded ingestion queue is full | | `503` | Kafka delivery is unavailable | ### API Observability ```text GET http://localhost:8000/health GET http://localhost:8000/metrics GET http://localhost:8000/stats ``` ### Worker Observability ```text GET http://localhost:8001/health GET http://localhost:8001/metrics GET http://localhost:8001/internal/stats ``` --- ## Micro-Batching The worker never inserts CDRs into ClickHouse row-by-row. Normalized records are buffered and flushed when either condition occurs: ```text Batch reaches 1,000 records OR 2 seconds pass after the batch starts ``` Configuration: ```env BATCH_SIZE=1000 BATCH_FLUSH_INTERVAL_SECONDS=2 ``` This reduces ClickHouse insert overhead while still keeping low-volume traffic latency bounded. On `SIGTERM`, the worker stops consuming new records, flushes the remaining buffer, commits the corresponding Kafka offsets, and then exits. --- ## Delivery and Offset Strategy The worker uses **at-least-once delivery**. Kafka auto-commit is disabled. The normal processing order is: ```text Kafka consume ↓ Validate + Normalize ↓ Micro-Batch ↓ ClickHouse INSERT ↓ ClickHouse ACK ↓ Kafka offset commit ``` Offsets are never committed before the corresponding processing outcome is safely completed. ### ClickHouse Failure A failed ClickHouse batch is retried with exponential backoff: ```text Attempt 1 ↓ failure 0.5s delay Attempt 2 ↓ failure 1.0s delay Attempt 3 ↓ failure Entire batch → DLQ ``` Offsets are committed only after every DLQ publication succeeds. If DLQ publishing fails, the offsets remain uncommitted so Kafka can replay the records. ### Worker Crash After ClickHouse Write If ClickHouse successfully stores a batch but the worker crashes before committing Kafka offsets, Kafka delivers the records again after restart. This can produce duplicate physical inserts, which is handled by the deduplication strategy below. ### Partial Batch Failure A batch is treated as a single processing unit. Offsets are never partially committed for a failed ClickHouse batch. --- ## Deduplication Each CDR gets a deterministic UUIDv5 generated from its `raw_id`: ```python uuid.uuid5(uuid.NAMESPACE_DNS, raw_id) ``` The same `raw_id` therefore always generates the same `cdr_id`. ClickHouse uses: ```text ReplacingMergeTree(processed_at) ``` with `cdr_id` included in the sorting key. This allows records replayed because of at-least-once Kafka delivery to eventually collapse into one logical record. For immediate query-time deduplication: ```sql SELECT * FROM telecom.cdr_records FINAL; ``` --- ## Dead Letter Queue Invalid or unprocessable Kafka records are published to: ```text cdr.raw.dlq ``` DLQ records include: ```json { "original_payload": {}, "failure_reason": "invalid_rat_value", "failed_field": "rat", "failed_at": "2026-09-20T12:00:00Z", "worker_instance": "worker-1" } ``` Supported failure reasons: ```text missing_required_field invalid_rat_value invalid_epoch malformed_json normalization_error ``` A bad record does not crash the worker. --- ## Monitoring ### API Metrics ```bash curl http://localhost:8000/metrics ``` Includes: ```text cdr_api_received_total ``` ### Worker Metrics ```bash curl http://localhost:8001/metrics ``` Includes: ```text cdr_processed_total cdr_dlq_total cdr_kafka_to_clickhouse_latency_seconds cdr_batch_flush_duration_seconds cdr_batch_size_at_flush cdr_in_flight_buffer_size cdr_kafka_consumer_lag ``` ### Pipeline Stats ```bash curl http://localhost:8000/stats ``` Example: ```json { "pipeline": { "total_received": 1000, "total_processed": 995, "total_dlq": 5, "in_flight_buffer": 0, "last_flush_at": "2026-09-20T12:00:00Z" }, "breakdown": { "by_service_type": {}, "by_radio_access_type": {}, "dlq_by_reason": {} } } ``` --- ## ClickHouse Verification Open a ClickHouse client: ```bash docker compose exec clickhouse \ clickhouse-client \ --user telecom \ --password telecom_dev_password \ --database telecom ``` Count records by service: ```sql SELECT service_type, count() AS total FROM cdr_records GROUP BY service_type; ``` Average duration by RAT: ```sql SELECT radio_access_type, avg(duration_seconds) AS avg_duration FROM cdr_records GROUP BY radio_access_type; ``` Top cells by traffic: ```sql SELECT cell_id, sum(total_bytes) AS total_traffic FROM cdr_records GROUP BY cell_id ORDER BY total_traffic DESC LIMIT 10; ``` Check logical deduplication: ```sql SELECT raw_id, count() FROM cdr_records FINAL GROUP BY raw_id HAVING count() > 1; ``` Expected result: ```text 0 rows ``` --- ## Automated Tests Install development dependencies: ```bash pip install -e ".[dev]" ``` Run all unit and integration tests: ```bash pytest -v ``` --- ## Functional Traffic Verification The functional traffic script exercises the pipeline with the assessment scenarios, including valid records, invalid requests, and duplicate `raw_id` values. Run: ```bash python scripts/functional_traffic.py ``` The assessment scenarios are: | Scenario | Count | Expected | | ------------------------- | ----: | --------------------------------------------- | | Valid records | 1,000 | Stored in ClickHouse | | Missing `caller_num` | 20 | Rejected / DLQ when worker is tested directly | | Invalid JSON | 10 | API `400`; DLQ when worker is bypassed | | Unknown `rat=99` | 10 | Rejected / DLQ when worker is tested directly | | `epoch_start > epoch_end` | 10 | Rejected / DLQ when worker is tested directly | | Duplicate `raw_id` | 20 | One logical ClickHouse record per `raw_id` | Invalid records submitted through the public API are rejected before Kafka. Worker-side DLQ handling is independently exercised by publishing malformed or invalid records directly to `cdr.raw` and by the automated worker tests. ### Functional Traffic Result Paste the final `functional_traffic.py` output below: ```text Total requests: 1070 Elapsed: 5.21s duplicate status=202 count=20 invalid_epoch status=400 count=10 invalid_json status=400 count=10 invalid_rat status=400 count=10 missing_caller_num status=400 count=20 valid status=202 count=1000 Expected: valid → 1000 x 202 duplicate → 20 x 202 missing_caller → 20 x 400 invalid_json → 10 x 400 invalid_rat → 10 x 400 invalid_epoch → 10 x 400 ``` --- ## Required k6 Burst Test The assessment requires: ```text 500 records in less than 5 seconds 0 × 429 responses No data loss ``` The included k6 script uses a configurable constant arrival rate. Run the required assessment scenario with: ```bash docker run --rm -i \ --network=host \ -e RATE=125 \ -e DURATION=4s \ -e PRE_ALLOCATED_VUS=500 \ -e MAX_VUS=750 \ grafana/k6 run - \ < scripts/load_test.js ``` This schedules approximately: ```text 125 requests/second × 4 seconds = 500 requests ``` A successful result should show approximately: ```text iterations 500 http_reqs 500 cdr_accepted 500 cdr_backpressure 0 cdr_unavailable 0 http_req_failed 0% dropped_iterations 0 ``` After the run, allow the worker to finish processing and verify that no records were lost. Check worker lag: ```bash curl -s http://localhost:8001/metrics \ | grep cdr_kafka_consumer_lag ``` Check pipeline statistics: ```bash curl http://localhost:8000/stats ``` ### Required k6 Result Paste the final required 500-record burst result below: ```text TOTAL RESULTS checks_total.......: 4087 272.979956/s checks_succeeded...: 34.47% 1409 out of 4087 checks_failed......: 65.52% 2678 out of 4087 ✗ accepted or gracefully rejected ↳ 34% — ✓ 1409 / ✗ 2678 CUSTOM cdr_accepted...................: 1409 94.110291/s cdr_unavailable................: 2678 178.869665/s HTTP http_req_duration..............: avg=5.26s min=44.55ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s { expected_response:true }...: avg=2.33s min=44.55ms med=1.75s max=9.94s p(90)=5.75s p(95)=6.1s http_req_failed................: 65.52% 2678 out of 4087 http_reqs......................: 4087 272.979956/s EXECUTION dropped_iterations.............: 5912 394.875826/s iteration_duration.............: avg=5.26s min=46.48ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s iterations.....................: 4087 272.979956/s vus............................: 214 min=214 max=2000 vus_max........................: 2000 min=1500 max=2000 NETWORK data_received..................: 862 kB 58 kB/s data_sent......................: 1.6 MB 105 kB/s running (15.0s), 0000/2000 VUs, 4087 complete and 0 interrupted iterations ingestion ✓ [ 100% ] 0000/2000 VUs 10s 1000.00 iters/s ``` --- ## Optional Stress Test The required assessment test is the 500-record burst above. A larger stress test can optionally be executed with: ```bash docker run --rm -i \ --network=host \ -e RATE=1000 \ -e DURATION=10s \ -e PRE_ALLOCATED_VUS=1500 \ -e MAX_VUS=2000 \ grafana/k6 run - \ < scripts/load_test.js ``` This is useful for identifying throughput and latency limits but is not required for the assessment's 500-record burst criterion. --- ## Configuration Reference Configuration is provided through environment variables. Local defaults are documented in `.env.example`. | Variable | Default | Purpose | | ----------------------------------------- | -------------------------------------- | -------------------------------------- | | `REDPANDA_VERSION` | `v26.2.3` | Redpanda image version | | `KAFKA_BOOTSTRAP_SERVERS` | `localhost:19092` | Kafka address for host applications | | `KAFKA_INTERNAL_BOOTSTRAP_SERVERS` | `redpanda:9092` | Kafka address inside Docker | | `KAFKA_RAW_TOPIC` | `cdr.raw` | Raw CDR topic | | `KAFKA_RAW_TOPIC_PARTITIONS` | `3` | Raw topic partitions | | `KAFKA_RAW_TOPIC_REPLICATION_FACTOR` | `1` | Raw topic replication | | `KAFKA_RAW_TOPIC_RETENTION_MS` | `86400000` | Raw topic retention | | `KAFKA_RAW_TOPIC_COMPRESSION` | `lz4` | Raw topic compression | | `KAFKA_DLQ_TOPIC` | `cdr.raw.dlq` | Dead Letter Queue topic | | `KAFKA_DLQ_TOPIC_PARTITIONS` | `1` | DLQ partitions | | `KAFKA_DLQ_TOPIC_REPLICATION_FACTOR` | `1` | DLQ replication | | `KAFKA_DLQ_TOPIC_RETENTION_MS` | `604800000` | DLQ retention | | `KAFKA_CONSUMER_GROUP` | `cdr-processing-workers` | Worker consumer group | | `KAFKA_AUTO_OFFSET_RESET` | `earliest` | Consumer fallback offset | | `KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS` | `1` | Worker Kafka poll timeout | | `KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS` | `5` | DLQ delivery timeout | | `KAFKA_DELIVERY_TIMEOUT_SECONDS` | `4` | API Kafka acknowledgement timeout | | `KAFKA_MESSAGE_TIMEOUT_MS` | `3000` | Kafka message timeout | | `KAFKA_LINGER_MS` | `5` | Producer batching delay | | `KAFKA_BATCH_SIZE` | `131072` | Kafka producer batch size | | `KAFKA_PRODUCER_MAX_MESSAGES` | `100000` | Kafka producer internal queue limit | | `API_QUEUE_SIZE` | `10000` | Bounded API ingestion queue | | `CLICKHOUSE_VERSION` | `26.8.7.19` | ClickHouse image version | | `CLICKHOUSE_HOST` | `localhost` | ClickHouse host for local applications | | `CLICKHOUSE_INTERNAL_HOST` | `clickhouse` | ClickHouse Docker service hostname | | `CLICKHOUSE_PORT` | `8123` | ClickHouse HTTP port | | `CLICKHOUSE_HTTP_HOST_PORT` | `8123` | Host-exposed ClickHouse port | | `CLICKHOUSE_DATABASE` | `telecom` | ClickHouse database | | `CLICKHOUSE_USER` | `telecom` | ClickHouse user | | `CLICKHOUSE_PASSWORD` | `telecom_dev_password` | Local development password | | `BATCH_SIZE` | `1000` | Worker micro-batch maximum size | | `BATCH_FLUSH_INTERVAL_SECONDS` | `2` | Worker time-based flush interval | | `CLICKHOUSE_MAX_INSERT_ATTEMPTS` | `3` | Maximum ClickHouse insert attempts | | `CLICKHOUSE_RETRY_BASE_DELAY_SECONDS` | `0.5` | Exponential retry base delay | | `CLICKHOUSE_CONNECT_TIMEOUT_SECONDS` | `5` | ClickHouse connection timeout | | `CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS` | `10` | ClickHouse request timeout | | `WORKER_INSTANCE_ID` | `worker-1` | Worker instance identifier | | `WORKER_HTTP_HOST` | `0.0.0.0` | Worker observability bind address | | `WORKER_HTTP_PORT` | `8001` | Worker observability port | | `WORKER_STATS_URL` | `http://localhost:8001/internal/stats` | Worker stats URL for host-run API | | `WORKER_STATS_TIMEOUT_SECONDS` | `2` | Worker stats HTTP timeout | | `API_HOST_PORT` | `8000` | API host port | | `WORKER_HTTP_HOST_PORT` | `8001` | Worker observability host port | Inside Docker Compose, service-to-service addresses override host defaults where required: ```text Kafka: redpanda:9092 ClickHouse: clickhouse:8123 Worker: http://worker:8001 ``` --- ## Shutdown Stop the stack while preserving Kafka and ClickHouse data: ```bash docker compose down ``` Remove containers and persistent development data: ```bash docker compose down -v ``` Use `-v` only when a completely clean local environment is desired.