# Telecom CDR Pipeline

High-throughput telecom Call Detail Record ingestion and analytics pipeline.

## Architecture

The system consists of:

* FastAPI ingestion API
* Redpanda as a Kafka-compatible message broker
* Processing and normalization worker
* ClickHouse analytical storage
* Dead Letter Queue for failed records

Current data flow:

```text
Network Element
    ↓
FastAPI API
    ↓
Validation
    ↓
Bounded Async Queue
    ↓
Redpanda / cdr.raw
    ↓
Processing Worker
    ↓
Validation + Normalization
    ├── invalid → cdr.raw.dlq
    └── valid
          ↓
      Micro-Batch
          ↓
      ClickHouse
```

Infrastructure already available:

```text
Redpanda
├── cdr.raw
└── cdr.raw.dlq

ClickHouse
└── telecom.cdr_records
```

worker processing is not connected yet.

## Current Features

* Dockerized Redpanda and ClickHouse
* Automatic and idempotent Kafka topic provisioning
* Automatic ClickHouse schema initialization
* Persistent Redpanda and ClickHouse volumes
* Raw and normalized CDR domain models
* Required-field and type validation
* RAT validation and normalization
* E.164 phone-number normalization
* Deterministic UUIDv5 `cdr_id` generation
* Epoch and traffic-value validation
* FastAPI `POST /api/v1/cdr/raw`
* FastAPI `GET /health`
* Structured JSON error responses
* Unit and API integration tests
* Bounded asynchronous ingestion queue for backpressure
* Long-lived asynchronous Kafka producer
* Kafka delivery acknowledgement before `202 Accepted`
* `429 Too Many Requests` on ingestion queue saturation
* `503 Service Unavailable` on Kafka delivery failure
* Kafka consumer with manual offset management
* Worker-side JSON decoding and validation
* Worker-side CDR normalization
* DLQ routing for malformed and unprocessable Kafka records
* DLQ records with failure reason, failed field, timestamp, and worker instance
* Kafka offsets committed only after the current processing outcome succeeds
* ClickHouse bulk storage through the official Python client
* Configurable CDR micro-batching
* Flush on 1,000 records or 2-second timeout
* Kafka offsets committed only after successful ClickHouse batch writes
* Remaining worker buffer flushed during graceful shutdown
* ClickHouse insert retries with exponential backoff
* Failed ClickHouse batches routed to the DLQ after three attempts
* Whole-batch offset commits with no partial Kafka acknowledgement
* Replay-safe storage using deterministic UUIDv5 identifiers and `ReplacingMergeTree`
* Graceful worker shutdown flushes buffered records before exit
* Separate Prometheus `/metrics` endpoints for the API and worker
* API accepted-CDR counter
* Worker processed and DLQ counters
* Kafka-to-ClickHouse and batch flush latency histograms
* Batch size and in-flight buffer metrics
* Kafka consumer lag by partition
* Aggregated human-readable `/stats` endpoint
* Fully containerized API and processing worker
* Health-based Docker Compose startup ordering
* Automatic Kafka topic and ClickHouse schema provisioning
* Persistent Redpanda and ClickHouse storage
* Complete stack startup with `docker compose up --build`

## Observability Architecture

Observability is strictly service-owned; each service manages its own telemetry and monitoring components:

* **API Observability** (`core.api.observability`):
  * Prometheus metrics: `cdr_api_received_total` counter.
  * API runtime stats: In-memory counter for accepted CDR requests.
  * Worker stats HTTP client: `HttpWorkerStatsClient` implementing `WorkerStatsProvider` to query the worker's internal stats endpoint.
  * Observability routes: Dedicated router mounted on the API for `GET /metrics` and `GET /stats`.
  * The `GET /stats` endpoint aggregates API reception stats with worker processing stats retrieved over HTTP.

* **Worker Observability** (`core.worker.observability`):
  * Prometheus metrics: Processed and DLQ counters, batch flush duration and Kafka-to-ClickHouse latency histograms, in-flight buffer and consumer lag gauges.
  * Runtime stats: Thread-safe in-memory breakdown by service type, radio access type, DLQ failure reason, and buffer status.
  * Internal observability HTTP server: `WorkerObservabilityServer` running a background ASGI server providing `GET /health`, `GET /metrics`, and `GET /internal/stats`.

## Project Structure

