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:

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:

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

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

docker compose up -d --build

Verify services:

docker compose ps -a

Expected:

Redpanda       healthy
redpanda-init  exited successfully
ClickHouse     healthy

Available endpoints:

API:

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

Worker:

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

A valid CDR currently returns:

{
  "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:

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:

   pip install -e ".[load-test]"
   

2. Start the Locust Web GUI:

   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:

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:

   curl -s http://localhost:8000/metrics | grep cdr_api_received_total
   

2. Check worker consumer lag:

   curl -s http://localhost:8001/metrics | grep cdr_kafka_consumer_lag
   

3. Check aggregated pipeline statistics:

   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.