# 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
│   └── 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.
