feat(common): complete Phase 1 - project config, structured logging, and ClickHouse DDL

.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

app/common/__init__.py

@@ -0,0 +1 @@
+# app/common package

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()

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

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;

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.
+- [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.
-- [ ] 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.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.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
 

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

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;

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()