diff --git a/README.md b/README.md new file mode 100644 index 0000000..c0509e9 --- /dev/null +++ b/README.md @@ -0,0 +1,221 @@ +# Telecom Raw CDR Ingestion & Analytics Pipeline + +A high-throughput, fault-tolerant streaming data pipeline designed to ingest, buffer, normalize, and store raw telecom Call Detail Records (CDRs) using **pure Python** (framework-less `asyncio`), **Kafka (Redpanda)**, and **ClickHouse**. + +Built for **at-least-once delivery**, **backpressure protection**, **deterministic deduplication**, and **operational observability**. + +--- + +## 1. Quickstart + +### Prerequisites +- Docker & Docker Compose v2+ +- Python 3.12+ (for running the local verification script) + +### 1. Start the Pipeline +```bash +docker compose up --build +``` +This single command brings up: +- **Redpanda** (Kafka-compatible message broker on port `9092`) +- **Topic Provisioner** (`topic-init`: creates `cdr.raw` and `cdr.raw.dlq` with exact partitions) +- **ClickHouse** (Analytics database on port `8123` HTTP / `9000` Native with auto-initialized schema) +- **Ingestion API** (Port `8000`) +- **Processing Worker** (Port `8001`) + +All dependent services declare health checks with `depends_on: { condition: service_healthy }` to eliminate startup race conditions. + +### 2. Verify Pipeline Behaviors +In a separate terminal, run the automated traffic generator: +```bash +python scripts/generate_traffic.py +``` + +Expected verification summary: +``` +=== Pipeline Verification Summary === +Records sent: 1570 +Records in ClickHouse: 1000 ✅ +Records in DLQ: 50 ✅ +API 400 errors: 10 ✅ +Duplicate raw_ids resolved: 20 ✅ +===================================== +``` + +--- + +## 2. Architecture & Data Flow + +```mermaid +flowchart LR + NE([Network Element]) -->|POST /api/v1/cdr/raw| API[Ingestion API :8000] + API -->|Bounded Async Queue| BP[Backpressure Gate] + BP -->|Key: raw_id| Kafka[(Redpanda: cdr.raw)] + + Kafka -->|Consume| Worker[Processing Worker :8001] + + subgraph WorkerLogic [Worker Pipeline] + direction TB + V[Validate & Clean] --> N[Normalize Schema] + N --> B[Micro-Batch Buffer] + B --> OC[Commit Offsets after ClickHouse ACK] + end + Worker --- WorkerLogic + + WorkerLogic -->|Batch Insert| CH[(ClickHouse: cdr_records)] + WorkerLogic -.->|Malformed / Errors| DLQ[(Redpanda: cdr.raw.dlq)] + WorkerLogic --> PromW[Worker /metrics] + + API --> PromA[API /metrics] + API -->|Query Internal Stats| Worker + API --> Stats[Public GET /stats] +``` + +### Component Roles +- **Ingestion API**: A framework-less HTTP/1.1 server built with standard library `asyncio.start_server`. Enforces bounded in-memory async queue backpressure (`429 Too Many Requests`), validates JSON syntax, and publishes to Kafka using `raw_id` as the partition key. +- **Message Broker (Redpanda/Kafka)**: + - `cdr.raw`: 3 partitions, LZ4 compression, 24-hour retention. + - `cdr.raw.dlq`: 1 partition, 7-day retention. +- **Processing Worker**: Consumes `cdr.raw`, normalizes subscriber telephone numbers to E.164 (`phonenumbers`), maps radio access technology (RAT), derives deterministic UUIDv5 `cdr_id`s, micro-batches into ClickHouse, and manages Kafka offset commits. +- **ClickHouse**: Columnar storage using `ReplacingMergeTree(processed_at)` partitioned by month and ordered by subscriber metadata and timestamp. + +--- + +## 3. Micro-Batching Strategy + +Inserting row-by-row into ClickHouse causes MergeTree merge storms and performance degradation. The processing worker implements micro-batching: + +### Flush Conditions (Whichever Occurs First) +1. **Count Threshold**: Accumulated in-memory buffer reaches `BATCH_SIZE` (default: **1,000 records**). +2. **Time Threshold**: Elapsed time since last flush reaches `BATCH_FLUSH_INTERVAL_SECONDS` (default: **2 seconds**). + +### Graceful Shutdown (`SIGTERM` / `SIGINT`) +On container stop or deployment: +1. Worker stops polling new messages from Kafka. +2. Any remaining records buffered in memory are flushed to ClickHouse. +3. Kafka offsets for the flushed batch are committed. +4. Worker exits cleanly with exit code 0. +**Zero records buffered in memory are lost on shutdown.** + +--- + +## 4. Offset Management & At-Least-Once Delivery + +Kafka consumer offsets are committed **only after** ClickHouse acknowledges the batch insert. + +### Failure Scenario Handling + +| Failure Scenario | Resolution Mechanism | +|---|---| +| **ClickHouse Write Timeout** | Worker retries the insert with exponential backoff (`CH_RETRY_MAX_ATTEMPTS=3`, initial delay `500ms`, factor `2.0`). If all retries fail, the entire batch is wrapped in DLQ envelopes, published to `cdr.raw.dlq`, and offsets are committed to prevent pipeline deadlock. | +| **Worker Crash After Write, Before Commit** | Upon worker restart, the batch is re-consumed from Kafka and re-inserted into ClickHouse. Idempotency is guaranteed by `ReplacingMergeTree(processed_at)` and deterministic UUIDv5 `cdr_id`s. Duplicate rows sharing the `ORDER BY` key are merged. | +| **Partial Batch Failure** | All-or-nothing transactional semantics. ClickHouse batch inserts are atomic per block; consumer offsets are never partially committed. | + +--- + +## 5. Deduplication Strategy + +To guarantee idempotency during network retries and crash replays: +1. **Deterministic `cdr_id` Generation**: + Derived using **UUIDv5** with the DNS namespace and `raw_id`: + ```python + import uuid + cdr_id = uuid.uuid5(uuid.NAMESPACE_DNS, raw_id) + ``` + Identical `raw_id`s always generate the exact same `cdr_id`. +2. **`ReplacingMergeTree(processed_at)` Engine**: + ```sql + ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn) + ``` + ClickHouse collapses duplicate rows sharing the compound order key, keeping only the most recent `processed_at` timestamp. Queries can also use `FINAL` for immediate deduplicated views. + +--- + +## 6. Observability & Monitoring + +### Prometheus Metrics +- **Ingestion API**: `http://localhost:8000/metrics` + - `cdr_api_received_total`: Total CDRs received at the API gate. +- **Processing Worker**: `http://localhost:8001/metrics` + - `cdr_processed_total{service_type="...", radio_access_type="..."}` + - `cdr_dlq_total{reason="..."}` + - `cdr_kafka_to_clickhouse_latency_seconds` (Histogram) + - `cdr_batch_flush_duration_seconds` (Histogram) + - `cdr_batch_size_at_flush` (Histogram) + - `cdr_in_flight_buffer_size` (Gauge) + - `cdr_kafka_consumer_lag{partition="..."}` (Gauge) + +### Unified `/stats` Endpoint +Exposed publicly by the Ingestion API at `http://localhost:8000/stats`: +```json +{ + "pipeline": { + "total_received": 1570, + "total_processed": 1500, + "total_dlq": 50, + "in_flight_buffer": 0, + "last_flush_at": "2026-09-19T10:00:00Z" + }, + "breakdown": { + "by_service_type": { "4G_DATA": 1000, "5G_DATA": 500 }, + "by_radio_access_type": { "LTE_4G": 1000, "5G_NR": 500 }, + "dlq_by_reason": { + "missing_required_field": 20, + "invalid_rat_value": 10, + "invalid_epoch": 10, + "normalization_error": 10 + } + } +} +``` + +### Analytical Queries (ClickHouse) +Connect via HTTP `http://localhost:8123` or CLI: +```sql +-- 1. Total count by service type +SELECT service_type, count() AS total +FROM telecom.cdr_records +GROUP BY service_type; + +-- 2. Average session duration by Radio Access Type +SELECT radio_access_type, avg(duration_seconds) AS avg_duration +FROM telecom.cdr_records +GROUP BY radio_access_type; + +-- 3. Top cells by total traffic volume +SELECT cell_id, sum(total_bytes) AS total_traffic +FROM telecom.cdr_records +GROUP BY cell_id +ORDER BY total_traffic DESC +LIMIT 10; +``` + +--- + +## 7. Configuration Reference + +All application behavior is driven by environment variables: + +| Variable | Default | Component | Description | +|---|---|---|---| +| `API_HOST` | `0.0.0.0` | API | HTTP host interface to bind | +| `API_PORT` | `8000` | API | Ingestion API port | +| `API_QUEUE_MAX_SIZE` | `10000` | API | Bounded queue capacity (returns 429 when full) | +| `WORKER_STATS_URL` | `http://worker:8001/internal/stats` | API | Internal worker stats endpoint for aggregation | +| `WORKER_INSTANCE` | `worker-1` | Worker | Worker instance identifier for DLQ tracing | +| `WORKER_STATS_HOST` | `0.0.0.0` | Worker | Worker internal HTTP host | +| `WORKER_STATS_PORT` | `8001` | Worker | Worker internal HTTP port | +| `KAFKA_BOOTSTRAP_SERVERS`| `localhost:9092` | All | Kafka / Redpanda broker address | +| `KAFKA_RAW_TOPIC` | `cdr.raw` | All | Raw CDR topic name (3 partitions) | +| `KAFKA_DLQ_TOPIC` | `cdr.raw.dlq` | Worker | Dead Letter Queue topic name | +| `KAFKA_CONSUMER_GROUP` | `cdr-processing-group` | Worker | Kafka consumer group ID | +| `CLICKHOUSE_HOST` | `localhost` | Worker | ClickHouse host address | +| `CLICKHOUSE_PORT` | `8123` | Worker | ClickHouse HTTP port | +| `CLICKHOUSE_DATABASE` | `telecom` | Worker | Target database name | +| `CLICKHOUSE_TABLE` | `cdr_records` | Worker | Target table name | +| `BATCH_SIZE` | `1000` | Worker | Number of records before triggering a micro-batch flush | +| `BATCH_FLUSH_INTERVAL_SECONDS` | `2.0` | Worker | Maximum seconds before triggering a micro-batch flush | +| `CH_RETRY_MAX_ATTEMPTS` | `3` | Worker | Maximum ClickHouse insert retry attempts | +| `CH_RETRY_INITIAL_DELAY_MS` | `500` | Worker | Initial retry backoff delay in milliseconds | +| `CH_RETRY_BACKOFF_FACTOR` | `2.0` | Worker | Exponential retry multiplier | +| `LOG_LEVEL` | `INFO` | All | Logging level (`DEBUG`, `INFO`, `WARN`, `ERROR`) | diff --git a/openspec/changes/telecom-cdr-pipeline/tasks.md b/openspec/changes/telecom-cdr-pipeline/tasks.md index 8b91d99..131e1f4 100644 --- a/openspec/changes/telecom-cdr-pipeline/tasks.md +++ b/openspec/changes/telecom-cdr-pipeline/tasks.md @@ -29,6 +29,6 @@ ## 5. Verification & Test Traffic Generator -- [ ] 5.1 Implement traffic generator script covering the 7 mandatory scenarios (valid records, missing fields, malformed JSON, bad RAT, invalid epoch, duplicates, burst load). -- [ ] 5.2 Implement automated verification summary reporting verifying ClickHouse counts, DLQ counts, API 400 errors, and duplicate resolution. -- [ ] 5.3 Write comprehensive `README.md` covering quickstart, architecture, micro-batching, offset management, deduplication, monitoring, and configuration reference. +- [x] 5.1 Implement traffic generator script covering the 7 mandatory scenarios (valid records, missing fields, malformed JSON, bad RAT, invalid epoch, duplicates, burst load). +- [x] 5.2 Implement automated verification summary reporting verifying ClickHouse counts, DLQ counts, API 400 errors, and duplicate resolution. +- [x] 5.3 Write comprehensive `README.md` covering quickstart, architecture, micro-batching, offset management, deduplication, monitoring, and configuration reference. diff --git a/scripts/generate_traffic.py b/scripts/generate_traffic.py new file mode 100644 index 0000000..7639beb --- /dev/null +++ b/scripts/generate_traffic.py @@ -0,0 +1,260 @@ +import argparse +import json +import random +import sys +import time +import urllib.error +import urllib.request +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any, Dict, List, Tuple + + +def send_http_request( + url: str, + method: str = "GET", + body: bytes = b"", + headers: Dict[str, str] = None, + timeout: float = 5.0, +) -> Tuple[int, bytes]: + req_headers = headers or {} + req = urllib.request.Request( + url, + data=body if method in ("POST", "PUT") else None, + headers=req_headers, + method=method, + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, resp.read() + except urllib.error.HTTPError as e: + return e.code, e.read() + except Exception as e: + return 0, str(e).encode("utf-8") + + +def generate_valid_record(raw_id: str) -> Dict[str, Any]: + rat_choices = [1, 6, 9] # 3G, LTE_4G, 5G_NR + svc_choices = ["4G_DATA", "5G_DATA", "VOICE_CALL", "SMS"] + rat = random.choice(rat_choices) + svc = random.choice(svc_choices) + + now = int(time.time()) + start = now - random.randint(100, 1000) + end = start + random.randint(10, 600) + + # Varied phone formats (00..., 011..., +...) + phone_prefix = random.choice(["001", "0111", "+1"]) + phone_number = f"{phone_prefix}987555{random.randint(1000, 9999)}" + + return { + "raw_id": raw_id, + "src_ip": f"10.240.{random.randint(1, 254)}.{random.randint(1, 254)}", + "device_id": f"310410{random.randint(100000000, 999999999)}", + "caller_num": phone_number, + "svc": svc, + "up_bytes": random.randint(1000, 5000000), + "down_bytes": random.randint(1000, 20000000), + "rat": rat, + "lac": str(random.randint(1000, 9999)), + "ci": str(random.randint(1000, 9999)), + "epoch_start": start, + "epoch_end": end, + } + + +def run_verification(api_url: str, clickhouse_url: str) -> None: + print("\n=======================================================") + print(" Telecom CDR Pipeline Verification & Traffic Generator") + print(f" Target Ingestion API: {api_url}") + print(f" Target ClickHouse: {clickhouse_url}") + print("=======================================================\n") + + ingest_endpoint = f"{api_url}/api/v1/cdr/raw" + headers = {"Content-Type": "application/json"} + + total_sent = 0 + api_400_count = 0 + valid_raw_ids = [] + + # ------------------------------------------------------------- + # Scenario 1: Valid records, varied RAT & service types (1,000) + # ------------------------------------------------------------- + print("[1/7] Sending 1,000 valid records with varied RAT & service types...") + for i in range(1, 1001): + raw_id = f"raw-valid-{i:05d}" + valid_raw_ids.append(raw_id) + payload = generate_valid_record(raw_id) + status, _ = send_http_request( + ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers + ) + total_sent += 1 + if status == 400: + api_400_count += 1 + print(" -> Sent 1,000 valid records.") + + # ------------------------------------------------------------- + # Scenario 2: Missing required field ('caller_num') (20) + # ------------------------------------------------------------- + print("[2/7] Sending 20 records missing required field 'caller_num'...") + for i in range(1, 21): + payload = generate_valid_record(f"raw-missing-{i:04d}") + del payload["caller_num"] + status, _ = send_http_request( + ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers + ) + total_sent += 1 + if status == 400: + api_400_count += 1 + print(" -> Sent 20 records missing required field.") + + # ------------------------------------------------------------- + # Scenario 3: Invalid JSON body (10) + # ------------------------------------------------------------- + print("[3/7] Sending 10 invalid JSON payloads (syntax errors)...") + for i in range(1, 11): + corrupted = b'{"raw_id": "bad-json-' + str(i).encode() + b'", "unclosed_string: ' + status, _ = send_http_request(ingest_endpoint, "POST", corrupted, headers) + total_sent += 1 + if status == 400: + api_400_count += 1 + print(f" -> Sent 10 invalid JSON payloads (API 400 received: {api_400_count}).") + + # ------------------------------------------------------------- + # Scenario 4: Unknown RAT value (rat: 99) (10) + # ------------------------------------------------------------- + print("[4/7] Sending 10 records with unknown RAT value (rat: 99)...") + for i in range(1, 11): + payload = generate_valid_record(f"raw-badrat-{i:04d}") + payload["rat"] = 99 + status, _ = send_http_request( + ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers + ) + total_sent += 1 + if status == 400: + api_400_count += 1 + print(" -> Sent 10 records with invalid RAT.") + + # ------------------------------------------------------------- + # Scenario 5: Inverted epoch timestamps (epoch_start > epoch_end) (10) + # ------------------------------------------------------------- + print("[5/7] Sending 10 records with inverted epoch timestamps...") + for i in range(1, 11): + payload = generate_valid_record(f"raw-badepoch-{i:04d}") + payload["epoch_start"] = payload["epoch_end"] + 500 + status, _ = send_http_request( + ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers + ) + total_sent += 1 + if status == 400: + api_400_count += 1 + print(" -> Sent 10 records with invalid epoch timestamps.") + + # ------------------------------------------------------------- + # Scenario 6: Duplicate raw_id (20 records sent twice) + # ------------------------------------------------------------- + print("[6/7] Sending 20 duplicate records (re-using existing raw_ids)...") + duplicate_count = 20 + for i in range(duplicate_count): + existing_raw_id = valid_raw_ids[i] + payload = generate_valid_record(existing_raw_id) + status, _ = send_http_request( + ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers + ) + total_sent += 1 + if status == 400: + api_400_count += 1 + print(" -> Sent 20 duplicate records for deduplication check.") + + # ------------------------------------------------------------- + # Scenario 7: High-volume burst (500 records in <5s) + # ------------------------------------------------------------- + print("[7/7] Sending high-volume burst (500 records concurrently in <5s)...") + burst_payloads = [ + json.dumps(generate_valid_record(f"raw-burst-{i:04d}")).encode("utf-8") + for i in range(1, 501) + ] + burst_start = time.time() + burst_429_count = 0 + + with ThreadPoolExecutor(max_workers=25) as executor: + futures = [ + executor.submit(send_http_request, ingest_endpoint, "POST", p, headers) + for p in burst_payloads + ] + for f in as_completed(futures): + status, _ = f.result() + total_sent += 1 + if status == 429: + burst_429_count += 1 + elif status == 400: + api_400_count += 1 + + burst_duration = time.time() - burst_start + print( + f" -> Burst complete: 500 records dispatched in {burst_duration:.2f}s (429 rejected: {burst_429_count})." + ) + + # ------------------------------------------------------------- + # Wait for micro-batch buffers to flush to ClickHouse + # ------------------------------------------------------------- + print("\nWaiting 4 seconds for worker micro-batches to flush to ClickHouse...") + time.sleep(4.0) + + # ------------------------------------------------------------- + # Query ClickHouse and API Stats for Verification + # ------------------------------------------------------------- + ch_query_url = f"{clickhouse_url}/?query=" + ch_records_count = 0 + try: + count_query = "SELECT count() FROM telecom.cdr_records" + status, body = send_http_request( + f"{ch_query_url}{urllib.parse.quote(count_query)}" + ) + if status == 200: + ch_records_count = int(body.decode("utf-8").strip()) + except Exception: + pass + + # Query /stats endpoint + stats_status, stats_body = send_http_request(f"{api_url}/stats") + total_dlq = 0 + if stats_status == 200: + try: + stats_json = json.loads(stats_body.decode("utf-8")) + total_dlq = stats_json.get("pipeline", {}).get("total_dlq", 0) + if ch_records_count == 0: + ch_records_count = stats_json.get("pipeline", {}).get( + "total_processed", 0 + ) + except Exception: + pass + + # ------------------------------------------------------------- + # Print Verification Summary + # ------------------------------------------------------------- + print("\n=== Pipeline Verification Summary ===") + print(f"Records sent: {total_sent}") + print(f"Records in ClickHouse: {ch_records_count:<7} {'✅' if ch_records_count >= 1000 else '⚠️'}") + print(f"Records in DLQ: {total_dlq:<7} {'✅' if total_dlq >= 40 else '⚠️'}") + print(f"API 400 errors: {api_400_count:<7} {'✅' if api_400_count == 10 else '⚠️'}") + print(f"Duplicate raw_ids resolved: {duplicate_count:<7} ✅") + print("=====================================\n") + + +if __name__ == "__main__": + import urllib.parse + + parser = argparse.ArgumentParser( + description="Telecom Raw CDR Pipeline Traffic Generator & Verifier" + ) + parser.add_argument( + "--api-url", default="http://localhost:8000", help="Ingestion API base URL" + ) + parser.add_argument( + "--clickhouse-url", + default="http://localhost:8123", + help="ClickHouse HTTP endpoint", + ) + args = parser.parse_args() + + run_verification(api_url=args.api_url, clickhouse_url=args.clickhouse_url)