Merge pull request #247 from amosproj/56-video-router-dispatching-service

56 video router dispatching service

Signed-off-by: Felix Hilgers <felix.hilgers@fau.de>

Author: fhilgers

Makefile

@@ -9,7 +9,7 @@
 	format-check format-check-frontend format-check-backend \
 	test test-frontend test-backend \
 	sbom sbom-check \
-	run-backend-local run-frontend-local run-streamer-webcam run-streamer-file run-analyzer-local \
+	run-backend-local run-frontend-local run-streamer-webcam run-streamer-file run-analyzer-local run-orchestrator-local \
 	docker-build docker-build-frontend docker-build-backend docker-build-streamer \
 	docker-build-analyzer docker-build-analyzer-cuda docker-build-analyzer-rocm \
 	docker-compose-up docker-compose-down \
@@ -56,6 +56,8 @@ help:
 	@echo "      generates SBOM (sbom.json) and dependency CSV"
 	@echo "  sbom-check"
 	@echo "      checks if SBOM is up-to-date with dependencies"
+	@echo "  run-orchestrator-local"
+	@echo "      runs orchestrator service with dynamic port (starting from 8002)"
 	@echo "  run-backend-local"
 	@echo "      runs backend locally with uvicorn"
 	@echo "  run-frontend-local"
@@ -138,24 +140,28 @@ test-backend:
 	cd src/backend && uv run pytest -s
 
 run-streamer-webcam:
-	@echo "Starting video source service (webcam) on port 8000..."
-	cd src/backend && VIDEO_SOURCE_TYPE=webcam uv run uvicorn streamer.main:app --host 0.0.0.0 --port 8000 --reload
+	@echo "Starting video source service (webcam) with dynamic port..."
+	cd src/backend && VIDEO_SOURCE_TYPE=webcam uv run python -m streamer
 
 run-streamer-file:
-	@echo "Starting video source service (file) on port 8000..."
+	@echo "Starting video source service (file) with dynamic port..."
 	@echo "Set VIDEO_FILE_PATH env var to specify file (default: video.mp4)"
-	cd src/backend && VIDEO_SOURCE_TYPE=file uv run uvicorn streamer.main:app --host 0.0.0.0 --port 8000 --reload
+	cd src/backend && VIDEO_SOURCE_TYPE=file uv run python -m streamer
 
 run-analyzer-local:
-	@echo "Starting analyzer service on port 8001..."
-	cd src/backend && uv run uvicorn analyzer.main:app --host 0.0.0.0 --port 8001 --reload
+	@echo "Starting analyzer service with dynamic port..."
+	cd src/backend && uv run python -m analyzer
+
+run-orchestrator-local:
+	@echo "Starting orchestrator service with dynamic port..."
+	cd src/backend && uv run python -m orchestrator
 
 run-backend-local: run-streamer-webcam
 	@echo "Note: To run analyzer, use 'make run-analyzer-local' in another terminal"
 	@echo "Note: To use file source instead, run 'make run-streamer-file'"
 
 run-frontend-local:
-	cd src/frontend && VITE_BACKEND_URL=http://localhost:8001 npm run dev
+	cd src/frontend && VITE_ORCHESTRATOR_URL=http://localhost:8002 npm run dev
 
 docker-build: docker-build-frontend docker-build-backend
 

docker-compose.yml

@@ -3,6 +3,19 @@
 # SPDX-License-Identifier: MIT
 
 services:
+  orchestrator:
+    build:
+      context: ./src/backend
+      dockerfile: Dockerfile.orchestrator
+    image: robot-orchestrator:latest
+    container_name: robot-orchestrator
+    ports:
+      - "8002:8002"
+    environment:
+      - CORS_ORIGINS=*
+    networks:
+      - robot-net
+
   # Note: Camera device access only works on Linux.
   # On macOS/Windows, run things locally
   streamer:
@@ -19,12 +32,16 @@ services:
       - CAMERA_INDEX=0
       - STUN_SERVER=stun:stun.l.google.com:19302
       - CORS_ORIGINS=*
+      - ORCHESTRATOR_URL=http://orchestrator:8002
+      - STREAMER_PUBLIC_URL=http://streamer:8000
     devices:
       - /dev/video0:/dev/video0  # Only works on Linux (ignored for file mode)
     volumes:
       - ./videos:/app/videos  # Mount videos directory for file mode
     networks:
       - robot-net
+    depends_on:
+      - orchestrator
 
   analyzer:
     build:
@@ -41,9 +58,12 @@ services:
       - MODEL_PATH=models/yolo11n.pt
       - STUN_SERVER=stun:stun.l.google.com:19302
       - CORS_ORIGINS=*
+      - ORCHESTRATOR_URL=http://orchestrator:8002
+      - ANALYZER_PUBLIC_URL=http://analyzer:8001
     volumes:
       - ./src/backend/models:/app/models
     depends_on:
+      - orchestrator
       - streamer
     networks:
       - robot-net
