diff --git a/internal/pkg/mediaconvert/scheduler.go b/internal/pkg/mediaconvert/scheduler.go index 2471749906..24fc394425 100644 --- a/internal/pkg/mediaconvert/scheduler.go +++ b/internal/pkg/mediaconvert/scheduler.go @@ -121,20 +121,37 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) { var destinationFolder = filepath.Join(p.cfg.OutputDir, task.ID) var _, filename = filepath.Split(task.Source) var sourceFilePath = filepath.Join(destinationFolder, filename) - _ = os.MkdirAll(destinationFolder, os.ModePerm) + err = os.MkdirAll(destinationFolder, os.ModePerm) + if err != nil { + logger.Error("can not create temporary folder", zap.Error(err)) + return + } logger.Debug("phase 3: get the remote file") remoteStorage, err := storage.NewStorageByURL(ctx, p.cfg.Endpoint(), p.cfg.EndpointURL.Scheme, tokenString, task.Workspace) if err != nil { - logger.Error("can not create storage by url", zap.Error(err)) + logger.Error("can not create storage by url", zap.Error(err), zap.String("url", p.cfg.EndpointURL.String())) + _ = os.RemoveAll(destinationFolder) + return + } + + stat, err := remoteStorage.StatFile(ctx, task.Source) + if err != nil { + logger.Error("can not stat a file", zap.Error(err), zap.String("filepath", task.Source)) + _ = os.RemoveAll(destinationFolder) + return + } + + if !IsSupportedMediaType(stat.Type) { + logger.Info("unsupported media type", zap.String("type", stat.Type)) _ = os.RemoveAll(destinationFolder) return } if err = remoteStorage.GetFile(ctx, task.Source, sourceFilePath); err != nil { - logger.Error("can not download a file", zap.Error(err)) + logger.Error("can not download a file", zap.Error(err), zap.String("filepath", task.Source)) _ = os.RemoveAll(destinationFolder) // TODO: reschedule return @@ -177,7 +194,12 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) { SourceFile: sourceFilePath, }) - _ = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID) + err = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID) + if err != nil { + logger.Error("can not generate hls playlist", zap.String("out", p.cfg.OutputDir), zap.String("uploadID", opts.UploadID)) + _ = os.RemoveAll(destinationFolder) + return + } go uploader.Start() @@ -240,3 +262,15 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) { } } } + +func IsSupportedMediaType(mediaType string) bool { + // Explicitly disable conversion for video/mp2t and video/x-mpegurl + switch mediaType { + case "video/mp2t", "video/x-mpegurl": + return false + case "video/mp4": + return true + default: + return false + } +} diff --git a/internal/pkg/storage/datalake.go b/internal/pkg/storage/datalake.go index 068c22e8cd..25d07268fc 100644 --- a/internal/pkg/storage/datalake.go +++ b/internal/pkg/storage/datalake.go @@ -271,5 +271,37 @@ func (d *DatalakeStorage) GetFile(ctx context.Context, filename, destination str return nil } +func (d *DatalakeStorage) StatFile(ctx context.Context, filename string) (*BlobInfo, error) { + var logger = d.logger.With(zap.String("head", d.workspace), zap.String("fileName", filename)) + logger.Debug("start") + + var objectKey = getObjectKey(filename) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(d.baseURL + "/blob/" + d.workspace + "/" + objectKey) + req.Header.SetMethod(fasthttp.MethodHead) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + return nil, err + } + + // Check the response status code + if resp.StatusCode() != fasthttp.StatusOK { + var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode()) + logger.Debug("bad status code", zap.Error(err)) + return nil, err + } + + var info BlobInfo + info.Size = int64(resp.Header.ContentLength()) + info.Type = string(resp.Header.ContentType()) + info.ETag = string(resp.Header.Peek("ETag")) + return &info, nil +} + var _ Storage = (*DatalakeStorage)(nil) var _ MetaProvider = (*DatalakeStorage)(nil) diff --git a/internal/pkg/storage/s3.go b/internal/pkg/storage/s3.go index 3864b4fa18..3a4baca952 100644 --- a/internal/pkg/storage/s3.go +++ b/internal/pkg/storage/s3.go @@ -164,3 +164,24 @@ func (u *S3Storage) GetFile(ctx context.Context, filename, dest string) error { return nil } + +func (u *S3Storage) StatFile(ctx context.Context, filename string) (*BlobInfo, error) { + var logger = u.logger.With(zap.String("head", u.bucketName), zap.String("fileName", filename)) + + var head, err = u.client.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: &u.bucketName, + Key: &filename, + }) + + if err != nil { + logger.Error("failed to head object", zap.Error(err)) + return nil, err + } + + var info BlobInfo + info.Size = *head.ContentLength + info.Type = *head.ContentType + info.ETag = *head.ETag + + return &info, nil +} diff --git a/internal/pkg/storage/storage.go b/internal/pkg/storage/storage.go index 79b65ec9c2..8eb726a337 100644 --- a/internal/pkg/storage/storage.go +++ b/internal/pkg/storage/storage.go @@ -30,11 +30,18 @@ type MetaProvider interface { PatchMeta(ctx context.Context, filename string, value *Metadata) error } +type BlobInfo struct { + Size int64 + Type string + ETag string +} + // Storage represents file-based storage type Storage interface { PutFile(ctx context.Context, fileName string) error DeleteFile(ctx context.Context, fileName string) error GetFile(ctx context.Context, fileName, destination string) error + StatFile(ctx context.Context, fileName string) (*BlobInfo, error) } // NewStorageByURL creates a new storage based on the type from the url scheme, for example "datalake://my-datalake-endpoint"