diff --git a/.env.sample b/.env.sample index f2b352e..4860c01 100644 --- a/.env.sample +++ b/.env.sample @@ -23,6 +23,18 @@ KAFKA_DLQ_TOPIC_PARTITIONS=1 KAFKA_DLQ_TOPIC_REPLICATION_FACTOR=1 KAFKA_DLQ_TOPIC_RETENTION_MS=604800000 + +# ========================= +# Processing Worker +# ========================= + +WORKER_INSTANCE_ID=worker-1 + +KAFKA_AUTO_OFFSET_RESET=earliest +KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS=1 + +KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS=5 + KAFKA_CONSUMER_GROUP=cdr-processing-workers diff --git a/README.md b/README.md index 1407419..558d142 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,12 @@ Validation Bounded Async Queue ↓ Redpanda / cdr.raw + ↓ +Processing Worker + ↓ +Validation + Normalization + ├── valid → processed by worker + └── invalid → cdr.raw.dlq ``` Infrastructure already available: @@ -60,34 +66,67 @@ worker processing is not connected yet. * 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 ## Project Structure ```text -src/core/ -├── config.py +telecom-cdr-pipeline/ +├── infrastructure/ +│ └── clickhouse/ +│ └── init.sql │ -├── api/ -│ ├── __init__.py -│ ├── kafka_publisher.py -│ ├── main.py -│ ├── publisher.py -│ └── routes.py +├── scripts/ +│ ├── sample-valid-cdr.json +│ └── sample-invalid-rat.json │ -└── domain/ - ├── __init__.py - ├── errors.py - ├── models.py - ├── normalization.py - └── validation.py - -infrastructure/ -└── clickhouse/ - └── init.sql - -tests/ -├── integration/ -└── unit/ +├── src/ +│ └── core/ +│ ├── __init__.py +│ ├── config.py +│ │ +│ ├── api/ +│ │ ├── __init__.py +│ │ ├── kafka_publisher.py +│ │ ├── main.py +│ │ ├── publisher.py +│ │ └── routes.py +│ │ +│ ├── domain/ +│ │ ├── __init__.py +│ │ ├── errors.py +│ │ ├── models.py +│ │ ├── normalization.py +│ │ └── validation.py +│ │ +│ └── worker/ +│ ├── __init__.py +│ ├── consumer.py +│ ├── dlq.py +│ ├── main.py +│ └── processor.py +│ +├── tests/ +│ ├── integration/ +│ │ └── test_ingestion_api.py +│ │ +│ └── unit/ +│ ├── test_dlq.py +│ ├── test_normalization.py +│ ├── test_project_setup.py +│ ├── test_validation.py +│ └── test_worker_processor.py +│ +├── .env.example +├── .gitignore +├── docker-compose.yml +├── pyproject.toml +└── README.md ``` ## Configuration diff --git a/src/core/config.py b/src/core/config.py index 795aa6b..db82f95 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -22,6 +22,14 @@ class Settings(BaseSettings): kafka_batch_size: int = 131_072 kafka_producer_max_messages: int = 100_000 + kafka_dlq_topic: str = "cdr.raw.dlq" + kafka_consumer_group: str = "cdr-processing-workers" + kafka_auto_offset_reset: str = "earliest" + + kafka_consumer_poll_timeout_seconds: float = 1.0 + kafka_dlq_delivery_timeout_seconds: float = 5.0 + + worker_instance_id: str = "worker-1" @lru_cache def get_settings() -> Settings: diff --git a/src/core/worker/__init__.py b/src/core/worker/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/core/worker/consumer.py b/src/core/worker/consumer.py new file mode 100644 index 0000000..b00b82c --- /dev/null +++ b/src/core/worker/consumer.py @@ -0,0 +1,215 @@ +import logging +from typing import Any + +from confluent_kafka import Consumer, KafkaException, Message + +from core.config import Settings +from core.domain.errors import ( + CDRProcessingError, + FailureReason, +) +from core.worker.dlq import ( + DLQPublishError, + DLQPublisher, +) +from core.worker.processor import ( + decode_payload, + process_payload, +) + + +logger = logging.getLogger(__name__) + + +class ProcessingWorker: + def __init__( + self, + settings: Settings, + ) -> None: + self._settings = settings + + self._consumer = Consumer( + { + "bootstrap.servers": + settings.kafka_bootstrap_servers, + + "group.id": + settings.kafka_consumer_group, + + "client.id": + settings.worker_instance_id, + + "auto.offset.reset": + settings.kafka_auto_offset_reset, + + "enable.auto.commit": False, + } + ) + + self._dlq = DLQPublisher(settings) + + self._running = False + + def run(self) -> None: + self._running = True + + self._consumer.subscribe( + [self._settings.kafka_raw_topic] + ) + + logger.info( + "Worker started. topic=%s group=%s instance=%s", + self._settings.kafka_raw_topic, + self._settings.kafka_consumer_group, + self._settings.worker_instance_id, + ) + + try: + while self._running: + message = self._consumer.poll( + self._settings.kafka_consumer_poll_timeout_seconds + ) + + if message is None: + continue + + if message.error(): + logger.error( + "Kafka consumer error: %s", + message.error(), + ) + continue + + self._handle_message(message) + + finally: + self._shutdown() + + def _handle_message( + self, + message: Message, + ) -> None: + value = message.value() or b"" + + original_payload: Any = { + "raw": value.decode( + "utf-8", + errors="replace", + ) + } + + try: + payload = decode_payload(value) + + original_payload = payload + + normalized = process_payload(payload) + + except CDRProcessingError as exc: + self._route_to_dlq( + message=message, + original_payload=original_payload, + error=exc, + ) + + return + + except Exception as exc: + logger.exception( + "Unexpected CDR processing error." + ) + + error = CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail=f"Unexpected processing error: {exc}", + ) + + self._route_to_dlq( + message=message, + original_payload=original_payload, + error=error, + ) + + return + + logger.info( + ( + "Normalized CDR. " + "raw_id=%s cdr_id=%s " + "service_type=%s rat=%s" + ), + normalized.raw_id, + normalized.cdr_id, + normalized.service_type, + normalized.radio_access_type, + ) + + # Temporary until ClickHouse batching is implemented. + self._commit(message) + + def _route_to_dlq( + self, + message: Message, + original_payload: Any, + error: CDRProcessingError, + ) -> None: + logger.warning( + ( + "Routing CDR to DLQ. " + "reason=%s field=%s " + "partition=%s offset=%s" + ), + error.reason.value, + error.failed_field, + message.partition(), + message.offset(), + ) + + try: + self._dlq.publish( + original_payload=original_payload, + error=error, + ) + + except DLQPublishError: + logger.exception( + "DLQ publishing failed. " + "Original Kafka offset will not be committed." + ) + + raise + + self._commit(message) + + def _commit( + self, + message: Message, + ) -> None: + committed = self._consumer.commit( + message=message, + asynchronous=False, + ) + + for partition in committed or []: + if partition.error is not None: + raise KafkaException( + partition.error + ) + + logger.debug( + "Committed Kafka offset. partition=%s offset=%s", + message.partition(), + message.offset(), + ) + + def stop(self) -> None: + self._running = False + + def _shutdown(self) -> None: + logger.info("Worker shutting down.") + + try: + self._dlq.close() + + finally: + self._consumer.close() diff --git a/src/core/worker/dlq.py b/src/core/worker/dlq.py new file mode 100644 index 0000000..dc093c5 --- /dev/null +++ b/src/core/worker/dlq.py @@ -0,0 +1,124 @@ +import json +import time +from datetime import datetime, timezone +from typing import Any + +from confluent_kafka import KafkaException, Producer + +from core.config import Settings +from core.domain.errors import CDRProcessingError + + +class DLQPublishError(Exception): + pass + + +def build_dlq_envelope( + original_payload: Any, + error: CDRProcessingError, + worker_instance: str, +) -> dict[str, Any]: + return { + "original_payload": original_payload, + "failure_reason": error.reason.value, + "failed_field": error.failed_field, + "failed_at": ( + datetime.now(timezone.utc) + .isoformat() + .replace("+00:00", "Z") + ), + "worker_instance": worker_instance, + } + + +class DLQPublisher: + def __init__( + self, + settings: Settings, + ) -> None: + self._settings = settings + + self._producer = Producer( + { + "bootstrap.servers": + settings.kafka_bootstrap_servers, + + "client.id": + f"{settings.worker_instance_id}-dlq", + + "acks": "all", + "enable.idempotence": True, + + "compression.type": "lz4", + + "message.timeout.ms": + settings.kafka_message_timeout_ms, + } + ) + + def publish( + self, + original_payload: Any, + error: CDRProcessingError, + ) -> None: + envelope = build_dlq_envelope( + original_payload=original_payload, + error=error, + worker_instance=self._settings.worker_instance_id, + ) + + value = json.dumps( + envelope, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + + delivered = False + delivery_error = None + + def on_delivery(err, message) -> None: + nonlocal delivered + nonlocal delivery_error + + delivery_error = err + delivered = True + + try: + self._producer.produce( + topic=self._settings.kafka_dlq_topic, + value=value, + on_delivery=on_delivery, + ) + + except (BufferError, KafkaException) as exc: + raise DLQPublishError( + f"Unable to queue DLQ message: {exc}" + ) from exc + + deadline = ( + time.monotonic() + + self._settings.kafka_dlq_delivery_timeout_seconds + ) + + while not delivered: + self._producer.poll(0.1) + + if time.monotonic() >= deadline: + raise DLQPublishError( + "DLQ Kafka delivery timed out." + ) + + if delivery_error is not None: + raise DLQPublishError( + f"DLQ delivery failed: {delivery_error}" + ) + + def close(self) -> None: + remaining = self._producer.flush( + self._settings.kafka_dlq_delivery_timeout_seconds + ) + + if remaining > 0: + raise DLQPublishError( + f"{remaining} DLQ message(s) were not delivered." + ) diff --git a/src/core/worker/main.py b/src/core/worker/main.py new file mode 100644 index 0000000..7a31f73 --- /dev/null +++ b/src/core/worker/main.py @@ -0,0 +1,51 @@ +import logging +import signal + +from core.config import get_settings +from core.worker.consumer import ProcessingWorker + + +logging.basicConfig( + level=logging.INFO, + format=( + "%(asctime)s " + "%(levelname)s " + "%(name)s " + "%(message)s" + ), +) + + +def main() -> None: + settings = get_settings() + + worker = ProcessingWorker( + settings=settings, + ) + + def handle_shutdown( + signum, + frame, + ) -> None: + logging.getLogger(__name__).info( + "Shutdown signal received: %s", + signum, + ) + + worker.stop() + + signal.signal( + signal.SIGINT, + handle_shutdown, + ) + + signal.signal( + signal.SIGTERM, + handle_shutdown, + ) + + worker.run() + + +if __name__ == "__main__": + main() diff --git a/src/core/worker/processor.py b/src/core/worker/processor.py new file mode 100644 index 0000000..48d73ae --- /dev/null +++ b/src/core/worker/processor.py @@ -0,0 +1,38 @@ +import json +from typing import Any + +from core.domain.errors import ( + CDRProcessingError, + FailureReason, +) +from core.domain.models import NormalizedCDR +from core.domain.normalization import normalize_cdr +from core.domain.validation import parse_raw_cdr + + +def decode_payload(value: bytes) -> dict[str, Any]: + try: + text = value.decode("utf-8") + payload = json.loads(text) + + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise CDRProcessingError( + reason=FailureReason.MALFORMED_JSON, + detail=f"Unable to decode Kafka message as JSON: {exc}", + ) from exc + + if not isinstance(payload, dict): + raise CDRProcessingError( + reason=FailureReason.NORMALIZATION_ERROR, + detail="Kafka JSON payload must be an object.", + ) + + return payload + + +def process_payload( + payload: dict[str, Any], +) -> NormalizedCDR: + raw_cdr = parse_raw_cdr(payload) + + return normalize_cdr(raw_cdr) diff --git a/tests/unit/test_dlq.py b/tests/unit/test_dlq.py new file mode 100644 index 0000000..e6e00c0 --- /dev/null +++ b/tests/unit/test_dlq.py @@ -0,0 +1,36 @@ +from core.domain.errors import ( + CDRProcessingError, + FailureReason, +) +from core.worker.dlq import build_dlq_envelope + + +def test_build_dlq_envelope() -> None: + original = { + "raw_id": "raw-invalid-001", + "rat": 99, + } + + error = CDRProcessingError( + reason=FailureReason.INVALID_RAT_VALUE, + detail="Unsupported RAT value: 99.", + failed_field="rat", + ) + + envelope = build_dlq_envelope( + original_payload=original, + error=error, + worker_instance="worker-test", + ) + + assert envelope["original_payload"] == original + + assert ( + envelope["failure_reason"] + == "invalid_rat_value" + ) + + assert envelope["failed_field"] == "rat" + assert envelope["worker_instance"] == "worker-test" + + assert envelope["failed_at"].endswith("Z") diff --git a/tests/unit/test_worker_processor.py b/tests/unit/test_worker_processor.py new file mode 100644 index 0000000..59f1ea3 --- /dev/null +++ b/tests/unit/test_worker_processor.py @@ -0,0 +1,78 @@ +import json + +import pytest + +from core.domain.errors import ( + CDRProcessingError, + FailureReason, +) +from core.worker.processor import ( + decode_payload, + process_payload, +) + + +def valid_payload() -> dict: + return { + "raw_id": "raw-worker-001", + "src_ip": "10.240.12.1", + "device_id": "310410998877665", + "caller_num": "0019875551234", + "svc": "4G_DATA", + "up_bytes": 100, + "down_bytes": 200, + "rat": 6, + "lac": "4021", + "ci": "8812", + "epoch_start": 1789900000, + "epoch_end": 1789900300, + } + + +def test_decode_valid_json_object() -> None: + value = json.dumps( + valid_payload() + ).encode("utf-8") + + payload = decode_payload(value) + + assert payload == valid_payload() + + +def test_malformed_json() -> None: + with pytest.raises( + CDRProcessingError + ) as exc_info: + decode_payload( + b'{"raw_id": invalid}' + ) + + assert ( + exc_info.value.reason + == FailureReason.MALFORMED_JSON + ) + + +def test_json_array_is_rejected() -> None: + with pytest.raises( + CDRProcessingError + ) as exc_info: + decode_payload( + b'["not", "a", "cdr"]' + ) + + assert ( + exc_info.value.reason + == FailureReason.NORMALIZATION_ERROR + ) + + +def test_worker_normalizes_valid_payload() -> None: + normalized = process_payload( + valid_payload() + ) + + assert normalized.raw_id == "raw-worker-001" + assert normalized.radio_access_type == "LTE_4G" + assert normalized.total_bytes == 300 + assert normalized.duration_seconds == 300