feat(load-test): replace k6 with locust for ongoing continuous testing and web GUI

README.md

@@ -99,6 +99,22 @@ worker processing is not connected yet.
 * Persistent Redpanda and ClickHouse storage
 * Complete stack startup with `docker compose up --build`
 
+## Observability Architecture
+
+Observability is strictly service-owned; each service manages its own telemetry and monitoring components:
+
+* **API Observability** (`core.api.observability`):
+  * Prometheus metrics: `cdr_api_received_total` counter.
+  * API runtime stats: In-memory counter for accepted CDR requests.
+  * Worker stats HTTP client: `HttpWorkerStatsClient` implementing `WorkerStatsProvider` to query the worker's internal stats endpoint.
+  * Observability routes: Dedicated router mounted on the API for `GET /metrics` and `GET /stats`.
+  * The `GET /stats` endpoint aggregates API reception stats with worker processing stats retrieved over HTTP.
+
+* **Worker Observability** (`core.worker.observability`):
+  * Prometheus metrics: Processed and DLQ counters, batch flush duration and Kafka-to-ClickHouse latency histograms, in-flight buffer and consumer lag gauges.
+  * Runtime stats: Thread-safe in-memory breakdown by service type, radio access type, DLQ failure reason, and buffer status.
+  * Internal observability HTTP server: `WorkerObservabilityServer` running a background ASGI server providing `GET /health`, `GET /metrics`, and `GET /internal/stats`.
+
 ## Project Structure
 
 ```text
@@ -108,8 +124,9 @@ telecom-cdr-pipeline/
 │       └── init.sql
 │
 ├── scripts/
-│   ├── sample-valid-cdr.json
+│   ├── locustfile.py
-│   └── sample-invalid-rat.json
+│   ├── functional_traffic.py
+│   └── sample-valid-cdr.json
 │
 ├── src/
 │   └── core/
@@ -120,10 +137,13 @@ telecom-cdr-pipeline/
 │       │   ├── __init__.py
 │       │   ├── kafka_publisher.py
 │       │   ├── main.py
-│       │   ├── metrics.py
 │       │   ├── publisher.py
 │       │   ├── routes.py
-│       │   └── stats.py
+│       │   └── observability/
+│       │       ├── __init__.py
+│       │       ├── metrics.py
+│       │       ├── routes.py
+│       │       └── stats.py
 │       │
 │       ├── domain/
 │       │   ├── __init__.py
@@ -139,23 +159,27 @@ telecom-cdr-pipeline/
 │           ├── consumer.py
 │           ├── dlq.py
 │           ├── main.py
-│           ├── metrics.py
-│           ├── observability.py
 │           ├── processor.py
-│           └── stats.py
+│           └── observability/
+│               ├── __init__.py
+│               ├── metrics.py
+│               ├── server.py
+│               └── stats.py
 │
 ├── tests/
 │   ├── integration/
 │   │   └── test_ingestion_api.py
 │   │
 │   └── unit/
+│       ├── test_api_observability.py
 │       ├── test_batch.py
 │       ├── test_dlq.py
 │       ├── test_normalization.py
 │       ├── test_project_setup.py
 │       ├── test_validation.py
+│       ├── test_worker_observability.py
 │       ├── test_worker_processor.py
-│       └── test_worker_stats.py.py
+│       └── test_worker_stats.py
 │
 ├── .env.example
 ├── .gitignore
@@ -233,6 +257,71 @@ Once the full stack starts cleanly, a real CDR reaches ClickHouse, and:
 pytest -v
 ```
 
