Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

uploader: Improve upload retry policy #61

Merged
merged 7 commits into from
Aug 22, 2024
Merged

uploader: Improve upload retry policy #61

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
01f77b3
core: Improve retry policy for uploads
victorges Jun 21, 2024
bab205f
core: Avoid keeping files in memory
victorges Jun 21, 2024
8e1577c
uploader: Fix error returned when backup URL can't be built
victorges Jun 25, 2024
bdd1bf3
Merge branch 'main' into vg/feat/better-retries
mjh1 Aug 9, 2024
38ab542
tweak retries and add time taken log
mjh1 Aug 12, 2024
02957ca
Add segment write timeout config param
mjh1 Aug 22, 2024
5d8ffd0
fix test
mjh1 Aug 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 82 additions & 46 deletions core/uploader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,17 +29,25 @@ func (bc *ByteCounter) Write(p []byte) (n int, err error) {
return n, nil
}

func newExponentialBackOffExecutor() *backoff.ExponentialBackOff {
func newExponentialBackOffExecutor(initial, max, totalMax time.Duration) *backoff.ExponentialBackOff {
backOff := backoff.NewExponentialBackOff()
backOff.InitialInterval = 30 * time.Second
backOff.MaxInterval = 2 * time.Minute
backOff.MaxElapsedTime = 0 // don't impose a timeout as part of the retries
backOff.InitialInterval = initial
backOff.MaxInterval = max
backOff.MaxElapsedTime = totalMax
backOff.Reset()
return backOff
}

func NoRetries() backoff.BackOff {
return &backoff.StopBackOff{}
}

func UploadRetryBackoff() backoff.BackOff {
return backoff.WithMaxRetries(newExponentialBackOffExecutor(), 2)
return newExponentialBackOffExecutor(2*time.Minute, 5*time.Minute, 1*time.Hour)
}

func SingleRequestRetryBackoff() backoff.BackOff {
return newExponentialBackOffExecutor(5*time.Second, 10*time.Second, 30*time.Second)
}

const segmentWriteTimeout = 5 * time.Minute
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reminder that we need to lower this value.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should take this in consideration. Looks like Storj writes are really slow.
https://eu-metrics-monitoring.livepeer.live/grafana/d/JZNaFMv4z/vod-monitoring?orgId=1&refresh=5s&from=now-24h&to=now&viewPanel=7

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've added a timetaken field to the "succeeded" log line so that we can get an idea of how long these uploads are taking. So then i'll lower it in a further PR does that sound ok?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

WDYT of making this an env var config, like the backup storage? So at least we can tune it better without a new code change

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah good idea 👍

Expand All @@ -49,29 +57,43 @@ var expiryField = map[string]string{
}

func Upload(input io.Reader, outputURI *url.URL, waitBetweenWrites, writeTimeout time.Duration, storageFallbackURLs map[string]string) (*drivers.SaveDataOutput, error) {
if strings.HasSuffix(outputURI.Path, ".ts") || strings.HasSuffix(outputURI.Path, ".mp4") {
ext := filepath.Ext(outputURI.Path)
inputFile, err := os.CreateTemp("", "upload-*"+ext)
if err != nil {
return nil, fmt.Errorf("failed to write to temp file: %w", err)
}
inputFileName := inputFile.Name()
defer os.Remove(inputFileName)

if ext == ".ts" || ext == ".mp4" {
// For segments we just write them in one go here and return early.
// (Otherwise the incremental write logic below caused issues with clipping since it results in partial segments being written.)
fileContents, err := io.ReadAll(input)
_, err = io.Copy(inputFile, input)
if err != nil {
return nil, fmt.Errorf("failed to read file")
return nil, fmt.Errorf("failed to write to temp file: %w", err)
}
if err := inputFile.Close(); err != nil {
return nil, fmt.Errorf("failed to close input file: %w", err)
}

out, bytesWritten, err := uploadFileWithBackup(outputURI, fileContents, nil, segmentWriteTimeout, true, storageFallbackURLs)
out, bytesWritten, err := uploadFileWithBackup(outputURI, inputFileName, nil, segmentWriteTimeout, true, storageFallbackURLs)
if err != nil {
return nil, fmt.Errorf("failed to upload video %s: (%d bytes) %w", outputURI.Redacted(), bytesWritten, err)
}

if err = extractThumb(outputURI, fileContents, storageFallbackURLs); err != nil {
if err = extractThumb(outputURI, inputFileName, storageFallbackURLs); err != nil {
glog.Errorf("extracting thumbnail failed for %s: %v", outputURI.Redacted(), err)
}
return out, nil
}

// For the manifest files we want a very short cache ttl as the files are updating every few seconds
fields := &drivers.FileProperties{CacheControl: "max-age=1"}
var fileContents []byte
var lastWrite = time.Now()
// Keep the file handle closed while we wait for input data
if err := inputFile.Close(); err != nil {
return nil, fmt.Errorf("failed to close input file: %w", err)
}

scanner := bufio.NewScanner(input)

Expand All @@ -89,11 +111,21 @@ func Upload(input io.Reader, outputURI *url.URL, waitBetweenWrites, writeTimeout

for scanner.Scan() {
b := scanner.Bytes()
fileContents = append(fileContents, b...)

inputFile, err = os.OpenFile(inputFileName, os.O_WRONLY|os.O_APPEND, 0644)
if err != nil {
return nil, fmt.Errorf("failed to open file: %w", err)
}
if _, err := inputFile.Write(b); err != nil {
return nil, fmt.Errorf("failed to append to input file: %w", err)
}
if err := inputFile.Close(); err != nil {
return nil, fmt.Errorf("failed to close input file: %w", err)
}

// Only write the latest version of the data that's been piped in if enough time has elapsed since the last write
if lastWrite.Add(waitBetweenWrites).Before(time.Now()) {
if _, _, err := uploadFileWithBackup(outputURI, fileContents, fields, writeTimeout, false, storageFallbackURLs); err != nil {
if _, _, err := uploadFileWithBackup(outputURI, inputFileName, fields, writeTimeout, false, storageFallbackURLs); err != nil {
// Just log this error, since it'll effectively be retried after the next interval
glog.Errorf("Failed to write: %v", err)
} else {
Expand All @@ -107,28 +139,40 @@ func Upload(input io.Reader, outputURI *url.URL, waitBetweenWrites, writeTimeout
}

// We have to do this final write, otherwise there might be final data that's arrived since the last periodic write
if _, _, err := uploadFileWithBackup(outputURI, fileContents, fields, writeTimeout, true, storageFallbackURLs); err != nil {
if _, _, err := uploadFileWithBackup(outputURI, inputFileName, fields, writeTimeout, false, storageFallbackURLs); err != nil {
// Don't ignore this error, since there won't be any further attempts to write
return nil, fmt.Errorf("failed to write final save: %w", err)
}
glog.Infof("Completed writing %s to storage", outputURI.Redacted())
return nil, nil
}

func uploadFileWithBackup(outputURI *url.URL, fileContents []byte, fields *drivers.FileProperties, writeTimeout time.Duration, withRetries bool, storageFallbackURLs map[string]string) (out *drivers.SaveDataOutput, bytesWritten int64, err error) {
out, bytesWritten, primaryErr := uploadFile(outputURI, fileContents, fields, writeTimeout, withRetries)
if primaryErr == nil {
return out, bytesWritten, nil
func uploadFileWithBackup(outputURI *url.URL, fileName string, fields *drivers.FileProperties, writeTimeout time.Duration, withRetries bool, storageFallbackURLs map[string]string) (out *drivers.SaveDataOutput, bytesWritten int64, err error) {
retryPolicy := NoRetries()
if withRetries {
retryPolicy = UploadRetryBackoff()
}
err = backoff.Retry(func() error {
var primaryErr error
out, bytesWritten, primaryErr = uploadFile(outputURI, fileName, fields, writeTimeout, withRetries)
if primaryErr == nil {
return nil
}

backupURI, err := buildBackupURI(outputURI, storageFallbackURLs)
if err != nil {
glog.Errorf("failed to build backup URL: %v", err)
return nil, 0, primaryErr
}
backupURI, err := buildBackupURI(outputURI, storageFallbackURLs)
if err != nil {
glog.Errorf("failed to build backup URL: %v", err)
return primaryErr
}
glog.Warningf("Primary upload failed, uploading to backupURL=%s primaryErr=%q", backupURI.Redacted(), primaryErr)

glog.Warningf("Primary upload failed, uploading to backupURL=%s primaryErr=%q", backupURI.Redacted(), primaryErr)
return uploadFile(backupURI, fileContents, fields, writeTimeout, withRetries)
out, bytesWritten, err = uploadFile(backupURI, fileName, fields, writeTimeout, withRetries)
if err == nil {
return nil
}
return fmt.Errorf("upload file errors: primary: %w; backup: %w", primaryErr, err)
}, retryPolicy)
return out, bytesWritten, err
}

func buildBackupURI(outputURI *url.URL, storageFallbackURLs map[string]string) (*url.URL, error) {
Expand All @@ -142,7 +186,7 @@ func buildBackupURI(outputURI *url.URL, storageFallbackURLs map[string]string) (
return nil, fmt.Errorf("no backup URL found for %s", outputURI.Redacted())
}

func uploadFile(outputURI *url.URL, fileContents []byte, fields *drivers.FileProperties, writeTimeout time.Duration, withRetries bool) (out *drivers.SaveDataOutput, bytesWritten int64, err error) {
func uploadFile(outputURI *url.URL, fileName string, fields *drivers.FileProperties, writeTimeout time.Duration, withRetries bool) (out *drivers.SaveDataOutput, bytesWritten int64, err error) {
outputStr := outputURI.String()
// While we wait for storj to implement an easier method for global object deletion we are hacking something
// here to allow us to have recording objects deleted after 7 days.
Expand All @@ -161,14 +205,20 @@ func uploadFile(outputURI *url.URL, fileContents []byte, fields *drivers.FilePro
}
session := driver.NewSession("")

var retryPolicy backoff.BackOff = &backoff.StopBackOff{} // no retries by default
retryPolicy := NoRetries()
if withRetries {
retryPolicy = UploadRetryBackoff()
retryPolicy = SingleRequestRetryBackoff()
}
err = backoff.Retry(func() error {
file, err := os.Open(fileName)
if err != nil {
return fmt.Errorf("failed to open file: %w", err)
}
defer file.Close()

// To count how many bytes we are trying to read then write (upload) to s3 storage
byteCounter := &ByteCounter{}
teeReader := io.TeeReader(bytes.NewReader(fileContents), byteCounter)
teeReader := io.TeeReader(file, byteCounter)

out, err = session.SaveData(context.Background(), "", teeReader, fields, writeTimeout)
bytesWritten = byteCounter.Count
Expand All @@ -182,20 +232,16 @@ func uploadFile(outputURI *url.URL, fileContents []byte, fields *drivers.FilePro
return out, bytesWritten, err
}

func extractThumb(outputURI *url.URL, segment []byte, storageFallbackURLs map[string]string) error {
func extractThumb(outputURI *url.URL, segmentFileName string, storageFallbackURLs map[string]string) error {
tmpDir, err := os.MkdirTemp(os.TempDir(), "thumb-*")
if err != nil {
return fmt.Errorf("temp file creation failed: %w", err)
}
defer os.RemoveAll(tmpDir)
outFile := filepath.Join(tmpDir, "out.jpg")
inFile := filepath.Join(tmpDir, filepath.Base(outputURI.Path))
if err = os.WriteFile(inFile, segment, 0644); err != nil {
return fmt.Errorf("failed to write input file: %w", err)
}

args := []string{
"-i", inFile,
"-i", segmentFileName,
"-ss", "00:00:00",
"-vframes", "1",
"-vf", "scale=854:480:force_original_aspect_ratio=decrease",
Expand All @@ -217,16 +263,6 @@ func extractThumb(outputURI *url.URL, segment []byte, storageFallbackURLs map[st
return fmt.Errorf("ffmpeg failed[%s] [%s]: %w", outputBuf.String(), stdErr.String(), err)
}

f, err := os.Open(outFile)
if err != nil {
return fmt.Errorf("opening file failed: %w", err)
}
defer f.Close()
thumbData, err := io.ReadAll(f)
if err != nil {
return fmt.Errorf("failed to read file: %w", err)
}

// two thumbs, one at session level, the other at stream level
thumbURLs := []*url.URL{outputURI.JoinPath("../latest.jpg"), outputURI.JoinPath("../../../latest.jpg")}
fields := &drivers.FileProperties{CacheControl: "max-age=5"}
Expand All @@ -235,7 +271,7 @@ func extractThumb(outputURI *url.URL, segment []byte, storageFallbackURLs map[st
for _, thumbURL := range thumbURLs {
thumbURL := thumbURL
errGroup.Go(func() error {
_, _, err = uploadFileWithBackup(thumbURL, thumbData, fields, 10*time.Second, true, storageFallbackURLs)
_, _, err = uploadFileWithBackup(thumbURL, outFile, fields, 10*time.Second, true, storageFallbackURLs)
if err != nil {
return fmt.Errorf("saving thumbnail failed: %w", err)
}
Expand Down
5 changes: 4 additions & 1 deletion core/uploader_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,9 @@ func TestUploadFileWithBackup(t *testing.T) {
require.NoError(t, err)
defer os.RemoveAll(dir)

testFile := filepath.Join(dir, "input.txt")
require.NoError(t, os.WriteFile(testFile, []byte("test"), 0644))

fakeStorage := "s3+https://fake.service.livepeer.com/bucket/"
backupStorage := filepath.Join(dir, "backup") + "/"
fakeOutput := fakeStorage + "hls/123/file.txt"
Expand All @@ -75,7 +78,7 @@ func TestUploadFileWithBackup(t *testing.T) {
storageFallbackURLs := map[string]string{
fakeStorage: "file://" + backupStorage,
}
out, written, err := uploadFileWithBackup(mustParseURL(fakeOutput), []byte("test"), nil, 0, false, storageFallbackURLs)
out, written, err := uploadFileWithBackup(mustParseURL(fakeOutput), testFile, nil, 0, false, storageFallbackURLs)
require.NoError(t, err)
require.Equal(t, expectedOutFile, out.URL)
require.Equal(t, int64(4), written)
Expand Down
Loading