From 060719d17dad5433f8df4ad2074b35ab2f87fc34 Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Mon, 29 Sep 2025 14:01:39 -0700 Subject: [PATCH] add config for user data recording (#3966) * add config for user data recording * missing file * wire * deps --- go.mod | 2 +- go.sum | 4 ++-- pkg/agent/client.go | 21 ++++++++++++--------- pkg/agent/config.go | 5 +++++ pkg/config/config.go | 2 ++ pkg/service/wire.go | 5 +++++ pkg/service/wire_gen.go | 21 +++++++++++++-------- 7 files changed, 40 insertions(+), 20 deletions(-) create mode 100644 pkg/agent/config.go diff --git a/go.mod b/go.mod index 83e4079be..73e26942c 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397 - github.com/livekit/protocol v1.42.1-0.20250929055041-fff85ca09992 + github.com/livekit/protocol v1.42.1-0.20250929205334-7ce7550cfaef github.com/livekit/psrpc v0.7.0 github.com/mackerelio/go-osstat v0.2.6 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index 11d3c74e4..39246b964 100644 --- a/go.sum +++ b/go.sum @@ -171,8 +171,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397 h1:Z7j2mY+bvG05UC80MpnJkitlJju8sSDWsr0Bb4dPceo= github.com/livekit/mediatransportutil v0.0.0-20250922175932-f537f0880397/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A= -github.com/livekit/protocol v1.42.1-0.20250929055041-fff85ca09992 h1:iH1dOme/hIVO9msd7/1DO54clbYmz80mYlgzcW+/oRQ= -github.com/livekit/protocol v1.42.1-0.20250929055041-fff85ca09992/go.mod h1:vhMS30QoEyH2p34vi6X1eWkC4EMV72ZGZwQb74ajY7A= +github.com/livekit/protocol v1.42.1-0.20250929205334-7ce7550cfaef h1:74nizub8BWqw393ynJ9DoQI5GXmVDkY4Lp3Ndttnllg= +github.com/livekit/protocol v1.42.1-0.20250929205334-7ce7550cfaef/go.mod h1:vhMS30QoEyH2p34vi6X1eWkC4EMV72ZGZwQb74ajY7A= github.com/livekit/psrpc v0.7.0 h1:rtfqfjYN06WJYloE/S0nmkJ/Y04x4pxLQLe8kQ4FVHU= github.com/livekit/psrpc v0.7.0/go.mod h1:AuDC5uOoEjQJEc69v4Li3t77Ocz0e0NdjQEuFfO+vfk= github.com/mackerelio/go-osstat v0.2.6 h1:gs4U8BZeS1tjrL08tt5VUliVvSWP26Ai2Ob8Lr7f2i0= diff --git a/pkg/agent/client.go b/pkg/agent/client.go index 925ca105f..4a465255a 100644 --- a/pkg/agent/client.go +++ b/pkg/agent/client.go @@ -67,6 +67,7 @@ type JobRequest struct { type agentClient struct { client rpc.AgentInternalClient + config Config mu sync.RWMutex @@ -87,7 +88,7 @@ type agentClient struct { subDone chan struct{} } -func NewAgentClient(bus psrpc.MessageBus) (Client, error) { +func NewAgentClient(bus psrpc.MessageBus, config Config) (Client, error) { client, err := rpc.NewAgentInternalClient(bus) if err != nil { return nil, err @@ -95,6 +96,7 @@ func NewAgentClient(bus psrpc.MessageBus) (Client, error) { c := &agentClient{ client: client, + config: config, workers: workerpool.New(50), subDone: make(chan struct{}), } @@ -159,14 +161,15 @@ func (c *agentClient) LaunchJob(ctx context.Context, desc *JobRequest) *serverut defer wg.Done() // The cached agent parameters do not provide the exact combination of available job type/agent name/namespace, so some of the JobRequest RPC may not trigger any worker job := &livekit.Job{ - Id: utils.NewGuid(utils.AgentJobPrefix), - DispatchId: desc.DispatchId, - Type: desc.JobType, - Room: desc.Room, - Participant: desc.Participant, - Namespace: curNs, - AgentName: desc.AgentName, - Metadata: desc.Metadata, + Id: utils.NewGuid(utils.AgentJobPrefix), + DispatchId: desc.DispatchId, + Type: desc.JobType, + Room: desc.Room, + Participant: desc.Participant, + Namespace: curNs, + AgentName: desc.AgentName, + Metadata: desc.Metadata, + EnableRecording: c.config.EnableUserDataRecording, } resp, err := c.client.JobRequest(context.Background(), topic, jobTypeTopic, job) if err != nil { diff --git a/pkg/agent/config.go b/pkg/agent/config.go new file mode 100644 index 000000000..c4d0fa84c --- /dev/null +++ b/pkg/agent/config.go @@ -0,0 +1,5 @@ +package agent + +type Config struct { + EnableUserDataRecording bool `yaml:"enable_user_data_recording"` +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 0faf481f1..83b399a3b 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -26,6 +26,7 @@ import ( "github.com/urfave/cli/v3" "gopkg.in/yaml.v3" + "github.com/livekit/livekit-server/pkg/agent" "github.com/livekit/livekit-server/pkg/metric" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/bwe/remotebwe" @@ -75,6 +76,7 @@ type Config struct { LogLevel string `yaml:"log_level,omitempty"` Logging LoggingConfig `yaml:"logging,omitempty"` Limit LimitConfig `yaml:"limit,omitempty"` + Agents agent.Config `yaml:"agents,omitempty"` Development bool `yaml:"development,omitempty"` diff --git a/pkg/service/wire.go b/pkg/service/wire.go index c079ef126..48bd84dbc 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -80,6 +80,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live NewWHIPService, NewAgentService, NewAgentDispatchService, + getAgentConfig, agent.NewAgentClient, getAgentStore, getSignalRelayConfig, @@ -269,3 +270,7 @@ func newInProcessTurnServer(conf *config.Config, authHandler turn.AuthHandler) ( func getNodeStatsConfig(config *config.Config) config.NodeStatsConfig { return config.NodeStats } + +func getAgentConfig(config *config.Config) agent.Config { + return config.Agents +} diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 3b8913f0c..22b1e0c1c 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -89,23 +89,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService, objectStore) topicFormatter := rpc.NewTopicFormatter() - roomClient, err := rpc.NewTypedRoomClient(clientParams) + v, err := rpc.NewTypedRoomClient(clientParams) if err != nil { return nil, err } - participantClient, err := rpc.NewTypedParticipantClient(clientParams) + v2, err := rpc.NewTypedParticipantClient(clientParams) if err != nil { return nil, err } - roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient) + roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2) if err != nil { return nil, err } - agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) + v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) if err != nil { return nil, err } - agentDispatchService := NewAgentDispatchService(agentDispatchInternalClient, topicFormatter, roomAllocator, router) + agentDispatchService := NewAgentDispatchService(v3, topicFormatter, roomAllocator, router) egressService := NewEgressService(egressClient, rtcEgressLauncher, ioInfoService, roomService) ingressConfig := getIngressConfig(conf) ingressClient, err := rpc.NewIngressClient(clientParams) @@ -120,11 +120,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService) rtcService := NewRTCService(conf, roomAllocator, router, telemetryService) - whipParticipantClient, err := rpc.NewTypedWHIPParticipantClient(clientParams) + v4, err := rpc.NewTypedWHIPParticipantClient(clientParams) if err != nil { return nil, err } - serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, whipParticipantClient) + serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, v4) if err != nil { return nil, err } @@ -132,7 +132,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } - client, err := agent.NewAgentClient(messageBus) + agentConfig := getAgentConfig(conf) + client, err := agent.NewAgentClient(messageBus, agentConfig) if err != nil { return nil, err } @@ -335,3 +336,7 @@ func newInProcessTurnServer(conf *config.Config, authHandler turn.AuthHandler) ( func getNodeStatsConfig(config2 *config.Config) config.NodeStatsConfig { return config2.NodeStats } + +func getAgentConfig(config2 *config.Config) agent.Config { + return config2.Agents +}