Colibri msg initiated pcap capture

jitsi-media-transform/pom.xml

@@ -135,6 +135,12 @@
             <version>2.3.1</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.9.0</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -63,6 +63,8 @@ abstract class RtpReceiver :
     abstract fun forceMuteAudio(shouldMute: Boolean)
 
     abstract fun forceMuteVideo(shouldMute: Boolean)
+
+    abstract fun setComplianceRecording(comRec: String?, contextId: String?)
 }
 
 interface RtpReceiverEventHandler {

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

@@ -32,6 +32,7 @@ import org.jitsi.nlj.stats.RtpReceiverStats
 import org.jitsi.nlj.transform.NodeEventVisitor
 import org.jitsi.nlj.transform.NodeStatsVisitor
 import org.jitsi.nlj.transform.NodeTeardownVisitor
+import org.jitsi.nlj.transform.node.CompliancePcapWriter
 import org.jitsi.nlj.transform.node.ConsumerNode
 import org.jitsi.nlj.transform.node.Node
 import org.jitsi.nlj.transform.node.PacketLossConfig
@@ -140,6 +141,7 @@ class RtpReceiverImpl @JvmOverloads constructor(
         })
     }
     private val toggleablePcapWriter = ToggleablePcapWriter(logger, "$id-rx")
+    private val compliancePcapWriter = CompliancePcapWriter(logger, streamInformationStore, id, "ne")
     private val videoBitrateCalculator = VideoBitrateCalculator(parentLogger)
     private val audioBitrateCalculator = BitrateCalculator("Audio bitrate calculator")
 
@@ -225,6 +227,7 @@ class RtpReceiverImpl @JvmOverloads constructor(
                         // This reads audio levels from packets that use cryptex. TODO: should it go in the Audio path?
                         node(audioLevelReader.postDecryptNode)
                         node(toggleablePcapWriter.newObserverNode())
+                        node(compliancePcapWriter.newObserverNode())
                         node(statsTracker)
                         node(PaddingTermination(logger))
                         demux("Media Type") {
@@ -350,9 +353,14 @@ class RtpReceiverImpl @JvmOverloads constructor(
         NodeTeardownVisitor().visit(inputTreeRoot)
         incomingPacketQueue.close()
         toggleablePcapWriter.disable()
+        compliancePcapWriter.disable()
     }
 
     override fun onRttUpdate(newRttMs: Double) {
         remoteBandwidthEstimator.onRttUpdate(newRttMs)
     }
+
+    override fun setComplianceRecording(comRec: String?, contextId: String?) {
+        compliancePcapWriter.configure(comRec, contextId)
+    }
 }

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

@@ -47,6 +47,7 @@ abstract class RtpSender :
     abstract fun setFeature(feature: Features, enabled: Boolean)
     abstract fun isFeatureEnabled(feature: Features): Boolean
     abstract fun tearDown()
+    abstract fun setComplianceRecording(comRec: String?, contextId: String?)
 
     abstract val bandwidthEstimator: BandwidthEstimator
 }

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

@@ -32,6 +32,7 @@ import org.jitsi.nlj.transform.NodeEventVisitor
 import org.jitsi.nlj.transform.NodeStatsVisitor
 import org.jitsi.nlj.transform.NodeTeardownVisitor
 import org.jitsi.nlj.transform.node.AudioRedHandler
+import org.jitsi.nlj.transform.node.CompliancePcapWriter
 import org.jitsi.nlj.transform.node.ConsumerNode
 import org.jitsi.nlj.transform.node.Node
 import org.jitsi.nlj.transform.node.PacketCacher
