From d2109fad095753554f6a7801503626e745b2e7ae Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sun, 20 Sep 2026 16:58:56 +0330 Subject: [PATCH] feat: add pipeline observability and stats --- .env.sample | 12 + README.md | 30 +- pyproject.toml | 1 + src/core/api/main.py | 97 ++++- src/core/api/metrics.py | 24 ++ src/core/api/routes.py | 3 + src/core/api/stats.py | 67 +++ src/core/config.py | 6 + src/core/worker/batch.py | 1 + src/core/worker/consumer.py | 547 +++++++++++++++++++----- src/core/worker/main.py | 24 +- src/core/worker/metrics.py | 156 +++++++ src/core/worker/observability.py | 89 ++++ src/core/worker/stats.py | 104 +++++ tests/integration/test_ingestion_api.py | 101 +++++ tests/unit/test_batch.py | 2 + tests/unit/test_worker_stats.py | 95 ++++ 17 files changed, 1237 insertions(+), 122 deletions(-) create mode 100644 src/core/api/metrics.py create mode 100644 src/core/api/stats.py create mode 100644 src/core/worker/metrics.py create mode 100644 src/core/worker/observability.py create mode 100644 src/core/worker/stats.py create mode 100644 tests/unit/test_worker_stats.py diff --git a/.env.sample b/.env.sample index 1a01ed2..6a52874 100644 --- a/.env.sample +++ b/.env.sample @@ -88,3 +88,15 @@ CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS=10 BATCH_SIZE=1000 BATCH_FLUSH_INTERVAL_SECONDS=2 + + +# ========================= +# Worker Observability +# ========================= + +WORKER_HTTP_HOST=0.0.0.0 +WORKER_HTTP_PORT=8001 + +# API uses this while both services run directly on the host. +WORKER_STATS_URL=http://localhost:8001/internal/stats +WORKER_STATS_TIMEOUT_SECONDS=2 diff --git a/README.md b/README.md index 2b28edd..8ad1861 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,13 @@ worker processing is not connected yet. * Whole-batch offset commits with no partial Kafka acknowledgement * Replay-safe storage using deterministic UUIDv5 identifiers and `ReplacingMergeTree` * Graceful worker shutdown flushes buffered records before exit +* Separate Prometheus `/metrics` endpoints for the API and worker +* API accepted-CDR counter +* Worker processed and DLQ counters +* Kafka-to-ClickHouse and batch flush latency histograms +* Batch size and in-flight buffer metrics +* Kafka consumer lag by partition +* Aggregated human-readable `/stats` endpoint ## Project Structure @@ -108,8 +115,10 @@ telecom-cdr-pipeline/ │ │ ├── __init__.py │ │ ├── kafka_publisher.py │ │ ├── main.py +│ │ ├── metrics.py │ │ ├── publisher.py -│ │ └── routes.py +│ │ ├── routes.py +│ │ └── stats.py │ │ │ ├── domain/ │ │ ├── __init__.py @@ -125,7 +134,10 @@ telecom-cdr-pipeline/ │ ├── consumer.py │ ├── dlq.py │ ├── main.py -│ └── processor.py +│ ├── metrics.py +│ ├── observability.py +│ ├── processor.py +│ └── stats.py │ ├── tests/ │ ├── integration/ @@ -137,7 +149,8 @@ telecom-cdr-pipeline/ │ ├── test_normalization.py │ ├── test_project_setup.py │ ├── test_validation.py -│ └── test_worker_processor.py +│ ├── test_worker_processor.py +│ └── test_worker_stats.py.py │ ├── .env.example ├── .gitignore @@ -185,11 +198,22 @@ uvicorn core.api.main:app --reload --host 0.0.0.0 --port 8000 Available endpoints: +API: + ```text GET /health +GET /metrics +GET /stats POST /api/v1/cdr/raw ``` +Worker: + +```text +GET /metrics +GET /internal/stats +``` + A valid CDR currently returns: ```json diff --git a/pyproject.toml b/pyproject.toml index 5d217b1..0cc6831 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "confluent-kafka>=2.15,<3.0", "pydantic-settings>=2.0,<3.0", "clickhouse-connect>=1.8,<2.0", + "prometheus-client>=0.26,<1.0", ] [project.optional-dependencies] diff --git a/src/core/api/main.py b/src/core/api/main.py index 3e8cff2..6ae26f3 100644 --- a/src/core/api/main.py +++ b/src/core/api/main.py @@ -1,27 +1,61 @@ from contextlib import asynccontextmanager from fastapi import FastAPI +from fastapi.responses import JSONResponse, Response + +from prometheus_client import CONTENT_TYPE_LATEST from core.api.kafka_publisher import KafkaPublisher from core.api.publisher import Publisher from core.api.routes import router from core.config import get_settings +from core.api.metrics import ApiMetrics +from core.api.stats import ( + ApiRuntimeStats, + HttpWorkerStatsClient, + WorkerStatsProvider, + WorkerStatsUnavailable, +) def create_app( publisher: Publisher | None = None, + worker_stats_client: WorkerStatsProvider | None = None, ) -> FastAPI: + settings = get_settings() + + api_metrics = ApiMetrics() + api_stats = ApiRuntimeStats() + @asynccontextmanager async def lifespan(app: FastAPI): active_publisher = publisher if active_publisher is None: active_publisher = KafkaPublisher( - get_settings() + settings + ) + + active_worker_stats_client = ( + worker_stats_client + ) + + if active_worker_stats_client is None: + active_worker_stats_client = ( + HttpWorkerStatsClient( + url=settings.worker_stats_url, + timeout_seconds=( + settings.worker_stats_timeout_seconds + ), + ) ) app.state.publisher = active_publisher + app.state.worker_stats_client = ( + active_worker_stats_client + ) + await active_publisher.start() try: @@ -30,12 +64,17 @@ def create_app( finally: await active_publisher.stop() + await active_worker_stats_client.close() + app = FastAPI( title="Telecom CDR Ingestion API", version="0.1.0", lifespan=lifespan, ) + app.state.api_metrics = api_metrics + app.state.api_stats = api_stats + app.include_router( router, prefix="/api/v1", @@ -47,6 +86,62 @@ def create_app( "status": "ok", } + @app.get( + "/metrics", + include_in_schema=False, + ) + async def metrics() -> Response: + return Response( + content=api_metrics.render(), + media_type=CONTENT_TYPE_LATEST, + ) + + @app.get("/stats") + async def stats(): + try: + worker_stats = await ( + app.state + .worker_stats_client + .get_stats() + ) + + except WorkerStatsUnavailable as exc: + return JSONResponse( + status_code=503, + content={ + "error": "worker_stats_unavailable", + "detail": str(exc), + }, + ) + + api_snapshot = api_stats.snapshot() + + worker_pipeline = ( + worker_stats["pipeline"] + ) + + return { + "pipeline": { + "total_received": + api_snapshot["total_received"], + + "total_processed": + worker_pipeline["total_processed"], + + "total_dlq": + worker_pipeline["total_dlq"], + + "in_flight_buffer": + worker_pipeline["in_flight_buffer"], + + "last_flush_at": + worker_pipeline["last_flush_at"], + }, + + "breakdown": + worker_stats["breakdown"], + } + return app diff --git a/src/core/api/metrics.py b/src/core/api/metrics.py new file mode 100644 index 0000000..ce241f7 --- /dev/null +++ b/src/core/api/metrics.py @@ -0,0 +1,24 @@ +from prometheus_client import ( + CollectorRegistry, + Counter, + generate_latest, +) + + +class ApiMetrics: + def __init__(self) -> None: + self.registry = CollectorRegistry() + + self._received = Counter( + "cdr_api_received_total", + "Total CDRs successfully accepted by the ingestion API.", + registry=self.registry, + ) + + def record_received(self) -> None: + self._received.inc() + + def render(self) -> bytes: + return generate_latest( + self.registry + ) diff --git a/src/core/api/routes.py b/src/core/api/routes.py index cc46ea5..31d289c 100644 --- a/src/core/api/routes.py +++ b/src/core/api/routes.py @@ -74,6 +74,9 @@ async def ingest_raw_cdr(request: Request) -> JSONResponse: }, ) + request.app.state.api_metrics.record_received() + request.app.state.api_stats.record_received() + return JSONResponse( status_code=status.HTTP_202_ACCEPTED, content={ diff --git a/src/core/api/stats.py b/src/core/api/stats.py new file mode 100644 index 0000000..7355ec3 --- /dev/null +++ b/src/core/api/stats.py @@ -0,0 +1,67 @@ +import threading +from typing import Any, Protocol + +import httpx2 + + +class ApiRuntimeStats: + def __init__(self) -> None: + self._lock = threading.Lock() + self._total_received = 0 + + def record_received(self) -> None: + with self._lock: + self._total_received += 1 + + def snapshot(self) -> dict[str, int]: + with self._lock: + return { + "total_received": self._total_received, + } + + +class WorkerStatsUnavailable(Exception): + pass + + +class WorkerStatsProvider(Protocol): + async def get_stats( + self, + ) -> dict[str, Any]: + ... + + async def close(self) -> None: + ... + + +class HttpWorkerStatsClient: + def __init__( + self, + url: str, + timeout_seconds: float, + ) -> None: + self._url = url + + self._client = httpx2.AsyncClient( + timeout=timeout_seconds, + ) + + async def get_stats( + self, + ) -> dict[str, Any]: + try: + response = await self._client.get( + self._url + ) + + response.raise_for_status() + + return response.json() + + except httpx2.HTTPError as exc: + raise WorkerStatsUnavailable( + f"Unable to retrieve worker stats: {exc}" + ) from exc + + async def close(self) -> None: + await self._client.aclose() diff --git a/src/core/config.py b/src/core/config.py index 2454c6a..2806a1d 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -46,6 +46,12 @@ class Settings(BaseSettings): batch_size: int = 1_000 batch_flush_interval_seconds: float = 2.0 + worker_http_host: str = "0.0.0.0" + worker_http_port: int = 8001 + + worker_stats_url: str = "http://localhost:8001/internal/stats" + worker_stats_timeout_seconds: float = 2.0 + @lru_cache def get_settings() -> Settings: diff --git a/src/core/worker/batch.py b/src/core/worker/batch.py index 98d6d35..fb7ed7d 100644 --- a/src/core/worker/batch.py +++ b/src/core/worker/batch.py @@ -17,6 +17,7 @@ class BufferedCDR: cdr: NormalizedCDR original_payload: dict[str, Any] position: KafkaPosition + kafka_timestamp_ms: int | None class MicroBatch: diff --git a/src/core/worker/consumer.py b/src/core/worker/consumer.py index 1c9f8d6..d84a861 100644 --- a/src/core/worker/consumer.py +++ b/src/core/worker/consumer.py @@ -1,4 +1,5 @@ import logging +import time from typing import Any from confluent_kafka import ( @@ -8,6 +9,11 @@ from confluent_kafka import ( TopicPartition, ) +from core.config import Settings +from core.domain.errors import ( + CDRProcessingError, + FailureReason, +) from core.worker.batch import ( BufferedCDR, KafkaPosition, @@ -17,19 +23,16 @@ from core.worker.clickhouse import ( ClickHouseWriteError, ClickHouseWriter, ) -from core.config import Settings -from core.domain.errors import ( - CDRProcessingError, - FailureReason, -) from core.worker.dlq import ( DLQPublishError, DLQPublisher, ) +from core.worker.metrics import WorkerMetrics from core.worker.processor import ( decode_payload, process_payload, ) +from core.worker.stats import WorkerRuntimeStats logger = logging.getLogger(__name__) @@ -39,8 +42,12 @@ class ProcessingWorker: def __init__( self, settings: Settings, + metrics: WorkerMetrics, + stats: WorkerRuntimeStats, ) -> None: self._settings = settings + self._metrics = metrics + self._stats = stats self._consumer = Consumer( { @@ -56,11 +63,15 @@ class ProcessingWorker: "auto.offset.reset": settings.kafka_auto_offset_reset, + # We commit manually only after the processing + # outcome has safely completed. "enable.auto.commit": False, } ) - self._dlq = DLQPublisher(settings) + self._dlq = DLQPublisher( + settings + ) self._clickhouse = ClickHouseWriter( settings @@ -79,11 +90,16 @@ class ProcessingWorker: self._running = True self._consumer.subscribe( - [self._settings.kafka_raw_topic] + [ + self._settings.kafka_raw_topic, + ] ) logger.info( - "Worker started. topic=%s group=%s instance=%s", + ( + "Worker started. " + "topic=%s group=%s instance=%s" + ), self._settings.kafka_raw_topic, self._settings.kafka_consumer_group, self._settings.worker_instance_id, @@ -92,7 +108,8 @@ class ProcessingWorker: try: while self._running: message = self._consumer.poll( - self._settings.kafka_consumer_poll_timeout_seconds + self._settings + .kafka_consumer_poll_timeout_seconds ) if message is None: @@ -104,9 +121,14 @@ class ProcessingWorker: "Kafka consumer error: %s", message.error(), ) + + self._flush_if_due() continue - self._handle_message(message) + self._handle_message( + message + ) + self._flush_if_due() finally: @@ -118,6 +140,8 @@ class ProcessingWorker: ) -> None: value = message.value() or b"" + # If JSON decoding fails, we still want to preserve + # the original raw Kafka payload inside the DLQ. original_payload: Any = { "raw": value.decode( "utf-8", @@ -126,13 +150,25 @@ class ProcessingWorker: } try: - payload = decode_payload(value) + payload = decode_payload( + value + ) original_payload = payload - normalized = process_payload(payload) + normalized = process_payload( + payload + ) except CDRProcessingError as exc: + # Important: + # + # There may already be valid records from this + # partition sitting uncommitted in the batch. + # + # Before committing this invalid record after + # sending it to the DLQ, flush those earlier + # records first. if not self._batch.is_empty: self._flush_batch() @@ -150,8 +186,12 @@ class ProcessingWorker: ) error = CDRProcessingError( - reason=FailureReason.NORMALIZATION_ERROR, - detail=f"Unexpected processing error: {exc}", + reason=( + FailureReason.NORMALIZATION_ERROR + ), + detail=( + f"Unexpected processing error: {exc}" + ), ) if not self._batch.is_empty: @@ -171,86 +211,59 @@ class ProcessingWorker: offset=message.offset(), ) + _, kafka_timestamp_ms = ( + message.timestamp() + ) + + if ( + kafka_timestamp_ms is not None + and kafka_timestamp_ms < 0 + ): + kafka_timestamp_ms = None + self._batch.add( BufferedCDR( cdr=normalized, original_payload=payload, position=position, + kafka_timestamp_ms=kafka_timestamp_ms, ) ) + self._metrics.set_buffer_size( + self._batch.size + ) + + self._stats.set_buffer_size( + self._batch.size + ) + logger.debug( ( "Buffered normalized CDR. " - "raw_id=%s batch_size=%s" + "raw_id=%s " + "partition=%s " + "offset=%s " + "batch_size=%s" ), normalized.raw_id, + message.partition(), + message.offset(), self._batch.size, ) if self._batch.is_full: self._flush_batch() - def _route_to_dlq( + def _flush_if_due( 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 _flush_if_due(self) -> None: if self._batch.is_due(): self._flush_batch() - def _flush_batch(self) -> None: + def _flush_batch( + self, + ) -> None: items = self._batch.snapshot() if not items: @@ -266,58 +279,296 @@ class ProcessingWorker: for item in items ] + batch_size = len( + records + ) + + self._metrics.observe_batch_size( + batch_size + ) + logger.info( "Flushing ClickHouse batch. size=%s", - len(records), + batch_size, ) + flush_started = time.monotonic() + try: self._clickhouse.insert( records ) except ClickHouseWriteError as exc: - logger.exception( - "ClickHouse batch permanently failed." + self._metrics.observe_batch_flush_duration( + time.monotonic() + - flush_started ) + logger.exception( + ( + "ClickHouse batch permanently " + "failed after retries." + ) + ) + + # The assessment requires an exhausted + # ClickHouse batch to be routed to the DLQ. + # + # _route_batch_to_dlq only returns if every + # DLQ publication was acknowledged. self._route_batch_to_dlq( items=items, storage_error=exc, ) - # Only reached if EVERY DLQ publish succeeded. + # Every record now has a successful processing + # outcome via the DLQ, so the batch offsets can + # safely be committed. self._commit_positions( positions ) self._batch.clear() + self._metrics.set_buffer_size( + 0 + ) + + self._stats.set_buffer_size( + 0 + ) + logger.warning( ( - "Failed ClickHouse batch routed to DLQ " - "and Kafka offsets committed. size=%s" + "Failed ClickHouse batch routed " + "to DLQ and offsets committed. " + "size=%s" ), - len(records), + batch_size, ) return + clickhouse_written_at = ( + time.time() + ) + + self._metrics.observe_batch_flush_duration( + time.monotonic() + - flush_started + ) + + # Record end-to-end Kafka -> ClickHouse latency. + for item in items: + if ( + item.kafka_timestamp_ms + is None + ): + continue + + latency_seconds = ( + clickhouse_written_at + - ( + item.kafka_timestamp_ms + / 1000 + ) + ) + + self._metrics.observe_kafka_to_clickhouse_latency( + latency_seconds + ) + + # Critical ordering: + # + # ClickHouse ACK + # ↓ + # Kafka commit + # + # Never commit before the database write. self._commit_positions( positions ) + for item in items: + self._metrics.record_processed( + item.cdr + ) + + self._stats.record_processed( + item.cdr + ) + + self._stats.record_flush() + self._batch.clear() + self._metrics.set_buffer_size( + 0 + ) + + self._stats.set_buffer_size( + 0 + ) + logger.info( - "ClickHouse batch flushed successfully. size=%s", - len(records), + ( + "ClickHouse batch flushed " + "successfully. size=%s" + ), + batch_size, + ) + + 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." + ) + ) + + # Do not continue consuming and later commit + # beyond this record. + raise + + self._metrics.record_dlq( + error.reason.value + ) + + self._stats.record_dlq( + error.reason.value + ) + + # DLQ publication has been acknowledged, + # therefore this source Kafka record has a + # successful terminal outcome. + self._commit( + message + ) + + def _route_batch_to_dlq( + self, + items: list[BufferedCDR], + storage_error: ClickHouseWriteError, + ) -> None: + # The assessment restricts failure_reason to the + # existing enum values and does not define a + # dedicated storage_error reason. + error = CDRProcessingError( + reason=( + FailureReason.NORMALIZATION_ERROR + ), + detail=( + "ClickHouse batch insert failed " + "after retries: " + f"{storage_error}" + ), + ) + + logger.error( + ( + "Routing failed ClickHouse batch " + "to DLQ. batch_size=%s" + ), + len(items), + ) + + for item in items: + # If any one of these fails, the exception + # propagates and the source Kafka offsets are + # NOT committed. + self._dlq.publish( + original_payload=( + item.original_payload + ), + error=error, + ) + + self._metrics.record_dlq( + error.reason.value + ) + + self._stats.record_dlq( + error.reason.value + ) + + 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 + ) + + committed_offset = ( + message.offset() + 1 + ) + + self._update_consumer_lag( + topic=message.topic(), + partition=message.partition(), + committed_offset=committed_offset, + ) + + logger.debug( + ( + "Committed Kafka offset. " + "partition=%s offset=%s" + ), + message.partition(), + committed_offset, ) def _commit_positions( self, positions: list[KafkaPosition], ) -> None: + # Only the highest successfully processed offset + # from each partition needs to be committed. + # + # Kafka committed offsets mean: + # "the NEXT offset to consume". highest_offsets: dict[ tuple[str, int], int, @@ -333,13 +584,19 @@ class ProcessingWorker: position.offset + 1 ) - current = highest_offsets.get(key) + current = ( + highest_offsets.get( + key + ) + ) if ( current is None or next_offset > current ): - highest_offsets[key] = next_offset + highest_offsets[ + key + ] = next_offset offsets = [ TopicPartition( @@ -350,57 +607,111 @@ class ProcessingWorker: for ( topic, partition, - ), offset in highest_offsets.items() + ), offset in ( + highest_offsets.items() + ) ] - committed = self._consumer.commit( - offsets=offsets, - asynchronous=False, + if not offsets: + return + + committed = ( + self._consumer.commit( + offsets=offsets, + asynchronous=False, + ) ) - for partition in committed or []: - if partition.error is not None: + for partition in ( + committed or [] + ): + if ( + partition.error + is not None + ): raise KafkaException( partition.error ) - def _route_batch_to_dlq( - self, - items: list[BufferedCDR], - storage_error: ClickHouseWriteError, - ) -> None: - error = CDRProcessingError( - reason=FailureReason.NORMALIZATION_ERROR, - detail=( - "ClickHouse batch insert failed after retries: " - f"{storage_error}" - ), - ) - - logger.error( - ( - "Routing failed ClickHouse batch to DLQ. " - "batch_size=%s" - ), - len(items), - ) - - for item in items: - self._dlq.publish( - original_payload=item.original_payload, - error=error, + self._update_consumer_lag( + topic=partition.topic, + partition=partition.partition, + committed_offset=( + partition.offset + ), ) - def stop(self) -> None: + logger.debug( + "Committed Kafka batch offsets: %s", + offsets, + ) + + def _update_consumer_lag( + self, + topic: str, + partition: int, + committed_offset: int, + ) -> None: + try: + _, high_watermark = ( + self._consumer + .get_watermark_offsets( + TopicPartition( + topic, + partition, + ), + timeout=1.0, + ) + ) + + except KafkaException: + logger.exception( + ( + "Unable to retrieve " + "consumer lag. partition=%s" + ), + partition, + ) + + return + + lag = max( + high_watermark + - committed_offset, + 0, + ) + + self._metrics.set_consumer_lag( + partition=partition, + lag=lag, + ) + + def stop( + self, + ) -> None: + logger.info( + "Worker stop requested." + ) + self._running = False - def _shutdown(self) -> None: - logger.info("Worker shutting down.") + def _shutdown( + self, + ) -> None: + logger.info( + "Worker shutting down." + ) try: + # The assessment requires all remaining + # in-memory records to be flushed before exit. if not self._batch.is_empty: logger.info( - "Flushing remaining records before shutdown." + ( + "Flushing remaining records " + "before shutdown. size=%s" + ), + self._batch.size, ) self._flush_batch() @@ -415,3 +726,7 @@ class ProcessingWorker: finally: self._consumer.close() + + logger.info( + "Worker shutdown complete." + ) \ No newline at end of file diff --git a/src/core/worker/main.py b/src/core/worker/main.py index 7a31f73..a3c2701 100644 --- a/src/core/worker/main.py +++ b/src/core/worker/main.py @@ -3,7 +3,9 @@ import signal from core.config import get_settings from core.worker.consumer import ProcessingWorker - +from core.worker.metrics import WorkerMetrics +from core.worker.observability import WorkerObservabilityServer +from core.worker.stats import WorkerRuntimeStats logging.basicConfig( level=logging.INFO, @@ -19,8 +21,20 @@ logging.basicConfig( def main() -> None: settings = get_settings() + metrics = WorkerMetrics() + + stats = WorkerRuntimeStats() + worker = ProcessingWorker( settings=settings, + metrics=metrics, + stats=stats, + ) + + observability = WorkerObservabilityServer( + settings=settings, + metrics=metrics, + stats=stats, ) def handle_shutdown( @@ -44,7 +58,13 @@ def main() -> None: handle_shutdown, ) - worker.run() + observability.start() + + try: + worker.run() + + finally: + observability.stop() if __name__ == "__main__": diff --git a/src/core/worker/metrics.py b/src/core/worker/metrics.py new file mode 100644 index 0000000..0813d5a --- /dev/null +++ b/src/core/worker/metrics.py @@ -0,0 +1,156 @@ +from prometheus_client import ( + CollectorRegistry, + Counter, + Gauge, + Histogram, + generate_latest, +) + +from core.domain.errors import FailureReason +from core.domain.models import NormalizedCDR + + +class WorkerMetrics: + def __init__(self) -> None: + self.registry = CollectorRegistry() + + self._processed = Counter( + "cdr_processed_total", + "Total CDRs successfully processed and stored.", + labelnames=( + "service_type", + "radio_access_type", + ), + registry=self.registry, + ) + + self._dlq = Counter( + "cdr_dlq_total", + "Total CDRs routed to the DLQ.", + labelnames=( + "reason", + ), + registry=self.registry, + ) + + self._kafka_to_clickhouse_latency = Histogram( + "cdr_kafka_to_clickhouse_latency_seconds", + "Time from Kafka message creation to ClickHouse write.", + buckets=( + 0.05, + 0.1, + 0.25, + 0.5, + 1, + 2.5, + 5, + ), + registry=self.registry, + ) + + self._batch_flush_duration = Histogram( + "cdr_batch_flush_duration_seconds", + "Duration of ClickHouse batch flushes.", + registry=self.registry, + ) + + self._batch_size = Histogram( + "cdr_batch_size_at_flush", + "Number of CDRs in each flushed batch.", + buckets=( + 10, + 50, + 100, + 250, + 500, + 1000, + ), + registry=self.registry, + ) + + self._buffer_size = Gauge( + "cdr_in_flight_buffer_size", + "Current number of CDRs buffered for ClickHouse.", + registry=self.registry, + ) + + self._consumer_lag = Gauge( + "cdr_kafka_consumer_lag", + "Kafka consumer lag by partition.", + labelnames=( + "partition", + ), + registry=self.registry, + ) + + # Ensure all required DLQ reasons appear even before + # their first failure. + for reason in FailureReason: + self._dlq.labels( + reason=reason.value + ) + + def record_processed( + self, + cdr: NormalizedCDR, + ) -> None: + self._processed.labels( + service_type=cdr.service_type, + radio_access_type=cdr.radio_access_type, + ).inc() + + def record_dlq( + self, + reason: str, + ) -> None: + self._dlq.labels( + reason=reason + ).inc() + + def observe_kafka_to_clickhouse_latency( + self, + seconds: float, + ) -> None: + self._kafka_to_clickhouse_latency.observe( + max(seconds, 0.0) + ) + + def observe_batch_flush_duration( + self, + seconds: float, + ) -> None: + self._batch_flush_duration.observe( + seconds + ) + + def observe_batch_size( + self, + size: int, + ) -> None: + self._batch_size.observe( + size + ) + + def set_buffer_size( + self, + size: int, + ) -> None: + self._buffer_size.set( + size + ) + + def set_consumer_lag( + self, + partition: int, + lag: int, + ) -> None: + self._consumer_lag.labels( + partition=str(partition) + ).set( + max(lag, 0) + ) + + def render(self) -> bytes: + return generate_latest( + self.registry + ) diff --git a/src/core/worker/observability.py b/src/core/worker/observability.py new file mode 100644 index 0000000..31ea702 --- /dev/null +++ b/src/core/worker/observability.py @@ -0,0 +1,89 @@ +import logging +import threading + +import uvicorn +from fastapi import FastAPI +from fastapi.responses import Response +from prometheus_client import CONTENT_TYPE_LATEST + +from core.config import Settings +from core.worker.metrics import WorkerMetrics +from core.worker.stats import WorkerRuntimeStats + + +logger = logging.getLogger(__name__) + + +def create_worker_observability_app( + metrics: WorkerMetrics, + stats: WorkerRuntimeStats, +) -> FastAPI: + app = FastAPI( + title="Telecom CDR Worker Observability", + version="0.1.0", + ) + + @app.get( + "/metrics", + include_in_schema=False, + ) + async def metrics_endpoint() -> Response: + return Response( + content=metrics.render(), + media_type=CONTENT_TYPE_LATEST, + ) + + @app.get("/internal/stats") + async def internal_stats() -> dict: + return stats.snapshot() + + return app + + +class WorkerObservabilityServer: + def __init__( + self, + settings: Settings, + metrics: WorkerMetrics, + stats: WorkerRuntimeStats, + ) -> None: + app = create_worker_observability_app( + metrics=metrics, + stats=stats, + ) + + config = uvicorn.Config( + app=app, + host=settings.worker_http_host, + port=settings.worker_http_port, + log_level="warning", + access_log=False, + ) + + self._server = uvicorn.Server( + config=config + ) + + self._thread = threading.Thread( + target=self._server.run, + name="worker-observability", + daemon=True, + ) + + def start(self) -> None: + logger.info( + "Starting worker observability server." + ) + + self._thread.start() + + def stop(self) -> None: + logger.info( + "Stopping worker observability server." + ) + + self._server.should_exit = True + + self._thread.join( + timeout=5, + ) diff --git a/src/core/worker/stats.py b/src/core/worker/stats.py new file mode 100644 index 0000000..2ed5621 --- /dev/null +++ b/src/core/worker/stats.py @@ -0,0 +1,104 @@ +import threading +from collections import defaultdict +from datetime import datetime, timezone +from typing import Any + +from core.domain.models import NormalizedCDR + + +class WorkerRuntimeStats: + def __init__(self) -> None: + self._lock = threading.Lock() + + self._total_processed = 0 + self._total_dlq = 0 + + self._in_flight_buffer = 0 + + self._last_flush_at: str | None = None + + self._by_service_type: dict[str, int] = defaultdict(int) + + self._by_radio_access_type: dict[str, int] = defaultdict(int) + + self._dlq_by_reason: dict[str, int] = defaultdict(int) + + def record_processed( + self, + cdr: NormalizedCDR, + ) -> None: + with self._lock: + self._total_processed += 1 + + self._by_service_type[ + cdr.service_type + ] += 1 + + self._by_radio_access_type[ + cdr.radio_access_type + ] += 1 + + def record_dlq( + self, + reason: str, + ) -> None: + with self._lock: + self._total_dlq += 1 + + self._dlq_by_reason[ + reason + ] += 1 + + def set_buffer_size( + self, + size: int, + ) -> None: + with self._lock: + self._in_flight_buffer = size + + def record_flush(self) -> None: + now = ( + datetime.now(timezone.utc) + .isoformat() + .replace("+00:00", "Z") + ) + + with self._lock: + self._last_flush_at = now + + def snapshot( + self, + ) -> dict[str, Any]: + with self._lock: + return { + "pipeline": { + "total_processed": + self._total_processed, + + "total_dlq": + self._total_dlq, + + "in_flight_buffer": + self._in_flight_buffer, + + "last_flush_at": + self._last_flush_at, + }, + + "breakdown": { + "by_service_type": + dict( + self._by_service_type + ), + + "by_radio_access_type": + dict( + self._by_radio_access_type + ), + + "dlq_by_reason": + dict( + self._dlq_by_reason + ), + }, + } diff --git a/tests/integration/test_ingestion_api.py b/tests/integration/test_ingestion_api.py index 98b4c99..fc7bb5a 100644 --- a/tests/integration/test_ingestion_api.py +++ b/tests/integration/test_ingestion_api.py @@ -38,6 +38,35 @@ class FakePublisher: ) +class FakeWorkerStatsClient: + async def get_stats( + self, + ) -> dict[str, Any]: + return { + "pipeline": { + "total_processed": 10, + "total_dlq": 2, + "in_flight_buffer": 3, + "last_flush_at": + "2026-09-20T12:00:00Z", + }, + "breakdown": { + "by_service_type": { + "4G_DATA": 10, + }, + "by_radio_access_type": { + "LTE_4G": 10, + }, + "dlq_by_reason": { + "invalid_rat_value": 2, + }, + }, + } + + async def close(self) -> None: + pass + + @pytest.fixture def publisher() -> FakePublisher: return FakePublisher() @@ -49,6 +78,7 @@ def client( ): app = create_app( publisher=publisher, + worker_stats_client=FakeWorkerStatsClient(), ) with TestClient(app) as test_client: @@ -226,3 +256,74 @@ def test_kafka_failure_returns_503( assert response.status_code == 503 assert response.json()["error"] == "kafka_unavailable" + + +def test_metrics_endpoint( + client: TestClient, +) -> None: + response = client.get( + "/metrics" + ) + + assert response.status_code == 200 + + assert ( + "cdr_api_received_total" + in response.text + ) + + +def test_stats_endpoint( + client: TestClient, +) -> None: + response = client.get( + "/stats" + ) + + assert response.status_code == 200 + + body = response.json() + + assert ( + body["pipeline"]["total_processed"] + == 10 + ) + + assert ( + body["pipeline"]["total_dlq"] + == 2 + ) + + assert ( + body["pipeline"]["in_flight_buffer"] + == 3 + ) + + assert ( + body["breakdown"] + ["by_service_type"] + ["4G_DATA"] + == 10 + ) + + +def test_successful_request_increments_api_metric( + client: TestClient, +) -> None: + response = client.post( + "/api/v1/cdr/raw", + json=valid_payload(), + ) + + assert response.status_code == 202 + + metrics = client.get( + "/metrics" + ) + + assert ( + 'cdr_api_received_total 1.0' + in metrics.text + ) + + diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py index e5e7e4b..a047be5 100644 --- a/tests/unit/test_batch.py +++ b/tests/unit/test_batch.py @@ -58,6 +58,8 @@ def buffered_item( partition=0, offset=offset, ), + + kafka_timestamp_ms=None, ) diff --git a/tests/unit/test_worker_stats.py b/tests/unit/test_worker_stats.py new file mode 100644 index 0000000..5ef8d29 --- /dev/null +++ b/tests/unit/test_worker_stats.py @@ -0,0 +1,95 @@ +from datetime import datetime, timezone +from ipaddress import IPv4Address +from uuid import uuid4 + +from core.domain.models import NormalizedCDR +from core.worker.stats import WorkerRuntimeStats + + +def normalized_cdr() -> NormalizedCDR: + now = datetime.now( + timezone.utc + ) + + return NormalizedCDR( + cdr_id=uuid4(), + raw_id="raw-stats-1", + + msisdn="+19875551234", + imsi="310410998877665", + + service_type="4G_DATA", + radio_access_type="LTE_4G", + cell_id="4021_8812", + + bytes_uploaded=100, + bytes_downloaded=200, + total_bytes=300, + duration_seconds=300, + + session_start_time=now, + session_end_time=now, + + source_ip=IPv4Address( + "10.240.12.1" + ), + ) + + +def test_worker_stats() -> None: + stats = WorkerRuntimeStats() + + stats.record_processed( + normalized_cdr() + ) + + stats.record_dlq( + "invalid_rat_value" + ) + + stats.set_buffer_size(4) + + stats.record_flush() + + snapshot = stats.snapshot() + + assert ( + snapshot["pipeline"]["total_processed"] + == 1 + ) + + assert ( + snapshot["pipeline"]["total_dlq"] + == 1 + ) + + assert ( + snapshot["pipeline"]["in_flight_buffer"] + == 4 + ) + + assert ( + snapshot["breakdown"] + ["by_service_type"] + ["4G_DATA"] + == 1 + ) + + assert ( + snapshot["breakdown"] + ["by_radio_access_type"] + ["LTE_4G"] + == 1 + ) + + assert ( + snapshot["breakdown"] + ["dlq_by_reason"] + ["invalid_rat_value"] + == 1 + ) + + assert ( + snapshot["pipeline"]["last_flush_at"] + is not None + )