Merge pull request #39 from OdyseeTeam/implement-multiple-storage

Implement multiple storage

Author: anbsky

.github/workflows/test.yaml

@@ -16,17 +16,15 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-      - run: |
-          docker compose up -d minio db redis &&
-          docker compose up -d cworker conductor &&
-          docker compose up minio-prepare
       - uses: actions/setup-go@v5
         with:
-          go-version: "1.22"
+          go-version: "1.23"
       - uses: FedericoCarboni/setup-ffmpeg@v3
         id: setup-ffmpeg
         with:
           ffmpeg-version: "6.1.0"
+      - run: |
+          make test_prepare
       - name: Run tests
         run: go test -covermode=count -coverprofile=coverage.out ./...
   golangci:
@@ -36,10 +34,10 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5
         with:
-          go-version: "1.22"
+          go-version: "1.23"
           cache: false
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v6
         with:
-          version: v1.58
+          version: v1.63
           args: -v

Makefile

@@ -1,25 +1,22 @@
-CC=x86_64-linux-musl-gcc
-CXX=x86_64-linux-musl-g++
-GOARCH=amd64
-GOOS=linux
-LDFLAGS=-ldflags "-linkmode external -extldflags -static"
-GO_BUILD=go build
-BUILD_DIR=dist
-LOCAL_ARCH=$(shell uname)
-VERSION := $(shell git describe --tags --match 'v*'|sed -e 's/v//')
+LOCAL_ARCH ?= $(shell uname)
+VERSION ?= $(shell git describe --tags --match 'v*'|sed -e 's/v//')
 TRANSCODER_VERSION ?= $(shell git describe --tags --match 'transcoder-v*'|sed 's/transcoder-v\([0-9.]*\).*/\1/')
+BUILD_DIR ?= dist
+GOOS ?= linux
+GOARCH ?= amd64
+GO_BUILD ?= go build
 
-transcoder: $(BUILD_DIR)/$(GOOS)_$(GOARCH)/transcoder
+transcoder:
 	GOARCH=$(GOARCH) GOOS=$(GOOS) CGO_ENABLED=0 \
   	$(GO_BUILD) -o $(BUILD_DIR)/$(GOOS)_$(GOARCH)/transcoder \
 	  -ldflags "-s -w -X github.com/OdyseeTeam/transcoder/internal/version.Version=$(TRANSCODER_VERSION)" \
 	  ./pkg/conductor/cmd/
 
-conductor_image:
+conductor_image: $(BUILD_DIR)/$(GOOS)_$(GOARCH)/transcoder
 	docker buildx build -f docker/Dockerfile-conductor -t odyseeteam/transcoder-conductor:$(TRANSCODER_VERSION) --platform linux/amd64 .
 	docker tag odyseeteam/transcoder-conductor:$(TRANSCODER_VERSION) odyseeteam/transcoder-conductor:latest
 
-cworker_image:
+cworker_image: $(BUILD_DIR)/$(GOOS)_$(GOARCH)/transcoder
 	docker buildx build -f docker/Dockerfile-cworker -t odyseeteam/transcoder-cworker:$(TRANSCODER_VERSION) --platform linux/amd64 .
 	docker tag odyseeteam/transcoder-cworker:$(TRANSCODER_VERSION) odyseeteam/transcoder-cworker:latest
 
@@ -30,9 +27,12 @@ test_down:
 	docker-compose down
 
 test_prepare:
-	docker-compose up -d minio db redis
-	docker-compose up -d cworker conductor
-	docker-compose up minio-prepare
+	make transcoder
+	make conductor_image
+	make cworker_image
+	docker compose -p transcoder up -d minio db redis
+	docker compose -p transcoder up -d cworker conductor
+	docker compose -p transcoder up minio-prepare
 
 test: test_prepare
 	go test -covermode=count -coverprofile=coverage.out ./...

conductor.ex.yml

@@ -1,10 +1,22 @@
-S3:
-  Name: local
-  Endpoint: http://minio:9000
-  Bucket: transcoded
-  Key: ody
-  Secret: odyseetes3
-  MaxSize: 1TB
+Storages:
+  - Name: local
+    Type: S3
+    Endpoint: http://minio:9000
+    Region: us-east-1
+    Bucket: transcoded
+    Key: ody
+    Secret: odyseetes3
+    MaxSize: 1TB
+    CreateBucket: true
+  - Name: remote
+    Type: S3
+    Endpoint: https://s3.wasabisys.com
+    Region: us-east-1
+    Bucket: production-videos
+    Key: AKIAXXXXXXXXXXXXXXXX
+    Secret: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
+    MaxSize: 5TB
+    CreateBucket: false
 
 AdaptiveQueue:
   MinHits: 1

