Add a timeseries log for bitrate as received and sent. (#2290)

Author: JonathanLennox

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

@@ -125,7 +125,7 @@ class RtpReceiverImpl @JvmOverloads constructor(
     private val silenceDiscarder = DiscardableDiscarder("Silence discarder", false)
     private val paddingOnlyDiscarder = DiscardableDiscarder("Padding-only discarder", true)
     private val statsTracker = IncomingStatisticsTracker(streamInformationStore)
-    private val packetStreamStats = PacketStreamStatsNode()
+    private val packetStreamStats = PacketStreamStatsNode(diagnosticContext, "receive")
     private val rtcpRrGenerator = RtcpRrGenerator(backgroundExecutor, rtcpSender, statsTracker) {
         remoteBandwidthEstimator.createRemb()?.let {
             listOf(it)

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

@@ -115,7 +115,7 @@ class RtpSenderImpl(
     private val headerExtensionStripper = HeaderExtStripper(streamInformationStore)
     private val absSendTime = AbsSendTime(streamInformationStore)
     private val statsTracker = OutgoingStatisticsTracker()
-    private val packetStreamStats = PacketStreamStatsNode()
+    private val packetStreamStats = PacketStreamStatsNode(diagnosticContext, "send")
     private val rtcpSrUpdater = RtcpSrUpdater(statsTracker)
     private val keyframeRequester = KeyframeRequester(streamInformationStore, logger)
     private val probingDataSender: ProbingDataSender

jitsi-media-transform/src/main/kotlin/org/jitsi/nlj/transform/node/PacketStreamStatsNode.kt

@@ -17,13 +17,19 @@ package org.jitsi.nlj.transform.node
 
 import org.jitsi.nlj.PacketInfo
 import org.jitsi.nlj.stats.PacketStreamStats
+import org.jitsi.utils.logging.DiagnosticContext
+import org.jitsi.utils.logging.TimeSeriesLogger
 
 /**
  * A [Node] which keeps track of the basic statistics for a stream of packets (packet and bit rates)
  *
  * @author Boris Grozev
  */
-class PacketStreamStatsNode(private val packetStreamStats: PacketStreamStats = PacketStreamStats()) :
+class PacketStreamStatsNode(
+    private val diagnosticContext: DiagnosticContext,
+    private val direction: String,
+    private val packetStreamStats: PacketStreamStats = PacketStreamStats()
+) :
     ObserverNode("PacketStreamStats") {
 
     override fun observe(packetInfo: PacketInfo) {
@@ -32,13 +38,25 @@ class PacketStreamStatsNode(private val packetStreamStats: PacketStreamStats = P
 
     override fun trace(f: () -> Unit) = f.invoke()
 
-    fun snapshot() = packetStreamStats.snapshot()
+    fun snapshot() = packetStreamStats.snapshot().also {
+        if (timeseriesLogger.isTraceEnabled) {
+            timeseriesLogger.trace(
+                diagnosticContext.makeTimeSeriesPoint("${direction}_packet_stream_stats")
+                    .addField("bitrate_bps", it.bitrate.bps)
+                    .addField("packet_rate", it.packetRate)
+            )
+        }
+    }
 
     fun getBitrate() = snapshot().bitrate
 
     /**
      * Creates a new [Node] instance which shares the same [packetStreamStats]. Useful when we want to add nodes to
      * different branches of a [Node] tree.
      */
-    fun createNewNode() = PacketStreamStatsNode(packetStreamStats)
+    fun createNewNode() = PacketStreamStatsNode(diagnosticContext, direction, packetStreamStats)
+
+    companion object {
+        private val timeseriesLogger = TimeSeriesLogger.getTimeSeriesLogger(PacketStreamStatsNode::class.java)
+    }
 }