diff --git a/pkg/service/errors.go b/pkg/service/errors.go index a1473c3a2..92093d682 100644 --- a/pkg/service/errors.go +++ b/pkg/service/errors.go @@ -24,6 +24,7 @@ var ( ErrIdentityEmpty = psrpc.NewErrorf(psrpc.InvalidArgument, "identity cannot be empty") ErrIngressNotConnected = psrpc.NewErrorf(psrpc.Internal, "ingress not connected (redis required)") ErrIngressNotFound = psrpc.NewErrorf(psrpc.NotFound, "ingress does not exist") + ErrIngressNonReusable = psrpc.NewErrorf(psrpc.InvalidArgument, "ingress is not reusable and cannot be modified") ErrMetadataExceedsLimits = psrpc.NewErrorf(psrpc.InvalidArgument, "metadata size exceeds limits") ErrOperationFailed = psrpc.NewErrorf(psrpc.Internal, "operation cannot be completed") ErrParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "participant does not exist") diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index 45c0bb789..bf4a316c2 100644 --- a/pkg/service/ingress.go +++ b/pkg/service/ingress.go @@ -16,6 +16,8 @@ package service import ( "context" + "fmt" + "net/url" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/telemetry" @@ -27,6 +29,10 @@ import ( "github.com/livekit/psrpc" ) +type IngressLauncher interface { + LaunchPullIngress(ctx context.Context, info *livekit.IngressInfo) (*livekit.IngressInfo, error) +} + type IngressService struct { conf *config.IngressConfig nodeID livekit.NodeID @@ -35,6 +41,30 @@ type IngressService struct { store IngressStore roomService livekit.RoomService telemetry telemetry.TelemetryService + launcher IngressLauncher +} + +func NewIngressServiceWithIngressLauncher( + conf *config.IngressConfig, + nodeID livekit.NodeID, + bus psrpc.MessageBus, + psrpcClient rpc.IngressClient, + store IngressStore, + rs livekit.RoomService, + ts telemetry.TelemetryService, + launcher IngressLauncher, +) *IngressService { + + return &IngressService{ + conf: conf, + nodeID: nodeID, + bus: bus, + psrpcClient: psrpcClient, + store: store, + roomService: rs, + telemetry: ts, + launcher: launcher, + } } func NewIngressService( @@ -46,16 +76,11 @@ func NewIngressService( rs livekit.RoomService, ts telemetry.TelemetryService, ) *IngressService { + s := NewIngressServiceWithIngressLauncher(conf, nodeID, bus, psrpcClient, store, rs, ts, nil) - return &IngressService{ - conf: conf, - nodeID: nodeID, - bus: bus, - psrpcClient: psrpcClient, - store: store, - roomService: rs, - telemetry: ts, - } + s.launcher = s + + return s } func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateIngressRequest) (*livekit.IngressInfo, error) { @@ -70,17 +95,18 @@ func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateI AppendLogFields(ctx, fields...) }() - var urlPrefix string + var url string switch req.InputType { case livekit.IngressInput_RTMP_INPUT: - urlPrefix = s.conf.RTMPBaseURL + url = s.conf.RTMPBaseURL case livekit.IngressInput_WHIP_INPUT: - urlPrefix = s.conf.WHIPBaseURL + url = s.conf.WHIPBaseURL + case livekit.IngressInput_URL_INPUT: default: return nil, ingress.ErrInvalidIngressType } - ig, err := s.CreateIngressWithUrlPrefix(ctx, urlPrefix, req) + ig, err := s.CreateIngressWithUrl(ctx, url, req) if err != nil { return nil, err } @@ -89,7 +115,7 @@ func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateI return ig, nil } -func (s *IngressService) CreateIngressWithUrlPrefix(ctx context.Context, urlPrefix string, req *livekit.CreateIngressRequest) (*livekit.IngressInfo, error) { +func (s *IngressService) CreateIngressWithUrl(ctx context.Context, urlStr string, req *livekit.CreateIngressRequest) (*livekit.IngressInfo, error) { err := EnsureIngressAdminPermission(ctx) if err != nil { return nil, twirpAuthError(err) @@ -98,13 +124,28 @@ func (s *IngressService) CreateIngressWithUrlPrefix(ctx context.Context, urlPref return nil, ErrIngressNotConnected } + if req.InputType == livekit.IngressInput_URL_INPUT { + if req.Url == "" { + return nil, ingress.ErrInvalidIngress("missing URL parameter") + } + urlObj, err := url.Parse(req.Url) + if err != nil { + return nil, psrpc.NewError(psrpc.InvalidArgument, err) + } + if urlObj.Scheme != "http" && urlObj.Scheme != "https" { + return nil, ingress.ErrInvalidIngress(fmt.Sprintf("invalid url scheme %s", urlObj.Scheme)) + } + // Marshall the URL again for sanitization + urlStr = urlObj.String() + } + sk := utils.NewGuid("") info := &livekit.IngressInfo{ IngressId: utils.NewGuid(utils.IngressPrefix), Name: req.Name, StreamKey: sk, - Url: urlPrefix, + Url: urlStr, InputType: req.InputType, Audio: req.Audio, Video: req.Video, @@ -112,14 +153,32 @@ func (s *IngressService) CreateIngressWithUrlPrefix(ctx context.Context, urlPref RoomName: req.RoomName, ParticipantIdentity: req.ParticipantIdentity, ParticipantName: req.ParticipantName, - Reusable: req.InputType == livekit.IngressInput_RTMP_INPUT, State: &livekit.IngressState{}, } + switch req.InputType { + case livekit.IngressInput_RTMP_INPUT, + livekit.IngressInput_WHIP_INPUT: + info.Reusable = true + } + if err := ingress.ValidateForSerialization(info); err != nil { return nil, err } + if req.InputType == livekit.IngressInput_URL_INPUT { + retInfo, err := s.launcher.LaunchPullIngress(ctx, info) + if retInfo != nil { + info = retInfo + } else { + info.State.Status = livekit.IngressState_ENDPOINT_ERROR + info.State.Error = err.Error() + } + if err != nil { + return info, err + } + } + if err = s.store.StoreIngress(ctx, info); err != nil { logger.Errorw("could not write ingress info", err) return nil, err @@ -129,6 +188,14 @@ func (s *IngressService) CreateIngressWithUrlPrefix(ctx context.Context, urlPref return info, nil } +func (s *IngressService) LaunchPullIngress(ctx context.Context, info *livekit.IngressInfo) (*livekit.IngressInfo, error) { + req := &rpc.StartIngressRequest{ + Info: info, + } + + return s.psrpcClient.StartIngress(ctx, req) +} + func updateInfoUsingRequest(req *livekit.UpdateIngressRequest, info *livekit.IngressInfo) error { if req.Name != "" { info.Name = req.Name @@ -183,6 +250,11 @@ func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateI return nil, err } + if !info.Reusable { + logger.Infow("ingress update attempted on non reusable ingress", "ingressID", info.IngressId) + return info, ErrIngressNonReusable + } + switch info.State.Status { case livekit.IngressState_ENDPOINT_ERROR: info.State.Status = livekit.IngressState_ENDPOINT_INACTIVE