Merge pull request #15 from ATNoG/hotfix/kafkabridge_start

run kafkabridge on another  thread

Author: alxmra

main.py

@@ -1,28 +1,46 @@
 import os
+import asyncio
+from threading import Thread
 from contextlib import asynccontextmanager
 from fastapi import FastAPI
+
 from src.routers.v1 import v1_router
 from src.routers.v1.latency import ClickHouse
 from src.sink import KafkaSinkManager
-import asyncio
 
 KAFKA_HOST = os.getenv("KAFKA_HOST", "localhost")
 KAFKA_PORT = os.getenv("KAFKA_PORT", "9092")
-KAFKA_TOPICS = ["raw-data","processed-data"]
+KAFKA_TOPICS = ["raw-data", "processed-data"]
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # Connect to databases
     ClickHouse.service.connect()
 
-    # Initialize and start Kafka sink manager
     sink_manager = KafkaSinkManager(KAFKA_HOST, KAFKA_PORT)
 
-    try:
-        await asyncio.to_thread(sink_manager.start, *KAFKA_TOPICS)
-    except Exception as e:
-        print(f"Warning: Failed to start Kafka sink: {e}")
-        print("API will continue running without Kafka sink")
+    def kafka_worker():
+        """
+        Kafka runs in its own thread with its own event loop.
+        This prevents rdkafka from blocking FastAPI startup.
+        """
+        try:
+            asyncio.run(sink_manager.start(*KAFKA_TOPICS))
+            print("Kafka consumer started successfully")
+        except Exception as e:
+            print(f"Kafka worker crashed: {e}")
+
+    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})"
+    )
 
     yield
 
@@ -33,6 +51,7 @@ async def lifespan(app: FastAPI):
 
     ClickHouse.service.client.close()
 
+
 app = FastAPI(lifespan=lifespan)
 
 app.include_router(v1_router, prefix="/api/v1", tags=["v1"])

sql/01_create_processed_latency_table.sql

@@ -1,51 +1,44 @@
--- Create processed_latency table for storing aggregated network metrics
-CREATE TABLE IF NOT EXISTS processed_latency (
-    -- Window temporal information
+CREATE DATABASE IF NOT EXISTS analytics;
+
+CREATE TABLE IF NOT EXISTS analytics.processed_latency
+(
     window_start_time DateTime64(3),
     window_end_time DateTime64(3),
     window_duration_seconds Float64,
 
-    -- Cell metadata
     cell_index Int32,
     network String,
 
-    -- RSRP statistics (Signal power)
     rsrp_mean Nullable(Float64),
     rsrp_max Nullable(Float64),
     rsrp_min Nullable(Float64),
     rsrp_std Nullable(Float64),
 
-    -- SINR statistics (Interference + noise)
     sinr_mean Nullable(Float64),
     sinr_max Nullable(Float64),
     sinr_min Nullable(Float64),
     sinr_std Nullable(Float64),
 
-    -- RSRQ statistics (Signal quality)
     rsrq_mean Nullable(Float64),
     rsrq_max Nullable(Float64),
     rsrq_min Nullable(Float64),
     rsrq_std Nullable(Float64),
 
-    -- Latency statistics
     latency_mean Nullable(Float64),
     latency_max Nullable(Float64),
     latency_min Nullable(Float64),
     latency_std Nullable(Float64),
 
-    -- CQI statistics (Cell quality)
     cqi_mean Nullable(Float64),
     cqi_max Nullable(Float64),
     cqi_min Nullable(Float64),
     cqi_std Nullable(Float64),
 
-    -- Bandwidth information
     primary_bandwidth Nullable(Float64),
     ul_bandwidth Nullable(Float64),
 
-    -- Sample metadata (higher count = more confidence)
     sample_count Int32
 )
-ENGINE = MergeTree()
+ENGINE = MergeTree
 ORDER BY (cell_index, window_start_time)
 SETTINGS index_granularity = 8192;