diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..43db9cd --- /dev/null +++ b/.dockerignore @@ -0,0 +1,7 @@ +.git +.env +.venv +venv +__pycache__ +*.pyc +.agents diff --git a/.env.example b/.env.example index d10c1c9..9b86c16 100644 --- a/.env.example +++ b/.env.example @@ -4,7 +4,8 @@ API_HOST=0.0.0.0 API_PORT=8000 API_QUEUE_MAX_SIZE=10000 -WORKER_STATS_URL=http://worker:8001/internal/stats +# Optional override; Compose otherwise derives this from WORKER_STATS_PORT. +# WORKER_STATS_URL=http://worker:8001/internal/stats # ========================================== # Processing Worker Configuration @@ -16,7 +17,7 @@ WORKER_STATS_PORT=8001 # ========================================== # Kafka / Redpanda Broker Configuration # ========================================== -KAFKA_BOOTSTRAP_SERVERS=kafka:9092 +KAFKA_BOOTSTRAP_SERVERS=redpanda:29092 KAFKA_RAW_TOPIC=cdr.raw KAFKA_DLQ_TOPIC=cdr.raw.dlq KAFKA_CONSUMER_GROUP=cdr-processing-group @@ -26,6 +27,7 @@ KAFKA_AUTO_OFFSET_RESET=earliest # ClickHouse Database Configuration # ========================================== CLICKHOUSE_HOST=clickhouse +# Published host HTTP port. Compose services always use clickhouse:8123. CLICKHOUSE_PORT=8123 CLICKHOUSE_USER=default CLICKHOUSE_PASSWORD= @@ -49,3 +51,14 @@ CH_RETRY_BACKOFF_FACTOR=2.0 # Observability & Logging # ========================================== LOG_LEVEL=INFO + +# Delivery and HTTP limits +KAFKA_DELIVERY_TIMEOUT_SECONDS=10 +API_MAX_BODY_BYTES=65536 +API_READ_TIMEOUT_SECONDS=10 +SHUTDOWN_TIMEOUT_SECONDS=45 +# Published infrastructure ports +KAFKA_PORT=9092 +KAFKA_ADVERTISED_HOST=localhost +REDPANDA_ADMIN_PORT=9644 +CLICKHOUSE_NATIVE_PORT=9000 diff --git a/app/common/bootstrap.py b/app/common/bootstrap.py new file mode 100644 index 0000000..8f26757 --- /dev/null +++ b/app/common/bootstrap.py @@ -0,0 +1,77 @@ +"""Idempotent infrastructure provisioning; never delete existing data.""" +import re +from pathlib import Path + +from app.common.config import config + +SCHEMA = Path(__file__).parents[1] / "worker" / "schema.sql" +SORT_KEY = "service_type, radio_access_type, cell_id, session_start_time, msisdn, cdr_id" + + +def identifier(value): + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", value): + raise ValueError(f"Invalid SQL identifier: {value!r}") + return value + + +def verify_table(client, database, table): + rows = client.query("SELECT engine, sorting_key FROM system.tables WHERE database={db:String} AND name={name:String}", + parameters={"db": database, "name": table}).result_rows + if not rows or rows[0][0] != "ReplacingMergeTree" or rows[0][1] != SORT_KEY: + raise RuntimeError("Incompatible ClickHouse table. Use the explicit assessment reset in README; no data was deleted.") + expected = re.findall(r"^ (\w+)\s+(UUID|String|LowCardinality\(String\)|UInt64|UInt32|DateTime64\(3\)|DateTime|IPv4)[,\s]", SCHEMA.read_text(), re.M) + actual = client.query("SELECT name, type FROM system.columns WHERE database={db:String} AND table={name:String} ORDER BY position", + parameters={"db": database, "name": table}).result_rows + if list(map(tuple, actual)) != expected: + raise RuntimeError("Incompatible ClickHouse columns; use the explicit assessment reset in README.") + ddl = client.command(f"SHOW CREATE TABLE {identifier(database)}.{identifier(table)}") + if "ReplacingMergeTree(processed_at)" not in ddl or "toYYYYMM(session_start_time)" not in ddl or "TTL session_start_time + toIntervalDay(90)" not in ddl: + raise RuntimeError("Incompatible ClickHouse engine/partition/TTL; use the explicit assessment reset in README.") + + +def provision_database(): + import clickhouse_connect + ch = config.clickhouse + database, table = identifier(ch.database), identifier(ch.table) + client = clickhouse_connect.get_client(host=ch.host, port=ch.port, username=ch.user, + password=ch.password, database="default") + try: + existing = client.query("SELECT count() FROM system.tables WHERE database={db:String} AND name={name:String}", + parameters={"db": database, "name": table}).first_row[0] + if not existing: + for statement in SCHEMA.read_text().format(database=database, table=table).split(";"): + if statement.strip(): + client.command(statement) + verify_table(client, database, table) + finally: + client.close() + + +def provision_topics(): + from confluent_kafka.admin import AdminClient, NewTopic, ConfigResource, ResourceType + admin = AdminClient({"bootstrap.servers": config.kafka.bootstrap_servers}) + required = {config.kafka.raw_topic: (3, {"retention.ms": "86400000", "compression.type": "lz4"}), + config.kafka.dlq_topic: (1, {"retention.ms": "604800000"})} + if len(required) != 2: + raise ValueError("Raw and DLQ topic names must differ") + existing = admin.list_topics(timeout=10).topics + missing = [NewTopic(name, num_partitions=count, replication_factor=1, config=settings) + for name, (count, settings) in required.items() if name not in existing] + for future in admin.create_topics(missing).values() if missing else []: + future.result(timeout=15) + metadata = admin.list_topics(timeout=10) + for name, (count, settings) in required.items(): + topic = metadata.topics[name] + if topic.error or len(topic.partitions) != count or any(len(p.replicas) != 1 for p in topic.partitions.values()): + raise RuntimeError(f"Incompatible topic partitions/replication: {name}") + resource = ConfigResource(ResourceType.TOPIC, name) + actual = admin.describe_configs([resource])[resource].result(timeout=10) + for key, value in settings.items(): + if actual[key].value != value: + raise RuntimeError(f"Topic {name} requires {key}={value}; found {actual[key].value}") + + +if __name__ == "__main__": + provision_topics() + provision_database() + print("Topics and ClickHouse schema verified") diff --git a/docker/clickhouse/init.sql b/app/worker/schema.sql similarity index 87% rename from docker/clickhouse/init.sql rename to app/worker/schema.sql index e39076a..b4f82d7 100644 --- a/docker/clickhouse/init.sql +++ b/app/worker/schema.sql @@ -1,10 +1,10 @@ -- ClickHouse Initialization Script for Telecom CDR Ingestion Pipeline -- Database: telecom --- Table: telecom.cdr_records +-- Table: {database}.{table} -CREATE DATABASE IF NOT EXISTS telecom; +CREATE DATABASE IF NOT EXISTS {database}; -CREATE TABLE IF NOT EXISTS telecom.cdr_records ( +CREATE TABLE IF NOT EXISTS {database}.{table} ( -- Identifiers cdr_id UUID, raw_id String, @@ -36,6 +36,6 @@ CREATE TABLE IF NOT EXISTS telecom.cdr_records ( ) ENGINE = ReplacingMergeTree(processed_at) PARTITION BY toYYYYMM(session_start_time) -ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn) +ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn, cdr_id) TTL session_start_time + INTERVAL 90 DAY SETTINGS index_granularity = 8192; diff --git a/docker-compose.yml b/docker-compose.yml index 30be86a..e6dc237 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,10 +1,37 @@ +x-runtime: &runtime + API_HOST: ${API_HOST:-0.0.0.0} + API_PORT: ${API_PORT:-8000} + API_QUEUE_MAX_SIZE: ${API_QUEUE_MAX_SIZE:-10000} + API_MAX_BODY_BYTES: ${API_MAX_BODY_BYTES:-65536} + API_READ_TIMEOUT_SECONDS: ${API_READ_TIMEOUT_SECONDS:-10} + SHUTDOWN_TIMEOUT_SECONDS: ${SHUTDOWN_TIMEOUT_SECONDS:-45} + WORKER_STATS_HOST: ${WORKER_STATS_HOST:-0.0.0.0} + WORKER_STATS_PORT: ${WORKER_STATS_PORT:-8001} + WORKER_STATS_URL: ${WORKER_STATS_URL:-http://worker:${WORKER_STATS_PORT:-8001}/internal/stats} + WORKER_INSTANCE: ${WORKER_INSTANCE:-worker-1} + KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_BOOTSTRAP_SERVERS:-redpanda:29092} + 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-group} + KAFKA_AUTO_OFFSET_RESET: ${KAFKA_AUTO_OFFSET_RESET:-earliest} + KAFKA_DELIVERY_TIMEOUT_SECONDS: ${KAFKA_DELIVERY_TIMEOUT_SECONDS:-10} + CLICKHOUSE_HOST: ${CLICKHOUSE_HOST:-clickhouse} + # CLICKHOUSE_PORT is the published host port; the server listens on 8123 internally. + CLICKHOUSE_PORT: 8123 + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-default} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-} + CLICKHOUSE_DATABASE: ${CLICKHOUSE_DATABASE:-telecom} + CLICKHOUSE_TABLE: ${CLICKHOUSE_TABLE:-cdr_records} + BATCH_SIZE: ${BATCH_SIZE:-1000} + BATCH_FLUSH_INTERVAL_SECONDS: ${BATCH_FLUSH_INTERVAL_SECONDS:-2} + CH_RETRY_MAX_ATTEMPTS: ${CH_RETRY_MAX_ATTEMPTS:-3} + CH_RETRY_INITIAL_DELAY_MS: ${CH_RETRY_INITIAL_DELAY_MS:-500} + CH_RETRY_BACKOFF_FACTOR: ${CH_RETRY_BACKOFF_FACTOR:-2} + LOG_LEVEL: ${LOG_LEVEL:-INFO} + services: - # ========================================== - # Message Broker: Redpanda (Kafka-compatible) - # ========================================== redpanda: image: docker.redpanda.com/redpandadata/redpanda:v24.1.8 - container_name: redpanda command: - redpanda start - --smp 1 @@ -13,126 +40,83 @@ services: - --overprovisioned - --node-id 0 - --check=false - - --kafka-addr PLAINTEXT://0.0.0.0:29092,OUTSIDE://0.0.0.0:9092 - - --advertise-kafka-addr PLAINTEXT://redpanda:29092,OUTSIDE://localhost:9092 + - --kafka-addr PLAINTEXT://0.0.0.0:29092,OUTSIDE://0.0.0.0:${KAFKA_PORT:-9092} + - --advertise-kafka-addr PLAINTEXT://redpanda:29092,OUTSIDE://${KAFKA_ADVERTISED_HOST:-localhost}:${KAFKA_PORT:-9092} ports: - - "9092:9092" - - "9644:9644" + - "${KAFKA_PORT:-9092}:${KAFKA_PORT:-9092}" + - "${REDPANDA_ADMIN_PORT:-9644}:9644" + volumes: + - kafka-data:/var/lib/redpanda/data healthcheck: - test: ["CMD", "rpk", "cluster", "health"] + test: ["CMD", "rpk", "cluster", "health", "--exit-when-healthy"] interval: 5s timeout: 5s retries: 12 - start_period: 10s - # Automated topic provisioner ensuring exact partitions & compression - topic-init: - image: docker.redpanda.com/redpandadata/redpanda:v24.1.8 - container_name: redpanda-topic-init + clickhouse: + image: clickhouse/clickhouse-server:24.3-alpine + environment: + CLICKHOUSE_USER: ${CLICKHOUSE_USER:-default} + CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-} + CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1 + ports: + - "${CLICKHOUSE_PORT:-8123}:8123" + - "${CLICKHOUSE_NATIVE_PORT:-9000}:9000" + volumes: + - clickhouse-data:/var/lib/clickhouse + healthcheck: + test: ["CMD-SHELL", "clickhouse-client --user \"$$CLICKHOUSE_USER\" --password \"$$CLICKHOUSE_PASSWORD\" --query 'SELECT 1'"] + interval: 5s + timeout: 5s + retries: 12 + + init: + build: + context: . + dockerfile: Dockerfile.worker + command: ["python", "-m", "app.common.bootstrap"] + environment: *runtime depends_on: redpanda: condition: service_healthy - entrypoint: /bin/sh - command: - - -c - - | - rpk topic create cdr.raw -p 3 -r 1 -c compression.type=lz4 -c retention.ms=86400000 --brokers=redpanda:29092 || true - rpk topic create cdr.raw.dlq -p 1 -r 1 -c retention.ms=604800000 --brokers=redpanda:29092 || true + clickhouse: + condition: service_healthy restart: "no" - # ========================================== - # Analytics Database: ClickHouse - # ========================================== - clickhouse: - image: clickhouse/clickhouse-server:24.3-alpine - container_name: clickhouse - ports: - - "8123:8123" - - "9000:9000" - environment: - CLICKHOUSE_DB: telecom - CLICKHOUSE_USER: default - CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1 - volumes: - - ./docker/clickhouse/init.sql:/docker-entrypoint-initdb.d/init.sql:ro - healthcheck: - test: ["CMD", "clickhouse-client", "--query", "SELECT 1"] - interval: 5s - timeout: 5s - retries: 10 - start_period: 10s - - # ========================================== - # Service 1: Ingestion API Gate - # ========================================== api: build: context: . dockerfile: Dockerfile.api - container_name: ingestion-api + environment: *runtime ports: - - "8000:8000" - environment: - API_HOST: 0.0.0.0 - API_PORT: 8000 - API_QUEUE_MAX_SIZE: 10000 - WORKER_STATS_URL: http://worker:8001/internal/stats - KAFKA_BOOTSTRAP_SERVERS: redpanda:29092 - KAFKA_RAW_TOPIC: cdr.raw - LOG_LEVEL: INFO + - "${API_PORT:-8000}:${API_PORT:-8000}" + stop_grace_period: 60s depends_on: - redpanda: - condition: service_healthy - topic-init: + init: condition: service_completed_successfully healthcheck: - test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"] + test: ["CMD", "python", "-c", "import os, urllib.request; urllib.request.urlopen('http://127.0.0.1:' + os.environ['API_PORT'] + '/health')"] interval: 5s timeout: 5s - retries: 10 - start_period: 5s + retries: 12 - # ========================================== - # Service 2: Processing Worker - # ========================================== worker: build: context: . dockerfile: Dockerfile.worker - container_name: processing-worker + environment: *runtime ports: - - "8001:8001" - environment: - WORKER_INSTANCE: worker-1 - WORKER_STATS_HOST: 0.0.0.0 - WORKER_STATS_PORT: 8001 - KAFKA_BOOTSTRAP_SERVERS: redpanda:29092 - KAFKA_RAW_TOPIC: cdr.raw - KAFKA_DLQ_TOPIC: cdr.raw.dlq - KAFKA_CONSUMER_GROUP: cdr-processing-group - KAFKA_AUTO_OFFSET_RESET: earliest - CLICKHOUSE_HOST: clickhouse - CLICKHOUSE_PORT: 8123 - CLICKHOUSE_USER: default - CLICKHOUSE_PASSWORD: "" - CLICKHOUSE_DATABASE: telecom - CLICKHOUSE_TABLE: cdr_records - BATCH_SIZE: 1000 - BATCH_FLUSH_INTERVAL_SECONDS: 2 - CH_RETRY_MAX_ATTEMPTS: 3 - CH_RETRY_INITIAL_DELAY_MS: 500 - CH_RETRY_BACKOFF_FACTOR: 2.0 - LOG_LEVEL: INFO + - "${WORKER_STATS_PORT:-8001}:${WORKER_STATS_PORT:-8001}" + stop_grace_period: 60s depends_on: - redpanda: - condition: service_healthy - topic-init: + init: condition: service_completed_successfully - clickhouse: - condition: service_healthy healthcheck: - test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8001/health')"] + test: ["CMD", "python", "-c", "import os, urllib.request; urllib.request.urlopen('http://127.0.0.1:' + os.environ['WORKER_STATS_PORT'] + '/health')"] interval: 5s timeout: 5s - retries: 10 - start_period: 5s + retries: 12 + +volumes: + kafka-data: + clickhouse-data: diff --git a/scripts/init_db.sql b/scripts/init_db.sql deleted file mode 100644 index e39076a..0000000 --- a/scripts/init_db.sql +++ /dev/null @@ -1,41 +0,0 @@ --- ClickHouse Initialization Script for Telecom CDR Ingestion Pipeline --- Database: telecom --- Table: telecom.cdr_records - -CREATE DATABASE IF NOT EXISTS telecom; - -CREATE TABLE IF NOT EXISTS telecom.cdr_records ( - -- Identifiers - cdr_id UUID, - raw_id String, - - -- Subscriber - msisdn String, - imsi String, - - -- Session metadata - service_type LowCardinality(String), - radio_access_type LowCardinality(String), - cell_id LowCardinality(String), - - -- Traffic - bytes_uploaded UInt64, - bytes_downloaded UInt64, - total_bytes UInt64, - duration_seconds UInt32, - - -- Timing - session_start_time DateTime, - session_end_time DateTime, - - -- Network - source_ip IPv4, - - -- Audit - processed_at DateTime64(3) DEFAULT now() - -) ENGINE = ReplacingMergeTree(processed_at) -PARTITION BY toYYYYMM(session_start_time) -ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn) -TTL session_start_time + INTERVAL 90 DAY -SETTINGS index_granularity = 8192; diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py new file mode 100644 index 0000000..f3daa85 --- /dev/null +++ b/tests/test_bootstrap.py @@ -0,0 +1,19 @@ +import unittest +from unittest.mock import Mock + +from app.common.bootstrap import identifier, verify_table + + +class TestBootstrap(unittest.TestCase): + def test_identifiers_cannot_inject_sql(self): + self.assertEqual(identifier("cdr_records"), "cdr_records") + for value in ("a; DROP TABLE b", "a.b", "", "1table"): + with self.subTest(value=value), self.assertRaises(ValueError): + identifier(value) + + def test_old_sorting_key_fails_without_mutating_data(self): + client = Mock() + client.query.return_value.result_rows = [("ReplacingMergeTree", "service_type, radio_access_type, cell_id, session_start_time, msisdn")] + with self.assertRaisesRegex(RuntimeError, "reset"): + verify_table(client, "telecom", "cdr_records") + client.command.assert_not_called()