```text
telecom-cdr-pipeline/
├── infrastructure/
│   └── clickhouse/
│       └── init.sql
│
├── scripts/
│   ├── locustfile.py
│   ├── functional_traffic.py
│   └── sample-valid-cdr.json
│
├── src/
│   └── core/
│       ├── __init__.py
│       ├── config.py
│       │
│       ├── api/
│       │   ├── __init__.py
│       │   ├── kafka_publisher.py
│       │   ├── main.py
│       │   ├── publisher.py
│       │   ├── routes.py
│       │   └── observability/
│       │       ├── __init__.py
│       │       ├── metrics.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
│           ├── processor.py
│           └── observability/
│               ├── __init__.py
│               ├── metrics.py
│               ├── server.py
│               └── stats.py
│
├── tests/
│   ├── integration/
│   │   └── test_ingestion_api.py
│   │
│   └── unit/
│       ├── test_api_observability.py
│       ├── test_batch.py
│       ├── test_dlq.py
│       ├── test_normalization.py
│       ├── test_project_setup.py
│       ├── test_validation.py
│       ├── test_worker_observability.py
│       ├── test_worker_processor.py
│       └── test_worker_stats.py
│
├── .env.example
├── .gitignore
├── docker-compose.yml
├── pyproject.toml
└── README.md
```

## Configuration

Configuration is provided through environment variables documented in `.env.example`.

This includes:

* Redpanda version and topic settings
* Kafka broker and consumer configuration
* ClickHouse connection settings
* Worker batch size and flush interval

## Running Pipeline

```bash
docker compose up -d --build
```

Verify services:

```bash
docker compose ps -a
```

Expected:

```text
Redpanda       healthy
redpanda-init  exited successfully
ClickHouse     healthy
```

### Available endpoints:

#### API:

```text
GET  http://localhost:8000/health
GET  http://localhost:8000/metrics
GET  http://localhost:8000/stats
POST http://localhost:8000/api/v1/cdr/raw
```

#### Worker:

```text
GET  http://localhost:8001/health
GET  http://localhost:8001/metrics
GET  http://localhost:8001/internal/stats
```

A valid CDR currently returns:

```json
{
  "status": "accepted",
  "raw_id": "raw-test-001"
}
```

Invalid payloads return structured `400` responses.

## Testing

Once the full stack starts cleanly, a real CDR reaches ClickHouse, and:

```bash
pytest -v
```

## Load Testing with Locust (Ongoing Test & Web GUI)

A high-throughput Locust load testing suite is provided in `scripts/locustfile.py` using `FastHttpUser` (`geventhttpclient`). It runs an **ongoing test that executes continuously until you stop it** and exposes an interactive **Web GUI** to monitor real-time throughput, latency distributions, and backpressure.

### Option 1: Run Locust via Local Python

1. Ensure dependencies are installed:
   ```bash
   pip install -e ".[load-test]"
   ```
2. Start the Locust Web GUI:
   ```bash
   locust -f scripts/locustfile.py --host http://localhost:8000
   ```
3. Open your browser at **http://localhost:8089**.

### Option 2: Run Locust via Docker Compose

Locust is configured as an on-demand service under the `load-test` profile in `docker-compose.yml`:

```bash
docker compose --profile load-test up locust
```

Open your browser at **http://localhost:8089**. The target host is pre-configured to `http://api:8000`.

---

### Controlling the Ongoing Test from the Web GUI

1. Open **http://localhost:8089**.
2. Configure your test parameters:
   - **Number of users**: e.g., `50` (or `500` for high stress).
   - **Spawn rate**: e.g., `10` users per second.
   - **Host**: `http://localhost:8000` (or `http://api:8000` inside Docker).
3. Click **Start swarming**:
   - The test will **run continuously and never stop** automatically.
   - You can inspect real-time metrics across tabs:
     - **Statistics**: Total requests, RPS, 50th/90th/95th/99th percentiles, and failure counts.
     - **Charts**: Live charts for Total Requests per Second, Response Times (ms), and Number of Active Users.
     - **Failures**: Distinguishes normal graceful backpressure (`429 Too Many Requests`) from broker errors (`503 Service Unavailable`).
     - **Download Data**: Export requests/failures CSVs or generate full HTML test reports.
4. Adjust or Stop:
   - You can change the user count in real-time from the GUI top bar.
   - Click **Stop swarming** whenever you want to end the test.

---

### Correlating with Pipeline Observability

While the Locust test is running ongoing traffic:

1. Check API metrics:
   ```bash
   curl -s http://localhost:8000/metrics | grep cdr_api_received_total
   ```
2. Check worker consumer lag:
   ```bash
   curl -s http://localhost:8001/metrics | grep cdr_kafka_consumer_lag
   ```
3. Check aggregated pipeline statistics:
   ```bash
   curl -s http://localhost:8000/stats
   ```

## Delivery Semantics

The worker uses at-least-once delivery.

Kafka offsets are committed only after a ClickHouse batch is successfully written or, after exhausted retries, the complete failed batch is acknowledged by the DLQ.

Repeated records use deterministic UUIDv5 identifiers and `ReplacingMergeTree` to support replay-safe storage.