From f45c457dc8f74bbe779fdc299f91c5d979c4f5ea Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sun, 20 Sep 2026 10:40:24 +0330 Subject: [PATCH] feat(metrics): distinguish received and acknowledged CDRs --- app/api/metrics.py | 18 ++++++++++++++++-- app/api/server.py | 3 ++- app/api/stats.py | 3 +++ scripts/generate_traffic.py | 26 +++++++++++++++++++++----- tests/test_api_server.py | 21 +++++++++++++++++++++ 5 files changed, 63 insertions(+), 8 deletions(-) diff --git a/app/api/metrics.py b/app/api/metrics.py index 2b902b0..4463f2c 100644 --- a/app/api/metrics.py +++ b/app/api/metrics.py @@ -10,27 +10,41 @@ class APIMetrics: def __init__(self): self._lock = threading.Lock() self._received_total: int = 0 + self._acknowledged_total: int = 0 def increment_received(self, count: int = 1) -> None: with self._lock: self._received_total += count + def increment_acknowledged(self, count: int = 1) -> None: + with self._lock: + self._acknowledged_total += count + @property def total_received(self) -> int: with self._lock: return self._received_total + @property + def total_acknowledged(self) -> int: + with self._lock: + return self._acknowledged_total + def get_prometheus_metrics(self) -> str: """ Formats metrics into standard Prometheus text format. """ with self._lock: - val = self._received_total + val_received = self._received_total + val_ack = self._acknowledged_total return ( "# HELP cdr_api_received_total Total CDRs received by the API\n" "# TYPE cdr_api_received_total counter\n" - f"cdr_api_received_total {val}\n" + f"cdr_api_received_total {val_received}\n" + "# HELP cdr_api_acknowledged_total Total CDRs successfully acknowledged by Kafka\n" + "# TYPE cdr_api_acknowledged_total counter\n" + f"cdr_api_acknowledged_total {val_ack}\n" ) diff --git a/app/api/server.py b/app/api/server.py index fae84c4..a825247 100644 --- a/app/api/server.py +++ b/app/api/server.py @@ -133,6 +133,7 @@ class IngestionAPIServer: await self._error(writer, 404, "not_found", f"Cannot {method} {path}") async def _handle_raw_cdr_ingestion(self, writer, body): + api_metrics.increment_received(1) try: payload = parse_payload(body) validate_structure(payload) @@ -158,7 +159,7 @@ class IngestionAPIServer: logger.warning(str(exc), extra={"raw_id": payload["raw_id"], "event": "kafka_delivery_failed"}) await self._error(writer, 503, "broker_unavailable", "Delivery unconfirmed; retry with the same raw_id and payload") return - api_metrics.increment_received(1) + api_metrics.increment_acknowledged(1) await self._json(writer, 202, {"status": "accepted", "raw_id": payload["raw_id"], "queued": True}) async def _error(self, writer, status, error, detail): diff --git a/app/api/stats.py b/app/api/stats.py index 60a0a07..bef5eb9 100644 --- a/app/api/stats.py +++ b/app/api/stats.py @@ -25,6 +25,7 @@ async def get_aggregated_pipeline_stats(worker_stats_url: str) -> Dict[str, Any] from the Processing Worker. """ total_received = api_metrics.total_received + total_acknowledged = api_metrics.total_acknowledged try: loop = asyncio.get_running_loop() @@ -38,6 +39,7 @@ async def get_aggregated_pipeline_stats(worker_stats_url: str) -> Dict[str, Any] return { "pipeline": { "total_received": total_received, + "total_acknowledged": total_acknowledged, "total_processed": pipeline_data.get("total_processed", 0), "total_dlq": pipeline_data.get("total_dlq", 0), "in_flight_buffer": pipeline_data.get("in_flight_buffer", 0), @@ -58,6 +60,7 @@ async def get_aggregated_pipeline_stats(worker_stats_url: str) -> Dict[str, Any] return { "pipeline": { "total_received": total_received, + "total_acknowledged": total_acknowledged, "total_processed": 0, "total_dlq": 0, "in_flight_buffer": 0, diff --git a/scripts/generate_traffic.py b/scripts/generate_traffic.py index fe3acc5..989a28e 100644 --- a/scripts/generate_traffic.py +++ b/scripts/generate_traffic.py @@ -153,6 +153,7 @@ def stats_snapshot(stats_url: str) -> Optional[Dict[str, Any]]: dlq = breakdown.get("dlq_by_reason", {}) if isinstance(breakdown.get("dlq_by_reason"), dict) else {} return { "total_received": int(pipeline.get("total_received", 0) or 0), + "total_acknowledged": int(pipeline.get("total_acknowledged", 0) or 0), "total_processed": int(pipeline.get("total_processed", 0) or 0), "total_dlq": int(pipeline.get("total_dlq", 0) or 0), "in_flight_buffer": int(pipeline.get("in_flight_buffer", 0) or 0), @@ -164,6 +165,19 @@ def delta(after: int, before: int) -> int: return after - before +def assessment_metrics_match(before: Dict[str, Any], after: Dict[str, Any]) -> bool: + expected_deltas = { + "total_received": 1570, + "total_acknowledged": 1540, + "total_processed": 1520, + "total_dlq": 50, + } + return all( + delta(int(after.get(name, 0)), int(before.get(name, 0))) == expected + for name, expected in expected_deltas.items() + ) + + def dlq_reason_delta( before: Optional[Dict[str, Any]], after: Optional[Dict[str, Any]], reason: str ) -> Optional[int]: @@ -522,11 +536,12 @@ def run_verification( f"verified={dlq_verified}; records={len(dlq_records)}/50; reasons={dict(reasons)}")) metrics_ok = False if baseline_stats is not None and final_stats is not None: - expected_dlq_metric = 50 - metrics_ok = (delta(final_stats["total_received"], baseline_stats["total_received"]) == 1540 - and delta(final_stats["total_processed"], baseline_stats["total_processed"]) == 1520 - and delta(final_stats["total_dlq"], baseline_stats["total_dlq"]) == expected_dlq_metric) - results.append(ScenarioResult("Acknowledged metrics", metrics_ok, f"received=1540; processed=1520; DLQ=50 expected")) + metrics_ok = assessment_metrics_match(baseline_stats, final_stats) + results.append(ScenarioResult( + "Pipeline metrics", + metrics_ok, + "received=1570; acknowledged=1540; processed=1520; DLQ=50 expected", + )) storage_ok, storage_detail = verify_normalized_rows(clickhouse_url, run_prefix, valid_payloads + burst_payloads) results.append(ScenarioResult("Normalized storage", storage_ok, storage_detail)) print("=== Assessment Verification Summary ===") @@ -537,6 +552,7 @@ def run_verification( if baseline_stats is not None and final_stats is not None: print(f"/stats received delta: {delta(final_stats['total_received'], baseline_stats['total_received'])}") + print(f"/stats acknowledged delta: {delta(final_stats['total_acknowledged'], baseline_stats['total_acknowledged'])}") print(f"/stats processed delta: {delta(final_stats['total_processed'], baseline_stats['total_processed'])}") print(f"/stats DLQ delta: {delta(final_stats['total_dlq'], baseline_stats['total_dlq'])} (expected 50)") print(f"/stats in-flight buffer: {final_stats.get('in_flight_buffer', 'n/a')}") diff --git a/tests/test_api_server.py b/tests/test_api_server.py index 5251a80..53663aa 100644 --- a/tests/test_api_server.py +++ b/tests/test_api_server.py @@ -160,6 +160,7 @@ class TestIngestionAPIServer(unittest.IsolatedAsyncioTestCase): self.assertEqual(status, 200) content = body.decode("utf-8") self.assertIn("cdr_api_received_total", content) + self.assertIn("cdr_api_acknowledged_total", content) async def test_stats_endpoint(self): status, _, body = await self._http_request("GET", "/stats") @@ -168,6 +169,26 @@ class TestIngestionAPIServer(unittest.IsolatedAsyncioTestCase): self.assertIn("pipeline", data) self.assertIn("breakdown", data) self.assertIn("total_received", data["pipeline"]) + self.assertIn("total_acknowledged", data["pipeline"]) + + async def test_metrics_separated_counters(self): + from app.api.metrics import api_metrics + received_before = api_metrics.total_received + ack_before = api_metrics.total_acknowledged + + # Send invalid payload (400 Bad Request) + status, _, _ = await self._http_request("POST", "/api/v1/cdr/raw", body=b"{}", headers={"Content-Type": "application/json"}) + self.assertEqual(status, 400) + self.assertEqual(api_metrics.total_received, received_before + 1) + self.assertEqual(api_metrics.total_acknowledged, ack_before) + + # Send valid payload (202 Accepted) + from tests.test_api_delivery import payload + valid_body = json.dumps(payload()).encode("utf-8") + status, _, _ = await self._http_request("POST", "/api/v1/cdr/raw", body=valid_body, headers={"Content-Type": "application/json"}) + self.assertEqual(status, 202) + self.assertEqual(api_metrics.total_received, received_before + 2) + self.assertEqual(api_metrics.total_acknowledged, ack_before + 1) async def test_not_found_returns_404(self): status, _, body = await self._http_request("GET", "/unknown/route")