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()