@@ -108,6 +109,7 @@ class RtpSenderImpl(
     private val srtpEncryptWrapper = SrtpEncryptNode()
     private val srtcpEncryptWrapper = SrtcpEncryptNode()
     private val toggleablePcapWriter = ToggleablePcapWriter(logger, "$id-tx")
+    private val compliancePcapWriter = CompliancePcapWriter(logger, streamInformationStore, id, "fe")
     private val outgoingPacketCache = PacketCacher()
     private val absSendTime = AbsSendTime(streamInformationStore)
     private val statsTracker = OutgoingStatisticsTracker()
@@ -148,6 +150,7 @@ class RtpSenderImpl(
             node(TccSeqNumTagger(transportCcEngine, streamInformationStore))
             node(HeaderExtEncoder(streamInformationStore, logger))
             node(toggleablePcapWriter.newObserverNode())
+            node(compliancePcapWriter.newObserverNode())
             node(srtpEncryptWrapper)
             node(packetStreamStats.createNewNode())
             node(PacketLossNode(packetLossConfig), condition = { packetLossConfig.enabled })
@@ -321,6 +324,11 @@ class RtpSenderImpl(
         NodeTeardownVisitor().reverseVisit(outputPipelineTerminationNode)
         incomingPacketQueue.close()
         toggleablePcapWriter.disable()
+        compliancePcapWriter.disable()
+    }
+
+    override fun setComplianceRecording(comRec: String?, contextId: String?) {
+        compliancePcapWriter.configure(comRec, contextId)
     }
 
     companion object {

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

@@ -379,6 +379,11 @@ class Transceiver(
         return rtpReceiver.isFeatureEnabled(feature)
     }
 
+    fun setComplianceRecording(comRec: String?, contextId: String?) {
+        rtpReceiver.setComplianceRecording(comRec, contextId)
+        rtpSender.setComplianceRecording(comRec, contextId)
+    }
+
     companion object {
         init {
 //            Node.plugins.add(BufferTracePlugin)

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

@@ -0,0 +1,263 @@
+/*
+ * Copyright @ 2018 - Present, 8x8 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 org.jitsi.nlj.transform.node
+
+import com.google.gson.GsonBuilder
+import com.google.gson.JsonElement
+import com.google.gson.JsonParser
+import org.jitsi.config.JitsiConfig
+import org.jitsi.metaconfig.config
+import org.jitsi.metaconfig.from
+import org.jitsi.nlj.PacketInfo
+import org.jitsi.nlj.rtp.AudioRtpPacket
+import org.jitsi.nlj.rtp.VideoRtpPacket
+import org.jitsi.nlj.util.ReadOnlyStreamInformationStore
+import org.jitsi.utils.MediaType
+import org.jitsi.utils.logging2.Logger
+import org.json.simple.JSONArray
+import org.json.simple.JSONObject
+import java.io.File
+import java.nio.file.Files
+import java.nio.file.Paths
+
+class CompliancePcapWriter(
+    private val logger: Logger,
+    private val streamInformationStore: ReadOnlyStreamInformationStore,
+    private val id: String,
+    private val captureEnd: String
+) {
+    private var writer: PcapWriter? = null
+    private val lock = Any()
+    private var contextId: String? = null
+    private var mode: String = CAP_MODE_NONE
+
+    private var captureAudio = false
+    private var captureVideo = false
+
+    fun capId(): String {
+        return "${id}_$captureEnd"
+    }
+
+    fun filename(): String {
+        return "$basePath/${contextId}_${capId()}"
+    }
+
+    fun pcapFilename(): String {
+        return "${filename()}.pcap"
+    }
+
+    fun jsonFilename(): String {
+        return "${filename()}.json"
+    }
+
+    fun validJmtConfig(): Boolean {
+        if (!allowed) {
+            logger.info("Compliance recording:${capId()} is not allowed in jmt.compliance-recording.enabled")
+            return false
+        }
+
+        if (basePath.isEmpty()) {
+            logger.error("Compliance recording:${capId()} jmt.compliance-recording.base-path is not configured")
+            return false
+        }
+
+        if (!Files.isDirectory(Paths.get(basePath))) {
+            logger.error("Compliance recording:${capId()} base-path:$basePath is not a valid directory path")
+            return false
+        }
+
+        return true
+    }
+
+    fun writeMetadata(): Boolean {
+        val meta = JSONObject()
+
+        meta.put("endpoint_id", id)
+        meta.put("context_id", contextId)
+        meta.put("capture_end", captureEnd)
+        meta.put("capture_mode", mode)
+
+        val payloadsMap = JSONArray()
+        streamInformationStore.rtpPayloadTypes.forEach {
+            if ((it.value.mediaType == MediaType.AUDIO && (mode == CAP_MODE_AUDIO || mode == CAP_MODE_AUDIO_VIDEO)) ||
+                (it.value.mediaType == MediaType.VIDEO && (mode == CAP_MODE_VIDEO || mode == CAP_MODE_AUDIO_VIDEO))
+            ) {
+                val entry = JSONObject()
+                entry["payload_type"] = it.value.pt
+                entry["encoding"] = it.value.encoding.toString().lowercase()
+                entry["media_type"] = it.value.mediaType.name.lowercase()
+                entry["clock_rate"] = it.value.clockRate
+                payloadsMap.add(entry)
+            }
+        }
+
+        meta.put("payload_map", payloadsMap)
+
+        // json-simple produces one long flat line
+        // File(jsonFilename()).writeText(meta.toJSONString())
+
+        // gson produces elements on separate lines
+        val gson = GsonBuilder().setPrettyPrinting().create()
+        val je: JsonElement = JsonParser.parseString(meta.toJSONString())
+
+        try {
+            File(jsonFilename()).writeText(gson.toJson(je))
+        } catch (ex: Exception) {
+            logger.error("Compliance recording:${capId()} exception: $ex")
+            return false
+        }
+
+        return true
+    }
+
+    fun configure(newMode: String?, newContextId: String?) {
+        // TODO - how to trace long line exceeding 120 chars
+        logger.info("Compliance recording:${capId()} configure mode:$mode -> $newMode")
+        logger.info("Compliance recording:${capId()} contextId:$contextId -> $newContextId")
+
+        // Disable conditions
+        if (newMode.isNullOrEmpty() || newMode == CAP_MODE_NONE) {
+            disable()
+            synchronized(lock) {
+                mode = "none"
+                contextId = null
+            }
+            return
+        }
+
+        // Checks for valid enable conditions
+        if (!validJmtConfig()) {
+            return
+        }
+
+        if (newMode !in listOf(CAP_MODE_AUDIO, CAP_MODE_VIDEO, CAP_MODE_AUDIO_VIDEO)) {
+            logger.error("Compliance recording:${capId()} invalid mode:$newMode")
+            return
+        }
+
+        if (newContextId.isNullOrEmpty()) {
+            logger.error("Compliance recording:${capId()} invalid context-id:$newContextId")
+            return
+        }
+
+        synchronized(lock) {
+            if (mode == newMode && contextId == newContextId) {
+                // nothing to do
+                logger.warn("Compliance recording:${capId()} duplicate configuration request ignored.")
+                return
+            }
+        }
+
+        if (isEnabled()) {
+            // reconfiguring to different mode on the fly?
+            logger.warn("Compliance recording:${capId()} is already enabled - resetting by disabling first.")
+            disable()
+        }
+
+        synchronized(lock) {
+            mode = newMode
+            contextId = newContextId
+            if (!writeMetadata()) {
+                contextId = null
+                mode = "none"
+                return
+            }
+        }
+
+        // Enable / open the writer with the new mode and context-id
+        enable()
+    }
+
+    fun enable() {
+        logger.info("Compliance recording:${capId()} enable ${pcapFilename()}")
+
+        if (!validJmtConfig()) {
+            return
+        }
+
+        if (contextId.isNullOrEmpty()) {
+            logger.error("Compliance recording:${capId()} is not configured with context-id")
+            return
+        }
+
+        synchronized(lock) {
+            captureAudio = false
+            captureVideo = false
+            when (mode) {
+                CAP_MODE_AUDIO -> captureAudio = true
+                CAP_MODE_VIDEO -> captureVideo = true
+                CAP_MODE_AUDIO_VIDEO -> {
+                    captureAudio = true
+                    captureVideo = true
+                }
+                else -> {
+                    logger.error("Compliance recording:${capId()} invalid mode:$mode")
+                    return
+                }
+            }
+
+            if (writer == null) {
+                logger.info("Compliance recording:${capId()} enable ${pcapFilename()}")
+                writer = PcapWriter(logger, pcapFilename())
+            }
+        }
+    }
+
+    fun disable() {
+        synchronized(lock) {
+            captureAudio = false
+            captureVideo = false
+
+            if (writer != null) {
+                logger.info("Compliance recording:${capId()} disable ${pcapFilename()}")
+                writer?.close()
+                writer = null
+            }
+        }
+    }
+
+    fun isEnabled(): Boolean = writer != null
+
+    fun newObserverNode(): Node = PcapWriterNode("Compliance recording:${capId()} pcap writer")
+
+    private inner class PcapWriterNode(name: String) : ObserverNode(name) {
+        override fun observe(packetInfo: PacketInfo) {
+            synchronized(lock) {
+                if (packetInfo.packet is AudioRtpPacket) {
+                    if (captureAudio) {
+                        writer?.processPacket(packetInfo)
+                    }
+                } else if (packetInfo.packet is VideoRtpPacket) {
+                    if (captureVideo) {
+                        writer?.processPacket(packetInfo)
+                    }
+                }
+            }
+        }
+
+        override fun trace(f: () -> Unit) = f.invoke()
+    }
+
+    companion object {
+        private val allowed: Boolean by config("jmt.compliance-recording.enabled".from(JitsiConfig.newConfig))
+        private val basePath: String by config("jmt.compliance-recording.base-path".from(JitsiConfig.newConfig))
+
+        private const val CAP_MODE_NONE = "none"
+        private const val CAP_MODE_AUDIO = "audio"
+        private const val CAP_MODE_VIDEO = "video"
+        private const val CAP_MODE_AUDIO_VIDEO = "audio-video"
+    }
+}

jitsi-media-transform/src/main/resources/reference.conf

@@ -160,4 +160,12 @@ jmt {
       }
     }
   }
+
+  compliance-recording {
+    enabled=true
+
+    // Base path for storing the compliance recording pcap files
+    base-path="/tmp/com-rec"
+  }
+
 }