267 lines
9.6 KiB
Python
267 lines
9.6 KiB
Python
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()
|