dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 05:04:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

add config for user data recording (#3966)

Browse Source 
* add config for user data recording

* missing file

* wire

* deps
This commit is contained in:
Paul Wells
2025-09-29 14:01:39 -07:00
committed by GitHub
parent b3ee219ccb
commit 060719d17d
7 changed files with 40 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/agent/client.go
+12 -9
View File
@@ -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 {
pkg/agent/config.go
+5
View File
@@ -0,0 +1,5 @@
package agent
type Config struct {
EnableUserDataRecording bool `yaml:"enable_user_data_recording"`
}
pkg/config/config.go
+2
View File
@@ -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"`
pkg/service/wire.go
+5
View File
@@ -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
}
pkg/service/wire_gen.go
+13 -8
View File
@@ -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
}