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

server: Fix timeout for stream recording background jobs #2573

Merged
merged 3 commits into from
Aug 31, 2022
Merged

server: Fix timeout for stream recording background jobs #2573

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
92ddc50
server: Use separate timeout for recording segments
victorges Aug 31, 2022
e1b98f2
server: Increase segment save attempts to 3
victorges Aug 31, 2022
7284abf
CHANGELOG
victorges Aug 31, 2022
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
1 change: 1 addition & 0 deletions CHANGELOG_PENDING.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
#### General

#### Broadcaster
- [#2573](https://github.com/livepeer/go-livepeer/pull/2573) server: Fix timeout for stream recording background jobs (@victorges)

#### Orchestrator

Expand Down
6 changes: 6 additions & 0 deletions clog/clog.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"strconv"
"strings"
"sync"
"time"

"github.com/golang/glog"
)
Expand Down Expand Up @@ -68,6 +69,11 @@ func Clone(parentCtx, logCtx context.Context) context.Context {
return context.WithValue(parentCtx, clogContextKey, newCmap)
}

func WithTimeout(parentCtx, logCtx context.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
ctx := Clone(parentCtx, logCtx)
return context.WithTimeout(ctx, timeout)
}

func AddManifestID(ctx context.Context, val string) context.Context {
return AddVal(ctx, manifestID, val)
}
Expand Down
10 changes: 8 additions & 2 deletions server/broadcast.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ var maxDurationSec = common.MaxDuration.Seconds()
// Max threshold for # of broadcast sessions under which we will refresh the session list
var maxRefreshSessionsThreshold = 8.0

var recordSegmentsMaxTimeout = 1 * time.Minute

var Policy *verification.Policy
var BroadcastCfg = &BroadcastConfig{}
var MaxAttempts = 3
Expand Down Expand Up @@ -787,8 +789,10 @@ func processSegment(ctx context.Context, cxn *rtmpConnection, seg *stream.HLSSeg
hasZeroVideoFrame := seg.IsZeroFrame
if ros != nil && !hasZeroVideoFrame {
go func() {
ctx, cancel := clog.WithTimeout(context.Background(), ctx, recordSegmentsMaxTimeout)
defer cancel()
now := time.Now()
uri, err := drivers.SaveRetried(ctx, ros, name, seg.Data, map[string]string{"duration": segDurMs}, 2)
uri, err := drivers.SaveRetried(ctx, ros, name, seg.Data, map[string]string{"duration": segDurMs}, 3)
took := time.Since(now)
if err != nil {
clog.Errorf(ctx, "Error saving name=%s bytes=%d to record store err=%q",
Expand Down Expand Up @@ -1184,11 +1188,13 @@ func downloadResults(ctx context.Context, cxn *rtmpConnection, seg *stream.HLSSe

if bros != nil {
go func() {
ctx, cancel := clog.WithTimeout(context.Background(), ctx, recordSegmentsMaxTimeout)
defer cancel()
ext, _ := common.ProfileFormatExtension(profile.Format)
name := fmt.Sprintf("%s/%d%s", profile.Name, seg.SeqNo, ext)
segDurMs := getSegDurMsString(seg)
now := time.Now()
uri, err := drivers.SaveRetried(ctx, bros, name, data, map[string]string{"duration": segDurMs}, 2)
uri, err := drivers.SaveRetried(ctx, bros, name, data, map[string]string{"duration": segDurMs}, 3)
took := time.Since(now)
if err != nil {
clog.Errorf(ctx, "Error saving nonce=%d manifestID=%s name=%s to record store err=%q", nonce, cxn.mid, name, err)
Expand Down