From c333b7a026fd85029b46cc9593146a049df117e5 Mon Sep 17 00:00:00 2001 From: Benjamin Pracht Date: Thu, 4 Aug 2022 09:48:53 -0700 Subject: [PATCH] Add support for Ingress service (#868) Also check for Join permission on ingress API calls --- config-sample.yaml | 7 +++++++ pkg/service/ingress.go | 27 ++++++++++++++++++++------ pkg/service/server.go | 42 ++++++++++++++++++++++++----------------- pkg/service/wire.go | 13 +++++++++++++ pkg/service/wire_gen.go | 15 ++++++++++++++- 5 files changed, 80 insertions(+), 24 deletions(-) diff --git a/config-sample.yaml b/config-sample.yaml index 99b124dd7..989a3254d 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -176,6 +176,13 @@ keys: # # cert_file: /path/to/cert.pem # # key_file: /path/to/key.pem +# ingress server +# ingress: +# # Prefix used to generate RTMP URLs for RTMP ingress. +# # The stream_key will be appended to this base and returned as part of the +# # ingress info +# rtmp_base_url: "rtmp://my.domain.com/live" + # Region of the current node. Required if using regionaware node selector # region: us-west-2 diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index ec6483c80..5ee8c3243 100644 --- a/pkg/service/ingress.go +++ b/pkg/service/ingress.go @@ -25,7 +25,7 @@ type IngressService struct { } func NewIngressService( - conf *config.IngressConfig, + conf *config.Config, rpc ingress.RPC, store IngressStore, rs livekit.RoomService, @@ -33,7 +33,7 @@ func NewIngressService( ) *IngressService { return &IngressService{ - conf: conf, + conf: &conf.Ingress, rpc: rpc, store: store, roomService: rs, @@ -54,7 +54,11 @@ func (s *IngressService) Stop() { } func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateIngressRequest) (*livekit.IngressInfo, error) { - if err := EnsureRecordPermission(ctx); err != nil { + roomName, err := EnsureJoinPermission(ctx) + if err != nil { + return nil, twirpAuthError(err) + } + if req.RoomName != "" && req.RoomName != string(roomName) { return nil, twirpAuthError(err) } @@ -86,9 +90,14 @@ func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateI } func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateIngressRequest) (*livekit.IngressInfo, error) { - if err := EnsureRecordPermission(ctx); err != nil { + roomName, err := EnsureJoinPermission(ctx) + if err != nil { return nil, twirpAuthError(err) } + if req.RoomName != "" && req.RoomName != string(roomName) { + return nil, twirpAuthError(err) + } + if s.rpc == nil { return nil, ErrIngressNotConnected } @@ -146,9 +155,14 @@ func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateI } func (s *IngressService) ListIngress(ctx context.Context, req *livekit.ListIngressRequest) (*livekit.ListIngressResponse, error) { - if err := EnsureRecordPermission(ctx); err != nil { + roomName, err := EnsureJoinPermission(ctx) + if err != nil { return nil, twirpAuthError(err) } + if req.RoomName != "" && req.RoomName != string(roomName) { + return nil, twirpAuthError(err) + } + if s.rpc == nil { return nil, ErrIngressNotConnected } @@ -163,9 +177,10 @@ func (s *IngressService) ListIngress(ctx context.Context, req *livekit.ListIngre } func (s *IngressService) DeleteIngress(ctx context.Context, req *livekit.DeleteIngressRequest) (*livekit.IngressInfo, error) { - if err := EnsureRecordPermission(ctx); err != nil { + if _, err := EnsureJoinPermission(ctx); err != nil { return nil, twirpAuthError(err) } + if s.rpc == nil { return nil, ErrIngressNotConnected } diff --git a/pkg/service/server.go b/pkg/service/server.go index 8f0ba1ea2..7d398a2f0 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -27,23 +27,25 @@ import ( ) type LivekitServer struct { - config *config.Config - egressService *EgressService - rtcService *RTCService - httpServer *http.Server - promServer *http.Server - router routing.Router - roomManager *RoomManager - turnServer *turn.Server - currentNode routing.LocalNode - running atomic.Bool - doneChan chan struct{} - closedChan chan struct{} + config *config.Config + egressService *EgressService + ingressService *IngressService + rtcService *RTCService + httpServer *http.Server + promServer *http.Server + router routing.Router + roomManager *RoomManager + turnServer *turn.Server + currentNode routing.LocalNode + running atomic.Bool + doneChan chan struct{} + closedChan chan struct{} } func NewLivekitServer(conf *config.Config, roomService livekit.RoomService, egressService *EgressService, + ingressService *IngressService, rtcService *RTCService, keyProvider auth.KeyProvider, router routing.Router, @@ -52,11 +54,12 @@ func NewLivekitServer(conf *config.Config, currentNode routing.LocalNode, ) (s *LivekitServer, err error) { s = &LivekitServer{ - config: conf, - egressService: egressService, - rtcService: rtcService, - router: router, - roomManager: roomManager, + config: conf, + egressService: egressService, + ingressService: ingressService, + rtcService: rtcService, + router: router, + roomManager: roomManager, // turn server starts automatically turnServer: turnServer, currentNode: currentNode, @@ -80,6 +83,7 @@ func NewLivekitServer(conf *config.Config, roomServer := livekit.NewRoomServiceServer(roomService) egressServer := livekit.NewEgressServer(egressService) + ingressServer := livekit.NewIngressServer(ingressService) mux := http.NewServeMux() if conf.Development { @@ -90,6 +94,7 @@ func NewLivekitServer(conf *config.Config, } mux.Handle(roomServer.PathPrefix(), roomServer) mux.Handle(egressServer.PathPrefix(), egressServer) + mux.Handle(ingressServer.PathPrefix(), ingressServer) mux.Handle("/rtc", rtcService) mux.HandleFunc("/rtc/validate", rtcService.Validate) mux.HandleFunc("/", s.healthCheck) @@ -150,6 +155,8 @@ func (s *LivekitServer) Start() error { return err } + s.ingressService.Start() + addresses := s.config.BindAddresses if addresses == nil { addresses = []string{""} @@ -238,6 +245,7 @@ func (s *LivekitServer) Start() error { s.roomManager.Stop() s.egressService.Stop() + s.ingressService.Stop() close(s.closedChan) return nil diff --git a/pkg/service/wire.go b/pkg/service/wire.go index 8fa768742..c6bf2e300 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -16,6 +16,7 @@ import ( "github.com/livekit/protocol/auth" "github.com/livekit/protocol/egress" + "github.com/livekit/protocol/ingress" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/webhook" @@ -44,6 +45,9 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live egress.NewRedisRPCClient, getEgressStore, NewEgressService, + ingress.NewRedisRPC, + getIngressStore, + NewIngressService, NewRoomAllocator, NewRoomService, NewRTCService, @@ -175,6 +179,15 @@ func getEgressStore(s ObjectStore) EgressStore { } } +func getIngressStore(s ObjectStore) IngressStore { + switch store := s.(type) { + case *RedisStore: + return store + default: + return nil + } +} + func createClientConfiguration() clientconfiguration.ClientConfigurationManager { return clientconfiguration.NewStaticClientConfigurationManager(clientconfiguration.StaticConfigurations) } diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index d15c07306..650b96e6f 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -17,6 +17,7 @@ import ( "github.com/livekit/livekit-server/pkg/telemetry" "github.com/livekit/protocol/auth" "github.com/livekit/protocol/egress" + "github.com/livekit/protocol/ingress" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/webhook" @@ -61,6 +62,9 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live analyticsService := telemetry.NewAnalyticsService(conf, currentNode) telemetryService := telemetry.NewTelemetryService(notifier, analyticsService) egressService := NewEgressService(rpcClient, objectStore, egressStore, roomService, telemetryService) + rpc := ingress.NewRedisRPC(nodeID, client) + ingressStore := getIngressStore(objectStore) + ingressService := NewIngressService(conf, rpc, ingressStore, roomService, telemetryService) rtcService := NewRTCService(conf, roomAllocator, objectStore, router, currentNode) clientConfigurationManager := createClientConfiguration() roomManager, err := NewLocalRoomManager(conf, objectStore, currentNode, router, telemetryService, clientConfigurationManager) @@ -72,7 +76,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } - livekitServer, err := NewLivekitServer(conf, roomService, egressService, rtcService, keyProvider, router, roomManager, server, currentNode) + livekitServer, err := NewLivekitServer(conf, roomService, egressService, ingressService, rtcService, keyProvider, router, roomManager, server, currentNode) if err != nil { return nil, err } @@ -201,6 +205,15 @@ func getEgressStore(s ObjectStore) EgressStore { } } +func getIngressStore(s ObjectStore) IngressStore { + switch store := s.(type) { + case *RedisStore: + return store + default: + return nil + } +} + func createClientConfiguration() clientconfiguration.ClientConfigurationManager { return clientconfiguration.NewStaticClientConfigurationManager(clientconfiguration.StaticConfigurations) }