From b47da082e62e510ef74f406967750b4a4defbf8b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=97=B0=ED=98=B8?= Date: Fri, 16 Oct 2026 20:20:21 +0900 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20AI=20=EC=9B=8C=EC=BB=A4=EB=B3=84=20?= =?UTF-8?q?=ED=99=9C=EC=84=B1=20=EC=8A=A4=ED=8A=B8=EB=A6=BC=20=EA=B2=8C?= =?UTF-8?q?=EC=9D=B4=EC=A7=80=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- internal/metrics/registry.go | 11 +++++++++++ internal/metrics/registry_test.go | 22 ++++++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/internal/metrics/registry.go b/internal/metrics/registry.go index 91e4156..757adb1 100644 --- a/internal/metrics/registry.go +++ b/internal/metrics/registry.go @@ -57,6 +57,7 @@ type Registry struct { aiTargetSessions map[string]uint64 egressUnitsByPlan map[string]int64 aiTargetReady map[string]int64 + aiTargetActive map[string]int64 nvencSessions map[string]int64 nvencSlotsPerCard map[string]int64 framesReceived map[string]uint64 @@ -87,6 +88,7 @@ func New() *Registry { aiInputPausedFrames: make(map[string]uint64), aiTargetSessions: make(map[string]uint64), aiTargetReady: make(map[string]int64), + aiTargetActive: make(map[string]int64), nvencSessions: make(map[string]int64), nvencSlotsPerCard: make(map[string]int64), framesReceived: make(map[string]uint64), @@ -209,6 +211,14 @@ func (r *Registry) IncAITargetSession(target string) { r.increment(r.aiTargetSessions, target) } +// SetAITargetActiveStreams는 한 AI worker에 지금 배정된 영상 파이프라인 수다. +// 워커 하나가 프레임을 차례로 처리하므로 이 값의 편중이 곧 지연 편중이다(#477). +func (r *Registry) SetAITargetActiveStreams(target string, active int) { + r.mu.Lock() + r.aiTargetActive[target] = int64(active) + r.mu.Unlock() +} + // SetAITargetReady는 한 AI worker가 마지막 preflight 프로브에서 실제로 추론을 // 돌려줬는지를 기록한다. 프로세스가 살아 있어도 런타임이 죽어 있는 상태를 // 드러내려는 것이므로(#162), 값의 근거는 언제나 왕복 프로브여야 한다. @@ -383,6 +393,7 @@ func (r *Registry) WritePrometheus(w io.Writer) { writeLabeledCountersWithKey(w, "innolive_ai_target_sessions_total", "Number of sessions assigned to each AI worker target.", "target", r.aiTargetSessions) writeLabeledGaugesWithKey(w, "innolive_nvenc_sessions", "NVENC encoder sessions reported by nvidia-smi for each GPU.", "gpu", r.nvencSessions) writeLabeledGaugesWithKey(w, "innolive_nvenc_slots_accounted", "Egress slots the server accounts for on each GPU.", "gpu", r.nvencSlotsPerCard) + writeLabeledGaugesWithKey(w, "innolive_ai_target_active_streams", "Video pipelines currently assigned to each AI worker target.", "target", r.aiTargetActive) writeLabeledGaugesWithKey(w, "innolive_ai_target_ready", "Whether the AI worker target answered the most recent preflight probe (1) or not (0).", "target", r.aiTargetReady) writeLabeledCounters(w, "innolive_frame_received_total", "Number of complete video frames received for processing.", r.framesReceived) writeLabeledCounters(w, "innolive_frame_processed_total", "Number of successfully processed video frames.", r.framesProcessed) diff --git a/internal/metrics/registry_test.go b/internal/metrics/registry_test.go index 726e726..763a984 100644 --- a/internal/metrics/registry_test.go +++ b/internal/metrics/registry_test.go @@ -70,6 +70,28 @@ func TestProcessTreeKeepsCompletedChildCPU(t *testing.T) { } } +// TestAITargetActiveStreamsGauge: 워커별 활성 파이프라인 수가 오르내린 그대로 +// 보여야 편중을 운영 메트릭에서 바로 읽을 수 있다(#477). +func TestAITargetActiveStreamsGauge(t *testing.T) { + r := New() + r.SetAITargetActiveStreams("127.0.0.1:50051", 2) + r.SetAITargetActiveStreams("127.0.0.1:50052", 1) + r.SetAITargetActiveStreams("127.0.0.1:50051", 1) + + var output bytes.Buffer + r.WritePrometheus(&output) + text := output.String() + for _, expected := range []string{ + "# TYPE innolive_ai_target_active_streams gauge", + `innolive_ai_target_active_streams{target="127.0.0.1:50051"} 1`, + `innolive_ai_target_active_streams{target="127.0.0.1:50052"} 1`, + } { + if !strings.Contains(text, expected) { + t.Errorf("metrics output missing %q", expected) + } + } +} + // TestAITargetReadyGauge: 게이지는 target별로 갈라져야 하고, not-ready로 떨어진 // 뒤 다시 ready가 되면 0에 머무르지 않고 올라와야 한다(#162). func TestAITargetReadyGauge(t *testing.T) { From ed89913768be0bc0f684463af74315995a52b279 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=97=B0=ED=98=B8?= Date: Fri, 16 Oct 2026 20:20:21 +0900 Subject: [PATCH 2/3] =?UTF-8?q?fix:=20AI=20=EC=9B=8C=EC=BB=A4=20=EB=B0=B0?= =?UTF-8?q?=EC=A0=95=EC=9D=84=20=ED=99=9C=EC=84=B1=20=EC=8A=A4=ED=8A=B8?= =?UTF-8?q?=EB=A6=BC=20=EC=B5=9C=EC=86=8C=20=EA=B8=B0=EC=A4=80=EC=9C=BC?= =?UTF-8?q?=EB=A1=9C=20=EB=B3=80=EA=B2=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- internal/ai/pool.go | 77 ++++++++++++++++++++++++++++------ internal/ai/pool_test.go | 83 ++++++++++++++++++++++++++++++++++--- internal/session/manager.go | 11 ++++- 3 files changed, 153 insertions(+), 18 deletions(-) diff --git a/internal/ai/pool.go b/internal/ai/pool.go index d568284..1b82870 100644 --- a/internal/ai/pool.go +++ b/internal/ai/pool.go @@ -5,7 +5,7 @@ import ( "errors" "fmt" "log/slog" - "sync/atomic" + "sync" "time" aiv1 "inno-live-server/api/gen/aiv1" @@ -14,13 +14,34 @@ import ( "google.golang.org/grpc/status" ) -// Pool은 AI 워커 대상마다 클라이언트를 하나씩 두고 세션을 라운드로빈으로 -// 배정한다. MPS(또는 소프트웨어 자원 상한)로 워커 프로세스마다 GPU 몫이 정해져 -// 있으므로, 세션을 워커에 나눠야 그 격리가 스트림별 격리가 된다. +// Pool은 AI 워커 대상마다 클라이언트를 하나씩 두고 세션을 활성 스트림이 가장 +// 적은 대상에 배정한다. MPS(또는 소프트웨어 자원 상한)로 워커 프로세스마다 GPU +// 몫이 정해져 있으므로, 세션을 워커에 나눠야 그 격리가 스트림별 격리가 된다. type Pool struct { clients []*Client - next atomic.Uint64 logger *slog.Logger + + mu sync.Mutex + // active는 대상별로 지금 열려 있는 영상 파이프라인 수다. + active []int + // next는 활성 수가 같은 대상 사이에서 시작 위치를 돌려 고르게 퍼뜨린다. + next int + // observer는 대상별 활성 수가 바뀔 때마다 불린다(메트릭 게이지). + observer func(target string, active int) +} + +// SetStreamObserver는 대상별 활성 스트림 수가 바뀔 때마다 부를 함수를 붙인다. +// 지금 값을 한 번씩 먼저 알려 게이지가 0부터 보이게 한다. +func (p *Pool) SetStreamObserver(observer func(target string, active int)) { + p.mu.Lock() + defer p.mu.Unlock() + p.observer = observer + if observer == nil { + return + } + for index, client := range p.clients { + observer(client.Address(), p.active[index]) + } } // SetLogger는 로거를 붙인다. 화이트리스트 롤백처럼 최선 노력 경로가 재시도를 다 @@ -79,13 +100,45 @@ func NewPool(targets []string, timeout time.Duration) (*Pool, error) { } clients = append(clients, client) } - return &Pool{clients: clients}, nil + return &Pool{clients: clients, active: make([]int, len(clients))}, nil +} + +// Acquire는 활성 스트림이 가장 적은 대상을 골라 스트림 하나를 배정하고, 그 +// 배정을 되돌리는 release를 함께 돌려준다. 호출자는 파이프라인이 끝날 때 release를 +// 한 번 불러야 한다(여러 번 불러도 한 번만 반영된다). +// +// 워커 한 프로세스는 프레임을 차례로 처리하므로 같은 워커에 몰린 스트림 수가 곧 +// 프레임 지연이다(#477). 종전 라운드로빈은 끝난 세션·파이프라인 재시작에도 +// 카운터가 넘어가, 동시 2세션이 한 워커에 몰리고 다른 워커가 놀 수 있었다. +func (p *Pool) Acquire() (*Client, func()) { + p.mu.Lock() + defer p.mu.Unlock() + best := p.next % len(p.clients) + for offset := 1; offset < len(p.clients); offset++ { + index := (p.next + offset) % len(p.clients) + if p.active[index] < p.active[best] { + best = index + } + } + p.next = (best + 1) % len(p.clients) + p.adjust(best, 1) + var once sync.Once + release := func() { + once.Do(func() { + p.mu.Lock() + defer p.mu.Unlock() + p.adjust(best, -1) + }) + } + return p.clients[best], release } -// Next는 다음 세션에 쓸 클라이언트를 대상들을 돌아가며 돌려준다. -func (p *Pool) Next() *Client { - index := (p.next.Add(1) - 1) % uint64(len(p.clients)) - return p.clients[index] +// adjust는 p.mu를 쥔 채 불린다. +func (p *Pool) adjust(index, delta int) { + p.active[index] += delta + if p.observer != nil { + p.observer(p.clients[index].Address(), p.active[index]) + } } func (p *Pool) Clients() []*Client { @@ -118,8 +171,8 @@ type WhitelistResult struct { // 요청 컨텍스트는 이미 취소된 경우가 많아, 롤백은 분리해 돌리고 자기 기한을 둔다. const partialAddRollbackTimeout = 5 * time.Second -// AddWhitelist는 클라이언트의 기준 얼굴을 모든 AI 워커에 등록한다(세션이 대상에 -// 라운드로빈으로 퍼지므로 화이트리스트가 전부에 있어야 한다). +// AddWhitelist는 클라이언트의 기준 얼굴을 모든 AI 워커에 등록한다(세션이 어느 +// 대상에 배정될지 미리 알 수 없으므로 화이트리스트가 전부에 있어야 한다). // // 오류가 나면 얼굴은 어디에도 등록되지 않은 상태다. 받아들인 워커는 여기서 // 되돌린다. 호출자는 이를 할 수 없다 — 엔트리 id는 워커마다 발급되고 부분 성공의 diff --git a/internal/ai/pool_test.go b/internal/ai/pool_test.go index 0ff3bc5..5cb6d0a 100644 --- a/internal/ai/pool_test.go +++ b/internal/ai/pool_test.go @@ -20,18 +20,24 @@ import ( "google.golang.org/grpc/test/bufconn" ) -func TestPoolNextCyclesRoundRobin(t *testing.T) { +func newTestPool(t *testing.T, targets ...string) *Pool { + t.Helper() // grpc.NewClient는 지연 연결이라 닿지 않는 대상도 여기서는 괜찮다. - pool, err := NewPool([]string{"a:1", "b:1", "c:1"}, time.Second) + pool, err := NewPool(targets, time.Second) if err != nil { t.Fatal(err) } - defer pool.Close() + t.Cleanup(func() { _ = pool.Close() }) + return pool +} + +func TestPoolAcquireSpreadsEqualLoadInTurn(t *testing.T) { + pool := newTestPool(t, "a:1", "b:1", "c:1") want := []string{"a:1", "b:1", "c:1", "a:1", "b:1", "c:1"} for index, expected := range want { - if got := pool.Next().Address(); got != expected { - t.Fatalf("Next() #%d = %q, want %q", index, got, expected) + if got, _ := pool.Acquire(); got.Address() != expected { + t.Fatalf("Acquire() #%d = %q, want %q", index, got.Address(), expected) } } if clients := pool.Clients(); len(clients) != 3 { @@ -39,6 +45,73 @@ func TestPoolNextCyclesRoundRobin(t *testing.T) { } } +// 짧게 끝난 파이프라인이 순번만 넘기면 라운드로빈은 다음 세션을 이미 붐비는 +// 워커에 보낸다(#477). 활성 수로 고르면 비어 있는 워커로 간다. +func TestPoolAcquirePrefersTheLeastActiveTarget(t *testing.T) { + pool := newTestPool(t, "a:1", "b:1") + + first, _ := pool.Acquire() + _, releaseShort := pool.Acquire() + releaseShort() + second, _ := pool.Acquire() + + if first.Address() != "a:1" || second.Address() != "b:1" { + t.Fatalf("Acquire() = %q then %q, want a:1 then b:1", first.Address(), second.Address()) + } +} + +func TestPoolReleaseIsIdempotentAndReported(t *testing.T) { + pool := newTestPool(t, "a:1", "b:1") + var mu sync.Mutex + observed := map[string]int{} + pool.SetStreamObserver(func(target string, active int) { + mu.Lock() + observed[target] = active + mu.Unlock() + }) + if observed["a:1"] != 0 || observed["b:1"] != 0 || len(observed) != 2 { + t.Fatalf("initial observation = %v, want both targets at 0", observed) + } + + _, releaseA := pool.Acquire() + _, releaseB := pool.Acquire() + _, releaseA2 := pool.Acquire() + if observed["a:1"] != 2 || observed["b:1"] != 1 { + t.Fatalf("after three acquires = %v, want a:1=2 b:1=1", observed) + } + releaseA() + releaseA() + releaseB() + if observed["a:1"] != 1 || observed["b:1"] != 0 { + t.Fatalf("after releases = %v, want a:1=1 b:1=0", observed) + } + releaseA2() + if observed["a:1"] != 0 { + t.Fatalf("a:1 = %d after every release, want 0", observed["a:1"]) + } +} + +func TestPoolAcquireIsSafeUnderConcurrency(t *testing.T) { + pool := newTestPool(t, "a:1", "b:1") + var wait sync.WaitGroup + for range 64 { + wait.Add(1) + go func() { + defer wait.Done() + _, release := pool.Acquire() + release() + }() + } + wait.Wait() + pool.mu.Lock() + defer pool.mu.Unlock() + for index, active := range pool.active { + if active != 0 { + t.Fatalf("active[%d] = %d after every release, want 0", index, active) + } + } +} + func TestNewPoolRejectsEmptyTargetList(t *testing.T) { if _, err := NewPool(nil, time.Second); err == nil { t.Fatal("NewPool(nil) should fail") diff --git a/internal/session/manager.go b/internal/session/manager.go index d1e9811..3170b8c 100644 --- a/internal/session/manager.go +++ b/internal/session/manager.go @@ -503,6 +503,9 @@ func NewManager(cfg config.Config, logger *slog.Logger, registry *metrics.Regist if cfg.PrivacyMode == config.PrivacyModeReal && aiPool == nil { return nil, errors.New("real privacy mode requires an AI client pool") } + if aiPool != nil && registry != nil { + aiPool.SetStreamObserver(registry.SetAITargetActiveStreams) + } var settingEngine webrtc.SettingEngine if cfg.UDPMuxPort > 0 { // 모든 ICE UDP 트래픽을 단일 포트에 multiplex한다. 기본값처럼 미디어를 @@ -2047,8 +2050,12 @@ func (m *Manager) startVideoPipeline(ctx context.Context, s *Session, track *web s.mu.Unlock() var aiStream media.AIStream + // releaseAI는 이 파이프라인의 AI 워커 배정을 돌려준다. 파이프라인이 끝나거나 + // 시작에 실패하면 반드시 불린다 — 그래야 다음 세션이 덜 붐비는 워커로 간다. + releaseAI := func() {} if s.privacyMode == config.PrivacyModeReal { - client := m.ai.Next() + client, release := m.ai.Acquire() + releaseAI = release m.metrics.IncAITargetSession(client.Address()) // AI 서버는 비어 있지 않은 세션 scope를 요구한다. 서버가 계산한 AI // client ID를 사용해 요청 metadata로 사용자별 reference-face bucket과 @@ -2066,6 +2073,7 @@ func (m *Manager) startVideoPipeline(ctx context.Context, s *Session, track *web processor, err := media.NewProcessor(s.privacyMode, m.cfg.PrivacyFixedDelay, aiStream, m.metrics, m.logger.With("session_id", s.ID), m.cfg.AIWireFormat, m.cfg.AIFailurePolicy, m.cfg.AITimeoutLatchThreshold) if err != nil { m.logger.Error("create video processor failed", "session_id", s.ID, "error", err) + releaseAI() close(pipelineDone) return } @@ -2099,6 +2107,7 @@ func (m *Manager) startVideoPipeline(ctx context.Context, s *Session, track *web } go func() { defer close(pipelineDone) + defer releaseAI() m.mu.Lock() m.pipelines[s.ID] = struct{}{} m.mu.Unlock() From 9d99b350d79c5e3a00ec7c65f452159c4efe6617 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=97=B0=ED=98=B8?= Date: Fri, 16 Oct 2026 20:20:22 +0900 Subject: [PATCH 3/3] =?UTF-8?q?docs:=20AI=20=EC=9B=8C=EC=BB=A4=20=EB=B0=B0?= =?UTF-8?q?=EC=A0=95=20=EB=B0=A9=EC=8B=9D=EA=B3=BC=20=ED=99=9C=EC=84=B1=20?= =?UTF-8?q?=EC=8A=A4=ED=8A=B8=EB=A6=BC=20=EB=A9=94=ED=8A=B8=EB=A6=AD=20?= =?UTF-8?q?=EC=84=A4=EB=AA=85=20=EA=B0=B1=EC=8B=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- README.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 82fe4bb..f29e75d 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ ## 주요 기능 - **WebRTC 인제스트/전달**: [pion/webrtc](https://github.com/pion/webrtc) 기반. STUN/TURN, ICE 재연결 유예, 네트워크 전환 시 ICE restart 복구를 지원합니다. -- **AI 프라이버시 파이프라인**: 프레임을 AI 서버로 보내 얼굴을 블러 처리합니다. `bypass`, `fixed_delay`, `real` 세 모드가 있고, 워커 여러 대에 라운드로빈으로 분배합니다. +- **AI 프라이버시 파이프라인**: 프레임을 AI 서버로 보내 얼굴을 블러 처리합니다. `bypass`, `fixed_delay`, `real` 세 모드가 있고, 워커 여러 대 중 활성 스트림이 가장 적은 워커에 분배합니다. - **fail-closed 실패 처리**: AI 처리가 지연되거나 실패하면 미처리 원본 대신 검은 화면을 내보냅니다. - **AI 워커 상시 감시**: 주기마다 실제 프레임을 왕복시키는 프로브를 돌려, 프로세스는 살아 있는데 추론만 죽은 워커를 잡아냅니다. - **플랫폼 송출**: 사용자가 연결한 YouTube 계정으로 방송을 만들고 FFmpeg로 RTMP 송출합니다. 방송 준비와 라이브 전환을 나눠서, 시청자에게 보이기 전에 화면을 확인할 수 있습니다. @@ -428,7 +428,7 @@ Apple·YouTube·AI worker·파일 저장소 또는 데이터베이스 단계에 `GET /metrics`가 `innolive_` 접두사의 Prometheus 메트릭을 노출합니다. 파이프라인을 진단할 때 먼저 보는 메트릭입니다. - **세션**: `innolive_active_sessions`, `innolive_sessions_rejected_total`, `innolive_connection_failures_total` -- **AI 처리**: `innolive_ai_duration_seconds`, `innolive_ai_fallback_frames_total`, `innolive_ai_fallback_latched_total`, `innolive_ai_target_ready` +- **AI 처리**: `innolive_ai_duration_seconds`, `innolive_ai_fallback_frames_total`, `innolive_ai_fallback_latched_total`, `innolive_ai_target_ready`, `innolive_ai_target_active_streams` - **프레임 흐름**: `innolive_frame_received_total`, `innolive_frame_processed_total`, `innolive_frame_dropped_total`, `innolive_processing_queue_size` - **송출**: `innolive_egress_frames_dropped_total`, `innolive_egress_reconnect_total`, `innolive_audio_samples_dropped_total` - **WebRTC 복구**: `innolive_peer_recovery_started_total`, `innolive_peer_recovery_succeeded_total`, `innolive_peer_recovery_exhausted_total` @@ -437,6 +437,8 @@ Apple·YouTube·AI worker·파일 저장소 또는 데이터베이스 단계에 `innolive_ai_target_ready`는 상시 감시 프로브가 워커마다 갱신하는 게이지입니다. 프로세스가 살아 있어도 추론이 죽은 워커는 여기서 `0`이 됩니다. +`innolive_ai_target_active_streams`는 워커마다 지금 배정된 영상 파이프라인 수입니다. 워커 하나가 프레임을 차례로 처리하므로 이 값이 한 워커에 몰리면 그 워커의 세션 지연이 늘어납니다. + ## 배포 `main` 브랜치에 머지하면 GitHub Actions가 이미지를 빌드해 레지스트리에 올리고, 프로덕션 서버가 그 이미지를 받아 컨테이너를 재기동합니다. 배포 경로는 이 하나뿐입니다. 서버에 들어가 소스나 바이너리를 고치지 마세요. 자세한 절차와 비상 경로는 [`deploy/README.md`](deploy/README.md)에 있습니다.