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

Compare commits

base: amiirkhl:d2a678264ec30afcdcc0a30180afc7038a99c5c8
Branches Tags
amiirkhl:main
...
compare: amiirkhl:main
Branches Tags
amiirkhl:main

2 Commits
d2a678264e ... main

Author SHA1 Message Date
Amirhossein Khalili
081abdeb11 feat(load-test): replace k6 with locust for ongoing continuous testing and web GUI 2026-09-22 17:20:04 +03:30
Amirhossein Khalili
98fa1f6441 refactor(observability): restructure into service-owned packages 2026-09-22 17:19:57 +03:30
21 changed files with 560 additions and 226 deletions
Expand all files Collapse all files

105
README.md
View File

@@ -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
│ └── sample-invalid-rat.json
│ ├── locustfile.py
│ ├── 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.

17
docker-compose.yml
View File

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

4
pyproject.toml
View File

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

131
scripts/load_test.js
View File

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

72
scripts/locustfile.py Normal file
View File

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

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

@@ -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.routes import router
from core.config import get_settings
from core.api.metrics import ApiMetrics
from core.api.stats import (
from core.api.observability.metrics import ApiMetrics
from core.api.observability.routes import create_observability_router
from core.api.observability.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

2
src/core/api/observability/__init__.py Normal file
View File

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

0
src/core/api/metrics.py → src/core/api/observability/metrics.py
View File

59
src/core/api/observability/routes.py Normal file
View File

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

0
src/core/api/stats.py → src/core/api/observability/stats.py
View File

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

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

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

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

2
src/core/worker/observability/__init__.py Normal file
View File

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

0
src/core/worker/metrics.py → src/core/worker/observability/metrics.py
View File

4
src/core/worker/observability.py → src/core/worker/observability/server.py
View File

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

0
src/core/worker/stats.py → src/core/worker/observability/stats.py
View File

29
tests/integration/test_ingestion_api.py
View File

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

135
tests/unit/test_api_observability.py Normal file
View File

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

109
tests/unit/test_worker_observability.py Normal file
View File

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

2
tests/unit/test_worker_stats.py
View File

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

1
tree
View File

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