feat: Implement a Prometheus histogram of rtp transit time.

jitsi-media-transform/src/main/kotlin/org/jitsi/nlj/stats/DelayStats.kt

@@ -16,36 +16,24 @@
 
 package org.jitsi.nlj.stats
 
-import org.jitsi.nlj.PacketInfo
 import org.jitsi.utils.OrderedJsonObject
 import org.jitsi.utils.stats.BucketStats
-import java.time.Clock
-import java.time.Duration
 import java.util.concurrent.atomic.LongAdder
 
 open class DelayStats(thresholds: List<Long> = defaultThresholds) :
     BucketStats(thresholds, "_delay_ms", "_ms") {
 
-    fun addDelay(delay: Duration) {
-        addDelay(delay.toMillis())
-    }
     fun addDelay(delayMs: Long) = addValue(delayMs)
 
     companion object {
         val defaultThresholds = listOf(0, 5, 50, 500, Long.MAX_VALUE)
     }
 }
 
-class PacketDelayStats(
-    thresholds: List<Long> = defaultThresholds,
-    private val clock: Clock = Clock.systemUTC()
-) : DelayStats(thresholds) {
+class PacketDelayStats(thresholds: List<Long> = defaultThresholds) : DelayStats(thresholds) {
     private val numPacketsWithoutTimestamps = LongAdder()
-    fun addPacket(packetInfo: PacketInfo) {
-        packetInfo.receivedTime?.let { addDelay(Duration.between(it, clock.instant())) } ?: run {
-            numPacketsWithoutTimestamps.increment()
-        }
-    }
+
+    fun addUnknown() = numPacketsWithoutTimestamps.increment()
 
     override fun toJson(format: Format): OrderedJsonObject {
         return super.toJson(format).apply {

jvb/src/main/java/org/jitsi/videobridge/EndpointMessageTransport.java

@@ -562,7 +562,7 @@ public BridgeChannelMessage endpointStats(@NotNull EndpointStats message)
 
         Conference conference = endpoint.getConference();
 
-        if (conference == null || conference.isExpired())
+        if (conference.isExpired())
         {
             getLogger().warn("Unable to send EndpointStats, conference is null or expired");
             return null;

jvb/src/main/java/org/jitsi/videobridge/Videobridge.java

@@ -607,7 +607,12 @@ public OrderedJsonObject getDebugState(String conferenceId, String endpointId, b
         debugState.put("time", System.currentTimeMillis());
 
         debugState.put("load-management", jvbLoadManager.getStats());
-        debugState.put("overall_bridge_jitter", PacketTransitStats.getBridgeJitter());
+
+        Double jitter = PacketTransitStats.getBridgeJitter();
+        if (jitter != null)
+        {
+            debugState.put("overall_bridge_jitter", jitter);
+        }
 
         JSONObject conferences = new JSONObject();
         debugState.put("conferences", conferences);

jvb/src/main/kotlin/org/jitsi/videobridge/stats/PacketTransitStats.kt

@@ -15,14 +15,20 @@
  */
 package org.jitsi.videobridge.stats
 
+import org.jitsi.config.JitsiConfig
+import org.jitsi.metaconfig.config
 import org.jitsi.nlj.PacketInfo
 import org.jitsi.nlj.stats.BridgeJitterStats
 import org.jitsi.nlj.stats.PacketDelayStats
 import org.jitsi.rtp.extensions.looksLikeRtcp
 import org.jitsi.rtp.extensions.looksLikeRtp
 import org.jitsi.utils.OrderedJsonObject
+import org.jitsi.utils.logging2.createLogger
 import org.jitsi.utils.stats.BucketStats
 import org.jitsi.videobridge.Endpoint
+import org.jitsi.videobridge.metrics.VideobridgeMetricsContainer
+import java.time.Clock
+import java.time.Duration
 
 /**
  * Track how long it takes for all RTP and RTCP packets to make their way through the bridge.
@@ -31,18 +37,66 @@ import org.jitsi.videobridge.Endpoint
  * for packets going out to all endpoints.
  */
 object PacketTransitStats {
-    private val rtpPacketDelayStats = PacketDelayStats()
-    private val rtcpPacketDelayStats = PacketDelayStats()
+    private val jsonEnabled: Boolean by config {
+        "videobridge.stats.transit-time.enable-json".from(JitsiConfig.newConfig)
+    }
+    private val prometheusEnabled: Boolean by config {
+        "videobridge.stats.transit-time.enable-prometheus".from(JitsiConfig.newConfig)
+    }
+    private val jitterEnabled: Boolean by config {
+        "videobridge.stats.transit-time.enable-jitter".from(JitsiConfig.newConfig)
+    }
+    private val enabled = jsonEnabled || prometheusEnabled || jitterEnabled
+
+    private val logger = createLogger()
+    private val clock: Clock = Clock.systemUTC()
+
+    init {
+        logger.info(
+            "Initializing, jsonEnabled=$jsonEnabled, prometheusEnabled=$prometheusEnabled, " +
+                "jitterEnabled=$jitterEnabled"
+        )
+    }
+
+    private val rtpPacketDelayStats = if (jsonEnabled) PacketDelayStats() else null
+    private val rtcpPacketDelayStats = if (jsonEnabled) PacketDelayStats() else null
+    private val prometheusRtpDelayStats = if (prometheusEnabled) {
+        PrometheusPacketDelayStats("rtp_transit_time")
+    } else {
+        null
+    }
+    private val prometheusRtcpDelayStats = if (prometheusEnabled) {
+        PrometheusPacketDelayStats("rtcp_transit_time")
+    } else {
+        null
+    }
 
-    private val bridgeJitterStats = BridgeJitterStats()
+    private val bridgeJitterStats = if (jitterEnabled) BridgeJitterStats() else null
 
     @JvmStatic
     fun packetSent(packetInfo: PacketInfo) {
+        if (!enabled) {
+            return
+        }
+
+        val delayMs = packetInfo.receivedTime?.let { Duration.between(it, clock.instant()).toMillis() }
         if (packetInfo.packet.looksLikeRtp()) {
-            rtpPacketDelayStats.addPacket(packetInfo)
-            bridgeJitterStats.packetSent(packetInfo)
+            if (delayMs != null) {
+                rtpPacketDelayStats?.addDelay(delayMs)
+                prometheusRtpDelayStats?.addDelay(delayMs)
+                bridgeJitterStats?.packetSent(packetInfo)
+            } else {
+                rtpPacketDelayStats?.addUnknown()
+                prometheusRtpDelayStats?.addUnknown()
+            }
         } else if (packetInfo.packet.looksLikeRtcp()) {
-            rtcpPacketDelayStats.addPacket(packetInfo)
+            if (delayMs != null) {
+                rtcpPacketDelayStats?.addDelay(delayMs)
+                prometheusRtcpDelayStats?.addDelay(delayMs)
+            } else {
+                rtcpPacketDelayStats?.addUnknown()
+                prometheusRtcpDelayStats?.addUnknown()
+            }
         }
     }
 
@@ -51,16 +105,44 @@ object PacketTransitStats {
         get() {
             val stats = OrderedJsonObject()
             stats["e2e_packet_delay"] = getPacketDelayStats()
-            stats["overall_bridge_jitter"] = bridgeJitterStats.jitter
+            bridgeJitterStats?.let {
+                stats["overall_bridge_jitter"] = it.jitter
+            }
             return stats
         }
 
     @JvmStatic
     val bridgeJitter
-        get() = bridgeJitterStats.jitter
+        get() = bridgeJitterStats?.jitter
 
     private fun getPacketDelayStats() = OrderedJsonObject().apply {
-        put("rtp", rtpPacketDelayStats.toJson(format = BucketStats.Format.CumulativeRight))
-        put("rtcp", rtcpPacketDelayStats.toJson(format = BucketStats.Format.CumulativeRight))
+        rtpPacketDelayStats?.let {
+            put("rtp", it.toJson(format = BucketStats.Format.CumulativeRight))
+        }
+        rtcpPacketDelayStats?.let {
+            put("rtcp", it.toJson(format = BucketStats.Format.CumulativeRight))
+        }
+    }
+}
+
+class PrometheusPacketDelayStats(name: String) {
+    private val histogram = VideobridgeMetricsContainer.instance.registerHistogram(
+        name,
+        "Packet delay stats for $name",
+        0.0,
+        5.0,
+        50.0,
+        500.0
+    )
+    private val numPacketsWithoutTimestamps = VideobridgeMetricsContainer.instance.registerCounter(
+        "${name}_unknown_delay",
+        "Number of packets without an unknown delay ($name)"
+    )
+
+    fun addUnknown() {
+        numPacketsWithoutTimestamps.inc()
+    }
+    fun addDelay(delayMs: Long) {
+        histogram.histogram.observe(delayMs.toDouble())
     }
 }

jvb/src/main/resources/reference.conf

@@ -219,6 +219,19 @@ videobridge {
   stats {
     // The interval at which stats are gathered.
     interval = 5 seconds
+
+    // Statistics about the transit time of RTP/RTCP packets. Note that the collection code for the JSON and Prometheus
+    // outputs is different, and each has a slight performance impact, so the format(s) that are not needed should be
+    // kept disabled.
+    transit-time {
+      // Enable collection of transit time stats in JSON format. Available through /debug/jvb/stats/transit-time
+      enable-json = true
+      // Enable collection of transit time stats in Prometheus. Available through /metrics.
+      enable-prometheus = false
+      // Enable collection of internal jitter (difference in processing time). Available through
+      // /debug/jvb/stats/transit-time
+      enable-jitter = false
+    }
   }
   websockets {
     enabled = false