update egress obsrv (#1481)

Author: frostbyte73

egress/types.go

@@ -73,22 +73,14 @@ type UploadRequest interface {
 	GetAliOSS() *livekit.AliOSSUpload
 }
 
-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())
-	}
-
+func GetTypes(request any) (string, string) {
 	switch req := request.(type) {
+	// case *livekit.EgressInfo_Egress:
+	// 	return getSourceTypeV2(req.Egress), GetOutputTypeV2(req.Egress.Outputs)
+
+	case *livekit.EgressInfo_Replay:
+		return getSourceTypeV2(req.Replay), GetOutputTypeV2(req.Replay.Outputs)
+	
 	case *livekit.EgressInfo_RoomComposite:
 		return EgressTypeRoomComposite, GetOutputType(req.RoomComposite)
 
@@ -108,6 +100,17 @@ func GetTypes(request interface{}) (string, string) {
 	return Unknown, Unknown
 }
 
+func getSourceTypeV2(req EgressRequest) string {
+	if req.GetMedia() != nil {
+		return EgressTypeMedia
+	} else if req.GetTemplate() != nil {
+		return EgressTypeTemplate
+	} else if req.GetWeb() != nil {
+		return EgressTypeWeb
+	}
+	return Unknown
+}
+
 func GetOutputTypeV2(outputs []*livekit.Output) string {
 	if len(outputs) == 0 {
 		return Unknown

observability/egressobs/egress.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/pkg/errors"
 
+	"github.com/livekit/protocol/egress"
 	"github.com/livekit/protocol/livekit"
 )
 
@@ -18,18 +19,42 @@ type EgressResults struct {
 }
 
 func GetSourceType(info *livekit.EgressInfo) SessionSourceType {
-	switch info.SourceType {
-	case livekit.EgressSourceType_EGRESS_SOURCE_TYPE_WEB:
-		return SessionSourceTypeWeb
-	case livekit.EgressSourceType_EGRESS_SOURCE_TYPE_SDK:
-		return SessionSourceTypeSdk
+	switch r := info.Request.(type) {
+	// case *livekit.EgressInfo_Egress:
+	// 	return getSourceTypeV2(r.Egress)
+	case *livekit.EgressInfo_Replay:
+		return getSourceTypeV2(r.Replay)
 	default:
-		return SessionSourceTypeUndefined
+		switch info.SourceType {
+		case livekit.EgressSourceType_EGRESS_SOURCE_TYPE_WEB:
+			return SessionSourceTypeWeb
+		case livekit.EgressSourceType_EGRESS_SOURCE_TYPE_SDK:
+			return SessionSourceTypeSdk
+		default:
+			return SessionSourceTypeUndefined
+		}
 	}
 }
 
+func getSourceTypeV2(r egress.EgressRequest) SessionSourceType {
+	if r.GetMedia() != nil {
+		return SessionSourceTypeMedia
+	}
+	if r.GetTemplate() != nil {
+		return SessionSourceTypeTemplate
+	}
+	if r.GetWeb() != nil {
+		return SessionSourceTypeWeb
+	}
+	return SessionSourceTypeUndefined
+}
+
 func GetRequestType(info *livekit.EgressInfo) EgressRequestType {
 	switch info.Request.(type) {
+	// case *livekit.EgressInfo_Egress:
+	// 	return EgressRequestTypeEgress
+	case *livekit.EgressInfo_Replay:
+		return EgressRequestTypeReplay
 	case *livekit.EgressInfo_RoomComposite:
 		return EgressRequestTypeRoomComposite
 	case *livekit.EgressInfo_Web:
@@ -68,6 +93,18 @@ func GetStatus(info *livekit.EgressInfo) SessionStatus {
 
 func GetRequest(info *livekit.EgressInfo) (string, error) {
 	switch req := info.Request.(type) {
+	// case *livekit.EgressInfo_Egress:
+	// 	b, err := protojson.Marshal(req.Egress)
+	// 	if err != nil {
+	// 		return "", errors.Wrap(err, "failed to marshal egress request")
+	// 	}
+	// 	return string(b), nil
+	case *livekit.EgressInfo_Replay:
+		b, err := protojson.Marshal(req.Replay)
+		if err != nil {
+			return "", errors.Wrap(err, "failed serializing Replay request")
+		}
+		return string(b), nil
 	case *livekit.EgressInfo_RoomComposite:
 		b, err := protojson.Marshal(req.RoomComposite)
 		if err != nil {