From 7ff1898b3021c7c5085c73460108237c74fc3294 Mon Sep 17 00:00:00 2001 From: Dawin Yurtseven Date: Mon, 25 Nov 2024 14:41:41 +0100 Subject: [PATCH] added edge to cfg for the streams --- api/runner_grpc.go | 17 ++++++++++------- config.yaml | 3 +++ runner/actions/stream.go | 18 +++--------------- runner/handlers.go | 1 - runner/runner.go | 1 - tools/config.go | 11 ++++++++++- 6 files changed, 26 insertions(+), 25 deletions(-) diff --git a/api/runner_grpc.go b/api/runner_grpc.go index e27942572..97b3c21cc 100644 --- a/api/runner_grpc.go +++ b/api/runner_grpc.go @@ -7,6 +7,7 @@ import ( "fmt" "github.com/TUM-Dev/gocast/dao" "github.com/TUM-Dev/gocast/model" + "github.com/TUM-Dev/gocast/tools" "github.com/getsentry/sentry-go" log "github.com/sirupsen/logrus" "github.com/tum-dev/gocast/runner/protobuf" @@ -327,8 +328,10 @@ func (g GrpcRunnerServer) NotifyStreamStarted(ctx context.Context, request *prot logger.Error("Can't set StreamLiveNowTimestamp", "err", err) } + hlsUrl := fmt.Sprintf("%v:%v/%v", tools.Cfg.Edge.Domain, tools.Cfg.Edge.Port, request.HLSUrl) + time.Sleep(time.Second * 5) - if !isHLSUrlOk(request.HLSUrl) { + if !isHLSUrlOk(hlsUrl) { sentry.WithScope(func(scope *sentry.Scope) { scope.SetExtra("URL", request.HLSUrl) scope.SetExtra("StreamID", request.StreamID) @@ -337,18 +340,18 @@ func (g GrpcRunnerServer) NotifyStreamStarted(ctx context.Context, request *prot scope.SetExtra("Version", request.SourceType) sentry.CaptureException(errors.New("DVR URL 404s")) }) - request.HLSUrl = strings.ReplaceAll(request.HLSUrl, "?dvr", "") + hlsUrl = strings.ReplaceAll(hlsUrl, "?dvr", "") } - logger.Info("hls url", "url", request.HLSUrl) + logger.Info("hls url", "url", hlsUrl) switch request.Version { case "CAM": - g.StreamsDao.SaveCAMURL(&stream, request.HLSUrl) + g.StreamsDao.SaveCAMURL(&stream, hlsUrl) case "PRES": - g.StreamsDao.SavePRESURL(&stream, request.HLSUrl) + g.StreamsDao.SavePRESURL(&stream, hlsUrl) default: - g.StreamsDao.SaveCOMBURL(&stream, request.HLSUrl) + g.StreamsDao.SaveCOMBURL(&stream, hlsUrl) } NotifyViewersLiveState(stream.Model.ID, true) @@ -598,7 +601,7 @@ func AssignRunnerAction(dao dao.DaoWrapper, action *model.Action) error { values := map[string]interface{}{} err = json.Unmarshal([]byte(action.Values), &values) for key, value := range values { - logger.Info("values", "value", value) + //logger.Info("values", "value", value) ctx = context.WithValue(ctx, key, value) } ctx = context.WithValue(ctx, "actionID", fmt.Sprintf("%v", action.ID)) diff --git a/config.yaml b/config.yaml index d89a8bdcf..c48f9fea4 100644 --- a/config.yaml +++ b/config.yaml @@ -30,6 +30,9 @@ db: database: tumlive password: example user: root +edge: + domain: http://localhost + port: 8089 ingestbase: rtmp://ingest.tum.live/ jwtkey: # This is an example key, delete and restart to generate a proper one | diff --git a/runner/actions/stream.go b/runner/actions/stream.go index a46d60aaa..7a5c1938a 100644 --- a/runner/actions/stream.go +++ b/runner/actions/stream.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/tum-dev/gocast/runner/protobuf" "log/slog" - "net/url" "os" "os/exec" "path/filepath" @@ -35,10 +34,6 @@ func (a *ActionProvider) StreamAction() *Action { if !ok { return ctx, fmt.Errorf("%w: context doesn't contain hostname", ErrRequiredContextValNotFound) } - edgeVM, ok := ctx.Value("Edge").(string) - if !ok { - return ctx, fmt.Errorf("%w: context doesn't contain edge", ErrRequiredContextValNotFound) - } streamID, ok := ctx.Value("stream").(uint64) if !ok { return ctx, fmt.Errorf("%w: context doesn't contain stream", ErrRequiredContextValNotFound) @@ -107,23 +102,16 @@ func (a *ActionProvider) StreamAction() *Action { continue } - localhls := fmt.Sprintf("%v/%d/%d/%s/%s/playlist.m3u8", hostname, courseID, streamID, version, end.Format("15-04-05")) + //this string is the local runner url that will be passed to TUMLive - edgeURL, err := url.Parse(fmt.Sprintf("http://%v:8089/%v", - edgeVM, localhls, - )) - if err != nil { - return ctx, fmt.Errorf("%w: cannot create urlPath", err) - } - log.Info("streaming", "edgeURL", edgeURL.String()) + localhls := fmt.Sprintf("%v/%d/%d/%s/%s/playlist.m3u8", hostname, courseID, streamID, version, end.Format("15-04-05")) resp := a.Server.NotifyStreamStarted(ctx, &protobuf.StreamStarted{ Hostname: hostname, StreamID: uint32(streamID), CourseID: uint32(courseID), Version: version, - HLSUrl: edgeURL.String(), - // fmt.Sprintf("http://%v:%v/%v:8187/%d/%d/%s/%s/playlist.m3u8", edgeVM, 8089, hostname, courseID, streamID, version, end.Format("15-04-05")), //edgeURL.String(), + HLSUrl: localhls, }) if resp.Ok != true { log.Warn("streamAction: NotifyStreamStarted failed") diff --git a/runner/handlers.go b/runner/handlers.go index dc0833878..f4329751f 100644 --- a/runner/handlers.go +++ b/runner/handlers.go @@ -30,7 +30,6 @@ func (r *Runner) RequestStream(ctx context.Context, req *protobuf.StreamRequest) ctx = context.WithValue(ctx, "URL", "") ctx = context.WithValue(ctx, "Hostname", r.cfg.Hostname) ctx = context.WithValue(ctx, "Port", r.cfg.Port) - ctx = context.WithValue(ctx, "Edge", r.cfg.EdgeServer) ctx = context.WithValue(ctx, "actionID", req.ActionID) r.log.Info("stream request", "jobID", req.ActionID) a := []*actions.Action{ diff --git a/runner/runner.go b/runner/runner.go index a9dbc4f2e..5f6d6b6ac 100644 --- a/runner/runner.go +++ b/runner/runner.go @@ -28,7 +28,6 @@ type envConfig struct { RecPath string `env:"REC_PATH" envDefault:"storage/rec"` GocastServer string `env:"GOCAST_SERVER" envDefault:"localhost:50056"` Hostname string `env:"REALHOST" envDefault:"localhost"` - EdgeServer string `env:"EDGE_SERVER" envDefault:"localhost"` Version string `env:"VERSION" envDefault:"dev"` } diff --git a/tools/config.go b/tools/config.go index d439f32d4..9fa652c2e 100644 --- a/tools/config.go +++ b/tools/config.go @@ -51,7 +51,12 @@ func initConfig() { if err != nil { panic(fmt.Errorf("fatal error config file: %v", err)) } - + if Cfg.Edge.Domain == "" { + logger.Error("No domain for edge found, can't proxy streams") + } + if Cfg.Edge.Port == 0 { + logger.Error("No port for edge found, can't proxy streams") + } // set defaults if Cfg.WorkerToken == "" { Cfg.WorkerToken = uuid.NewV4().String() @@ -113,6 +118,10 @@ type Config struct { Host string `yaml:"host"` Port uint `yaml:"port"` } `yaml:"db"` + Edge struct { + Domain string `yaml:"domain"` + Port int `yaml:"port"` + } `yaml:"edge"` Campus struct { Base string `yaml:"base"` Tokens []string `yaml:"tokens"`