Make sure to update stats monitor on ingress started/ended (#122)

livekit · Jun 23, 2023 · 2c186c5 · 2c186c5
1 parent 7ac35ee
commit 2c186c5
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 16 deletions.
diff --git a/pkg/config/config.go b/pkg/config/config.go
@@ -50,12 +50,6 @@ type Config struct {
 	NodeID      string `yaml:"-"`
 }
 
-type WhipConfig struct {
-	// TODO add IceLite, NAT1To1IPs
-	ICEPortRange            []uint16 `yaml:"ice_port_range"`
-	EnableLoopbackCandidate bool     `yaml:"enable_loopback_candidate"`
-}
-
 type CPUCostConfig struct {
 	RTMPCpuCost                  float64 `yaml:"rtmp_cpu_cost"`
 	WHIPCpuCost                  float64 `yaml:"whip_cpu_cost"`

diff --git a/pkg/service/process_manager.go b/pkg/service/process_manager.go
@@ -99,7 +99,7 @@ func (s *ProcessManager) launchHandler(ctx context.Context, resp *rpc.GetIngress
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 
-	s.sm.IngressStarted(resp.Info.IngressId, SessionType_HandlerProcess)
+	s.sm.IngressStarted(resp.Info, SessionType_HandlerProcess)
 	h := &process{
 		info:   resp.Info,
 		cmd:    cmd,
@@ -122,7 +122,7 @@ func (s *ProcessManager) awaitCleanup(h *process) {
 	}
 
 	h.closed.Break()
-	s.sm.IngressEnded(h.info.IngressId)
+	s.sm.IngressEnded(h.info)
 
 	s.mu.Lock()
 	defer s.mu.Unlock()

diff --git a/pkg/service/service.go b/pkg/service/service.go
@@ -181,7 +181,7 @@ func (s *Service) HandleWHIPPublishRequest(streamKey, resourceId string, ihs rpc
 			p.SetStatus(livekit.IngressState_ENDPOINT_PUBLISHING, "")
 			p.SendStateUpdate(ctx)
 
-			s.sm.IngressStarted(p.IngressId, SessionType_Service)
+			s.sm.IngressStarted(p.IngressInfo, SessionType_Service)
 		} else {
 			extraParams.MimeTypes = mimeTypes
 
@@ -202,7 +202,7 @@ func (s *Service) HandleWHIPPublishRequest(streamKey, resourceId string, ihs rpc
 			}
 
 			p.SendStateUpdate(ctx)
-			s.sm.IngressEnded(p.IngressId)
+			s.sm.IngressEnded(p.IngressInfo)
 			DeregisterIngressRpcHandlers(rpcServer, p.IngressInfo, p.ExtraParams)
 		}
 	}

diff --git a/pkg/service/session_manager.go b/pkg/service/session_manager.go
@@ -4,6 +4,7 @@ import (
 	"sync"
 
 	"github.com/livekit/ingress/pkg/stats"
+	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 )
 
@@ -28,22 +29,26 @@ func NewSessionManager(monitor *stats.Monitor) *SessionManager {
 	}
 }
 
-func (sm *SessionManager) IngressStarted(ingressID string, t SessionType) {
-	logger.Infow("ingress started", "ingressID", ingressID, "type", t)
+func (sm *SessionManager) IngressStarted(info *livekit.IngressInfo, t SessionType) {
+	logger.Infow("ingress started", "ingressID", info.IngressId, "type", t)
 
 	sm.lock.Lock()
 	defer sm.lock.Unlock()
 
-	sm.sessions[ingressID] = t
+	sm.sessions[info.IngressId] = t
+
+	sm.monitor.IngressStarted(info)
 }
 
-func (sm *SessionManager) IngressEnded(ingressID string) {
-	logger.Infow("ingress ended", "ingressID", ingressID)
+func (sm *SessionManager) IngressEnded(info *livekit.IngressInfo) {
+	logger.Infow("ingress ended", "ingressID", info.IngressId)
 
 	sm.lock.Lock()
 	defer sm.lock.Unlock()
 
-	delete(sm.sessions, ingressID)
+	delete(sm.sessions, info.IngressId)
+
+	sm.monitor.IngressEnded(info)
 }
 
 func (sm *SessionManager) IsIdle() bool {