From 43df200eeb738d6aaaee801fdb25ee75fa045bae Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sat, 19 Sep 2026 13:12:04 +0330 Subject: [PATCH] feat(common): complete Phase 1 - project config, structured logging, and ClickHouse DDL --- .env.example | 51 +++++++++++ app/common/__init__.py | 1 + app/common/config.py | 83 ++++++++++++++++++ app/common/logging.py | 87 +++++++++++++++++++ docker/clickhouse/init.sql | 41 +++++++++ .../changes/telecom-cdr-pipeline/tasks.md | 6 +- requirements.txt | 7 ++ scripts/init_db.sql | 41 +++++++++ tests/test_logging.py | 50 +++++++++++ 9 files changed, 364 insertions(+), 3 deletions(-) create mode 100644 .env.example create mode 100644 app/common/__init__.py create mode 100644 app/common/config.py create mode 100644 app/common/logging.py create mode 100644 docker/clickhouse/init.sql create mode 100644 requirements.txt create mode 100644 scripts/init_db.sql create mode 100644 tests/test_logging.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..d10c1c9 --- /dev/null +++ b/.env.example @@ -0,0 +1,51 @@ +# ========================================== +# Ingestion API Configuration +# ========================================== +API_HOST=0.0.0.0 +API_PORT=8000 +API_QUEUE_MAX_SIZE=10000 +WORKER_STATS_URL=http://worker:8001/internal/stats + +# ========================================== +# Processing Worker Configuration +# ========================================== +WORKER_INSTANCE=worker-1 +WORKER_STATS_HOST=0.0.0.0 +WORKER_STATS_PORT=8001 + +# ========================================== +# Kafka / Redpanda Broker Configuration +# ========================================== +KAFKA_BOOTSTRAP_SERVERS=kafka:9092 +KAFKA_RAW_TOPIC=cdr.raw +KAFKA_DLQ_TOPIC=cdr.raw.dlq +KAFKA_CONSUMER_GROUP=cdr-processing-group +KAFKA_AUTO_OFFSET_RESET=earliest + +# ========================================== +# ClickHouse Database Configuration +# ========================================== +CLICKHOUSE_HOST=clickhouse +CLICKHOUSE_PORT=8123 +CLICKHOUSE_USER=default +CLICKHOUSE_PASSWORD= +CLICKHOUSE_DATABASE=telecom +CLICKHOUSE_TABLE=cdr_records + +# ========================================== +# Micro-Batching Settings +# ========================================== +BATCH_SIZE=1000 +BATCH_FLUSH_INTERVAL_SECONDS=2 + +# ========================================== +# ClickHouse Timeout Retry Policy +# ========================================== +CH_RETRY_MAX_ATTEMPTS=3 +CH_RETRY_INITIAL_DELAY_MS=500 +CH_RETRY_BACKOFF_FACTOR=2.0 + +# ========================================== +# Observability & Logging +# ========================================== +LOG_LEVEL=INFO diff --git a/app/common/__init__.py b/app/common/__init__.py new file mode 100644 index 0000000..d8189a9 --- /dev/null +++ b/app/common/__init__.py @@ -0,0 +1 @@ +# app/common package diff --git a/app/common/config.py b/app/common/config.py new file mode 100644 index 0000000..01bc1db --- /dev/null +++ b/app/common/config.py @@ -0,0 +1,83 @@ +import os +from dataclasses import dataclass + + +def get_env_str(key: str, default: str) -> str: + return os.getenv(key, default) + + +def get_env_int(key: str, default: int) -> int: + val = os.getenv(key) + if val is None or not val.strip(): + return default + return int(val) + + +def get_env_float(key: str, default: float) -> float: + val = os.getenv(key) + if val is None or not val.strip(): + return default + return float(val) + + +@dataclass(frozen=True) +class APIConfig: + host: str = get_env_str("API_HOST", "0.0.0.0") + port: int = get_env_int("API_PORT", 8000) + queue_max_size: int = get_env_int("API_QUEUE_MAX_SIZE", 10000) + worker_stats_url: str = get_env_str( + "WORKER_STATS_URL", "http://worker:8001/internal/stats" + ) + + +@dataclass(frozen=True) +class WorkerConfig: + instance_id: str = get_env_str("WORKER_INSTANCE", "worker-1") + stats_host: str = get_env_str("WORKER_STATS_HOST", "0.0.0.0") + stats_port: int = get_env_int("WORKER_STATS_PORT", 8001) + + +@dataclass(frozen=True) +class KafkaConfig: + bootstrap_servers: str = get_env_str("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092") + raw_topic: str = get_env_str("KAFKA_RAW_TOPIC", "cdr.raw") + dlq_topic: str = get_env_str("KAFKA_DLQ_TOPIC", "cdr.raw.dlq") + consumer_group: str = get_env_str("KAFKA_CONSUMER_GROUP", "cdr-processing-group") + auto_offset_reset: str = get_env_str("KAFKA_AUTO_OFFSET_RESET", "earliest") + + +@dataclass(frozen=True) +class ClickHouseConfig: + host: str = get_env_str("CLICKHOUSE_HOST", "localhost") + port: int = get_env_int("CLICKHOUSE_PORT", 8123) + user: str = get_env_str("CLICKHOUSE_USER", "default") + password: str = get_env_str("CLICKHOUSE_PASSWORD", "") + database: str = get_env_str("CLICKHOUSE_DATABASE", "telecom") + table: str = get_env_str("CLICKHOUSE_TABLE", "cdr_records") + + +@dataclass(frozen=True) +class BatchConfig: + size: int = get_env_int("BATCH_SIZE", 1000) + flush_interval_seconds: float = get_env_float("BATCH_FLUSH_INTERVAL_SECONDS", 2.0) + + +@dataclass(frozen=True) +class RetryConfig: + max_attempts: int = get_env_int("CH_RETRY_MAX_ATTEMPTS", 3) + initial_delay_ms: int = get_env_int("CH_RETRY_INITIAL_DELAY_MS", 500) + backoff_factor: float = get_env_float("CH_RETRY_BACKOFF_FACTOR", 2.0) + + +@dataclass(frozen=True) +class AppConfig: + api: APIConfig = APIConfig() + worker: WorkerConfig = WorkerConfig() + kafka: KafkaConfig = KafkaConfig() + clickhouse: ClickHouseConfig = ClickHouseConfig() + batch: BatchConfig = BatchConfig() + retry: RetryConfig = RetryConfig() + log_level: str = get_env_str("LOG_LEVEL", "INFO") + + +config = AppConfig() diff --git a/app/common/logging.py b/app/common/logging.py new file mode 100644 index 0000000..24227e9 --- /dev/null +++ b/app/common/logging.py @@ -0,0 +1,87 @@ +import json +import logging +import sys +from datetime import datetime, timezone +from typing import Any, Dict, Optional + + +class StructuredJsonFormatter(logging.Formatter): + """ + JSON log formatter emitting structured log records with standard telecom pipeline fields: + - timestamp (ISO 8601 UTC) + - level (INFO, WARN, ERROR, DEBUG) + - service (service name) + - raw_id (record identifier or None) + - event (message string) + """ + + def __init__(self, service_name: str): + super().__init__() + self.service_name = service_name + + def format(self, record: logging.LogRecord) -> str: + log_entry: Dict[str, Any] = { + "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "level": record.levelname, + "service": getattr(record, "service", self.service_name), + "raw_id": getattr(record, "raw_id", None), + "event": record.getMessage(), + } + + # Include exception info if present + if record.exc_info: + log_entry["exception"] = self.formatException(record.exc_info) + + # Include additional extra context fields if present + for key, value in record.__dict__.items(): + if key not in { + "name", + "msg", + "args", + "levelname", + "levelno", + "pathname", + "filename", + "module", + "exc_info", + "exc_text", + "stack_info", + "lineno", + "funcName", + "created", + "msecs", + "relativeCreated", + "thread", + "threadName", + "processName", + "process", + "raw_id", + "service", + }: + log_entry[key] = value + + return json.dumps(log_entry, default=str) + + +def setup_logger( + service_name: str, level: Optional[str] = None +) -> logging.Logger: + """ + Configures and returns a logger configured with StructuredJsonFormatter on stdout. + """ + if level is None: + from app.common.config import config + + level = config.log_level + + logger = logging.getLogger(service_name) + logger.setLevel(getattr(logging, level.upper(), logging.INFO)) + logger.propagate = False + + # Avoid duplicate handlers if setup is called multiple times + if not logger.handlers: + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(StructuredJsonFormatter(service_name=service_name)) + logger.addHandler(handler) + + return logger diff --git a/docker/clickhouse/init.sql b/docker/clickhouse/init.sql new file mode 100644 index 0000000..e39076a --- /dev/null +++ b/docker/clickhouse/init.sql @@ -0,0 +1,41 @@ +-- 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/openspec/changes/telecom-cdr-pipeline/tasks.md b/openspec/changes/telecom-cdr-pipeline/tasks.md index 319a9b1..ca56c00 100644 --- a/openspec/changes/telecom-cdr-pipeline/tasks.md +++ b/openspec/changes/telecom-cdr-pipeline/tasks.md @@ -1,8 +1,8 @@ ## 1. Project Setup & Common Utilities -- [ ] 1.1 Create project structure, `requirements.txt` (with `confluent-kafka`, `clickhouse-connect`, `phonenumbers`), and `.env.example` defining all environment variables and verify file presence. -- [ ] 1.2 Implement standard library structured JSON logging module emitting `timestamp`, `level`, `service`, `raw_id`, and `event`, and verify output formatting with unit test. -- [ ] 1.3 Create idempotent ClickHouse initialization script (`init.sql`) creating database `telecom` and `ReplacingMergeTree` table `cdr_records` with TTL and indexing settings, and verify schema correctness. +- [x] 1.1 Create project structure, `requirements.txt` (with `confluent-kafka`, `clickhouse-connect`, `phonenumbers`), and `.env.example` defining all environment variables and verify file presence. +- [x] 1.2 Implement standard library structured JSON logging module emitting `timestamp`, `level`, `service`, `raw_id`, and `event`, and verify output formatting with unit test. +- [x] 1.3 Create idempotent ClickHouse initialization script (`init.sql`) creating database `telecom` and `ReplacingMergeTree` table `cdr_records` with TTL and indexing settings, and verify schema correctness. ## 2. Framework-less Async Ingestion API diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..5ff230f --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +# Core Dependencies (Pure Python / Driver Libraries - No Web Frameworks) +confluent-kafka==2.8.2 +clickhouse-connect==0.8.15 +phonenumbers==8.13.55 + +# Testing & Development +pytest==8.3.5 diff --git a/scripts/init_db.sql b/scripts/init_db.sql new file mode 100644 index 0000000..e39076a --- /dev/null +++ b/scripts/init_db.sql @@ -0,0 +1,41 @@ +-- 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_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..bb72b09 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,50 @@ +import io +import json +import logging +import unittest +from app.common.logging import StructuredJsonFormatter, setup_logger + + +class TestStructuredJsonFormatter(unittest.TestCase): + def test_log_format_structure(self): + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setFormatter(StructuredJsonFormatter(service_name="test-service")) + + logger = logging.getLogger("test_logger") + logger.setLevel(logging.INFO) + logger.addHandler(handler) + + logger.info("Batch processed successfully", extra={"raw_id": "raw-12345"}) + + output = stream.getvalue().strip() + data = json.loads(output) + + self.assertIn("timestamp", data) + self.assertEqual(data["level"], "INFO") + self.assertEqual(data["service"], "test-service") + self.assertEqual(data["raw_id"], "raw-12345") + self.assertEqual(data["event"], "Batch processed successfully") + + def test_log_format_null_raw_id(self): + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setFormatter(StructuredJsonFormatter(service_name="test-service")) + + logger = logging.getLogger("test_logger_null") + logger.setLevel(logging.INFO) + logger.addHandler(handler) + + logger.info("Server started on port 8000") + + output = stream.getvalue().strip() + data = json.loads(output) + + self.assertIn("timestamp", data) + self.assertEqual(data["level"], "INFO") + self.assertIsNone(data["raw_id"]) + self.assertEqual(data["event"], "Server started on port 8000") + + +if __name__ == "__main__": + unittest.main()