Live video on the AT Protocol

livepeer: keep two segments in flight (and add metrics)

See merge request streamplace/streamplace!125

Changelog: feature

Eli Mallon ec183b20 a10df6e6

+32 -15
+4 -3
pkg/atproto/firehose.go
··· 227 227 228 228 case repomgr.EvtKindDeleteRecord: 229 229 if collection.String() == constants.APP_BSKY_GRAPH_FOLLOW { 230 + if r == nil { 231 + log.Debug(ctx, "no repo found for follow", "userDID", evt.Repo, "subjectDID", rkey.String()) 232 + continue 233 + } 230 234 log.Debug(ctx, "deleting follow", "userDID", evt.Repo, "subjectDID", rkey.String()) 231 235 err := atsync.Model.DeleteFollow(ctx, evt.Repo, rkey.String()) 232 236 if err != nil { ··· 246 250 } 247 251 } 248 252 249 - if err != nil { 250 - return err 251 - } 252 253 default: 253 254 log.Error(ctx, "unexpected record op kind") 254 255 }
+1
pkg/director/stream_session.go
··· 135 135 } else { 136 136 log.Log(ctx, "transcoded segment", "took", took) 137 137 } 138 + spmetrics.QueuedTranscodeDuration.WithLabelValues(spseg.Creator).Set(float64(time.Since(start).Milliseconds())) 138 139 }() 139 140 } 140 141
+15 -11
pkg/livepeer/livepeer.go
··· 10 10 "mime/multipart" 11 11 "net/http" 12 12 "strings" 13 - "sync" 14 13 "time" 15 14 16 15 "golang.org/x/net/context/ctxhttp" 17 16 "stream.place/streamplace/pkg/aqhttp" 18 17 "stream.place/streamplace/pkg/log" 18 + "stream.place/streamplace/pkg/spmetrics" 19 19 "stream.place/streamplace/pkg/streamplace" 20 20 ) 21 21 22 + const SEGMENTS_IN_FLIGHT = 2 23 + 22 24 type LivepeerSession struct { 23 25 SessionID string 24 26 Count int 25 27 GatewayURL string 26 - SegLock sync.Mutex 28 + Guard chan struct{} 27 29 } 28 30 29 31 // borrowed from catalyst-api ··· 43 45 SessionID: fmt.Sprintf("%s-%s", did, sessionID), 44 46 Count: 0, 45 47 GatewayURL: gatewayURL, 48 + Guard: make(chan struct{}, SEGMENTS_IN_FLIGHT), 46 49 }, nil 47 50 } 48 51 49 - func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, seg *streamplace.Segment) ([][]byte, error) { 52 + func (ls *LivepeerSession) PostSegmentToGateway(ctx context.Context, buf []byte, spseg *streamplace.Segment) ([][]byte, error) { 50 53 ctx = log.WithLogValues(ctx, "func", "PostSegmentToGateway") 51 - ls.SegLock.Lock() 54 + ls.Guard <- struct{}{} 55 + start := time.Now() 52 56 // check if context is done since we were waiting for the lock 53 57 if ctx.Err() != nil { 54 - ls.SegLock.Unlock() 58 + <-ls.Guard 55 59 return nil, ctx.Err() 56 60 } 57 61 ctx, cancel := context.WithTimeout(ctx, time.Minute*5) ··· 59 63 url := fmt.Sprintf("%s/live/%s/%d.mp4", ls.GatewayURL, ls.SessionID, ls.Count) 60 64 ls.Count++ 61 65 62 - dur := time.Duration(*seg.Duration) 66 + dur := time.Duration(*spseg.Duration) 63 67 durationMs := int(dur.Milliseconds()) 64 68 log.Debug(ctx, "posting segment to livepeer gateway", "duration_ms", durationMs, "url", url) 65 69 66 - vid := seg.Video[0] 70 + vid := spseg.Video[0] 67 71 width := int(vid.Width) 68 72 height := int(vid.Height) 69 73 70 74 req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(buf)) 71 75 if err != nil { 72 - ls.SegLock.Unlock() 76 + <-ls.Guard 73 77 return nil, fmt.Errorf("failed to create request: %w", err) 74 78 } 75 79 req.Header.Set("Accept", "multipart/mixed") ··· 78 82 79 83 resp, err := ctxhttp.Do(ctx, &aqhttp.Client, req) 80 84 if err != nil { 81 - ls.SegLock.Unlock() 85 + <-ls.Guard 82 86 return nil, fmt.Errorf("failed to send segment to gateway: %w", err) 83 87 } 84 - ls.SegLock.Unlock() 88 + <-ls.Guard 85 89 defer resp.Body.Close() 86 90 87 91 if resp.StatusCode != http.StatusOK { ··· 113 117 out = append(out, bs) 114 118 } 115 119 } 116 - 120 + spmetrics.TranscodeDuration.WithLabelValues(spseg.Creator).Observe(float64(time.Since(start).Milliseconds())) 117 121 return out, nil 118 122 }
+12 -1
pkg/spmetrics/spmetrics.go
··· 20 20 var Viewers = promauto.NewGaugeVec(prometheus.GaugeOpts{ 21 21 Name: "streamplace_viewers", 22 22 Help: "number of current viewers per user", 23 - }, []string{"user"}) 23 + }, []string{"streamer"}) 24 24 25 25 var ViewersTotal = promauto.NewGauge(prometheus.GaugeOpts{ 26 26 Name: "streamplace_viewers_total", ··· 41 41 Name: "streamplace_transcode_errors_total", 42 42 Help: "total number of transcode errors", 43 43 }) 44 + 45 + var TranscodeDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ 46 + Name: "streamplace_transcode_duration_ms", 47 + Help: "duration of transcode in ms", 48 + Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000}, 49 + }, []string{"streamer"}) 50 + 51 + var QueuedTranscodeDuration = promauto.NewGaugeVec(prometheus.GaugeOpts{ 52 + Name: "streamplace_queued_transcode_duration_ms", 53 + Help: "duration of transcode in ms, including time spent waiting", 54 + }, []string{"streamer"}) 44 55 45 56 var Version = promauto.NewCounterVec(prometheus.CounterOpts{ 46 57 Name: "streamplace_version",