fix: ai-least-loaded-worker/#477

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)에 있습니다.

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는 워커마다 발급되고 부분 성공의

internal/ai/pool_test.go

@@ -20,25 +20,98 @@ 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 {
 		t.Fatalf("Clients() length = %d, want 3", len(clients))
 	}
 }
 
+// 짧게 끝난 파이프라인이 순번만 넘기면 라운드로빈은 다음 세션을 이미 붐비는
+// 워커에 보낸다(#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")

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)

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) {

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()