Egress v2 (#1454)

* egress v2

* generated protobuf

* undo deprecations

* generated protobuf

* deprecate unused quality fields

* generated protobuf

* remove mcap related fields for now

* generated protobuf

* update psrpc handler

* generated protobuf

* aliOss -> aliOSS

* generated protobuf

* update helpers

* replay only

* generated protobuf

---------

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

Author: frostbyte73

.github/workflows/slack-notifier.yaml

@@ -2,7 +2,7 @@ name: PR Slack Notifier
 
 on:
   pull_request:
-    types: [review_requested, reopened, closed]
+    types: [review_requested, reopened, closed, synchronize]
   pull_request_review:
     types: [submitted]
 

egress/notify_options.go

@@ -31,6 +31,14 @@ func GetEgressNotifyOptions(egressInfo *livekit.EgressInfo) []webhook.NotifyOpti
 	var whs []*livekit.WebhookConfig
 
 	switch req := egressInfo.Request.(type) {
+	case *livekit.EgressInfo_Egress:
+		if req.Egress != nil {
+			whs = req.Egress.Webhooks
+		}
+	case *livekit.EgressInfo_Replay:
+		if req.Replay != nil {
+			whs = req.Replay.Webhooks
+		}
 	case *livekit.EgressInfo_RoomComposite:
 		if req.RoomComposite != nil {
 			whs = req.RoomComposite.Webhooks

egress/redact.go

@@ -19,6 +19,17 @@ import (
 	"github.com/livekit/protocol/utils"
 )
 
+// RedactStartEgressRequest redacts secrets from a V2 StartEgressRequest.
+func RedactStartEgressRequest(req EgressRequest) {
+	RedactUpload(req.GetStorage())
+	for _, output := range req.GetOutputs() {
+		RedactUpload(output.Storage)
+		if stream := output.GetStream(); stream != nil {
+			RedactStreamKeys(stream)
+		}
+	}
+}
+
 func RedactUpload(req UploadRequest) {
 	if s3 := req.GetS3(); s3 != nil {
 		s3.AccessKey = utils.Redact(s3.AccessKey, "{access_key}")

egress/types.go

@@ -17,8 +17,11 @@ package egress
 import "github.com/livekit/protocol/livekit"
 
 const (
+	EgressTypeTemplate = "template"
+	EgressTypeWeb      = "web"
+	EgressTypeMedia    = "media"
+
 	EgressTypeRoomComposite  = "room_composite"
-	EgressTypeWeb            = "web"
 	EgressTypeParticipant    = "participant"
 	EgressTypeTrackComposite = "track_composite"
 	EgressTypeTrack          = "track"
@@ -55,6 +58,14 @@ type DirectOutput interface {
 	GetWebsocketUrl() string
 }
 
+type EgressRequest interface {
+	GetMedia() *livekit.MediaSource
+	GetTemplate() *livekit.TemplateSource
+	GetWeb() *livekit.WebSource
+	GetOutputs() []*livekit.Output
+	GetStorage() *livekit.StorageConfig
+}
+
 type UploadRequest interface {
 	GetS3() *livekit.S3Upload
 	GetGcp() *livekit.GCPUpload
@@ -63,6 +74,20 @@ type UploadRequest interface {
 }
 
 func GetTypes(request interface{}) (string, string) {
+	if r, ok := request.(EgressRequest); ok {
+		var egressType string
+		if r.GetMedia() != nil {
+			egressType = EgressTypeMedia
+		} else if r.GetTemplate() != nil {
+			egressType = EgressTypeTemplate
+		} else if r.GetWeb() != nil {
+			egressType = EgressTypeWeb
+		} else {
+			egressType = Unknown
+		}
+		return egressType, GetOutputTypeV2(r.GetOutputs())
+	}
+
 	switch req := request.(type) {
 	case *livekit.EgressInfo_RoomComposite:
 		return EgressTypeRoomComposite, GetOutputType(req.RoomComposite)
@@ -83,6 +108,27 @@ func GetTypes(request interface{}) (string, string) {
 	return Unknown, Unknown
 }
 
+func GetOutputTypeV2(outputs []*livekit.Output) string {
+	if len(outputs) == 0 {
+		return Unknown
+	}
+	if len(outputs) > 1 {
+		return OutputTypeMultiple
+	}
+	switch outputs[0].Config.(type) {
+	case *livekit.Output_File:
+		return OutputTypeFile
+	case *livekit.Output_Stream:
+		return OutputTypeStream
+	case *livekit.Output_Segments:
+		return OutputTypeSegments
+	case *livekit.Output_Images:
+		return OutputTypeImages
+	default:
+		return Unknown
+	}
+}
+
 func GetOutputType(req interface{}) string {
 	if r, ok := req.(EncodedOutput); ok {
 		outputs := make([]string, 0)