+## Load Testing with Locust (Ongoing Test & Web GUI)
+
+A high-throughput Locust load testing suite is provided in `scripts/locustfile.py` using `FastHttpUser` (`geventhttpclient`). It runs an **ongoing test that executes continuously until you stop it** and exposes an interactive **Web GUI** to monitor real-time throughput, latency distributions, and backpressure.
+
+### Option 1: Run Locust via Local Python
+
+1. Ensure dependencies are installed:
+   ```bash
+   pip install -e ".[load-test]"
+   ```
+2. Start the Locust Web GUI:
+   ```bash
+   locust -f scripts/locustfile.py --host http://localhost:8000
+   ```
+3. Open your browser at **http://localhost:8089**.
+
+### Option 2: Run Locust via Docker Compose
+
+Locust is configured as an on-demand service under the `load-test` profile in `docker-compose.yml`:
+
+```bash
+docker compose --profile load-test up locust
+```
+
+Open your browser at **http://localhost:8089**. The target host is pre-configured to `http://api:8000`.
+
+---
+
+### Controlling the Ongoing Test from the Web GUI
+
+1. Open **http://localhost:8089**.
+2. Configure your test parameters:
+   - **Number of users**: e.g., `50` (or `500` for high stress).
+   - **Spawn rate**: e.g., `10` users per second.
+   - **Host**: `http://localhost:8000` (or `http://api:8000` inside Docker).
+3. Click **Start swarming**:
+   - The test will **run continuously and never stop** automatically.
+   - You can inspect real-time metrics across tabs:
+     - **Statistics**: Total requests, RPS, 50th/90th/95th/99th percentiles, and failure counts.
+     - **Charts**: Live charts for Total Requests per Second, Response Times (ms), and Number of Active Users.
+     - **Failures**: Distinguishes normal graceful backpressure (`429 Too Many Requests`) from broker errors (`503 Service Unavailable`).
+     - **Download Data**: Export requests/failures CSVs or generate full HTML test reports.
+4. Adjust or Stop:
+   - You can change the user count in real-time from the GUI top bar.
+   - Click **Stop swarming** whenever you want to end the test.
+
+---
+
+### Correlating with Pipeline Observability
+
+While the Locust test is running ongoing traffic:
+
+1. Check API metrics:
+   ```bash
+   curl -s http://localhost:8000/metrics | grep cdr_api_received_total
+   ```
+2. Check worker consumer lag:
+   ```bash
+   curl -s http://localhost:8001/metrics | grep cdr_kafka_consumer_lag
+   ```
+3. Check aggregated pipeline statistics:
+   ```bash
+   curl -s http://localhost:8000/stats
+   ```
+
 ## Delivery Semantics
 
 The worker uses at-least-once delivery.

docker-compose.yml

@@ -247,6 +247,23 @@ services:
     restart: unless-stopped
 
 
+  locust:
+    image: locustio/locust:latest
+    container_name: telecom-locust
+    profiles:
+      - load-test
+    depends_on:
+      api:
+        condition: service_healthy
+    environment:
+      LOCUST_HOST: http://api:8000
+    ports:
+      - "${LOCUST_HTTP_PORT:-8089}:8089"
+    volumes:
+      - ./scripts:/mnt/locust
+    command: -f /mnt/locust/locustfile.py
+
+
 volumes:
   redpanda-data:
   clickhouse-data:

pyproject.toml

