diff --git a/cmd/cli/main.go b/cmd/cli/main.go index 8720e8725..b0d640d94 100644 --- a/cmd/cli/main.go +++ b/cmd/cli/main.go @@ -20,7 +20,7 @@ func main() { app.Commands = append(app.Commands, commands.RTCCommands...) app.Commands = append(app.Commands, commands.TokenCommands...) - logger.InitDevelopment() + logger.InitDevelopment("") if err := app.Run(os.Args); err != nil { fmt.Println(err) } diff --git a/cmd/server/main.go b/cmd/server/main.go index e53548ad3..5a3b1b064 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -1,7 +1,6 @@ package main import ( - "bytes" "context" "errors" "fmt" @@ -39,14 +38,13 @@ func main() { EnvVars: []string{"LIVEKIT_CONFIG"}, }, &cli.StringFlag{ - Name: "key-file", - Usage: "path to file that contains API keys/secrets", - EnvVars: []string{"KEY_FILE"}, + Name: "key-file", + Usage: "path to file that contains API keys/secrets", }, &cli.StringFlag{ Name: "keys", Usage: "API keys/secret pairs (key:secret, one per line)", - EnvVars: []string{"KEY_FILE"}, + EnvVars: []string{"LIVEKIT_KEYS"}, }, &cli.StringFlag{ Name: "cpuprofile", @@ -58,7 +56,7 @@ func main() { }, &cli.BoolFlag{ Name: "dev", - Usage: "when set, token validation will be disabled", + Usage: "sets log-level to debug, and console formatter", }, }, Action: startServer, @@ -92,13 +90,14 @@ func startServer(c *cli.Context) error { return err } - conf.UpdateFromCLI(c) - var keyProvider auth.KeyProvider + if err = conf.UpdateFromCLI(c); err != nil { + return err + } if conf.Development { - logger.InitDevelopment() + logger.InitDevelopment(conf.LogLevel) } else { - logger.InitProduction() + logger.InitProduction(conf.LogLevel) } if cpuProfile != "" { @@ -127,10 +126,11 @@ func startServer(c *cli.Context) error { } // require a key provider - if keyProvider, err = createKeyProvider(c.String("key-file"), c.String("keys")); err != nil { + keyProvider, err := createKeyProvider(conf) + if err != nil { return err } - logger.Infow("auth enabled", "num_keys", keyProvider.NumKeys()) + logger.Infow("configured key provider", "num_keys", keyProvider.NumKeys()) currentNode, err := routing.NewLocalNode(conf) if err != nil { @@ -181,25 +181,24 @@ func createRouterAndStore(config *config.Config, node routing.LocalNode) (router return } -func createKeyProvider(keyFile, keys string) (auth.KeyProvider, error) { +func createKeyProvider(conf *config.Config) (auth.KeyProvider, error) { // prefer keyfile if set - if keyFile != "" { - if st, err := os.Stat(keyFile); err != nil { + if conf.KeyFile != "" { + if st, err := os.Stat(conf.KeyFile); err != nil { return nil, err } else if st.Mode().Perm() != 0600 { return nil, fmt.Errorf("key file must have permission set to 600") } - f, err := os.Open(keyFile) + f, err := os.Open(conf.KeyFile) if err != nil { return nil, err } defer f.Close() - return auth.NewFileBasedKeyProvider(f) + return auth.NewFileBasedKeyProviderFromReader(f) } - if keys != "" { - r := bytes.NewReader([]byte(keys)) - return auth.NewFileBasedKeyProvider(r) + if conf.Keys != nil { + return auth.NewFileBasedKeyProviderFromMap(conf.Keys), nil } return nil, errors.New("one of key-file or keys must be provided in order to support a secure installation") diff --git a/go.mod b/go.mod index b5dab816c..83f6c397a 100644 --- a/go.mod +++ b/go.mod @@ -16,6 +16,7 @@ require ( github.com/magefile/mage v1.10.0 github.com/manifoldco/promptui v0.8.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 + github.com/mitchellh/go-homedir v1.1.0 github.com/pion/ion-log v1.0.0 github.com/pion/ion-sfu v1.7.7 github.com/pion/rtcp v1.2.6 diff --git a/go.sum b/go.sum index 8cd728c83..47db9d471 100644 --- a/go.sum +++ b/go.sum @@ -265,6 +265,8 @@ github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0/go.mod h1:fcEyUyXZXoV4Abw8DX0t7w github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc= github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= +github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y= +github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI= github.com/mitchellh/gox v0.4.0/go.mod h1:Sd9lOJ0+aimLBi73mGofS1ycjY8lL3uZM3JPS42BGNg= github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0QubkSMEySY= diff --git a/pkg/auth/provider.go b/pkg/auth/provider.go index 76a06134b..9feb5ca23 100644 --- a/pkg/auth/provider.go +++ b/pkg/auth/provider.go @@ -10,7 +10,7 @@ type FileBasedKeyProvider struct { keys map[string]string } -func NewFileBasedKeyProvider(r io.Reader) (p *FileBasedKeyProvider, err error) { +func NewFileBasedKeyProviderFromReader(r io.Reader) (p *FileBasedKeyProvider, err error) { keys := make(map[string]string) decoder := yaml.NewDecoder(r) if err = decoder.Decode(&keys); err != nil { @@ -23,6 +23,12 @@ func NewFileBasedKeyProvider(r io.Reader) (p *FileBasedKeyProvider, err error) { return } +func NewFileBasedKeyProviderFromMap(keys map[string]string) *FileBasedKeyProvider { + return &FileBasedKeyProvider{ + keys: keys, + } +} + func (p *FileBasedKeyProvider) GetSecret(key string) string { return p.keys[key] } diff --git a/pkg/auth/provider_test.go b/pkg/auth/provider_test.go index 5013fae1e..d0aad6468 100644 --- a/pkg/auth/provider_test.go +++ b/pkg/auth/provider_test.go @@ -31,7 +31,7 @@ func TestFileBasedKeyProvider(t *testing.T) { r, err := os.Open(f.Name()) defer r.Close() - p, err := auth.NewFileBasedKeyProvider(r) + p, err := auth.NewFileBasedKeyProviderFromReader(r) assert.NoError(t, err) for key, val := range keys { diff --git a/pkg/config/config.go b/pkg/config/config.go index 87bdf4ee7..675c18ebb 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -1,14 +1,20 @@ package config import ( + "os" + + "github.com/mitchellh/go-homedir" "github.com/urfave/cli/v2" "gopkg.in/yaml.v3" ) type Config struct { - Port uint32 `yaml:"port"` - RTC RTCConfig `yaml:"rtc"` - Redis RedisConfig `yaml:"redis"` + Port uint32 `yaml:"port"` + RTC RTCConfig `yaml:"rtc"` + Redis RedisConfig `yaml:"redis"` + KeyFile string `yaml:"key_file"` + Keys map[string]string `yaml:"keys"` + LogLevel string `yaml:"log_level"` // multi-node configuration, MultiNode bool `yaml:"multi_node"` @@ -51,8 +57,25 @@ func NewConfig(confString string) (*Config, error) { return conf, nil } -func (conf *Config) UpdateFromCLI(c *cli.Context) { +func (conf *Config) UpdateFromCLI(c *cli.Context) error { if c.IsSet("dev") { conf.Development = c.Bool("dev") } + if c.IsSet("key-file") { + conf.KeyFile = c.String("file") + } + if c.IsSet("keys") { + conf.Keys = make(map[string]string) + if err := yaml.Unmarshal([]byte(c.String("keys")), conf.Keys); err != nil { + return err + } + } + + // expand env vars in filenames + file, err := homedir.Expand(os.ExpandEnv(conf.KeyFile)) + if err != nil { + return err + } + conf.KeyFile = file + return nil } diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index 67f085a7a..7aed981de 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -2,30 +2,38 @@ package logger import ( "go.uber.org/zap" + "go.uber.org/zap/zapcore" ) var ( - logger *zap.SugaredLogger - zapOptions = []zap.Option{ - zap.AddCallerSkip(1), - } + logger *zap.SugaredLogger ) func getLogger() *zap.SugaredLogger { if logger == nil { - InitDevelopment() + InitDevelopment("") } return logger } -func InitProduction() { - l, _ := zap.NewProduction(zapOptions...) +func initLogger(config zap.Config, level string) { + if level != "" { + lvl := zapcore.Level(0) + if err := lvl.UnmarshalText([]byte(level)); err == nil { + config.Level = zap.NewAtomicLevelAt(lvl) + } + } + l, _ := config.Build(zap.AddCallerSkip(1)) logger = l.Sugar() } -func InitDevelopment() { - l, _ := zap.NewDevelopment(zapOptions...) - logger = l.Sugar() +func InitProduction(logLevel string) { + initLogger(zap.NewProductionConfig(), logLevel) + +} + +func InitDevelopment(logLevel string) { + initLogger(zap.NewDevelopmentConfig(), logLevel) } func Debugw(msg string, keysAndValues ...interface{}) { diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index 882995d48..835b7db86 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -25,12 +25,15 @@ type ParticipantCallback func(roomId, participantId, participantName string, req //counterfeiter:generate . Router type Router interface { - GetNodeIdForRoom(roomName string) (string, error) + GetNodeForRoom(roomName string) (string, error) + SetNodeForRoom(roomName string, nodeId string) error RegisterNode() error UnregisterNode() error GetNode(nodeId string) (*livekit.Node, error) + ListNodes() ([]*livekit.Node, error) SetParticipantRTCNode(participantId, nodeId string) error + // functions for websocket handler GetRequestSink(participantId string) (MessageSink, error) GetResponseSource(participantId string) (MessageSource, error) diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 72872736c..eb7654f45 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -25,10 +25,14 @@ func NewLocalRouter(currentNode LocalNode) *LocalRouter { } } -func (r *LocalRouter) GetNodeIdForRoom(roomName string) (string, error) { +func (r *LocalRouter) GetNodeForRoom(roomName string) (string, error) { return r.currentNode.Id, nil } +func (r *LocalRouter) SetNodeForRoom(roomName string, nodeId string) error { + return nil +} + func (r *LocalRouter) RegisterNode() error { return nil } @@ -44,6 +48,12 @@ func (r *LocalRouter) GetNode(nodeId string) (*livekit.Node, error) { return nil, ErrNotFound } +func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) { + return []*livekit.Node{ + r.currentNode, + }, nil +} + func (r *LocalRouter) StartParticipant(roomName, participantId, participantName string) error { // treat it as a new participant connecting if r.onNewParticipant == nil { diff --git a/pkg/routing/redis.go b/pkg/routing/redis.go index e8441975a..3d08bfc03 100644 --- a/pkg/routing/redis.go +++ b/pkg/routing/redis.go @@ -68,10 +68,17 @@ type RedisSink struct { rc *redis.Client nodeId string participantId string - channel string onClose func() } +func NewRedisSink(rc *redis.Client, nodeId, participantId string) *RedisSink { + return &RedisSink{ + rc: rc, + nodeId: nodeId, + participantId: participantId, + } +} + func (s *RedisSink) WriteMessage(msg proto.Message) error { return publishRouterMessage(s.rc, s.nodeId, s.participantId, msg) } diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 36dc0b073..769dff763 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -5,6 +5,7 @@ import ( "time" "github.com/go-redis/redis/v8" + "github.com/pkg/errors" "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/logger" @@ -34,6 +35,7 @@ func NewRedisRouter(currentNode LocalNode, rc *redis.Client, useLocal bool) *Red LocalRouter: *NewLocalRouter(currentNode), useLocal: useLocal, rc: rc, + redisSinks: make(map[string]*RedisSink), } rr.ctx, rr.cancel = context.WithCancel(context.Background()) rr.cr = utils.NewCachedRedis(rr.ctx, rr.rc) @@ -46,7 +48,10 @@ func (r *RedisRouter) RegisterNode() error { return err } r.cr.ExpireHash(NodesKey, r.currentNode.Id) - return r.rc.HSet(r.ctx, NodesKey, data).Err() + if err := r.rc.HSet(r.ctx, NodesKey, r.currentNode.Id, data).Err(); err != nil { + return errors.Wrap(err, "could not register node") + } + return nil } func (r *RedisRouter) UnregisterNode() error { @@ -54,8 +59,17 @@ func (r *RedisRouter) UnregisterNode() error { return nil } -func (r *RedisRouter) GetNodeIdForRoom(roomName string) (string, error) { - return r.cr.CachedHGet(NodeRoomKey, roomName) +func (r *RedisRouter) GetNodeForRoom(roomName string) (string, error) { + val, err := r.cr.CachedHGet(NodeRoomKey, roomName) + if err != nil { + err = errors.Wrap(err, "could not get node for room") + } + return val, err +} + +func (r *RedisRouter) SetNodeForRoom(roomName string, nodeId string) error { + // TODO: how do we clear this periodically to remove old rooms? + return r.rc.HSet(r.ctx, NodeRoomKey, roomName, nodeId).Err() } func (r *RedisRouter) GetNode(nodeId string) (*livekit.Node, error) { @@ -70,9 +84,29 @@ func (r *RedisRouter) GetNode(nodeId string) (*livekit.Node, error) { return &n, nil } +func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) { + items, err := r.rc.HVals(r.ctx, NodesKey).Result() + if err != nil { + return nil, errors.Wrap(err, "could not list nodes") + } + nodes := make([]*livekit.Node, 0, len(items)) + for _, item := range items { + n := livekit.Node{} + if err := proto.Unmarshal([]byte(item), &n); err != nil { + return nil, err + } + nodes = append(nodes, &n) + } + return nodes, nil +} + func (r *RedisRouter) SetParticipantRTCNode(participantId, nodeId string) error { r.cr.Expire(participantRTCKey(participantId)) - return r.rc.Set(r.ctx, participantRTCKey(participantId), nodeId, 0).Err() + err := r.rc.Set(r.ctx, participantRTCKey(participantId), nodeId, 0).Err() + if err != nil { + err = errors.Wrap(err, "could not set rtc node") + } + return err } // for a local router, sink and source are pointing to the same spot @@ -107,9 +141,10 @@ func (r *RedisRouter) GetResponseSource(participantId string) (MessageSource, er return source, nil } +// StartParticipant always called on the signal node func (r *RedisRouter) StartParticipant(roomName, participantId, participantName string) error { // find the node where the room is hosted at - rtcNode, err := r.GetNodeIdForRoom(roomName) + rtcNode, err := r.GetNodeForRoom(roomName) if err != nil { return err } @@ -123,6 +158,11 @@ func (r *RedisRouter) StartParticipant(roomName, participantId, participantName return r.LocalRouter.StartParticipant(roomName, participantId, participantId) } + err = r.setParticipantSignalNode(participantId, r.currentNode.Id) + if err != nil { + return err + } + // find signal node to send responses back signalNode, err := r.getParticipantSignalNode(participantId) if err != nil { @@ -152,6 +192,14 @@ func (r *RedisRouter) Stop() { r.cancel() } +func (r *RedisRouter) setParticipantSignalNode(participantId, nodeId string) error { + r.cr.Expire(participantSignalKey(participantId)) + if err := r.rc.Set(r.ctx, participantSignalKey(participantId), nodeId, 0).Err(); err != nil { + return errors.Wrap(err, "could not set signal node") + } + return nil +} + func (r *RedisRouter) getOrCreateRedisSink(nodeId string, participantId string) *RedisSink { r.lock.RLock() sink := r.redisSinks[participantId] @@ -161,12 +209,7 @@ func (r *RedisRouter) getOrCreateRedisSink(nodeId string, participantId string) return sink } - sink = &RedisSink{ - rc: r.rc, - nodeId: nodeId, - participantId: participantId, - channel: nodeChannel(nodeId), - } + sink = NewRedisSink(r.rc, nodeId, participantId) sink.OnClose(func() { r.lock.Lock() delete(r.redisSinks, participantId) @@ -175,7 +218,7 @@ func (r *RedisRouter) getOrCreateRedisSink(nodeId string, participantId string) r.lock.Lock() r.redisSinks[participantId] = sink r.lock.Unlock() - return nil + return sink } func (r *RedisRouter) getParticipantRTCNode(participantId string) (string, error) { @@ -235,7 +278,7 @@ func (r *RedisRouter) subscribeWorker() { case *livekit.RouterMessage_EndSession: signalNode, err := r.getParticipantRTCNode(pId) - if err == nil { + if err != nil { logger.Errorw("could not get participant RTC node", "error", err) continue diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index b23e8b8d1..7be6d975e 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -22,16 +22,16 @@ type FakeRouter struct { result1 *livekit.Node result2 error } - GetNodeIdForRoomStub func(string) (string, error) - getNodeIdForRoomMutex sync.RWMutex - getNodeIdForRoomArgsForCall []struct { + GetNodeForRoomStub func(string) (string, error) + getNodeForRoomMutex sync.RWMutex + getNodeForRoomArgsForCall []struct { arg1 string } - getNodeIdForRoomReturns struct { + getNodeForRoomReturns struct { result1 string result2 error } - getNodeIdForRoomReturnsOnCall map[int]struct { + getNodeForRoomReturnsOnCall map[int]struct { result1 string result2 error } @@ -61,6 +61,18 @@ type FakeRouter struct { result1 routing.MessageSource result2 error } + ListNodesStub func() ([]*livekit.Node, error) + listNodesMutex sync.RWMutex + listNodesArgsForCall []struct { + } + listNodesReturns struct { + result1 []*livekit.Node + result2 error + } + listNodesReturnsOnCall map[int]struct { + result1 []*livekit.Node + result2 error + } OnNewParticipantStub func(routing.ParticipantCallback) onNewParticipantMutex sync.RWMutex onNewParticipantArgsForCall []struct { @@ -76,6 +88,18 @@ type FakeRouter struct { registerNodeReturnsOnCall map[int]struct { result1 error } + SetNodeForRoomStub func(string, string) error + setNodeForRoomMutex sync.RWMutex + setNodeForRoomArgsForCall []struct { + arg1 string + arg2 string + } + setNodeForRoomReturns struct { + result1 error + } + setNodeForRoomReturnsOnCall map[int]struct { + result1 error + } SetParticipantRTCNodeStub func(string, string) error setParticipantRTCNodeMutex sync.RWMutex setParticipantRTCNodeArgsForCall []struct { @@ -193,16 +217,16 @@ func (fake *FakeRouter) GetNodeReturnsOnCall(i int, result1 *livekit.Node, resul }{result1, result2} } -func (fake *FakeRouter) GetNodeIdForRoom(arg1 string) (string, error) { - fake.getNodeIdForRoomMutex.Lock() - ret, specificReturn := fake.getNodeIdForRoomReturnsOnCall[len(fake.getNodeIdForRoomArgsForCall)] - fake.getNodeIdForRoomArgsForCall = append(fake.getNodeIdForRoomArgsForCall, struct { +func (fake *FakeRouter) GetNodeForRoom(arg1 string) (string, error) { + fake.getNodeForRoomMutex.Lock() + ret, specificReturn := fake.getNodeForRoomReturnsOnCall[len(fake.getNodeForRoomArgsForCall)] + fake.getNodeForRoomArgsForCall = append(fake.getNodeForRoomArgsForCall, struct { arg1 string }{arg1}) - stub := fake.GetNodeIdForRoomStub - fakeReturns := fake.getNodeIdForRoomReturns - fake.recordInvocation("GetNodeIdForRoom", []interface{}{arg1}) - fake.getNodeIdForRoomMutex.Unlock() + stub := fake.GetNodeForRoomStub + fakeReturns := fake.getNodeForRoomReturns + fake.recordInvocation("GetNodeForRoom", []interface{}{arg1}) + fake.getNodeForRoomMutex.Unlock() if stub != nil { return stub(arg1) } @@ -212,46 +236,46 @@ func (fake *FakeRouter) GetNodeIdForRoom(arg1 string) (string, error) { return fakeReturns.result1, fakeReturns.result2 } -func (fake *FakeRouter) GetNodeIdForRoomCallCount() int { - fake.getNodeIdForRoomMutex.RLock() - defer fake.getNodeIdForRoomMutex.RUnlock() - return len(fake.getNodeIdForRoomArgsForCall) +func (fake *FakeRouter) GetNodeForRoomCallCount() int { + fake.getNodeForRoomMutex.RLock() + defer fake.getNodeForRoomMutex.RUnlock() + return len(fake.getNodeForRoomArgsForCall) } -func (fake *FakeRouter) GetNodeIdForRoomCalls(stub func(string) (string, error)) { - fake.getNodeIdForRoomMutex.Lock() - defer fake.getNodeIdForRoomMutex.Unlock() - fake.GetNodeIdForRoomStub = stub +func (fake *FakeRouter) GetNodeForRoomCalls(stub func(string) (string, error)) { + fake.getNodeForRoomMutex.Lock() + defer fake.getNodeForRoomMutex.Unlock() + fake.GetNodeForRoomStub = stub } -func (fake *FakeRouter) GetNodeIdForRoomArgsForCall(i int) string { - fake.getNodeIdForRoomMutex.RLock() - defer fake.getNodeIdForRoomMutex.RUnlock() - argsForCall := fake.getNodeIdForRoomArgsForCall[i] +func (fake *FakeRouter) GetNodeForRoomArgsForCall(i int) string { + fake.getNodeForRoomMutex.RLock() + defer fake.getNodeForRoomMutex.RUnlock() + argsForCall := fake.getNodeForRoomArgsForCall[i] return argsForCall.arg1 } -func (fake *FakeRouter) GetNodeIdForRoomReturns(result1 string, result2 error) { - fake.getNodeIdForRoomMutex.Lock() - defer fake.getNodeIdForRoomMutex.Unlock() - fake.GetNodeIdForRoomStub = nil - fake.getNodeIdForRoomReturns = struct { +func (fake *FakeRouter) GetNodeForRoomReturns(result1 string, result2 error) { + fake.getNodeForRoomMutex.Lock() + defer fake.getNodeForRoomMutex.Unlock() + fake.GetNodeForRoomStub = nil + fake.getNodeForRoomReturns = struct { result1 string result2 error }{result1, result2} } -func (fake *FakeRouter) GetNodeIdForRoomReturnsOnCall(i int, result1 string, result2 error) { - fake.getNodeIdForRoomMutex.Lock() - defer fake.getNodeIdForRoomMutex.Unlock() - fake.GetNodeIdForRoomStub = nil - if fake.getNodeIdForRoomReturnsOnCall == nil { - fake.getNodeIdForRoomReturnsOnCall = make(map[int]struct { +func (fake *FakeRouter) GetNodeForRoomReturnsOnCall(i int, result1 string, result2 error) { + fake.getNodeForRoomMutex.Lock() + defer fake.getNodeForRoomMutex.Unlock() + fake.GetNodeForRoomStub = nil + if fake.getNodeForRoomReturnsOnCall == nil { + fake.getNodeForRoomReturnsOnCall = make(map[int]struct { result1 string result2 error }) } - fake.getNodeIdForRoomReturnsOnCall[i] = struct { + fake.getNodeForRoomReturnsOnCall[i] = struct { result1 string result2 error }{result1, result2} @@ -385,6 +409,62 @@ func (fake *FakeRouter) GetResponseSourceReturnsOnCall(i int, result1 routing.Me }{result1, result2} } +func (fake *FakeRouter) ListNodes() ([]*livekit.Node, error) { + fake.listNodesMutex.Lock() + ret, specificReturn := fake.listNodesReturnsOnCall[len(fake.listNodesArgsForCall)] + fake.listNodesArgsForCall = append(fake.listNodesArgsForCall, struct { + }{}) + stub := fake.ListNodesStub + fakeReturns := fake.listNodesReturns + fake.recordInvocation("ListNodes", []interface{}{}) + fake.listNodesMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeRouter) ListNodesCallCount() int { + fake.listNodesMutex.RLock() + defer fake.listNodesMutex.RUnlock() + return len(fake.listNodesArgsForCall) +} + +func (fake *FakeRouter) ListNodesCalls(stub func() ([]*livekit.Node, error)) { + fake.listNodesMutex.Lock() + defer fake.listNodesMutex.Unlock() + fake.ListNodesStub = stub +} + +func (fake *FakeRouter) ListNodesReturns(result1 []*livekit.Node, result2 error) { + fake.listNodesMutex.Lock() + defer fake.listNodesMutex.Unlock() + fake.ListNodesStub = nil + fake.listNodesReturns = struct { + result1 []*livekit.Node + result2 error + }{result1, result2} +} + +func (fake *FakeRouter) ListNodesReturnsOnCall(i int, result1 []*livekit.Node, result2 error) { + fake.listNodesMutex.Lock() + defer fake.listNodesMutex.Unlock() + fake.ListNodesStub = nil + if fake.listNodesReturnsOnCall == nil { + fake.listNodesReturnsOnCall = make(map[int]struct { + result1 []*livekit.Node + result2 error + }) + } + fake.listNodesReturnsOnCall[i] = struct { + result1 []*livekit.Node + result2 error + }{result1, result2} +} + func (fake *FakeRouter) OnNewParticipant(arg1 routing.ParticipantCallback) { fake.onNewParticipantMutex.Lock() fake.onNewParticipantArgsForCall = append(fake.onNewParticipantArgsForCall, struct { @@ -470,6 +550,68 @@ func (fake *FakeRouter) RegisterNodeReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeRouter) SetNodeForRoom(arg1 string, arg2 string) error { + fake.setNodeForRoomMutex.Lock() + ret, specificReturn := fake.setNodeForRoomReturnsOnCall[len(fake.setNodeForRoomArgsForCall)] + fake.setNodeForRoomArgsForCall = append(fake.setNodeForRoomArgsForCall, struct { + arg1 string + arg2 string + }{arg1, arg2}) + stub := fake.SetNodeForRoomStub + fakeReturns := fake.setNodeForRoomReturns + fake.recordInvocation("SetNodeForRoom", []interface{}{arg1, arg2}) + fake.setNodeForRoomMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeRouter) SetNodeForRoomCallCount() int { + fake.setNodeForRoomMutex.RLock() + defer fake.setNodeForRoomMutex.RUnlock() + return len(fake.setNodeForRoomArgsForCall) +} + +func (fake *FakeRouter) SetNodeForRoomCalls(stub func(string, string) error) { + fake.setNodeForRoomMutex.Lock() + defer fake.setNodeForRoomMutex.Unlock() + fake.SetNodeForRoomStub = stub +} + +func (fake *FakeRouter) SetNodeForRoomArgsForCall(i int) (string, string) { + fake.setNodeForRoomMutex.RLock() + defer fake.setNodeForRoomMutex.RUnlock() + argsForCall := fake.setNodeForRoomArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeRouter) SetNodeForRoomReturns(result1 error) { + fake.setNodeForRoomMutex.Lock() + defer fake.setNodeForRoomMutex.Unlock() + fake.SetNodeForRoomStub = nil + fake.setNodeForRoomReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeRouter) SetNodeForRoomReturnsOnCall(i int, result1 error) { + fake.setNodeForRoomMutex.Lock() + defer fake.setNodeForRoomMutex.Unlock() + fake.SetNodeForRoomStub = nil + if fake.setNodeForRoomReturnsOnCall == nil { + fake.setNodeForRoomReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.setNodeForRoomReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeRouter) SetParticipantRTCNode(arg1 string, arg2 string) error { fake.setParticipantRTCNodeMutex.Lock() ret, specificReturn := fake.setParticipantRTCNodeReturnsOnCall[len(fake.setParticipantRTCNodeArgsForCall)] @@ -730,16 +872,20 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.getNodeMutex.RLock() defer fake.getNodeMutex.RUnlock() - fake.getNodeIdForRoomMutex.RLock() - defer fake.getNodeIdForRoomMutex.RUnlock() + fake.getNodeForRoomMutex.RLock() + defer fake.getNodeForRoomMutex.RUnlock() fake.getRequestSinkMutex.RLock() defer fake.getRequestSinkMutex.RUnlock() fake.getResponseSourceMutex.RLock() defer fake.getResponseSourceMutex.RUnlock() + fake.listNodesMutex.RLock() + defer fake.listNodesMutex.RUnlock() fake.onNewParticipantMutex.RLock() defer fake.onNewParticipantMutex.RUnlock() fake.registerNodeMutex.RLock() defer fake.registerNodeMutex.RUnlock() + fake.setNodeForRoomMutex.RLock() + defer fake.setNodeForRoomMutex.RUnlock() fake.setParticipantRTCNodeMutex.RLock() defer fake.setParticipantRTCNodeMutex.RUnlock() fake.startMutex.RLock() diff --git a/pkg/service/errors.go b/pkg/service/errors.go index eb5fdecf8..d71a11be7 100644 --- a/pkg/service/errors.go +++ b/pkg/service/errors.go @@ -3,5 +3,6 @@ package service import "errors" var ( - ErrRoomNotFound = errors.New("requested room does not exist") + ErrRoomNotFound = errors.New("requested room does not exist") + ErrNoRegisteredNodes = errors.New("there are no registered nodes") ) diff --git a/pkg/service/redisroomstore.go b/pkg/service/redisroomstore.go index 15fcf712f..9ffeaa65a 100644 --- a/pkg/service/redisroomstore.go +++ b/pkg/service/redisroomstore.go @@ -4,6 +4,7 @@ import ( "context" "github.com/go-redis/redis/v8" + "github.com/pkg/errors" "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/proto/livekit" @@ -24,7 +25,8 @@ type RedisRoomStore struct { func NewRedisRoomStore(rc *redis.Client) *RedisRoomStore { return &RedisRoomStore{ - rc: rc, + ctx: context.Background(), + rc: rc, } } @@ -38,7 +40,10 @@ func (p *RedisRoomStore) CreateRoom(room *livekit.Room) error { if err != nil { return err } - return p.rc.HSet(p.ctx, RoomsKey, room.Name, data).Err() + if err := p.rc.HSet(p.ctx, RoomsKey, room.Name, data).Err(); err != nil { + return errors.Wrap(err, "could not create room") + } + return nil } func (p *RedisRoomStore) GetRoom(idOrName string) (*livekit.Room, error) { @@ -68,7 +73,7 @@ func (p *RedisRoomStore) GetRoom(idOrName string) (*livekit.Room, error) { func (p *RedisRoomStore) ListRooms() ([]*livekit.Room, error) { items, err := p.rc.HVals(p.ctx, RoomsKey).Result() if err != nil && err != redis.Nil { - return nil, err + return nil, errors.Wrap(err, "could not get rooms") } rooms := make([]*livekit.Room, 0, len(items)) diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 7b4ca5d49..21c976b69 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -4,8 +4,10 @@ import ( "context" "time" + "github.com/thoas/go-funk" "github.com/twitchtv/twirp" + "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/livekit-server/proto/livekit" ) @@ -13,11 +15,13 @@ import ( // A rooms service that supports a single node type RoomService struct { roomProvider RoomStore + router routing.Router } -func NewRoomService(rp RoomStore) (svc *RoomService, err error) { +func NewRoomService(rp RoomStore, router routing.Router) (svc *RoomService, err error) { svc = &RoomService{ roomProvider: rp, + router: router, } return @@ -40,6 +44,22 @@ func (s *RoomService) CreateRoom(ctx context.Context, req *livekit.CreateRoomReq return } + // allocate room to a node + nodes, err := s.router.ListNodes() + if err != nil { + return + } + + if len(nodes) == 0 { + return nil, ErrNoRegisteredNodes + } + + idx := funk.RandomInt(0, len(nodes)) + node := nodes[idx] + if err = s.router.SetNodeForRoom(req.Name, node.Id); err != nil { + return + } + return } diff --git a/pkg/service/rtcrunner.go b/pkg/service/rtcrunner.go index a92220901..ff9bda886 100644 --- a/pkg/service/rtcrunner.go +++ b/pkg/service/rtcrunner.go @@ -11,9 +11,8 @@ import ( "github.com/livekit/livekit-server/proto/livekit" ) -// RTC runner manages the lifecycles of a WebRTC connection +// RTC runner manages the lifecycles of all WebRTC connections // it creates a new goroutine for each participant it manages. - type RTCRunner struct { lock sync.RWMutex roomProvider RoomStore @@ -106,12 +105,16 @@ func (r *RTCRunner) sessionWorker(room *rtc.Room, participant types.Participant, ) // remove peer from room when participant leaves room room.RemoveParticipant(participant.ID()) + + // TODO: notify router to cleanup? }() defer rtc.Recover() for { obj, err := requestSource.ReadMessage() if err == io.EOF { + // TODO: when request is EOF, we might be better off requesting a new connection and waiting + // RTC connection terminating should be the only case that we exit return } diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 48f6b3ab8..3e4b01460 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -81,7 +81,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { participantId := utils.NewGuid(utils.ParticipantPrefix) err = s.router.StartParticipant(roomName, participantId, pName) if err != nil { - handleError(w, http.StatusInternalServerError, "could not set signal node: "+err.Error()) + handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error()) return } diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index c2e13f383..3dc821522 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -15,7 +15,7 @@ import ( // Injectors from wire.go: func InitializeServer(conf *config.Config, keyProvider auth.KeyProvider, roomStore RoomStore, router routing.Router, currentNode routing.LocalNode) (*LivekitServer, error) { - roomService, err := NewRoomService(roomStore) + roomService, err := NewRoomService(roomStore, router) if err != nil { return nil, err } diff --git a/pkg/utils/cachedredis.go b/pkg/utils/cachedredis.go index 30b386b99..30a0cc773 100644 --- a/pkg/utils/cachedredis.go +++ b/pkg/utils/cachedredis.go @@ -6,6 +6,7 @@ import ( "github.com/go-redis/redis/v8" "github.com/karlseguin/ccache/v2" + "github.com/pkg/errors" ) const ( @@ -34,7 +35,7 @@ func (r *CachedRedis) CachedHGet(key, hashKey string) (string, error) { } val, err := r.rc.HGet(r.ctx, key, hashKey).Result() if err != nil { - return "", err + return "", errors.Wrapf(err, "could not hget %s[%s]", key, hashKey) } r.cache.Set(key, val, defaultCacheTTL) return val, nil @@ -47,7 +48,7 @@ func (r *CachedRedis) CachedGet(key string) (string, error) { } val, err := r.rc.Get(r.ctx, key).Result() if err != nil { - return "", err + return "", errors.Wrapf(err, "could not get %s", key) } r.cache.Set(key, val, defaultCacheTTL) diff --git a/test/integration_test.go b/test/integration_test.go index d9244a2c3..9d70a7ee5 100644 --- a/test/integration_test.go +++ b/test/integration_test.go @@ -73,7 +73,7 @@ func TestSinglePublisher(t *testing.T) { } func TestMain(m *testing.M) { - logger.InitDevelopment() + logger.InitDevelopment("") s := createServer() go func() { s.Start()