From 5342edacf638d28ebb19259594268db1af535396 Mon Sep 17 00:00:00 2001 From: denis-tingaikin Date: Sun, 16 Mar 2025 01:00:35 +0300 Subject: [PATCH 1/3] add possible to schedule transcoding tasks Signed-off-by: denis-tingaikin --- .github/workflows/docker-push.yaml | 4 +- .golangci.yaml | 4 +- Dockerfile | 18 +- README.md | 100 +++-- cmd/huly-stream/main.go | 104 ----- cmd/stream/main.go | 71 ++++ go.mod | 12 +- go.sum | 18 +- internal/pkg/api/v1/recording/handler.go | 82 ++++ internal/pkg/api/v1/transcoding/handler.go | 66 +++ internal/pkg/config/config.go | 28 +- internal/pkg/log/zap.go | 4 +- internal/pkg/manifest/hls.go | 2 +- internal/pkg/manifest/hls_test.go | 2 +- internal/pkg/pprof/pprof.go | 3 +- internal/pkg/resconv/resconv_test.go | 2 +- .../pkg/sharedpipe/shared_pipe_bench_test.go | 18 +- internal/pkg/storage/datalake.go | 254 ++++++++++++ internal/pkg/{uploader => storage}/s3.go | 89 ++-- internal/pkg/storage/storage.go | 55 +++ internal/pkg/token/token.go | 77 ++++ .../postpone_test.go => token/token_test.go} | 37 +- .../{transcoding => transcoder}/command.go | 14 +- .../command_test.go | 16 +- internal/pkg/transcoder/coordinator.go | 190 +++++++++ internal/pkg/transcoder/queue.go | 234 +++++++++++ .../worker.go => transcoder/stream.go} | 63 +-- internal/pkg/transcoding/limiter.go | 78 ---- internal/pkg/transcoding/limiter_test.go | 88 ---- internal/pkg/transcoding/scheduler.go | 167 -------- internal/pkg/uploader/datalake.go | 130 ------ internal/pkg/uploader/options.go | 14 +- internal/pkg/uploader/postpone.go | 48 --- internal/pkg/uploader/uploader.go | 381 +++++++++++------- 34 files changed, 1523 insertions(+), 950 deletions(-) delete mode 100644 cmd/huly-stream/main.go create mode 100644 cmd/stream/main.go create mode 100644 internal/pkg/api/v1/recording/handler.go create mode 100644 internal/pkg/api/v1/transcoding/handler.go create mode 100644 internal/pkg/storage/datalake.go rename internal/pkg/{uploader => storage}/s3.go (61%) create mode 100644 internal/pkg/storage/storage.go create mode 100644 internal/pkg/token/token.go rename internal/pkg/{uploader/postpone_test.go => token/token_test.go} (50%) rename internal/pkg/{transcoding => transcoder}/command.go (93%) rename internal/pkg/{transcoding => transcoder}/command_test.go (56%) create mode 100644 internal/pkg/transcoder/coordinator.go create mode 100644 internal/pkg/transcoder/queue.go rename internal/pkg/{transcoding/worker.go => transcoder/stream.go} (72%) delete mode 100644 internal/pkg/transcoding/limiter.go delete mode 100644 internal/pkg/transcoding/limiter_test.go delete mode 100644 internal/pkg/transcoding/scheduler.go delete mode 100644 internal/pkg/uploader/datalake.go delete mode 100644 internal/pkg/uploader/postpone.go diff --git a/.github/workflows/docker-push.yaml b/.github/workflows/docker-push.yaml index ad67bcee7c..47baea039f 100644 --- a/.github/workflows/docker-push.yaml +++ b/.github/workflows/docker-push.yaml @@ -29,7 +29,7 @@ jobs: id: metaci uses: docker/metadata-action@v3 with: - images: hardcoreeng/huly-stream + images: hardcoreeng/stream:${{ inputs.version }} tags: | type=ref,event=pr type=sha,prefix= @@ -41,4 +41,4 @@ jobs: context: . platforms: linux/amd64,linux/arm64 push: true - tags: hardcoreeng/huly-stream:${{ inputs.version }} + tags: hardcoreeng/stream:${{ inputs.version }} diff --git a/.golangci.yaml b/.golangci.yaml index 0099487e6f..a5335cf208 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -61,8 +61,8 @@ linters-settings: dupl: threshold: 150 funlen: - lines: 120 - statements: 60 + lines: 140 + statements: 80 goconst: min-len: 2 min-occurrences: 2 diff --git a/Dockerfile b/Dockerfile index 0b4dca5d82..4304349055 100644 --- a/Dockerfile +++ b/Dockerfile @@ -11,7 +11,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -FROM --platform=$BUILDPLATFORM golang:1.23.5 AS builder +FROM --platform=linux/amd64 golang:1.24.1 AS builder ENV GO111MODULE=on ENV CGO_ENABLED=0 ENV GOBIN=/bin @@ -19,18 +19,18 @@ ARG BUILDARCH=amd64 COPY . ./ -RUN set -xe && GOOS=$TARGETOS GOARCH=$TARGETARCH go build -o /go/bin/huly-stream ./cmd/huly-stream +RUN set -xe && GOOS=$TARGETOS GOARCH=$TARGETARCH go build -o /go/bin/stream ./cmd/stream -FROM alpine +FROM --platform=linux/amd64 alpine RUN set -xe && apk add --no-cache ffmpeg RUN apk add --no-cache ca-certificates jq bash \ - && addgroup -g 1000 huly-stream \ - && adduser -u 1000 -G huly-stream -s /bin/sh -D huly-stream \ - && chown huly-stream:huly-stream /. -COPY --from=builder /go/bin/huly-stream /huly-stream + && addgroup -g 1000 stream \ + && adduser -u 1000 -G stream -s /bin/sh -D stream \ + && chown stream:stream /. +COPY --from=builder /go/bin/stream /stream EXPOSE 1080 -USER huly-stream +USER stream -ENTRYPOINT ["/huly-stream"] \ No newline at end of file +ENTRYPOINT ["/stream"] \ No newline at end of file diff --git a/README.md b/README.md index f977167dba..529b764313 100644 --- a/README.md +++ b/README.md @@ -1,91 +1,96 @@ -# Huly Stream +# Stream [![X (formerly Twitter) Follow](https://img.shields.io/twitter/follow/huly_io?style=for-the-badge)](https://x.com/huly_io) ![GitHub License](https://img.shields.io/github/license/hcengineering/platform?style=for-the-badge) ## About -The Huly Stream high-performance HTTP-based transcoding service. Huly-stream is built around the **TUS protocol**, enabling reliable, resumable file uploads and downloads. Designed for seamless and consistent media processing,it supports advanced transcoding features with robust integration options. +The Stream is high-performance HTTP-based transcoding service. *Stream* supports **TUS protocol**, enabling reliable, +resumable transcodings. Designed for seamless and consistent media processing,it supports advanced transcoding features +with robust integration options. --- ## Features ### TUS Protocol Support -- **Resumable transcoding**: Leveraging the TUS protocol, Huly-stream ensures reliable and efficient stream processing. -### Input Support +- **Resumable transcoding**: Leveraging the TUS protocol, *Stream* ensures reliable and efficient transcoding bucket + processing. + +#### Input Support + - **Supported Input Formats**: - - `mp4` - - `webm` + - `mp4` + - `webm` + +#### Output Options -### Output Options - **Supported Output Formats**: - - `aac` - - `hls` + - `hls` + +#### Upload options -### Upload options -- **TUS Upload**: Resumable file uploads via TUS protocol. - **s3 Upload**: Direct upload to S3 storage. - **datalake Upload**: Upload to datalake storage. -### Key Functionalities +#### Key Functionalities + - **Live transcoing with minimal upload time**: Transcoding results are going to be avaible after stream completion. - **Transcoding Cancelation**: Cancel or pause ongoing transcoding in real-time. - **Transcoding Resumption**: Resume incomplete transcoding tasks efficiently. ---- +### Transcoding scheduling ## Installation ### Prerequisites + - [Go](https://golang.org/dl/) (v1.23+ recommended) - [ffmpeg](https://www.ffmpeg.org/download.html) (ensure it’s installed and available in your system's PATH) ### Steps 1. Install dependencies: - ```bash - go mod tidy - ``` + +```bash +go mod tidy +``` 2. Build the service: - ```bash - docker build . -t hcengineering/huly-stream:latest - ``` +```bash +docker build . -t hcengineering/sream:latest +``` --- ## Configuraiton ### App env configuraiton + The following environment variables can be used: + ``` -KEY TYPE DEFAULT REQUIRED DESCRIPTION -STREAM_SECRET_TOKEN String secret token for authorize requests -STREAM_LOG_LEVEL String debug sets log level for the application -STREAM_PPROF_ENABLED True or False false starts profile server on localhost:6060 if true -STREAM_INSECURE True or False false ignores authorization check if true -STREAM_SERVE_URL String 0.0.0.0:1080 app listen url -STREAM_ENDPOINT_URL URL S3 or Datalake endpoint, example: s3://my-ip-address, datalake://my-ip-address -STREAM_MAX_CAPACITY Integer 6220800 represents the amount of maximum possible capacity for the transcoding. The default value is 1920 * 1080 * 3. -STREAM_MAX_THREADS Integer 4 means upper bound for the transcoing provider. -STREAM_OUTPUT_DIR String /tmp/transcoing/ path to the directory with transcoding result. -STREAM_REMOVE_CONTENT_ON_UPLOAD True or False true deletes all content when content delivered if true -STREAM_UPLOAD_RAW_CONTENT True or False false uploads content in raw quality to the endpoint if true +KEY TYPE DEFAULT STREAM_LOG_LEVEL String debug sets log level for the application +STREAM_SERVER_SECRET String server secret required to generate and verify tokens +STREAM_PPROF_ENABLED True or False true starts profile server on localhost:6060 if true +STREAM_INSECURE True or False false ignores authorization check if true +STREAM_SERVE_URL String 0.0.0.0:1080 listen on url +STREAM_ENDPOINT_URL URL s3://127.0.0.1:9000 S3 or Datalake endpoint, example: s3://my-ip-address, datalake://my-ip-address +STREAM_MAX_PARALLEL_SCALING_COUNT Integer 2 how much parallel scaling can be processed +STREAM_MAX_THREAD_COUNT Integer 4 max number of threads for transcoder +STREAM_OUTPUT_DIR String /tmp/transcoing/ path to the directory with tra ``` ### Metadata -**resolutions:** if passed, set the resolution for the output, for example, 'resolutions: 1920:1080, 1280:720.' +**resolution:** if passed, set the resolution for the output, for example, 'resolution: 1920:1080'. **token:** must be provided to be authorized in the Huly's datalake service. **workspace:** required for uploading content to the datalake storage. - - #### S3 Env configuration if you're working with S3 storage type, these envs must be provided: @@ -94,16 +99,35 @@ if you're working with S3 storage type, these envs must be provided: ## Usage -The service exposes an HTTP API. Below are some examples of how to interact with it. +The service exposes an HTTP API. + +Below are some examples of how to interact with it. + +### Trnascode via TUS -### Upload a File for Transcoding via TUS ```bash -curl -X POST http://localhost:1080/transcoing \ +curl -X POST http://localhost:1080/recording \ -H "Tus-Resumable: 1.0.0" \ -H "Upload-Length: " \ --data-binary @path/to/your/file.mp4 ``` +Note: tus client is required, to play with a service locally you can use tus-js-client example +with [video](https://github.com/tus/tus-js-client/blob/main/demos/browser/video.html) + +### Schedule a transcoding + +```bash +curl -X POST http://localhost:1080/transcoding \ + -H "Content-Type: application/json" \ + -H "Authorization: Bearer " \ + -d '{ + "source": "", + "format": "hls", + "workspace": "test" + }' +``` + ## Contributing We welcome contributions! To get started: @@ -120,4 +144,4 @@ This project is licensed under the [MIT License](LICENSE). --- -Enjoy seamless transcoding with huly-stream! 🚀 \ No newline at end of file +Enjoy seamless transcoding with *Stream*! 🚀 \ No newline at end of file diff --git a/cmd/huly-stream/main.go b/cmd/huly-stream/main.go deleted file mode 100644 index d7c5d9690b..0000000000 --- a/cmd/huly-stream/main.go +++ /dev/null @@ -1,104 +0,0 @@ -// 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 main provides huly-stream entry point function -package main - -import ( - "context" - "net/http" - - "os" - "os/signal" - "syscall" - - "go.uber.org/zap" - "golang.org/x/exp/slog" - - "github.com/huly-stream/internal/pkg/config" - "github.com/huly-stream/internal/pkg/log" - "github.com/huly-stream/internal/pkg/pprof" - "github.com/huly-stream/internal/pkg/transcoding" - tusd "github.com/tus/tusd/v2/pkg/handler" -) - -const basePath = "/recording" - -func main() { - var ctx, cancel = signal.NotifyContext( - context.Background(), - os.Interrupt, - syscall.SIGHUP, - syscall.SIGTERM, - syscall.SIGQUIT, - ) - defer cancel() - ctx = log.WithLoggerFields(ctx) - - var logger = log.FromContext(ctx) - var conf = must(config.FromEnv()) - logger.Sugar().Debugf("provided config is %v", conf) - - logger.Sugar().Info(conf.Endpoint()) - - mustNoError(os.MkdirAll(conf.OutputDir, os.ModePerm)) - if conf.PprofEnabled { - go pprof.ListenAndServe(ctx, "localhost:6060") - } - scheduler := transcoding.NewScheduler(ctx, conf) - - tusComposer := tusd.NewStoreComposer() - tusComposer.UseCore(scheduler) - tusComposer.UseTerminater(scheduler) - tusComposer.UseConcater(scheduler) - tusComposer.UseLengthDeferrer(scheduler) - - var handler = must(tusd.NewHandler(tusd.Config{ - BasePath: basePath, - StoreComposer: tusComposer, - Logger: slog.New(slog.NewTextHandler(discardTextHandler{}, nil)), - })) - - http.Handle("/recording/", http.StripPrefix("/recording/", handler)) - http.Handle("/recording", http.StripPrefix("/recording", handler)) - - go func() { - logger.Info("started to listen") - defer logger.Info("server has finished") - // #nosec - var err = http.ListenAndServe(conf.ServeURL, nil) - if err != nil { - cancel() - logger.Debug("unable to listen", zap.Error(err)) - } - }() - - <-ctx.Done() -} - -type discardTextHandler struct{} - -func (discardTextHandler) Write([]byte) (int, error) { - return 0, nil -} - -func mustNoError(err error) { - if err != nil { - panic(err.Error()) - } -} - -func must[T any](val T, err error) T { - mustNoError(err) - return val -} diff --git a/cmd/stream/main.go b/cmd/stream/main.go new file mode 100644 index 0000000000..9b34801c02 --- /dev/null +++ b/cmd/stream/main.go @@ -0,0 +1,71 @@ +// 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 main provides huly-stream entry point function +package main + +import ( + "context" + "net/http" + + "os" + "os/signal" + "syscall" + + "go.uber.org/zap" + + "github.com/hcengineering/stream/internal/pkg/api/v1/recording" + "github.com/hcengineering/stream/internal/pkg/api/v1/transcoding" + "github.com/hcengineering/stream/internal/pkg/config" + "github.com/hcengineering/stream/internal/pkg/log" +) + +func main() { + var ctx, cancel = signal.NotifyContext( + context.Background(), + os.Interrupt, + syscall.SIGHUP, + syscall.SIGTERM, + syscall.SIGQUIT, + ) + defer cancel() + ctx = log.WithFields(ctx) + + var logger = log.FromContext(ctx) + var cfg, err = config.FromEnv() + if err != nil { + panic(err.Error()) + } + logger.Sugar().Debugf("parsed config is %v", cfg) + + var recordingHandler = recording.NewHandler(ctx, cfg) + var transcodingHandler = transcoding.NewHandler(ctx, cfg) + + http.Handle("/recording/", http.StripPrefix("/recording/", recordingHandler)) + http.Handle("/recording", http.StripPrefix("/recording", recordingHandler)) + http.Handle("/transcoding", http.StripPrefix("/transcoding", transcodingHandler)) + + go func() { + logger.Info("server started serving", zap.String("ServeURL", cfg.ServeURL)) + defer logger.Info("server finished") + + // #nosec + var err = http.ListenAndServe(cfg.ServeURL, nil) + if err != nil { + cancel() + logger.Debug("unable to listen", zap.Error(err)) + } + }() + + <-ctx.Done() +} diff --git a/go.mod b/go.mod index 3775572bbf..0671d43070 100644 --- a/go.mod +++ b/go.mod @@ -1,14 +1,13 @@ -module github.com/huly-stream +module github.com/hcengineering/stream -go 1.23.2 +go 1.24.1 require ( github.com/aws/aws-sdk-go-v2 v1.36.1 github.com/aws/aws-sdk-go-v2/config v1.29.6 github.com/aws/aws-sdk-go-v2/credentials v1.17.59 github.com/aws/aws-sdk-go-v2/service/s3 v1.77.0 - github.com/aws/smithy-go v1.22.3 - github.com/fsnotify/fsnotify v1.8.0 + github.com/golang-jwt/jwt/v5 v5.2.1 github.com/google/uuid v1.6.0 github.com/kelseyhightower/envconfig v1.4.0 github.com/pkg/errors v0.9.1 @@ -17,6 +16,8 @@ require ( github.com/valyala/fasthttp v1.59.0 go.uber.org/zap v1.27.0 golang.org/x/exp v0.0.0-20250215185904-eff6e970281f + gopkg.in/vansante/go-ffprobe.v2 v2.2.1 + k8s.io/utils v0.0.0-20241210054802-24370beab758 ) require ( @@ -34,11 +35,12 @@ require ( github.com/aws/aws-sdk-go-v2/service/sso v1.24.15 // indirect github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.14 // indirect github.com/aws/aws-sdk-go-v2/service/sts v1.33.14 // indirect + github.com/aws/smithy-go v1.22.3 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/klauspost/compress v1.17.11 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/valyala/bytebufferpool v1.0.0 // indirect go.uber.org/multierr v1.11.0 // indirect - golang.org/x/sys v0.30.0 // indirect + golang.org/x/net v0.37.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 1ffa95ca7c..6ef7c10d82 100644 --- a/go.sum +++ b/go.sum @@ -40,8 +40,8 @@ github.com/aws/smithy-go v1.22.3 h1:Z//5NuZCSW6R4PhQ93hShNbyBbn8BWCmCVCt+Q8Io5k= github.com/aws/smithy-go v1.22.3/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M= -github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= +github.com/golang-jwt/jwt/v5 v5.2.1 h1:OuVbFODueb089Lh128TAcimifWaLhJwVflnrgM17wHk= +github.com/golang-jwt/jwt/v5 v5.2.1/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc= github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= @@ -72,13 +72,15 @@ go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= golang.org/x/exp v0.0.0-20250215185904-eff6e970281f h1:oFMYAjX0867ZD2jcNiLBrI9BdpmEkvPyi5YrBGXbamg= golang.org/x/exp v0.0.0-20250215185904-eff6e970281f/go.mod h1:BHOTPb3L19zxehTsLoJXVaTktb06DFgmdW6Wb9s8jqk= -golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8= -golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk= -golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= -golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM= -golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY= +golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c= +golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8= +golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY= +golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/vansante/go-ffprobe.v2 v2.2.1 h1:sFV08OT1eZ1yroLCZVClIVd9YySgCh9eGjBWO0oRayI= +gopkg.in/vansante/go-ffprobe.v2 v2.2.1/go.mod h1:qF0AlAjk7Nqzqf3y333Ly+KxN3cKF2JqA3JT5ZheUGE= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +k8s.io/utils v0.0.0-20241210054802-24370beab758 h1:sdbE21q2nlQtFh65saZY+rRM6x6aJJI8IUa1AmH/qa0= +k8s.io/utils v0.0.0-20241210054802-24370beab758/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= diff --git a/internal/pkg/api/v1/recording/handler.go b/internal/pkg/api/v1/recording/handler.go new file mode 100644 index 0000000000..4282261f76 --- /dev/null +++ b/internal/pkg/api/v1/recording/handler.go @@ -0,0 +1,82 @@ +// 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 recording provides recording http handler. +package recording + +import ( + "context" + + "golang.org/x/exp/slog" + + "net/http" + "sync" + + "github.com/hcengineering/stream/internal/pkg/config" + "github.com/hcengineering/stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/transcoder" + "go.uber.org/zap" + + tusd "github.com/tus/tusd/v2/pkg/handler" +) + +type recordingHandler struct { + logger *zap.Logger + once sync.Once + cfg *config.Config + ctx context.Context + tusHandler http.Handler +} + +func (h *recordingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if !h.cfg.Insecure { + r.Header.Set("X-Forwarded-Proto", "https") + } + h.once.Do(h.initialize) + h.tusHandler.ServeHTTP(w, r) +} + +// NewHandler creates a new recording http handler, requires context and config. +func NewHandler(ctx context.Context, cfg *config.Config) http.Handler { + return &recordingHandler{ + logger: log.FromContext(ctx).With(zap.String("handler", "recording")), + cfg: cfg, + ctx: ctx, + } +} + +func (h *recordingHandler) initialize() { + scheduler := transcoder.NewStreamCoordinator(h.ctx, h.cfg) + + tusComposer := tusd.NewStoreComposer() + tusComposer.UseCore(scheduler) + tusComposer.UseTerminater(scheduler) + tusComposer.UseConcater(scheduler) + tusComposer.UseLengthDeferrer(scheduler) + + var tusHandler, err = tusd.NewHandler(tusd.Config{ + BasePath: "/recording", + StoreComposer: tusComposer, + RespectForwardedHeaders: true, + DisableDownload: true, + Cors: &tusd.DefaultCorsConfig, + NetworkTimeout: h.cfg.Timeout, + Logger: slog.Default(), + }) + + if err != nil { + panic(err.Error()) + } + + h.tusHandler = tusHandler +} diff --git a/internal/pkg/api/v1/transcoding/handler.go b/internal/pkg/api/v1/transcoding/handler.go new file mode 100644 index 0000000000..0391ea74da --- /dev/null +++ b/internal/pkg/api/v1/transcoding/handler.go @@ -0,0 +1,66 @@ +// 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 transcoding provides transcoding http handler. +package transcoding + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + + "github.com/hcengineering/stream/internal/pkg/config" + "github.com/hcengineering/stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/transcoder" + "go.uber.org/zap" +) + +type trascodeHandler struct { + taskQueue *transcoder.Scheduler + logger *zap.Logger +} + +func (t *trascodeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "" { + w.WriteHeader(http.StatusBadRequest) + _, _ = fmt.Fprint(w, "uri is not allowed") + return + } + + if r.Header.Get("Authorization") == "" { + w.WriteHeader(http.StatusUnauthorized) + _, _ = fmt.Fprint(w, "missed Authorization header") + return + } + + var decoder = json.NewDecoder(r.Body) + var task transcoder.Task + + if err := decoder.Decode(&task); err != nil { + w.WriteHeader(http.StatusBadRequest) + _, _ = fmt.Fprint(w, "can not decode request body") + return + } + + t.taskQueue.Schedule(&task) + w.WriteHeader(http.StatusOK) +} + +// NewHandler creates a new trnascoding http handler, requires context and config. +func NewHandler(ctx context.Context, cfg *config.Config) http.Handler { + return &trascodeHandler{ + taskQueue: transcoder.NewScheduler(ctx, cfg), + logger: log.FromContext(ctx).With(zap.String("handler", "transcoding")), + } +} diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 6a748d5bae..23fee3afc7 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -18,22 +18,24 @@ import ( "net/url" "time" + "github.com/pkg/errors" + "github.com/kelseyhightower/envconfig" ) // Config represents configuration for the huly-stream application. type Config struct { - SecretToken string `split_words:"true" desc:"secret token for authorize requests"` - LogLevel string `split_words:"true" default:"debug" desc:"sets log level for the application"` - PprofEnabled bool `default:"false" split_words:"true" desc:"starts profile server on localhost:6060 if true"` - Insecure bool `default:"false" desc:"ignores authorization check if true"` - ServeURL string `split_words:"true" desc:"app listen 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"` - AuthURL *url.URL `split_words:"true" desc:"url to auth the upload"` - MaxCapacity int64 `split_words:"true" default:"6220800" desc:"represents the amount of maximum possible capacity for the transcoding. The default value is 1920 * 1080 * 3."` - MaxThreads int `split_words:"true" default:"4" desc:"means upper bound for the transcoing provider."` - OutputDir string `split_words:"true" default:"/tmp/transcoing/" desc:"path to the directory with transcoding result."` - Timeout time.Duration `default:"5m" desc:"timeout for the upload"` + 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"` + + OutputDir string `split_words:"true" default:"/tmp/transcoing/" desc:"path to the directory with transcoding result."` + Timeout time.Duration `default:"5m" desc:"timeout for the upload"` } // FromEnv creates new Config from env @@ -52,6 +54,10 @@ func FromEnv() (*Config, error) { result.EndpointURL = nil } + if !result.Insecure && result.ServerSecret == "" { + return nil, errors.New("server secret must be provided for secure confgiuration") + } + return &result, nil } diff --git a/internal/pkg/log/zap.go b/internal/pkg/log/zap.go index 72e03ac379..b85fcfbe5b 100644 --- a/internal/pkg/log/zap.go +++ b/internal/pkg/log/zap.go @@ -22,8 +22,8 @@ import ( type contextKey struct{} -// WithLoggerFields createsa new context with zap.Logger and passed fields -func WithLoggerFields(ctx context.Context, fields ...zap.Field) context.Context { +// WithFields createsa new context with zap.Logger and passed fields +func WithFields(ctx context.Context, fields ...zap.Field) context.Context { var logger = FromContext(ctx) if logger == nil { var err error diff --git a/internal/pkg/manifest/hls.go b/internal/pkg/manifest/hls.go index d8dd7a5361..dcb421fe50 100644 --- a/internal/pkg/manifest/hls.go +++ b/internal/pkg/manifest/hls.go @@ -20,7 +20,7 @@ import ( "path/filepath" "strings" - "github.com/huly-stream/internal/pkg/resconv" + "github.com/hcengineering/stream/internal/pkg/resconv" ) // GenerateHLSPlaylist generates master file for master files for resolution levels diff --git a/internal/pkg/manifest/hls_test.go b/internal/pkg/manifest/hls_test.go index 624e3a27a9..a37c63b874 100644 --- a/internal/pkg/manifest/hls_test.go +++ b/internal/pkg/manifest/hls_test.go @@ -18,7 +18,7 @@ import ( "path/filepath" "testing" - "github.com/huly-stream/internal/pkg/manifest" + "github.com/hcengineering/stream/internal/pkg/manifest" "github.com/stretchr/testify/require" ) diff --git a/internal/pkg/pprof/pprof.go b/internal/pkg/pprof/pprof.go index e7a5eb0302..081e9603df 100644 --- a/internal/pkg/pprof/pprof.go +++ b/internal/pkg/pprof/pprof.go @@ -20,7 +20,7 @@ import ( "net/http/pprof" "time" - "github.com/huly-stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/log" "go.uber.org/zap" ) @@ -48,7 +48,6 @@ func ListenAndServe(ctx context.Context, listenOn string) { if err := server.ListenAndServe(); err != nil { log.FromContext(ctx).Debug("Failed to start profiler", zap.Error(err)) } - <-ctx.Done() _ = server.Close() } diff --git a/internal/pkg/resconv/resconv_test.go b/internal/pkg/resconv/resconv_test.go index 547fab0582..48fd291d57 100644 --- a/internal/pkg/resconv/resconv_test.go +++ b/internal/pkg/resconv/resconv_test.go @@ -16,7 +16,7 @@ package resconv_test import ( "testing" - "github.com/huly-stream/internal/pkg/resconv" + "github.com/hcengineering/stream/internal/pkg/resconv" "github.com/stretchr/testify/require" ) diff --git a/internal/pkg/sharedpipe/shared_pipe_bench_test.go b/internal/pkg/sharedpipe/shared_pipe_bench_test.go index 3633e1fdf4..cafe282dc0 100644 --- a/internal/pkg/sharedpipe/shared_pipe_bench_test.go +++ b/internal/pkg/sharedpipe/shared_pipe_bench_test.go @@ -196,36 +196,35 @@ func Test_Consistent(t *testing.T) { } // Benchmark_DefaultPipe-8 (4 b) 61956 19177 ns/op 48 B/op 1 allocs/op -// Benchmark_DefaultPipe-8 (8 mb) 22 49187741 ns/op 118 B/op 1 allocs/op +// Benchmark_DefaultPipe-8 (8 mb) 24 48471316 ns/op 257 B/op 1 allocs/op func Benchmark_DefaultPipe(b *testing.B) { var data [sendMessageSize]byte var buffer = make([]byte, len(data)) var readers []io.Reader var writers []io.Writer - for i := 0; i < readerCount; i++ { + for range readerCount { r, w := io.Pipe() readers = append(readers, r) writers = append(writers, w) } b.ReportAllocs() - b.ResetTimer() - for range b.N { + for b.Loop() { go func() { - for i := 0; i < readerCount; i++ { + for i := range readerCount { _, _ = writers[i].Write(data[:]) } }() - for i := 0; i < readerCount; i++ { + for i := range readerCount { _, _ = readers[i].Read(buffer) } } } // Benchmark_SharedPipe-8 (4 b) 161847 8131 ns/op 160 B/op 2 allocs/op -// Benchmark_SharedPipe-8 (8 mb) 69 15880031 ns/op 160 B/op 2 allocs/op +// Benchmark_SharedPipe-8 (8 mb) 75 15710658 ns/op 161 B/op 2 allocs/op func Benchmark_SharedPipe(b *testing.B) { var data [sendMessageSize]byte var buffer = make([]byte, len(data)) @@ -237,11 +236,10 @@ func Benchmark_SharedPipe(b *testing.B) { } b.ReportAllocs() - b.ResetTimer() - for range b.N { + for b.Loop() { _, _ = writer.Write(data[:]) - for i := 0; i < readerCount; i++ { + for i := range readerCount { _, _ = readers[i].Read(buffer) } } diff --git a/internal/pkg/storage/datalake.go b/internal/pkg/storage/datalake.go new file mode 100644 index 0000000000..abe2f69459 --- /dev/null +++ b/internal/pkg/storage/datalake.go @@ -0,0 +1,254 @@ +// 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 storage + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "mime/multipart" + "os" + "path/filepath" + + "github.com/hcengineering/stream/internal/pkg/log" + "github.com/pkg/errors" + "github.com/valyala/fasthttp" + "go.uber.org/zap" +) + +// DatalakeStorage represents datalake storage +type DatalakeStorage struct { + baseURL string + workspace string + token string + logger *zap.Logger + client fasthttp.Client +} + +// NewDatalakeStorage creates a new datalake client +func NewDatalakeStorage(ctx context.Context, baseURL, workspace, token string) Storage { + return &DatalakeStorage{ + baseURL: baseURL, + token: token, + workspace: workspace, + logger: log.FromContext(ctx).With(zap.String("storage", "datalake")), + } +} + +// PutFile uploads file to the datalake +func (d *DatalakeStorage) PutFile(ctx context.Context, fileName string) error { + // #nosec + file, err := os.Open(fileName) + if err != nil { + return err + } + defer func() { + _ = file.Close() + }() + + var objectKey = getObjectKey(fileName) + var logger = d.logger.With(zap.String("upload", d.workspace), zap.String("fileName", fileName)) + + logger.Debug("start") + + body := &bytes.Buffer{} + writer := multipart.NewWriter(body) + + part, err := writer.CreateFormFile("file", objectKey) + if err != nil { + return errors.Wrapf(err, "failed to create form file") + } + + _, err = io.Copy(part, file) + if err != nil { + return errors.Wrapf(err, "failed to copy file data") + } + + _ = writer.Close() + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + + res := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(res) + + req.SetRequestURI(d.baseURL + "/upload/form-data/" + d.workspace) + req.Header.SetMethod(fasthttp.MethodPost) + req.Header.Add("Authorization", "Bearer "+d.token) + req.Header.SetContentType(writer.FormDataContentType()) + req.SetBody(body.Bytes()) + + if err := d.client.Do(req, res); err != nil { + logger.Error("upload failed", zap.Error(err)) + return errors.Wrapf(err, "upload failed") + } + + logger.Debug("uploaded") + + return nil +} + +// DeleteFile deletes file from the datalake +func (d *DatalakeStorage) DeleteFile(ctx context.Context, fileName string) error { + var logger = d.logger.With(zap.String("delete", d.workspace), zap.String("fileName", fileName)) + logger.Debug("start") + + var objectKey = getObjectKey(fileName) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + + res := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(res) + + req.SetRequestURI(d.baseURL + "/blob/" + d.workspace + "/" + objectKey) + req.Header.SetMethod(fasthttp.MethodDelete) + req.Header.Add("Authorization", "Bearer "+d.token) + + if err := d.client.Do(req, res); err != nil { + logger.Error("delete failed", zap.Error(err)) + return errors.Wrapf(err, "delete failed") + } + + logger.Debug("deleted") + + return nil +} + +func getObjectKey(s string) string { + var _, objectKey = filepath.Split(s) + return objectKey +} + +// PatchMeta patches metadata for the object +func (d *DatalakeStorage) PatchMeta(ctx context.Context, filename string, md *Metadata) error { + var logger = d.logger.With(zap.String("patch meta", d.workspace), zap.String("fileName", filename)) + logger.Debug("start") + defer logger.Debug("finished") + + var objectKey = getObjectKey(filename) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(d.baseURL + "/meta/" + d.workspace + "/" + objectKey) + req.Header.SetMethod(fasthttp.MethodPatch) + req.Header.Add("Authorization", "Bearer "+d.token) + req.Header.SetContentType("application/json") + + b, err := json.Marshal(md) + + if err != nil { + return err + } + req.SetBody(b) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + return err + } + + if resp.StatusCode() != fasthttp.StatusOK { + var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode()) + logger.Debug("bad status code", zap.Error(err)) + return err + } + + fmt.Println(string(resp.Body())) + + return nil +} + +// GetMeta gets metadata related to the object +func (d *DatalakeStorage) GetMeta(ctx context.Context, filename string) (*Metadata, error) { + var logger = d.logger.With(zap.String("get meta", d.workspace), zap.String("fileName", filename)) + logger.Debug("start") + + var objectKey = getObjectKey(filename) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(d.baseURL + "/meta/" + d.workspace + "/" + objectKey) + req.Header.SetMethod(fasthttp.MethodGet) + req.Header.Add("Authorization", "Bearer "+d.token) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + return nil, err + } + + if resp.StatusCode() != fasthttp.StatusOK { + var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode()) + logger.Debug("bad status code", zap.Error(err)) + return nil, err + } + + var md Metadata + fmt.Println(string(resp.Body())) + var err = json.Unmarshal(resp.Body(), &md) + + return &md, err +} + +// GetFile gets file from the storage +func (d *DatalakeStorage) GetFile(ctx context.Context, filename, destination string) error { + var logger = d.logger.With(zap.String("get", d.workspace), zap.String("fileName", filename), zap.String("destination", destination)) + logger.Debug("start") + + var objectKey = getObjectKey(filename) + + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + req.SetRequestURI(d.baseURL + "/blob/" + d.workspace + "/" + objectKey) + req.Header.SetMethod(fasthttp.MethodGet) + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := d.client.Do(req, resp); err != nil { + return err + } + + // Check the response status code + if resp.StatusCode() != fasthttp.StatusOK { + var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode()) + logger.Debug("bad status code", zap.Error(err)) + return err + } + + // #nosec + file, err := os.Create(destination) + if err != nil { + logger.Debug("can't create a file", zap.Error(err)) + return err + } + defer func() { + _ = file.Close() + }() + if err := resp.BodyWriteTo(file); err != nil { + logger.Debug("can't write to file", zap.Error(err)) + return err + } + + logger.Debug("file downloaded successfully") + return nil +} + +var _ Storage = (*DatalakeStorage)(nil) +var _ MetaProvider = (*DatalakeStorage)(nil) diff --git a/internal/pkg/uploader/s3.go b/internal/pkg/storage/s3.go similarity index 61% rename from internal/pkg/uploader/s3.go rename to internal/pkg/storage/s3.go index d211e0347f..41da2eaaf1 100644 --- a/internal/pkg/uploader/s3.go +++ b/internal/pkg/storage/s3.go @@ -11,14 +11,13 @@ // See the License for the specific language governing permissions and // limitations under the License. -package uploader +// Package storage provdies simple storage interface for the remote storages. +package storage import ( "context" "fmt" - "github.com/pkg/errors" - "os" "path/filepath" "strings" @@ -28,8 +27,7 @@ import ( "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" "github.com/aws/aws-sdk-go-v2/service/s3" - "github.com/aws/smithy-go" - "github.com/huly-stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/log" "go.uber.org/zap" ) @@ -44,7 +42,6 @@ type S3Storage struct { func NewS3(ctx context.Context, endpoint, bucketName string) Storage { var accessKeyID = os.Getenv("AWS_ACCESS_KEY_ID") var accessKeySecret = os.Getenv("AWS_SECRET_ACCESS_KEY") - var logger = log.FromContext(ctx).With(zap.String("s3", "storage")) cfg, err := config.LoadDefaultConfig(ctx, config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(accessKeyID, accessKeySecret, "")), @@ -62,7 +59,7 @@ func NewS3(ctx context.Context, endpoint, bucketName string) Storage { return &S3Storage{ client: s3Client, bucketName: bucketName, - logger: logger, + logger: log.FromContext(ctx).With(zap.String("s3", "storage")), } } @@ -71,7 +68,7 @@ func getContentType(objectKey string) string { return "video/mp2t" } if strings.HasSuffix(objectKey, ".m3u8") { - return "application/x-mpegurl" + return "video/x-mpegurl" } return "application/octet-stream" } @@ -79,9 +76,10 @@ func getContentType(objectKey string) string { // DeleteFile deletes file from the s3 storage func (u *S3Storage) DeleteFile(ctx context.Context, fileName string) error { var _, objectKey = filepath.Split(fileName) - var logger = log.FromContext(ctx).With(zap.String("s3 delete", u.bucketName), zap.String("fileName", fileName)) + var logger = u.logger.With(zap.String("delete", u.bucketName), zap.String("fileName", fileName)) + + logger.Debug("start") - logger.Debug("start deleting") input := &s3.DeleteObjectInput{ Bucket: aws.String(u.bucketName), Key: aws.String(objectKey), @@ -91,15 +89,17 @@ func (u *S3Storage) DeleteFile(ctx context.Context, fileName string) error { if err != nil { return fmt.Errorf("failed to delete file from S3: %w", err) } - logger.Debug("file deleted") + + logger.Debug("deleted") return nil } -// UploadFile uploads file to the s3 storage -func (u *S3Storage) UploadFile(ctx context.Context, fileName string) error { +// PutFile uploads file to the s3 storage +func (u *S3Storage) PutFile(ctx context.Context, fileName string) error { var _, objectKey = filepath.Split(fileName) - var logger = log.FromContext(ctx).With(zap.String("s3 upload", u.bucketName), zap.String("fileName", fileName)) - logger.Debug("start upload file") + var logger = u.logger.With(zap.String("upload", u.bucketName), zap.String("fileName", fileName)) + + logger.Debug("start") // #nosec var file, err = os.Open(fileName) @@ -112,6 +112,7 @@ func (u *S3Storage) UploadFile(ctx context.Context, fileName string) error { defer func() { _ = file.Close() }() + _, err = u.client.PutObject(ctx, &s3.PutObjectInput{ Bucket: aws.String(u.bucketName), Key: aws.String(objectKey), @@ -120,23 +121,57 @@ func (u *S3Storage) UploadFile(ctx context.Context, fileName string) error { }) if err != nil { - var apiErr smithy.APIError - if errors.As(err, &apiErr) && apiErr.ErrorCode() == "EntityTooLarge" { - logger.Error("Error while uploading object. The object is too large." + - "To upload objects larger than 5GB, use the S3 console (160GB max)" + - "or the multipart upload API (5TB max).") - } else { - logger.Error("Couldn't upload file", zap.Error(err)) - } - return apiErr + logger.Error("couldn't upload file", zap.Error(err)) + return err } err = s3.NewObjectExistsWaiter(u.client).Wait( ctx, &s3.HeadObjectInput{Bucket: aws.String(u.bucketName), Key: aws.String(objectKey)}, time.Minute) + if err != nil { - logger.Debug("Failed attempt to wait for object to exist.") + logger.Error("Failed attempt to wait for object to exist") + return err } - logger.Debug("file has uploaded") - return err + logger.Debug("uploaded") + return nil +} + +// GetFile gets file from the storage and stores it to destination +func (u *S3Storage) GetFile(ctx context.Context, filename, dest string) error { + var logger = u.logger.With(zap.String("get", u.bucketName), zap.String("fileName", filename), zap.String("destination", dest)) + + var result, err = u.client.GetObject(ctx, &s3.GetObjectInput{ + Bucket: &u.bucketName, + Key: &filename, + }) + + if err != nil { + logger.Error("failed to get object", zap.Error(err)) + return err + } + defer func() { + _ = result.Body.Close() + }() + + // Create a local file to save the downloaded content + // #nosec + file, err := os.Create(dest) + if err != nil { + logger.Error("failed to create file", zap.Error(err)) + return err + } + defer func() { + _ = file.Close() + }() + // Copy the S3 object content to the local file + _, err = file.ReadFrom(result.Body) + if err != nil { + logger.Error("failed to write to file", zap.Error(err)) + return err + } + + logger.Debug("file downloaded successfully") + + return nil } diff --git a/internal/pkg/storage/storage.go b/internal/pkg/storage/storage.go new file mode 100644 index 0000000000..e45250284d --- /dev/null +++ b/internal/pkg/storage/storage.go @@ -0,0 +1,55 @@ +// 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 storage + +import ( + "context" + "net/url" + + "github.com/pkg/errors" +) + +// Metadata represents file's metadata +type Metadata map[string]any + +// MetaProvider provides simple api for working with files metadata +type MetaProvider interface { + GetMeta(ctx context.Context, filename string) (*Metadata, error) + PatchMeta(ctx context.Context, filename string, value *Metadata) error +} + +// Storage represents file-based storage +type Storage interface { + PutFile(ctx context.Context, fileName string) error + DeleteFile(ctx context.Context, fileName string) error + GetFile(ctx context.Context, fileName, destination string) error +} + +// 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, worksapce string) (Storage, error) { + if worksapce == "" { + return nil, errors.New("workspace is missed") + } + switch storageType { + case "datalake": + if token == "" { + return nil, errors.New("token is missed") + } + return NewDatalakeStorage(ctx, u.String(), worksapce, token), nil + case "s3": + return NewS3(ctx, u.String(), worksapce), nil + default: + return nil, errors.New("unknown scheme") + } +} diff --git a/internal/pkg/token/token.go b/internal/pkg/token/token.go new file mode 100644 index 0000000000..1f81a76484 --- /dev/null +++ b/internal/pkg/token/token.go @@ -0,0 +1,77 @@ +// 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 token provides functions to work with platform tokens +package token + +import ( + "fmt" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/google/uuid" +) + +// Token represents Claims for the platform token +type Token struct { + jwt.RegisteredClaims + Account string `json:"account"` + Workspace string `json:"workspace,omitempty"` + Extra map[string]interface{} `json:"extra,omitempty"` +} + +// NewToken creates a new platform token +func NewToken(serverSecret, workspace, issuer, audience string) (string, error) { + var res = Token{ + Account: uuid.NewString(), + Workspace: workspace, + RegisteredClaims: jwt.RegisteredClaims{ + Issuer: issuer, + Audience: jwt.ClaimStrings{audience}, + ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour * 12)), + }, + } + return res.Encode(serverSecret) +} + +// Encode signes a token object and converts it to string +func (t *Token) Encode(serverSecret string) (string, error) { + var token = jwt.NewWithClaims(jwt.SigningMethodHS256, t) + var tokenString, err = token.SignedString([]byte(serverSecret)) + + if err != nil { + return "", err + } + + return tokenString, nil +} + +// Decode decodes a token by a passed configuration +func Decode(secretKey, tokenString string) (*Token, error) { + token, err := jwt.ParseWithClaims(tokenString, &Token{}, func(token *jwt.Token) (interface{}, error) { + if _, ok := token.Method.(*jwt.SigningMethodHMAC); !ok { + return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"]) + } + return []byte(secretKey), nil + }) + + if err != nil { + return nil, fmt.Errorf("failed to parse token: %w", err) + } + + if claims, ok := token.Claims.(*Token); ok && token.Valid { + return claims, nil + } + + return nil, fmt.Errorf("invalid token: can't parse claims") +} diff --git a/internal/pkg/uploader/postpone_test.go b/internal/pkg/token/token_test.go similarity index 50% rename from internal/pkg/uploader/postpone_test.go rename to internal/pkg/token/token_test.go index 68758451e7..93d97bdb53 100644 --- a/internal/pkg/uploader/postpone_test.go +++ b/internal/pkg/token/token_test.go @@ -11,35 +11,30 @@ // See the License for the specific language governing permissions and // limitations under the License. -package uploader +package token_test import ( - "context" - "sync/atomic" "testing" "time" + "github.com/golang-jwt/jwt/v5" + "github.com/hcengineering/stream/internal/pkg/token" "github.com/stretchr/testify/require" ) -func Test_Postpone(t *testing.T) { - var u = uploader{ - postponeDuration: time.Second / 4, - } - var counter atomic.Int32 - u.postpone("1", func(context.Context) { counter.Add(1) }) - time.Sleep(time.Second / 8) - u.postpone("1", func(context.Context) { counter.Add(1) }) - time.Sleep(time.Second / 2) - require.Equal(t, int32(1), counter.Load()) - time.Sleep(time.Second / 2) - require.Equal(t, int32(1), counter.Load()) +func Test_GenerateSimpleServiceToken(t *testing.T) { + var _, err = token.NewToken("secret", "ws", "issuer", "aud") + require.NoError(t, err) } -func Test_WithoutPostpone(t *testing.T) { - var counter atomic.Int32 - var u uploader - u.postpone("1", func(context.Context) { counter.Add(1) }) - time.Sleep(time.Second / 10) - require.Equal(t, int32(1), counter.Load()) +func Test_ParseSimpleServiceToken(t *testing.T) { + const secret = "secret" + tokenString, err := token.NewToken(secret, "ws", "issuer", "aud") + require.NoError(t, err) + tok, err := token.Decode(secret, tokenString) + require.NoError(t, err) + require.Equal(t, tok.Issuer, "issuer") + require.Equal(t, tok.Audience, jwt.ClaimStrings{"aud"}) + require.Equal(t, tok.Workspace, "ws") + require.True(t, tok.ExpiresAt.After(time.Now())) } diff --git a/internal/pkg/transcoding/command.go b/internal/pkg/transcoder/command.go similarity index 93% rename from internal/pkg/transcoding/command.go rename to internal/pkg/transcoder/command.go index 0065fa3196..9cc65d9637 100644 --- a/internal/pkg/transcoding/command.go +++ b/internal/pkg/transcoder/command.go @@ -13,7 +13,7 @@ // limitations under the License. // -package transcoding +package transcoder import ( "context" @@ -25,13 +25,13 @@ import ( "github.com/pkg/errors" - "github.com/huly-stream/internal/pkg/log" - "github.com/huly-stream/internal/pkg/resconv" + "github.com/hcengineering/stream/internal/pkg/log" "go.uber.org/zap" ) // Options represents configuration for the ffmpeg command type Options struct { + Input string OuputDir string ScalingLevels []string Level string @@ -58,9 +58,8 @@ func newFfmpegCommand(ctx context.Context, in io.Reader, args []string) (*exec.C func buildCommonComamnd(opts *Options) []string { return []string{ - "-nostdin", "-threads", fmt.Sprint(opts.Threads), - "-i", "pipe:0", + "-i", opts.Input, } } @@ -79,8 +78,6 @@ func BuildRawVideoCommand(opts *Options) []string { return append(buildCommonComamnd(opts), "-c:v", "copy", - "-fps_mode", - "vfr", "-hls_time", "5", "-hls_list_size", "0", "-hls_segment_filename", filepath.Join(opts.OuputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", opts.Level)), @@ -90,9 +87,10 @@ func BuildRawVideoCommand(opts *Options) []string { // BuildScalingVideoCommand returns flags for ffmpeg for video scaling func BuildScalingVideoCommand(opts *Options) []string { var result = buildCommonComamnd(opts) + for _, level := range opts.ScalingLevels { result = append(result, - "-vf", "scale="+resconv.Resolution(level), + "-vf", "scale=-2:"+level[:len(level)-1], "-c:v", "libx264", "-preset", "veryfast", diff --git a/internal/pkg/transcoding/command_test.go b/internal/pkg/transcoder/command_test.go similarity index 56% rename from internal/pkg/transcoding/command_test.go rename to internal/pkg/transcoder/command_test.go index 719c38467a..db382a555e 100644 --- a/internal/pkg/transcoding/command_test.go +++ b/internal/pkg/transcoder/command_test.go @@ -11,39 +11,41 @@ // See the License for the specific language governing permissions and // limitations under the License. -package transcoding_test +package transcoder_test import ( "strings" "testing" - "github.com/huly-stream/internal/pkg/resconv" - "github.com/huly-stream/internal/pkg/transcoding" + "github.com/hcengineering/stream/internal/pkg/resconv" + "github.com/hcengineering/stream/internal/pkg/transcoder" "github.com/stretchr/testify/require" ) func Test_BuildVideoCommand_Scaling(t *testing.T) { - var scaleCommand = transcoding.BuildScalingVideoCommand(&transcoding.Options{ + var scaleCommand = transcoder.BuildScalingVideoCommand(&transcoder.Options{ OuputDir: "test", + Input: "pipe:0", UploadID: "1", Threads: 4, ScalingLevels: []string{"720p", "480p"}, }) - const expected = `-nostdin -threads 4 -i pipe:0 -vf scale=1280:720 -c:v libx264 -preset veryfast -crf 23 -g 60 -hls_time 5 -hls_list_size 0 -hls_segment_filename test/1/1_%03d_720p.ts test/1/1_720p_master.m3u8 -vf scale=640:480 -c:v libx264 -preset veryfast -crf 23 -g 60 -hls_time 5 -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8` + const expected = `-threads 4 -i pipe:0 -vf scale=-2:720 -c:v libx264 -preset veryfast -crf 23 -g 60 -hls_time 5 -hls_list_size 0 -hls_segment_filename test/1/1_%03d_720p.ts test/1/1_720p_master.m3u8 -vf scale=-2:480 -c:v libx264 -preset veryfast -crf 23 -g 60 -hls_time 5 -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(t *testing.T) { - var rawCommand = transcoding.BuildRawVideoCommand(&transcoding.Options{ + var rawCommand = transcoder.BuildRawVideoCommand(&transcoder.Options{ OuputDir: "test", + Input: "pipe:0", UploadID: "1", Threads: 4, Level: resconv.Level("651:490"), }) - const expected = `-nostdin -threads 4 -i pipe:0 -c:v copy -fps_mode vfr -hls_time 5 -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8` + const expected = `"-threads 4 -i pipe:0 -c:v copy -hls_time 5 -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, " ")) } diff --git a/internal/pkg/transcoder/coordinator.go b/internal/pkg/transcoder/coordinator.go new file mode 100644 index 0000000000..d0705cb0ce --- /dev/null +++ b/internal/pkg/transcoder/coordinator.go @@ -0,0 +1,190 @@ +// +// 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 transcoder + +import ( + "context" + "path/filepath" + "sync" + "sync/atomic" + "time" + + "github.com/pkg/errors" + + "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" + "github.com/tus/tusd/v2/pkg/handler" + "go.uber.org/zap" +) + +// StreamCoordinator represents manager for streams. It creates a new stream for a client and manages it's life cycle. +type StreamCoordinator struct { + conf *config.Config + uploadOptions uploader.Options + + activeScalling int32 + + mainContext context.Context + logger *zap.Logger + + streams sync.Map + cancels sync.Map +} + +// NewStreamCoordinator creates a new scheduler for transcode operations. +func NewStreamCoordinator(ctx context.Context, c *config.Config) *StreamCoordinator { + return &StreamCoordinator{ + conf: c, + uploadOptions: uploader.Options{ + RetryDelay: time.Millisecond * 100, + Timeout: c.Timeout, + WorkerCount: uint32(c.MaxThreadCount), + RetryCount: 5, + BufferSize: 128, + Dir: c.OutputDir, + }, + mainContext: ctx, + logger: log.FromContext(ctx).With(zap.String("Scheduler", c.OutputDir)), + } +} + +// NewUpload creates a new worker with passed parameters +func (s *StreamCoordinator) NewUpload(ctx context.Context, info handler.FileInfo) (handler.Upload, error) { + if info.ID == "" { + info.ID = uuid.NewString() + } + s.logger.Sugar().Debugf("stream: %v", info) + s.logger.Debug("NewUpload", zap.String("ID", info.ID)) + + var stream = &Stream{ + writer: sharedpipe.NewWriter(), + info: info, + logger: log.FromContext(s.mainContext).With(zap.String("worker", info.ID)), + 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) + s.logger.Debug("run out of resources for scaling") + scaling = nil + } + + var commandOptions = Options{ + Input: "pipe:0", + OuputDir: s.conf.OutputDir, + Threads: s.conf.MaxThreadCount, + UploadID: info.ID, + Level: level, + ScalingLevels: scaling, + } + + if s.conf.EndpointURL != nil { + s.logger.Sugar().Debugf("initializing uploader for %v", info) + 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"]) + if err != nil { + s.logger.Error("can not create storage by url") + return nil, err + } + var contentUploader = uploader.New(s.mainContext, storage, opts) + + stream.contentUploader = contentUploader + } + + s.streams.Store(stream.info.ID, stream) + if err := stream.start(s.mainContext, &commandOptions); err != nil { + return nil, err + } + + go func() { + stream.commandGroup.Wait() + if scaling != nil { + atomic.AddInt32(&s.activeScalling, -1) + } + s.logger.Debug("returned capacity", zap.Int64("capacity", cost)) + close(stream.done) + }() + + s.manageTimeout(stream) + + s.logger.Debug("NewUpload", zap.String("done", info.ID)) + return stream, nil +} + +// GetUpload returns current a worker based on upload id +func (s *StreamCoordinator) GetUpload(ctx context.Context, id string) (upload handler.Upload, err error) { + if v, ok := s.streams.Load(id); ok { + s.logger.Debug("GetUpload: found worker by id", zap.String("id", id)) + var w = v.(*Stream) + s.manageTimeout(w) + return w, nil + } + s.logger.Debug("GetUpload: worker not found", zap.String("id", id)) + return nil, errors.New("bad 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 +} + +// AsLengthDeclarableUpload returns tusd handler.LengthDeclarableUpload +func (s *StreamCoordinator) AsLengthDeclarableUpload(upload handler.Upload) handler.LengthDeclarableUpload { + s.logger.Debug("AsLengthDeclarableUpload") + return upload.(*Stream) +} + +func (s *StreamCoordinator) manageTimeout(w *Stream) { + var cancelCtx, cancel = context.WithCancel(context.Background()) + if v, ok := s.cancels.Load(w.info.ID); ok { + v.(context.CancelFunc)() + } + s.cancels.Store(w.info.ID, cancel) + go func() { + select { + case <-w.done: + w.logger.Debug("stream has finished") + s.cancels.Delete(w.info.ID) + return + case <-cancelCtx.Done(): + w.logger.Debug("stream timeout has refreshed") + return + case <-time.After(s.conf.Timeout): + w.logger.Error("stream timeout") + s.cancels.Delete(w.info.ID) + var terminateCtx, terminateCancel = context.WithTimeout(context.Background(), s.conf.Timeout) + defer terminateCancel() + _ = w.Terminate(terminateCtx) + } + }() +} diff --git a/internal/pkg/transcoder/queue.go b/internal/pkg/transcoder/queue.go new file mode 100644 index 0000000000..4a82ae107b --- /dev/null +++ b/internal/pkg/transcoder/queue.go @@ -0,0 +1,234 @@ +// +// 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 transcoder + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "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" + "go.uber.org/zap" + "gopkg.in/vansante/go-ffprobe.v2" +) + +// HLS represents metadata for transcoding result +type HLS struct { + Source string `json:"source"` +} + +// Task represents transcoding task +type Task struct { + ID string + Status string + Source string + Format string + Workspace string + Metadata map[string]string +} + +// Scheduler manages transcoding tasks by passed config +type Scheduler struct { + logger *zap.Logger + taskCh chan *Task + cfg *config.Config + ctx context.Context +} + +// Schedule schedules a task to transcode +func (p *Scheduler) Schedule(t *Task) { + t.ID = uuid.NewString() + t.Status = "planned" + + select { + case p.taskCh <- t: + p.logger.Sugar().Debugf("task %v is scheduled", t) + default: + p.logger.Error("task channel is full") + } +} + +// NewScheduler creates a new instance of transcoding task scheduler +func NewScheduler(ctx context.Context, cfg *config.Config) *Scheduler { + var p = &Scheduler{ + taskCh: make(chan *Task, 128), + cfg: cfg, + ctx: ctx, + logger: log.FromContext(ctx).With(zap.String("transcoding", "planner")), + } + + go p.start() + + return p +} + +func (p *Scheduler) start() { + go func() { + <-p.ctx.Done() + close(p.taskCh) + }() + + for range p.cfg.MaxParallelScalingCount { + go func() { + for task := range p.taskCh { + p.processTask(p.ctx, task) + } + }() + } +} + +func (p *Scheduler) processTask(ctx context.Context, task *Task) { + var logger = p.logger.With(zap.String("task-id", task.ID)) + + logger.Debug("start") + defer logger.Debug("finished") + + logger.Debug("phase 1: get a token") + var tokenString, err = token.NewToken(p.cfg.ServerSecret, task.Workspace, "stream", "datalake") + if err != nil { + logger.Error("can not create token", zap.Error(err)) + return + } + + logger.Debug("phase 2: preparing fs") + var destinationFolder = filepath.Join(p.cfg.OutputDir, task.ID) + var _, filename = filepath.Split(task.Source) + var sourceFilePath = filepath.Join(destinationFolder, filename) + _ = os.MkdirAll(destinationFolder, os.ModePerm) + + logger.Debug("phase 3: get the remote file") + + remoteStorage, err := storage.NewStorageByURL(ctx, p.cfg.Endpoint(), p.cfg.EndpointURL.Scheme, tokenString, task.Workspace) + + if err != nil { + logger.Error("can not create storage by url", zap.Error(err)) + _ = os.RemoveAll(destinationFolder) + return + } + + if err = remoteStorage.GetFile(ctx, task.Source, sourceFilePath); err != nil { + logger.Error("can not download a file", zap.Error(err)) + _ = os.RemoveAll(destinationFolder) + // TODO: reschedule + return + } + + logger.Debug("phase 4: prepare to transcode") + probe, err := ffprobe.ProbeURL(ctx, sourceFilePath) + if err != nil { + logger.Error("can not get probe for a file", zap.Error(err), zap.String("filepath", sourceFilePath)) + _ = os.RemoveAll(destinationFolder) + return + } + + var res = fmt.Sprintf("%v:%v", probe.FirstVideoStream().Width, probe.FirstVideoStream().Height) + var level = resconv.Level(res) + var opts = Options{ + Input: sourceFilePath, + OuputDir: p.cfg.OutputDir, + Level: level, + ScalingLevels: append(resconv.SubLevels(res), level), + UploadID: task.ID, + Threads: p.cfg.MaxThreadCount, + } + + logger.Debug("phase 5: start async upload process") + var uploader = uploader.New(ctx, remoteStorage, uploader.Options{ + Dir: destinationFolder, + WorkerCount: uint32(opts.Threads), + BufferSize: 128, + RetryCount: 5, + RetryDelay: time.Millisecond * 100, + Timeout: p.cfg.Timeout, + SourceFile: sourceFilePath, + }) + + go uploader.Start() + + logger.Debug("phase 6: start async transcode process") + var rawCommand, scaleCommand *exec.Cmd + + rawCommand, err = newFfmpegCommand(ctx, nil, BuildRawVideoCommand(&opts)) + if err != nil { + logger.Error("can not create ffmpeg command", zap.Error(err)) + go uploader.Cancel() + return + } + + scaleCommand, err = newFfmpegCommand(ctx, nil, BuildScalingVideoCommand(&opts)) + if err != nil { + logger.Error("can not create ffmpeg command", zap.Error(err)) + go uploader.Cancel() + return + } + + _ = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID) + + if err = rawCommand.Start(); err != nil { + logger.Error("can not run raw ffmpeg command", zap.Error(err)) + go uploader.Cancel() + return + } + + if err = scaleCommand.Start(); err != nil { + logger.Error("can not run scale ffmpeg command", zap.Error(err)) + go uploader.Cancel() + return + } + + logger.Debug("phase 7: wait for the result") + if err = scaleCommand.Wait(); err != nil { + logger.Error("can not scale ", zap.Error(err)) + go uploader.Cancel() + return + } + if err = rawCommand.Wait(); err != nil { + logger.Error("can not process raw", zap.Error(err)) + go uploader.Cancel() + return + } + + logger.Debug("phase 8: schedule cleanup") + go uploader.Stop() + + logger.Debug("phase 9: try to set metadata") + var resultURL = p.cfg.Endpoint().JoinPath("blob", task.Workspace, task.ID+"_master.m3u8") + + if metaProvider, ok := remoteStorage.(storage.MetaProvider); ok { + var hls = &HLS{Source: resultURL.String()} + logger.Debug("applying metadata", zap.Stringer("url", resultURL), zap.String("source", task.Source)) + err = metaProvider.PatchMeta( + ctx, + task.Source, + &storage.Metadata{ + "hls": hls, + }, + ) + if err != nil { + logger.Error("can not patch the source file", zap.Error(err)) + } + } +} diff --git a/internal/pkg/transcoding/worker.go b/internal/pkg/transcoder/stream.go similarity index 72% rename from internal/pkg/transcoding/worker.go rename to internal/pkg/transcoder/stream.go index c2aaca3d34..d59c12c09f 100644 --- a/internal/pkg/transcoding/worker.go +++ b/internal/pkg/transcoder/stream.go @@ -11,8 +11,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package transcoding provides objects and functions for video trnascoding -package transcoding +// Package transcoder provides types and functions for video trnascoding +package transcoder import ( "context" @@ -21,27 +21,27 @@ import ( "github.com/pkg/errors" - "github.com/huly-stream/internal/pkg/manifest" - "github.com/huly-stream/internal/pkg/sharedpipe" - "github.com/huly-stream/internal/pkg/uploader" + "github.com/hcengineering/stream/internal/pkg/manifest" + "github.com/hcengineering/stream/internal/pkg/sharedpipe" + "github.com/hcengineering/stream/internal/pkg/uploader" "github.com/tus/tusd/v2/pkg/handler" "go.uber.org/zap" ) -// Worker manages client's input and transcodes it based on the passsed configuration -type Worker struct { +// Stream manages client's input and transcodes it based on the passsed configuration +type Stream struct { contentUploader uploader.Uploader logger *zap.Logger info handler.FileInfo writer *sharedpipe.Writer reader *sharedpipe.Reader - wg sync.WaitGroup - done chan struct{} + commandGroup sync.WaitGroup + done chan struct{} } // WriteChunk calls when client sends a chunk of raw data -func (w *Worker) WriteChunk(ctx context.Context, _ int64, src io.Reader) (int64, error) { +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) @@ -52,7 +52,7 @@ func (w *Worker) WriteChunk(ctx context.Context, _ int64, src io.Reader) (int64, } // DeclareLength sets length of the video input -func (w *Worker) DeclareLength(ctx context.Context, length int64) error { +func (w *Stream) DeclareLength(ctx context.Context, length int64) error { w.info.Size = length w.info.SizeIsDeferred = false w.logger.Debug("DeclareLength", zap.Int64("size", length), zap.Bool("SizeIsDeferred", w.info.SizeIsDeferred)) @@ -60,65 +60,67 @@ func (w *Worker) DeclareLength(ctx context.Context, length int64) error { } // GetInfo returns info about transcoing status -func (w *Worker) GetInfo(ctx context.Context) (handler.FileInfo, error) { +func (w *Stream) GetInfo(ctx context.Context) (handler.FileInfo, error) { w.logger.Debug("GetInfo is executed") return w.info, nil } -// GetReader returns worker's bytes stream -func (w *Worker) GetReader(ctx context.Context) (io.ReadCloser, error) { +// GetReader returns Stream's bytes stream +func (w *Stream) GetReader(ctx context.Context) (io.ReadCloser, error) { w.logger.Debug("GetReader is executed, creating current reader...") return w.reader, nil } // Terminate calls when upload has failed -func (w *Worker) Terminate(ctx context.Context) error { +func (w *Stream) Terminate(ctx context.Context) error { w.logger.Debug("Terminating...") if w.contentUploader != nil { go func() { - w.wg.Wait() - w.contentUploader.Rollback() + w.commandGroup.Wait() + w.contentUploader.Cancel() }() } return w.writer.Close() } // ConcatUploads calls when upload resumed after fail -func (w *Worker) ConcatUploads(ctx context.Context, partialUploads []handler.Upload) error { +func (w *Stream) ConcatUploads(ctx context.Context, partialUploads []handler.Upload) error { w.logger.Debug("ConcatUploads was executed, it's not implemented") // - // TODO: load raw source from the Buckup bucket, terminate all workers with same ID and start process again. + // TODO: load raw source from the Buckup bucket, terminate all Streams with same ID and start process again. // return errors.New("not implemented") } // FinishUpload calls when upload finished without errors on the client side -func (w *Worker) FinishUpload(ctx context.Context) error { +func (w *Stream) FinishUpload(ctx context.Context) error { w.logger.Debug("finishing upload...") + if w.contentUploader != nil { go func() { - w.wg.Wait() - w.contentUploader.Terminate() + w.commandGroup.Wait() + w.contentUploader.Stop() }() } + return w.writer.Close() } // AsConcatableUpload returns tusd handler.ConcatableUpload -func (s *Scheduler) AsConcatableUpload(upload handler.Upload) handler.ConcatableUpload { +func (s *StreamCoordinator) AsConcatableUpload(upload handler.Upload) handler.ConcatableUpload { s.logger.Debug("AsConcatableUpload is executed") - return upload.(*Worker) + return upload.(*Stream) } -func (w *Worker) start(ctx context.Context, options *Options) error { +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.OuputDir, options.UploadID); err != nil { return err } - w.wg.Add(1) + w.commandGroup.Add(1) go func() { - defer w.wg.Done() + defer w.commandGroup.Done() var logger = w.logger.With(zap.String("command", "raw")) defer logger.Debug("done") @@ -134,10 +136,11 @@ func (w *Worker) start(ctx context.Context, options *Options) error { }() if len(options.ScalingLevels) > 0 { - w.wg.Add(1) + w.commandGroup.Add(1) var scalingCommandReader = w.writer.Transpile() + go func() { - defer w.wg.Done() + defer w.commandGroup.Done() var logger = w.logger.With(zap.String("command", "scaling")) defer logger.Debug("done") @@ -153,5 +156,7 @@ func (w *Worker) start(ctx context.Context, options *Options) error { }() } + go w.contentUploader.Start() + return nil } diff --git a/internal/pkg/transcoding/limiter.go b/internal/pkg/transcoding/limiter.go deleted file mode 100644 index f1fdf51d9f..0000000000 --- a/internal/pkg/transcoding/limiter.go +++ /dev/null @@ -1,78 +0,0 @@ -// 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 transcoding - -import "sync/atomic" - -// Limiter is a simple CAS data structure for managing resources. -type Limiter struct { - capacity int64 - maxCapacity int64 -} - -// NewLimiter creates a new limiter with the given initial capacity. -func NewLimiter(capacity int64) *Limiter { - return &Limiter{ - capacity: capacity, - maxCapacity: capacity, - } -} - -// TryConsume attempts to consume the specified amount of capacity. -// Returns true if successful, false otherwise. -func (l *Limiter) TryConsume(amount int64) bool { - if amount <= 0 { - return false - } - - for { - current := atomic.LoadInt64(&l.capacity) - if current < amount { - return false - } - updated := current - amount - if atomic.CompareAndSwapInt64(&l.capacity, current, updated) { - return true - } - } -} - -// ReturnCapacity adds the specified amount back to the limiter's capacity. -// Does not exceed the maximum capacity. -func (l *Limiter) ReturnCapacity(amount int64) { - if amount <= 0 { - return - } - - for { - current := atomic.LoadInt64(&l.capacity) - updated := current + amount - if updated > l.maxCapacity { - updated = l.maxCapacity - } - if atomic.CompareAndSwapInt64(&l.capacity, current, updated) { - break - } - } -} - -// GetCapacity retrieves the current capacity for debugging or monitoring purposes. -func (l *Limiter) GetCapacity() int64 { - return atomic.LoadInt64(&l.capacity) -} - -// GetMaxCapacity retrieves the maximum capacity. -func (l *Limiter) GetMaxCapacity() int64 { - return l.maxCapacity -} diff --git a/internal/pkg/transcoding/limiter_test.go b/internal/pkg/transcoding/limiter_test.go deleted file mode 100644 index df7db82ba0..0000000000 --- a/internal/pkg/transcoding/limiter_test.go +++ /dev/null @@ -1,88 +0,0 @@ -// 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 transcoding_test - -import ( - "sync" - "sync/atomic" - "testing" - - "github.com/huly-stream/internal/pkg/transcoding" - "github.com/stretchr/testify/require" -) - -func TestLimiter(t *testing.T) { - limiter := transcoding.NewLimiter(10) - - t.Run("Initial capacity", func(t *testing.T) { - require.Equal(t, int64(10), limiter.GetCapacity()) - }) - - t.Run("Successful consume", func(t *testing.T) { - success := limiter.TryConsume(5) - require.True(t, success) - require.Equal(t, int64(5), limiter.GetCapacity()) - }) - - t.Run("Failed consume", func(t *testing.T) { - success := limiter.TryConsume(10) - require.False(t, success) - require.Equal(t, int64(5), limiter.GetCapacity()) - }) - - t.Run("Return capacity", func(t *testing.T) { - limiter.ReturnCapacity(3) - require.Equal(t, int64(8), limiter.GetCapacity()) - }) - - t.Run("Exceeding max capacity", func(t *testing.T) { - limiter.ReturnCapacity(10) - require.Equal(t, int64(10), limiter.GetCapacity()) - }) -} - -func TestLimiterConcurrency(t *testing.T) { - limiter := transcoding.NewLimiter(10) - var wg sync.WaitGroup - - for i := 0; i < 10; i++ { - wg.Add(1) - go func() { - defer wg.Done() - limiter.TryConsume(2) - }() - } - - wg.Wait() - require.LessOrEqual(t, limiter.GetCapacity(), int64(0)) -} - -func TestLimiterCAS(t *testing.T) { - limiter := transcoding.NewLimiter(10) - var successful int64 - var wg sync.WaitGroup - - for i := 0; i < 1000; i++ { - wg.Add(1) - go func() { - defer wg.Done() - if limiter.TryConsume(1) { - atomic.AddInt64(&successful, 1) - } - }() - } - wg.Wait() - - require.Equal(t, int64(10), successful) - require.Equal(t, int64(0), limiter.GetCapacity()) -} diff --git a/internal/pkg/transcoding/scheduler.go b/internal/pkg/transcoding/scheduler.go deleted file mode 100644 index 080aeedbb5..0000000000 --- a/internal/pkg/transcoding/scheduler.go +++ /dev/null @@ -1,167 +0,0 @@ -// -// 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 transcoding - -import ( - "context" - "sync" - "time" - - "github.com/pkg/errors" - - "github.com/google/uuid" - "github.com/huly-stream/internal/pkg/config" - "github.com/huly-stream/internal/pkg/log" - "github.com/huly-stream/internal/pkg/resconv" - "github.com/huly-stream/internal/pkg/sharedpipe" - "github.com/huly-stream/internal/pkg/uploader" - "github.com/tus/tusd/v2/pkg/handler" - "go.uber.org/zap" -) - -// Scheduler represents manager for worker. It creates a new worker for clients and manages its life cycle. -type Scheduler struct { - conf *config.Config - - limiter *Limiter - - mainContext context.Context - logger *zap.Logger - workers sync.Map - cancels sync.Map -} - -// NewScheduler creates a new scheduler for transcode operations. -func NewScheduler(ctx context.Context, c *config.Config) *Scheduler { - return &Scheduler{ - conf: c, - limiter: NewLimiter(c.MaxCapacity), - mainContext: ctx, - logger: log.FromContext(ctx).With(zap.String("Scheduler", c.OutputDir)), - } -} - -// NewUpload creates a new worker with passed parameters -func (s *Scheduler) NewUpload(ctx context.Context, info handler.FileInfo) (handler.Upload, error) { - if info.ID == "" { - info.ID = uuid.NewString() - } - s.logger.Sugar().Debugf("upload: %v", info) - s.logger.Debug("NewUpload", zap.String("ID", info.ID)) - - var worker = &Worker{ - writer: sharedpipe.NewWriter(), - info: info, - logger: log.FromContext(s.mainContext).With(zap.String("worker", info.ID)), - 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 !s.limiter.TryConsume(cost) { - s.logger.Debug("run out of resources for scaling") - scaling = nil - } - - var commandOptions = Options{ - OuputDir: s.conf.OutputDir, - Threads: s.conf.MaxThreads, - UploadID: info.ID, - Level: level, - ScalingLevels: scaling, - } - - if s.conf.EndpointURL != nil { - s.logger.Sugar().Debugf("initializing uploader for %v", info) - var contentUploader, err = uploader.New(s.mainContext, s.conf.OutputDir, s.conf.EndpointURL, info) - if err != nil { - s.logger.Error("can not create uploader", zap.Error(err)) - return nil, err - } - - worker.contentUploader = contentUploader - go func() { - var serverErr = worker.contentUploader.Serve() - worker.logger.Debug("content uploader has finished", zap.Error(serverErr)) - }() - } - s.workers.Store(worker.info.ID, worker) - if err := worker.start(s.mainContext, &commandOptions); err != nil { - return nil, err - } - - go func() { - worker.wg.Wait() - s.limiter.ReturnCapacity(cost) - s.logger.Debug("returned capacity", zap.Int64("capacity", cost)) - close(worker.done) - }() - - s.logger.Debug("NewUpload", zap.String("done", info.ID)) - return worker, nil -} - -// GetUpload returns current a worker based on upload id -func (s *Scheduler) GetUpload(ctx context.Context, id string) (upload handler.Upload, err error) { - if v, ok := s.workers.Load(id); ok { - s.logger.Debug("GetUpload: found worker by id", zap.String("id", id)) - var w = v.(*Worker) - var cancelCtx, cancel = context.WithCancel(context.Background()) - if v, ok := s.cancels.Load(id); ok { - v.(context.CancelFunc)() - } - s.cancels.Store(id, cancel) - go func() { - select { - case <-w.done: - w.logger.Debug("upload timeout just canceled") - s.cancels.Delete(id) - return - case <-cancelCtx.Done(): - w.logger.Debug("upload refreshed") - return - case <-time.After(s.conf.Timeout): - w.logger.Debug("upload timeout") - s.cancels.Delete(id) - var terminateCtx, terminateCancel = context.WithTimeout(context.Background(), s.conf.Timeout) - defer terminateCancel() - _ = w.Terminate(terminateCtx) - } - }() - return w, nil - } - s.logger.Debug("GetUpload: worker not found", zap.String("id", id)) - return nil, errors.New("bad id") -} - -// AsTerminatableUpload returns tusd handler.TerminatableUpload -func (s *Scheduler) AsTerminatableUpload(upload handler.Upload) handler.TerminatableUpload { - var worker = upload.(*Worker) - s.logger.Debug("AsTerminatableUpload") - return worker -} - -// AsLengthDeclarableUpload returns tusd handler.LengthDeclarableUpload -func (s *Scheduler) AsLengthDeclarableUpload(upload handler.Upload) handler.LengthDeclarableUpload { - s.logger.Debug("AsLengthDeclarableUpload") - return upload.(*Worker) -} diff --git a/internal/pkg/uploader/datalake.go b/internal/pkg/uploader/datalake.go deleted file mode 100644 index 097bf0a6b5..0000000000 --- a/internal/pkg/uploader/datalake.go +++ /dev/null @@ -1,130 +0,0 @@ -// 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 uploader - -import ( - "bytes" - "context" - "io" - "mime/multipart" - "os" - "path/filepath" - - "github.com/huly-stream/internal/pkg/log" - "github.com/pkg/errors" - "github.com/valyala/fasthttp" - "go.uber.org/zap" -) - -// DatalakeStorage represents datalake storage -type DatalakeStorage struct { - baseURL string - workspace string - token string -} - -// NewDatalakeStorage creates a new datalake client -func NewDatalakeStorage(baseURL, workspace, token string) Storage { - return &DatalakeStorage{ - baseURL: baseURL, - token: token, - workspace: workspace, - } -} - -// UploadFile uploads file to the datalake -func (d *DatalakeStorage) UploadFile(ctx context.Context, fileName string) error { - // #nosec - file, err := os.Open(fileName) - if err != nil { - return err - } - defer func() { - _ = file.Close() - }() - - var objectKey = getObjectKey(fileName) - var logger = log.FromContext(ctx).With(zap.String("datalake upload", d.workspace), zap.String("fileName", fileName)) - - logger.Debug("start uploading") - - body := &bytes.Buffer{} - writer := multipart.NewWriter(body) - - part, err := writer.CreateFormFile("file", objectKey) - if err != nil { - return errors.Wrapf(err, "failed to create form file") - } - - _, err = io.Copy(part, file) - if err != nil { - return errors.Wrapf(err, "failed to copy file data") - } - - _ = writer.Close() - - req := fasthttp.AcquireRequest() - defer fasthttp.ReleaseRequest(req) - - res := fasthttp.AcquireResponse() - defer fasthttp.ReleaseResponse(res) - - req.SetRequestURI(d.baseURL + "/upload/form-data/" + d.workspace) - req.Header.SetMethod(fasthttp.MethodPost) - req.Header.Add("Authorization", "Bearer "+d.token) - req.Header.SetContentType(writer.FormDataContentType()) - req.SetBody(body.Bytes()) - - client := fasthttp.Client{} - if err := client.Do(req, res); err != nil { - return errors.Wrapf(err, "upload failed") - } - - logger.Debug("file uploaded") - - return nil -} - -// DeleteFile deletes file from the datalake -func (d *DatalakeStorage) DeleteFile(ctx context.Context, fileName string) error { - var logger = log.FromContext(ctx).With(zap.String("datalake delete", d.workspace), zap.String("fileName", fileName)) - logger.Debug("start deleting") - - var objectKey = getObjectKey(fileName) - - req := fasthttp.AcquireRequest() - defer fasthttp.ReleaseRequest(req) - - res := fasthttp.AcquireResponse() - defer fasthttp.ReleaseResponse(res) - - req.SetRequestURI(d.baseURL + "/blob/" + d.workspace + "/" + objectKey) - req.Header.SetMethod(fasthttp.MethodDelete) - req.Header.Add("Authorization", "Bearer "+d.token) - - client := fasthttp.Client{} - if err := client.Do(req, res); err != nil { - logger.Error("failed to del", zap.Error(err)) - return errors.Wrapf(err, "delete failed") - } - - logger.Debug("file deleted") - - return nil -} - -func getObjectKey(s string) string { - var _, objectKey = filepath.Split(s) - return objectKey -} diff --git a/internal/pkg/uploader/options.go b/internal/pkg/uploader/options.go index 076e2c70af..97ba5cc508 100644 --- a/internal/pkg/uploader/options.go +++ b/internal/pkg/uploader/options.go @@ -13,7 +13,15 @@ package uploader -type options struct{} +import "time" -// Option provides option for storages -type Option func(*options) +// Options reprents options for uploading files +type Options struct { + Dir string + WorkerCount uint32 + BufferSize int + RetryCount int + RetryDelay time.Duration + Timeout time.Duration + SourceFile string +} diff --git a/internal/pkg/uploader/postpone.go b/internal/pkg/uploader/postpone.go deleted file mode 100644 index 29b308dc5e..0000000000 --- a/internal/pkg/uploader/postpone.go +++ /dev/null @@ -1,48 +0,0 @@ -// 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 uploader - -import ( - "context" - "time" - - "github.com/huly-stream/internal/pkg/log" - "go.uber.org/zap" -) - -func (u *uploader) postpone(fileName string, action func(ctx context.Context)) { - u.waitJobs.Add(1) - var ctx, cancel = context.WithCancel(context.Background()) - ctx = log.WithLoggerFields(ctx, zap.String("pospone", "action")) - var startCh = time.After(u.postponeDuration) - - if v, ok := u.contexts.Load(fileName); ok { - (*v.(*context.CancelFunc))() - } - u.contexts.Store(fileName, &cancel) - - go func() { - u.waitJobs.Done() - defer cancel() - select { - case <-ctx.Done(): - return - case <-startCh: - action(ctx) - if ctx.Err() == nil { - u.contexts.CompareAndDelete(fileName, &cancel) - } - } - }() -} diff --git a/internal/pkg/uploader/uploader.go b/internal/pkg/uploader/uploader.go index 8d04618f57..d251d4b4da 100644 --- a/internal/pkg/uploader/uploader.go +++ b/internal/pkg/uploader/uploader.go @@ -11,204 +11,289 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package uploader provides objects and functions to work with uploading and monitoring files +// Package uploader provides a mechanism for uploading files to a remote storage. package uploader import ( "context" - "net/url" + "hash/fnv" "os" "path/filepath" "strings" "sync" "time" + "unsafe" - "github.com/pkg/errors" - "github.com/tus/tusd/v2/pkg/handler" - - "github.com/fsnotify/fsnotify" - "github.com/huly-stream/internal/pkg/config" - "github.com/huly-stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/log" + "github.com/hcengineering/stream/internal/pkg/storage" "go.uber.org/zap" + "k8s.io/utils/inotify" ) -type uploader struct { - done chan struct{} - waitJobs sync.WaitGroup - ctx context.Context - cancel context.CancelFunc - baseDir string - uploadID string - postponeDuration time.Duration - sentFiles sync.Map - storage Storage - contexts sync.Map - retryCount int - eventBufferCount uint +// See at https://man7.org/linux/man-pages/man7/inotify.7.html +const inotifyCloseWrite uint32 = 0x8 // IN_CLOSE_WRITE +const inotifyMovedTo uint32 = 0x80 // IN_MOVED_TO + +// Uploader represents file uploader +type Uploader interface { + Start() + Stop() + Cancel() } -func (u *uploader) retry(action func() error) { - var delay = time.Millisecond * 50 - for range u.retryCount { - if err := action(); err == nil { - return - } - time.Sleep(delay) - delay *= 2 +type uploaderImpl struct { + logger *zap.Logger + options *Options + storage storage.Storage + + filesCh chan string + + sentFiles sync.Map // key: string, value: struct{} + + shouldDeleteOnStop func(string) bool + + workersCh []chan func() + + watcherStopCh chan struct{} + watcherDoneCh chan struct{} + + uploadCtx context.Context + uploadCancel context.CancelFunc + + workerWaitGroup sync.WaitGroup +} + +// New creates a new instance of uploader +func New(ctx context.Context, s storage.Storage, opts Options) Uploader { + if s == nil { + panic("storage should not be nil") } -} -// Rollback deletes all delivered files and also deletes all local content by uploadID -func (u *uploader) Rollback() { - logger := log.FromContext(u.ctx).With(zap.String("uploader", "Rollback")) - logger.Debug("starting") - defer logger.Debug("done") + var res = &uploaderImpl{ + options: &opts, + storage: s, + logger: log.FromContext(ctx).With(zap.String("uploader", opts.Dir)), + shouldDeleteOnStop: func(s string) bool { + return strings.HasSuffix(s, "m3u8") + }, + filesCh: make(chan string, opts.BufferSize), + watcherStopCh: make(chan struct{}), + watcherDoneCh: make(chan struct{}), + } - u.postpone("", func(ctx context.Context) { - u.sentFiles.Range(func(key, value any) bool { - logger.Debug("deleting remote file", zap.String("key", key.(string))) - u.retry(func() error { return u.storage.DeleteFile(ctx, key.(string)) }) - return true - }) - }) + res.workersCh = make([]chan func(), opts.WorkerCount) - u.Terminate() -} + for i := range opts.WorkerCount { + res.workersCh[i] = make(chan func(), opts.BufferSize) + } -// Terminate deletes -func (u *uploader) Terminate() { - logger := log.FromContext(u.ctx).With(zap.String("uploader", "Terminate")) - logger.Debug("starting") + res.logger.Sugar().Debugf("uploader config is %v", opts) + + res.uploadCtx, res.uploadCancel = context.WithCancel(context.Background()) + + _ = os.MkdirAll(opts.Dir, os.ModePerm) + res.workerWaitGroup.Add(1) go func() { - defer logger.Debug("done") - u.waitJobs.Wait() - u.cancel() + defer res.workerWaitGroup.Done() + initFiles, _ := os.ReadDir(opts.Dir) + for _, f := range initFiles { + var filePath = filepath.Join(opts.Dir, f.Name()) + if filePath == opts.SourceFile { + continue + } + res.filesCh <- filePath + } }() + + return res } -func (u *uploader) uploadAndDelte(fileName string) { - u.postpone(fileName+"-del", func(context.Context) {}) - u.postpone(fileName, func(ctx context.Context) { - u.retry(func() error { return u.storage.UploadFile(ctx, fileName) }) - u.postpone(fileName+"-del", func(context.Context) { - _ = os.Remove(fileName) +func (u *uploaderImpl) Stop() { + u.stop(false) +} + +func (u *uploaderImpl) Cancel() { + u.stop(true) +} + +func (u *uploaderImpl) stop(rollback bool) { + close(u.watcherStopCh) + <-u.watcherDoneCh + u.logger.Debug("file watch stopped") + + if rollback { + u.logger.Debug("starting rollback...") + var i uint32 + u.sentFiles.Range(func(key, _ any) bool { + i++ + var filename = key.(string) + u.workersCh[i%u.options.WorkerCount] <- func() { + u.deleteRemoteFile(filename) + } + return true }) - }) + u.logger.Debug("rollback done") + } + close(u.filesCh) + u.workerWaitGroup.Wait() + u.logger.Debug("workers done") - u.sentFiles.Store(fileName, struct{}{}) + u.uploadCancel() + _ = os.RemoveAll(u.options.Dir) + u.sentFiles.Clear() + + u.logger.Debug("finish done", zap.Bool("cancel", rollback)) } -func (u *uploader) Serve() error { - var logger = log.FromContext(u.ctx).With(zap.String("uploader", u.uploadID), zap.String("dir", u.baseDir)) - var watcher, err = fsnotify.NewBufferedWatcher(u.eventBufferCount) - defer close(u.done) +func (u *uploaderImpl) Start() { + u.startWorkers() + u.startWatch() +} + +func (u *uploaderImpl) startWorkers() { + go func() { + var logger = u.logger.With(zap.String("func", "startWorkers")) + logger.Debug("fanout goroutine started") + defer logger.Debug("fanout goroutine stopped") + + h := fnv.New32a() + for f := range u.filesCh { + // #nosec + bytes := unsafe.Slice(unsafe.StringData(f), len(f)) + _, _ = h.Write(bytes) + id := h.Sum32() % u.options.WorkerCount + u.workersCh[id] <- func() { + u.uploadAndDelete(f) + } + h.Reset() + } + for i := range u.options.WorkerCount { + close(u.workersCh[i]) + } + }() + + for i := range u.options.WorkerCount { + var logger = u.logger.With(zap.Uint32("worker", i)) + + u.workerWaitGroup.Add(1) + go func(index uint32) { + logger.Debug("start") + defer logger.Debug("finished") + + defer u.workerWaitGroup.Done() + for work := range u.workersCh[index] { + work() + } + }(i) + } +} + +func (u *uploaderImpl) deleteRemoteFile(f string) { + var logger = u.logger.With(zap.String("delete remote", f)) + logger.Debug("deleting remote file") + + for range u.options.RetryCount { + var ctx, cancel = context.WithTimeout(u.uploadCtx, u.options.Timeout) + var err = u.storage.DeleteFile(ctx, f) + cancel() + + if err != nil { + logger.Error("attempt failed", zap.Error(err)) + } else { + logger.Debug("file deleted in remote storage") + break + } + + time.Sleep(u.options.RetryDelay) + } + + u.logger.Error("can not delete remote file") +} + +func (u *uploaderImpl) uploadAndDelete(f string) { + var logger = u.logger.With(zap.String("upload and delete", f)) + logger.Debug("uploading file") + + var _, ok = u.sentFiles.Load(f) + + if ok && !u.shouldDeleteOnStop(f) { + logger.Debug("file already uploaded") + return + } + + for range u.options.RetryCount { + var ctx, cancel = context.WithTimeout(u.uploadCtx, u.options.Timeout) + var err = u.storage.PutFile(ctx, f) + cancel() + + if err != nil { + logger.Error("attempt failed", zap.Error(err)) + } else { + if !u.shouldDeleteOnStop(f) { + _ = os.Remove(f) + logger.Debug("removed file locally") + } + u.sentFiles.Store(f, struct{}{}) + logger.Debug("file uploaded") + break + } + + time.Sleep(u.options.RetryDelay) + } +} + +func (u *uploaderImpl) startWatch() { + var logger = u.logger.With(zap.String("func", "startWatch")) + var watcher, err = inotify.NewWatcher() if err != nil { - logger.Error("can not start watcher") - return err + logger.Error("can not start file watcher", zap.Error(err)) + return } - _ = os.MkdirAll(u.baseDir, os.ModePerm) - initFiles, _ := os.ReadDir(u.baseDir) - for _, f := range initFiles { - var filePath = filepath.Join(u.baseDir, f.Name()) - u.uploadAndDelte(filePath) + if err := watcher.AddWatch(u.options.Dir, inotifyCloseWrite); err != nil { + logger.Error("can not start watching for close write", zap.Error(err)) + return } - - if err := watcher.Add(u.baseDir); err != nil { - return err + if err := watcher.AddWatch(u.options.Dir, inotifyMovedTo); err != nil { + logger.Error("can not start watching for moved to", zap.Error(err)) + return } - defer func() { _ = watcher.Close() + close(u.watcherDoneCh) }() - logger.Debug("uploader has initialized and started watching") + logger.Debug("watching for file updates") defer logger.Debug("done") for { select { - case <-u.ctx.Done(): - return u.ctx.Err() - case event, ok := <-watcher.Events: + case <-u.watcherStopCh: + return + case event, ok := <-watcher.Event: if !ok { - return u.ctx.Err() + logger.Error("file channel was closed") + return + } + if !strings.Contains(event.Name, u.options.Dir) { + continue + } + if event.Name == u.options.SourceFile { + continue } if strings.HasSuffix(event.Name, "tmp") { continue } - if !strings.Contains(event.Name, u.uploadID) { - continue - } - u.uploadAndDelte(event.Name) - case err, ok := <-watcher.Errors: + logger.Debug("received an event", zap.String("event", event.Name), zap.Uint32("mask", event.Mask)) + + u.filesCh <- event.Name + case err, ok := <-watcher.Error: if !ok { - return u.ctx.Err() + logger.Error("error channel was closed") } - logger.Error("get an error", zap.Error(err)) + logger.Error("received an error", zap.Error(err)) } } } - -// Uploader manages content delivering -type Uploader interface { - Terminate() - Rollback() - Serve() error -} - -// Storage represents file-based storage -type Storage interface { - UploadFile(ctx context.Context, fileName string) error - DeleteFile(ctx context.Context, fileName string) error -} - -// New creates a new instance of Uplaoder -func New(ctx context.Context, baseDir string, endpointURL *url.URL, uploadInfo handler.FileInfo) (Uploader, error) { - var uploaderCtx, uploadCancel = context.WithCancel(context.Background()) - uploaderCtx = log.WithLoggerFields(uploaderCtx) - - var storage Storage - var err error - - storage, err = NewStorageByURL(uploaderCtx, endpointURL, uploadInfo.MetaData) - if err != nil { - uploadCancel() - return nil, err - } - - return &uploader{ - ctx: uploaderCtx, - cancel: uploadCancel, - done: make(chan struct{}), - uploadID: uploadInfo.ID, - postponeDuration: time.Second * 2, - storage: storage, - retryCount: 10, - baseDir: filepath.Join(baseDir, uploadInfo.ID), - eventBufferCount: 100, - }, nil -} - -// NewStorageByURL creates a new storage basd on the type from the url scheme, for example "datalake://my-datalake-endpoint" -func NewStorageByURL(ctx context.Context, u *url.URL, headers map[string]string) (Storage, error) { - var workspace = headers["workspace"] - if workspace == "" { - return nil, errors.New("missed workspace in the client's metadata") - } - c, _ := config.FromEnv() - switch u.Scheme { - case "tus": - return nil, errors.New("not imlemented yet") - case "datalake": - if headers["token"] == "" { - return nil, errors.New("missed auth token in the client's metadata") - } - return NewDatalakeStorage(c.Endpoint().String(), workspace, headers["token"]), nil - case "s3": - return NewS3(ctx, c.Endpoint().String(), workspace), nil - default: - return nil, errors.New("unknown scheme") - } -} From 9c82eed873e14bb1625bb2ecf137fca2cb7f15db Mon Sep 17 00:00:00 2001 From: denis-tingaikin Date: Sun, 16 Mar 2025 01:09:22 +0300 Subject: [PATCH 2/3] update ci for go 1.24.1 Signed-off-by: denis-tingaikin --- .github/workflows/main.yaml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index 635804cf49..35e5f78798 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -36,7 +36,7 @@ jobs: - name: Setup Go uses: actions/setup-go@v5 with: - go-version: 1.23.5 + go-version: 1.24.1 - name: Build run: go build -race ./... - name: Test @@ -47,7 +47,7 @@ jobs: - uses: actions/checkout@v4 - uses: actions/setup-go@v5 with: - go-version: 1.23.5 + go-version: 1.24.1 - run: go mod tidy - name: Check for changes in go.mod or go.sum run: | @@ -63,9 +63,9 @@ jobs: - name: Setup Go uses: actions/setup-go@v5 with: - go-version: 1.23.5 + go-version: 1.24.1 - name: golangci-lint uses: golangci/golangci-lint-action@v4 with: - version: v1.60.3 + version: v1.64.7 args: --timeout 3m --verbose From c6a7bae060ca93dde948bb65a6f57d389d905366 Mon Sep 17 00:00:00 2001 From: denis-tingaikin Date: Sun, 16 Mar 2025 01:26:40 +0300 Subject: [PATCH 3/3] apply self code review Signed-off-by: denis-tingaikin --- Dockerfile | 4 ++-- internal/pkg/transcoder/{queue.go => scheduler.go} | 0 2 files changed, 2 insertions(+), 2 deletions(-) rename internal/pkg/transcoder/{queue.go => scheduler.go} (100%) diff --git a/Dockerfile b/Dockerfile index 4304349055..9fd38ffc7e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -11,7 +11,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -FROM --platform=linux/amd64 golang:1.24.1 AS builder +FROM --platform=$BUILDPLATFORM golang:1.24.1 AS builder ENV GO111MODULE=on ENV CGO_ENABLED=0 ENV GOBIN=/bin @@ -21,7 +21,7 @@ COPY . ./ RUN set -xe && GOOS=$TARGETOS GOARCH=$TARGETARCH go build -o /go/bin/stream ./cmd/stream -FROM --platform=linux/amd64 alpine +FROM --platform=$BUILDPLATFORM alpine RUN set -xe && apk add --no-cache ffmpeg RUN apk add --no-cache ca-certificates jq bash \ diff --git a/internal/pkg/transcoder/queue.go b/internal/pkg/transcoder/scheduler.go similarity index 100% rename from internal/pkg/transcoder/queue.go rename to internal/pkg/transcoder/scheduler.go