332 lines
10 KiB
Markdown
332 lines
10 KiB
Markdown
# 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.
|