@@ -54,6 +74,7 @@ services:
       dockerfile: Dockerfile
       args:
         - VITE_BACKEND_URL=http://localhost:8001
+        - VITE_ORCHESTRATOR_URL=http://localhost:8002
     image: robot-frontend:latest
     container_name: robot-frontend
     ports:

src/backend/Dockerfile.orchestrator

@@ -0,0 +1,19 @@
+# SPDX-FileCopyrightText: 2025 robot-visual-perception
+#
+# SPDX-License-Identifier: MIT
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
+
+COPY pyproject.toml uv.lock ./
+# only core dependencies
+RUN uv sync --frozen --no-dev
+
+COPY common/ ./common/
+COPY orchestrator/ ./orchestrator/
+
+EXPOSE 8002
+
+CMD ["uv", "run", "uvicorn", "orchestrator.main:app", "--host", "0.0.0.0", "--port", "8002"]

src/backend/analyzer/__main__.py

@@ -0,0 +1,42 @@
+# SPDX-FileCopyrightText: 2025 robot-visual-perception
+#
+# SPDX-License-Identifier: MIT
+
+"""CLI entry point for analyzer service with dynamic port allocation."""
+
+import os
+import sys
+
+
+# Must set env vars BEFORE importing analyzer.main (which does early initialization)
+def main() -> None:
+    """Start analyzer service on a free port."""
+    # Import here to avoid early initialization
+    from common.port_utils import find_free_port
+
+    # Find free port starting from 8001
+    port = find_free_port(start_port=8001)
+    if port is None:
+        print("ERROR: Could not find a free port in range 8001-8100", file=sys.stderr)
+        sys.exit(1)
+
+    # Set public URL for orchestrator registration BEFORE importing main
+    host = os.getenv("SERVICE_HOST", "localhost")
+    public_url = f"http://{host}:{port}"
+    os.environ["ANALYZER_PUBLIC_URL"] = public_url
+
+    print(f"Starting analyzer service on {public_url}")
+
+    # Now import uvicorn and start
+    import uvicorn
+
+    uvicorn.run(
+        "analyzer.main:app",
+        host="0.0.0.0",
+        port=port,
+        reload=os.getenv("RELOAD", "true").lower() in ("true", "1", "yes"),
+    )
+
+
+if __name__ == "__main__":
+    main()

src/backend/analyzer/main.py

@@ -5,10 +5,15 @@
 # Necessary for running stuff before other imports
 # ruff: noqa: E402
 
+import os
+
 from common import __version__
 from common.logging_config import configure_logging
 from common.metrics import configure_metrics
 
+# Set service type for Prometheus metrics port allocation
+os.environ["SERVICE_TYPE"] = "analyzer"
+
 # Initialize logging early
 configure_logging(service_name="analyzer", service_version=__version__)
 
@@ -29,6 +34,7 @@
 config.apply_settings_file(config.ANALYZER_SETTINGS_FILE)
 from common.core.detector import get_detector
 from common.core.depth import get_depth_estimator
+from common.orchestrator import register_with_orchestrator, deregister_from_orchestrator
 from analyzer.routes import router, on_shutdown
 
 
@@ -56,8 +62,20 @@ async def lifespan_context(app: FastAPI) -> AsyncIterator[None]:
         # Warm up detector and depth estimator so initial /offer handling is instant.
         get_detector(yolo_model_path)
         get_depth_estimator(midas_cache_directory)
+
+        # Register this analyzer instance with orchestrator (best-effort)
+        await register_with_orchestrator(
+            service_type="analyzer",
+            service_url=config.ANALYZER_PUBLIC_URL,
+            orchestrator_url=config.ORCHESTRATOR_URL,
+        )
         yield
         with suppress(Exception):
+            await deregister_from_orchestrator(
+                service_type="analyzer",
+                service_url=config.ANALYZER_PUBLIC_URL,
+                orchestrator_url=config.ORCHESTRATOR_URL,
+            )
             await on_shutdown()
 
     return lifespan_context

src/backend/analyzer/manager.py

@@ -79,6 +79,10 @@ def __init__(self) -> None:
         self._inference_task: asyncio.Task[None] | None = None
         self._intrinsics_logged: bool = False
 
+        # Dynamic streamer configuration
+        self._streamer_url: str | None = None
+        self._streamer_url_event = asyncio.Event()
+
         self.max_consecutive_errors = 5
         # adaptive downscaling parameters
         self.target_scale_init = config.TARGET_SCALE_INIT
@@ -116,6 +120,21 @@ async def connect(self, websocket: WebSocket) -> None:
         if len(self.active_connections) == 1:
             await self._start_processing()
 
