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.

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

upload: add TeeReader to count bytes during upload #54

Merged
merged 1 commit into from
Feb 14, 2024
Merged
Changes from all commits
Commits
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
19 changes: 16 additions & 3 deletions core/uploader.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,15 @@
"github.com/livepeer/go-tools/drivers"
)

type ByteCounter struct {
Count int64
}

func (bc *ByteCounter) Write(p []byte) (n int, err error) {
bc.Count += int64(len(p))
return n, nil

Check warning on line 27 in core/uploader.go

View check run for this annotation

Codecov / codecov/patch

core/uploader.go#L25-L27

Added lines #L25 - L27 were not covered by tests
}

func newExponentialBackOffExecutor() *backoff.ExponentialBackOff {
backOff := backoff.NewExponentialBackOff()
backOff.InitialInterval = 200 * time.Millisecond
Expand Down Expand Up @@ -55,6 +64,7 @@
}
}

byteCounter := &ByteCounter{}
if strings.HasSuffix(output, ".ts") || strings.HasSuffix(output, ".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.)
Expand All @@ -63,16 +73,19 @@
return nil, fmt.Errorf("failed to read file")
}

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

Check warning on line 78 in core/uploader.go

View check run for this annotation

Codecov / codecov/patch

core/uploader.go#L77-L78

Added lines #L77 - L78 were not covered by tests
var out *drivers.SaveDataOutput
err = backoff.Retry(func() error {
out, err = session.SaveData(context.Background(), "", bytes.NewReader(fileContents), fields, segmentWriteTimeout)
out, err = session.SaveData(context.Background(), "", teeReader, fields, segmentWriteTimeout)

Check warning on line 81 in core/uploader.go

View check run for this annotation

Codecov / codecov/patch

core/uploader.go#L81

Added line #L81 was not covered by tests
if err != nil {
glog.Errorf("failed upload attempt for %s: %v", outputURI.Redacted(), err)
glog.Errorf("failed upload attempt for %s (%d bytes): %v", outputURI.Redacted(), byteCounter.Count, err)

Check warning on line 83 in core/uploader.go

View check run for this annotation

Codecov / codecov/patch

core/uploader.go#L83

Added line #L83 was not covered by tests
}
return err
}, UploadRetryBackoff())
if err != nil {
return nil, fmt.Errorf("failed to upload video %s: %w", outputURI.Redacted(), err)
return nil, fmt.Errorf("failed to upload video %s: (%d bytes) %w", outputURI.Redacted(), byteCounter.Count, err)

Check warning on line 88 in core/uploader.go

View check run for this annotation

Codecov / codecov/patch

core/uploader.go#L88

Added line #L88 was not covered by tests
}

if err = extractThumb(session, output, fileContents); err != nil {
Expand Down
Loading