@@ -17,6 +17,10 @@ dependencies = [
 [project.optional-dependencies]
 dev = [
     "pytest>=8.0",
+    "locust>=2.31.0",
+]
+load-test = [
+    "locust>=2.31.0",
 ]
 
 [build-system]

scripts/load_test.js

@@ -1,131 +0,0 @@
-import http from 'k6/http';
-import { check } from 'k6';
-import { Counter } from 'k6/metrics';
-
-
-const accepted = new Counter(
-    'cdr_accepted'
-);
-
-const backpressure = new Counter(
-    'cdr_backpressure'
-);
-
-const unavailable = new Counter(
-    'cdr_unavailable'
-);
-
-const unexpected = new Counter(
-    'cdr_unexpected'
-);
-
-
-const RATE = Number(
-    __ENV.RATE || 1000
-);
-
-const PRE_ALLOCATED_VUS = Number(
-    __ENV.PRE_ALLOCATED_VUS || 250
-);
-
-const MAX_VUS = Number(
-    __ENV.MAX_VUS || 1000
-);
-
-const DURATION = (
-    __ENV.DURATION || '10s'
-);
-
-const BASE_URL = (
-    __ENV.BASE_URL ||
-    'http://localhost:8000'
-);
-
-
-export const options = {
-    discardResponseBodies: true,
-
-    scenarios: {
-        ingestion: {
-            executor: 'constant-arrival-rate',
-
-            rate: RATE,
-            timeUnit: '1s',
-
-            duration: DURATION,
-
-            preAllocatedVUs: PRE_ALLOCATED_VUS,
-            maxVUs: MAX_VUS,
-        },
-    },
-};
-
-
-export default function () {
-    const now = Math.floor(
-        Date.now() / 1000
-    );
-
-    const rawId =
-        `k6-${__VU}-${__ITER}-${Date.now()}`;
-
-    const payload = JSON.stringify({
-        raw_id: rawId,
-
-        src_ip: '10.240.12.1',
-
-        device_id:
-            '310410998877665',
-
-        caller_num:
-            '+14155552671',
-
-        svc: '4G_DATA',
-
-        up_bytes: 1024,
-        down_bytes: 4096,
-
-        rat: 6,
-
-        lac: '4021',
-        ci: '8812',
-
-        epoch_start: now,
-        epoch_end: now + 300,
-    });
-
-    const response = http.post(
-        `${BASE_URL}/api/v1/cdr/raw`,
-        payload,
-        {
-            headers: {
-                'Content-Type':
-                    'application/json',
-            },
-
-            tags: {
-                endpoint: 'cdr_ingest',
-            },
-        }
-    );
-
-    if (response.status === 202) {
-        accepted.add(1);
-
-    } else if (response.status === 429) {
-        backpressure.add(1);
-
-    } else if (response.status === 503) {
-        unavailable.add(1);
-
-    } else {
-        unexpected.add(1);
-    }
-
-    check(response, {
-        'accepted or gracefully rejected':
-            (r) =>
-                r.status === 202 ||
-                r.status === 429,
-    });
-}

scripts/locustfile.py

@@ -0,0 +1,72 @@
+import os
+import random
+import time
+import uuid
+from locust import FastHttpUser, task, between, events
+
+# Configuration defaults with environment variable overrides
+MIN_WAIT = float(os.getenv("LOCUST_MIN_WAIT", "0.001"))
+MAX_WAIT = float(os.getenv("LOCUST_MAX_WAIT", "0.010"))
+TREAT_429_AS_SUCCESS = os.getenv("TREAT_429_AS_SUCCESS", "false").lower() in ("true", "1", "yes")
+
+# Valid Radio Access Types supported by the pipeline validation layer
+VALID_RATS = [1, 6, 9]
+
+# Service types
+SERVICE_TYPES = ["4G_DATA", "5G_DATA", "VOICE", "SMS"]
+
+
+class CDRLoadUser(FastHttpUser):
+    """
+    High-performance Locust user for generating and ingesting Telecom CDR records.
+    Uses FastHttpUser (geventhttpclient) for maximum throughput with minimal CPU overhead.
+    Runs continuously until manually stopped in the Web GUI or terminal.
+    """
+
+    # Low wait time between iterations to maintain continuous traffic flow
+    wait_time = between(MIN_WAIT, MAX_WAIT)
+
+    @task
+    def ingest_cdr(self) -> None:
+        """
+        Generates a valid CDR payload and posts it to the raw CDR ingestion endpoint.
+        """
+        now = int(time.time())
+        raw_id = f"locust-{uuid.uuid4().hex}"
+
+        payload = {
+            "raw_id": raw_id,
+            "src_ip": "10.240.12.1",
+            "device_id": "310410998877665",
+            "caller_num": "+14155552671",
+            "svc": random.choice(SERVICE_TYPES),
+            "up_bytes": random.randint(512, 8192),
+            "down_bytes": random.randint(1024, 32768),
+            "rat": random.choice(VALID_RATS),
+            "lac": "4021",
+            "ci": "8812",
+            "epoch_start": now,
+            "epoch_end": now + 300,
+        }
+
+        with self.client.post(
+            "/api/v1/cdr/raw",
+            json=payload,
+            name="/api/v1/cdr/raw",
+            catch_response=True,
+        ) as response:
+            if response.status_code == 202:
+                response.success()
+            elif response.status_code == 429:
+                # 429 indicates bounded queue saturation (graceful backpressure)
+                if TREAT_429_AS_SUCCESS:
+                    response.success()
+                else:
+                    response.failure("429 Too Many Requests (Queue Backpressure)")
+            elif response.status_code == 503:
+                response.failure("503 Service Unavailable (Kafka Broker Failure)")
+            elif response.status_code == 400:
+                response.failure(f"400 Bad Request: {response.text}")
+            else:
+                response.failure(f"Unexpected HTTP {response.status_code}: {response.text}")
+

src/core/api/main.py

@@ -1,21 +1,18 @@
 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.observability.metrics import ApiMetrics
-from core.api.routes import router
+from core.api.observability.routes import create_observability_router
-from core.config import get_settings
+from core.api.observability.stats import (
-from core.api.metrics import ApiMetrics
-from core.api.stats import (
     ApiRuntimeStats,
     HttpWorkerStatsClient,
     WorkerStatsProvider,
-    WorkerStatsUnavailable,
 )
+from core.api.publisher import Publisher
+from core.api.routes import router
+from core.config import get_settings
 
 
 def create_app(
@@ -27,6 +24,15 @@ def create_app(
     api_metrics = ApiMetrics()
     api_stats = ApiRuntimeStats()
 
+    active_worker_stats_client = (
+        worker_stats_client
+        if worker_stats_client is not None
+        else HttpWorkerStatsClient(
+            url=settings.worker_stats_url,
+            timeout_seconds=settings.worker_stats_timeout_seconds,
+        )
+    )
+
     @asynccontextmanager
     async def lifespan(app: FastAPI):
         active_publisher = publisher
@@ -36,20 +42,6 @@ def create_app(
                 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 = (
@@ -75,73 +67,27 @@ def create_app(
     app.state.api_metrics = api_metrics
     app.state.api_stats = api_stats
 
+    observability_router = create_observability_router(
+        metrics=api_metrics,
+        stats=api_stats,
+        worker_stats_provider=active_worker_stats_client,
+    )
+
     app.include_router(
         router,
         prefix="/api/v1",
     )
 
+    app.include_router(
+        observability_router,
+    )
+
     @app.get("/health")
     async def health() -> dict[str, str]:
         return {
             "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
 
 

src/core/api/observability/__init__.py

@@ -0,0 +1,2 @@
+"""API observability package."""
+

src/core/api/observability/routes.py

@@ -0,0 +1,59 @@
+from typing import Any
+
+from fastapi import APIRouter, Response
+from fastapi.responses import JSONResponse
+from prometheus_client import CONTENT_TYPE_LATEST
+
+from core.api.observability.metrics import ApiMetrics
+from core.api.observability.stats import (
+    ApiRuntimeStats,
+    WorkerStatsProvider,
+    WorkerStatsUnavailable,
+)
+
+
+def create_observability_router(
+    metrics: ApiMetrics,
+    stats: ApiRuntimeStats,
+    worker_stats_provider: WorkerStatsProvider,
+) -> APIRouter:
+    router = APIRouter()
+
+    @router.get(
+        "/metrics",
+        include_in_schema=False,
+    )
+    async def metrics_endpoint() -> Response:
+        return Response(
+            content=metrics.render(),
+            media_type=CONTENT_TYPE_LATEST,
+        )
+
+    @router.get("/stats")
+    async def stats_endpoint():
+        try:
+            worker_stats = await worker_stats_provider.get_stats()
+        except WorkerStatsUnavailable as exc:
+            return JSONResponse(
+                status_code=503,
+                content={
+                    "error": "worker_stats_unavailable",
+                    "detail": str(exc),
+                },
+            )
+
+        api_snapshot = 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 router

src/core/worker/consumer.py

@@ -27,12 +27,12 @@ from core.worker.dlq import (
     DLQPublishError,
     DLQPublisher,
 )
-from core.worker.metrics import WorkerMetrics
+from core.worker.observability.metrics import WorkerMetrics
 from core.worker.processor import (
     decode_payload,
     process_payload,
 )
-from core.worker.stats import WorkerRuntimeStats
+from core.worker.observability.stats import WorkerRuntimeStats
 
 
 logger = logging.getLogger(__name__)

src/core/worker/main.py

@@ -3,9 +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.metrics import WorkerMetrics
-from core.worker.observability import WorkerObservabilityServer
+from core.worker.observability.server import WorkerObservabilityServer
-from core.worker.stats import WorkerRuntimeStats
+from core.worker.observability.stats import WorkerRuntimeStats
 
 logging.basicConfig(
     level=logging.INFO,

src/core/worker/observability/__init__.py

@@ -0,0 +1,2 @@
+"""Worker observability package."""
+

src/core/worker/observability/server.py

@@ -7,8 +7,8 @@ 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.observability.metrics import WorkerMetrics
-from core.worker.stats import WorkerRuntimeStats
+from core.worker.observability.stats import WorkerRuntimeStats
 
 
 logger = logging.getLogger(__name__)

tests/integration/test_ingestion_api.py

@@ -5,6 +5,7 @@ import pytest
 from fastapi.testclient import TestClient
 
 from core.api.main import create_app
+from core.api.observability.stats import WorkerStatsUnavailable
 from core.api.publisher import (
     IngestionQueueFull,
     KafkaPublishError,
@@ -327,3 +328,31 @@ def test_successful_request_increments_api_metric(
     )
 
 
+def test_stats_endpoint_when_worker_unavailable(
+    publisher: FakePublisher,
+) -> None:
+    class FailingWorkerStatsClient:
+        async def get_stats(self) -> dict[str, Any]:
+            raise WorkerStatsUnavailable(
+                "Worker stats endpoint unreachable"
+            )
+
+        async def close(self) -> None:
+            pass
+
+    app = create_app(
+        publisher=publisher,
+        worker_stats_client=FailingWorkerStatsClient(),
+    )
+
+    with TestClient(app) as test_client:
+        response = test_client.get("/stats")
+
+        assert response.status_code == 503
+
+        body = response.json()
+        assert body["error"] == "worker_stats_unavailable"
+        assert "Worker stats endpoint unreachable" in body["detail"]
+
+
+

tests/unit/test_api_observability.py

@@ -0,0 +1,135 @@
+import httpx2
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from prometheus_client import CONTENT_TYPE_LATEST
+
+from core.api.observability.metrics import ApiMetrics
+from core.api.observability.routes import create_observability_router
+from core.api.observability.stats import (
+    ApiRuntimeStats,
+    HttpWorkerStatsClient,
+    WorkerStatsUnavailable,
+)
+
+
+@pytest.mark.anyio
+async def test_http_worker_stats_client_success() -> None:
+    async def fake_handler(request: httpx2.Request) -> httpx2.Response:
+        assert request.url == "http://test-worker/internal/stats"
+        return httpx2.Response(
+            status_code=200,
+            json={
+                "pipeline": {
+                    "total_processed": 5,
+                    "total_dlq": 1,
+                    "in_flight_buffer": 2,
+                    "last_flush_at": "2026-09-22T12:00:00Z",
+                },
+                "breakdown": {
+                    "by_service_type": {"4G_DATA": 5},
+                    "by_radio_access_type": {"LTE_4G": 5},
+                    "dlq_by_reason": {"timeout": 1},
+                },
+            },
+        )
+
+    transport = httpx2.MockTransport(fake_handler)
+    client = HttpWorkerStatsClient(
+        url="http://test-worker/internal/stats",
+        timeout_seconds=2.0,
+    )
+    # Inject mock transport
+    client._client = httpx2.AsyncClient(transport=transport)
+
+    stats = await client.get_stats()
+    assert stats["pipeline"]["total_processed"] == 5
+    await client.close()
+
+
+@pytest.mark.anyio
+async def test_http_worker_stats_client_error() -> None:
+    async def failing_handler(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(status_code=500)
+
+    transport = httpx2.MockTransport(failing_handler)
+    client = HttpWorkerStatsClient(
+        url="http://test-worker/internal/stats",
+        timeout_seconds=2.0,
+    )
+    client._client = httpx2.AsyncClient(transport=transport)
+
+    with pytest.raises(WorkerStatsUnavailable) as exc_info:
+        await client.get_stats()
+
+    assert "Unable to retrieve worker stats" in str(exc_info.value)
+    await client.close()
+
+
+def test_api_observability_router_metrics() -> None:
+    metrics = ApiMetrics()
+    stats = ApiRuntimeStats()
+
+    class DummyWorkerStatsProvider:
+        async def get_stats(self) -> dict:
+            return {}
+
+        async def close(self) -> None:
+            pass
+
+    app = FastAPI()
+    router = create_observability_router(
+        metrics=metrics,
+        stats=stats,
+        worker_stats_provider=DummyWorkerStatsProvider(),
+    )
+    app.include_router(router)
+
+    client = TestClient(app)
+    response = client.get("/metrics")
+    assert response.status_code == 200
+    assert CONTENT_TYPE_LATEST in response.headers["content-type"]
+    assert "cdr_api_received_total" in response.text
+
+
+def test_api_observability_router_stats_aggregation() -> None:
+    metrics = ApiMetrics()
+    stats = ApiRuntimeStats()
+    stats.record_received()
+    stats.record_received()
+
+    class FakeProvider:
+        async def get_stats(self) -> dict:
+            return {
+                "pipeline": {
+                    "total_processed": 15,
+                    "total_dlq": 3,
+                    "in_flight_buffer": 0,
+                    "last_flush_at": "2026-09-22T10:00:00Z",
+                },
+                "breakdown": {
+                    "by_service_type": {"4G_DATA": 15},
+                    "by_radio_access_type": {"LTE_4G": 15},
+                    "dlq_by_reason": {},
+                },
+            }
+
+        async def close(self) -> None:
+            pass
+
+    app = FastAPI()
+    router = create_observability_router(
+        metrics=metrics,
+        stats=stats,
+        worker_stats_provider=FakeProvider(),
+    )
+    app.include_router(router)
+
+    client = TestClient(app)
+    response = client.get("/stats")
+    assert response.status_code == 200
+    body = response.json()
+    assert body["pipeline"]["total_received"] == 2
+    assert body["pipeline"]["total_processed"] == 15
+    assert body["pipeline"]["total_dlq"] == 3
+

tests/unit/test_worker_observability.py

@@ -0,0 +1,109 @@
+from datetime import datetime, timezone
+from ipaddress import IPv4Address
+from uuid import uuid4
+
+from fastapi.testclient import TestClient
+from prometheus_client import CONTENT_TYPE_LATEST
+
+from core.domain.models import NormalizedCDR
+from core.worker.observability.metrics import WorkerMetrics
+from core.worker.observability.server import create_worker_observability_app
+from core.worker.observability.stats import WorkerRuntimeStats
+
+
+def create_sample_cdr() -> NormalizedCDR:
+    now = datetime.now(timezone.utc)
+    return NormalizedCDR(
+        cdr_id=uuid4(),
+        raw_id="raw-obs-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_observability_health_endpoint() -> None:
+    metrics = WorkerMetrics()
+    stats = WorkerRuntimeStats()
+    app = create_worker_observability_app(metrics=metrics, stats=stats)
+    client = TestClient(app)
+
+    response = client.get("/health")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}
+
+
+def test_worker_observability_metrics_endpoint() -> None:
+    metrics = WorkerMetrics()
+    stats = WorkerRuntimeStats()
+    app = create_worker_observability_app(metrics=metrics, stats=stats)
+    client = TestClient(app)
+
+    response = client.get("/metrics")
+    assert response.status_code == 200
+    assert CONTENT_TYPE_LATEST in response.headers["content-type"]
+    assert "cdr_processed_total" in response.text
+    assert "cdr_dlq_total" in response.text
+
+
+def test_worker_observability_internal_stats_endpoint() -> None:
+    metrics = WorkerMetrics()
+    stats = WorkerRuntimeStats()
+    app = create_worker_observability_app(metrics=metrics, stats=stats)
+    client = TestClient(app)
+
+    response = client.get("/internal/stats")
+    assert response.status_code == 200
+    body = response.json()
+    assert "pipeline" in body
+    assert "breakdown" in body
+    assert body["pipeline"]["total_processed"] == 0
+    assert body["pipeline"]["total_dlq"] == 0
+
+
+def test_shared_instance_behavior_between_processing_and_observability() -> None:
+    # Processing and observability must share the identical metrics and stats instances
+    metrics = WorkerMetrics()
+    stats = WorkerRuntimeStats()
+    app = create_worker_observability_app(metrics=metrics, stats=stats)
+    client = TestClient(app)
+
+    # Simulate worker processing activity updating the instances
+    cdr = create_sample_cdr()
+    metrics.record_processed(cdr)
+    stats.record_processed(cdr)
+
+    metrics.record_dlq("invalid_rat_value")
+    stats.record_dlq("invalid_rat_value")
+
+    metrics.set_buffer_size(5)
+    stats.set_buffer_size(5)
+
+    # Verify that the observability server sees the exact same updates
+    stats_response = client.get("/internal/stats")
+    assert stats_response.status_code == 200
+    stats_body = stats_response.json()
+
+    assert stats_body["pipeline"]["total_processed"] == 1
+    assert stats_body["pipeline"]["total_dlq"] == 1
+    assert stats_body["pipeline"]["in_flight_buffer"] == 5
+    assert stats_body["breakdown"]["by_service_type"]["4G_DATA"] == 1
+    assert stats_body["breakdown"]["by_radio_access_type"]["LTE_4G"] == 1
+    assert stats_body["breakdown"]["dlq_by_reason"]["invalid_rat_value"] == 1
+
+    # Verify metrics endpoint output contains the recorded metrics
+    metrics_response = client.get("/metrics")
+    assert metrics_response.status_code == 200
+    assert 'cdr_processed_total{radio_access_type="LTE_4G",service_type="4G_DATA"} 1.0' in metrics_response.text
+    assert 'cdr_dlq_total{reason="invalid_rat_value"} 1.0' in metrics_response.text
+    assert 'cdr_in_flight_buffer_size 5.0' in metrics_response.text
+

tests/unit/test_worker_stats.py

@@ -3,7 +3,7 @@ from ipaddress import IPv4Address
 from uuid import uuid4
 
 from core.domain.models import NormalizedCDR
-from core.worker.stats import WorkerRuntimeStats
+from core.worker.observability.stats import WorkerRuntimeStats
 
 
 def normalized_cdr() -> NormalizedCDR:

tree

@@ -86,6 +86,7 @@ telecom-cdr-pipeline/
 ├── scripts/
 │   ├── functional_traffic.py
 │   ├── load_test.js
+│   ├── locustfile.py
 │   └── sample-valid-cdr.json
 │
 ├── src/