mirror of
https://github.com/hanzoai/livekit.git
synced 2026-10-30 18:29:34 +00:00
zeekay
f143a420ac
This repo did not compile, and had not before any of the KV work: at the commit prior to it, `go build ./...` failed identically in pkg/telemetry/prometheus. That failure masked every package downstream of it, which is why the KV rename could only be type-checked in isolation. The telemetry code was written against an older luxfi/metric API: metric.CounterVec/HistogramVec/GaugeVec are INTERFACES, but 28 declarations said *metric.X — a pointer to an interface, which nothing satisfies. metric.MustRegister does not exist, and is not needed: NewHistogramVec and friends already register into DefaultRegistry at construction. 36 calls removed. metric.ExponentialBucketsRange does not exist; the bucket helper comes from prometheus/client_golang. rpc.InitPSRPCStats / webhook.InitWebhookStats come from livekit/protocol and take prometheus.Labels, not metric.Labels. Those two call sites now pass the type the upstream signature actually declares. metric.DefaultGatherer does not exist. Replaced with metric.DefaultRegistry, which satisfies Gatherer (Registry embeds it). NOT metric.Handler() — that wraps a FRESH NewRegistry(), so the endpoint would have served zero metrics while looking healthy. `go build ./...` and `go vet ./pkg/...` now pass.
334 lines
11 KiB
Go
334 lines
11 KiB
Go
// Copyright 2023 LiveKit, Inc.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package prometheus
|
|
|
|
import (
|
|
metric "github.com/luxfi/metric"
|
|
"go.uber.org/atomic"
|
|
|
|
"github.com/livekit/protocol/livekit"
|
|
)
|
|
|
|
type Direction string
|
|
|
|
const (
|
|
Incoming Direction = "incoming"
|
|
Outgoing Direction = "outgoing"
|
|
)
|
|
|
|
type TransmissionType string
|
|
|
|
const (
|
|
TransmissionInitial TransmissionType = "initial"
|
|
TransmissionRetransmit TransmissionType = "retransmit"
|
|
)
|
|
|
|
var (
|
|
bytesIn atomic.Uint64
|
|
bytesOut atomic.Uint64
|
|
packetsIn atomic.Uint64
|
|
packetsOut atomic.Uint64
|
|
nackTotal atomic.Uint64
|
|
retransmitBytes atomic.Uint64
|
|
retransmitPackets atomic.Uint64
|
|
participantSignalConnected atomic.Uint64
|
|
participantRTCConnected atomic.Uint64
|
|
participantRTCInit atomic.Uint64
|
|
participantRTCCanceled atomic.Uint64
|
|
forwardLatency atomic.Uint32
|
|
forwardJitter atomic.Uint32
|
|
|
|
promPacketLabels = []string{"direction", "transmission", "country"}
|
|
promPacketTotal metric.CounterVec
|
|
promPacketBytes metric.CounterVec
|
|
promRTCPLabels = []string{"direction", "country"}
|
|
promStreamLabels = []string{"direction", "source", "type", "country"}
|
|
promNackTotal metric.CounterVec
|
|
promPliTotal metric.CounterVec
|
|
promFirTotal metric.CounterVec
|
|
promPacketLossTotal metric.CounterVec
|
|
promPacketLoss metric.HistogramVec
|
|
promPacketOutOfOrderTotal metric.CounterVec
|
|
promPacketOutOfOrder metric.HistogramVec
|
|
promJitter metric.HistogramVec
|
|
promRTT metric.HistogramVec
|
|
promParticipantJoin metric.CounterVec
|
|
promConnections metric.GaugeVec
|
|
promForwardLatency metric.Gauge
|
|
promForwardJitter metric.Gauge
|
|
promForwardLatencyHist metric.Histogram
|
|
)
|
|
|
|
func initPacketStats(nodeID string, nodeType livekit.NodeType) {
|
|
promPacketTotal = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "packet",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, promPacketLabels)
|
|
promPacketBytes = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "packet",
|
|
Name: "bytes",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, promPacketLabels)
|
|
promNackTotal = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "nack",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, promRTCPLabels)
|
|
promPliTotal = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "pli",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, promRTCPLabels)
|
|
promFirTotal = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "fir",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, promRTCPLabels)
|
|
promPacketLossTotal = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "packet_loss",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, promStreamLabels)
|
|
promPacketLoss = metric.NewHistogramVec(metric.HistogramOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "packet_loss",
|
|
Name: "percent",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
Buckets: []float64{0.0, 0.1, 0.3, 0.5, 0.7, 1, 5, 10, 40, 100},
|
|
}, promStreamLabels)
|
|
promPacketOutOfOrderTotal = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "packet_out_of_order",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, promStreamLabels)
|
|
promPacketOutOfOrder = metric.NewHistogramVec(metric.HistogramOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "packet_out_of_order",
|
|
Name: "percent",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
Buckets: []float64{0.0, 0.1, 0.3, 0.5, 0.7, 1, 5, 10, 40, 100},
|
|
}, promStreamLabels)
|
|
promJitter = metric.NewHistogramVec(metric.HistogramOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "jitter",
|
|
Name: "us",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
// 1ms, 10ms, 30ms, 50ms, 70ms, 100ms, 300ms, 600ms, 1s
|
|
Buckets: []float64{1000, 10000, 30000, 50000, 70000, 100000, 300000, 600000, 1000000},
|
|
}, promStreamLabels)
|
|
promRTT = metric.NewHistogramVec(metric.HistogramOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "rtt",
|
|
Name: "ms",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
Buckets: []float64{50, 100, 150, 200, 250, 500, 750, 1000, 5000, 10000},
|
|
}, promStreamLabels)
|
|
promParticipantJoin = metric.NewCounterVec(metric.CounterOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "participant_join",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, []string{"state"})
|
|
promConnections = metric.NewGaugeVec(metric.GaugeOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "connection",
|
|
Name: "total",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
}, []string{"kind"})
|
|
promForwardLatency = metric.NewGauge(metric.GaugeOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "forward",
|
|
Name: "latency",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
})
|
|
promForwardJitter = metric.NewGauge(metric.GaugeOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "forward",
|
|
Name: "jitter",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
})
|
|
promForwardLatencyHist = metric.NewHistogram(metric.HistogramOpts{
|
|
Namespace: livekitNamespace,
|
|
Subsystem: "forward_latency",
|
|
Name: "ns",
|
|
ConstLabels: metric.Labels{"node_id": nodeID, "node_type": nodeType.String()},
|
|
// 50us, 100us, 250us, 500us, 1ms, 2ms, 3ms, 5ms, 10ms, 20ms
|
|
Buckets: []float64{
|
|
50 * 1000,
|
|
100 * 1000,
|
|
250 * 1000,
|
|
500 * 1000,
|
|
1 * 1000 * 1000,
|
|
2 * 1000 * 1000,
|
|
3 * 1000 * 1000,
|
|
5 * 1000 * 1000,
|
|
10 * 1000 * 1000,
|
|
20 * 1000 * 1000,
|
|
},
|
|
})
|
|
}
|
|
|
|
func IncrementPackets(country string, direction Direction, count uint64, retransmit bool) {
|
|
var transmission TransmissionType
|
|
if retransmit {
|
|
transmission = TransmissionRetransmit
|
|
} else {
|
|
transmission = TransmissionInitial
|
|
}
|
|
promPacketTotal.WithLabelValues(string(direction), string(transmission), country).Add(float64(count))
|
|
|
|
if direction == Incoming {
|
|
packetsIn.Add(count)
|
|
} else {
|
|
packetsOut.Add(count)
|
|
if retransmit {
|
|
retransmitPackets.Add(count)
|
|
}
|
|
}
|
|
}
|
|
|
|
func IncrementBytes(country string, direction Direction, count uint64, retransmit bool) {
|
|
var transmission TransmissionType
|
|
if retransmit {
|
|
transmission = TransmissionRetransmit
|
|
} else {
|
|
transmission = TransmissionInitial
|
|
}
|
|
promPacketBytes.WithLabelValues(string(direction), string(transmission), country).Add(float64(count))
|
|
|
|
if direction == Incoming {
|
|
bytesIn.Add(count)
|
|
} else {
|
|
bytesOut.Add(count)
|
|
if retransmit {
|
|
retransmitBytes.Add(count)
|
|
}
|
|
}
|
|
}
|
|
|
|
func IncrementRTCP(country string, direction Direction, nack, pli, fir uint32) {
|
|
if nack > 0 {
|
|
promNackTotal.WithLabelValues(string(direction), country).Add(float64(nack))
|
|
nackTotal.Add(uint64(nack))
|
|
}
|
|
if pli > 0 {
|
|
promPliTotal.WithLabelValues(string(direction), country).Add(float64(pli))
|
|
}
|
|
if fir > 0 {
|
|
promFirTotal.WithLabelValues(string(direction), country).Add(float64(fir))
|
|
}
|
|
}
|
|
|
|
func RecordPacketLoss(
|
|
country string,
|
|
direction Direction,
|
|
trackSource livekit.TrackSource,
|
|
trackType livekit.TrackType,
|
|
lost uint32,
|
|
total uint32,
|
|
) {
|
|
if total > 0 {
|
|
promPacketLoss.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(lost) / float64(total) * 100)
|
|
}
|
|
if lost > 0 {
|
|
promPacketLossTotal.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Add(float64(lost))
|
|
}
|
|
}
|
|
|
|
func RecordPacketOutOfOrder(country string, direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, ooo, total uint32) {
|
|
if total > 0 {
|
|
promPacketOutOfOrder.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(ooo) / float64(total) * 100)
|
|
}
|
|
if ooo > 0 {
|
|
promPacketOutOfOrderTotal.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Add(float64(ooo))
|
|
}
|
|
}
|
|
|
|
func RecordJitter(country string, direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, jitter uint32) {
|
|
if jitter > 0 {
|
|
promJitter.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(jitter))
|
|
}
|
|
}
|
|
|
|
func RecordRTT(country string, direction Direction, trackSource livekit.TrackSource, trackType livekit.TrackType, rtt uint32) {
|
|
if rtt > 0 {
|
|
promRTT.WithLabelValues(string(direction), trackSource.String(), trackType.String(), country).Observe(float64(rtt))
|
|
}
|
|
}
|
|
|
|
func IncrementParticipantJoin(join uint32) {
|
|
if join > 0 {
|
|
participantSignalConnected.Add(uint64(join))
|
|
promParticipantJoin.WithLabelValues("signal_connected").Add(float64(join))
|
|
}
|
|
}
|
|
|
|
func IncrementParticipantJoinFail(join uint32) {
|
|
if join > 0 {
|
|
promParticipantJoin.WithLabelValues("signal_failed").Add(float64(join))
|
|
}
|
|
}
|
|
|
|
func IncrementParticipantRtcInit(join uint32) {
|
|
if join > 0 {
|
|
participantRTCInit.Add(uint64(join))
|
|
promParticipantJoin.WithLabelValues("rtc_init").Add(float64(join))
|
|
}
|
|
}
|
|
|
|
func IncrementParticipantRtcConnected(join uint32) {
|
|
if join > 0 {
|
|
participantRTCConnected.Add(uint64(join))
|
|
promParticipantJoin.WithLabelValues("rtc_connected").Add(float64(join))
|
|
}
|
|
}
|
|
|
|
func IncrementParticipantRtcCanceled(numCancels uint64) {
|
|
if numCancels > 0 {
|
|
participantRTCConnected.Add(numCancels)
|
|
promParticipantJoin.WithLabelValues("rtc_canceled").Add(float64(numCancels))
|
|
}
|
|
}
|
|
|
|
func AddConnection(direction Direction) {
|
|
promConnections.WithLabelValues(string(direction)).Add(1)
|
|
}
|
|
|
|
func SubConnection(direction Direction) {
|
|
promConnections.WithLabelValues(string(direction)).Sub(1)
|
|
}
|
|
|
|
func RecordForwardLatencySample(forwardLatency int64) {
|
|
promForwardLatencyHist.Observe(float64(forwardLatency))
|
|
}
|
|
|
|
func RecordForwardLatency(longTermLatencyAvg uint32) {
|
|
forwardLatency.Store(longTermLatencyAvg)
|
|
promForwardLatency.Set(float64(longTermLatencyAvg))
|
|
}
|
|
|
|
func RecordForwardJitter(longTermJitterAvg uint32) {
|
|
forwardJitter.Store(longTermJitterAvg)
|
|
promForwardJitter.Set(float64(longTermJitterAvg))
|
|
}
|