diff --git a/.env.example b/.env.example index eb99246..a8ada78 100644 --- a/.env.example +++ b/.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 diff --git a/.gitignore b/.gitignore index 2ac5406..47aa6c0 100644 --- a/.gitignore +++ b/.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 diff --git a/AI_ARCHITECTURE.md b/AI_ARCHITECTURE.md index 2dfb36c..95336cb 100644 --- a/AI_ARCHITECTURE.md +++ b/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가 실패하면 원본이 노출되는가? diff --git a/app/api/routes/root.py b/app/api/routes/root.py index b1612e2..13dfb88 100644 --- a/app/api/routes/root.py +++ b/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) diff --git a/app/core/config.py b/app/core/config.py index d3cad36..083b692 100644 --- a/app/core/config.py +++ b/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: diff --git a/app/services/ai/filter.py b/app/services/ai/filter.py index 3bc0317..5bf2da2 100644 --- a/app/services/ai/filter.py +++ b/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), diff --git a/app/services/ai/track.py b/app/services/ai/track.py index 215e386..e6a6c07 100644 --- a/app/services/ai/track.py +++ b/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: diff --git a/app/services/metrics.py b/app/services/metrics.py new file mode 100644 index 0000000..3be84ee --- /dev/null +++ b/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) diff --git a/app/services/sessions/manager.py b/app/services/sessions/manager.py index 90f7ce3..bcbf349 100644 --- a/app/services/sessions/manager.py +++ b/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 diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index c1f370c..25723be 100644 --- a/docker-compose.prod.yml +++ b/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" diff --git a/docker-compose.test.yml b/docker-compose.test.yml new file mode 100644 index 0000000..3894102 --- /dev/null +++ b/docker-compose.test.yml @@ -0,0 +1,26 @@ +services: + prometheus: + image: prom/prometheus:v3.7.3 + command: + - --config.file=/etc/prometheus/prometheus.yml + - --storage.tsdb.retention.time=7d + volumes: + - ./observability/prometheus.yml:/etc/prometheus/prometheus.yml:ro + extra_hosts: + - host.docker.internal:host-gateway + ports: + - "9090:9090" + + grafana: + image: grafana/grafana:12.3.1 + environment: + GF_SECURITY_ADMIN_USER: admin + GF_SECURITY_ADMIN_PASSWORD: admin + GF_USERS_ALLOW_SIGN_UP: "false" + volumes: + - ./observability/grafana/provisioning:/etc/grafana/provisioning:ro + - ./observability/grafana/dashboards:/var/lib/grafana/dashboards:ro + ports: + - "3000:3000" + depends_on: + - prometheus diff --git a/loadtest/k6/sessions.js b/loadtest/k6/sessions.js new file mode 100644 index 0000000..f6e4f28 --- /dev/null +++ b/loadtest/k6/sessions.js @@ -0,0 +1,40 @@ +import http from "k6/http"; +import { check, sleep } from "k6"; + +const baseUrl = __ENV.BASE_URL || "http://127.0.0.1:8000"; + +export const options = { + scenarios: { + session_control: { + executor: "ramping-vus", + startVUs: 0, + stages: [ + { duration: "30s", target: Number(__ENV.VUS || 10) }, + { duration: __ENV.HOLD || "2m", target: Number(__ENV.VUS || 10) }, + { duration: "15s", target: 0 } + ] + } + }, + thresholds: { + http_req_failed: ["rate<0.01"], + http_req_duration: ["p(95)<500"] + } +}; + +export default function () { + const create = http.post( + `${baseUrl}/sessions`, + JSON.stringify({ metadata: { source: "k6", vu: String(__VU) } }), + { headers: { "Content-Type": "application/json" } } + ); + check(create, { "session created": (response) => response.status === 201 }); + if (create.status !== 201) return; + + const sessionId = create.json("session_id"); + const read = http.get(`${baseUrl}/sessions/${sessionId}`); + check(read, { "session readable": (response) => response.status === 200 }); + sleep(Number(__ENV.SESSION_SECONDS || 5)); + + const remove = http.del(`${baseUrl}/sessions/${sessionId}`); + check(remove, { "session deleted": (response) => response.status === 204 }); +} diff --git a/loadtest/pion-client/README.md b/loadtest/pion-client/README.md new file mode 100644 index 0000000..53b21aa --- /dev/null +++ b/loadtest/pion-client/README.md @@ -0,0 +1,93 @@ +# InnoLive Pion media load client + +This client creates real InnoLive sessions and one Pion PeerConnection per +virtual broadcaster. Every client loops the same VP8/IVF file, receives the +processed video track, exchanges offer/answer and trickle ICE over the server's +signaling WebSocket, and writes interval samples to JSON and CSV. + +## Prerequisites + +- Go 1.23 or newer +- FFmpeg (only needed to prepare the input) +- A running InnoLive server + +Create a deterministic 720p/30 FPS input from any MP4: + +```bash +mkdir -p loadtest/videos +ffmpeg -i input.mp4 -an -vf "scale=1280:720,fps=30" \ + -c:v libvpx -deadline good -cpu-used 2 -b:v 2M \ + loadtest/videos/720p30.ivf +``` + +The load client intentionally accepts IVF rather than MP4. IVF contains VP8 +frames Pion can packetize directly, so decoding and re-encoding in the load +generator does not consume CPU and distort the server-capacity result. + +## Build + +```bash +cd loadtest/pion-client +go mod download +go build -o pion-load . +``` + +## Run scenarios + +Step load (add five streams every two minutes): + +```bash +./pion-load \ + -base-url http://127.0.0.1:8000 \ + -input ../videos/720p30.ivf \ + -scenario step -clients 50 -start-clients 5 \ + -step-size 5 -step-duration 2m \ + -output results/step-720p30 +``` + +Spike load (reach 50 streams in 10 seconds and hold for five minutes): + +```bash +./pion-load -base-url http://127.0.0.1:8000 \ + -input ../videos/720p30.ivf -scenario spike -clients 50 \ + -ramp-duration 10s -duration 5m -output results/spike-720p30 +``` + +Soak load (gradually reach 30 streams and hold for two hours): + +```bash +./pion-load -base-url http://127.0.0.1:8000 \ + -input ../videos/720p30.ivf -scenario soak -clients 30 \ + -start-clients 1 -ramp-duration 5m -duration 2h \ + -output results/soak-720p30 +``` + +Press Ctrl+C for a graceful stop. Active PeerConnections and REST sessions are +closed, then partial results are still written. + +## Output fields + +Both files contain one row per active client at each `-stats-interval` (default +5 seconds): + +- connection and ICE state, success/failure, and last error +- sent and received frame totals plus interval send/receive FPS +- received RTP packets, inferred missing packets, and cumulative loss percent +- RFC 3550 receive jitter in milliseconds +- RTT in milliseconds derived from RTCP receiver reports +- received payload bytes + +RTT is `0` until the first usable RTCP receiver report arrives. Packet loss and +jitter describe the processed video received by the client. The JSON file also +contains the complete command configuration. + +## Capacity-test notes + +- Run the load generator on a different machine for final capacity numbers. +- Keep input, codec, resolution, FPS, server mode, and network conditions fixed. +- Check server Prometheus/Grafana metrics alongside the client CSV. +- Start with `AI_PRIVACY_MODE=bypass`, then `fixed_delay`, then `real` to isolate + WebRTC overhead from AI cost. +- A single process is suitable for dozens to low hundreds of connections; + distribute identical runs across load-generator machines if that host itself + becomes CPU- or network-bound. diff --git a/loadtest/pion-client/client.go b/loadtest/pion-client/client.go new file mode 100644 index 0000000..7ea6e04 --- /dev/null +++ b/loadtest/pion-client/client.go @@ -0,0 +1,347 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "sync" + "sync/atomic" + "time" + + "github.com/gorilla/websocket" + "github.com/pion/rtcp" + "github.com/pion/webrtc/v4" +) + +type receiveCounters struct { + frames atomic.Uint64 + packets atomic.Uint64 + bytes atomic.Uint64 + lost atomic.Uint64 + jitterMicros atomic.Uint64 + rttMicros atomic.Uint64 +} + +type loadClient struct { + ordinal int + sessionID string + startedAt time.Time + connected atomic.Bool + failed atomic.Bool + errorMu sync.RWMutex + errorText string + stateMu sync.RWMutex + pcState string + iceState string + sent mediaCounters + received receiveCounters + cancel context.CancelFunc + pc *webrtc.PeerConnection + ws *signalSocket + http *http.Client + baseURL string + closeOnce sync.Once +} + +func startLoadClient(parent context.Context, cfg config, ordinal int, iceServers []webrtc.ICEServer, httpClient *http.Client) (*loadClient, error) { + ctx, cancel := context.WithCancel(parent) + c := &loadClient{ordinal: ordinal, startedAt: time.Now(), pcState: "new", iceState: "new", cancel: cancel, http: httpClient, baseURL: cfg.BaseURL} + + setupCtx, setupCancel := context.WithTimeout(ctx, cfg.ConnectTimeout) + defer setupCancel() + var err error + c.sessionID, err = createSession(setupCtx, httpClient, cfg.BaseURL, ordinal) + if err != nil { + cancel() + return nil, err + } + cleanup := func() { + c.Close() + } + + wsURL, err := websocketURL(cfg.BaseURL) + if err != nil { + cleanup() + return nil, err + } + conn, _, err := websocket.DefaultDialer.DialContext(setupCtx, wsURL, nil) + if err != nil { + cleanup() + return nil, fmt.Errorf("open signaling websocket: %w", err) + } + c.ws = &signalSocket{conn: conn} + + pc, err := webrtc.NewPeerConnection(webrtc.Configuration{ICEServers: iceServers}) + if err != nil { + cleanup() + return nil, err + } + c.pc = pc + connectedCh := make(chan struct{}, 1) + answerCh := make(chan error, 1) + c.installHandlers(ctx, cfg, connectedCh) + + track, err := webrtc.NewTrackLocalStaticSample( + webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000}, + "video", fmt.Sprintf("load-%d", ordinal), + ) + if err != nil { + cleanup() + return nil, err + } + transceiver, err := pc.AddTransceiverFromTrack(track, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendrecv}) + if err != nil { + cleanup() + return nil, fmt.Errorf("add video transceiver: %w", err) + } + go c.readSenderRTCP(ctx, transceiver.Sender()) + go c.readSignaling(ctx, answerCh) + + offer, err := pc.CreateOffer(nil) + if err == nil { + err = pc.SetLocalDescription(offer) + } + if err == nil { + err = c.ws.write(map[string]any{"type": "offer", "session_id": c.sessionID, "sdp": offer.SDP}) + } + if err != nil { + cleanup() + return nil, fmt.Errorf("send offer: %w", err) + } + + select { + case err = <-answerCh: + if err != nil { + cleanup() + return nil, err + } + case <-setupCtx.Done(): + cleanup() + return nil, fmt.Errorf("wait for answer: %w", setupCtx.Err()) + } + select { + case <-connectedCh: + case <-setupCtx.Done(): + cleanup() + return nil, fmt.Errorf("wait for WebRTC connection: %w", setupCtx.Err()) + } + go func() { + if err := streamIVF(ctx, cfg.Input, cfg.FPS, track, &c.sent); err != nil && !errors.Is(err, context.Canceled) { + c.setError(err) + } + }() + return c, nil +} + +func (c *loadClient) installHandlers(ctx context.Context, cfg config, connectedCh chan<- struct{}) { + c.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) { + if c.ws == nil { + return + } + message := signalMessage{Type: "ice_candidate", SessionID: c.sessionID} + if candidate != nil { + jsonCandidate := candidate.ToJSON() + message.Candidate = &jsonCandidate.Candidate + message.SDPMid = jsonCandidate.SDPMid + message.SDPMLineIndex = jsonCandidate.SDPMLineIndex + } + if err := c.ws.write(message); err != nil && ctx.Err() == nil { + c.setError(fmt.Errorf("send ICE candidate: %w", err)) + } + }) + c.pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { + c.stateMu.Lock() + c.pcState = state.String() + c.stateMu.Unlock() + if state == webrtc.PeerConnectionStateConnected && c.connected.CompareAndSwap(false, true) { + select { + case connectedCh <- struct{}{}: + default: + } + } + if state == webrtc.PeerConnectionStateFailed { + c.setError(fmt.Errorf("PeerConnection failed")) + } + }) + c.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) { + c.stateMu.Lock() + c.iceState = state.String() + c.stateMu.Unlock() + }) + c.pc.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) { + if track.Kind() == webrtc.RTPCodecTypeVideo { + go c.readProcessedTrack(ctx, track) + } + }) +} + +func (c *loadClient) readSignaling(ctx context.Context, answerCh chan<- error) { + answered := false + for { + var message signalMessage + if err := c.ws.conn.ReadJSON(&message); err != nil { + if ctx.Err() == nil { + c.setError(fmt.Errorf("read signaling: %w", err)) + } + if !answered { + select { + case answerCh <- err: + default: + } + } + return + } + switch message.Type { + case "answer": + if message.SessionID != c.sessionID { + continue + } + err := c.pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeAnswer, SDP: message.SDP}) + answered = true + select { + case answerCh <- err: + default: + } + case "error": + var payload map[string]any + _ = json.Unmarshal(message.Error, &payload) + err := fmt.Errorf("signaling error: %v", payload) + c.setError(err) + if !answered { + select { + case answerCh <- err: + default: + } + } + } + } +} + +func (c *loadClient) readProcessedTrack(ctx context.Context, track *webrtc.TrackRemote) { + var base, max uint16 + var cycles uint64 + var received uint64 + var initialized bool + var transit int64 + var jitter float64 + clockRate := int64(track.Codec().ClockRate) + for { + packet, _, err := track.ReadRTP() + if err != nil { + if ctx.Err() == nil && !errors.Is(err, io.EOF) { + c.setError(fmt.Errorf("read processed RTP: %w", err)) + } + return + } + arrival := time.Now().UnixNano() * clockRate / int64(time.Second) + currentTransit := arrival - int64(packet.Timestamp) + if initialized { + d := currentTransit - transit + if d < 0 { + d = -d + } + jitter += (float64(d) - jitter) / 16 + if packet.SequenceNumber < max && max-packet.SequenceNumber > 0x8000 { + cycles += 1 << 16 + max = packet.SequenceNumber + } else if packet.SequenceNumber > max && packet.SequenceNumber-max < 0x8000 { + max = packet.SequenceNumber + } + } else { + base, max, initialized = packet.SequenceNumber, packet.SequenceNumber, true + } + transit = currentTransit + received++ + expected := cycles + uint64(max) - uint64(base) + 1 + if expected > received { + c.received.lost.Store(expected - received) + } else { + c.received.lost.Store(0) + } + c.received.jitterMicros.Store(uint64(jitter * 1_000_000 / float64(clockRate))) + c.received.packets.Add(1) + c.received.bytes.Add(uint64(len(packet.Payload))) + if packet.Marker { + c.received.frames.Add(1) + } + } +} + +func (c *loadClient) readSenderRTCP(ctx context.Context, sender *webrtc.RTPSender) { + for { + packets, _, err := sender.ReadRTCP() + if err != nil { + return + } + for _, packet := range packets { + report, ok := packet.(*rtcp.ReceiverReport) + if !ok { + continue + } + for _, reception := range report.Reports { + if reception.LastSenderReport == 0 { + continue + } + now := compactNTP(time.Now()) + rtt := now - reception.LastSenderReport - reception.Delay + if rtt < 10*65536 { + c.received.rttMicros.Store(uint64(rtt) * 1_000_000 / 65536) + } + } + } + if ctx.Err() != nil { + return + } + } +} + +func compactNTP(t time.Time) uint32 { + const ntpEpochOffset = 2208988800 + seconds := uint64(t.Unix()+ntpEpochOffset) & 0xffff + fraction := uint64(t.Nanosecond()) * (1 << 32) / uint64(time.Second) + return uint32(seconds<<16 | (fraction >> 16)) +} + +func (c *loadClient) setError(err error) { + if err == nil { + return + } + c.failed.Store(true) + c.errorMu.Lock() + c.errorText = err.Error() + c.errorMu.Unlock() + log.Printf("client=%d session=%s error=%v", c.ordinal, c.sessionID, err) +} + +func (c *loadClient) states() (string, string, string) { + c.stateMu.RLock() + pc, ice := c.pcState, c.iceState + c.stateMu.RUnlock() + c.errorMu.RLock() + errText := c.errorText + c.errorMu.RUnlock() + return pc, ice, errText +} + +func (c *loadClient) Close() { + c.closeOnce.Do(func() { + c.cancel() + if c.pc != nil { + _ = c.pc.Close() + } + if c.ws != nil { + _ = c.ws.conn.Close() + } + if c.sessionID != "" { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := deleteSession(ctx, c.http, c.baseURL, c.sessionID); err != nil { + c.setError(err) + } + } + }) +} diff --git a/loadtest/pion-client/client_test.go b/loadtest/pion-client/client_test.go new file mode 100644 index 0000000..1fbc1b2 --- /dev/null +++ b/loadtest/pion-client/client_test.go @@ -0,0 +1,38 @@ +package main + +import ( + "strings" + "testing" + "time" +) + +func TestWebsocketURL(t *testing.T) { + got, err := websocketURL("https://example.test:8443/api") + if err != nil { + t.Fatal(err) + } + if got != "wss://example.test:8443/signaling" { + t.Fatalf("unexpected URL: %s", got) + } +} + +func TestCompactNTPAdvancesAt65536TicksPerSecond(t *testing.T) { + start := time.Unix(1_700_000_000, 0) + delta := compactNTP(start.Add(time.Second)) - compactNTP(start) + if delta != 65536 { + t.Fatalf("unexpected compact NTP delta: %d", delta) + } +} + +func TestRecorderWritesJSONAndCSV(t *testing.T) { + dir := t.TempDir() + r := newRecorder(config{Scenario: "step"}) + r.rows = append(r.rows, sampleRow{Timestamp: time.Now(), Client: 1, SessionID: "test"}) + jsonPath, csvPath, err := r.write(dir + "/result") + if err != nil { + t.Fatal(err) + } + if !strings.HasSuffix(jsonPath, ".json") || !strings.HasSuffix(csvPath, ".csv") { + t.Fatalf("unexpected result paths: %s %s", jsonPath, csvPath) + } +} diff --git a/loadtest/pion-client/config.go b/loadtest/pion-client/config.go new file mode 100644 index 0000000..35dbb8a --- /dev/null +++ b/loadtest/pion-client/config.go @@ -0,0 +1,62 @@ +package main + +import ( + "flag" + "fmt" + "strings" + "time" +) + +type config struct { + BaseURL string + Input string + Scenario string + Clients int + StartClients int + StepSize int + StepDuration time.Duration + RampDuration time.Duration + Duration time.Duration + StatsInterval time.Duration + ConnectTimeout time.Duration + FPS float64 + Output string +} + +func parseConfig() (config, error) { + var cfg config + flag.StringVar(&cfg.BaseURL, "base-url", "http://127.0.0.1:8000", "InnoLive HTTP base URL") + flag.StringVar(&cfg.Input, "input", "", "VP8 IVF video file (required)") + flag.StringVar(&cfg.Scenario, "scenario", "step", "load shape: step, spike, or soak") + flag.IntVar(&cfg.Clients, "clients", 10, "maximum concurrent PeerConnections") + flag.IntVar(&cfg.StartClients, "start-clients", 1, "initial clients for step scenario") + flag.IntVar(&cfg.StepSize, "step-size", 1, "clients added at each step") + flag.DurationVar(&cfg.StepDuration, "step-duration", 2*time.Minute, "time held at each step") + flag.DurationVar(&cfg.RampDuration, "ramp-duration", 30*time.Second, "time to reach target in spike/soak") + flag.DurationVar(&cfg.Duration, "duration", 10*time.Minute, "hold time after ramp (spike/soak)") + flag.DurationVar(&cfg.StatsInterval, "stats-interval", 5*time.Second, "statistics sampling interval") + flag.DurationVar(&cfg.ConnectTimeout, "connect-timeout", 30*time.Second, "per-client connection timeout") + flag.Float64Var(&cfg.FPS, "fps", 0, "override input FPS; 0 uses IVF header") + flag.StringVar(&cfg.Output, "output", "results/pion-load", "output path prefix (.json and .csv are appended)") + flag.Parse() + + cfg.BaseURL = strings.TrimRight(cfg.BaseURL, "/") + if cfg.Input == "" { + return cfg, fmt.Errorf("-input is required") + } + if cfg.Clients < 1 || cfg.StartClients < 0 || cfg.StartClients > cfg.Clients { + return cfg, fmt.Errorf("clients must be >= 1 and start-clients must be between 0 and clients") + } + if cfg.StepSize < 1 || cfg.StepDuration <= 0 || cfg.RampDuration < 0 || cfg.Duration <= 0 { + return cfg, fmt.Errorf("scenario counts and durations must be positive") + } + if cfg.StatsInterval <= 0 || cfg.ConnectTimeout <= 0 || cfg.FPS < 0 { + return cfg, fmt.Errorf("stats/connect durations must be positive and fps cannot be negative") + } + switch cfg.Scenario { + case "step", "spike", "soak": + default: + return cfg, fmt.Errorf("unsupported scenario %q", cfg.Scenario) + } + return cfg, nil +} diff --git a/loadtest/pion-client/go.mod b/loadtest/pion-client/go.mod new file mode 100644 index 0000000..2a65df4 --- /dev/null +++ b/loadtest/pion-client/go.mod @@ -0,0 +1,31 @@ +module innolive/loadtest/pion-client + +go 1.23.0 + +require ( + github.com/gorilla/websocket v1.5.3 + github.com/pion/rtcp v1.2.15 + github.com/pion/webrtc/v4 v4.1.2 +) + +require ( + github.com/google/uuid v1.6.0 // indirect + github.com/pion/datachannel v1.5.10 // indirect + github.com/pion/dtls/v3 v3.0.6 // indirect + github.com/pion/ice/v4 v4.0.10 // indirect + github.com/pion/interceptor v0.1.40 // indirect + github.com/pion/logging v0.2.3 // indirect + github.com/pion/mdns/v2 v2.0.7 // indirect + github.com/pion/randutil v0.1.0 // indirect + github.com/pion/rtp v1.8.22 // indirect + github.com/pion/sctp v1.8.39 // indirect + github.com/pion/sdp/v3 v3.0.13 // indirect + github.com/pion/srtp/v3 v3.0.5 // indirect + github.com/pion/stun/v3 v3.0.0 // indirect + github.com/pion/transport/v3 v3.0.7 // indirect + github.com/pion/turn/v4 v4.0.0 // indirect + github.com/wlynxg/anet v0.0.5 // indirect + golang.org/x/crypto v0.33.0 // indirect + golang.org/x/net v0.35.0 // indirect + golang.org/x/sys v0.30.0 // indirect +) diff --git a/loadtest/pion-client/go.sum b/loadtest/pion-client/go.sum new file mode 100644 index 0000000..f1a1d40 --- /dev/null +++ b/loadtest/pion-client/go.sum @@ -0,0 +1,52 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/pion/datachannel v1.5.10 h1:ly0Q26K1i6ZkGf42W7D4hQYR90pZwzFOjTq5AuCKk4o= +github.com/pion/datachannel v1.5.10/go.mod h1:p/jJfC9arb29W7WrxyKbepTU20CFgyx5oLo8Rs4Py/M= +github.com/pion/dtls/v3 v3.0.6 h1:7Hkd8WhAJNbRgq9RgdNh1aaWlZlGpYTzdqjy9x9sK2E= +github.com/pion/dtls/v3 v3.0.6/go.mod h1:iJxNQ3Uhn1NZWOMWlLxEEHAN5yX7GyPvvKw04v9bzYU= +github.com/pion/ice/v4 v4.0.10 h1:P59w1iauC/wPk9PdY8Vjl4fOFL5B+USq1+xbDcN6gT4= +github.com/pion/ice/v4 v4.0.10/go.mod h1:y3M18aPhIxLlcO/4dn9X8LzLLSma84cx6emMSu14FGw= +github.com/pion/interceptor v0.1.40 h1:e0BjnPcGpr2CFQgKhrQisBU7V3GXK6wrfYrGYaU6Jq4= +github.com/pion/interceptor v0.1.40/go.mod h1:Z6kqH7M/FYirg3frjGJ21VLSRJGBXB/KqaTIrdqnOic= +github.com/pion/logging v0.2.3 h1:gHuf0zpoh1GW67Nr6Gj4cv5Z9ZscU7g/EaoC/Ke/igI= +github.com/pion/logging v0.2.3/go.mod h1:z8YfknkquMe1csOrxK5kc+5/ZPAzMxbKLX5aXpbpC90= +github.com/pion/mdns/v2 v2.0.7 h1:c9kM8ewCgjslaAmicYMFQIde2H9/lrZpjBkN8VwoVtM= +github.com/pion/mdns/v2 v2.0.7/go.mod h1:vAdSYNAT0Jy3Ru0zl2YiW3Rm/fJCwIeM0nToenfOJKA= +github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= +github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= +github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo= +github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0= +github.com/pion/rtp v1.8.22 h1:8NCVDDF+uSJmMUkjLJVnIr/HX7gPesyMV1xFt5xozXc= +github.com/pion/rtp v1.8.22/go.mod h1:rF5nS1GqbR7H/TCpKwylzeq6yDM+MM6k+On5EgeThEM= +github.com/pion/sctp v1.8.39 h1:PJma40vRHa3UTO3C4MyeJDQ+KIobVYRZQZ0Nt7SjQnE= +github.com/pion/sctp v1.8.39/go.mod h1:cNiLdchXra8fHQwmIoqw0MbLLMs+f7uQ+dGMG2gWebE= +github.com/pion/sdp/v3 v3.0.13 h1:uN3SS2b+QDZnWXgdr69SM8KB4EbcnPnPf2Laxhty/l4= +github.com/pion/sdp/v3 v3.0.13/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E= +github.com/pion/srtp/v3 v3.0.5 h1:8XLB6Dt3QXkMkRFpoqC3314BemkpMQK2mZeJc4pUKqo= +github.com/pion/srtp/v3 v3.0.5/go.mod h1:r1G7y5r1scZRLe2QJI/is+/O83W2d+JoEsuIexpw+uM= +github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw= +github.com/pion/stun/v3 v3.0.0/go.mod h1:HvCN8txt8mwi4FBvS3EmDghW6aQJ24T+y+1TKjB5jyU= +github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1o0= +github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo= +github.com/pion/turn/v4 v4.0.0 h1:qxplo3Rxa9Yg1xXDxxH8xaqcyGUtbHYw4QSCvmFWvhM= +github.com/pion/turn/v4 v4.0.0/go.mod h1:MuPDkm15nYSklKpN8vWJ9W2M0PlyQZqYt1McGuxG7mA= +github.com/pion/webrtc/v4 v4.1.2 h1:mpuUo/EJ1zMNKGE79fAdYNFZBX790KE7kQQpLMjjR54= +github.com/pion/webrtc/v4 v4.1.2/go.mod h1:xsCXiNAmMEjIdFxAYU0MbB3RwRieJsegSB2JZsGN+8U= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= +github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= +golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus= +golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M= +golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8= +golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/loadtest/pion-client/main.go b/loadtest/pion-client/main.go new file mode 100644 index 0000000..c477c96 --- /dev/null +++ b/loadtest/pion-client/main.go @@ -0,0 +1,202 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "sync" + "syscall" + "time" + + "github.com/pion/webrtc/v4" +) + +type controller struct { + cfg config + ctx context.Context + http *http.Client + iceServers []webrtc.ICEServer + recorder *recorder + mu sync.RWMutex + clients []*loadClient + attempts []*loadClient + target int + nextOrdinal int +} + +func main() { + cfg, err := parseConfig() + if err != nil { + log.Fatal(err) + } + file, _, _, err := openIVF(cfg.Input, cfg.FPS) + if err != nil { + log.Fatal(err) + } + _ = file.Close() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + httpClient := &http.Client{Timeout: cfg.ConnectTimeout} + iceServers, err := fetchICEServers(ctx, httpClient, cfg.BaseURL) + if err != nil { + log.Fatal(err) + } + c := &controller{cfg: cfg, ctx: ctx, http: httpClient, iceServers: iceServers, recorder: newRecorder(cfg)} + done := make(chan struct{}) + go func() { c.collectStats(); close(done) }() + + err = c.runScenario() + c.snapshot() + c.closeAll() + stop() + <-done + jsonPath, csvPath, writeErr := c.recorder.write(cfg.Output) + if writeErr != nil { + log.Fatalf("write results: %v", writeErr) + } + log.Printf("results: %s, %s", jsonPath, csvPath) + if err != nil && !errors.Is(err, context.Canceled) { + log.Fatal(err) + } +} + +func (c *controller) runScenario() error { + log.Printf("scenario=%s clients=%d input=%s", c.cfg.Scenario, c.cfg.Clients, c.cfg.Input) + switch c.cfg.Scenario { + case "step": + start := c.cfg.StartClients + if start == 0 { + start = c.cfg.StepSize + } + for target := start; ; target += c.cfg.StepSize { + if target > c.cfg.Clients { + target = c.cfg.Clients + } + if err := c.scaleTo(target, 0); err != nil { + log.Printf("scale warning: %v", err) + } + if err := waitContext(c.ctx, c.cfg.StepDuration); err != nil { + return err + } + if target == c.cfg.Clients { + break + } + } + case "spike": + if err := c.scaleTo(c.cfg.Clients, c.cfg.RampDuration); err != nil { + log.Printf("scale warning: %v", err) + } + return waitContext(c.ctx, c.cfg.Duration) + case "soak": + if c.cfg.StartClients > 0 { + if err := c.scaleTo(c.cfg.StartClients, 0); err != nil { + log.Printf("scale warning: %v", err) + } + } + if err := c.scaleTo(c.cfg.Clients, c.cfg.RampDuration); err != nil { + log.Printf("scale warning: %v", err) + } + return waitContext(c.ctx, c.cfg.Duration) + } + return nil +} + +func (c *controller) scaleTo(target int, ramp time.Duration) error { + c.mu.Lock() + current := len(c.clients) + c.target = target + c.mu.Unlock() + if target <= current { + return nil + } + delay := time.Duration(0) + if ramp > 0 { + delay = ramp / time.Duration(target-current) + } + var failures int + for attempt := 0; attempt < target-current; attempt++ { + if c.ctx.Err() != nil { + return c.ctx.Err() + } + c.mu.Lock() + c.nextOrdinal++ + ordinal := c.nextOrdinal + c.mu.Unlock() + client, err := startLoadClient(c.ctx, c.cfg, ordinal, c.iceServers, c.http) + if err != nil { + failures++ + log.Printf("client=%d start failed: %v", ordinal, err) + _, cancel := context.WithCancel(c.ctx) + failed := &loadClient{ordinal: ordinal, startedAt: time.Now(), pcState: "failed", iceState: "failed", cancel: cancel, http: c.http, baseURL: c.cfg.BaseURL} + failed.setError(err) + c.mu.Lock() + c.attempts = append(c.attempts, failed) + c.mu.Unlock() + } else { + c.mu.Lock() + c.clients = append(c.clients, client) + c.attempts = append(c.attempts, client) + c.mu.Unlock() + log.Printf("client=%d session=%s connected", ordinal, client.sessionID) + } + if delay > 0 && attempt+1 < target-current { + if err := waitContext(c.ctx, delay); err != nil { + return err + } + } + } + if failures > 0 { + return fmt.Errorf("%d of %d new clients failed", failures, target-current) + } + return nil +} + +func (c *controller) collectStats() { + ticker := time.NewTicker(c.cfg.StatsInterval) + defer ticker.Stop() + for { + select { + case <-c.ctx.Done(): + return + case <-ticker.C: + c.snapshot() + } + } +} + +func (c *controller) snapshot() { + c.mu.RLock() + clients := append([]*loadClient(nil), c.attempts...) + target := c.target + c.mu.RUnlock() + c.recorder.sample(target, clients) + log.Printf("stats %s", c.recorder.summary(clients)) +} + +func (c *controller) closeAll() { + c.mu.RLock() + clients := append([]*loadClient(nil), c.attempts...) + c.mu.RUnlock() + var wg sync.WaitGroup + for _, client := range clients { + wg.Add(1) + go func(client *loadClient) { defer wg.Done(); client.Close() }(client) + } + wg.Wait() +} + +func waitContext(ctx context.Context, duration time.Duration) error { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/loadtest/pion-client/media.go b/loadtest/pion-client/media.go new file mode 100644 index 0000000..0649e54 --- /dev/null +++ b/loadtest/pion-client/media.go @@ -0,0 +1,84 @@ +package main + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "sync/atomic" + "time" + + "github.com/pion/webrtc/v4/pkg/media" + "github.com/pion/webrtc/v4/pkg/media/ivfreader" +) + +type mediaCounters struct { + frames atomic.Uint64 + bytes atomic.Uint64 +} + +func streamIVF(ctx context.Context, path string, fpsOverride float64, track sampleWriter, counters *mediaCounters) error { + for { + file, reader, frameDuration, err := openIVF(path, fpsOverride) + if err != nil { + return err + } + err = playIVF(ctx, file, reader, frameDuration, track, counters) + _ = file.Close() + if errors.Is(err, io.EOF) { + continue + } + return err + } +} + +type sampleWriter interface { + WriteSample(media.Sample) error +} + +func openIVF(path string, fpsOverride float64) (*os.File, *ivfreader.IVFReader, time.Duration, error) { + file, err := os.Open(path) + if err != nil { + return nil, nil, 0, fmt.Errorf("open input: %w", err) + } + reader, header, err := ivfreader.NewWith(file) + if err != nil { + _ = file.Close() + return nil, nil, 0, fmt.Errorf("read IVF header: %w", err) + } + fps := fpsOverride + if fps == 0 { + if header.TimebaseNumerator == 0 { + _ = file.Close() + return nil, nil, 0, fmt.Errorf("IVF timebase numerator is zero; supply -fps") + } + fps = float64(header.TimebaseDenominator) / float64(header.TimebaseNumerator) + } + if fps <= 0 { + _ = file.Close() + return nil, nil, 0, fmt.Errorf("input FPS must be greater than zero") + } + return file, reader, time.Duration(float64(time.Second) / fps), nil +} + +func playIVF(ctx context.Context, file *os.File, reader *ivfreader.IVFReader, frameDuration time.Duration, track sampleWriter, counters *mediaCounters) error { + ticker := time.NewTicker(frameDuration) + defer ticker.Stop() + for { + frame, _, err := reader.ParseNextFrame() + if err != nil { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + if err := track.WriteSample(media.Sample{Data: frame, Duration: frameDuration}); err != nil { + return fmt.Errorf("write video sample: %w", err) + } + counters.frames.Add(1) + counters.bytes.Add(uint64(len(frame))) + } +} diff --git a/loadtest/pion-client/media_test.go b/loadtest/pion-client/media_test.go new file mode 100644 index 0000000..1660e9b --- /dev/null +++ b/loadtest/pion-client/media_test.go @@ -0,0 +1,75 @@ +package main + +import ( + "context" + "encoding/binary" + "errors" + "os" + "sync" + "testing" + + "github.com/pion/webrtc/v4/pkg/media" +) + +type cancelWriter struct { + mu sync.Mutex + frames int + cancel context.CancelFunc +} + +func (w *cancelWriter) WriteSample(media.Sample) error { + w.mu.Lock() + defer w.mu.Unlock() + w.frames++ + if w.frames == 3 { + w.cancel() + } + return nil +} + +func TestStreamIVFLoopsAtEOF(t *testing.T) { + path := writeTestIVF(t) + ctx, cancel := context.WithCancel(context.Background()) + writer := &cancelWriter{cancel: cancel} + err := streamIVF(ctx, path, 1000, writer, &mediaCounters{}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected cancellation, got %v", err) + } + writer.mu.Lock() + frames := writer.frames + writer.mu.Unlock() + if frames != 3 { + t.Fatalf("expected looped third frame, got %d", frames) + } +} + +func writeTestIVF(t *testing.T) string { + t.Helper() + path := t.TempDir() + "/input.ivf" + file, err := os.Create(path) + if err != nil { + t.Fatal(err) + } + defer file.Close() + header := make([]byte, 32) + copy(header[0:4], "DKIF") + binary.LittleEndian.PutUint16(header[6:8], 32) + copy(header[8:12], "VP80") + binary.LittleEndian.PutUint16(header[12:14], 2) + binary.LittleEndian.PutUint16(header[14:16], 2) + binary.LittleEndian.PutUint32(header[16:20], 30) + binary.LittleEndian.PutUint32(header[20:24], 1) + binary.LittleEndian.PutUint32(header[24:28], 2) + if _, err := file.Write(header); err != nil { + t.Fatal(err) + } + for i := uint64(0); i < 2; i++ { + frameHeader := make([]byte, 12) + binary.LittleEndian.PutUint32(frameHeader[0:4], 1) + binary.LittleEndian.PutUint64(frameHeader[4:12], i) + if _, err := file.Write(append(frameHeader, byte(i))); err != nil { + t.Fatal(err) + } + } + return path +} diff --git a/loadtest/pion-client/results.go b/loadtest/pion-client/results.go new file mode 100644 index 0000000..fa4d7cd --- /dev/null +++ b/loadtest/pion-client/results.go @@ -0,0 +1,153 @@ +package main + +import ( + "encoding/csv" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strconv" + "sync" + "time" +) + +type sampleRow struct { + Timestamp time.Time `json:"timestamp"` + ElapsedSeconds float64 `json:"elapsed_seconds"` + TargetClients int `json:"target_clients"` + Client int `json:"client"` + SessionID string `json:"session_id"` + Connected bool `json:"connected"` + Failed bool `json:"failed"` + PCState string `json:"pc_state"` + ICEState string `json:"ice_state"` + SendFPS float64 `json:"send_fps"` + ReceiveFPS float64 `json:"receive_fps"` + SentFrames uint64 `json:"sent_frames"` + ReceivedFrames uint64 `json:"received_frames"` + ReceivedPackets uint64 `json:"received_packets"` + LostPackets uint64 `json:"lost_packets"` + PacketLossPct float64 `json:"packet_loss_pct"` + JitterMS float64 `json:"jitter_ms"` + RTTMS float64 `json:"rtt_ms"` + ReceivedBytes uint64 `json:"received_bytes"` + Error string `json:"error,omitempty"` +} + +type resultFile struct { + StartedAt time.Time `json:"started_at"` + FinishedAt time.Time `json:"finished_at"` + Scenario string `json:"scenario"` + Config config `json:"config"` + Samples []sampleRow `json:"samples"` +} + +type previousFrames struct { + sent, received uint64 + at time.Time +} + +type recorder struct { + mu sync.Mutex + started time.Time + cfg config + rows []sampleRow + previous map[int]previousFrames +} + +func newRecorder(cfg config) *recorder { + return &recorder{started: time.Now(), cfg: cfg, previous: make(map[int]previousFrames)} +} + +func (r *recorder) sample(target int, clients []*loadClient) { + r.mu.Lock() + defer r.mu.Unlock() + now := time.Now() + for _, client := range clients { + sentFrames, receivedFrames := client.sent.frames.Load(), client.received.frames.Load() + previous := r.previous[client.ordinal] + seconds := now.Sub(previous.at).Seconds() + var sendFPS, receiveFPS float64 + if !previous.at.IsZero() && seconds > 0 { + sendFPS = float64(sentFrames-previous.sent) / seconds + receiveFPS = float64(receivedFrames-previous.received) / seconds + } + r.previous[client.ordinal] = previousFrames{sent: sentFrames, received: receivedFrames, at: now} + packets, lost := client.received.packets.Load(), client.received.lost.Load() + var lossPct float64 + if packets+lost > 0 { + lossPct = float64(lost) * 100 / float64(packets+lost) + } + pcState, iceState, errText := client.states() + r.rows = append(r.rows, sampleRow{ + Timestamp: now, ElapsedSeconds: now.Sub(r.started).Seconds(), TargetClients: target, + Client: client.ordinal, SessionID: client.sessionID, Connected: client.connected.Load(), Failed: client.failed.Load(), + PCState: pcState, ICEState: iceState, SendFPS: sendFPS, ReceiveFPS: receiveFPS, + SentFrames: sentFrames, ReceivedFrames: receivedFrames, ReceivedPackets: packets, LostPackets: lost, + PacketLossPct: lossPct, JitterMS: float64(client.received.jitterMicros.Load()) / 1000, + RTTMS: float64(client.received.rttMicros.Load()) / 1000, ReceivedBytes: client.received.bytes.Load(), Error: errText, + }) + } +} + +func (r *recorder) write(prefix string) (string, string, error) { + r.mu.Lock() + defer r.mu.Unlock() + if err := os.MkdirAll(filepath.Dir(prefix), 0o755); err != nil && filepath.Dir(prefix) != "." { + return "", "", err + } + jsonPath, csvPath := prefix+".json", prefix+".csv" + jsonFile, err := os.Create(jsonPath) + if err != nil { + return "", "", err + } + encoder := json.NewEncoder(jsonFile) + encoder.SetIndent("", " ") + err = encoder.Encode(resultFile{StartedAt: r.started, FinishedAt: time.Now(), Scenario: r.cfg.Scenario, Config: r.cfg, Samples: r.rows}) + closeErr := jsonFile.Close() + if err != nil { + return "", "", err + } + if closeErr != nil { + return "", "", closeErr + } + + csvFile, err := os.Create(csvPath) + if err != nil { + return "", "", err + } + w := csv.NewWriter(csvFile) + _ = w.Write([]string{"timestamp", "elapsed_seconds", "target_clients", "client", "session_id", "connected", "failed", "pc_state", "ice_state", "send_fps", "receive_fps", "sent_frames", "received_frames", "received_packets", "lost_packets", "packet_loss_pct", "jitter_ms", "rtt_ms", "received_bytes", "error"}) + for _, row := range r.rows { + _ = w.Write([]string{row.Timestamp.Format(time.RFC3339Nano), f(row.ElapsedSeconds), strconv.Itoa(row.TargetClients), strconv.Itoa(row.Client), row.SessionID, strconv.FormatBool(row.Connected), strconv.FormatBool(row.Failed), row.PCState, row.ICEState, f(row.SendFPS), f(row.ReceiveFPS), u(row.SentFrames), u(row.ReceivedFrames), u(row.ReceivedPackets), u(row.LostPackets), f(row.PacketLossPct), f(row.JitterMS), f(row.RTTMS), u(row.ReceivedBytes), row.Error}) + } + w.Flush() + err = w.Error() + closeErr = csvFile.Close() + if err != nil { + return "", "", err + } + if closeErr != nil { + return "", "", closeErr + } + return jsonPath, csvPath, nil +} + +func f(value float64) string { return strconv.FormatFloat(value, 'f', 3, 64) } +func u(value uint64) string { return strconv.FormatUint(value, 10) } + +func (r *recorder) summary(clients []*loadClient) string { + connected, failed := 0, 0 + var sent, received uint64 + for _, client := range clients { + if client.connected.Load() { + connected++ + } + if client.failed.Load() { + failed++ + } + sent += client.sent.frames.Load() + received += client.received.frames.Load() + } + return fmt.Sprintf("attempts=%d connected=%d failed=%d sent_frames=%d received_frames=%d", len(clients), connected, failed, sent, received) +} diff --git a/loadtest/pion-client/signaling.go b/loadtest/pion-client/signaling.go new file mode 100644 index 0000000..4f38668 --- /dev/null +++ b/loadtest/pion-client/signaling.go @@ -0,0 +1,131 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + + "github.com/gorilla/websocket" + "github.com/pion/webrtc/v4" +) + +type signalMessage struct { + Type string `json:"type"` + SessionID string `json:"session_id"` + SDP string `json:"sdp,omitempty"` + Candidate *string `json:"candidate"` + SDPMid *string `json:"sdpMid,omitempty"` + SDPMLineIndex *uint16 `json:"sdpMLineIndex,omitempty"` + Error json.RawMessage `json:"error,omitempty"` +} + +type signalSocket struct { + conn *websocket.Conn + mu sync.Mutex +} + +func (s *signalSocket) write(v any) error { + s.mu.Lock() + defer s.mu.Unlock() + return s.conn.WriteJSON(v) +} + +func createSession(ctx context.Context, client *http.Client, baseURL string, ordinal int) (string, error) { + body, _ := json.Marshal(map[string]any{"metadata": map[string]string{ + "load_test": "pion", "client": fmt.Sprintf("%d", ordinal), + }}) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/sessions", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + return "", fmt.Errorf("create session: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusCreated { + data, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return "", fmt.Errorf("create session: HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(data))) + } + var result struct { + SessionID string `json:"session_id"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", fmt.Errorf("decode session response: %w", err) + } + if result.SessionID == "" { + return "", fmt.Errorf("decode session response: session_id is empty") + } + return result.SessionID, nil +} + +func deleteSession(ctx context.Context, client *http.Client, baseURL, sessionID string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, baseURL+"/sessions/"+url.PathEscape(sessionID), nil) + if err != nil { + return err + } + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotFound { + return fmt.Errorf("delete session: HTTP %d", resp.StatusCode) + } + return nil +} + +func fetchICEServers(ctx context.Context, client *http.Client, baseURL string) ([]webrtc.ICEServer, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, baseURL+"/webrtc/config", nil) + if err != nil { + return nil, err + } + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("get WebRTC config: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("get WebRTC config: HTTP %d", resp.StatusCode) + } + var payload struct { + ICEServers []struct { + URLs []string `json:"urls"` + Username string `json:"username"` + Credential string `json:"credential"` + } `json:"iceServers"` + } + if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil { + return nil, err + } + servers := make([]webrtc.ICEServer, 0, len(payload.ICEServers)) + for _, server := range payload.ICEServers { + servers = append(servers, webrtc.ICEServer{URLs: server.URLs, Username: server.Username, Credential: server.Credential}) + } + return servers, nil +} + +func websocketURL(baseURL string) (string, error) { + u, err := url.Parse(baseURL) + if err != nil { + return "", err + } + switch u.Scheme { + case "http": + u.Scheme = "ws" + case "https": + u.Scheme = "wss" + default: + return "", fmt.Errorf("base URL must use http or https") + } + u.Path = "/signaling" + u.RawQuery = "" + return u.String(), nil +} diff --git a/loadtest/scripts/collect-server-baseline.sh b/loadtest/scripts/collect-server-baseline.sh new file mode 100644 index 0000000..25ff404 --- /dev/null +++ b/loadtest/scripts/collect-server-baseline.sh @@ -0,0 +1,71 @@ +#!/usr/bin/env bash +set -euo pipefail + +if [[ $# -lt 2 || $# -gt 3 ]]; then + echo "usage: $0 OUTPUT_DIR DURATION_SECONDS [INTERVAL_SECONDS]" >&2 + exit 2 +fi + +output_dir=$1 +duration_seconds=$2 +interval_seconds=${3:-1} + +mkdir -p "$output_dir" + +server_pid=$(pgrep -f '/opt/inno-live-venv/bin/python -m uvicorn app.main:app' | head -n 1 || true) +if [[ -z "$server_pid" ]]; then + echo "uvicorn server process not found" >&2 + exit 1 +fi + +echo "waiting for an active WebRTC session (timeout: 600s)" +wait_deadline=$((SECONDS + 600)) +while ((SECONDS < wait_deadline)); do + active_sessions=$(curl --fail --silent http://127.0.0.1:8000/metrics | + awk '$1 == "innolive_active_sessions" { print $2; exit }') + if awk -v value="${active_sessions:-0}" 'BEGIN { exit !(value > 0) }'; then + echo "active session detected; starting ${duration_seconds}s collection" + break + fi + sleep 1 +done + +if ((SECONDS >= wait_deadline)); then + echo "timed out waiting for an active WebRTC session" >&2 + exit 1 +fi + +curl --fail --silent http://127.0.0.1:8000/metrics > "$output_dir/metrics-initial.prom" +curl --fail --silent http://127.0.0.1:8000/health > "$output_dir/health-initial.json" + +echo "timestamp,gpu_util_pct,gpu_memory_mib" > "$output_dir/gpu.csv" +echo "timestamp,active_sessions,queue_size,received_real,processed_real,dropped_real,failures_real" > "$output_dir/app.csv" + +pidstat -h -r -u -p "$server_pid" "$interval_seconds" "$duration_seconds" > "$output_dir/pidstat.txt" & +pidstat_pid=$! +/opt/inno-live-venv/bin/py-spy record --pid "$server_pid" --duration "$duration_seconds" --rate 50 --format speedscope --output "$output_dir/py-spy-speedscope.json" & +py_spy_pid=$! + +samples=$((duration_seconds / interval_seconds)) +for ((sample = 0; sample < samples; sample++)); do + timestamp=$(date --iso-8601=seconds) + gpu_values=$(nvidia-smi --query-gpu=utilization.gpu,memory.used --format=csv,noheader,nounits | head -n 1 | tr -d ' ') + echo "$timestamp,$gpu_values" >> "$output_dir/gpu.csv" + + metrics=$(curl --fail --silent http://127.0.0.1:8000/metrics) + metric_value() { + local pattern=$1 + local value + value=$(awk -v pattern="$pattern" '$0 ~ pattern { print $NF; exit }' <<< "$metrics") + echo "${value:-0}" + } + echo "$timestamp,$(metric_value '^innolive_active_sessions '),$(metric_value '^innolive_processing_queue_size '),$(metric_value '^innolive_frame_received_total\\{mode="real"\\} '),$(metric_value '^innolive_frame_processed_total\\{mode="real"\\} '),$(metric_value '^innolive_frame_dropped_total\\{mode="real"\\} '),$(metric_value '^innolive_frame_failures_total\\{mode="real"\\} ')" >> "$output_dir/app.csv" + + sleep "$interval_seconds" +done + +wait "$pidstat_pid" +wait "$py_spy_pid" +curl --fail --silent http://127.0.0.1:8000/health > "$output_dir/health-final.json" +curl --fail --silent http://127.0.0.1:8000/sessions > "$output_dir/sessions-final.json" +curl --fail --silent http://127.0.0.1:8000/metrics > "$output_dir/metrics-final.prom" diff --git a/loadtest/videos/README.md b/loadtest/videos/README.md new file mode 100644 index 0000000..f0218cf --- /dev/null +++ b/loadtest/videos/README.md @@ -0,0 +1,13 @@ +# Local test videos + +Video assets in this directory are intentionally ignored by Git. Keep the +source MP4 as `input.mp4` and the Pion-ready VP8 file as `720p30.ivf`. + +Generate the load-test input with: + +```bash +ffmpeg -i input.mp4 -an \ + -vf "scale=1280:720:force_original_aspect_ratio=decrease,pad=1280:720:(ow-iw)/2:(oh-ih)/2,fps=30" \ + -c:v libvpx -deadline good -cpu-used 4 -b:v 2M -maxrate 2M -bufsize 4M \ + 720p30.ivf +``` diff --git a/loadtest/windows/configure-firewall.ps1 b/loadtest/windows/configure-firewall.ps1 new file mode 100644 index 0000000..79d2ba1 --- /dev/null +++ b/loadtest/windows/configure-firewall.ps1 @@ -0,0 +1,66 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [ValidatePattern('^(?:\d{1,3}\.){3}\d{1,3}$')] + [string]$RemoteAddress +) + +$ErrorActionPreference = 'Stop' +$wslCreatorId = '{40E0AC32-46A5-438A-A0B2-2B479E8F2E90}' + +$rules = @( + @{ + DisplayName = 'InnoLive Mac TCP 8000' + Protocol = 'TCP' + LocalPort = '8000' + }, + @{ + DisplayName = 'InnoLive Mac WebRTC UDP' + Protocol = 'UDP' + LocalPort = '50000-60000' + } +) + +foreach ($definition in $rules) { + $existing = Get-NetFirewallRule ` + -DisplayName $definition.DisplayName ` + -ErrorAction SilentlyContinue + + if ($null -ne $existing) { + Remove-NetFirewallRule -DisplayName $definition.DisplayName + } + + New-NetFirewallRule ` + -DisplayName $definition.DisplayName ` + -Direction Inbound ` + -Action Allow ` + -Protocol $definition.Protocol ` + -LocalPort $definition.LocalPort ` + -RemoteAddress $RemoteAddress ` + -Profile Any | Out-Null + + # WSL2 mirrored networking has a separate Hyper-V firewall layer. Windows + # Defender rules alone do not allow inbound traffic through that layer. + $hyperVRuleName = ($definition.DisplayName -replace '[^A-Za-z0-9]', '-') + $existingHyperVRule = Get-NetFirewallHyperVRule ` + -Name $hyperVRuleName ` + -ErrorAction SilentlyContinue + + if ($null -ne $existingHyperVRule) { + Remove-NetFirewallHyperVRule -Name $hyperVRuleName + } + + New-NetFirewallHyperVRule ` + -Name $hyperVRuleName ` + -DisplayName $definition.DisplayName ` + -Direction Inbound ` + -VMCreatorId $wslCreatorId ` + -Action Allow ` + -Enabled True ` + -Profiles Any ` + -Protocol $definition.Protocol ` + -LocalPorts $definition.LocalPort ` + -RemoteAddresses $RemoteAddress | Out-Null +} + +Write-Host "Configured Windows and WSL Hyper-V firewall access for $RemoteAddress" diff --git a/observability/grafana/dashboards/innolive.json b/observability/grafana/dashboards/innolive.json new file mode 100644 index 0000000..55aab69 --- /dev/null +++ b/observability/grafana/dashboards/innolive.json @@ -0,0 +1,39 @@ +{ + "title": "InnoLive Load Test", + "uid": "innolive-load-test", + "schemaVersion": 42, + "refresh": "5s", + "time": {"from": "now-30m", "to": "now"}, + "panels": [ + { + "type": "stat", + "title": "Active sessions", + "gridPos": {"h": 6, "w": 6, "x": 0, "y": 0}, + "targets": [{"expr": "innolive_active_sessions", "refId": "A"}] + }, + { + "type": "timeseries", + "title": "Processed FPS by mode", + "gridPos": {"h": 8, "w": 9, "x": 6, "y": 0}, + "targets": [{"expr": "sum by (mode) (rate(innolive_frame_processed_total[1m]))", "refId": "A"}] + }, + { + "type": "timeseries", + "title": "Frame processing p95", + "gridPos": {"h": 8, "w": 9, "x": 15, "y": 0}, + "targets": [{"expr": "histogram_quantile(0.95, sum by (le, mode) (rate(innolive_frame_processing_duration_seconds_bucket[5m])))", "refId": "A"}] + }, + { + "type": "timeseries", + "title": "Dropped frames", + "gridPos": {"h": 8, "w": 12, "x": 0, "y": 8}, + "targets": [{"expr": "sum by (mode) (rate(innolive_frame_dropped_total[1m]))", "refId": "A"}] + }, + { + "type": "timeseries", + "title": "AI queue depth", + "gridPos": {"h": 8, "w": 12, "x": 12, "y": 8}, + "targets": [{"expr": "innolive_processing_queue_size", "refId": "A"}] + } + ] +} diff --git a/observability/grafana/provisioning/dashboards/dashboards.yml b/observability/grafana/provisioning/dashboards/dashboards.yml new file mode 100644 index 0000000..802e0bf --- /dev/null +++ b/observability/grafana/provisioning/dashboards/dashboards.yml @@ -0,0 +1,7 @@ +apiVersion: 1 + +providers: + - name: InnoLive + type: file + options: + path: /var/lib/grafana/dashboards diff --git a/observability/grafana/provisioning/datasources/prometheus.yml b/observability/grafana/provisioning/datasources/prometheus.yml new file mode 100644 index 0000000..86fd346 --- /dev/null +++ b/observability/grafana/provisioning/datasources/prometheus.yml @@ -0,0 +1,8 @@ +apiVersion: 1 + +datasources: + - name: Prometheus + type: prometheus + access: proxy + url: http://prometheus:9090 + isDefault: true diff --git a/observability/prometheus.yml b/observability/prometheus.yml new file mode 100644 index 0000000..680442a --- /dev/null +++ b/observability/prometheus.yml @@ -0,0 +1,8 @@ +global: + scrape_interval: 5s + evaluation_interval: 5s + +scrape_configs: + - job_name: innolive + static_configs: + - targets: ["host.docker.internal:8000"] diff --git a/privacy_blur/core.py b/privacy_blur/core.py index c76e209..6b39b07 100644 --- a/privacy_blur/core.py +++ b/privacy_blur/core.py @@ -2,6 +2,7 @@ from dataclasses import dataclass import threading +from time import perf_counter from typing import Any import numpy as np @@ -393,9 +394,13 @@ async def apply_async(self, frame: Any) -> Any: return await self.engine.process_async(frame) video_frame_type = _video_frame_type() if video_frame_type is not None and isinstance(frame, video_frame_type): + decode_started = perf_counter() image = frame.to_ndarray(format="bgr24") + self.engine.runtime.observe_stage("decode", perf_counter() - decode_started) processed = await self.engine.process_async(image) + encode_started = perf_counter() protected = video_frame_type.from_ndarray(processed, format="bgr24") + self.engine.runtime.observe_stage("encode", perf_counter() - encode_started) protected.pts = frame.pts protected.time_base = frame.time_base return protected diff --git a/privacy_blur/runtime.py b/privacy_blur/runtime.py index 7cffa69..7c66e17 100644 --- a/privacy_blur/runtime.py +++ b/privacy_blur/runtime.py @@ -3,6 +3,7 @@ import asyncio import atexit from collections import defaultdict +from contextlib import nullcontext from concurrent.futures import Future from dataclasses import dataclass import logging @@ -10,6 +11,7 @@ from queue import Empty, Full, Queue import threading import time +from collections.abc import Callable from typing import TYPE_CHECKING, Any import cv2 @@ -180,6 +182,7 @@ def __init__(self, config: PrivacyBlurConfig): maxsize=config.max_pending_frames * config.max_batch_size ) self._closed = False + self._stage_observer: Callable[[str, float], None] | None = None self._stats_lock = threading.Lock() self._batches = 0 self._frames = 0 @@ -214,6 +217,20 @@ def __init__(self, config: PrivacyBlurConfig): self._worker.start() self._identity_worker.start() + def set_stage_observer( + self, observer: Callable[[str, float], None] | None + ) -> None: + self._stage_observer = observer + + def observe_stage(self, stage: str, seconds: float) -> None: + observer = self._stage_observer + if observer is None: + return + try: + observer(stage, seconds) + except Exception: + LOGGER.exception("AI stage observer failed: stage=%s", stage) + def _create_ort_session(self, path: Path) -> ort.InferenceSession: available = ort.get_available_providers() if self.cuda_enabled and "CUDAExecutionProvider" not in available: @@ -267,7 +284,29 @@ def _load_reference_features( config: PrivacyBlurConfig, *, sface_session: ort.InferenceSession | None = None, + detector: Any | None = None, + device: str | None = None, + half: bool | None = None, ) -> torch.Tensor: + reference_faces = self._reference_faces_from_config( + config, + detector=detector, + device=device, + half=half, + ) + return self._reference_features_from_faces( + reference_faces, + sface_session=sface_session, + ) + + def _reference_faces_from_config( + self, + config: PrivacyBlurConfig, + *, + detector: Any | None = None, + device: str | None = None, + half: bool | None = None, + ) -> list[tuple[np.ndarray, np.ndarray, np.ndarray | None]]: references: list[tuple[Path, np.ndarray]] = [] for reference_path in config.identity_image_paths(): reference = cv2.imread(str(reference_path)) @@ -280,6 +319,9 @@ def _load_reference_features( detections_by_reference = self._predict_yolo( [reference for _, reference in references], config=config, + detector=detector, + device=device, + half=half, ) reference_faces: list[ tuple[np.ndarray, np.ndarray, np.ndarray | None] @@ -306,7 +348,14 @@ def _load_reference_features( if not reference_faces: raise ReferenceFaceNotDetectedError("No reference images configured") + return reference_faces + def _reference_features_from_faces( + self, + reference_faces: list[tuple[np.ndarray, np.ndarray, np.ndarray | None]], + *, + sface_session: ort.InferenceSession | None = None, + ) -> torch.Tensor: embeddings, valid = self._face_embeddings_from_landmarks( reference_faces, sface_session=sface_session, @@ -448,14 +497,28 @@ def _identity_worker_loop(self) -> None: def prepare_identity(self, config: PrivacyBlurConfig) -> _PreparedIdentity: """Build a reference embedding without pausing live frame batches.""" + if self._closed: + raise RuntimeError("The shared inference runtime is closed") if not config.enable_identity_exclusion: raise ValueError("identity exclusion must be enabled") for reference_path in config.identity_image_paths(): _require_file(reference_path) + _require_file(config.face_model_path) _require_file(config.face_recognition_model_path) + # Use an isolated CPU detector so uploads do not take the live detector lock. + reference_detector = YOLO(str(config.face_model_path)) + reference_faces = self._reference_faces_from_config( + config, + detector=reference_detector, + device="cpu", + half=False, + ) + with self._identity_lock: + if self._closed: + raise RuntimeError("The shared inference runtime is closed") sface_session = ( self.sface_session if self._sface_model_path == config.face_recognition_model_path @@ -465,8 +528,8 @@ def prepare_identity(self, config: PrivacyBlurConfig) -> _PreparedIdentity: sface_session = self._create_ort_session( config.face_recognition_model_path, ) - reference_features = self._load_reference_features( - config, + reference_features = self._reference_features_from_faces( + reference_faces, sface_session=sface_session, ) return _PreparedIdentity( @@ -500,10 +563,12 @@ def _run_batch(self, requests: list[_FrameRequest]) -> list[np.ndarray]: ] detections: dict[int, list[FaceDetection]] = {} if detect_indices: + detection_started = time.perf_counter() detected = self._predict_yolo( [requests[index].frame for index in detect_indices], config=runtime_config, ) + self.observe_stage("detection", time.perf_counter() - detection_started) detections = dict(zip(detect_indices, detected)) identity_work: list[_IdentityRequest] = [] @@ -529,11 +594,14 @@ def _run_batch(self, requests: list[_FrameRequest]) -> list[np.ndarray]: if identity_work: self._submit_identity_work(identity_work) privacy_boxes = [request.engine.finish_frame() for request in requests] - return self._blur_frames( + blur_started = time.perf_counter() + blurred = self._blur_frames( [request.frame for request in requests], privacy_boxes, config=runtime_config, ) + self.observe_stage("blur", time.perf_counter() - blur_started) + return blurred def _submit_identity_work(self, requests: list[_IdentityRequest]) -> None: dropped: list[_IdentityRequest] = [] @@ -552,6 +620,7 @@ def _submit_identity_work(self, requests: list[_IdentityRequest]) -> None: request.engine.apply_identity_matches([(request.track, -1.0)]) def _run_identity_batch(self, requests: list[_IdentityRequest]) -> None: + identity_started = time.perf_counter() active_requests = [ request for request in requests @@ -578,6 +647,7 @@ def _run_identity_batch(self, requests: list[_IdentityRequest]) -> None: for request in active_requests ] ) + self.observe_stage("identity", time.perf_counter() - identity_started) grouped: dict[FacePrivacyEngine, list[tuple[FaceTrack, float]]] = defaultdict( list ) @@ -591,22 +661,31 @@ def _predict_yolo( frames: list[np.ndarray], *, config: PrivacyBlurConfig | None = None, + detector: Any | None = None, + device: str | None = None, + half: bool | None = None, ) -> list[list[FaceDetection]]: from .core import FaceDetection if not frames: return [] config = config or self.config - with self._detector_lock: - results = self.detector.predict( + active_detector = detector if detector is not None else self.detector + detector_context = ( + nullcontext() if detector is not None else self._detector_lock + ) + predict_device = device if device is not None else str(self.device) + predict_half = half if half is not None else self.half_enabled + with detector_context: + results = active_detector.predict( source=frames, imgsz=config.face_imgsz, conf=config.face_conf_threshold, iou=config.face_iou_threshold, max_det=config.face_max_detections, verbose=False, - device=str(self.device), - half=self.half_enabled, + device=predict_device, + quantize=16 if predict_half else None, batch=len(frames), ) diff --git a/requirements.txt b/requirements.txt index cd62e22..85a37d9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,9 +8,10 @@ av==16.1.0 python-multipart==0.0.31 httpx==0.28.1 python-dotenv==1.2.2 +prometheus-client==0.23.1 numpy>=1.24 opencv-python-headless>=4.8 -ultralytics>=8.0 +ultralytics>=8.4.92 torch>=2.0 torchvision>=0.15 onnx>=1.17 diff --git a/test-docs/test-backlog.md b/test-docs/test-backlog.md new file mode 100644 index 0000000..4e6507a --- /dev/null +++ b/test-docs/test-backlog.md @@ -0,0 +1,123 @@ +# 부하 테스트 백로그 + +## 현재 상태 + +- [x] Go/Pion 실제 영상 부하 클라이언트 구현 +- [x] k6 REST 세션 부하 스크립트 구현 +- [x] Prometheus 애플리케이션 메트릭 추가 +- [x] Prometheus/Grafana Compose 및 대시보드 파일 추가 +- [x] 720p 30FPS VP8/IVF 기준 영상 1종 준비 +- [x] Windows/WSL2 Ubuntu 24.04 서버 환경 구성 +- [x] RTX 3060 Ti CUDA PyTorch 실제 연산 확인 +- [x] Python 전체 테스트 통과: 278개 +- [x] WSL mirrored networking 적용 +- [x] 맥북과 Linux 서버 간 최초 WebRTC 미디어 연결 + +맥북 공인 IP로 범위를 제한한 Windows 방화벽 규칙을 적용했고, `bypass` 모드에서 +30초 원격 미디어 smoke test에 성공했습니다. ICE는 `completed`, PeerConnection은 +`connected`에 도달했으며 송신 약 30FPS와 수신 약 29.5FPS를 확인했습니다. + +## 우선순위 1: 최초 원격 WebRTC smoke test + +1. [x] 맥북의 접속 IP 확인 +2. [x] 해당 IP로 제한한 Windows 방화벽 규칙 추가 +3. [x] 서버를 `AI_PRIVACY_MODE=bypass`로 실행 +4. [x] 맥북에서 `/health`, `/webrtc/config` 확인 +5. [x] Pion 클라이언트 1개를 30초간 실행 +6. [x] 송신·수신 FPS, ICE 상태, 오류 및 결과 CSV/JSON 확인 +7. [x] `fixed_delay`, `real` 모드에서 각각 1개 연결 확인 + +## 우선순위 2: Python 단일 세션 기준선 + +합격 기준은 720p 30FPS 입력의 90%인 27FPS입니다. + +- [x] `quantize=16` 적용 후 실제 GPU 예열 시험 +- [x] 예열 후 단일 세션 5분 기준선 시험 +- [x] 첫 30초를 제외한 수신·서버 처리 FPS 평균 및 p95 계산 +- [x] 클라이언트 프레임 손실과 서버 stale drop 기록 +- [x] Prometheus, `nvidia-smi`, `pidstat`, `py-spy` 결과 저장 +- [x] 종료 후 활성 세션과 처리 큐가 0인지 확인 + +2026-07-13 기준선은 수신 25.51FPS, 서버 처리 25.52FPS, 송수신 프레임 차이 +14.98%, stale drop 17.47%로 실패했습니다. 따라서 다중 세션 시험은 보류합니다. + +단일 세션에서 평균 27FPS 또는 드롭률 10% 기준을 만족하지 못하면 다중 세션 +시험을 중단하고 decode·AI·blur·encode 단계별 병목 계측을 먼저 구현합니다. + +## 우선순위 3: 계단식·지속·스파이크 부하 + +단일 세션 기준선을 통과한 뒤 진행합니다. + +- [ ] `1 → 2 → 4 → 8` 세션을 각 5분 유지 +- [ ] 최초 기준 위반 단계 주변을 1세션 단위로 재시험 +- [ ] 최대 안정 세션의 70~80%로 1~2시간 지속 시험 +- [ ] `0 → 최대 후보 → 0 → 최대 후보` 스파이크·재연결 시험 + +## 우선순위 4: End-to-End 지연 측정 + +현재 서버 내부 프레임 처리 시간만 측정합니다. 송신부터 처리 영상 수신까지 +측정하려면 다음 중 하나가 필요합니다. + +- 영상에 프레임 번호와 타임코드 삽입 +- RTP header extension으로 `frame_id`와 timestamp 전달 + +첫 구현은 대표 클라이언트 1~3개가 타임코드 영상을 판독하는 방식으로 진행합니다. + +## 우선순위 5: 단계별 처리 지연 + +병목 분석을 위해 전체 프레임 처리 시간을 다음 단계로 분리해야 합니다. + +- [x] Prometheus 단계별 histogram 추가 +- [x] decode 계측 +- [x] AI detection 계측 +- [x] identity comparison 계측 +- [x] blur 계측 +- [x] encode 입력 프레임 생성 계측 +- [x] 계측 적용 서버를 재시작하고 예열 후 단일 세션 5분 재프로파일 +- [x] 단계별 평균과 p95를 비교해 최우선 최적화 대상 선정 +- [ ] detection 호출·전처리·추론·후처리를 세분화하고 최적화 +- [ ] blur 마스크 생성·Gaussian 연산·CPU 반환을 세분화하고 최적화 +- [ ] 최적화 후 단일 세션 5분 기준선 재시험 + +현재 `encode` 단계는 BGR 배열을 AV 프레임으로 만드는 시간입니다. aiortc 내부 VP8 +코덱 압축 시간은 포함하지 않으며, 필요하면 aiortc encoder 계층에 별도 훅을 추가합니다. + +## 우선순위 6: 모니터링 스택 검증 + +Docker Desktop 또는 실제 Linux Docker 환경에서 다음 스택을 기동하고 검증합니다. + +```bash +docker compose -f docker-compose.test.yml up -d +``` + +- Prometheus: +- Grafana: (`admin` / `admin`) + +## 우선순위 7: 시스템·GPU 자원 메트릭 + +- CPU/RAM: Node Exporter +- 컨테이너: cAdvisor +- GPU/VRAM: DCGM Exporter 또는 `nvidia-smi` 주기 수집기 + +## 우선순위 8: 고정 테스트 영상 확대 + +현재 기준 영상 1종은 준비됐습니다. 본 시험에는 다음 3종을 고정해야 합니다. + +- 얼굴 1명, 움직임 적음 +- 얼굴 4명, 움직임 보통 +- 얼굴 다수, 움직임 많음 + +## 우선순위 9: 네트워크 장애 자동화 + +`tc netem` 시나리오와 자동 복구 스크립트가 필요합니다. + +- 지연 50/100/150ms +- jitter +- 손실 1/3/5% +- 3초 단절 +- 테스트 종료 시 자동 복구 + +## 우선순위 10: Go 서버 비교 구현 + +현재 저장소는 Python 서버입니다. Python 기준선 시험을 완료한 뒤 동일 API와 AI 의미를 +갖는 Go 서버를 구현해야 최종 언어 비교가 가능합니다. diff --git a/test-docs/test-plan.md b/test-docs/test-plan.md new file mode 100644 index 0000000..af22f8e --- /dev/null +++ b/test-docs/test-plan.md @@ -0,0 +1,637 @@ +## 결론 + +두 테스트를 따로 진행하기보다 **동일한 WebRTC 부하 생성기를 이용해 Python 서버와 Go 서버를 번갈아 테스트하는 통합 방식**이 가장 적절해. + +InnoLive는 클라이언트가 원본 영상을 WebRTC로 서버와 Vision AI에 보내고, 비식별화된 영상을 다시 받아 스트리밍 플랫폼으로 송출하는 구조다. 따라서 여기서 말하는 “수용 가능한 유저 수”는 일반적인 회원 수나 시청자 수가 아니라 **동시에 비식별화 방송을 실행하는 활성 스트림 수**로 정의해야 한다. 스트리밍 플랫폼의 시청자 트래픽은 InnoLive 서버가 직접 처리하지 않기 때문이다. + +즉, 최종적으로 다음 결과를 얻는 테스트를 설계하면 된다. + +> 동일 서버 사양에서 720p 30FPS 방송을 Python 서버는 최대 몇 개, Go 서버는 최대 몇 개까지 안정적으로 처리하는가? +> + +--- + +# 1. 권장 테스트 전체 구조 + +``` +┌──────────────────────────────────────┐ +│ 부하 발생 서버 │ +│ │ +│ k6 │ +│ - 로그인/API 호출 │ +│ - WebSocket 시그널링 │ +│ │ +│ Go + Pion WebRTC 가상 클라이언트 │ +│ - 테스트 영상 반복 송신 │ +│ - 처리 영상 수신 │ +│ - 지연 시간/FPS/손실 측정 │ +└──────────────────┬───────────────────┘ + │ WebRTC + ▼ +┌──────────────────────────────────────┐ +│ 테스트 대상 │ +│ │ +│ A. Python 서버 │ +│ B. Go 서버 │ +│ │ +│ 동일한 AI 모델 / 동일한 Vision AI 서버 │ +└──────────────────┬───────────────────┘ + │ + ▼ +┌──────────────────────────────────────┐ +│ Prometheus + Grafana │ +│ CPU / RAM / GPU / Network / FPS │ +│ 처리 지연 / 세션 수 / 오류율 │ +└──────────────────────────────────────┘ +``` + +부하 발생기는 반드시 테스트 대상 서버와 **다른 컴퓨터나 인스턴스**에서 실행하는 게 좋다. 한 컴퓨터에서 부하 생성과 서버를 동시에 실행하면 부하 생성기 자체가 CPU나 네트워크를 사용해서 결과가 왜곡될 수 있다. + +--- + +# 2. 사용할 도구 + +## 2.1 실제 영상 부하: Go + Pion WebRTC + +가장 중요한 도구다. + +일반적인 HTTP 부하 테스트 도구는 REST API 요청은 많이 만들 수 있지만, 실제 WebRTC 영상 트랙을 지속적으로 송수신하지는 못한다. 따라서 InnoLive에서는 **가상의 방송 클라이언트**를 직접 만들어야 한다. + +추천 구성: + +- Go +- Pion WebRTC +- 고정된 MP4 테스트 영상 +- 한 프로세스에서 여러 PeerConnection 생성 +- 각 PeerConnection이 영상 송신 및 처리 결과 수신 +- 세션마다 WebRTC 통계와 영상 처리 지연 기록 + +Pion의 `webrtc-bench` 프로젝트는 영상 파일을 WebRTC로 재생하면서 PeerConnection 수, CPU 사용량, 연결 단계별 지연 등을 측정하는 구조이므로 부하 생성기 구현의 출발점으로 활용하기 좋다. (GitHub) + +부하 생성기는 다음처럼 동작하면 된다. + +``` +1번 가상 클라이언트 연결 +2번 가상 클라이언트 연결 +3번 가상 클라이언트 연결 +... +N번 가상 클라이언트 연결 + +각 클라이언트: +- 같은 테스트 영상 반복 송신 +- 처리된 영상 수신 +- 수신 FPS 기록 +- 프레임 누락 기록 +- RTT, jitter, packet loss 기록 +- 연결 종료 여부 기록 +``` + +## 2.2 API와 시그널링: k6 + +k6는 다음 영역에 사용하면 된다. + +- 로그인 +- 방송 세션 생성 및 종료 +- Offer/Answer 교환 +- ICE candidate 전달 +- WebSocket 연결 +- 얼굴 등록 API +- 다수 사용자의 동시 재연결 + +현재 k6는 WebSocket 부하 테스트와 WebSocket 관련 연결·메시지 메트릭을 제공한다. 다만 **실제 RTP 영상 부하는 Pion 클라이언트가 담당**하고, k6는 제어 영역을 담당하도록 분리하는 게 좋다. (Grafana Labs) + +## 2.3 고정된 영상 입력: FFmpeg + +FFmpeg로 동일한 영상을 동일한 조건으로 준비한다. + +예시 테스트 프로필: + +| 프로필 | 해상도 | FPS | 얼굴 수 | 움직임 | +| --- | --- | --- | --- | --- | +| A | 1280×720 | 30 | 1명 | 적음 | +| B | 1280×720 | 30 | 4명 | 보통 | +| C | 1920×1080 | 30 | 4명 | 보통 | +| D | 1920×1080 | 30 | 10명 이상 | 많음 | + +얼굴 탐지·비식별화 서비스는 단순히 해상도만 같다고 부하가 동일하지 않다. 프레임 내 얼굴 수, 얼굴 크기, 움직임, 등록 인물 비교 여부에 따라 AI 연산량이 달라질 수 있으므로 테스트 영상도 여러 종류가 필요하다. + +FFmpeg는 해상도, FPS, 비트레이트, 코덱을 고정하고 테스트 영상을 반복 재생하는 데 사용할 수 있다. 필요하다면 FFmpeg의 필터를 이용해 출력 영상의 품질 비교도 할 수 있다. (FFmpeg) + +## 2.4 서버 모니터링: Prometheus + Grafana + +서버 내부에 다음 메트릭을 직접 추가하는 게 좋다. + +``` +innolive_active_sessions +innolive_connection_total +innolive_connection_failures_total + +innolive_frame_received_total +innolive_frame_processed_total +innolive_frame_dropped_total + +innolive_decode_duration_seconds +innolive_ai_duration_seconds +innolive_blur_duration_seconds +innolive_encode_duration_seconds +innolive_end_to_end_duration_seconds + +innolive_processing_queue_size +innolive_reconnect_total +``` + +Prometheus는 애플리케이션에 언어별 클라이언트 라이브러리를 붙여 내부 메트릭을 노출하는 방식을 지원한다. 호스트 자원은 Node Exporter, 컨테이너 자원은 cAdvisor, NVIDIA GPU를 사용한다면 DCGM Exporter로 수집할 수 있다. (Prometheus) + +Grafana 대시보드에는 최소한 다음 그래프가 있어야 한다. + +- 활성 WebRTC 세션 수 +- CPU 사용률 +- 프로세스 메모리 +- 세션당 CPU 및 메모리 +- GPU 사용률과 GPU 메모리 +- 초당 처리 프레임 +- 프레임 처리 지연 p50, p95, p99 +- WebRTC 패킷 손실률 +- 처리 큐 크기 +- 연결 실패 및 재연결 횟수 + +## 2.5 병목 분석 + +Python: + +- `py-spy` +- `cProfile` +- `pstats` + +Go: + +- `net/http/pprof` +- `go tool pprof` +- Go benchmark + +Python의 `cProfile`은 함수 호출 횟수와 누적 실행 시간을 확인할 수 있고, `py-spy`는 실행 중인 프로세스에 붙어서 비교적 낮은 부하로 샘플링 프로파일을 만들 수 있다. Go는 표준 `pprof` 엔드포인트를 통해 CPU, heap, goroutine, allocation 프로파일을 수집할 수 있다. (Python documentation) + +## 2.6 네트워크 환경 재현: Linux `tc netem` + +WebRTC는 실제 네트워크 상태에 따라 결과가 크게 달라지므로 정상 네트워크만 테스트하면 부족하다. + +`tc netem`을 사용하면 다음을 인위적으로 발생시킬 수 있다. + +- 네트워크 지연 +- jitter +- 패킷 손실 +- 패킷 중복 +- 패킷 순서 변경 + +`netem`은 실제 네트워크 특성을 시험하기 위해 delay, loss, duplication, corruption 등을 적용하는 Linux 네트워크 에뮬레이션 기능이다. (man7.org) + +예시 조건: + +``` +정상 환경: +delay 5ms / loss 0% + +일반 무선 환경: +delay 30ms ± 10ms / loss 0.5% + +불안정한 모바일 환경: +delay 80ms ± 30ms / loss 2% + +매우 불안정한 환경: +delay 150ms ± 50ms / loss 5% +``` + +--- + +# 3. 반드시 수집해야 할 성능 지표 + +## 서버 처리 지표 + +| 지표 | 의미 | +| --- | --- | +| 최대 동시 스트림 | SLO를 만족하면서 유지 가능한 방송 수 | +| 처리 FPS | 서버가 실제로 처리한 초당 프레임 | +| 프레임 처리 지연 | 디코딩부터 비식별화 및 인코딩까지 | +| End-to-End 지연 | 클라이언트 송신부터 처리 영상 수신까지 | +| 프레임 드롭률 | 입력됐지만 출력되지 못한 프레임 비율 | +| CPU 사용률 | 전체 및 세션당 사용량 | +| 메모리 사용량 | RSS, heap, 세션당 메모리 | +| GPU 사용률 | 연산 사용률과 VRAM | +| 큐 길이 | 처리 대기 중인 프레임 수 | +| 오류율 | 연결 실패, 추론 실패, 인코딩 실패 | + +## WebRTC 지표 + +- RTT +- jitter +- packetsLost +- framesDropped +- framesEncoded +- framesDecoded +- keyFramesEncoded +- outbound/inbound bitrate +- connection setup time +- ICE 연결 성공률 +- TURN 사용 비율 + +WebRTC의 `getStats()` 표준은 RTP 스트림, 코덱, 전송, ICE, 송수신 영상 등에 관한 통계를 제공하도록 정의되어 있다. 부하 클라이언트에서 해당 통계를 주기적으로 수집하면 된다. (W3C) + +--- + +# 4. End-to-End 지연 측정 방법 + +단순히 서버 함수 실행 시간만 재면 실제 사용자가 체감하는 지연을 알 수 없다. + +가장 정확한 방법은 각 프레임에 다음 정보를 부여하는 것이다. + +``` +frame_id +send_timestamp +session_id +``` + +가능한 방식은 두 가지다. + +### 방식 A: RTP Header Extension + +송신 프레임의 RTP 확장 헤더에 프레임 ID와 송신 시간을 넣고, 처리된 영상 반환 시 해당 값을 유지하거나 연결한다. + +장점: + +- 이미지 내용에 영향을 주지 않음 +- 자동화하기 좋음 +- 매우 정확함 + +단점: + +- 서버 파이프라인에서 메타데이터를 유지하도록 구현해야 함 + +### 방식 B: 영상 구석에 시간 코드 삽입 + +FFmpeg로 영상 한쪽 구석에 프레임 번호나 QR 형태의 시간 코드를 넣고, 반환 영상에서 읽는다. + +장점: + +- 서버 코드를 크게 수정하지 않아도 됨 +- 실제 영상 파이프라인 전체 지연 측정 가능 + +단점: + +- 시간 코드 인식 코드가 필요함 +- 블러 영역과 겹치지 않도록 해야 함 + +서로 다른 컴퓨터에서 시간을 비교할 때는 시스템 시계 오차가 들어갈 수 있으므로, 가능하면 **송신과 수신을 같은 부하 클라이언트 프로세스에서 측정**하는 것이 좋다. + +--- + +# 5. 테스트 시나리오 + +## 1단계: 단일 세션 기준선 + +사용자 1명만 연결한다. + +측정: + +- 기본 End-to-End 지연 +- 평균 CPU +- 평균 RAM +- AI 추론 시간 +- 인코딩 및 디코딩 시간 +- 정상 FPS +- 비식별화 정상 여부 + +이 결과가 Python과 Go 서버 비교의 기준선이 된다. + +--- + +## 2단계: 계단식 부하 테스트 + +다음과 같이 동시 스트림 수를 증가시킨다. + +``` +1개 → 5분 유지 +2개 → 5분 유지 +4개 → 5분 유지 +8개 → 5분 유지 +12개 → 5분 유지 +16개 → 5분 유지 +20개 → 5분 유지 +... +``` + +한계에 가까워지면 증가 단위를 줄인다. + +``` +20 → 22 → 24 → 26 → 28 +``` + +각 단계에서 다음 중 하나라도 발생하면 한계에 도달한 것으로 본다. + +- p95 지연이 설정한 기준 초과 +- 처리 FPS가 목표 이하로 감소 +- 프레임 드롭률이 기준 초과 +- 연결 실패 발생 +- 처리 큐가 계속 증가 +- CPU 또는 GPU가 장시간 포화 +- 메모리가 지속적으로 증가 +- 영상이 끊기거나 정지 + +--- + +## 3단계: 지속 부하 테스트 + +최대 처리량 테스트만 하면 메모리 누수나 장시간 연결 문제를 발견하기 어렵다. + +계단식 테스트에서 확인한 최대 안정 세션의 약 70~80% 수준을 이용해 다음 테스트를 한다. + +``` +동시 스트림 20개 +지속 시간 1~2시간 +``` + +확인 사항: + +- 메모리가 계속 증가하지 않는가 +- goroutine/thread가 누적되지 않는가 +- WebRTC 연결이 점차 끊기지 않는가 +- 처리 지연이 시간이 지나며 증가하지 않는가 +- GPU 메모리가 반환되는가 +- 프레임 처리 큐가 누적되지 않는가 + +--- + +## 4단계: 스파이크 테스트 + +모든 클라이언트가 동시에 방송을 시작하는 상황을 테스트한다. + +``` +0개 → 20개 즉시 연결 +20개 → 전체 연결 종료 +0개 → 20개 즉시 재연결 +``` + +이 테스트에서는 다음을 본다. + +- Offer/Answer 처리 병목 +- ICE 연결 실패 +- 서버 초기화 경쟁 +- AI 모델 요청 폭증 +- 소켓 및 파일 디스크립터 부족 +- 재연결 시 리소스 정리 실패 + +--- + +## 5단계: 네트워크 장애 테스트 + +각 동시 사용자 단계에서 다음 조건을 추가한다. + +``` +패킷 손실 1% +패킷 손실 3% +지연 50ms +지연 100ms + jitter 30ms +일시적인 3초 네트워크 단절 +``` + +WebRTC 연결이 복구되는지, 복구 후 영상과 서버 자원이 정상으로 돌아오는지 확인한다. + +--- + +# 6. Python과 Go의 공정한 비교 방법 + +언어만 다르게 하고 나머지는 모두 같아야 한다. + +## 고정해야 할 조건 + +- 같은 서버 장비 +- 같은 CPU 코어 수 +- 같은 메모리 제한 +- 같은 OS +- 같은 Docker 제한 +- 같은 WebRTC 코덱 +- 같은 해상도와 FPS +- 같은 비트레이트 +- 같은 AI 모델 +- 같은 AI 런타임 +- 같은 Vision AI 서버 +- 같은 테스트 영상 +- 같은 네트워크 조건 +- 같은 로그 레벨 +- 같은 부하 발생기 + +Python과 Go 서버를 동시에 실행하지 말고 다음처럼 번갈아 실행하는 게 좋다. + +``` +Python 1차 +Go 1차 +Python 2차 +Go 2차 +Python 3차 +Go 3차 +... +``` + +CPU 온도, 캐시, 백그라운드 프로세스 등의 영향을 줄이기 위해 각 조건을 최소 5회, 가능하면 10회 반복한다. + +--- + +# 7. AI 병목과 서버 언어 병목을 분리해야 함 + +가장 중요한 부분이다. + +전체 파이프라인에서 AI 추론이 대부분의 시간을 차지한다면 Python을 Go로 바꿔도 실제 End-to-End 속도는 거의 개선되지 않을 수 있다. + +따라서 다음 세 가지 모드를 각각 테스트해야 한다. + +## 모드 A: AI를 우회한 서버 테스트 + +``` +WebRTC 수신 +→ 영상 그대로 반환 +``` + +측정 대상: + +- WebRTC 처리 +- RTP 패킷 처리 +- 프레임 복사 +- 동시 연결 관리 +- 언어 런타임 차이 + +여기서는 Python과 Go 서버 자체의 차이가 가장 잘 드러난다. + +## 모드 B: 가짜 AI 지연 테스트 + +``` +WebRTC 수신 +→ 일정 시간 대기 +→ 영상 반환 +``` + +예를 들어 AI 추론 대신 항상 20ms를 기다린다. + +측정 대상: + +- 동일한 외부 처리 시간이 있을 때 서버의 동시성 처리 차이 +- Python 이벤트 루프와 Go goroutine 구조의 차이 + +## 모드 C: 실제 AI 포함 + +``` +WebRTC 수신 +→ 실제 얼굴 탐지 +→ 등록 인물 판별 +→ 블러 +→ 인코딩 +→ 반환 +``` + +이것이 실제 사용자 환경의 최종 성능이다. + +결과는 다음처럼 해석할 수 있다. + +``` +A에서는 Go가 크게 빠름 +C에서는 차이가 거의 없음 +→ AI 또는 인코딩이 병목 + +A와 C 모두 Go가 크게 빠름 +→ Python 서버의 패킷/프레임 처리 또는 동시성 구조가 병목 + +두 환경 모두 CPU는 여유롭지만 GPU가 100% +→ 서버 언어보다 AI 배치 처리 또는 GPU 확장이 우선 +``` + +--- + +# 8. 권장 성능 합격 기준 + +수치는 현재 단일 세션 측정 결과에 맞게 조정해야 하지만, 초기 기준은 다음처럼 잡을 수 있다. + +| 항목 | 예시 기준 | +| --- | --- | +| 연결 성공률 | 99% 이상 | +| 처리 FPS | 목표 FPS의 95% 이상 | +| 프레임 드롭률 | 1% 미만 | +| 서버 오류율 | 0.1% 미만 | +| End-to-End 지연 | 제품 목표 이하 | +| 메모리 증가 | 지속 테스트 중 안정화 | +| 처리 큐 | 지속적으로 증가하지 않음 | +| 비식별화 정확도 | 기존 Python보다 감소하지 않음 | + +InnoLive가 “초저지연”을 핵심 가치로 내세우고 있으므로, 단순히 서버가 죽지 않는지를 기준으로 하면 안 된다. 지연과 프레임 드롭, 비식별화 정상 동작을 모두 만족해야 수용 가능한 세션으로 계산해야 한다. + +최종 수용량은 다음처럼 정의하는 게 좋다. + +> 30분 이상 연결을 유지하면서 p95 지연, FPS, 프레임 드롭률, 오류율 기준을 모두 만족하는 최대 동시 방송 세션 수 +> + +운영 시에는 실험에서 나온 최대치의 100%를 사용하지 않고 약 20~30%의 여유 용량을 두는 편이 안전하다. + +--- + +# 9. 최종 결과표 형식 + +## 고정 사용자 수 성능 + +| 항목 | Python | Go | 차이 | +| --- | --- | --- | --- | +| 동시 스트림 | 10 | 10 | 동일 | +| p50 지연 | | | | +| p95 지연 | | | | +| p99 지연 | | | | +| 평균 처리 FPS | | | | +| 프레임 드롭률 | | | | +| CPU 사용률 | | | | +| 메모리 사용량 | | | | +| 세션당 메모리 | | | | +| GPU 사용률 | | | | + +## 최대 수용량 비교 + +| 항목 | Python | Go | +| --- | --- | --- | +| 최대 연결 가능 세션 | | | +| 최대 안정 세션 | | | +| 운영 권장 세션 | | | +| 세션당 CPU | | | +| 세션당 메모리 | | | +| 한계 도달 원인 | | | + +결과 계산은 다음처럼 하면 된다. + +``` +처리량 향상률 = +(Go 최대 세션 - Python 최대 세션) +÷ Python 최대 세션 × 100 + +p95 지연 개선 배수 = +Python p95 지연 ÷ Go p95 지연 + +메모리 절감률 = +(Python 메모리 - Go 메모리) +÷ Python 메모리 × 100 +``` + +예를 들어 결과가 다음과 같다면: + +``` +Python 최대 안정 세션: 12개 +Go 최대 안정 세션: 20개 +``` + +``` +처리량 향상률 += (20 - 12) / 12 × 100 += 약 66.7% +``` + +--- + +# 10. 실제 진행 순서 + +1. **성능 기준 정의** + - 720p 30FPS + - 코덱 및 비트레이트 고정 + - 허용 지연과 드롭률 결정 +2. **테스트 영상 제작** + - 얼굴 수와 움직임이 다른 영상 3~4개 준비 +3. **서버 메트릭 추가** + - 프레임 단계별 처리 시간 + - 활성 세션 + - 큐 길이 + - 실패 횟수 +4. **Pion 부하 클라이언트 개발** + - 영상 송신 + - 처리 영상 수신 + - WebRTC 통계 수집 + - 동시 세션 수 조절 +5. **Python 기준 테스트** + - 단일 세션 + - 계단식 부하 + - 지속 부하 + - 스파이크 및 네트워크 장애 +6. **Go 서버를 같은 인터페이스로 구현** +7. **동일한 부하 스크립트로 Go 테스트** +8. **AI 우회·가짜 AI·실제 AI 결과 비교** +9. **병목 프로파일링** + - Python: py-spy + - Go: pprof +10. **최대 안정 세션과 비용 산출** + +핵심 도구 구성은 다음으로 정리할 수 있다. + +``` +API·시그널링 부하: k6 +실제 WebRTC 영상 부하: Go + Pion +영상 소스 생성: FFmpeg +메트릭 수집: Prometheus +시각화: Grafana +컨테이너 측정: cAdvisor +GPU 측정: DCGM Exporter +네트워크 장애: tc netem +Python 분석: py-spy / cProfile +Go 분석: pprof +``` + +이 방식이면 **“현재 서버가 몇 명을 처리하는가”와 “Go로 전환했을 때 실제로 몇 배 개선되는가”를 동일한 실험에서 함께 증명**할 수 있다. \ No newline at end of file diff --git a/test-docs/test-result.md b/test-docs/test-result.md new file mode 100644 index 0000000..09a8921 --- /dev/null +++ b/test-docs/test-result.md @@ -0,0 +1,128 @@ +# 부하 테스트 결과 + +## 합격 기준 + +| 항목 | 기준 | +| --- | --- | +| 입력 프로필 | 1280×720, 30FPS, VP8/IVF | +| 평균 클라이언트 수신 FPS | 27FPS 이상 | +| 평균 서버 처리 FPS | 27FPS 이상 | +| 프레임 손실·드롭 | 10% 이하 | +| 연결 성공률 | 99% 이상 | +| 서버 처리 오류율 | 0.1% 이하 | +| 처리 큐 | 시험 종료까지 지속 증가하지 않음 | +| E2E p50/p95/p99 | 타임코드 측정 구현 전까지 측정 보류 | + +최종 최대 안정 세션은 위 기준을 모두 만족하며 30분 이상 유지 가능한 동시 방송 +세션 수로 정의합니다. 5분 시험은 후보 구간을 찾는 예비 용량 시험입니다. + +## 2026-07-12 원격 smoke test + +서버는 Windows/WSL2 Ubuntu 24.04, RTX 3060 Ti 8GB이고 부하 클라이언트는 +맥북에서 실행했습니다. 서버 방화벽은 맥북 공인 IP만 허용했습니다. + +| 모드 | 시간 | 송신 프레임 | 수신 프레임 | 수신 FPS | 결과 | +| --- | ---: | ---: | ---: | ---: | --- | +| bypass | 30초 | 899 | 884 | 약 29.5 | 통과 | +| fixed_delay 20ms | 30초 | 898 | 859 | 약 28.6 | 통과 | +| real cold | 30초 | 899 | 0 | 0 | 예열 결과, 판정 제외 | +| real warm | 30초 | 899 | 782 | 약 26.1 | 기준 미달 | + +### 서버 측 관찰 + +- `bypass`: 수신·처리 각각 898프레임, 처리 실패 0 +- `fixed_delay`: 서버 수신 876, 처리 875, stale drop 9, 평균 인위 지연 20.32ms +- `real cold`: 첫 AI 배치 약 46.4초로 30초 시험 종료 전 반환 프레임 없음 +- `real warm`: 서버 약 800프레임 처리, stale drop 98, 처리 실패 0 +- `real warm` 추론 약 24~30ms/프레임, GPU 메모리 약 1GB +- 모든 시험에서 ICE `completed`, PeerConnection `connected`, 종료 후 활성 세션 0 + +`real warm`은 현재 27FPS와 10% 드롭 기준을 만족하지 못합니다. 당시 Ultralytics가 +매 프레임 `half` 폐기 경고를 출력했으므로 `quantize=16` 적용 후 5분 기준선을 다시 +측정합니다. + +## Python 단일 세션 5분 기준선 + +첫 30초는 안정화 구간으로 제외하고 이후 4분 30초를 집계합니다. + +2026-07-13에 `quantize=16`과 별도 예열 실행을 적용해 측정했습니다. Pion 시험은 +1개 연결에 성공했고 8,999프레임을 송신해 7,595프레임을 수신했습니다. 아래 FPS와 +드롭 수치는 안정화 구간을 제외한 270초 창의 카운터 차이로 계산했습니다. + +| 항목 | 결과 | 합격 여부 | +| --- | ---: | --- | +| 평균 수신 FPS | 25.51 | 실패 (< 27) | +| 수신 FPS p95 | 27.60 | 참고값 | +| 평균 서버 처리 FPS | 25.52 | 실패 (< 27) | +| 서버 처리 지연 평균 / p95 버킷 | 31.76ms / 50ms 이하 | 참고값 | +| 클라이언트 송수신 프레임 차이 | 14.98% | 실패 (> 10%) | +| RTP 패킷 손실률 | 최종 0.177%, 최대 0.288% | 통과 | +| 서버 stale drop | 1,221 / 수신 6,991 (17.47%) | 실패 (> 10%) | +| 연결 성공률 | 100% (1/1) | 통과 | +| 처리 오류율 | 0% (0건) | 통과 | +| 평균/최대 CPU | 145.02% / 161% | 참고값 | +| 평균/최대 RSS | 3,479.57 / 3,479.59MiB | 참고값 | +| 평균/최대 GPU 사용률 | 33.57% / 58% | 참고값 | +| 평균/최대 VRAM | 1,043.53 / 1,082MiB | 참고값 | +| 종료 후 활성 세션/큐 | 0 / 0 | 통과 | +| E2E p50/p95/p99 | 측정 보류 | - | + +판정은 **단일 세션 기준선 실패**입니다. 네트워크 RTP 손실은 낮지만 서버 처리량이 +입력 30FPS보다 낮아 stale frame이 누적 폐기됐습니다. 계획에 따라 2·4·8 세션 확대는 +중단하고 단계별 병목 계측 후 단일 세션을 다시 측정합니다. + +수집 원본은 Git에서 제외된 `loadtest/server-results/real-single-5m/`에 저장했습니다. +`py-spy`는 12,783개 샘플을 기록했으며, CPU 프로파일은 +`py-spy-speedscope.json`에서 확인할 수 있습니다. + +## Python 단일 세션 단계별 재프로파일 + +2026-07-13에 단계별 Prometheus histogram을 적용하고 별도 예열 후 다시 5분간 +측정했습니다. 첫 30초를 제외한 270초 창의 결과입니다. + +| 항목 | 결과 | 판정 | +| --- | ---: | --- | +| 클라이언트 수신 FPS | 25.66 | 실패 (< 27) | +| 서버 처리 FPS | 25.67 | 실패 (< 27) | +| 송수신 프레임 차이 | 14.48% | 실패 (> 10%) | +| 서버 stale drop | 1,193 / 7,033 (16.96%) | 실패 (> 10%) | +| RTP 패킷 손실 | 0% | 통과 | +| 연결 성공 / 처리 오류 | 100% / 0% | 통과 | +| 평균/최대 CPU | 146.57% / 163% | 참고값 | +| 평균/최대 GPU | 33.32% / 57% | 참고값 | +| 평균/최대 RSS | 3,972.92 / 3,979.37MiB | 참고값 | +| 평균/최대 VRAM | 1,083.98 / 1,142MiB | 참고값 | +| 종료 후 활성 세션/큐 | 0 / 0 | 통과 | + +예열 누적 카운터를 차감한 단계별 평균은 다음과 같습니다. + +| 단계 | 평균 | 누적 histogram 기준 p95 버킷 | +| --- | ---: | ---: | +| decode | 0.64ms | 1ms 이하 | +| detection | 16.42ms | 33ms 이하 | +| identity | 미실행 | - | +| blur | 8.45ms | 20ms 이하 | +| encode 입력 프레임 생성 | 1.13ms | 5ms 이하 | + +이번 수집기는 종료 histogram만 저장했으므로 평균은 예열 직후 카운터를 차감해 +계산했지만 p95는 예열 샘플이 약 5.6% 포함된 누적 버킷입니다. 이후 수집부터 시작·종료 +Prometheus 스냅샷을 모두 저장하도록 수정했습니다. + +`py-spy` GPU worker 표본에서는 detection 경로가 포괄 시간의 62.8%, blur 경로가 +36.0%였습니다. 단일 세션 처리량 개선의 우선순위는 **detection 최적화 후 blur +최적화**입니다. GPU 사용률이 평균 33.32%라 GPU 연산 포화보다는 단일 프레임 호출과 +CPU/GPU 동기화 비용의 영향이 큰 상태입니다. + +## 향후 Python/Go 비교표 + +| 항목 | Python | Go | 차이 | +| --- | --- | --- | --- | +| 최대 연결 가능 세션 | | | | +| 최대 안정 세션 | | | | +| 운영 권장 세션 | | | | +| 평균 처리 FPS | | | | +| 프레임 드롭률 | | | | +| CPU 사용률 | | | | +| 메모리 사용량 | | | | +| GPU 사용률 | | | | +| 한계 도달 원인 | | | | diff --git a/test-docs/test-runbook.md b/test-docs/test-runbook.md new file mode 100644 index 0000000..e1cd1fb --- /dev/null +++ b/test-docs/test-runbook.md @@ -0,0 +1,63 @@ +# Local load-test runbook + +## Server modes + +Use the same command and change only `AI_PRIVACY_MODE`: + +- `bypass`: WebRTC frame forwarding without AI. +- `fixed_delay`: forwarding after `AI_PRIVACY_FIXED_DELAY_MS`. +- `real`: the production privacy pipeline. + +Example: + +```bash +AI_PRIVACY_ENABLED=true \ +AI_PRIVACY_MODE=real \ +AI_PRIVACY_DEVICE=cuda:0 \ +AI_PRIVACY_REQUIRE_GPU=true \ +AI_PRIVACY_USE_HALF=true \ +python -m uvicorn app.main:app --host 0.0.0.0 --port 8000 +``` + +## Metrics + +Prometheus metrics are exposed at `GET /metrics`. Start the supplied monitoring +stack with: + +```bash +docker compose -f docker-compose.test.yml up -d +``` + +- Prometheus: +- Grafana: (`admin` / `admin`) + +## Control-plane load + +The k6 script measures REST session lifecycle load only. It does not generate +RTP video traffic. + +```bash +BASE_URL=http://127.0.0.1:8000 VUS=10 HOLD=2m SESSION_SECONDS=5 \ +k6 run loadtest/k6/sessions.js +``` + +Do not use the k6 result as the maximum video-stream capacity. + +## Real WebRTC media load + +The Go/Pion client in `loadtest/pion-client` creates real media sessions. It +loops a VP8/IVF video, receives the processed track, drives step/spike/soak load, +and saves FPS, packet loss, jitter, RTT, connection state, and errors as JSON +and CSV. + +```bash +cd loadtest/pion-client +go build -o pion-load . +./pion-load -base-url http://127.0.0.1:8000 \ + -input ../videos/720p30.ivf \ + -scenario step -clients 20 -step-size 2 -step-duration 2m \ + -output results/step-real +``` + +See `loadtest/pion-client/README.md` for input conversion, all scenarios, and +result-field details. diff --git a/tests/test_api_sessions.py b/tests/test_api_sessions.py index bd3a587..bbbcf67 100644 --- a/tests/test_api_sessions.py +++ b/tests/test_api_sessions.py @@ -84,6 +84,15 @@ async def test_health_returns_ai_runtime_metrics(client): assert isinstance(body["ai_runtimes"], list) +async def test_metrics_exposes_prometheus_series(client): + response = await client.get("/metrics") + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/plain") + assert "innolive_active_sessions" in response.text + assert "innolive_frame_processing_duration_seconds" in response.text + + # --------------------------------------------------------------------------- # POST /sessions # --------------------------------------------------------------------------- diff --git a/tests/test_config_privacy.py b/tests/test_config_privacy.py new file mode 100644 index 0000000..af4909d --- /dev/null +++ b/tests/test_config_privacy.py @@ -0,0 +1,55 @@ +import pytest +from pydantic import ValidationError + +from app.core.config import Settings + + +@pytest.mark.parametrize("mode", ["bypass", "fixed_delay"]) +def test_production_rejects_non_real_privacy_mode(mode): + with pytest.raises( + ValidationError, + match="AI_PRIVACY_MODE must be real in production", + ): + Settings( + _env_file=None, + app_env="production", + ai_privacy_enabled=True, + ai_privacy_mode=mode, + ) + + +def test_production_rejects_disabled_privacy_filter(): + with pytest.raises( + ValidationError, + match="AI_PRIVACY_ENABLED must be true in production", + ): + Settings( + _env_file=None, + app_env="production", + ai_privacy_enabled=False, + ai_privacy_mode="real", + ) + + +def test_production_accepts_enabled_real_privacy_filter(): + settings = Settings( + _env_file=None, + app_env="production", + ai_privacy_enabled=True, + ai_privacy_mode="real", + ) + + assert settings.ai_privacy_enabled is True + assert settings.ai_privacy_mode == "real" + + +def test_non_production_accepts_benchmark_privacy_mode(): + settings = Settings( + _env_file=None, + app_env="test", + ai_privacy_enabled=False, + ai_privacy_mode="bypass", + ) + + assert settings.ai_privacy_enabled is False + assert settings.ai_privacy_mode == "bypass" diff --git a/tests/test_deployment_privacy.py b/tests/test_deployment_privacy.py index 497152c..adf2efa 100644 --- a/tests/test_deployment_privacy.py +++ b/tests/test_deployment_privacy.py @@ -7,7 +7,9 @@ def test_production_compose_forces_privacy_filter_enabled(): compose = (PROJECT_ROOT / "docker-compose.prod.yml").read_text() + assert 'APP_ENV: "production"' in compose assert 'AI_PRIVACY_ENABLED: "true"' in compose + assert 'AI_PRIVACY_MODE: "real"' in compose assert 'AI_PRIVACY_DEVICE: "cuda:0"' in compose assert 'AI_PRIVACY_REQUIRE_GPU: "true"' in compose assert "gpus: all" in compose diff --git a/tests/test_identity_performance.py b/tests/test_identity_performance.py index 8c212e0..c5ac646 100644 --- a/tests/test_identity_performance.py +++ b/tests/test_identity_performance.py @@ -1,4 +1,6 @@ -from unittest.mock import MagicMock +import threading +from types import SimpleNamespace +from unittest.mock import MagicMock, patch import numpy as np import torch @@ -24,6 +26,17 @@ def _detection(confidence: float = 0.9) -> FaceDetection: ) +def test_runtime_stage_observer_receives_measurements() -> None: + runtime = SharedGpuRuntime.__new__(SharedGpuRuntime) + measurements: list[tuple[str, float]] = [] + runtime._stage_observer = None + + runtime.set_stage_observer(lambda stage, seconds: measurements.append((stage, seconds))) + runtime.observe_stage("detection", 0.012) + + assert measurements == [("detection", 0.012)] + + def _engine(**overrides) -> FacePrivacyEngine: config = PrivacyBlurConfig( enable_identity_exclusion=True, @@ -188,3 +201,88 @@ def test_missing_landmarks_are_not_identity_matched() -> None: assert scores == [-1.0] runtime._run_sface.assert_not_called() + + +class _ForbiddenLock: + def __enter__(self): + raise AssertionError("live detector lock should not be used") + + def __exit__(self, *args): + return False + + +def test_external_reference_detector_does_not_take_live_detector_lock() -> None: + runtime = SharedGpuRuntime.__new__(SharedGpuRuntime) + runtime._detector_lock = _ForbiddenLock() + runtime.detector = MagicMock(name="live_detector") + runtime.device = torch.device("cpu") + runtime.half_enabled = True + reference_detector = MagicMock(name="reference_detector") + reference_detector.predict.return_value = [ + SimpleNamespace(boxes=None, keypoints=None) + ] + frame = np.zeros((32, 32, 3), dtype=np.uint8) + + detections = runtime._predict_yolo( + [frame], + config=PrivacyBlurConfig(enable_identity_exclusion=False), + detector=reference_detector, + device="cpu", + half=False, + ) + + assert detections == [[]] + runtime.detector.predict.assert_not_called() + reference_detector.predict.assert_called_once() + assert reference_detector.predict.call_args.kwargs["device"] == "cpu" + assert reference_detector.predict.call_args.kwargs["quantize"] is None + + +def test_prepare_identity_uses_isolated_cpu_detector(tmp_path) -> None: + face_model = tmp_path / "face.pt" + sface_model = tmp_path / "sface.onnx" + reference_image = tmp_path / "reference.jpg" + for path in (face_model, sface_model, reference_image): + path.write_bytes(b"model") + config = PrivacyBlurConfig( + face_model_path=face_model, + face_recognition_model_path=sface_model, + me_image_path=reference_image, + enable_identity_exclusion=True, + ) + runtime = SharedGpuRuntime.__new__(SharedGpuRuntime) + runtime._closed = False + runtime._identity_lock = threading.RLock() + runtime.sface_session = object() + runtime._sface_model_path = sface_model + runtime._create_ort_session = MagicMock(name="create_ort_session") + reference_faces = [ + ( + np.zeros((32, 32, 3), dtype=np.uint8), + np.array([1, 1, 20, 20], dtype=np.float32), + np.ones((5, 2), dtype=np.float32), + ) + ] + reference_features = torch.ones((1, 128), dtype=torch.float32) + runtime._reference_faces_from_config = MagicMock(return_value=reference_faces) + runtime._reference_features_from_faces = MagicMock( + return_value=reference_features + ) + + with patch("privacy_blur.runtime.YOLO") as yolo: + prepared = runtime.prepare_identity(config) + + yolo.assert_called_once_with(str(face_model)) + runtime._reference_faces_from_config.assert_called_once_with( + config, + detector=yolo.return_value, + device="cpu", + half=False, + ) + runtime._reference_features_from_faces.assert_called_once_with( + reference_faces, + sface_session=runtime.sface_session, + ) + runtime._create_ort_session.assert_not_called() + assert prepared.sface_session is runtime.sface_session + assert prepared.reference_features is reference_features diff --git a/tests/test_wrap_with_ai.py b/tests/test_wrap_with_ai.py index 5a4e742..6b3afca 100644 --- a/tests/test_wrap_with_ai.py +++ b/tests/test_wrap_with_ai.py @@ -67,3 +67,25 @@ def test_protected_track_has_correct_session_id(mock_track, mock_privacy_filter) result = _wrap_with_ai(mock_track, "session-abc") assert result._session_id == "session-abc" + + +@pytest.mark.parametrize("mode", ["bypass", "fixed_delay"]) +def test_benchmark_modes_return_instrumented_track( + mock_track, + settings_ai_enabled, + monkeypatch, + mode, +): + from app.core.config import get_settings + from app.services.ai.track import BenchmarkVideoTrack + from app.services.sessions.manager import _wrap_with_ai + + monkeypatch.setenv("AI_PRIVACY_MODE", mode) + monkeypatch.setenv("AI_PRIVACY_FIXED_DELAY_MS", "25") + get_settings.cache_clear() + + result = _wrap_with_ai(mock_track, "session-benchmark") + + assert isinstance(result, BenchmarkVideoTrack) + assert result._mode == mode + assert result._delay_seconds == (0.025 if mode == "fixed_delay" else 0.0)