test: load-test

.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

AI_ARCHITECTURE.md

@@ -177,8 +177,11 @@ return ProtectedVideoTrack(source=track, privacy_filter=privacy_filter, session_
 5. 프레임 큐 `_queue`와 신원 대조 큐 `_identity_queue`를 만든다.
 6. `privacy-blur-gpu-batcher`와 `privacy-blur-identity-worker` 두 데몬 스레드를 시작한다.
 
-두 워커는 같은 `_processing_lock`을 공유한다. 프레임 배치 처리, 신원 대조 처리,
-기준 얼굴 준비/적용이 같은 모델 상태를 만지기 때문에 동시에 실행되지 않게 막는다.
+프레임 워커는 라이브 YOLO detector를 `_detector_lock`으로 보호한다. 기준 얼굴
+등록은 별도의 CPU detector 인스턴스로 얼굴을 먼저 찾기 때문에 업로드 준비가
+라이브 detector 락을 점유하지 않는다. SFace 신원 대조와 기준 얼굴 feature 적용은
+`_identity_lock`으로 보호하지만, 프레임 워커는 기존 추적/feature 상태로 계속
+프레임을 반환한다.
 
 ### 8.1 프레임 배치 워커
 
@@ -311,7 +314,7 @@ data/reference_face.jpg
 `register_reference_faces()`는 바로 최종 파일을 쓰지 않는다.
 
 1. `.reference_face.{face_id}.tmp` 임시 파일을 쓴다.
-2. 런타임이 살아 있으면 `prepare_identity_update()`로 임시 파일까지 포함해 기준 얼굴 임베딩을 미리 만든다.
+2. 런타임이 살아 있으면 `prepare_identity_update()`로 임시 파일까지 포함해 기준 얼굴 임베딩을 미리 만든다. 이때 기준 얼굴 감지는 등록 전용 CPU detector로 수행해 라이브 스트림 detector를 막지 않는다.
 3. 검증이 성공하면 `os.replace(tmp, final)`로 최종 파일을 발행한다.
 4. `replace_existing=True`인 경우 기존 파일을 삭제한다.
 5. `commit_identity_update()`로 공유 런타임과 현재 살아 있는 필터들에 새 config/reference feature를 반영한다.
@@ -421,7 +424,9 @@ threshold 이상이면 해당 `FaceTrack.is_me=True`가 되고, `finish_frame()`
 
 새 reference embedding은 `prepare_identity_update()`에서 먼저 준비한다.
 준비가 성공한 뒤에만 파일을 최종 경로로 이동하고 `commit_identity_update()`를 호출한다.
-따라서 잘못된 업로드가 기존 라이브 설정을 덮어쓰지 않는다.
+준비 중에는 등록 전용 CPU detector가 기준 얼굴을 찾고, 기존 스트림은 현재 설정과
+feature로 계속 프레임을 처리한다. 따라서 잘못된 업로드가 기존 라이브 설정을
+덮어쓰지 않고, 업로드 준비가 라이브 detector 락을 점유하지 않는다.
 
 ### Q. AI가 실패하면 원본이 노출되는가?
 

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)