Feature/flex data schema

.env.example

@@ -0,0 +1,12 @@
+INFLUX_PORT=8086
+INFLUX_ADMIN_USER=admin
+INFLUX_ADMIN_PASSWORD=changeme
+INFLUX_ORG=nwdaf
+INFLUX_BUCKET=raw_data
+INFLUX_TOKEN=your-super-secret-token
+
+CLICKHOUSE_HTTP_PORT=8123
+CLICKHOUSE_TCP_PORT=9000
+CLICKHOUSE_DB=analytics
+CLICKHOUSE_USER=default
+CLICKHOUSE_PASSWORD=changeme

.gitignore

@@ -5,7 +5,7 @@ __pycache__/
 
 # C extensions
 *.so
-
+confs/*.yml
 # Distribution / packaging
 .Python
 build/
@@ -182,9 +182,9 @@ cython_debug/
 .abstra/
 
 # Visual Studio Code
-#  Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
+#  Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
 #  that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
-#  and can be added to the global gitignore or merged into this file. However, if you prefer, 
+#  and can be added to the global gitignore or merged into this file. However, if you prefer,
 #  you could uncomment the following to ignore the entire vscode folder
 # .vscode/
 

confs/core_fields.yml.example

@@ -0,0 +1,2 @@
+timestamp: datetime
+cell_index: integer

confs/extra_fields.yml.example

@@ -0,0 +1,12 @@
+# Optional/allowed extra fields for raw data points
+
+datarate: float
+mean_latency: float
+rsrp: float
+sinr: float
+rsrq: float
+direction: string
+network: string
+cqi: float
+primary_bandwidth: float
+ul_bandwidth: float

confs/tag_fields.yml.example

@@ -0,0 +1,2 @@
+- cell_index
+- network

docker-compose.yml

@@ -20,7 +20,9 @@ services:
     restart: unless-stopped
 
   clickhouse:
-    image: clickhouse/clickhouse-server:latest
+    build:
+      context: .
+      dockerfile: docker/Dockerfile.clickhouse
     container_name: clickhouse
     ports:
       - "${CLICKHOUSE_HTTP_PORT}:${CLICKHOUSE_HTTP_PORT}"
@@ -41,29 +43,13 @@ services:
       timeout: 3s
       retries: 5
 
-  clickhouse-init:
-    build:
-      context: .
-      dockerfile: Dockerfile.clickhouse-init
-    container_name: clickhouse-init
-    depends_on:
-      clickhouse:
-        condition: service_healthy
-    environment:
-      - CLICKHOUSE_HOST=clickhouse
-      - CLICKHOUSE_USER=${CLICKHOUSE_USER}
-      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_PASSWORD}
-    networks:
-      - nwdaf-network
-    restart: "no"
-
   data-storage:
     container_name: data-storage
     ports:
       - "8000:8000"
     build:
       context: .
-      dockerfile: Dockerfile
+      dockerfile: docker/Dockerfile
     environment:
       - INFLUX_URL=http://influxdb:8086
       - INFLUX_TOKEN=${INFLUX_TOKEN}

Dockerfile → docker/Dockerfile

@@ -27,6 +27,7 @@ COPY requirements.txt ./
 RUN uv pip install --system -r requirements.txt
 
 COPY src/ ./src/
+COPY confs/ ./confs/
 COPY main.py .
 
 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

docker/Dockerfile.clickhouse

@@ -0,0 +1,3 @@
+FROM clickhouse/clickhouse-server:24.1.2.5
+
+COPY sql/ /docker-entrypoint-initdb.d/

main.py

@@ -1,12 +1,13 @@
-import os
 import asyncio
-from threading import Thread
+import os
 from contextlib import asynccontextmanager
+from threading import Thread
+
 from fastapi import FastAPI
 
+from src.configs import load_all
 from src.routers.v1 import v1_router
 from src.services.databases import ClickHouse, Influx
-
 from src.sink import KafkaSinkManager
 
 KAFKA_HOST = os.getenv("KAFKA_HOST", "localhost")
@@ -19,7 +20,7 @@ async def lifespan(app: FastAPI):
     # Initialize database connections (singleton, handles connection internally)
     Influx.service  # Access triggers lazy initialization
     ClickHouse.service  # Access triggers lazy initialization
-
+    load_all()
     sink_manager = KafkaSinkManager(KAFKA_HOST, KAFKA_PORT)
 
     def kafka_worker():
@@ -33,17 +34,10 @@ def kafka_worker():
         except Exception as e:
             print(f"Kafka worker crashed: {e}")
 
-    kafka_thread = Thread(
-        target=kafka_worker,
-        daemon=True,
-        name="kafka-sink-thread"
-    )
+    kafka_thread = Thread(target=kafka_worker, daemon=True, name="kafka-sink-thread")
     kafka_thread.start()
 
-    print(
-        f"API started (Kafka connecting in background to "
-        f"{KAFKA_HOST}:{KAFKA_PORT})"
-    )
+    print(f"API started (Kafka connecting in background to {KAFKA_HOST}:{KAFKA_PORT})")
 
     yield
 

sql/01_create_processed_latency_table.sql → sql/01_create_processed_table.sql

@@ -1,13 +1,10 @@
 CREATE DATABASE IF NOT EXISTS analytics;
-
-CREATE TABLE IF NOT EXISTS analytics.processed_latency
+CREATE TABLE IF NOT EXISTS analytics.processed
 (
-    window_start_time DateTime64(3),
-    window_end_time DateTime64(3),
-    window_duration_seconds Float64,
-
     cell_index Int32,
-    network String,
+
+    network Nullable(String),
+    data_type Nullable(String),  -- e.g., 'latency', 'anomaly', etc.
 
     rsrp_mean Nullable(Float64),
     rsrp_max Nullable(Float64),
@@ -37,7 +34,10 @@ CREATE TABLE IF NOT EXISTS analytics.processed_latency
     primary_bandwidth Nullable(Float64),
     ul_bandwidth Nullable(Float64),
 
-    sample_count Int32
+    sample_count Int32,
+    window_start_time DateTime64(3),
+    window_end_time DateTime64(3),
+    window_duration_seconds Float64
 )
 ENGINE = MergeTree
 ORDER BY (cell_index, window_start_time)

src/configs/__init__.py

@@ -0,0 +1,13 @@
+from .clickhouse_conf import ClickhouseConf
+from .conf import Conf
+from .influx_conf import InfluxConf
+from .schema_conf import SchemaConf
+
+__all__ = ["ClickhouseConf", "InfluxConf", "SchemaConf"]
+
+
+def load_all() -> None:
+    """Load all configs"""
+    config: Conf
+    for config in [ClickhouseConf, InfluxConf, SchemaConf]:
+        config.load()

src/configs/clickhouse_conf.py

@@ -5,10 +5,11 @@
 
 logger = logging.getLogger("Config")
 
+
 class ClickhouseConf(Conf):
-    host:     str
-    port:     int
-    user:     str
+    host: str
+    port: int
+    user: str
     password: str
 
     _instance = None
@@ -26,9 +27,9 @@ def __init__(self):
     @classmethod
     def load_env(cls, file: str = ".env") -> None:
 
-        cls.host     = os.getenv("CLICKHOUSE_HOST", "localhost")
-        cls.port     = int(os.getenv("CLICKHOUSE_PORT", "9000"))
-        cls.user     = os.getenv("CLICKHOUSE_USER", "default")
+        cls.host = os.getenv("CLICKHOUSE_HOST", "localhost")
+        cls.port = int(os.getenv("CLICKHOUSE_PORT", "9000"))
+        cls.user = os.getenv("CLICKHOUSE_USER", "default")
         cls.password = os.getenv("CLICKHOUSE_PASSWORD", "")
 
         cls._loaded = True

src/configs/conf.py

@@ -1,15 +1,19 @@
 class Conf:
     @classmethod
-    def load_env(cls,file:str=".env")->None:
+    def load_env(cls, file: str = ".env") -> None:
         """
         Load configuration from env
         """
         raise NotImplementedError
 
     @classmethod
-    def get(cls)->dict:
+    def get(cls) -> dict:
         """
         Get a dict with the loaded confs.
         If not loaded yet, then load in the function
         """
         raise NotImplementedError
+
+    @classmethod
+    def load(cls) -> None:
+        cls.load_env()