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

test: add end-to-end and load verification

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-09-20 17:46:42 +03:30
parent 85d9003e43
commit d2a678264e
3 changed files with 1145 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

242
scripts/functional_traffic.py Normal file
View File

@@ -0,0 +1,242 @@
import json
import os
import random
import time
import uuid
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from copy import deepcopy
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
BASE_URL = os.getenv(
"BASE_URL",
"http://localhost:8000",
)
ENDPOINT = f"{BASE_URL}/api/v1/cdr/raw"
RUN_ID = os.getenv(
"RUN_ID",
uuid.uuid4().hex[:8],
)
def valid_payload(index: int) -> dict:
start = int(time.time())
return {
"raw_id": f"functional-{RUN_ID}-{index:04d}",
"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": start,
"epoch_end": start + 300,
}
def send_json(payload: dict) -> int | str:
body = json.dumps(
payload
).encode("utf-8")
return send_raw(body)
def send_raw(body: bytes) -> int | str:
request = Request(
ENDPOINT,
data=body,
method="POST",
headers={
"Content-Type": "application/json",
},
)
try:
with urlopen(
request,
timeout=10,
) as response:
return response.status
except HTTPError as exc:
return exc.code
except URLError as exc:
return f"network_error:{exc.reason}"
def main() -> None:
requests: list[
tuple[str, dict | bytes]
] = []
valid_records = [
valid_payload(index)
for index in range(1000)
]
# 1000 valid CDRs
for payload in valid_records:
requests.append(
("valid", payload)
)
# 20 missing caller_num
for index in range(20):
payload = valid_payload(
2000 + index
)
payload.pop(
"caller_num"
)
requests.append(
("missing_caller_num", payload)
)
# 10 malformed JSON requests
for _ in range(10):
requests.append(
(
"invalid_json",
b'{"raw_id": invalid}',
)
)
# 10 invalid RAT values
for index in range(10):
payload = valid_payload(
3000 + index
)
payload["rat"] = 99
requests.append(
("invalid_rat", payload)
)
# 10 invalid epochs
for index in range(10):
payload = valid_payload(
4000 + index
)
payload["epoch_start"] = (
payload["epoch_end"] + 100
)
requests.append(
("invalid_epoch", payload)
)
# 20 duplicates of already accepted valid records.
for payload in valid_records[:20]:
requests.append(
(
"duplicate",
deepcopy(payload),
)
)
random.shuffle(
requests
)
results = Counter()
started_at = time.perf_counter()
with ThreadPoolExecutor(
max_workers=50
) as executor:
futures = []
for request_type, payload in requests:
if isinstance(
payload,
bytes,
):
future = executor.submit(
send_raw,
payload,
)
else:
future = executor.submit(
send_json,
payload,
)
futures.append(
(
request_type,
future,
)
)
for request_type, future in futures:
status = future.result()
results[
(
request_type,
status,
)
] += 1
elapsed = (
time.perf_counter()
- started_at
)
print()
print(
f"Run ID: {RUN_ID}"
)
print(
f"Total requests: {len(requests)}"
)
print(
f"Elapsed: {elapsed:.2f}s"
)
print()
for (
request_type,
status,
), count in sorted(
results.items(),
key=lambda item: str(
item[0]
),
):
print(
f"{request_type:24} "
f"status={status!s:8} "
f"count={count}"
)
print()
print("Expected:")
print(" valid → 1000 x 202")
print(" duplicate → 20 x 202")
print(" missing_caller → 20 x 400")
print(" invalid_json → 10 x 400")
print(" invalid_rat → 10 x 400")
print(" invalid_epoch → 10 x 400")
if __name__ == "__main__":
main()

131
scripts/load_test.js Normal file
View File

@@ -0,0 +1,131 @@
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,
});
}

772
tree Normal file
View File

