diff --git a/app/worker/__init__.py b/app/worker/__init__.py new file mode 100644 index 0000000..3ca6f5b --- /dev/null +++ b/app/worker/__init__.py @@ -0,0 +1 @@ +# app/worker package diff --git a/app/worker/clickhouse.py b/app/worker/clickhouse.py new file mode 100644 index 0000000..5980d18 --- /dev/null +++ b/app/worker/clickhouse.py @@ -0,0 +1,182 @@ +import time +from abc import ABC, abstractmethod +from typing import Any, List, Optional + +from app.common.config import config +from app.common.logging import setup_logger +from app.worker.normalizer import NormalizedRecord + +logger = setup_logger("processing-worker") + +COLUMN_NAMES = [ + "cdr_id", + "raw_id", + "msisdn", + "imsi", + "service_type", + "radio_access_type", + "cell_id", + "bytes_uploaded", + "bytes_downloaded", + "total_bytes", + "duration_seconds", + "session_start_time", + "session_end_time", + "source_ip", +] + + +class BaseClickHouseClient(ABC): + @abstractmethod + def insert_batch(self, records: List[NormalizedRecord]) -> bool: + pass + + @abstractmethod + def is_healthy(self) -> bool: + pass + + @abstractmethod + def close(self) -> None: + pass + + +class ClickHouseService(BaseClickHouseClient): + """ + ClickHouse client using clickhouse-connect with exponential retry backoff. + """ + + def __init__( + self, + host: Optional[str] = None, + port: Optional[int] = None, + user: Optional[str] = None, + password: Optional[str] = None, + database: Optional[str] = None, + table: Optional[str] = None, + max_attempts: Optional[int] = None, + initial_delay_ms: Optional[int] = None, + backoff_factor: Optional[float] = None, + ): + try: + import clickhouse_connect + self._cc = clickhouse_connect + except ImportError: + raise ImportError( + "clickhouse-connect is not installed. Run 'pip install clickhouse-connect'." + ) + + self.host = host or config.clickhouse.host + self.port = port or config.clickhouse.port + self.user = user or config.clickhouse.user + self.password = password or config.clickhouse.password + self.database = database or config.clickhouse.database + self.table = table or config.clickhouse.table + + self.max_attempts = max_attempts or config.retry.max_attempts + self.initial_delay_ms = initial_delay_ms or config.retry.initial_delay_ms + self.backoff_factor = backoff_factor or config.retry.backoff_factor + + self._client = None + self._connect() + + def _connect(self) -> None: + try: + self._client = self._cc.get_client( + host=self.host, + port=self.port, + username=self.user, + password=self.password, + database=self.database, + connect_timeout=5, + send_receive_timeout=10, + ) + except Exception as e: + logger.error( + f"ClickHouse connection failed: {e}", + extra={"event": "clickhouse_connection_failed"}, + ) + self._client = None + + def is_healthy(self) -> bool: + if self._client is None: + self._connect() + if self._client is None: + return False + try: + return self._client.ping() + except Exception: + return False + + def insert_batch(self, records: List[NormalizedRecord]) -> bool: + """ + Inserts a micro-batch of NormalizedRecord into ClickHouse with exponential backoff retry. + Returns True if successful, raises exception if all retries fail. + """ + if not records: + return True + + rows = [record.to_row() for record in records] + last_exception = None + + for attempt in range(1, self.max_attempts + 1): + try: + if self._client is None: + self._connect() + if self._client is None: + raise ConnectionError("ClickHouse client is not connected") + + self._client.insert( + table=self.table, + data=rows, + column_names=COLUMN_NAMES, + ) + return True + except Exception as e: + last_exception = e + delay = (self.initial_delay_ms / 1000.0) * ( + self.backoff_factor ** (attempt - 1) + ) + logger.warning( + f"ClickHouse insert attempt {attempt}/{self.max_attempts} failed: {e}. " + f"Retrying in {delay:.2f}s...", + extra={"event": "clickhouse_insert_retry"}, + ) + time.sleep(delay) + + logger.error( + f"All {self.max_attempts} ClickHouse insert attempts failed: {last_exception}", + extra={"event": "clickhouse_batch_failed"}, + ) + raise last_exception or RuntimeError("ClickHouse insert failed") + + def close(self) -> None: + if self._client: + self._client.close() + self._client = None + + +class MockClickHouseClient(BaseClickHouseClient): + """ + In-memory ClickHouse mock client for offline tests. + """ + + def __init__(self, fail_attempts: int = 0): + self.rows: List[Any] = [] + self.fail_attempts = fail_attempts + self.attempt_count = 0 + self._healthy = True + + def is_healthy(self) -> bool: + return self._healthy + + def insert_batch(self, records: List[NormalizedRecord]) -> bool: + self.attempt_count += 1 + if self.attempt_count <= self.fail_attempts: + raise TimeoutError("ClickHouse insert timed out (simulated)") + + for record in records: + self.rows.append(record.to_row()) + return True + + def close(self) -> None: + pass diff --git a/app/worker/consumer.py b/app/worker/consumer.py new file mode 100644 index 0000000..7433f83 --- /dev/null +++ b/app/worker/consumer.py @@ -0,0 +1,266 @@ +import signal +import time +from datetime import datetime, timezone +from typing import Any, List, Optional, Tuple + +from app.api.producer import ConfluentKafkaProducer, MessageProducer +from app.common.config import config +from app.common.logging import setup_logger +from app.worker.clickhouse import BaseClickHouseClient, ClickHouseService +from app.worker.dlq import route_to_dlq +from app.worker.internal_server import WorkerInternalServer +from app.worker.metrics import worker_metrics +from app.worker.normalizer import NormalizedRecord, normalize_cdr_payload + +logger = setup_logger("processing-worker") + + +class ProcessingWorker: + """ + Consumes raw CDRs from Kafka, normalizes schema, micro-batches into ClickHouse, + routes malformed payloads to DLQ, and manages offsets with at-least-once semantics. + """ + + def __init__( + self, + consumer: Optional[Any] = None, + dlq_producer: Optional[MessageProducer] = None, + clickhouse: Optional[BaseClickHouseClient] = None, + batch_size: Optional[int] = None, + flush_interval: Optional[float] = None, + internal_server: Optional[WorkerInternalServer] = None, + ): + self.batch_size = batch_size or config.batch.size + self.flush_interval = flush_interval or config.batch.flush_interval_seconds + + # Kafka Consumer + self._consumer = consumer + # Kafka DLQ Producer + self.dlq_producer = dlq_producer or ConfluentKafkaProducer() + self.dlq_topic = config.kafka.dlq_topic + + # ClickHouse Client + self.clickhouse = clickhouse or ClickHouseService() + + # Internal HTTP Server for metrics & /stats + self.internal_server = internal_server or WorkerInternalServer() + + # Micro-batch buffer: list of (NormalizedRecord, message) + self.buffer: List[Tuple[NormalizedRecord, Any]] = [] + self.last_flush_time = time.time() + self.running = False + + def _init_consumer(self) -> None: + if self._consumer is None: + try: + import confluent_kafka + + conf = { + "bootstrap.servers": config.kafka.bootstrap_servers, + "group.id": config.kafka.consumer_group, + "auto.offset.reset": config.kafka.auto_offset_reset, + "enable.auto.commit": False, # Manual offset commit post ClickHouse ACK + } + self._consumer = confluent_kafka.Consumer(conf) + except ImportError: + raise ImportError( + "confluent-kafka is required. Run 'pip install confluent-kafka'." + ) + + def _handle_signal(self, signum: int, frame: Any) -> None: + sig_name = "SIGTERM" if signum == signal.SIGTERM else "SIGINT" + logger.info( + f"Received {sig_name}, initiating graceful shutdown...", + extra={"event": "shutdown_signal_received"}, + ) + self.running = False + + def flush_buffer(self) -> None: + """ + Flushes buffered normalized records to ClickHouse and commits Kafka offsets. + If ClickHouse write fails all retries, routes batch to DLQ. + """ + if not self.buffer: + self.last_flush_time = time.time() + return + + batch_records = [item[0] for item in self.buffer] + batch_messages = [item[1] for item in self.buffer] + batch_count = len(batch_records) + + start_time = time.time() + try: + self.clickhouse.insert_batch(batch_records) + flush_duration = time.time() - start_time + + # Commit offsets only after ClickHouse ACK + if self._consumer and hasattr(self._consumer, "commit"): + try: + self._consumer.commit(asynchronous=False) + except Exception as e: + logger.error( + f"Kafka offset commit failed after ClickHouse ACK: {e}", + extra={"event": "kafka_commit_failed"}, + ) + + # Record metrics + now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + worker_metrics.set_last_flush_at(now_iso) + worker_metrics.batch_flush_duration.observe(flush_duration) + worker_metrics.batch_size_at_flush.observe(batch_count) + + for rec in batch_records: + worker_metrics.record_processed( + rec.service_type, rec.radio_access_type, 1 + ) + + logger.info( + f"Successfully flushed {batch_count} records to ClickHouse in {flush_duration:.3f}s", + extra={"event": "clickhouse_batch_flushed"}, + ) + + except Exception as e: + logger.error( + f"Batch insert to ClickHouse failed after retries: {e}. Routing batch to DLQ.", + extra={"event": "ch_batch_fatal_error"}, + ) + # Route all records in batch to DLQ to prevent pipeline deadlock + for rec, msg in self.buffer: + route_to_dlq( + producer=self.dlq_producer, + topic=self.dlq_topic, + original_payload=vars(rec), + failure_reason="ch_insert_failed", + failed_field=None, + raw_id=rec.raw_id, + ) + + # Commit offsets to prevent infinite replay stall + if self._consumer and hasattr(self._consumer, "commit"): + try: + self._consumer.commit(asynchronous=False) + except Exception as ce: + logger.error(f"Commit after DLQ fallback failed: {ce}") + + finally: + self.buffer.clear() + worker_metrics.set_in_flight_buffer(0) + self.last_flush_time = time.time() + + def process_single_message(self, message: Any) -> None: + """ + Parses, validates, and either buffers or routes message to DLQ. + """ + value_bytes = message.value() if hasattr(message, "value") else message + if value_bytes is None: + return + + record, failure_reason, failed_field, raw_id = normalize_cdr_payload( + value_bytes + ) + + if failure_reason: + payload_str = ( + value_bytes.decode("utf-8", errors="replace") + if isinstance(value_bytes, bytes) + else str(value_bytes) + ) + route_to_dlq( + producer=self.dlq_producer, + topic=self.dlq_topic, + original_payload=payload_str, + failure_reason=failure_reason, + failed_field=failed_field, + raw_id=raw_id, + ) + # Commit DLQ record offset so it is not reprocessed + if self._consumer and hasattr(self._consumer, "commit"): + try: + self._consumer.commit(message=message, asynchronous=False) + except Exception: + pass + return + + if record: + self.buffer.append((record, message)) + worker_metrics.set_in_flight_buffer(len(self.buffer)) + + # Flush if batch size threshold reached + if len(self.buffer) >= self.batch_size: + self.flush_buffer() + + def start(self) -> None: + """ + Main worker execution loop with signal handling and micro-batching timer. + """ + self._init_consumer() + self.internal_server.start() + + # Register signals for graceful shutdown + signal.signal(signal.SIGTERM, self._handle_signal) + signal.signal(signal.SIGINT, self._handle_signal) + + self._consumer.subscribe([config.kafka.raw_topic]) + self.running = True + logger.info( + f"Processing Worker started. Subscribed to '{config.kafka.raw_topic}'. " + f"Batch size: {self.batch_size}, Interval: {self.flush_interval}s", + extra={"event": "worker_started"}, + ) + + try: + while self.running: + # Poll with short 100ms timeout + msg = self._consumer.poll(timeout=0.1) + + if msg is None: + # Check flush timer on idle / interval + if ( + self.buffer + and (time.time() - self.last_flush_time) >= self.flush_interval + ): + self.flush_buffer() + continue + + if msg.error(): + logger.error( + f"Consumer error: {msg.error()}", + extra={"event": "kafka_consumer_error"}, + ) + continue + + self.process_single_message(msg) + + # Check flush interval during message stream + if ( + self.buffer + and (time.time() - self.last_flush_time) >= self.flush_interval + ): + self.flush_buffer() + + finally: + self.stop() + + def stop(self) -> None: + """ + Graceful shutdown: flush any in-memory records and commit offsets. + """ + logger.info( + f"Stopping worker: flushing {len(self.buffer)} remaining in-memory records...", + extra={"event": "graceful_shutdown_flushing"}, + ) + if self.buffer: + self.flush_buffer() + + if self._consumer and hasattr(self._consumer, "close"): + self._consumer.close() + + self.dlq_producer.close() + self.clickhouse.close() + self.internal_server.stop() + logger.info("Processing Worker stopped cleanly", extra={"event": "worker_stopped"}) + + +if __name__ == "__main__": + worker = ProcessingWorker() + worker.start() diff --git a/app/worker/dlq.py b/app/worker/dlq.py new file mode 100644 index 0000000..1281e6f --- /dev/null +++ b/app/worker/dlq.py @@ -0,0 +1,66 @@ +import json +from datetime import datetime, timezone +from typing import Any, Dict, Optional + +from app.api.producer import MessageProducer +from app.common.config import config +from app.common.logging import setup_logger +from app.worker.metrics import worker_metrics + +logger = setup_logger("processing-worker") + +ALLOWED_FAILURE_REASONS = { + "missing_required_field", + "invalid_rat_value", + "invalid_epoch", + "malformed_json", + "normalization_error", + "ch_insert_failed", +} + + +def build_dlq_envelope( + original_payload: Any, + failure_reason: str, + failed_field: Optional[str] = None, + worker_instance: Optional[str] = None, +) -> Dict[str, Any]: + """ + Constructs a standardized DLQ envelope according to specification. + """ + return { + "original_payload": original_payload, + "failure_reason": failure_reason, + "failed_field": failed_field, + "failed_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "worker_instance": worker_instance or config.worker.instance_id, + } + + +def route_to_dlq( + producer: MessageProducer, + topic: str, + original_payload: Any, + failure_reason: str, + failed_field: Optional[str] = None, + raw_id: Optional[str] = None, +) -> bool: + """ + Wraps unprocessable record in a DLQ envelope, updates DLQ metric, + and publishes to the DLQ topic. + """ + envelope = build_dlq_envelope( + original_payload=original_payload, + failure_reason=failure_reason, + failed_field=failed_field, + ) + envelope_bytes = json.dumps(envelope, default=str).encode("utf-8") + + success = producer.produce(topic=topic, key=raw_id, value=envelope_bytes) + worker_metrics.record_dlq(failure_reason, 1) + + logger.warning( + f"Record routed to DLQ: reason={failure_reason}, field={failed_field}", + extra={"raw_id": raw_id, "event": "routed_to_dlq"}, + ) + return success diff --git a/app/worker/internal_server.py b/app/worker/internal_server.py new file mode 100644 index 0000000..2d5e51a --- /dev/null +++ b/app/worker/internal_server.py @@ -0,0 +1,95 @@ +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Optional + +from app.common.config import config +from app.common.logging import setup_logger +from app.worker.metrics import worker_metrics + +logger = setup_logger("processing-worker") + + +class WorkerHTTPRequestHandler(BaseHTTPRequestHandler): + """ + Lightweight HTTP request handler for worker internal metrics and stats. + """ + + def log_message(self, format: str, *args) -> None: + # Suppress default noisy stderr logging + pass + + def do_GET(self) -> None: + if self.path == "/internal/stats": + stats = worker_metrics.get_internal_stats() + body = json.dumps(stats, indent=2).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + + if self.path == "/metrics": + metrics_text = worker_metrics.get_prometheus_metrics() + body = metrics_text.encode("utf-8") + self.send_response(200) + self.send_header( + "Content-Type", "text/plain; version=0.0.4; charset=utf-8" + ) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + + if self.path in ("/health", "/healthz", "/live"): + body = b'{"status": "ok"}' + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + + self.send_response(404) + self.send_header("Content-Type", "application/json") + body = b'{"error": "not_found"}' + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + +class WorkerInternalServer: + """ + Background HTTP server serving /internal/stats and /metrics on worker port (8001). + """ + + def __init__(self, host: Optional[str] = None, port: Optional[int] = None): + self.host = host or config.worker.stats_host + self.port = port or config.worker.stats_port + self._server: Optional[ThreadingHTTPServer] = None + self._thread: Optional[threading.Thread] = None + + def start(self) -> None: + self._server = ThreadingHTTPServer( + (self.host, self.port), WorkerHTTPRequestHandler + ) + self._thread = threading.Thread( + target=self._server.serve_forever, daemon=True, name="WorkerInternalHTTP" + ) + self._thread.start() + logger.info( + f"Worker internal HTTP server started on http://{self.host}:{self.port}", + extra={"event": "worker_http_started"}, + ) + + def stop(self) -> None: + if self._server: + self._server.shutdown() + self._server.server_close() + if self._thread: + self._thread.join(timeout=2.0) + logger.info( + "Worker internal HTTP server stopped cleanly", + extra={"event": "worker_http_stopped"}, + ) diff --git a/app/worker/metrics.py b/app/worker/metrics.py new file mode 100644 index 0000000..84db66c --- /dev/null +++ b/app/worker/metrics.py @@ -0,0 +1,165 @@ +import threading +from collections import defaultdict +from typing import Any, Dict, List, Optional + + +class SimpleHistogram: + """ + Lightweight, thread-safe Prometheus-compatible histogram. + """ + + def __init__(self, name: str, help_text: str, buckets: List[float]): + self.name = name + self.help_text = help_text + self.buckets = sorted(buckets) + self._counts: Dict[float, int] = {b: 0 for b in self.buckets} + self._count = 0 + self._sum = 0.0 + self._lock = threading.Lock() + + def observe(self, value: float) -> None: + with self._lock: + self._count += 1 + self._sum += value + for b in self.buckets: + if value <= b: + self._counts[b] += 1 + + def to_prometheus(self) -> str: + lines = [ + f"# HELP {self.name} {self.help_text}", + f"# TYPE {self.name} histogram", + ] + with self._lock: + for b in self.buckets: + lines.append(f'{self.name}_bucket{{le="{b}"}} {self._counts[b]}') + lines.append(f'{self.name}_bucket{{le="+Inf"}} {self._count}') + lines.append(f"{self.name}_count {self._count}") + lines.append(f"{self.name}_sum {self._sum:.4f}") + return "\n".join(lines) + "\n" + + +class WorkerMetrics: + """ + Thread-safe operational metrics collector for the Processing Worker. + Provides data for /metrics (Prometheus) and /internal/stats (Ingestion API aggregator). + """ + + def __init__(self): + self._lock = threading.Lock() + self._processed_total: Dict[tuple, int] = defaultdict(int) + self._dlq_total: Dict[str, int] = defaultdict(int) + self._in_flight_buffer_size: int = 0 + self._last_flush_at: Optional[str] = None + self._consumer_lag: Dict[str, int] = defaultdict(int) + + # Histograms + self.kafka_to_ch_latency = SimpleHistogram( + name="cdr_kafka_to_clickhouse_latency_seconds", + help_text="End-to-end latency from Kafka produce to ClickHouse write", + buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0], + ) + self.batch_flush_duration = SimpleHistogram( + name="cdr_batch_flush_duration_seconds", + help_text="ClickHouse batch flush duration in seconds", + buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.0], + ) + self.batch_size_at_flush = SimpleHistogram( + name="cdr_batch_size_at_flush", + help_text="Batch sizes at flush time", + buckets=[10, 50, 100, 250, 500, 1000], + ) + + def record_processed( + self, service_type: str, radio_access_type: str, count: int = 1 + ) -> None: + with self._lock: + self._processed_total[(service_type, radio_access_type)] += count + + def record_dlq(self, reason: str, count: int = 1) -> None: + with self._lock: + self._dlq_total[reason] += count + + def set_in_flight_buffer(self, size: int) -> None: + with self._lock: + self._in_flight_buffer_size = size + + def set_last_flush_at(self, timestamp_iso: str) -> None: + with self._lock: + self._last_flush_at = timestamp_iso + + def set_consumer_lag(self, partition: int, lag: int) -> None: + with self._lock: + self._consumer_lag[str(partition)] = lag + + def get_internal_stats(self) -> Dict[str, Any]: + """ + Returns structured dictionary for GET /internal/stats. + """ + with self._lock: + total_processed = sum(self._processed_total.values()) + total_dlq = sum(self._dlq_total.values()) + in_flight = self._in_flight_buffer_size + last_flush = self._last_flush_at + + # Breakdowns + by_service: Dict[str, int] = defaultdict(int) + by_rat: Dict[str, int] = defaultdict(int) + for (svc, rat), cnt in self._processed_total.items(): + by_service[svc] += cnt + by_rat[rat] += cnt + + dlq_reasons = dict(self._dlq_total) + + return { + "pipeline": { + "total_processed": total_processed, + "total_dlq": total_dlq, + "in_flight_buffer": in_flight, + "last_flush_at": last_flush, + }, + "breakdown": { + "by_service_type": dict(by_service), + "by_radio_access_type": dict(by_rat), + "dlq_by_reason": dlq_reasons, + }, + } + + def get_prometheus_metrics(self) -> str: + """ + Formats all worker metrics into standard Prometheus exposition format. + """ + lines = [] + + # 1. cdr_processed_total + lines.append("# HELP cdr_processed_total Total CDRs successfully written to ClickHouse") + lines.append("# TYPE cdr_processed_total counter") + with self._lock: + for (svc, rat), count in self._processed_total.items(): + lines.append(f'cdr_processed_total{{service_type="{svc}",radio_access_type="{rat}"}} {count}') + + # 2. cdr_dlq_total + lines.append("# HELP cdr_dlq_total Total CDRs routed to DLQ by reason") + lines.append("# TYPE cdr_dlq_total counter") + for reason, count in self._dlq_total.items(): + lines.append(f'cdr_dlq_total{{reason="{reason}"}} {count}') + + # 3. Gauges + lines.append("# HELP cdr_in_flight_buffer_size Current in-memory buffer size") + lines.append("# TYPE cdr_in_flight_buffer_size gauge") + lines.append(f"cdr_in_flight_buffer_size {self._in_flight_buffer_size}") + + lines.append("# HELP cdr_kafka_consumer_lag Kafka consumer lag per partition") + lines.append("# TYPE cdr_kafka_consumer_lag gauge") + for part, lag in self._consumer_lag.items(): + lines.append(f'cdr_kafka_consumer_lag{{partition="{part}"}} {lag}') + + # 4. Histograms + lines.append(self.kafka_to_ch_latency.to_prometheus().strip()) + lines.append(self.batch_flush_duration.to_prometheus().strip()) + lines.append(self.batch_size_at_flush.to_prometheus().strip()) + + return "\n".join(lines) + "\n" + + +worker_metrics = WorkerMetrics() diff --git a/app/worker/normalizer.py b/app/worker/normalizer.py new file mode 100644 index 0000000..f6842dc --- /dev/null +++ b/app/worker/normalizer.py @@ -0,0 +1,200 @@ +import json +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Dict, Optional, Tuple + +import phonenumbers + +RAT_MAP = { + 1: "3G", + 6: "LTE_4G", + 9: "5G_NR", +} + +REQUIRED_FIELDS = [ + "raw_id", + "device_id", + "caller_num", + "svc", + "rat", + "epoch_start", + "epoch_end", + "up_bytes", + "down_bytes", + "lac", + "ci", +] + + +@dataclass(frozen=True) +class NormalizedRecord: + cdr_id: str + raw_id: str + msisdn: str + imsi: str + service_type: str + radio_access_type: str + cell_id: str + bytes_uploaded: int + bytes_downloaded: int + total_bytes: int + duration_seconds: int + session_start_time: str + session_end_time: str + source_ip: str + + def to_row(self) -> Tuple[Any, ...]: + """Returns ordered tuple matching ClickHouse cdr_records schema columns.""" + return ( + uuid.UUID(self.cdr_id), + self.raw_id, + self.msisdn, + self.imsi, + self.service_type, + self.radio_access_type, + self.cell_id, + self.bytes_uploaded, + self.bytes_downloaded, + self.total_bytes, + self.duration_seconds, + datetime.strptime(self.session_start_time, "%Y-%m-%d %H:%M:%S"), + datetime.strptime(self.session_end_time, "%Y-%m-%d %H:%M:%S"), + self.source_ip, + ) + + +def normalize_phone_number(raw_num: str) -> Optional[str]: + """ + Normalizes telephone number into E.164 format: + - Replaces leading '00' or '011' with '+' + - Preserves existing '+' + - Validates with phonenumbers library + """ + if not isinstance(raw_num, str): + return None + + cleaned = raw_num.strip() + if cleaned.startswith("00"): + cleaned = "+" + cleaned[2:] + elif cleaned.startswith("011"): + cleaned = "+" + cleaned[3:] + elif not cleaned.startswith("+"): + cleaned = "+" + cleaned + + try: + parsed = phonenumbers.parse(cleaned, None) + if not ( + phonenumbers.is_valid_number(parsed) + or phonenumbers.is_possible_number(parsed) + ): + return None + return phonenumbers.format_number( + parsed, phonenumbers.PhoneNumberFormat.E164 + ) + except Exception: + return None + + +def normalize_cdr_payload( + raw_payload_bytes: bytes, +) -> Tuple[Optional[NormalizedRecord], Optional[str], Optional[str], Optional[str]]: + """ + Validates and normalizes raw CDR payload. + Returns: + (record, failure_reason, failed_field, raw_id) + If record is None, failure_reason describes why it should be routed to DLQ. + """ + # 1. Parse JSON + try: + if isinstance(raw_payload_bytes, bytes): + payload_str = raw_payload_bytes.decode("utf-8") + else: + payload_str = str(raw_payload_bytes) + payload: Dict[str, Any] = json.loads(payload_str) + except Exception: + return None, "malformed_json", None, None + + if not isinstance(payload, dict): + return None, "malformed_json", None, None + + raw_id = payload.get("raw_id") + raw_id_str = str(raw_id) if raw_id is not None else None + + # 2. Check required fields + for field in REQUIRED_FIELDS: + if field not in payload or payload[field] is None: + return None, "missing_required_field", field, raw_id_str + + # 3. Check timestamps (epoch_start <= epoch_end) + try: + epoch_start = int(payload["epoch_start"]) + epoch_end = int(payload["epoch_end"]) + if epoch_start > epoch_end: + return None, "invalid_epoch", "epoch_start", raw_id_str + except (ValueError, TypeError): + return None, "invalid_epoch", "epoch_start", raw_id_str + + # 4. Check bytes (must be >= 0) + try: + up_bytes = int(payload["up_bytes"]) + down_bytes = int(payload["down_bytes"]) + if up_bytes < 0: + return None, "normalization_error", "up_bytes", raw_id_str + if down_bytes < 0: + return None, "normalization_error", "down_bytes", raw_id_str + except (ValueError, TypeError): + return None, "normalization_error", "bytes", raw_id_str + + # 5. Radio Access Type mapping + rat_raw = payload.get("rat") + try: + rat_int = int(rat_raw) + if rat_int not in RAT_MAP: + return None, "invalid_rat_value", "rat", raw_id_str + radio_access_type = RAT_MAP[rat_int] + except (ValueError, TypeError): + return None, "invalid_rat_value", "rat", raw_id_str + + # 6. E.164 phone normalization + caller_num = str(payload["caller_num"]) + msisdn = normalize_phone_number(caller_num) + if not msisdn: + return None, "normalization_error", "caller_num", raw_id_str + + # 7. Field mappings and derivations + imsi = str(payload["device_id"]) + service_type = str(payload["svc"]) + cell_id = f"{payload['lac']}_{payload['ci']}" + total_bytes = up_bytes + down_bytes + duration_seconds = epoch_end - epoch_start + + start_dt = datetime.fromtimestamp(epoch_start, tz=timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S" + ) + end_dt = datetime.fromtimestamp(epoch_end, tz=timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S" + ) + + src_ip = str(payload.get("src_ip") or "0.0.0.0") + + # Deterministic UUIDv5 cdr_id from raw_id + cdr_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, raw_id_str)) + + record = NormalizedRecord( + cdr_id=cdr_id, + raw_id=raw_id_str, + msisdn=msisdn, + imsi=imsi, + service_type=service_type, + radio_access_type=radio_access_type, + cell_id=cell_id, + bytes_uploaded=up_bytes, + bytes_downloaded=down_bytes, + total_bytes=total_bytes, + duration_seconds=duration_seconds, + session_start_time=start_dt, + session_end_time=end_dt, + source_ip=src_ip, + ) + return record, None, None, raw_id_str diff --git a/openspec/changes/telecom-cdr-pipeline/tasks.md b/openspec/changes/telecom-cdr-pipeline/tasks.md index 5dc6d98..1c6bcb0 100644 --- a/openspec/changes/telecom-cdr-pipeline/tasks.md +++ b/openspec/changes/telecom-cdr-pipeline/tasks.md @@ -14,13 +14,13 @@ ## 3. Processing Worker Implementation -- [ ] 3.1 Implement Kafka consumer loop with `confluent-kafka` consuming from `cdr.raw` and DLQ producer publishing to `cdr.raw.dlq`. -- [ ] 3.2 Implement safe parser routing malformed records to DLQ with structured envelope and reason taxonomy (`missing_required_field`, `invalid_rat_value`, `invalid_epoch`, `malformed_json`, `normalization_error`). -- [ ] 3.3 Implement schema normalizer: E.164 phone conversion via `phonenumbers`, RAT mapping (`1`→`3G`, `6`→`LTE_4G`, `9`→`5G_NR`), `cell_id` derivation (`lac_ci`), and deterministic UUIDv5 `cdr_id` generation. -- [ ] 3.4 Implement micro-batching buffer flushing on 1,000 records or 2 seconds elapsed, batch-inserting into ClickHouse via `clickhouse-connect`, and committing offsets only after ClickHouse ACK. -- [ ] 3.5 Implement exponential backoff retry for ClickHouse write timeouts (`CH_RETRY_MAX_ATTEMPTS`, `CH_RETRY_INITIAL_DELAY_MS`, `CH_RETRY_BACKOFF_FACTOR`) with fallback to DLQ. -- [ ] 3.6 Implement worker internal HTTP stats endpoint (`GET /internal/stats`) and worker Prometheus metrics (`cdr_processed_total`, `cdr_dlq_total`, histograms, gauges). -- [ ] 3.7 Implement `SIGTERM` and `SIGINT` signal handlers to stop consuming, flush remaining in-memory buffer to ClickHouse, commit offsets, and exit cleanly without data loss. +- [x] 3.1 Implement Kafka consumer loop with `confluent-kafka` consuming from `cdr.raw` and DLQ producer publishing to `cdr.raw.dlq`. +- [x] 3.2 Implement safe parser routing malformed records to DLQ with structured envelope and reason taxonomy (`missing_required_field`, `invalid_rat_value`, `invalid_epoch`, `malformed_json`, `normalization_error`). +- [x] 3.3 Implement schema normalizer: E.164 phone conversion via `phonenumbers`, RAT mapping (`1`→`3G`, `6`→`LTE_4G`, `9`→`5G_NR`), `cell_id` derivation (`lac_ci`), and deterministic UUIDv5 `cdr_id` generation. +- [x] 3.4 Implement micro-batching buffer flushing on 1,000 records or 2 seconds elapsed, batch-inserting into ClickHouse via `clickhouse-connect`, and committing offsets only after ClickHouse ACK. +- [x] 3.5 Implement exponential backoff retry for ClickHouse write timeouts (`CH_RETRY_MAX_ATTEMPTS`, `CH_RETRY_INITIAL_DELAY_MS`, `CH_RETRY_BACKOFF_FACTOR`) with fallback to DLQ. +- [x] 3.6 Implement worker internal HTTP stats endpoint (`GET /internal/stats`) and worker Prometheus metrics (`cdr_processed_total`, `cdr_dlq_total`, histograms, gauges). +- [x] 3.7 Implement `SIGTERM` and `SIGINT` signal handlers to stop consuming, flush remaining in-memory buffer to ClickHouse, commit offsets, and exit cleanly without data loss. ## 4. Containerization & Orchestration diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..65140f2 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# tests package diff --git a/tests/test_worker.py b/tests/test_worker.py new file mode 100644 index 0000000..09e94af --- /dev/null +++ b/tests/test_worker.py @@ -0,0 +1,287 @@ +import json +import time +import unittest +import urllib.request + +from app.api.producer import MockMessageProducer +from app.worker.clickhouse import MockClickHouseClient +from app.worker.consumer import ProcessingWorker +from app.worker.internal_server import WorkerInternalServer +from app.worker.metrics import worker_metrics +from app.worker.normalizer import ( + NormalizedRecord, + normalize_cdr_payload, + normalize_phone_number, +) + + +class MockKafkaMessage: + def __init__(self, value_bytes: bytes, key: str = "k1", partition: int = 0, offset: int = 0): + self._value = value_bytes + self._key = key + self._partition = partition + self._offset = offset + + def value(self): + return self._value + + def key(self): + return self._key.encode("utf-8") if self._key else None + + def partition(self): + return self._partition + + def offset(self): + return self._offset + + def error(self): + return None + + +class MockKafkaConsumer: + def __init__(self): + self.committed = False + self.closed = False + + def commit(self, message=None, asynchronous=False): + self.committed = True + + def close(self): + self.closed = True + + +class TestWorkerNormalizer(unittest.TestCase): + def test_phone_normalization(self): + # 00 prefix + self.assertEqual(normalize_phone_number("0019875551234"), "+19875551234") + # 011 prefix + self.assertEqual(normalize_phone_number("01119875551234"), "+19875551234") + # existing + prefix + self.assertEqual(normalize_phone_number("+19875551234"), "+19875551234") + # invalid + self.assertIsNone(normalize_phone_number("invalid-phone")) + + def test_valid_payload_normalization(self): + payload = { + "raw_id": "raw-test-100", + "src_ip": "10.240.12.1", + "device_id": "310410998877665", + "caller_num": "0019875551234", + "svc": "4G_DATA", + "up_bytes": 1000, + "down_bytes": 2000, + "rat": 6, + "lac": "4021", + "ci": "8812", + "epoch_start": 1726070000, + "epoch_end": 1726070300, + } + data = json.dumps(payload).encode("utf-8") + rec, failure, failed_field, raw_id = normalize_cdr_payload(data) + + self.assertIsNotNone(rec) + self.assertIsNone(failure) + self.assertEqual(rec.raw_id, "raw-test-100") + self.assertEqual(rec.msisdn, "+19875551234") + self.assertEqual(rec.radio_access_type, "LTE_4G") + self.assertEqual(rec.cell_id, "4021_8812") + self.assertEqual(rec.total_bytes, 3000) + self.assertEqual(rec.duration_seconds, 300) + self.assertEqual(rec.source_ip, "10.240.12.1") + self.assertTrue(len(rec.cdr_id) > 0) + + def test_invalid_rat_routing(self): + payload = { + "raw_id": "raw-test-101", + "device_id": "123", + "caller_num": "+19875551234", + "svc": "4G_DATA", + "up_bytes": 10, + "down_bytes": 20, + "rat": 99, # Unsupported RAT + "lac": "1", + "ci": "2", + "epoch_start": 100, + "epoch_end": 200, + } + rec, failure, failed_field, _ = normalize_cdr_payload(json.dumps(payload).encode("utf-8")) + self.assertIsNone(rec) + self.assertEqual(failure, "invalid_rat_value") + self.assertEqual(failed_field, "rat") + + def test_invalid_epoch_routing(self): + payload = { + "raw_id": "raw-test-102", + "device_id": "123", + "caller_num": "+19875551234", + "svc": "4G_DATA", + "up_bytes": 10, + "down_bytes": 20, + "rat": 6, + "lac": "1", + "ci": "2", + "epoch_start": 500, # epoch_start > epoch_end + "epoch_end": 200, + } + rec, failure, failed_field, _ = normalize_cdr_payload(json.dumps(payload).encode("utf-8")) + self.assertIsNone(rec) + self.assertEqual(failure, "invalid_epoch") + self.assertEqual(failed_field, "epoch_start") + + def test_negative_bytes_routing(self): + payload = { + "raw_id": "raw-test-103", + "device_id": "123", + "caller_num": "+19875551234", + "svc": "4G_DATA", + "up_bytes": -10, # Negative bytes + "down_bytes": 20, + "rat": 6, + "lac": "1", + "ci": "2", + "epoch_start": 100, + "epoch_end": 200, + } + rec, failure, failed_field, _ = normalize_cdr_payload(json.dumps(payload).encode("utf-8")) + self.assertIsNone(rec) + self.assertEqual(failure, "normalization_error") + self.assertEqual(failed_field, "up_bytes") + + def test_missing_required_field_routing(self): + payload = { + "raw_id": "raw-test-104", + "device_id": "123", + # caller_num missing + "svc": "4G_DATA", + "up_bytes": 10, + "down_bytes": 20, + "rat": 6, + "lac": "1", + "ci": "2", + "epoch_start": 100, + "epoch_end": 200, + } + rec, failure, failed_field, _ = normalize_cdr_payload(json.dumps(payload).encode("utf-8")) + self.assertIsNone(rec) + self.assertEqual(failure, "missing_required_field") + self.assertEqual(failed_field, "caller_num") + + +class TestProcessingWorker(unittest.TestCase): + def setUp(self): + self.consumer = MockKafkaConsumer() + self.dlq_producer = MockMessageProducer() + self.clickhouse = MockClickHouseClient() + self.internal_server = WorkerInternalServer(host="127.0.0.1", port=18001) + self.internal_server.start() + + self.worker = ProcessingWorker( + consumer=self.consumer, + dlq_producer=self.dlq_producer, + clickhouse=self.clickhouse, + batch_size=2, # Flush on 2 records + flush_interval=0.5, + internal_server=self.internal_server, + ) + + def tearDown(self): + self.internal_server.stop() + + def test_micro_batch_flush_on_count(self): + payload1 = { + "raw_id": "raw-1", + "device_id": "1", + "caller_num": "+19875551234", + "svc": "4G_DATA", + "up_bytes": 10, + "down_bytes": 20, + "rat": 6, + "lac": "1", + "ci": "2", + "epoch_start": 100, + "epoch_end": 200, + } + payload2 = { + "raw_id": "raw-2", + "device_id": "2", + "caller_num": "+19875551234", + "svc": "4G_DATA", + "up_bytes": 30, + "down_bytes": 40, + "rat": 6, + "lac": "1", + "ci": "2", + "epoch_start": 100, + "epoch_end": 200, + } + msg1 = MockKafkaMessage(json.dumps(payload1).encode("utf-8")) + msg2 = MockKafkaMessage(json.dumps(payload2).encode("utf-8")) + + self.worker.process_single_message(msg1) + self.assertEqual(len(self.worker.buffer), 1) + self.assertEqual(len(self.clickhouse.rows), 0) + + # Second message reaches batch_size=2 -> flushes to ClickHouse + self.worker.process_single_message(msg2) + self.assertEqual(len(self.worker.buffer), 0) + self.assertEqual(len(self.clickhouse.rows), 2) + self.assertTrue(self.consumer.committed) + + def test_dlq_routing_for_malformed_message(self): + bad_msg = MockKafkaMessage(b"{not json") + self.worker.process_single_message(bad_msg) + + self.assertEqual(len(self.dlq_producer.messages), 1) + dlq_data = json.loads(self.dlq_producer.messages[0]["value"].decode("utf-8")) + self.assertEqual(dlq_data["failure_reason"], "malformed_json") + + def test_clickhouse_retry_and_dlq_fallback(self): + # ClickHouse configured to fail 3 times (exceeding max_attempts=3) + ch_failing = MockClickHouseClient(fail_attempts=5) + worker = ProcessingWorker( + consumer=self.consumer, + dlq_producer=self.dlq_producer, + clickhouse=ch_failing, + batch_size=1, + flush_interval=1.0, + internal_server=self.internal_server, + ) + payload = { + "raw_id": "raw-ch-fail", + "device_id": "1", + "caller_num": "+19875551234", + "svc": "4G_DATA", + "up_bytes": 10, + "down_bytes": 20, + "rat": 6, + "lac": "1", + "ci": "2", + "epoch_start": 100, + "epoch_end": 200, + } + msg = MockKafkaMessage(json.dumps(payload).encode("utf-8")) + worker.process_single_message(msg) + + # Record should have fallen back to DLQ with ch_insert_failed + self.assertEqual(len(self.dlq_producer.messages), 1) + dlq_data = json.loads(self.dlq_producer.messages[0]["value"].decode("utf-8")) + self.assertEqual(dlq_data["failure_reason"], "ch_insert_failed") + + def test_internal_stats_and_metrics_endpoints(self): + # Verify /internal/stats + with urllib.request.urlopen("http://127.0.0.1:18001/internal/stats") as resp: + self.assertEqual(resp.status, 200) + data = json.loads(resp.read().decode("utf-8")) + self.assertIn("pipeline", data) + self.assertIn("breakdown", data) + + # Verify /metrics + with urllib.request.urlopen("http://127.0.0.1:18001/metrics") as resp: + self.assertEqual(resp.status, 200) + text = resp.read().decode("utf-8") + self.assertIn("cdr_processed_total", text) + self.assertIn("cdr_dlq_total", text) + + +if __name__ == "__main__": + unittest.main()