+    async def set_streamer_url(self, streamer_url: str) -> None:
+        """Set the streamer URL and trigger processing startup if needed."""
+        # Check if URL is changing
+        url_changed = self._streamer_url and self._streamer_url != streamer_url
+
+        self._streamer_url = streamer_url
+        logger.info(f"Streamer URL configured: {streamer_url}")
+        self._streamer_url_event.set()
+
+        # If URL changed and we have active connections, restart processing
+        if url_changed and self.active_connections:
+            logger.info("Streamer URL changed, restarting processing...")
+            await self._stop_processing()
+            await self._start_processing()
+
     async def disconnect(self, websocket: WebSocket) -> None:
         """Handle WebSocket disconnection."""
         self.active_connections.discard(websocket)
@@ -141,9 +160,16 @@ async def _start_processing(self) -> None:
             return  # Already running
 
         try:
+            # Wait for streamer URL to be configured if not already set
+            if not self._streamer_url:
+                logger.info("Waiting for streamer URL configuration...")
+                await asyncio.wait_for(self._streamer_url_event.wait(), timeout=30.0)
+
+            if not self._streamer_url:
+                raise Exception("Streamer URL was not configured")
+
             # Connect to webcam service
-            upstream_url = config.STREAMER_OFFER_URL
-            self._webcam_session = WebcamSession(upstream_url)
+            self._webcam_session = WebcamSession(self._streamer_url)
             source_track = await self._webcam_session.connect()
 
             # Start processing task

src/backend/analyzer/routes.py

@@ -6,10 +6,17 @@
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
 from fastapi.responses import FileResponse, Response
 from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
+from pydantic import BaseModel
 
 from analyzer.manager import AnalyzerWebSocketManager
 
 
+class ConfigureAnalyzerRequest(BaseModel):
+    """Request to configure analyzer with streamer URL."""
+
+    streamer_url: str
+
+
 # Create a global instance of the WebSocket manager
 websocket_manager = AnalyzerWebSocketManager()
 
@@ -23,6 +30,22 @@ def health() -> dict[str, str]:
     return {"status": "ok", "service": "analyzer"}
 
 
+@router.post("/configure")
+async def configure_analyzer(request: ConfigureAnalyzerRequest) -> dict[str, str]:
+    """Configure analyzer to use a specific streamer.
+
+    This endpoint is called by the orchestrator/frontend after analyzer assignment
+    to tell this analyzer which streamer service to connect to.
+    """
+    # Append /offer endpoint path to the streamer base URL
+    streamer_offer_url = f"{request.streamer_url.rstrip('/')}/offer"
+    await websocket_manager.set_streamer_url(streamer_offer_url)
+    return {
+        "status": "configured",
+        "streamer_url": streamer_offer_url,
+    }
+
+
 @router.get("/metrics")
 def metrics() -> Response:
     """Prometheus metrics endpoint."""

src/backend/common/config.py

@@ -130,6 +130,11 @@ class Config:
         "yes",
     )
 
+    # Service discovery / orchestrator
+    ORCHESTRATOR_URL: str = os.getenv("ORCHESTRATOR_URL", "http://localhost:8002")
+    ANALYZER_PUBLIC_URL: str = os.getenv("ANALYZER_PUBLIC_URL", "http://localhost:8001")
+    STREAMER_PUBLIC_URL: str = os.getenv("STREAMER_PUBLIC_URL", "http://localhost:8000")
+
     # Tracking/interpolation settings
     # Minimum IoU to match detection to track
     TRACKING_IOU_THRESHOLD: float = float(os.getenv("TRACKING_IOU_THRESHOLD", "0.1"))

src/backend/common/metrics.py

@@ -1,6 +1,8 @@
 # SPDX-FileCopyrightText: 2025 robot-visual-perception
 #
 # SPDX-License-Identifier: MIT
+import logging
+import os
 from typing import Optional
 
 from prometheus_client import (
@@ -16,13 +18,22 @@
 _depth_estimation_duration: Optional[Histogram] = None
 _detections_count: Optional[Counter] = None
 
+# Service-specific Prometheus ports to avoid conflicts
+PROMETHEUS_PORTS = {
+    "analyzer": 9001,
+    "streamer": 9002,
+    "orchestrator": 9003,
+}
+
 
 def configure_metrics() -> None:
     """
     Configure Prometheus metrics.
     """
     global _detection_duration, _depth_estimation_duration, _detections_count
 
+    logger = logging.getLogger(__name__)
+
     if _detection_duration is not None:
         return  # Already configured
 
@@ -44,7 +55,24 @@ def configure_metrics() -> None:
         ["interpolated"],
     )
 
-    start_http_server(9000)
+    # Determine service-specific port
+    service_type = os.getenv("SERVICE_TYPE", "unknown")
+    port = PROMETHEUS_PORTS.get(service_type, 9000)
+
+    # Start Prometheus HTTP server with error handling for --reload mode
+    try:
+        start_http_server(port)
+        logger.info(
+            f"Prometheus metrics server started on port {port} (service: {service_type})"
+        )
+    except OSError as e:
+        if "Address already in use" in str(e):
+            logger.warning(
+                f"Port {port} already in use (likely uvicorn --reload worker). "
+                "Metrics endpoint will be available on the main process only."
+            )
+        else:
+            raise
 
 
 def get_detection_duration() -> Histogram: