Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
17 kB · 374 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375package spmetrics
import ( "sync"
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto")
var viewersByStreamer = map[string]int{}var viewersByProtocol = map[string]int{}var viewersLock sync.RWMutex
var Viewers = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_viewers", Help: "number of current viewers per user",}, []string{"streamer"})
var ViewersTotal = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_viewers_total", Help: "total number of viewers",}, []string{"protocol"})
var StreamSessions = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_stream_sessions", Help: "number of open stream sessions per streamer",}, []string{"streamer"})
var SendSegmentCalls = promauto.NewGauge(prometheus.GaugeOpts{ Name: "streamplace_send_segment_calls", Help: "total number of send segment calls currently in flight",})
var SwarmPutCalls = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_swarm_put_calls", Help: "total number of swarm put calls currently in flight",}, []string{"streamer"})
var TranscodeAttemptsTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_transcode_attempts_total", Help: "total number of transcode attempts",})
var TranscodeSuccessesTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_transcode_successes_total", Help: "total number of transcode successes",})
var TranscodeSkippedTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_transcode_skipped_total", Help: "segments not transcoded because the transcoder was too far behind (see livepeer.MaxWaiting)",})
var TranscodeErrorsTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_transcode_errors_total", Help: "total number of transcode errors",})
var IngestNonIDRKeyframesTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_ingest_non_idr_keyframes_total", Help: "frames the parser flagged as keyframes that hold no IDR slice (open-GOP I-frames), kept inside their GoP instead of starting a segment",}, []string{"streamer"})
var TranscodeSkippedNoKeyframeTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_transcode_skipped_no_keyframe_total", Help: "segments not sent to the transcoder because their video does not start with an IDR and its parameter sets (a transcoder cannot decode them and drops the session)",})
var TranscodeManifestRotationsTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_transcode_manifest_rotations_total", Help: "times a stream moved to a fresh gateway manifest after the gateway refused every recent push (see livepeer.RotateAfterFailures)",})
var TranscodeDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ Name: "streamplace_transcode_duration_ms", Help: "duration of transcode in ms", Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000},}, []string{"streamer"})
var SigningDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ Name: "streamplace_signing_duration_ms", Help: "duration of transcode in ms", Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000, 20000, 30000, 60000},}, []string{"streamer"})
// SegmentDeliveryDuration measures wall time from when the muxer hands a// fresh segment to the sign callback through the end of validation —// i.e. the latency the user actually waits on before the segment is// available downstream. Wraps per-segment signing + ValidateMP4.var SegmentDeliveryDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ Name: "streamplace_segment_delivery_duration_ms", Help: "duration of sign + validate in ms (the user-visible per-segment latency)", Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000, 20000, 30000, 60000},}, []string{"streamer"})
var QueuedTranscodeDuration = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_queued_transcode_duration_ms", Help: "duration of transcode in ms, including time spent waiting",}, []string{"streamer"})
var Version = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_version", Help: "version of streamplace",}, []string{"version"})
var WebsocketsOpen = promauto.NewGauge(prometheus.GaugeOpts{ Name: "streamplace_websockets_open", Help: "number of open playback websockets",})
var ReplicationWebsocketsOpen = promauto.NewGauge(prometheus.GaugeOpts{ Name: "streamplace_replication_websockets_open", Help: "number of open replication websockets",})
// --- syndication (outbound) -------------------------------------------------//// The inbound side is ReplicationWebsocketsOpen (peers pulling from this// node). These are this node pulling from origins: what it does with each// broadcast origin it hears about, how many pulls it holds open, and how// often a pull fails. A node that is supposed to syndicate but shows only// not_syndicated or self outcomes, or connect errors, is misconfigured or// can't reach the origin.
// ReplicationOutboundOpen is the number of origin websockets this node is// currently pulling segments from, per streamer.var ReplicationOutboundOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_replication_outbound_open", Help: "outbound syndication websockets this node holds open to origin nodes, per streamer",}, []string{"streamer"})
// BroadcastOriginsTotal counts broadcast origin records handled, by what was// done with each: connect (a pull was opened), not_syndicated (the streamer// isn't in the syndicate role), self (the origin is this node),// already_connected, no_url (the record carries no websocket URL).var BroadcastOriginsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_broadcast_origins_total", Help: "broadcast origin records handled, by outcome (connect|not_syndicated|self|already_connected|no_url)",}, []string{"outcome"})
// ReplicationConnectErrorsTotal counts outbound syndication pulls that// failed: the dial, or a read after it.var ReplicationConnectErrorsTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_replication_connect_errors_total", Help: "outbound syndication websocket dials or reads that failed",})
var SegmentSubscriptionsOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_segment_subscriptions_open", Help: "number of open new segment subscriptions",}, []string{"streamer", "rendition"})
var LabelerFirehosesConnected = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_labeler_firehoses_connected", Help: "number of currently connected labeler firehoses",}, []string{"labeler"})
// FirehoseRelaysConnected is 1 while a relay's firehose is connected and 0// while it is reconnecting, labeled by relay host and protocol ("websocket" or// "moq"). With multi-relay support this shows at a glance how many of the// configured relays are currently feeding us.var FirehoseRelaysConnected = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_firehose_relays_connected", Help: "1 if the relay's firehose is currently connected, else 0",}, []string{"relay", "protocol"})
// FirehoseEventsReceivedTotal counts every event a relay delivered, before// cross-relay dedup, labeled by relay, protocol ("websocket"/"moq"), and kind.// Comparing rates across relays tells you whether they carry the same volume;// received minus deduped is each relay's first-seen (unique) contribution.var FirehoseEventsReceivedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_firehose_events_received_total", Help: "firehose events delivered by each relay before dedup, by relay/protocol/kind",}, []string{"relay", "protocol", "kind"})
// FirehoseEventsDedupedTotal counts events dropped because the same commit (or// identity update) already arrived from another relay, labeled by the relay that// delivered the duplicate, its protocol, and event kind. A relay whose deduped// count tracks its received count is fully redundant (a healthy mirror); a relay// with a high first-seen count (received minus deduped) is contributing events// the others are not — i.e. the relays are diverging.var FirehoseEventsDedupedTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_firehose_events_deduped_total", Help: "firehose events dropped as cross-relay duplicates, by relay/protocol/kind",}, []string{"relay", "protocol", "kind"})
// FirehoseBadFramesTotal counts firehose frames that failed strict decoding// and were skipped (the connection stays up; only the frame is dropped),// labeled by relay and protocol. A nonzero rate means a relay is delivering// bytes that violate the atproto event schema — either corruption in the// relay's serving path or genuinely invalid upstream data; the accompanying// warn log carries the decoded context (DID, seq) for investigation.var FirehoseBadFramesTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_firehose_bad_frames_total", Help: "firehose frames skipped because they failed decoding, by relay/protocol",}, []string{"relay", "protocol"})
// FirehoseRelayHighSeq is the highest upstream sequence number seen from each// relay. A WebSocket bsky relay and a MoQ relay that bridges the same upstream// both carry bsky's original seq, so the difference between two relays' high_seq// is a direct, units-of-events measure of how far apart they are: a small,// roughly-constant skew means they are in sync (just a latency offset); a// growing gap means one relay is falling behind or stuck.var FirehoseRelayHighSeq = promauto.NewGaugeVec(prometheus.GaugeOpts{ Name: "streamplace_firehose_relay_high_seq", Help: "highest upstream sequence number seen from each relay (cross-relay diff = parity/lag)",}, []string{"relay", "protocol"})
// --- isolated ingest workers ------------------------------------------------
// IngestWorkerStarts counts isolated ingest worker subprocesses spawned, by// transport ("mkv-fd" = fd-4 fallback, "mkv" = detached, "whip" = detached WHIP).var IngestWorkerStarts = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_ingest_worker_starts_total", Help: "isolated ingest worker subprocesses spawned, by transport",}, []string{"transport"})
// IngestWorkerExits counts isolated ingest worker exits by transport and outcome// ("clean" | "crash"). A rising crash rate is the signal that a stream is// repeatedly faulting — the contained fault the node now survives but which would// otherwise be invisible. A worker left running across a main shutdown is not an// exit and is not counted; "resumed" is a worker reattached after a restart.var IngestWorkerExits = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_ingest_worker_exits_total", Help: "isolated ingest worker exits, by transport and outcome (clean|crash)",}, []string{"transport", "outcome"})
// --- VOD processing ---------------------------------------------------------
// VODProcessAttemptsTotal increments once per task dequeued for VOD// processing, labeled by the input storage backend (file/s3). Pair with// VODProcessSuccessesTotal / VODProcessErrorsTotal to compute success// rate.var VODProcessAttemptsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_vod_process_attempts_total", Help: "total number of VOD processing attempts (one per dequeued task)",}, []string{"backend"})
var VODProcessSuccessesTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_vod_process_successes_total", Help: "total number of VOD processing runs that produced a content-addressed object",}, []string{"backend"})
// VODProcessErrorsTotal is labeled by the stage that failed so we can// see whether the bulk of failures are codec issues, S3 issues, etc.// Stage values: open_source, gstreamer_pipeline, muxl_drain,// s3_complete, content_address_copy.var VODProcessErrorsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_vod_process_errors_total", Help: "total number of VOD processing failures, by stage",}, []string{"stage"})
// VODProcessDurationMS is the end-to-end wall time of one ProcessVOD// call, from task dequeue to final CopyObject. The wide buckets cover// realistic upload sizes (small phone clips through multi-hour// long-form videos).var VODProcessDurationMS = promauto.NewHistogram(prometheus.HistogramOpts{ Name: "streamplace_vod_process_duration_ms", Help: "end-to-end wall time of one VOD processing run", Buckets: []float64{500, 1000, 2500, 5000, 10000, 30000, 60000, 120000, 300000, 600000, 1800000, 3600000},})
// VODInputBytes / VODOutputBytes are observed once per successful run.// Comparing them gives a transcode ratio (output/input) — e.g. opus->aac// usually produces a modestly larger AAC payload, while many users will// have inputs with surplus container overhead that gets trimmed.var VODInputBytes = promauto.NewHistogram(prometheus.HistogramOpts{ Name: "streamplace_vod_input_bytes", Help: "size in bytes of the user upload being processed", Buckets: prometheus.ExponentialBuckets(1<<20, 4, 8), // 1 MB ... 16 GB})
var VODOutputBytes = promauto.NewHistogram(prometheus.HistogramOpts{ Name: "streamplace_vod_output_bytes", Help: "size in bytes of the content-addressed fMP4 produced", Buckets: prometheus.ExponentialBuckets(1<<20, 4, 8),})
// S3 helpers — instrumented for both the VOD pipeline and any future// reusers of pkg/s3.ReaderAt / pkg/s3.MultipartWriter.
var S3ReaderAtOpensTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_s3_readerat_opens_total", Help: "total number of ranged GetObject opens issued by pkg/s3.ReaderAt",})
var S3ReaderAtReadsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "streamplace_s3_readerat_reads_total", Help: "total ReadAt calls, by whether they hit the open body (sequential) or forced a reopen",}, []string{"kind"}) // "sequential" or "seek"
var S3MultipartPartsUploadedTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_s3_multipart_parts_uploaded_total", Help: "total number of parts uploaded across all multipart uploads",})
var S3MultipartBytesUploadedTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "streamplace_s3_multipart_bytes_uploaded_total", Help: "total bytes written via pkg/s3.MultipartWriter",})
func ViewerInc(user string, protocol string) { go func() { viewersLock.Lock() defer viewersLock.Unlock() viewersByStreamer[user]++ viewersByProtocol[protocol]++ Viewers.WithLabelValues(user).Set(float64(viewersByStreamer[user])) ViewersTotal.WithLabelValues(protocol).Set(float64(viewersByProtocol[protocol])) }()}
func ViewerDec(user string, protocol string) { go func() { viewersLock.Lock() defer viewersLock.Unlock() viewersByStreamer[user]-- if viewersByStreamer[user] == 0 { Viewers.DeleteLabelValues(user) } else { Viewers.WithLabelValues(user).Set(float64(viewersByStreamer[user])) } viewersByProtocol[protocol]-- // Per-protocol series stay at 0 rather than vanishing: a node with // no viewers should read 0, not "no data". (Per-streamer ones are // unbounded, so those are still dropped.) ViewersTotal.WithLabelValues(protocol).Set(float64(viewersByProtocol[protocol])) }()}
func GetViewCount(user string) int { viewersLock.RLock() defer viewersLock.RUnlock() return viewersByStreamer[user]}
// A labelled metric only appears on /metrics once some label combination has// been touched, so an idle node shows nothing at all for it — no way to tell// "zero" from "not scraping". Touching the known label values (and an empty// sentinel for per-streamer ones, which Prometheus reads as the label being// absent) makes every one of these report 0 from boot. Per-relay and// per-labeler ones are touched where those are configured.func init() { for _, v := range []*prometheus.GaugeVec{Viewers, StreamSessions, SwarmPutCalls, ReplicationOutboundOpen} { v.WithLabelValues("") } SegmentSubscriptionsOpen.WithLabelValues("", "") for _, protocol := range []string{"webrtc", "hls", "local"} { ViewersTotal.WithLabelValues(protocol) } for _, outcome := range []string{"connect", "not_syndicated", "self", "already_connected", "no_url"} { BroadcastOriginsTotal.WithLabelValues(outcome) } for _, transport := range []string{"mp4-fd", "mp4", "whip"} { IngestWorkerStarts.WithLabelValues(transport) for _, outcome := range []string{"clean", "crash"} { IngestWorkerExits.WithLabelValues(transport, outcome) } } for _, backend := range []string{"file", "s3"} { VODProcessAttemptsTotal.WithLabelValues(backend) VODProcessSuccessesTotal.WithLabelValues(backend) }}
// TouchRelay makes a relay's per-relay series exist at 0 before its first// connection, so a relay that never connects is visible as 0 rather than// missing.func TouchRelay(relay, protocol string) { FirehoseRelaysConnected.WithLabelValues(relay, protocol) for _, kind := range []string{"commit", "identity"} { FirehoseEventsReceivedTotal.WithLabelValues(relay, protocol, kind) }}