diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..52e7a35 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,22 @@ +.git +.gitignore + +.venv +venv + +__pycache__ +*.py[cod] + +.pytest_cache +.coverage +htmlcov + +.env + +.vscode +.idea + +tests +scripts + +README.md \ No newline at end of file diff --git a/.env.sample b/.env.sample index 6a52874..21e3c15 100644 --- a/.env.sample +++ b/.env.sample @@ -100,3 +100,11 @@ WORKER_HTTP_PORT=8001 # API uses this while both services run directly on the host. WORKER_STATS_URL=http://localhost:8001/internal/stats WORKER_STATS_TIMEOUT_SECONDS=2 + + +# ========================= +# Docker Host Ports +# ========================= + +API_HOST_PORT=8000 +WORKER_HTTP_HOST_PORT=8001 diff --git a/Dockerfile.api b/Dockerfile.api new file mode 100644 index 0000000..be53105 --- /dev/null +++ b/Dockerfile.api @@ -0,0 +1,24 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 + +WORKDIR /app + +COPY pyproject.toml ./ +COPY src ./src + +RUN pip install --upgrade pip \ + && pip install . + +RUN useradd \ + --create-home \ + --uid 10001 \ + appuser + +USER appuser + +EXPOSE 8000 + +CMD ["uvicorn", "core.api.main:app", "--host", "0.0.0.0", "--port", "8000"] \ No newline at end of file diff --git a/Dockerfile.worker b/Dockerfile.worker new file mode 100644 index 0000000..eb67e27 --- /dev/null +++ b/Dockerfile.worker @@ -0,0 +1,24 @@ +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 + +WORKDIR /app + +COPY pyproject.toml ./ +COPY src ./src + +RUN pip install --upgrade pip \ + && pip install . + +RUN useradd \ + --create-home \ + --uid 10001 \ + appuser + +USER appuser + +EXPOSE 8001 + +CMD ["python", "-m", "core.worker.main"] \ No newline at end of file diff --git a/README.md b/README.md index 8ad1861..373b703 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,11 @@ worker processing is not connected yet. * Batch size and in-flight buffer metrics * Kafka consumer lag by partition * Aggregated human-readable `/stats` endpoint +* Fully containerized API and processing worker +* Health-based Docker Compose startup ordering +* Automatic Kafka topic and ClickHouse schema provisioning +* Persistent Redpanda and ClickHouse storage +* Complete stack startup with `docker compose up --build` ## Project Structure @@ -170,10 +175,10 @@ This includes: * ClickHouse connection settings * Worker batch size and flush interval -## Running Infrastructure +## Running Pipeline ```bash -docker compose up -d +docker compose up -d --build ``` Verify services: @@ -190,28 +195,23 @@ redpanda-init exited successfully ClickHouse healthy ``` -## Running the API +### Available endpoints: -```bash -uvicorn core.api.main:app --reload --host 0.0.0.0 --port 8000 -``` - -Available endpoints: - -API: +#### API: ```text -GET /health -GET /metrics -GET /stats -POST /api/v1/cdr/raw +GET http://localhost:8000/health +GET http://localhost:8000/metrics +GET http://localhost:8000/stats +POST http://localhost:8000/api/v1/cdr/raw ``` -Worker: +#### Worker: ```text -GET /metrics -GET /internal/stats +GET http://localhost:8001/health +GET http://localhost:8001/metrics +GET http://localhost:8001/internal/stats ``` A valid CDR currently returns: @@ -227,14 +227,12 @@ Invalid payloads return structured `400` responses. ## Testing -Run: +Once the full stack starts cleanly, a real CDR reaches ClickHouse, and: ```bash pytest -v ``` -The current test suite covers domain validation, normalization, deterministic ID generation, and API request validation. - ## Delivery Semantics The worker uses at-least-once delivery. diff --git a/docker-compose.yml b/docker-compose.yml index 13c41f5..4ec5039 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,34 +2,23 @@ name: telecom-cdr-pipeline services: redpanda: - image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-latest} - + image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-v26.2.3} container_name: telecom-redpanda command: - redpanda - start - - # Kafka listeners: - # 9092 -> other Docker containers - # 19092 -> applications running on our host machine - --kafka-addr=internal://0.0.0.0:9092,external://0.0.0.0:19092 - - # These are the addresses Redpanda tells Kafka clients to use. - --advertise-kafka-addr=internal://redpanda:9092,external://localhost:19092 - - # Internal Redpanda node communication. - --rpc-addr=redpanda:33145 - --advertise-rpc-addr=redpanda:33145 - - # Suitable settings for local Docker development. - --mode=dev-container - --smp=1 - --default-log-level=info ports: - "19092:19092" - - "9644:9644" # Redpanda's Admin API + - "9644:9644" volumes: - redpanda-data:/var/lib/redpanda/data @@ -48,28 +37,34 @@ services: retries: 10 start_period: 10s + restart: unless-stopped + + redpanda-init: - image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-latest} + image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-v26.2.3} + container_name: telecom-redpanda-init depends_on: redpanda: condition: service_healthy environment: - KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS} + KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS:-redpanda:9092} - KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC} - KAFKA_RAW_TOPIC_PARTITIONS: ${KAFKA_RAW_TOPIC_PARTITIONS} - KAFKA_RAW_TOPIC_REPLICATION_FACTOR: ${KAFKA_RAW_TOPIC_REPLICATION_FACTOR} - KAFKA_RAW_TOPIC_RETENTION_MS: ${KAFKA_RAW_TOPIC_RETENTION_MS} - KAFKA_RAW_TOPIC_COMPRESSION: ${KAFKA_RAW_TOPIC_COMPRESSION} + KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC:-cdr.raw} + KAFKA_RAW_TOPIC_PARTITIONS: ${KAFKA_RAW_TOPIC_PARTITIONS:-3} + KAFKA_RAW_TOPIC_REPLICATION_FACTOR: ${KAFKA_RAW_TOPIC_REPLICATION_FACTOR:-1} + KAFKA_RAW_TOPIC_RETENTION_MS: ${KAFKA_RAW_TOPIC_RETENTION_MS:-86400000} + KAFKA_RAW_TOPIC_COMPRESSION: ${KAFKA_RAW_TOPIC_COMPRESSION:-lz4} - KAFKA_DLQ_TOPIC: ${KAFKA_DLQ_TOPIC} - KAFKA_DLQ_TOPIC_PARTITIONS: ${KAFKA_DLQ_TOPIC_PARTITIONS} - KAFKA_DLQ_TOPIC_REPLICATION_FACTOR: ${KAFKA_DLQ_TOPIC_REPLICATION_FACTOR} - KAFKA_DLQ_TOPIC_RETENTION_MS: ${KAFKA_DLQ_TOPIC_RETENTION_MS} + KAFKA_DLQ_TOPIC: ${KAFKA_DLQ_TOPIC:-cdr.raw.dlq} + KAFKA_DLQ_TOPIC_PARTITIONS: ${KAFKA_DLQ_TOPIC_PARTITIONS:-1} + KAFKA_DLQ_TOPIC_REPLICATION_FACTOR: ${KAFKA_DLQ_TOPIC_REPLICATION_FACTOR:-1} + KAFKA_DLQ_TOPIC_RETENTION_MS: ${KAFKA_DLQ_TOPIC_RETENTION_MS:-604800000} - entrypoint: ["/bin/bash", "-c"] + entrypoint: + - /bin/bash + - -c command: - | @@ -92,15 +87,18 @@ services: echo "Kafka topics initialized successfully." - clickhouse: - image: clickhouse/clickhouse-server:${CLICKHOUSE_VERSION:-latest} + restart: "no" + + clickhouse: + image: clickhouse/clickhouse-server:${CLICKHOUSE_VERSION:-26.8.7.19} container_name: telecom-clickhouse environment: - CLICKHOUSE_DB: ${CLICKHOUSE_DATABASE} - CLICKHOUSE_USER: ${CLICKHOUSE_USER} - CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD} + CLICKHOUSE_DB: ${CLICKHOUSE_DATABASE:-telecom} + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-telecom} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-telecom_dev_password} + CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1 CLICKHOUSE_ALWAYS_RUN_INITDB_SCRIPTS: 1 @@ -127,6 +125,128 @@ services: retries: 10 start_period: 10s + restart: unless-stopped + + + api: + build: + context: . + dockerfile: Dockerfile.api + + container_name: telecom-api + + depends_on: + redpanda: + condition: service_healthy + + redpanda-init: + condition: service_completed_successfully + + environment: + KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS:-redpanda:9092} + KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC:-cdr.raw} + + API_QUEUE_SIZE: ${API_QUEUE_SIZE:-10000} + + KAFKA_DELIVERY_TIMEOUT_SECONDS: ${KAFKA_DELIVERY_TIMEOUT_SECONDS:-4} + KAFKA_MESSAGE_TIMEOUT_MS: ${KAFKA_MESSAGE_TIMEOUT_MS:-3000} + + KAFKA_LINGER_MS: ${KAFKA_LINGER_MS:-5} + KAFKA_BATCH_SIZE: ${KAFKA_BATCH_SIZE:-131072} + KAFKA_PRODUCER_MAX_MESSAGES: ${KAFKA_PRODUCER_MAX_MESSAGES:-100000} + + WORKER_STATS_URL: http://worker:8001/internal/stats + WORKER_STATS_TIMEOUT_SECONDS: ${WORKER_STATS_TIMEOUT_SECONDS:-2} + + ports: + - "${API_HOST_PORT:-8000}:8000" + + healthcheck: + test: + [ + "CMD", + "python", + "-c", + "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=2)" + ] + interval: 5s + timeout: 3s + retries: 10 + start_period: 5s + + restart: unless-stopped + + + worker: + build: + context: . + dockerfile: Dockerfile.worker + + container_name: telecom-worker + + depends_on: + redpanda: + condition: service_healthy + + redpanda-init: + condition: service_completed_successfully + + clickhouse: + condition: service_healthy + + environment: + KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS:-redpanda:9092} + + KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC:-cdr.raw} + KAFKA_DLQ_TOPIC: ${KAFKA_DLQ_TOPIC:-cdr.raw.dlq} + + KAFKA_CONSUMER_GROUP: ${KAFKA_CONSUMER_GROUP:-cdr-processing-workers} + KAFKA_AUTO_OFFSET_RESET: ${KAFKA_AUTO_OFFSET_RESET:-earliest} + KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS: ${KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS:-1} + + KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS: ${KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS:-5} + KAFKA_MESSAGE_TIMEOUT_MS: ${KAFKA_MESSAGE_TIMEOUT_MS:-3000} + + WORKER_INSTANCE_ID: ${WORKER_INSTANCE_ID:-worker-1} + + WORKER_HTTP_HOST: 0.0.0.0 + WORKER_HTTP_PORT: 8001 + + CLICKHOUSE_HOST: ${CLICKHOUSE_INTERNAL_HOST:-clickhouse} + CLICKHOUSE_PORT: ${CLICKHOUSE_PORT:-8123} + + CLICKHOUSE_DATABASE: ${CLICKHOUSE_DATABASE:-telecom} + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-telecom} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-telecom_dev_password} + + BATCH_SIZE: ${BATCH_SIZE:-1000} + BATCH_FLUSH_INTERVAL_SECONDS: ${BATCH_FLUSH_INTERVAL_SECONDS:-2} + + CLICKHOUSE_MAX_INSERT_ATTEMPTS: ${CLICKHOUSE_MAX_INSERT_ATTEMPTS:-3} + CLICKHOUSE_RETRY_BASE_DELAY_SECONDS: ${CLICKHOUSE_RETRY_BASE_DELAY_SECONDS:-0.5} + + CLICKHOUSE_CONNECT_TIMEOUT_SECONDS: ${CLICKHOUSE_CONNECT_TIMEOUT_SECONDS:-5} + CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS: ${CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS:-10} + + ports: + - "${WORKER_HTTP_HOST_PORT:-8001}:8001" + + healthcheck: + test: + [ + "CMD", + "python", + "-c", + "import urllib.request; urllib.request.urlopen('http://localhost:8001/health', timeout=2)" + ] + interval: 5s + timeout: 3s + retries: 10 + start_period: 5s + + restart: unless-stopped + + volumes: redpanda-data: - clickhouse-data: + clickhouse-data: \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 0cc6831..5035cc4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,12 +11,12 @@ dependencies = [ "pydantic-settings>=2.0,<3.0", "clickhouse-connect>=1.8,<2.0", "prometheus-client>=0.26,<1.0", + "httpx2>=2.4,<3.0", ] [project.optional-dependencies] dev = [ "pytest>=8.0", - "httpx2>=2.4,<3.0", ] [build-system] diff --git a/src/core/worker/observability.py b/src/core/worker/observability.py index 31ea702..83ee53e 100644 --- a/src/core/worker/observability.py +++ b/src/core/worker/observability.py @@ -23,6 +23,15 @@ def create_worker_observability_app( version="0.1.0", ) + @app.get( + "/health", + include_in_schema=False, + ) + async def health() -> dict[str, str]: + return { + "status": "ok", + } + @app.get( "/metrics", include_in_schema=False,