dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-22 07:49:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

config rework to handle distributed setup, fixed redis bugs

Browse Source
This commit is contained in:
David Zhao
2021-01-17 17:06:50 -08:00
parent 30e99e8c6b
commit 458241d1f1
21 changed files with 382 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/cli/main.go
+1 -1
View File
@@ -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)
}
cmd/server/main.go
+19 -20
View File
@@ -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")
go.mod
+1
View File
@@ -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
go.sum
+2
View File
@@ -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=
pkg/auth/provider.go
+7 -1
View File
@@ -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]
}
pkg/auth/provider_test.go
+1 -1
View File
@@ -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 {
pkg/config/config.go
+27 -4
View File
@@ -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
}
pkg/logger/logger.go
+18 -10
View File
@@ -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{}) {
pkg/routing/interfaces.go
+4 -1
View File
@@ -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)
pkg/routing/localrouter.go
+11 -1
View File
@@ -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 {
pkg/routing/redis.go
+8 -1
View File
@@ -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)
}
pkg/routing/redisrouter.go
+56 -13
View File
@@ -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
pkg/routing/routingfakes/fake_router.go
+185 -39
View File
@@ -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()
pkg/service/errors.go
+2 -1
View File
@@ -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")
)
pkg/service/redisroomstore.go
+8 -3
View File
@@ -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))
pkg/service/roomservice.go
+21 -1
View File
@@ -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
}
pkg/service/rtcrunner.go
+5 -2
View File
@@ -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
}
pkg/service/rtcservice.go
+1 -1
View File
@@ -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
}
pkg/service/wire_gen.go
+1 -1
View File
@@ -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
}
pkg/utils/cachedredis.go
+3 -2
View File
@@ -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)
test/integration_test.go
+1 -1
View File
@@ -73,7 +73,7 @@ func TestSinglePublisher(t *testing.T) {
}
func TestMain(m *testing.M) {
logger.InitDevelopment()
logger.InitDevelopment("")
s := createServer()
go func() {
s.Start()