diff --git a/README.md b/README.md index 373b703..3ca4fd4 100644 --- a/README.md +++ b/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 -│ └── 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. diff --git a/docker-compose.yml b/docker-compose.yml index 4ec5039..d8a0330 100644 --- a/docker-compose.yml +++ b/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: \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 5035cc4..dd998ed 100644 --- a/pyproject.toml +++ b/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] diff --git a/scripts/load_test.js b/scripts/load_test.js deleted file mode 100644 index 673c1f5..0000000 --- a/scripts/load_test.js +++ /dev/null @@ -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, - }); -} diff --git a/scripts/locustfile.py b/scripts/locustfile.py new file mode 100644 index 0000000..83fcec2 --- /dev/null +++ b/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}") + diff --git a/tree b/tree index 3ecc035..916c99d 100644 --- a/tree +++ b/tree @@ -86,6 +86,7 @@ telecom-cdr-pipeline/ ├── scripts/ │ ├── functional_traffic.py │ ├── load_test.js +│ ├── locustfile.py │ └── sample-valid-cdr.json │ ├── src/