@@ -0,0 +1,772 @@
# Telecom CDR Pipeline
High-throughput, fault-tolerant telecom Call Detail Record ingestion and analytics pipeline built with FastAPI, Redpanda/Kafka, ClickHouse, and Python.
## Architecture
```text
Network Element
│
▼
FastAPI Ingestion API
│
├── Validation
│
├── Bounded Async Queue
│
└── Kafka Producer
│
▼
Redpanda / cdr.raw
│
▼
Processing Worker
│
├── Decode + Validate
├── Normalize
├── Micro-Batch
│
├── valid ──────► ClickHouse
│
└── invalid ────► cdr.raw.dlq
```
The main services are:
* **FastAPI API** — validates incoming CDRs, applies backpressure, and publishes accepted records to Kafka.
* **Redpanda** — Kafka-compatible durable buffer between ingestion and processing.
* **Processing Worker** — validates, normalizes, batches, and persists CDRs.
* **ClickHouse** — analytical storage for normalized records.
* **DLQ** — stores malformed or unprocessable records with structured failure information.
## Features
* Bounded asynchronous ingestion queue
* `202`, `400`, `429`, and `503` ingestion responses
* Long-lived asynchronous Kafka producer
* Kafka delivery acknowledgement before `202 Accepted`
* Kafka consumer with manual offset management
* Worker-side validation and normalization
* E.164 phone-number normalization using `phonenumbers`
* RAT mapping for `3G`, `LTE_4G`, and `5G_NR`
* Deterministic UUIDv5 `cdr_id` generation
* Structured Dead Letter Queue
* Configurable ClickHouse micro-batching
* Flush on 1,000 records or 2-second timeout
* ClickHouse retries with exponential backoff
* At-least-once delivery semantics
* Replay-safe ClickHouse storage using `ReplacingMergeTree`
* Graceful worker shutdown with final buffer flush
* Separate Prometheus metrics endpoints for API and worker
* Kafka consumer lag monitoring
* Aggregated `/stats` endpoint
* Health-based Docker Compose startup ordering
* Automated Kafka topic provisioning
* Automated and idempotent ClickHouse initialization
* Unit, integration, functional, and load testing
---
## Project Structure
Generated directories such as `__pycache__` and `*.egg-info` are omitted below.
```text
telecom-cdr-pipeline/
├── Dockerfile.api
├── Dockerfile.worker
├── README.md
├── docker-compose.yml
├── pyproject.toml
│
├── infrastructure/
│ └── clickhouse/
│ └── init.sql
│
├── scripts/
│ ├── functional_traffic.py
│ ├── load_test.js
│ └── sample-valid-cdr.json
│
├── src/
│ └── core/
│ ├── __init__.py
│ ├── config.py
│ │
│ ├── api/
│ │ ├── __init__.py
│ │ ├── kafka_publisher.py
│ │ ├── main.py
│ │ ├── metrics.py
│ │ ├── publisher.py
│ │ ├── routes.py
│ │ └── stats.py
│ │
│ ├── domain/
│ │ ├── __init__.py
│ │ ├── errors.py
│ │ ├── models.py
│ │ ├── normalization.py
│ │ └── validation.py
│ │
│ └── worker/
│ ├── __init__.py
│ ├── batch.py
│ ├── clickhouse.py
│ ├── consumer.py
│ ├── dlq.py
│ ├── main.py
│ ├── metrics.py
│ ├── observability.py
│ ├── processor.py
│ └── stats.py
│
└── tests/
├── integration/
│ └── test_ingestion_api.py
│
└── unit/
├── test_batch.py
├── test_dlq.py
├── test_normalization.py
├── test_project_setup.py
├── test_validation.py
├── test_worker_processor.py
└── test_worker_stats.py
```
---
## Quickstart
### 1. Configure Environment
Create your local environment file from `.env.example`:
```bash
cp .env.example .env
```
### 2. Start the Complete Pipeline
```bash
docker compose up --build -d
```
Check service status:
```bash
docker compose ps -a
```
Expected state:
```text
telecom-redpanda healthy
telecom-clickhouse healthy
telecom-api healthy
telecom-worker healthy
telecom-redpanda-init exited (0)
```
### 3. Verify Health
API:
```bash
curl http://localhost:8000/health
```
Worker:
```bash
curl http://localhost:8001/health
```
### 4. Send a Sample CDR
```bash
curl -X POST \
http://localhost:8000/api/v1/cdr/raw \
-H "Content-Type: application/json" \
--data-binary "@scripts/sample-valid-cdr.json"
```
Expected response:
```json
{
"status": "accepted",
"raw_id": "raw-test-001"
}
```
---
## API Endpoints
### Ingestion API
```text
POST /api/v1/cdr/raw
```
Possible responses:
| Status | Meaning |
| ------ | ------------------------------------------- |
| `202` | Payload validated and acknowledged by Kafka |
| `400` | Invalid JSON or invalid CDR schema |
| `429` | Bounded ingestion queue is full |
| `503` | Kafka delivery is unavailable |
### API Observability
```text
GET http://localhost:8000/health
GET http://localhost:8000/metrics
GET http://localhost:8000/stats
```
### Worker Observability
```text
GET http://localhost:8001/health
GET http://localhost:8001/metrics
GET http://localhost:8001/internal/stats
```
---
## Micro-Batching
The worker never inserts CDRs into ClickHouse row-by-row.
Normalized records are buffered and flushed when either condition occurs:
```text
Batch reaches 1,000 records
OR
2 seconds pass after the batch starts
```
Configuration:
```env
BATCH_SIZE=1000
BATCH_FLUSH_INTERVAL_SECONDS=2
```
This reduces ClickHouse insert overhead while still keeping low-volume traffic latency bounded.
On `SIGTERM`, the worker stops consuming new records, flushes the remaining buffer, commits the corresponding Kafka offsets, and then exits.
---
## Delivery and Offset Strategy
The worker uses **at-least-once delivery**.
Kafka auto-commit is disabled.
The normal processing order is:
```text
Kafka consume
↓
Validate + Normalize
↓
Micro-Batch
↓
ClickHouse INSERT
↓
ClickHouse ACK
↓
Kafka offset commit
```
Offsets are never committed before the corresponding processing outcome is safely completed.
### ClickHouse Failure
A failed ClickHouse batch is retried with exponential backoff:
```text
Attempt 1
↓ failure
0.5s delay
Attempt 2
↓ failure
1.0s delay
Attempt 3
↓ failure
Entire batch → DLQ
```
Offsets are committed only after every DLQ publication succeeds.
If DLQ publishing fails, the offsets remain uncommitted so Kafka can replay the records.
### Worker Crash After ClickHouse Write
If ClickHouse successfully stores a batch but the worker crashes before committing Kafka offsets, Kafka delivers the records again after restart.
This can produce duplicate physical inserts, which is handled by the deduplication strategy below.
### Partial Batch Failure
A batch is treated as a single processing unit.
Offsets are never partially committed for a failed ClickHouse batch.
---
## Deduplication
Each CDR gets a deterministic UUIDv5 generated from its `raw_id`:
```python
uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)
```
The same `raw_id` therefore always generates the same `cdr_id`.
ClickHouse uses:
```text
ReplacingMergeTree(processed_at)
```
with `cdr_id` included in the sorting key.
This allows records replayed because of at-least-once Kafka delivery to eventually collapse into one logical record.
For immediate query-time deduplication:
```sql
SELECT *
FROM telecom.cdr_records FINAL;
```
---
## Dead Letter Queue
Invalid or unprocessable Kafka records are published to:
```text
cdr.raw.dlq
```
DLQ records include:
```json
{
"original_payload": {},
"failure_reason": "invalid_rat_value",
"failed_field": "rat",
"failed_at": "2026-09-20T12:00:00Z",
"worker_instance": "worker-1"
}
```
Supported failure reasons:
```text
missing_required_field
invalid_rat_value
invalid_epoch
malformed_json
normalization_error
```
A bad record does not crash the worker.
---
## Monitoring
### API Metrics
```bash
curl http://localhost:8000/metrics
```
Includes:
```text
cdr_api_received_total
```
### Worker Metrics
```bash
curl http://localhost:8001/metrics
```
Includes:
```text
cdr_processed_total
cdr_dlq_total
cdr_kafka_to_clickhouse_latency_seconds
cdr_batch_flush_duration_seconds
cdr_batch_size_at_flush
cdr_in_flight_buffer_size
cdr_kafka_consumer_lag
```
### Pipeline Stats
```bash
curl http://localhost:8000/stats
```
Example:
```json
{
"pipeline": {
"total_received": 1000,
"total_processed": 995,
"total_dlq": 5,
"in_flight_buffer": 0,
"last_flush_at": "2026-09-20T12:00:00Z"
},
"breakdown": {
"by_service_type": {},
"by_radio_access_type": {},
"dlq_by_reason": {}
}
}
```
---
## ClickHouse Verification
Open a ClickHouse client:
```bash
docker compose exec clickhouse \
clickhouse-client \
--user telecom \
--password telecom_dev_password \
--database telecom
```
Count records by service:
```sql
SELECT
service_type,
count() AS total
FROM cdr_records
GROUP BY service_type;
```
Average duration by RAT:
```sql
SELECT
radio_access_type,
avg(duration_seconds) AS avg_duration
FROM cdr_records
GROUP BY radio_access_type;
```
Top cells by traffic:
```sql
SELECT
cell_id,
sum(total_bytes) AS total_traffic
FROM cdr_records
GROUP BY cell_id
ORDER BY total_traffic DESC
LIMIT 10;
```
Check logical deduplication:
```sql
SELECT
raw_id,
count()
FROM cdr_records FINAL
GROUP BY raw_id
HAVING count() > 1;
```
Expected result:
```text
0 rows
```
---
## Automated Tests
Install development dependencies:
```bash
pip install -e ".[dev]"
```
Run all unit and integration tests:
```bash
pytest -v
```
---
## Functional Traffic Verification
The functional traffic script exercises the pipeline with the assessment scenarios, including valid records, invalid requests, and duplicate `raw_id` values.
Run:
```bash
python scripts/functional_traffic.py
```
The assessment scenarios are:
| Scenario | Count | Expected |
| ------------------------- | ----: | --------------------------------------------- |
| Valid records | 1,000 | Stored in ClickHouse |
| Missing `caller_num` | 20 | Rejected / DLQ when worker is tested directly |
| Invalid JSON | 10 | API `400`; DLQ when worker is bypassed |
| Unknown `rat=99` | 10 | Rejected / DLQ when worker is tested directly |
| `epoch_start > epoch_end` | 10 | Rejected / DLQ when worker is tested directly |
| Duplicate `raw_id` | 20 | One logical ClickHouse record per `raw_id` |
Invalid records submitted through the public API are rejected before Kafka. Worker-side DLQ handling is independently exercised by publishing malformed or invalid records directly to `cdr.raw` and by the automated worker tests.
### Functional Traffic Result
Paste the final `functional_traffic.py` output below:
```text
Total requests: 1070
Elapsed: 5.21s
duplicate status=202 count=20
invalid_epoch status=400 count=10
invalid_json status=400 count=10
invalid_rat status=400 count=10
missing_caller_num status=400 count=20
valid status=202 count=1000
Expected:
valid → 1000 x 202
duplicate → 20 x 202
missing_caller → 20 x 400
invalid_json → 10 x 400
invalid_rat → 10 x 400
invalid_epoch → 10 x 400
```
---
## Required k6 Burst Test
The assessment requires:
```text
500 records in less than 5 seconds
0 × 429 responses
No data loss
```
The included k6 script uses a configurable constant arrival rate.
Run the required assessment scenario with:
```bash
docker run --rm -i \
--network=host \
-e RATE=125 \
-e DURATION=4s \
-e PRE_ALLOCATED_VUS=500 \
-e MAX_VUS=750 \
grafana/k6 run - \
< scripts/load_test.js
```
This schedules approximately:
```text
125 requests/second × 4 seconds = 500 requests
```
A successful result should show approximately:
```text
iterations 500
http_reqs 500
cdr_accepted 500
cdr_backpressure 0
cdr_unavailable 0
http_req_failed 0%
dropped_iterations 0
```
After the run, allow the worker to finish processing and verify that no records were lost.
Check worker lag:
```bash
curl -s http://localhost:8001/metrics \
| grep cdr_kafka_consumer_lag
```
Check pipeline statistics:
```bash
curl http://localhost:8000/stats
```
### Required k6 Result
Paste the final required 500-record burst result below:
```text
TOTAL RESULTS
checks_total.......: 4087 272.979956/s
checks_succeeded...: 34.47% 1409 out of 4087
checks_failed......: 65.52% 2678 out of 4087
✗ accepted or gracefully rejected
↳ 34% — ✓ 1409 / ✗ 2678
CUSTOM
cdr_accepted...................: 1409 94.110291/s
cdr_unavailable................: 2678 178.869665/s
HTTP
http_req_duration..............: avg=5.26s min=44.55ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s
{ expected_response:true }...: avg=2.33s min=44.55ms med=1.75s max=9.94s p(90)=5.75s p(95)=6.1s
http_req_failed................: 65.52% 2678 out of 4087
http_reqs......................: 4087 272.979956/s
EXECUTION
dropped_iterations.............: 5912 394.875826/s
iteration_duration.............: avg=5.26s min=46.48ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s
iterations.....................: 4087 272.979956/s
vus............................: 214 min=214 max=2000
vus_max........................: 2000 min=1500 max=2000
NETWORK
data_received..................: 862 kB 58 kB/s
data_sent......................: 1.6 MB 105 kB/s
running (15.0s), 0000/2000 VUs, 4087 complete and 0 interrupted iterations
ingestion ✓ [ 100% ] 0000/2000 VUs 10s 1000.00 iters/s
```
---
## Optional Stress Test
The required assessment test is the 500-record burst above.
A larger stress test can optionally be executed with:
```bash
docker run --rm -i \
--network=host \
-e RATE=1000 \
-e DURATION=10s \
-e PRE_ALLOCATED_VUS=1500 \
-e MAX_VUS=2000 \
grafana/k6 run - \
< scripts/load_test.js
```
This is useful for identifying throughput and latency limits but is not required for the assessment's 500-record burst criterion.
---
## Configuration Reference
Configuration is provided through environment variables. Local defaults are documented in `.env.example`.
| Variable | Default | Purpose |
| ----------------------------------------- | -------------------------------------- | -------------------------------------- |
| `REDPANDA_VERSION` | `v26.2.3` | Redpanda image version |
| `KAFKA_BOOTSTRAP_SERVERS` | `localhost:19092` | Kafka address for host applications |
| `KAFKA_INTERNAL_BOOTSTRAP_SERVERS` | `redpanda:9092` | Kafka address inside Docker |
| `KAFKA_RAW_TOPIC` | `cdr.raw` | Raw CDR topic |
| `KAFKA_RAW_TOPIC_PARTITIONS` | `3` | Raw topic partitions |
| `KAFKA_RAW_TOPIC_REPLICATION_FACTOR` | `1` | Raw topic replication |
| `KAFKA_RAW_TOPIC_RETENTION_MS` | `86400000` | Raw topic retention |
| `KAFKA_RAW_TOPIC_COMPRESSION` | `lz4` | Raw topic compression |
| `KAFKA_DLQ_TOPIC` | `cdr.raw.dlq` | Dead Letter Queue topic |
| `KAFKA_DLQ_TOPIC_PARTITIONS` | `1` | DLQ partitions |
| `KAFKA_DLQ_TOPIC_REPLICATION_FACTOR` | `1` | DLQ replication |
| `KAFKA_DLQ_TOPIC_RETENTION_MS` | `604800000` | DLQ retention |
| `KAFKA_CONSUMER_GROUP` | `cdr-processing-workers` | Worker consumer group |
| `KAFKA_AUTO_OFFSET_RESET` | `earliest` | Consumer fallback offset |
| `KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS` | `1` | Worker Kafka poll timeout |
| `KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS` | `5` | DLQ delivery timeout |
| `KAFKA_DELIVERY_TIMEOUT_SECONDS` | `4` | API Kafka acknowledgement timeout |
| `KAFKA_MESSAGE_TIMEOUT_MS` | `3000` | Kafka message timeout |
| `KAFKA_LINGER_MS` | `5` | Producer batching delay |
| `KAFKA_BATCH_SIZE` | `131072` | Kafka producer batch size |
| `KAFKA_PRODUCER_MAX_MESSAGES` | `100000` | Kafka producer internal queue limit |
| `API_QUEUE_SIZE` | `10000` | Bounded API ingestion queue |
| `CLICKHOUSE_VERSION` | `26.8.7.19` | ClickHouse image version |
| `CLICKHOUSE_HOST` | `localhost` | ClickHouse host for local applications |
| `CLICKHOUSE_INTERNAL_HOST` | `clickhouse` | ClickHouse Docker service hostname |
| `CLICKHOUSE_PORT` | `8123` | ClickHouse HTTP port |
| `CLICKHOUSE_HTTP_HOST_PORT` | `8123` | Host-exposed ClickHouse port |
| `CLICKHOUSE_DATABASE` | `telecom` | ClickHouse database |
| `CLICKHOUSE_USER` | `telecom` | ClickHouse user |
| `CLICKHOUSE_PASSWORD` | `telecom_dev_password` | Local development password |
| `BATCH_SIZE` | `1000` | Worker micro-batch maximum size |
| `BATCH_FLUSH_INTERVAL_SECONDS` | `2` | Worker time-based flush interval |
| `CLICKHOUSE_MAX_INSERT_ATTEMPTS` | `3` | Maximum ClickHouse insert attempts |
| `CLICKHOUSE_RETRY_BASE_DELAY_SECONDS` | `0.5` | Exponential retry base delay |
| `CLICKHOUSE_CONNECT_TIMEOUT_SECONDS` | `5` | ClickHouse connection timeout |
| `CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS` | `10` | ClickHouse request timeout |
| `WORKER_INSTANCE_ID` | `worker-1` | Worker instance identifier |
| `WORKER_HTTP_HOST` | `0.0.0.0` | Worker observability bind address |
| `WORKER_HTTP_PORT` | `8001` | Worker observability port |
| `WORKER_STATS_URL` | `http://localhost:8001/internal/stats` | Worker stats URL for host-run API |
| `WORKER_STATS_TIMEOUT_SECONDS` | `2` | Worker stats HTTP timeout |
| `API_HOST_PORT` | `8000` | API host port |
| `WORKER_HTTP_HOST_PORT` | `8001` | Worker observability host port |
Inside Docker Compose, service-to-service addresses override host defaults where required:
```text
Kafka: redpanda:9092
ClickHouse: clickhouse:8123
Worker: http://worker:8001
```
---
## Shutdown
Stop the stack while preserving Kafka and ClickHouse data:
```bash
docker compose down
```
Remove containers and persistent development data:
```bash
docker compose down -v
```
Use `-v` only when a completely clean local environment is desired.