diff --git a/internal/pkg/api/v1/transcoding/handler.go b/internal/pkg/api/v1/transcoding/handler.go index c3a3ecfc92..b9871f7e12 100644 --- a/internal/pkg/api/v1/transcoding/handler.go +++ b/internal/pkg/api/v1/transcoding/handler.go @@ -68,7 +68,7 @@ func (t *trascodeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -// NewHandler creates a new trnascoding http handler, requires context and config. +// NewHandler creates a new transcoding http handler, requires context and config. func NewHandler(ctx context.Context, cfg *config.Config) http.Handler { return &trascodeHandler{ scheduler: mediaconvert.NewScheduler(ctx, cfg), diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index f182a61075..ca0e4aef22 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -25,15 +25,15 @@ import ( // Config represents configuration for the huly-stream application. type Config struct { - SentryDsn string `split_words:"true" default:"" desc:"sentry dsn value"` - LogLevel string `split_words:"true" default:"debug" desc:"sets log level for the application"` - ServerSecret string `split_words:"true" default:"" desc:"server secret required to generate and verify tokens"` - PprofEnabled bool `split_words:"true" default:"true" desc:"starts profile server on localhost:6060 if true"` - Insecure bool `split_words:"true" default:"false" desc:"ignores authorization check if true"` - ServeURL string `split_words:"true" desc:"listen on url" default:"0.0.0.0:1080"` - EndpointURL *url.URL `split_words:"true" default:"s3://127.0.0.1:9000" desc:"S3 or Datalake endpoint, example: s3://my-ip-address, datalake://my-ip-address"` - MaxParallelScalingCount int `split_words:"true" default:"2" desc:"how much parallel scaling can be processed"` - MaxThreadCount int `split_words:"true" default:"4" desc:"max number of threads for transcoder"` + SentryDsn string `split_words:"true" default:"" desc:"sentry dsn value"` + LogLevel string `split_words:"true" default:"debug" desc:"sets log level for the application"` + ServerSecret string `split_words:"true" default:"" desc:"server secret required to generate and verify tokens"` + PprofEnabled bool `split_words:"true" default:"true" desc:"starts profile server on localhost:6060 if true"` + Insecure bool `split_words:"true" default:"false" desc:"ignores authorization check if true"` + ServeURL string `split_words:"true" desc:"listen on url" default:"0.0.0.0:1080"` + EndpointURL *url.URL `split_words:"true" default:"s3://127.0.0.1:9000" desc:"S3 or Datalake endpoint, example: s3://my-ip-address, datalake://my-ip-address"` + MaxParallelTranscodingCount int `split_words:"true" default:"2" desc:"how much parallel transcodings can be processed"` + MaxThreadCount int `split_words:"true" default:"4" desc:"max number of threads for transcoder"` QueueConfig string `split_words:"true" default:"" desc:"queue config"` Region string `split_words:"true" default:"" desc:"service region"` diff --git a/internal/pkg/manifest/hls.go b/internal/pkg/manifest/hls.go index dcb421fe50..7b2934ea46 100644 --- a/internal/pkg/manifest/hls.go +++ b/internal/pkg/manifest/hls.go @@ -18,13 +18,12 @@ import ( "fmt" "os" "path/filepath" - "strings" - "github.com/hcengineering/stream/internal/pkg/resconv" + "github.com/hcengineering/stream/internal/pkg/profile" ) // GenerateHLSPlaylist generates master file for master files for resolution levels -func GenerateHLSPlaylist(levels []string, outputPath, uploadID string) error { +func GenerateHLSPlaylist(profiles []profile.VideoProfile, outputPath, uploadID string) error { p := filepath.Join(outputPath, uploadID, fmt.Sprintf("%v_master.m3u8", uploadID)) d := filepath.Dir(p) _ = os.MkdirAll(d, os.ModePerm) @@ -40,16 +39,16 @@ func GenerateHLSPlaylist(levels []string, outputPath, uploadID string) error { return err } - for _, res := range levels { - var bandwidth = resconv.Bandwidth(res) - var resolution = strings.ReplaceAll(resconv.Resolution(res), ":", "x") + for _, profile := range profiles { + var bandwidth = profile.Bandwidth + var resolution = fmt.Sprintf("%vx%v", profile.Width, profile.Height) _, err = file.WriteString(fmt.Sprintf("#EXT-X-STREAM-INF:BANDWIDTH=%d,RESOLUTION=%v\n", bandwidth, resolution)) if err != nil { return err } - _, err = file.WriteString(fmt.Sprintf("%s_%s_master.m3u8\n", uploadID, res)) + _, err = file.WriteString(fmt.Sprintf("%s_%s_master.m3u8\n", uploadID, profile.Name)) if err != nil { return err } diff --git a/internal/pkg/manifest/hls_test.go b/internal/pkg/manifest/hls_test.go index a37c63b874..5ad6dd6758 100644 --- a/internal/pkg/manifest/hls_test.go +++ b/internal/pkg/manifest/hls_test.go @@ -19,14 +19,24 @@ import ( "testing" "github.com/hcengineering/stream/internal/pkg/manifest" + "github.com/hcengineering/stream/internal/pkg/profile" "github.com/stretchr/testify/require" ) func TestGenerateHLSPlaylist(t *testing.T) { - resolutions := []string{"320p", "480p", "720p", "1080p", "4k", "8k"} + profiles := []profile.VideoProfile{ + profile.Profile360p, + profile.Profile480p, + profile.Profile720p, + profile.Profile1080p, + profile.Profile1440p, + } uploadID := "test123" + defer func() { + _ = os.RemoveAll(uploadID) + }() - err := manifest.GenerateHLSPlaylist(resolutions, "", uploadID) + err := manifest.GenerateHLSPlaylist(profiles, "", uploadID) require.NoError(t, err) outputPath := filepath.Join(uploadID, uploadID+"_master.m3u8") @@ -42,10 +52,8 @@ func TestGenerateHLSPlaylist(t *testing.T) { require.Contains(t, playlistContent, "#EXTM3U", "File must start with #EXTM3U") - for _, res := range resolutions { - expectedLine := uploadID + "_" + res + "_master.m3u8" + for _, prof := range profiles { + expectedLine := uploadID + "_" + prof.Name + "_master.m3u8" require.Contains(t, playlistContent, expectedLine, "Missing expected reference: "+expectedLine) } - - _ = os.RemoveAll(uploadID) } diff --git a/internal/pkg/mediaconvert/command.go b/internal/pkg/mediaconvert/command.go index 3b02624031..895b5e08c2 100644 --- a/internal/pkg/mediaconvert/command.go +++ b/internal/pkg/mediaconvert/command.go @@ -21,11 +21,13 @@ import ( "io" "os/exec" "path/filepath" + "strconv" "strings" "github.com/pkg/errors" "github.com/hcengineering/stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/profile" "go.uber.org/zap" ) @@ -55,14 +57,12 @@ const ( // Options represents configuration for the ffmpeg command type Options struct { - Input string - OutputDir string - ScalingLevels []string - Level string - LogLevel LogLevel - Transcode bool - Threads int - UploadID string + Input string + OutputDir string + LogLevel LogLevel + Threads int + UploadID string + Profiles []profile.VideoProfile } func newFfmpegCommand(ctx context.Context, in io.Reader, args []string) (*exec.Cmd, error) { @@ -70,14 +70,13 @@ func newFfmpegCommand(ctx context.Context, in io.Reader, args []string) (*exec.C return nil, errors.New("ctx should not be nil") } + var cmd = exec.CommandContext(ctx, "ffmpeg", args...) + cmd.Stdin = in + var logger = log.FromContext(ctx).With(zap.String("func", "newFFMpegCommand")) + logger.Debug("prepared command: ", zap.String("cmd", cmd.String())) - logger.Debug("prepared command: ", zap.Strings("args", args)) - - var result = exec.CommandContext(ctx, "ffmpeg", args...) - result.Stdin = in - - return result, nil + return cmd, nil } func buildCommonCommand(opts *Options) []string { @@ -90,6 +89,7 @@ func buildCommonCommand(opts *Options) []string { "-i", opts.Input, } + // If input is a URL, add HTTP specific parameters if strings.HasPrefix(opts.Input, "http://") || strings.HasPrefix(opts.Input, "https://") { result = append(result, "-reconnect", "1", @@ -101,6 +101,49 @@ func buildCommonCommand(opts *Options) []string { return result } +func buildHLSCommand(profile profile.VideoProfile, opts *Options) []string { + return []string{ + "-f", "hls", + "-hls_time", "5", + // Use HLS flags + // - split_by_time + // Allow segments to start on frames other than key frames. + // This improves behavior on some players when the time between key frames is inconsistent, + // but may make things worse on others, and can cause some oddities during seeking. + // This flag should be used with the hls_time option. + // - temp_file + // Write segment data to filename.tmp and rename to filename only once the segment is complete. + "-hls_flags", "split_by_time+temp_file", + // Do not limit number of HLS segments + "-hls_list_size", "0", + "-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", profile.Name)), + } +} + +func buildVideoCommand(profile profile.VideoProfile, opts *Options) []string { + crf := profile.CRF + if crf == 0 { + crf = 23 + } + command := []string{ + // Transcode only first video and optionally audio stream + "-map", "0:v:0", + "-map", "0:a?", + // Set up codecs + "-c:a", profile.AudioCodec, + "-c:v", profile.VideoCodec, + "-preset", "veryfast", + "-crf", strconv.Itoa(crf), + "-g", "60", + } + + if profile.VideoCodec != "copy" && profile.Scale { + command = append(command, "-vf", "scale=-2:"+strconv.Itoa(profile.Height)) + } + + return command +} + // BuildAudioCommand returns flags for getting the audio from the input func BuildAudioCommand(opts *Options) []string { var commonPart = buildCommonCommand(opts) @@ -111,34 +154,19 @@ func BuildAudioCommand(opts *Options) []string { ) } -// BuildRawVideoCommand returns an extremely lightweight ffmpeg command for converting raw video without extra cost. -func BuildRawVideoCommand(opts *Options) []string { - if opts.Transcode { - return append(buildCommonCommand(opts), - "-map", "0:v:0", - "-map", "0:a?", - "-c:a", "aac", - "-c:v", "libx264", - "-preset", "veryfast", - "-crf", "23", - "-g", "60", - "-f", "hls", - "-hls_time", "5", - "-hls_flags", "split_by_time+temp_file", - "-hls_list_size", "0", - "-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", opts.Level)), - filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, opts.Level))) +// BuildVideoCommand returns ffmpeg command for converting video. +func BuildVideoCommand(opts *Options) []string { + if len(opts.Profiles) == 0 { + return []string{} } - return append(buildCommonCommand(opts), - "-c:a", "copy", // Copy audio stream - "-c:v", "copy", // Copy video stream - "-f", "hls", - "-hls_time", "5", - "-hls_flags", "split_by_time+temp_file", - "-hls_list_size", "0", - "-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", opts.Level)), - filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, opts.Level))) + var command = buildCommonCommand(opts) + for _, profile := range opts.Profiles { + command = append(command, buildVideoCommand(profile, opts)...) + command = append(command, buildHLSCommand(profile, opts)...) + command = append(command, filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, profile.Name))) + } + return command } // BuildThumbnailCommand creates a command that creates a thumbnail for the input video @@ -150,48 +178,3 @@ func BuildThumbnailCommand(opts *Options) []string { filepath.Join(opts.OutputDir, opts.UploadID, opts.UploadID+".jpg"), ) } - -// BuildScalingVideoCommand returns flags for ffmpeg for video scaling -func BuildScalingVideoCommand(opts *Options) []string { - if len(opts.ScalingLevels) == 0 { - return []string{} - } - - if len(opts.ScalingLevels) == 1 && opts.ScalingLevels[0] == opts.Level { - return []string{} - } - - var result = buildCommonCommand(opts) - - for _, level := range opts.ScalingLevels { - if level == opts.Level { - continue - } - - result = append(result, - "-map", "0:v:0", - "-map", "0:a?", - "-vf", "scale=-2:"+level[:len(level)-1], - "-c:a", "aac", - "-c:v", "libx264", - "-preset", "veryfast", - "-crf", "23", - "-g", "60", - "-f", "hls", - "-hls_time", "5", - // Use HLS flags - // - split_by_time - // Allow segments to start on frames other than key frames. - // This improves behavior on some players when the time between key frames is inconsistent, - // but may make things worse on others, and can cause some oddities during seeking. - // This flag should be used with the hls_time option. - // - temp_file - // Write segment data to filename.tmp and rename to filename only once the segment is complete. - "-hls_flags", "split_by_time+temp_file", - "-hls_list_size", "0", - "-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", level)), - filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, level))) - } - - return result -} diff --git a/internal/pkg/mediaconvert/command_test.go b/internal/pkg/mediaconvert/command_test.go index 0b83a79ab6..67beff10b4 100644 --- a/internal/pkg/mediaconvert/command_test.go +++ b/internal/pkg/mediaconvert/command_test.go @@ -18,83 +18,79 @@ import ( "testing" "github.com/hcengineering/stream/internal/pkg/mediaconvert" - "github.com/hcengineering/stream/internal/pkg/resconv" + "github.com/hcengineering/stream/internal/pkg/profile" "github.com/stretchr/testify/require" ) +func Test_BuildVideoCommand_Empty(t *testing.T) { + var profiles []profile.VideoProfile + + var rawCommand = mediaconvert.BuildVideoCommand(&mediaconvert.Options{ + OutputDir: "test", + Input: "pipe:0", + UploadID: "1", + Threads: 4, + LogLevel: mediaconvert.LogLevelDebug, + Profiles: profiles, + }) + + require.Empty(t, rawCommand) +} + func Test_BuildVideoCommand_Scaling(t *testing.T) { - var scaleCommand = mediaconvert.BuildScalingVideoCommand(&mediaconvert.Options{ - OutputDir: "test", - Input: "pipe:0", - UploadID: "1", - Threads: 4, - LogLevel: mediaconvert.LogLevelDebug, - ScalingLevels: []string{"720p", "480p"}, - }) + var profiles = []profile.VideoProfile{ + profile.Profile720p, + profile.Profile480p, + } - const expected = `-y -v debug -err_detect ignore_err -fflags +discardcorrupt -threads 4 -i pipe:0 -map 0:v:0 -map 0:a? -vf scale=-2:720 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_720p.ts test/1/1_720p_master.m3u8 -map 0:v:0 -map 0:a? -vf scale=-2:480 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8` - - require.Contains(t, expected, strings.Join(scaleCommand, " ")) -} - -func Test_BuildVideoCommand_Scaling_NoRaw(t *testing.T) { - var scaleCommand = mediaconvert.BuildScalingVideoCommand(&mediaconvert.Options{ - OutputDir: "test", - Input: "pipe:0", - UploadID: "1", - Threads: 4, - LogLevel: mediaconvert.LogLevelDebug, - Level: "720p", - ScalingLevels: []string{"720p", "480p"}, - }) - - const expected = `-y -v debug -err_detect ignore_err -fflags +discardcorrupt -threads 4 -i pipe:0 -map 0:v:0 -map 0:a? -vf scale=-2:480 -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8` - - require.Contains(t, expected, strings.Join(scaleCommand, " ")) -} - -func Test_BuildVideoCommand_Raw_NoTranscode(t *testing.T) { - var rawCommand = mediaconvert.BuildRawVideoCommand(&mediaconvert.Options{ + var scaleCommand = mediaconvert.BuildVideoCommand(&mediaconvert.Options{ OutputDir: "test", Input: "pipe:0", UploadID: "1", Threads: 4, LogLevel: mediaconvert.LogLevelDebug, - Level: resconv.Level("651:490"), - Transcode: false, + Profiles: profiles, }) - const expected = `"-y -v debug -err_detect ignore_err -fflags +discardcorrupt -threads 4 -i pipe:0 -c:a copy -c:v copy -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8` + const expected = `-y -v debug -err_detect ignore_err -fflags +discardcorrupt -threads 4 -i pipe:0 -map 0:v:0 -map 0:a? -c:a aac -c:v libx264 -preset veryfast -crf 25 -g 60 -vf scale=-2:720 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_720p.ts test/1/1_720p_master.m3u8 -map 0:v:0 -map 0:a? -c:a aac -c:v libx264 -preset veryfast -crf 27 -g 60 -vf scale=-2:480 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8` - require.Contains(t, expected, strings.Join(rawCommand, " ")) + require.Contains(t, expected, strings.Join(scaleCommand, " ")) } -func Test_BuildVideoCommand_Raw_Transcode(t *testing.T) { - var rawCommand = mediaconvert.BuildRawVideoCommand(&mediaconvert.Options{ +func Test_BuildVideoCommand_Original(t *testing.T) { + var profiles = []profile.VideoProfile{ + profile.MakeProfileOriginal(640, 480), + } + + var rawCommand = mediaconvert.BuildVideoCommand(&mediaconvert.Options{ OutputDir: "test", Input: "pipe:0", UploadID: "1", Threads: 4, LogLevel: mediaconvert.LogLevelDebug, - Level: resconv.Level("651:490"), - Transcode: true, + Profiles: profiles, }) - const expected = `-y -v debug -err_detect ignore_err -fflags +discardcorrupt -threads 4 -i pipe:0 -map 0:v:0 -map 0:a? -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8` + const expected = `-y -v debug -err_detect ignore_err -fflags +discardcorrupt -threads 4 -i pipe:0 -map 0:v:0 -map 0:a? -c:a copy -c:v copy -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_orig.ts test/1/1_orig_master.m3u8` require.Contains(t, expected, strings.Join(rawCommand, " ")) } -func Test_BuildVideoCommand_Scaling_Small(t *testing.T) { - var scaleCommand = mediaconvert.BuildScalingVideoCommand(&mediaconvert.Options{ - OutputDir: "test", - Input: "pipe:0", - UploadID: "1", - Threads: 4, - LogLevel: mediaconvert.LogLevelDebug, - Level: "360p", - ScalingLevels: []string{"360p"}, +func Test_BuildVideoCommand_OriginalT(t *testing.T) { + var profiles = []profile.VideoProfile{ + profile.MakeProfileOriginalT(640, 480), + } + + var rawCommand = mediaconvert.BuildVideoCommand(&mediaconvert.Options{ + OutputDir: "test", + Input: "pipe:0", + UploadID: "1", + Threads: 4, + LogLevel: mediaconvert.LogLevelDebug, + Profiles: profiles, }) - require.Empty(t, scaleCommand) + const expected = `-y -v debug -err_detect ignore_err -fflags +discardcorrupt -threads 4 -i pipe:0 -map 0:v:0 -map 0:a? -c:a aac -c:v libx264 -preset veryfast -crf 23 -g 60 -f hls -hls_time 5 -hls_flags split_by_time+temp_file -hls_list_size 0 -hls_segment_filename test/1/1_%03d_orig.ts test/1/1_orig_master.m3u8` + + require.Contains(t, expected, strings.Join(rawCommand, " ")) } diff --git a/internal/pkg/mediaconvert/coordinator.go b/internal/pkg/mediaconvert/coordinator.go index d2721134ee..d20520a203 100644 --- a/internal/pkg/mediaconvert/coordinator.go +++ b/internal/pkg/mediaconvert/coordinator.go @@ -17,7 +17,11 @@ package mediaconvert import ( "context" + "fmt" "path/filepath" + "regexp" + "strconv" + "strings" "sync" "sync/atomic" "time" @@ -27,7 +31,6 @@ import ( "github.com/google/uuid" "github.com/hcengineering/stream/internal/pkg/config" "github.com/hcengineering/stream/internal/pkg/log" - "github.com/hcengineering/stream/internal/pkg/resconv" "github.com/hcengineering/stream/internal/pkg/sharedpipe" "github.com/hcengineering/stream/internal/pkg/storage" "github.com/hcengineering/stream/internal/pkg/uploader" @@ -40,7 +43,7 @@ type StreamCoordinator struct { conf *config.Config uploadOptions uploader.Options - activeScalling int32 + activeTranscoding int32 mainContext context.Context logger *zap.Logger @@ -49,6 +52,11 @@ type StreamCoordinator struct { cancels sync.Map } +var _ handler.DataStore = (*StreamCoordinator)(nil) +var _ handler.ConcaterDataStore = (*StreamCoordinator)(nil) +var _ handler.TerminaterDataStore = (*StreamCoordinator)(nil) +var _ handler.LengthDeferrerDataStore = (*StreamCoordinator)(nil) + // NewStreamCoordinator creates a new scheduler for transcode operations. func NewStreamCoordinator(ctx context.Context, c *config.Config) *StreamCoordinator { return &StreamCoordinator{ @@ -81,42 +89,64 @@ func (s *StreamCoordinator) NewUpload(ctx context.Context, info handler.FileInfo done: make(chan struct{}), } - var scaling = resconv.SubLevels(info.MetaData["resolution"]) - var level = resconv.Level(info.MetaData["resolution"]) - var cost int64 - - for _, scale := range scaling { - cost += int64(resconv.Pixels(resconv.Resolution(scale))) - } - - if atomic.AddInt32(&s.activeScalling, 1) > int32(s.conf.MaxParallelScalingCount) { - atomic.AddInt32(&s.activeScalling, -1) + if atomic.AddInt32(&s.activeTranscoding, 1) > int32(s.conf.MaxParallelTranscodingCount) { + atomic.AddInt32(&s.activeTranscoding, -1) s.logger.Debug("run out of resources for scaling") - scaling = nil + // TODO do not transcode } + width, err := strconv.Atoi(info.MetaData["width"]) + if err != nil { + return nil, errors.Wrapf(err, "can not parse video width: %v", info.MetaData["width"]) + } + + height, err := strconv.Atoi(info.MetaData["height"]) + if err != nil { + return nil, errors.Wrapf(err, "can not parse video height: %v", info.MetaData["height"]) + } + + meta := VideoMeta{ + Width: width, + Height: height, + Codec: extractCodec(info.MetaData["contentType"]), + ContentType: extractContentType(info.MetaData["contentType"]), + } + profiles := FastTranscodingProfiles(meta) + var commandOptions = Options{ - Input: "pipe:0", - OutputDir: s.conf.OutputDir, - Threads: s.conf.MaxThreadCount, - UploadID: info.ID, - Transcode: true, - Level: level, - ScalingLevels: scaling, + Input: "pipe:0", + OutputDir: s.conf.OutputDir, + Threads: s.conf.MaxThreadCount, + UploadID: info.ID, + Profiles: profiles, } if s.conf.EndpointURL != nil { s.logger.Sugar().Debugf("initializing uploader for %v", info) + + // setup content uploader for transcoded outputs var opts = s.uploadOptions opts.Dir = filepath.Join(opts.Dir, info.ID) - var storage, err = storage.NewStorageByURL(s.mainContext, s.conf.Endpoint(), s.conf.EndpointURL.Scheme, info.MetaData["token"], info.MetaData["workspace"]) + // create storage backend + var stg, err = storage.NewStorageByURL(s.mainContext, s.conf.Endpoint(), s.conf.EndpointURL.Scheme, info.MetaData["token"], info.MetaData["workspace"]) if err != nil { - s.logger.Error("can not create storage by url") - return nil, err + s.logger.Error("can not create storage by url", zap.Error(err)) + return nil, errors.Wrapf(err, "can not create storage") } - var contentUploader = uploader.New(s.mainContext, storage, opts) + stream.storage = stg + // if storage supports multipart, initialize raw upload + if ms, ok := stg.(storage.MultipartStorage); ok { + multipart, err := NewMultipartUpload(s.mainContext, ms, info, meta.ContentType) + if err != nil { + s.logger.Error("multipart upload failed", zap.Error(err)) + return nil, errors.Wrapf(err, "multipart upload failed") + } + stream.multipart = multipart + } + // uploader for processed outputs + var contentUploader = uploader.New(s.mainContext, stg, opts) stream.contentUploader = contentUploader } @@ -127,10 +157,7 @@ func (s *StreamCoordinator) NewUpload(ctx context.Context, info handler.FileInfo go func() { stream.commandGroup.Wait() - if scaling != nil { - atomic.AddInt32(&s.activeScalling, -1) - } - s.logger.Debug("returned capacity", zap.Int64("capacity", cost)) + atomic.AddInt32(&s.activeTranscoding, -1) close(stream.done) }() @@ -142,26 +169,26 @@ func (s *StreamCoordinator) NewUpload(ctx context.Context, info handler.FileInfo // GetUpload returns current a worker based on upload id func (s *StreamCoordinator) GetUpload(ctx context.Context, id string) (upload handler.Upload, err error) { + logger := s.logger.With(zap.String("func", "GetUpload")).With(zap.String("id", id)) + if v, ok := s.streams.Load(id); ok { - s.logger.Debug("GetUpload: found stream by id", zap.String("id", id)) + logger.Debug("found stream") var w = v.(*Stream) s.manageTimeout(w) return w, nil } - s.logger.Debug("GetUpload: stream not found", zap.String("id", id)) - return nil, errors.New("bad id") + + logger.Warn("stream not found") + return nil, fmt.Errorf("stream not found: %v", id) } // AsTerminatableUpload returns tusd handler.TerminatableUpload func (s *StreamCoordinator) AsTerminatableUpload(upload handler.Upload) handler.TerminatableUpload { - var worker = upload.(*Stream) - s.logger.Debug("AsTerminatableUpload") - return worker + return upload.(*Stream) } // AsLengthDeclarableUpload returns tusd handler.LengthDeclarableUpload func (s *StreamCoordinator) AsLengthDeclarableUpload(upload handler.Upload) handler.LengthDeclarableUpload { - s.logger.Debug("AsLengthDeclarableUpload") return upload.(*Stream) } @@ -191,3 +218,23 @@ func (s *StreamCoordinator) manageTimeout(w *Stream) { } }() } + +func extractCodec(mimeType string) string { + codecRegex := regexp.MustCompile(`codecs["\s=]+([^",\s]+)`) + matches := codecRegex.FindStringSubmatch(mimeType) + codec := "unknown" + if len(matches) > 1 { + codec = matches[1] + } + + return codec +} + +func extractContentType(mimeType string) string { + contentType := "video/mp4" + parts := strings.Split(mimeType, ";") + if parts[0] != "" { + contentType = strings.TrimSpace(parts[0]) + } + return contentType +} diff --git a/internal/pkg/mediaconvert/multipart.go b/internal/pkg/mediaconvert/multipart.go new file mode 100644 index 0000000000..f29e35ef7f --- /dev/null +++ b/internal/pkg/mediaconvert/multipart.go @@ -0,0 +1,188 @@ +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package mediaconvert provides types and functions for video transcoding +package mediaconvert + +import ( + "bytes" + "context" + "time" + + "github.com/pkg/errors" + + "github.com/hcengineering/stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/storage" + "github.com/tus/tusd/v2/pkg/handler" + "go.uber.org/zap" +) + +// minimum size for multipart parts (backend requires >=5MiB for all but last part) +const minPartSize = 5 * 1024 * 1024 + +type MultipartUpload struct { + logger *zap.Logger + buffer *bytes.Buffer + info handler.FileInfo + + storage storage.MultipartStorage + objectName string + uploadID string + parts []storage.MultipartPart + nextPartNum int + + terminated bool + completed bool + + bytesWritten int64 + bytesUploaded int64 +} + +func NewMultipartUpload( + ctx context.Context, + multipartStorage storage.MultipartStorage, + info handler.FileInfo, + contentType string, +) (*MultipartUpload, error) { + objectName := info.ID + uploadID, err := multipartStorage.MultipartUploadStart(ctx, objectName, contentType) + if err != nil { + return nil, errors.Wrap(err, "failed to initialize multipart upload") + } + + logger := log.FromContext(ctx).With(zap.String("multipart", "upload"), zap.String("uploadID", uploadID)) + + return &MultipartUpload{ + logger: logger, + buffer: bytes.NewBuffer(nil), + info: info, + storage: multipartStorage, + objectName: objectName, + uploadID: uploadID, + parts: make([]storage.MultipartPart, 0), + nextPartNum: 1, + }, nil +} + +// Write writes chunk of data to the storage +func (w *MultipartUpload) Write(ctx context.Context, data []byte) error { + if w.terminated || w.completed { + return errors.New("upload already terminated or completed") + } + + if err := ctx.Err(); err != nil { + return err + } + + _, err := w.buffer.Write(data) + if err != nil { + return errors.Wrap(err, "failed to write to buffer") + } + w.bytesWritten += int64(len(data)) + + // flush parts of at least minPartSize + for w.buffer.Len() >= minPartSize { + partNum := w.nextPartNum + partData := w.buffer.Next(minPartSize) + + if err := ctx.Err(); err != nil { + return err + } + + part, err := w.storage.MultipartUploadPart(ctx, w.objectName, w.uploadID, partNum, partData) + if err != nil { + w.logger.Error("multipart upload part failed", zap.Error(err), zap.Int("partNumber", partNum)) + return errors.Wrap(err, "failed to upload part") + } + + w.bytesUploaded += int64(len(partData)) + w.parts = append(w.parts, *part) + w.nextPartNum++ + } + + return nil +} + +// Terminate cancels the upload +func (w *MultipartUpload) Terminate(ctx context.Context) error { + if w.terminated || w.completed { + return nil + } + w.terminated = true + + w.logger.Debug("terminating multipart upload", zap.Int("parts", len(w.parts))) + + // create new context in case the main context is cancelled + cancelCtx := ctx + if ctx.Err() != nil { + var cancel context.CancelFunc + cancelCtx, cancel = context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + } + + if err := w.storage.MultipartUploadCancel(cancelCtx, w.objectName, w.uploadID); err != nil { + w.logger.Error("multipart upload cancel failed", zap.Error(err)) + return errors.Wrap(err, "failed to cancel multipart upload") + } + + w.logger.Debug("multipart upload terminated", zap.Int("parts", len(w.parts))) + + return nil +} + +// Complete uploads last bytes and completes the upload +func (w *MultipartUpload) Complete(ctx context.Context) error { + if w.completed { + return nil + } + + if w.terminated { + return errors.New("cannot complete terminated upload") + } + + w.logger.Debug("finishing multipart upload", zap.Int("parts", len(w.parts))) + + // flush any remaining data as last part + if w.buffer.Len() > 0 { + partNum := w.nextPartNum + lastData := w.buffer.Bytes() + + part, err := w.storage.MultipartUploadPart(ctx, w.objectName, w.uploadID, partNum, lastData) + if err != nil { + w.logger.Error("multipart upload last part failed", zap.Error(err), zap.Int("partNumber", partNum)) + return errors.Wrap(err, "failed to upload last part") + } + + w.bytesUploaded += int64(len(lastData)) + w.parts = append(w.parts, *part) + } + + if len(w.parts) == 0 { + w.logger.Warn("cannot complete upload with no parts") + return errors.New("cannot complete upload with no parts") + } + + if err := w.storage.MultipartUploadComplete(ctx, w.objectName, w.uploadID, w.parts); err != nil { + w.logger.Error("multipart upload complete failed", zap.Error(err)) + return errors.Wrap(err, "failed to complete multipart upload") + } + + w.completed = true + w.logger.Info( + "multipart upload completed", + zap.Int64("bytesUploaded", w.bytesUploaded), + zap.Int64("bytesWritten", w.bytesWritten), + ) + + return nil +} diff --git a/internal/pkg/mediaconvert/scheduler.go b/internal/pkg/mediaconvert/scheduler.go index da3190276b..2f3e2f6f55 100644 --- a/internal/pkg/mediaconvert/scheduler.go +++ b/internal/pkg/mediaconvert/scheduler.go @@ -21,13 +21,13 @@ import ( "os" "os/exec" "path/filepath" + "strings" "time" "github.com/google/uuid" "github.com/hcengineering/stream/internal/pkg/config" "github.com/hcengineering/stream/internal/pkg/log" "github.com/hcengineering/stream/internal/pkg/manifest" - "github.com/hcengineering/stream/internal/pkg/resconv" "github.com/hcengineering/stream/internal/pkg/storage" "github.com/hcengineering/stream/internal/pkg/token" "github.com/hcengineering/stream/internal/pkg/uploader" @@ -94,7 +94,7 @@ func (p *Scheduler) start() { close(p.taskCh) }() - for range p.cfg.MaxParallelScalingCount { + for range p.cfg.MaxParallelTranscodingCount { go func() { for task := range p.taskCh { p.processTask(p.ctx, task) @@ -185,17 +185,24 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) { logger.Debug("video stream found", zap.String("codec", videoStream.CodecName), zap.Int("width", videoStream.Width), zap.Int("height", videoStream.Height)) - var res = fmt.Sprintf("%v:%v", videoStream.Width, videoStream.Height) - var codec = videoStream.CodecName - var level = resconv.Level(res) + meta := VideoMeta{ + Width: videoStream.Width, + Height: videoStream.Height, + Codec: videoStream.CodecName, + ContentType: stat.Type, + } + + var profiles = DefaultTranscodingProfiles(meta) + var opts = Options{ - Input: sourceFilePath, - OutputDir: p.cfg.OutputDir, - Level: level, - Transcode: !IsHLSSupportedVideoCodec(codec), - ScalingLevels: append(resconv.SubLevels(res), level), - UploadID: task.ID, - Threads: p.cfg.MaxThreadCount, + Input: sourceFilePath, + OutputDir: p.cfg.OutputDir, + UploadID: task.ID, + Threads: p.cfg.MaxThreadCount, + Profiles: profiles, + // Level: level, + // Transcode: !IsHLSSupportedVideoCodec(codec), + // ScalingLevels: append(resconv.SubLevels(res), level), } logger.Debug("phase 5: start async upload process") @@ -210,7 +217,7 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) { SourceFile: sourceFilePath, }) - err = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID) + err = manifest.GenerateHLSPlaylist(profiles, 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) @@ -223,8 +230,7 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) { var argsSlice = [][]string{ BuildThumbnailCommand(&opts), - BuildRawVideoCommand(&opts), - BuildScalingVideoCommand(&opts), + BuildVideoCommand(&opts), } var cmds []*exec.Cmd @@ -295,6 +301,11 @@ func IsHLSSupportedVideoCodec(codec string) bool { case "h264", "h265": return true default: + if strings.HasPrefix(codec, "avc1") { + return true + } else if strings.HasPrefix(codec, "av1") { + return true + } return false } } diff --git a/internal/pkg/mediaconvert/strategy.go b/internal/pkg/mediaconvert/strategy.go new file mode 100644 index 0000000000..387db9de3f --- /dev/null +++ b/internal/pkg/mediaconvert/strategy.go @@ -0,0 +1,67 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package mediaconvert + +import ( + "fmt" + + "github.com/hcengineering/stream/internal/pkg/profile" + "github.com/hcengineering/stream/internal/pkg/resconv" +) + +type VideoMeta struct { + Width int + Height int + Codec string + ContentType string +} + +// DefaultTranscodingProfiles uses original resolution and two more resolutions +func DefaultTranscodingProfiles(meta VideoMeta) []profile.VideoProfile { + var profiles = make([]profile.VideoProfile, 0) + + var res = fmt.Sprintf("%v:%v", meta.Width, meta.Height) + var sublevels = resconv.SubLevels(res) + + if IsHLSSupportedVideoCodec(meta.Codec) { + profile := profile.MakeProfileOriginal(meta.Width, meta.Height) + profiles = append(profiles, profile) + } else { + profile := profile.MakeProfileOriginalT(meta.Width, meta.Height) + profiles = append(profiles, profile) + } + + for _, level := range sublevels { + if profile, ok := profile.GetProfileByName(level); ok { + profiles = append(profiles, profile) + } + } + + return profiles +} + +// FastTranscodingProfiles uses fastest possible video profile +func FastTranscodingProfiles(meta VideoMeta) []profile.VideoProfile { + if IsHLSSupportedVideoCodec(meta.Codec) { + return []profile.VideoProfile{ + profile.MakeProfileOriginal(meta.Width, meta.Height), + } + } + + return []profile.VideoProfile{ + profile.Profile360p, + } +} diff --git a/internal/pkg/mediaconvert/stream.go b/internal/pkg/mediaconvert/stream.go index e41042d2e0..135ce98997 100644 --- a/internal/pkg/mediaconvert/stream.go +++ b/internal/pkg/mediaconvert/stream.go @@ -11,44 +11,69 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package mediaconvert provides types and functions for video trnascoding +// Package mediaconvert provides types and functions for video transcoding package mediaconvert import ( "context" "io" + "os/exec" "sync" "github.com/pkg/errors" "github.com/hcengineering/stream/internal/pkg/manifest" "github.com/hcengineering/stream/internal/pkg/sharedpipe" + "github.com/hcengineering/stream/internal/pkg/storage" "github.com/hcengineering/stream/internal/pkg/uploader" "github.com/tus/tusd/v2/pkg/handler" "go.uber.org/zap" ) -// Stream manages client's input and transcodes it based on the passsed configuration +// Stream manages client's input and transcodes it based on the passed configuration type Stream struct { contentUploader uploader.Uploader logger *zap.Logger info handler.FileInfo writer *sharedpipe.Writer reader *sharedpipe.Reader + storage storage.Storage + multipart *MultipartUpload commandGroup sync.WaitGroup done chan struct{} } -// WriteChunk calls when client sends a chunk of raw data +var _ handler.Upload = (*Stream)(nil) +var _ handler.ConcatableUpload = (*Stream)(nil) +var _ handler.TerminatableUpload = (*Stream)(nil) +var _ handler.LengthDeclarableUpload = (*Stream)(nil) + +// WriteChunk is called when client sends a chunk of raw data func (w *Stream) WriteChunk(ctx context.Context, _ int64, src io.Reader) (int64, error) { w.logger.Debug("Write Chunk start", zap.Int64("offset", w.info.Offset)) - var bytes, err = io.ReadAll(src) - _, _ = w.writer.Write(bytes) - var n = int64(len(bytes)) + data, err := io.ReadAll(src) + if err != nil { + return 0, err + } + // write into pipeline for transcoding + written, err := w.writer.Write(data) + if err != nil { + return int64(written), err + } + + n := int64(len(data)) w.info.Offset += n - w.logger.Debug("Write Chunk end", zap.Int64("offset", w.info.Offset), zap.Error(err)) - return n, err + + if w.multipart != nil { + if writeErr := w.multipart.Write(ctx, data); writeErr != nil { + w.logger.Error("multipart upload part failed", zap.Error(writeErr)) + return n, writeErr + } + } + + w.logger.Debug("write chunk end", zap.Int64("offset", w.info.Offset), zap.Error(err)) + return n, nil } // DeclareLength sets length of the video input @@ -73,14 +98,40 @@ func (w *Stream) GetReader(ctx context.Context) (io.ReadCloser, error) { // Terminate calls when upload has failed func (w *Stream) Terminate(ctx context.Context) error { - w.logger.Debug("Terminating...") + w.logger.Debug("terminate upload") + + // Close the writer first to signal EOF to all readers + if err := w.writer.Close(); err != nil { + w.logger.Error("failed to close writer", zap.Error(err)) + return err + } + + var wg sync.WaitGroup + + // cancel upload if in progress if w.contentUploader != nil { + wg.Add(1) go func() { + defer wg.Done() w.commandGroup.Wait() w.contentUploader.Cancel() }() } - return w.writer.Close() + + // cancel multipart upload if in progress + if w.multipart != nil { + wg.Add(1) + go func() { + defer wg.Done() + if err := w.multipart.Terminate(ctx); err != nil { + w.logger.Error("multipart upload cancel failed", zap.Error(err)) + } + }() + } + + wg.Wait() + + return nil } // ConcatUploads calls when upload resumed after fail @@ -94,16 +145,56 @@ func (w *Stream) ConcatUploads(ctx context.Context, partialUploads []handler.Upl // FinishUpload calls when upload finished without errors on the client side func (w *Stream) FinishUpload(ctx context.Context) error { - w.logger.Debug("finishing upload...") + w.logger.Debug("finish upload") + + // Close the writer first to signal EOF to all readers + if err := w.writer.Close(); err != nil { + w.logger.Error("failed to close writer", zap.Error(err)) + return err + } + + var wg sync.WaitGroup if w.contentUploader != nil { + wg.Add(1) go func() { + defer wg.Done() w.commandGroup.Wait() w.contentUploader.Stop() }() } - return w.writer.Close() + // finalize raw multipart stream if supported + if w.multipart != nil { + wg.Add(1) + go func() { + defer wg.Done() + if err := w.multipart.Complete(ctx); err != nil { + w.logger.Error("multipart upload complete failed", zap.Error(err)) + return + } + + if metaProvider, ok := w.storage.(storage.MetaProvider); ok { + metaErr := metaProvider.PatchMeta( + ctx, + w.info.ID, + &storage.Metadata{ + "hls": map[string]any{ + "source": w.info.ID + "_master.m3u8", + "thumbnail": w.info.ID + ".jpg", + }, + }, + ) + if metaErr != nil { + w.logger.Error("can not patch the source file", zap.Error(metaErr)) + } + } + }() + } + + wg.Wait() + + return nil } // AsConcatableUpload returns tusd handler.ConcatableUpload @@ -115,48 +206,39 @@ func (s *StreamCoordinator) AsConcatableUpload(upload handler.Upload) handler.Co func (w *Stream) start(ctx context.Context, options *Options) error { defer w.logger.Debug("start done") w.reader = w.writer.Transpile() - if err := manifest.GenerateHLSPlaylist(append(options.ScalingLevels, options.Level), options.OutputDir, options.UploadID); err != nil { + if err := manifest.GenerateHLSPlaylist(options.Profiles, options.OutputDir, options.UploadID); err != nil { return err } + var argsSlice = [][]string{ + BuildThumbnailCommand(options), + BuildVideoCommand(options), + } + + var cmds []*exec.Cmd + for idx, args := range argsSlice { + reader := w.reader + if idx > 0 { + reader = w.writer.Transpile() + } + + cmd, cmdErr := newFfmpegCommand(ctx, reader, args) + if cmdErr != nil { + w.logger.Error("can not create a new command", zap.Error(cmdErr), zap.Strings("args", args)) + return errors.Wrapf(cmdErr, "can not create a new command") + } + cmds = append(cmds, cmd) + } + w.commandGroup.Add(1) go func() { defer w.commandGroup.Done() - var logger = w.logger.With(zap.String("command", "raw")) - defer logger.Debug("done") - - var args = BuildRawVideoCommand(options) - var convertSourceCommand, err = newFfmpegCommand(ctx, w.reader, args) - if err != nil { - logger.Debug("can not start", zap.Error(err)) - } - err = convertSourceCommand.Run() - if err != nil { - logger.Debug("finished with error", zap.Error(err)) + executor := NewCommandExecutor(ctx) + if execErr := executor.Execute(cmds); execErr != nil { + w.logger.Error("can not execute command", zap.Error(execErr)) } }() - if len(options.ScalingLevels) > 0 { - w.commandGroup.Add(1) - var scalingCommandReader = w.writer.Transpile() - - go func() { - defer w.commandGroup.Done() - var logger = w.logger.With(zap.String("command", "scaling")) - defer logger.Debug("done") - - var args = BuildScalingVideoCommand(options) - var convertSourceCommand, err = newFfmpegCommand(ctx, scalingCommandReader, args) - if err != nil { - logger.Debug("can not start", zap.Error(err)) - } - err = convertSourceCommand.Run() - if err != nil { - logger.Debug("finished with error", zap.Error(err)) - } - }() - } - go w.contentUploader.Start() return nil diff --git a/internal/pkg/mediaconvert/transcoder.go b/internal/pkg/mediaconvert/transcoder.go index f85d034532..1972f171dd 100644 --- a/internal/pkg/mediaconvert/transcoder.go +++ b/internal/pkg/mediaconvert/transcoder.go @@ -26,7 +26,6 @@ import ( "github.com/hcengineering/stream/internal/pkg/config" "github.com/hcengineering/stream/internal/pkg/log" "github.com/hcengineering/stream/internal/pkg/manifest" - "github.com/hcengineering/stream/internal/pkg/resconv" "github.com/hcengineering/stream/internal/pkg/storage" "github.com/hcengineering/stream/internal/pkg/token" "github.com/hcengineering/stream/internal/pkg/uploader" @@ -119,7 +118,7 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (*TaskResult, er videoStream := probe.FirstVideoStream() if videoStream == nil { logger.Error("no video stream found", zap.String("filepath", sourceFilePath)) - return nil, errors.Wrapf(err, "no video stream found") + return nil, fmt.Errorf("no video stream found") } logger.Debug("video stream found", zap.String("codec", videoStream.CodecName), zap.Int("width", videoStream.Width), zap.Int("height", videoStream.Height)) @@ -129,19 +128,22 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (*TaskResult, er logger.Info("no audio stream found", zap.String("filepath", sourceFilePath)) } - var res = fmt.Sprintf("%v:%v", videoStream.Width, videoStream.Height) - var codec = videoStream.CodecName - var level = resconv.Level(res) - var sublevels = resconv.SubLevels(res) + meta := VideoMeta{ + Width: videoStream.Width, + Height: videoStream.Height, + Codec: videoStream.CodecName, + ContentType: stat.Type, + } + + var profiles = DefaultTranscodingProfiles(meta) + var opts = Options{ - Input: sourceFilePath, - OutputDir: p.cfg.OutputDir, - Level: level, - LogLevel: LogLevel(p.cfg.LogLevel), - Transcode: !IsHLSSupportedVideoCodec(codec), - ScalingLevels: append(sublevels, level), - UploadID: task.ID, - Threads: p.cfg.MaxThreadCount, + Input: sourceFilePath, + OutputDir: p.cfg.OutputDir, + LogLevel: LogLevel(p.cfg.LogLevel), + Profiles: profiles, + UploadID: task.ID, + Threads: p.cfg.MaxThreadCount, } logger.Debug("phase 5: start async upload process") @@ -156,7 +158,7 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (*TaskResult, er SourceFile: sourceFilePath, }) - err = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID) + err = manifest.GenerateHLSPlaylist(profiles, 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)) return nil, errors.Wrapf(err, "can not generate hls playlist") @@ -168,8 +170,7 @@ func (p *Transcoder) Transcode(ctx context.Context, task *Task) (*TaskResult, er var argsSlice = [][]string{ BuildThumbnailCommand(&opts), - BuildRawVideoCommand(&opts), - BuildScalingVideoCommand(&opts), + BuildVideoCommand(&opts), } var cmds []*exec.Cmd diff --git a/internal/pkg/profile/profile.go b/internal/pkg/profile/profile.go new file mode 100644 index 0000000000..7716ef6914 --- /dev/null +++ b/internal/pkg/profile/profile.go @@ -0,0 +1,185 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +// Package profile provides video profiles +package profile + +import ( + "fmt" + + "github.com/hcengineering/stream/internal/pkg/resconv" +) + +// VideoProfile represents a video profile +type VideoProfile struct { + Name string + Width int // 0 means keep original + Height int // 0 means keep original + Bandwidth int + Scale bool + + // Codec settings + VideoCodec string + AudioCodec string + + // Advanced encoding settings + CRF int // Constant Rate Factor (0-51 for x264) +} + +// profileOriginal is a profile for original video without transcoding +var profileOriginal = VideoProfile{ + Name: "orig", + Scale: false, + VideoCodec: "copy", + AudioCodec: "copy", + CRF: 23, +} + +// profileOriginalT is a profile for transcoding video in the original resolution +var profileOriginalT = VideoProfile{ + Name: "orig", + Scale: false, + VideoCodec: "libx264", + AudioCodec: "aac", + CRF: 23, +} + +// Profile360p is a profile for transcoding video in 360p +var Profile360p = VideoProfile{ + Name: "360p", + Scale: true, + Width: 640, + Height: 360, + Bandwidth: 500000, + VideoCodec: "libx264", + AudioCodec: "aac", + CRF: 28, +} + +// Profile480p is a profile for transcoding video in 480p +var Profile480p = VideoProfile{ + Name: "480p", + Scale: true, + Width: 854, + Height: 480, + Bandwidth: 2000000, + VideoCodec: "libx264", + AudioCodec: "aac", + CRF: 27, +} + +// Profile720p is a profile for transcoding video in 720p +var Profile720p = VideoProfile{ + Name: "720p", + Scale: true, + Width: 1280, + Height: 720, + Bandwidth: 5000000, + VideoCodec: "libx264", + AudioCodec: "aac", + CRF: 25, +} + +// Profile1080p is a profile for transcoding video in 1080p +var Profile1080p = VideoProfile{ + Name: "1080p", + Scale: true, + Width: 1920, + Height: 1080, + Bandwidth: 8000000, + VideoCodec: "libx264", + AudioCodec: "aac", + CRF: 23, +} + +// Profile1440p is a profile for transcoding video in 1440p +var Profile1440p = VideoProfile{ + Name: "1440p", + Scale: true, + Width: 2560, + Height: 1440, + Bandwidth: 12000000, + VideoCodec: "libx264", + AudioCodec: "aac", + CRF: 23, +} + +// Profile2160p is a profile for transcoding video in 2160p +var Profile2160p = VideoProfile{ + Name: "2160p", + Scale: true, + Width: 3840, + Height: 2160, + Bandwidth: 25000000, + VideoCodec: "libx264", // Consider libx265 + AudioCodec: "aac", + CRF: 22, +} + +// Profile4320p is a profile for transcoding video in 360p +var Profile4320p = VideoProfile{ + Name: "4320p", + Scale: true, + Width: 7680, + Height: 4320, + Bandwidth: 50000000, + VideoCodec: "libx264", // Consider libx265 + AudioCodec: "aac", + CRF: 22, +} + +var Profiles = map[string]VideoProfile{ + "360p": Profile360p, + "480p": Profile480p, + "720p": Profile720p, + "1080p": Profile1080p, + "1440p": Profile1440p, + "2160p": Profile2160p, + "4320p": Profile4320p, +} + +func MakeProfileOriginal(width, height int) VideoProfile { + resolution := fmt.Sprintf("%v:%v", width, height) + level := resconv.Level(resolution) + bandwidth := resconv.Bandwidth(level) + + profile := profileOriginal + //profile.Name = level + profile.Width = width + profile.Height = height + profile.Bandwidth = bandwidth + + return profile +} + +func MakeProfileOriginalT(width, height int) VideoProfile { + resolution := fmt.Sprintf("%v:%v", width, height) + level := resconv.Level(resolution) + bandwidth := resconv.Bandwidth(level) + + profile := profileOriginalT + //profile.Name = level + profile.Width = width + profile.Height = height + profile.Bandwidth = bandwidth + + return profile +} + +// GetProfileByName returns a VideoProfile by name +func GetProfileByName(name string) (VideoProfile, bool) { + profile, ok := Profiles[name] + return profile, ok +} diff --git a/internal/pkg/profile/profile_test.go b/internal/pkg/profile/profile_test.go new file mode 100644 index 0000000000..a4ea1448fd --- /dev/null +++ b/internal/pkg/profile/profile_test.go @@ -0,0 +1,173 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package profile_test + +import ( + "testing" + + "github.com/hcengineering/stream/internal/pkg/profile" + "github.com/stretchr/testify/assert" +) + +func TestGetProfileByName(t *testing.T) { + tests := []struct { + name string + expected profile.VideoProfile + }{ + { + name: "360p", + expected: profile.Profile360p, + }, + { + name: "480p", + expected: profile.Profile480p, + }, + { + name: "720p", + expected: profile.Profile720p, + }, + { + name: "1080p", + expected: profile.Profile1080p, + }, + { + name: "1440p", + expected: profile.Profile1440p, + }, + { + name: "2160p", + expected: profile.Profile2160p, + }, + { + name: "4320p", + expected: profile.Profile4320p, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + profile, ok := profile.GetProfileByName(tt.name) + assert.True(t, ok) + assert.Equal(t, tt.expected, profile) + }) + } +} + +func TestGetProfileByName_Failure(t *testing.T) { + tests := []struct { + name string + }{ + { + name: "foo", + }, + { + name: "original", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, ok := profile.GetProfileByName(tt.name) + assert.False(t, ok) + }) + } +} + +func TestMakeProfileOriginal(t *testing.T) { + tests := []struct { + name string + width int + height int + expected profile.VideoProfile + }{ + { + name: "480p", + width: 640, + height: 480, + expected: profile.VideoProfile{ + Name: "orig", + VideoCodec: "copy", + AudioCodec: "copy", + Width: 640, + Height: 480, + Bandwidth: 2000000, + }, + }, + { + name: "1440p", + width: 2000, + height: 1200, + expected: profile.VideoProfile{ + Name: "orig", + VideoCodec: "copy", + AudioCodec: "copy", + Width: 2000, + Height: 1200, + Bandwidth: 8000000, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + profile := profile.MakeProfileOriginal(tt.width, tt.height) + assert.Equal(t, tt.expected, profile) + }) + } +} + +func TestMakeProfileOriginalT(t *testing.T) { + tests := []struct { + name string + width int + height int + expected profile.VideoProfile + }{ + { + name: "720p", + width: 1280, + height: 720, + expected: profile.VideoProfile{ + Name: "orig", + VideoCodec: "libx264", + AudioCodec: "aac", + Width: 1280, + Height: 720, + Bandwidth: 5000000, + }, + }, + { + name: "2160p", + width: 3840, + height: 2160, + expected: profile.VideoProfile{ + Name: "orig", + VideoCodec: "libx264", + AudioCodec: "aac", + Width: 3840, + Height: 2160, + Bandwidth: 25000000, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + profile := profile.MakeProfileOriginalT(tt.width, tt.height) + assert.Equal(t, tt.expected, profile) + }) + } +} diff --git a/internal/pkg/queue/queue.go b/internal/pkg/queue/queue.go index cb5c674519..d9f6a54208 100644 --- a/internal/pkg/queue/queue.go +++ b/internal/pkg/queue/queue.go @@ -17,6 +17,7 @@ package queue import ( "context" "encoding/json" + "fmt" "time" "github.com/hcengineering/stream/internal/pkg/log" @@ -168,7 +169,12 @@ func NewProducer(ctx context.Context, options ProducerOptions) Producer { // Send sends a message to the queue topic func (p *TProducer) Send(ctx context.Context, workspaceID string, data any) error { + if data == nil { + return fmt.Errorf("event data is empty") + } + value, err := json.Marshal(data) + if err != nil { return err } diff --git a/internal/pkg/queue/worker.go b/internal/pkg/queue/worker.go index 7c9dae02b4..60def900d5 100644 --- a/internal/pkg/queue/worker.go +++ b/internal/pkg/queue/worker.go @@ -111,7 +111,7 @@ func (w *Worker) processMessage(ctx context.Context, msg kafka.Message, logger * transcoder := mediaconvert.NewTranscoder(ctx, w.cfg) res, err := transcoder.Transcode(ctx, &task) - if err == nil { + if res != nil { result := TranscodeResult{ BlobID: req.BlobID, WorkspaceUUID: req.WorkspaceUUID, diff --git a/internal/pkg/storage/datalake.go b/internal/pkg/storage/datalake.go index 95c8755569..4000168aca 100644 --- a/internal/pkg/storage/datalake.go +++ b/internal/pkg/storage/datalake.go @@ -21,8 +21,10 @@ import ( "io" "mime/multipart" "net/textproto" + "net/url" "os" "path/filepath" + "strconv" "strings" "time" @@ -376,6 +378,140 @@ func (d *DatalakeStorage) SetParent(ctx context.Context, filename, parent string return nil } +func (d *DatalakeStorage) MultipartUploadStart(ctx context.Context, objectName, contentType string) (string, error) { + var logger = d.logger.With(zap.String("workspace", d.workspace), zap.String("objectName", objectName)) + url := fmt.Sprintf("%v/upload/multipart/%v/%v", d.baseURL, d.workspace, objectName) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(url) + req.Header.SetMethod(fasthttp.MethodPost) + req.Header.Add("Authorization", "Bearer "+d.token) + req.Header.SetContentType(contentType) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + logRequestError(logger, err, "request failed", resp) + return "", err + } + + if err := okResponse(resp); err != nil { + logRequestError(logger, err, "bad status code", resp) + return "", err + } + + var result struct { + UploadID string `json:"uploadId"` + } + err := json.Unmarshal(resp.Body(), &result) + + return result.UploadID, err +} + +func (d *DatalakeStorage) MultipartUploadPart(ctx context.Context, objectName, uploadID string, partNumber int, data []byte) (*MultipartPart, error) { + var logger = d.logger.With(zap.String("workspace", d.workspace), zap.String("uploadID", uploadID), zap.Int("partNumber", partNumber)) + params := url.Values{} + params.Add("uploadId", uploadID) + params.Add("partNumber", strconv.Itoa(partNumber)) + url := fmt.Sprintf("%v/upload/multipart/%v/%v/part?%v", d.baseURL, d.workspace, objectName, params.Encode()) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(url) + req.Header.SetMethod(fasthttp.MethodPut) + req.Header.Add("Authorization", "Bearer "+d.token) + req.Header.SetContentType("application/octet-stream") + req.Header.SetContentLength(len(data)) + req.SetBody(data) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + logRequestError(logger, err, "request failed", resp) + return nil, err + } + + if err := okResponse(resp); err != nil { + logRequestError(logger, err, "bad status code", resp) + return nil, err + } + + var part MultipartPart + err := json.Unmarshal(resp.Body(), &part) + + return &part, err +} + +func (d *DatalakeStorage) MultipartUploadComplete(ctx context.Context, objectName, uploadID string, parts []MultipartPart) error { + var logger = d.logger.With(zap.String("workspace", d.workspace), zap.String("uploadID", uploadID), zap.String("objectName", objectName)) + params := url.Values{} + params.Add("uploadId", uploadID) + url := fmt.Sprintf("%v/upload/multipart/%v/%v/complete?%v", d.baseURL, d.workspace, objectName, params.Encode()) + + body, err := json.Marshal(map[string]any{ + "parts": parts, + }) + + if err != nil { + logger.Debug("can not encode body", zap.Error(err)) + return err + } + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(url) + req.Header.SetMethod(fasthttp.MethodPost) + req.Header.Add("Authorization", "Bearer "+d.token) + req.Header.SetContentType("application/json") + req.SetBody(body) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + logRequestError(logger, err, "request failed", resp) + return err + } + + if err := okResponse(resp); err != nil { + logRequestError(logger, err, "bad status code", resp) + return err + } + + return nil +} + +func (d *DatalakeStorage) MultipartUploadCancel(ctx context.Context, objectName, uploadID string) error { + var logger = d.logger.With(zap.String("workspace", d.workspace), zap.String("uploadID", uploadID)) + params := url.Values{} + params.Add("uploadId", uploadID) + url := fmt.Sprintf("%v/upload/multipart/%v/%v/abort?%v", d.baseURL, d.workspace, objectName, params.Encode()) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(url) + req.Header.SetMethod(fasthttp.MethodPost) + req.Header.Add("Authorization", "Bearer "+d.token) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + logRequestError(logger, err, "request failed", resp) + return err + } + + if err := okResponse(resp); err != nil { + logRequestError(logger, err, "bad status code", resp) + return err + } + + return nil +} + func okResponse(res *fasthttp.Response) error { var statusOK = res.StatusCode() >= 200 && res.StatusCode() < 300 @@ -397,4 +533,5 @@ func logRequestError(logger *zap.Logger, err error, msg string, res *fasthttp.Re } var _ Storage = (*DatalakeStorage)(nil) +var _ MultipartStorage = (*DatalakeStorage)(nil) var _ MetaProvider = (*DatalakeStorage)(nil) diff --git a/internal/pkg/storage/storage.go b/internal/pkg/storage/storage.go index 0feb587825..78d28cdf13 100644 --- a/internal/pkg/storage/storage.go +++ b/internal/pkg/storage/storage.go @@ -46,6 +46,20 @@ type Storage interface { SetParent(ctx context.Context, fileName string, parentName string) error } +// MultipartPart represents uploaded multipart part +type MultipartPart struct { + PartNumber int `json:"partNumber"` + ETag string `json:"etag"` +} + +// MultipartStorage represents multipart-based storage +type MultipartStorage interface { + MultipartUploadStart(ctx context.Context, objectName, contentType string) (string, error) + MultipartUploadPart(ctx context.Context, objectName, uploadID string, partNumber int, data []byte) (*MultipartPart, error) + MultipartUploadComplete(ctx context.Context, objectName, uploadID string, parts []MultipartPart) error + MultipartUploadCancel(ctx context.Context, objectName, uploadID string) error +} + // NewStorageByURL creates a new storage based on the type from the url scheme, for example "datalake://my-datalake-endpoint" func NewStorageByURL(ctx context.Context, u *url.URL, storageType, token, workspace string) (Storage, error) { if workspace == "" {