amiirkhl/telecom-cdr-pipeline
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add pipeline observability and stats

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-09-20 16:58:56 +03:30
parent 1021fe55b4
commit d2109fad09
17 changed files with 1237 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
.env.sample
View File

@@ -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

30
README.md
View File

@@ -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

1
pyproject.toml
View File

@@ -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]

97
src/core/api/main.py
View File

@@ -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

24
src/core/api/metrics.py Normal file
View File

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

3
src/core/api/routes.py
View File

@@ -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={

67
src/core/api/stats.py Normal file
View File

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

6
src/core/config.py
View File

@@ -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:

1
src/core/worker/batch.py
View File

@@ -17,6 +17,7 @@ class BufferedCDR:
cdr: NormalizedCDR
original_payload: dict[str, Any]
position: KafkaPosition
kafka_timestamp_ms: int | None
class MicroBatch:

547
src/core/worker/consumer.py
View File

@@ -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."
)

24
src/core/worker/main.py
View File

@@ -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__":

156
src/core/worker/metrics.py Normal file
View File

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

89
src/core/worker/observability.py Normal file
View File

@@ -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,
)

104
src/core/worker/stats.py Normal file
View File

@@ -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
),
},
}

101
tests/integration/test_ingestion_api.py
View File

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

2
tests/unit/test_batch.py
View File

@@ -58,6 +58,8 @@ def buffered_item(
partition=0,
offset=offset,
),
kafka_timestamp_ms=None,
)

95
tests/unit/test_worker_stats.py Normal file
View File

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