WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-08-17 18:05:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: fast live recording

Browse Source 
Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>
This commit is contained in:
Alexander Onnikov
2025-06-26 23:26:26 +07:00
parent 9b74c1bc15
commit 0cb756d440
18 changed files with 1169 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/pkg/api/v1/transcoding/handler.go
+1 -1
View File
@@ -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),
internal/pkg/config/config.go
+9 -9
View File
@@ -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"`
internal/pkg/manifest/hls.go
+6 -7
View File
@@ -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
}
internal/pkg/manifest/hls_test.go
+14 -6
View File
@@ -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)
}
internal/pkg/mediaconvert/command.go
+68 -85
View File
@@ -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
}
internal/pkg/mediaconvert/command_test.go
+47 -51
View File
@@ -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, " "))
}
internal/pkg/mediaconvert/coordinator.go
+82 -35
View File
@@ -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
}
internal/pkg/mediaconvert/multipart.go
+188
View File
@@ -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
}
internal/pkg/mediaconvert/scheduler.go
+26 -15
View File
@@ -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
}
}
internal/pkg/mediaconvert/strategy.go
+67
View File
@@ -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,
}
}
internal/pkg/mediaconvert/stream.go
+127 -45
View File
@@ -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
internal/pkg/mediaconvert/transcoder.go
+18 -17
View File
@@ -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
internal/pkg/profile/profile.go
+185
View File
@@ -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
}
internal/pkg/profile/profile_test.go
+173
View File
@@ -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)
})
}
}
internal/pkg/queue/queue.go
+6
View File
@@ -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
}
internal/pkg/queue/worker.go
+1 -1
View File
@@ -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,
internal/pkg/storage/datalake.go
+137
View File
@@ -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)
internal/pkg/storage/storage.go
+14
View File
@@ -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 == "" {