merge: develop

.env.example

@@ -10,6 +10,8 @@ WEBRTC_UDP_PORT_MIN=50000
 WEBRTC_UDP_PORT_MAX=60000
 WEBRTC_DISCONNECTED_GRACE_SECONDS=10
 AI_PRIVACY_ENABLED=true
+AI_PRIVACY_MODE=real
+AI_PRIVACY_FIXED_DELAY_MS=20
 AI_PRIVACY_DEVICE=cuda:0
 AI_PRIVACY_REQUIRE_GPU=true
 AI_PRIVACY_USE_HALF=true

.gitignore

@@ -10,3 +10,9 @@ htmlcov/
 .DS_Store
 .idea
 data/
+loadtest/pion-client/results/
+loadtest/pion-client/.gocache/
+loadtest/pion-client/*.exe
+loadtest/server-results/
+loadtest/videos/*
+!loadtest/videos/README.md

app/api/routes/root.py

@@ -1,6 +1,9 @@
 from fastapi import APIRouter
+from fastapi.responses import Response
+from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
 
 from app.services.ai.filter import get_privacy_runtime_stats
+from app.services.metrics import update_runtime_metrics
 
 router = APIRouter()
 
@@ -17,3 +20,9 @@ async def health() -> dict[str, object]:
         "status": "ready",
         "ai_runtimes": get_privacy_runtime_stats(),
     }
+
+
+@router.get("/metrics", include_in_schema=False)
+async def metrics() -> Response:
+    update_runtime_metrics(get_privacy_runtime_stats())
+    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

app/core/config.py

@@ -1,5 +1,7 @@
 from functools import lru_cache
+from typing import Literal
 
+from pydantic import model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -16,6 +18,8 @@ class Settings(BaseSettings):
     webrtc_udp_port_max: int | None = 60000
     webrtc_disconnected_grace_seconds: float = 10.0
     ai_privacy_enabled: bool = True
+    ai_privacy_mode: Literal["real", "bypass", "fixed_delay"] = "real"
+    ai_privacy_fixed_delay_ms: float = 20.0
     ai_privacy_device: str | None = None
     ai_privacy_require_gpu: bool = False
     ai_privacy_use_half: bool = True
@@ -48,6 +52,17 @@ class Settings(BaseSettings):
         extra="ignore",
     )
 
+    @model_validator(mode="after")
+    def validate_production_privacy_settings(self) -> "Settings":
+        if self.app_env != "production":
+            return self
+
+        if not self.ai_privacy_enabled:
+            raise ValueError("AI_PRIVACY_ENABLED must be true in production")
+        if self.ai_privacy_mode != "real":
+            raise ValueError("AI_PRIVACY_MODE must be real in production")
+        return self
+
 
 @lru_cache
 def get_settings() -> Settings:

app/services/ai/filter.py

@@ -7,6 +7,7 @@
 from typing import TYPE_CHECKING, Any
 
 from app.core.config import Settings, get_settings
+from app.services.metrics import AI_STAGE_DURATION
 from privacy_blur import (
     FacePrivacyFilter,
     PrivacyBlurConfig,
@@ -108,9 +109,13 @@ def _build_config() -> PrivacyBlurConfig | None:
         return _config
 
     settings = get_settings()
-    if not settings.ai_privacy_enabled:
+    if not settings.ai_privacy_enabled or settings.ai_privacy_mode != "real":
         _config_initialized = True
-        logger.info("AI privacy filter is disabled")
+        logger.info(
+            "AI privacy runtime is disabled: enabled=%s mode=%s",
+            settings.ai_privacy_enabled,
+            settings.ai_privacy_mode,
+        )
         return None
 
     _config = build_privacy_config(settings)
@@ -127,6 +132,9 @@ def initialize_privacy_runtime() -> PrivacyBlurConfig | None:
             return None
         if _runtime is None:
             _runtime = initialize_runtime(config)
+            _runtime.set_stage_observer(
+                lambda stage, seconds: AI_STAGE_DURATION.labels(stage).observe(seconds)
+            )
             _store_identity_features(
                 getattr(config, "identity_client_id", None),
                 getattr(_runtime, "reference_features", None),

app/services/ai/track.py

@@ -2,12 +2,21 @@
 
 import asyncio
 import logging
+from time import perf_counter
 from typing import TYPE_CHECKING
 
 import numpy as np
 from aiortc.mediastreams import MediaStreamTrack, VideoStreamTrack
 from av import VideoFrame
 
+from app.services.metrics import (
+    FRAME_FAILURES,
+    FRAME_PROCESSING_DURATION,
+    FRAMES_DROPPED,
+    FRAMES_PROCESSED,
+    FRAMES_RECEIVED,
+)
+
 if TYPE_CHECKING:
     from privacy_blur import FacePrivacyFilter
 
@@ -35,21 +44,36 @@ def is_fallback_active(self) -> bool:
         return self._fallback
 
     async def recv(self):
-        frame = await _recv_latest_frame(self._source)
+        frame, dropped = await _recv_latest_frame_with_drop_count(self._source)
+        mode = "real"
+        FRAMES_RECEIVED.labels(mode).inc()
+        if dropped:
+            FRAMES_DROPPED.labels(mode).inc(dropped)
+        started = perf_counter()
         if self._fallback:
-            return _blackout_frame(frame)
+            result = _blackout_frame(frame)
+            FRAMES_PROCESSED.labels(mode).inc()
+            FRAME_PROCESSING_DURATION.labels(mode).observe(perf_counter() - started)
+            return result
         try:
             assert self._filter is not None
-            return await self._filter.apply_async(frame)
+            result = await self._filter.apply_async(frame)
+            FRAMES_PROCESSED.labels(mode).inc()
+            FRAME_PROCESSING_DURATION.labels(mode).observe(perf_counter() - started)
+            return result
         except Exception:
             self._fallback = True
+            FRAME_FAILURES.labels(mode).inc()
             logger.warning(
                 "AI filter failed, blocking video instead of exposing raw frames: "
                 "session_id=%s",
                 self._session_id,
                 exc_info=True,
             )
-            return _blackout_frame(frame)
+            result = _blackout_frame(frame)
+            FRAMES_PROCESSED.labels(mode).inc()
+            FRAME_PROCESSING_DURATION.labels(mode).observe(perf_counter() - started)
+            return result
 
     def stop(self) -> None:
         super().stop()
@@ -69,10 +93,17 @@ def stop(self) -> None:
 
 
 async def _recv_latest_frame(source: MediaStreamTrack) -> VideoFrame:
+    frame, _ = await _recv_latest_frame_with_drop_count(source)
+    return frame
+
+
+async def _recv_latest_frame_with_drop_count(
+    source: MediaStreamTrack,
+) -> tuple[VideoFrame, int]:
     frame = await source.recv()
     queue = getattr(source, "_queue", None)
     if not isinstance(queue, asyncio.Queue):
-        return frame
+        return frame, 0
 
     dropped = 0
     while True:
@@ -94,7 +125,47 @@ async def _recv_latest_frame(source: MediaStreamTrack) -> VideoFrame:
             "Dropped stale source frames before AI processing: count=%d",
             dropped,
         )
-    return frame
+    return frame, dropped
+
+
+class BenchmarkVideoTrack(VideoStreamTrack):
+    kind = "video"
+
+    def __init__(
+        self,
+        source: MediaStreamTrack,
+        *,
+        session_id: str,
+        mode: str,
+        delay_ms: float = 0.0,
+    ) -> None:
+        super().__init__()
+        self._source = source
+        self._session_id = session_id
+        self._mode = mode
+        self._delay_seconds = max(delay_ms, 0.0) / 1000.0
+
+    @property
+    def is_fallback_active(self) -> bool:
+        return False
+
+    async def recv(self) -> VideoFrame:
+        frame, dropped = await _recv_latest_frame_with_drop_count(self._source)
+        FRAMES_RECEIVED.labels(self._mode).inc()
+        if dropped:
+            FRAMES_DROPPED.labels(self._mode).inc(dropped)
+        started = perf_counter()
+        if self._delay_seconds:
+            await asyncio.sleep(self._delay_seconds)
+        FRAMES_PROCESSED.labels(self._mode).inc()
+        FRAME_PROCESSING_DURATION.labels(self._mode).observe(
+            perf_counter() - started
+        )
+        return frame
+
+    def stop(self) -> None:
+        super().stop()
+        self._source.stop()
 
 
 def _blackout_frame(frame: VideoFrame) -> VideoFrame:

app/services/metrics.py

@@ -0,0 +1,67 @@
+from prometheus_client import Counter, Gauge, Histogram
+
+
+ACTIVE_SESSIONS = Gauge(
+    "innolive_active_sessions",
+    "Number of active WebRTC sessions.",
+)
+CONNECTIONS = Counter(
+    "innolive_connection_total",
+    "Number of WebRTC sessions created.",
+)
+CONNECTION_FAILURES = Counter(
+    "innolive_connection_failures_total",
+    "Number of WebRTC connections that reached a failed state.",
+)
+FRAMES_RECEIVED = Counter(
+    "innolive_frame_received_total",
+    "Number of video frames received for processing.",
+    ("mode",),
+)
+FRAMES_PROCESSED = Counter(
+    "innolive_frame_processed_total",
+    "Number of video frames returned by the processing track.",
+    ("mode",),
+)
+FRAMES_DROPPED = Counter(
+    "innolive_frame_dropped_total",
+    "Number of stale source frames discarded before processing.",
+    ("mode",),
+)
+FRAME_FAILURES = Counter(
+    "innolive_frame_failures_total",
+    "Number of frame-processing failures.",
+    ("mode",),
+)
+FRAME_PROCESSING_DURATION = Histogram(
+    "innolive_frame_processing_duration_seconds",
+    "Time spent processing one frame.",
+    ("mode",),
+    buckets=(0.005, 0.01, 0.02, 0.033, 0.05, 0.075, 0.1, 0.2, 0.5, 1.0),
+)
+AI_STAGE_DURATION = Histogram(
+    "innolive_ai_stage_duration_seconds",
+    (
+        "Time spent in an AI video pipeline stage. The encode stage measures "
+        "BGR-to-AV-frame construction, not aiortc codec compression."
+    ),
+    ("stage",),
+    buckets=(0.001, 0.0025, 0.005, 0.01, 0.02, 0.033, 0.05, 0.075, 0.1, 0.2, 0.5, 1.0),
+)
+PROCESSING_QUEUE_SIZE = Gauge(
+    "innolive_processing_queue_size",
+    "Current shared AI processing queue depth.",
+)
+RECONNECTS = Counter(
+    "innolive_reconnect_total",
+    "Number of WebRTC sessions that recovered after disconnecting.",
+)
+
+
+def update_runtime_metrics(stats: list[dict[str, object]]) -> None:
+    queue_depth = 0
+    for runtime in stats:
+        value = runtime.get("queue_depth", 0)
+        if isinstance(value, int | float):
+            queue_depth += float(value)
+    PROCESSING_QUEUE_SIZE.set(queue_depth)

app/services/sessions/manager.py

@@ -11,6 +11,12 @@
 from app.core.config import get_settings
 from app.core.exceptions import BadRequestError, ConflictError, NotFoundError
 from app.services.rtmp.state import StreamState
+from app.services.metrics import (
+    ACTIVE_SESSIONS,
+    CONNECTION_FAILURES,
+    CONNECTIONS,
+    RECONNECTS,
+)
 from app.services.webrtc.metrics import PeerConnectionTiming
 from app.services.webrtc.peer_connection import create_peer_connection
 
@@ -124,6 +130,8 @@ async def create_session(
 
         async with self._lock:
             self._sessions[session_id] = session
+            ACTIVE_SESSIONS.set(len(self._sessions))
+        CONNECTIONS.inc()
 
         logger.info("Created live session: session_id=%s", session_id)
         return session
@@ -272,6 +280,7 @@ async def close_all(self) -> None:
         async with self._lock:
             sessions = list(self._sessions.values())
             self._sessions.clear()
+            ACTIVE_SESSIONS.set(0)
 
         for session in sessions:
             if session.disconnected_cleanup_task is not None:
@@ -300,6 +309,8 @@ async def handle_peer_connection_state_change(
             return
 
         if connection_state in TERMINAL_CONNECTION_STATES:
+            if connection_state == "failed":
+                CONNECTION_FAILURES.inc()
             await self.close_session(
                 session_id,
                 reason=f"peer_connection_{connection_state}",
@@ -328,6 +339,7 @@ async def close_session(
 
             session.status = "closing"
             self._sessions.pop(session_id, None)
+            ACTIVE_SESSIONS.set(len(self._sessions))
             cleanup_task = session.disconnected_cleanup_task
             session.disconnected_cleanup_task = None
 
@@ -536,6 +548,7 @@ async def _cancel_disconnected_cleanup(
 
         if cleanup_task is not None and not cleanup_task.done():
             cleanup_task.cancel()
+            RECONNECTS.inc()
             logger.info(
                 "Cancelled disconnected session cleanup: session_id=%s "
                 "connection_state=%s",
@@ -882,9 +895,24 @@ def _wrap_with_ai(
     from app.core.config import get_settings
     from app.services.ai.filter import get_privacy_filter
 
-    privacy_filter = get_privacy_filter(client_id=client_id)
-    if privacy_filter is None and not get_settings().ai_privacy_enabled:
+    settings = get_settings()
+    if not settings.ai_privacy_enabled:
         return track
+    if settings.ai_privacy_mode in {"bypass", "fixed_delay"}:
+        from app.services.ai.track import BenchmarkVideoTrack
+
+        return BenchmarkVideoTrack(
+            source=track,
+            session_id=session_id,
+            mode=settings.ai_privacy_mode,
+            delay_ms=(
+                settings.ai_privacy_fixed_delay_ms
+                if settings.ai_privacy_mode == "fixed_delay"
+                else 0.0
+            ),
+        )
+
+    privacy_filter = get_privacy_filter(client_id=client_id)
 
     from app.services.ai.track import ProtectedVideoTrack
 

docker-compose.prod.yml

@@ -11,6 +11,8 @@ services:
       - .env
 
     environment:
+      APP_ENV: "production"
       AI_PRIVACY_ENABLED: "true"
+      AI_PRIVACY_MODE: "real"
       AI_PRIVACY_DEVICE: "cuda:0"
       AI_PRIVACY_REQUIRE_GPU: "true"