docker-compose.yml

@@ -1,10 +1,8 @@
-version: '3.7'
-
 services:
   conductor:
-    image: odyseeteam/transcoder-conductor:24.2.3
+    image: odyseeteam/transcoder-conductor:latest
     platform: linux/amd64
-    container_name: tc-conductor
+    container_name: conductor
     command:
       - ./transcoder
       - conductor
@@ -26,9 +24,9 @@ services:
         max_attempts: 3
         window: 120s
   cworker:
-    image: odyseeteam/transcoder-cworker:24.2.3
+    image: odyseeteam/transcoder-cworker:latest
     platform: linux/amd64
-    container_name: tc-cworker
+    container_name: cworker
     command:
       - ./transcoder
       - worker
@@ -48,7 +46,7 @@ services:
   #     - '9090:8080'
   redis:
     image: redis:7.0
-    container_name: tc-redis
+    container_name: redis
     ports:
       - '6379:6379'
     volumes:
@@ -60,7 +58,7 @@ services:
     restart: unless-stopped
   db:
     image: postgres:14
-    container_name: tc-db
+    container_name: db
     ports:
       - "5432:5432"
     environment:
@@ -78,7 +76,7 @@ services:
       retries: 5
   minio:
     image: minio/minio
-    container_name: tc-minio
+    container_name: minio
     ports:
       - "9000:9000"
       - "38861:38861"
@@ -104,7 +102,6 @@ services:
     depends_on: ["minio"]
 
 volumes:
-
   db-data:
   minio-data:
   redis-data:

go.mod

@@ -1,6 +1,6 @@
 module github.com/OdyseeTeam/transcoder
 
-go 1.22
+go 1.23
 
 require (
 	github.com/Pallinder/go-randomdata v1.2.0

internal/config/config.go

@@ -0,0 +1,93 @@
+package config
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/spf13/viper"
+)
+
+type Storages []S3Config
+
+type ConductorConfig struct {
+	Storages      Storages
+	Redis         string
+	AdaptiveQueue AdaptiveQueue
+	Library       Library
+}
+
+type WorkerConfig struct {
+	Storage   S3Config
+	Redis     string
+	EdgeToken string
+}
+
+type S3Config struct {
+	Type         string
+	Name         string
+	Endpoint     string
+	Region       string
+	Bucket       string
+	Key          string
+	Secret       string
+	MaxSize      string
+	CreateBucket bool
+}
+
+type AdaptiveQueue struct {
+	MinHits int
+}
+
+type Library struct {
+	DSN          string
+	ManagerToken string
+}
+
+func ProjectRoot() (string, error) {
+	ex, err := os.Executable()
+	if err != nil {
+		return "", err
+	}
+	return filepath.Dir(ex), nil
+}
+
+func Read(name string, cfg any) error {
+	v := viper.New()
+	v.SetConfigName(name)
+
+	pp, err := ProjectRoot()
+	if err != nil {
+		return err
+	}
+	v.AddConfigPath(pp)
+	v.AddConfigPath(".")
+
+	if err := v.ReadInConfig(); err != nil {
+		return fmt.Errorf("fatal error reading config file: %w", err)
+	}
+
+	if err := v.Unmarshal(cfg); err != nil {
+		return fmt.Errorf("unable to decode into struct: %w", err)
+	}
+
+	return nil
+}
+
+func ReadConductorConfig() (*ConductorConfig, error) {
+	cfg := &ConductorConfig{}
+	return cfg, Read("conductor", cfg)
+}
+
+func ReadWorkerConfig() (*WorkerConfig, error) {
+	cfg := &WorkerConfig{}
+	return cfg, Read("worker", cfg)
+}
+
+func (s Storages) Endpoints() []string {
+	endpoints := []string{}
+	for _, v := range s {
+		endpoints = append(endpoints, v.Endpoint)
+	}
+	return endpoints
+}

library/library.go

@@ -22,37 +22,33 @@ const (
 )
 
 var ErrStreamNotFound = errors.New("stream not found")
-var storageURLs = map[string]string{
-	"wasabi": "https://s3.wasabisys.com/t-na2.odycdn.com",
-	"legacy": "https://na-storage-1.transcoder.odysee.com/t-na",
-}
 
 type Storage interface {
 	Name() string
-	GetURL(tid string) string
+	GetURL(item string) string
 	Put(stream *Stream, _ bool) error
 	Delete(tid string) error
 	DeleteFragments(tid string, fragments []string) error
 	// GetFragment(sdHash, name string) (storage.StreamFragment, error)
 }
 
 type Library struct {
-	db      *db.Queries
-	storage Storage
-	log     logging.KVLogger
+	db       *db.Queries
+	storages map[string]Storage
+	log      logging.KVLogger
 }
 
 type Config struct {
-	Storage Storage
-	DB      db.DBTX
-	Log     logging.KVLogger
+	Storages map[string]Storage
+	DB       db.DBTX
+	Log      logging.KVLogger
 }
 
 func New(config Config) *Library {
 	return &Library{
-		db:      db.New(config.DB),
-		log:     config.Log,
-		storage: config.Storage,
+		db:       db.New(config.DB),
+		log:      config.Log,
+		storages: config.Storages,
 	}
 }
 
@@ -137,7 +133,12 @@ func (lib *Library) RetireVideos(storageName string, maxSize uint64) (uint64, ui
 func (lib *Library) Retire(video db.Video) error {
 	var deleted bool
 
-	ll := lib.log.With("tid", video.TID, "sd_hash", video.SDHash)
+	ll := lib.log.With("tid", video.TID, "sd_hash", video.SDHash, "storage", video.Storage)
+
+	storage, ok := lib.storages[video.Storage]
+	if !ok {
+		return fmt.Errorf("storage %s not found", video.Storage)
+	}
 
 	if video.Manifest.Valid {
 		manifest := &Manifest{}
@@ -147,7 +148,7 @@ func (lib *Library) Retire(video db.Video) error {
 		} else if len(manifest.Files) == 0 {
 			ll.Warn("empty video manifest", "err", err)
 		} else {
-			err = lib.storage.DeleteFragments(video.TID, manifest.Files)
+			err = storage.DeleteFragments(video.TID, manifest.Files)
 			if err != nil {
 				ll.Warn("failed to delete fragments for remote video", "err", err)
 			} else {
@@ -156,7 +157,7 @@ func (lib *Library) Retire(video db.Video) error {
 		}
 	}
 	if !deleted {
-		err := lib.storage.Delete(video.TID)
+		err := storage.Delete(video.TID)
 		if err != nil {
 			ll.Warn("failed to delete remote video", "err", err)
 			return err
@@ -173,7 +174,7 @@ func (lib *Library) Retire(video db.Video) error {
 	return nil
 }
 
-// RetireVideos deletes older videos from S3, keeping total size of remote videos at maxSize.
+// ValidateStreams removes broken streams from the database.
 func (lib *Library) ValidateStreams(storageName string, offset, limit int32, remove bool) ([]string, []string, error) {
 	broken := []string{}
 	valid := []string{}
@@ -219,7 +220,7 @@ func (lib *Library) ValidateStreams(storageName string, offset, limit int32, rem
 	defer p.Release()
 
 	for _, v := range items {
-		url := fmt.Sprintf("%s/%s", storageURLs[v.Storage], v.Path)
+		url := lib.storages[v.Storage].GetURL(v.Path)
 		tids[url] = v.TID
 		wg.Add(1)
 		_ = p.Invoke(url)

library/library_test.go

@@ -41,8 +41,9 @@ func (s *librarySuite) TestAddChannel() {
 func (s *librarySuite) TestAddGetVideo() {
 	var err error
 
-	lib := New(Config{DB: s.DB, Storage: NewDummyStorage("storage1", "https://storage.host"), Log: zapadapter.NewKV(nil)})
-	newStream := GenerateDummyStream()
+	dummyStorage := NewDummyStorage("dummy1", "https://storage.host")
+	lib := New(Config{DB: s.DB, Storages: map[string]Storage{dummyStorage.Name(): dummyStorage}, Log: zapadapter.NewKV(nil)})
+	newStream := GenerateDummyStream(dummyStorage)
 
 	url, err := lib.GetVideoURL(newStream.SDHash())
 	s.ErrorIs(err, ErrStreamNotFound)