feat(telemetry): add client telemetry through the shared Rust core

.changeset/client-telemetry.md

@@ -0,0 +1,5 @@
+---
+"client-sdk-android": minor
+---
+
+Client telemetry through the shared Rust core: each Room reports its connect, reconnect, publish and subscribe spans, RTC statistics, SDK warnings and errors and device state to its LiveKit Cloud project when the token carries the observability grant; apps can add `Room.emitTelemetryEvent(name, attributes)` and `Room.setTelemetryAttribute(key, value)`, and opt out with `LiveKit.disableTelemetry()`.

.github/workflows/android.yml

@@ -58,6 +58,28 @@ jobs:
     - name: Build and test with Gradle
       run: ./gradlew assembleRelease livekit-android-test:testRelease
 
+      # The telemetry end-to-end test runs the real Rust core under Robolectric: a host build of
+      # livekit-uniffi at the release gradle/libs.versions.toml pins, posting OTLP to a collector
+      # that writes what the test reads back (livekit-android-test/src/test/resources/telemetry/otelcol.yaml).
+      # Only this step sets LK_TELEMETRY_ENDPOINT, and a missing collector or library fails the job
+      # instead of skipping the test. The telemetry package's platform tests that need the core run here too.
+    - name: Telemetry E2E test
+      run: |
+        version="$(sed -n 's/^livekit-uniffi = "\(.*\)"/\1/p' gradle/libs.versions.toml)"
+        git clone --depth 1 --branch "livekit-uniffi/v$version" https://github.com/livekit/rust-sdks.git "$RUNNER_TEMP/rust-sdks"
+        (cd "$RUNNER_TEMP/rust-sdks" && cargo build -p livekit-uniffi)
+        curl -sSfL -o "$RUNNER_TEMP/otelcol.tar.gz" "https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.162.0/otelcol-contrib_0.162.0_linux_amd64.tar.gz"
+        echo "fcc063749f730f8c21fe29f2d340ff174f5f1c5885bd3156fb6c985a3036fcc3  $RUNNER_TEMP/otelcol.tar.gz" | sha256sum -c -
+        tar -xzf "$RUNNER_TEMP/otelcol.tar.gz" -C "$RUNNER_TEMP" otelcol-contrib
+        nohup "$RUNNER_TEMP/otelcol-contrib" --config livekit-android-test/src/test/resources/telemetry/otelcol.yaml > "$RUNNER_TEMP/otelcol.log" 2>&1 &
+        timeout 30 bash -c 'until curl -s -o /dev/null http://127.0.0.1:4319; do sleep 1; done' || { cat "$RUNNER_TEMP/otelcol.log"; exit 1; }
+        ./gradlew livekit-android-test:testReleaseUnitTest --tests 'io.livekit.android.telemetry.*' --rerun \
+          -PlivekitUniffiLibraryPath="$RUNNER_TEMP/rust-sdks/target/debug"
+        grep -q 'skipped="0"' livekit-android-test/build/test-results/testReleaseUnitTest/TEST-io.livekit.android.telemetry.TelemetryMockE2ETest.xml \
+          || { echo "TelemetryMockE2ETest was skipped"; exit 1; }
+      env:
+        LK_TELEMETRY_ENDPOINT: http://127.0.0.1:4319
+
     - name: Run Detekt
       run: ./gradlew livekit-android-sdk:detektRelease
 

gradle/libs.versions.toml

@@ -33,6 +33,7 @@ lifecycleProcess = "2.8.7"
 agp = "8.7.2"
 kotlin = "1.9.25"
 livekit-uniffi = "0.1.12"
+jna = "5.16.0"
 
 [libraries]
 livekit-uniffi = { module = "io.livekit:livekit-uniffi-android", version.ref = "livekit-uniffi" }
@@ -103,6 +104,8 @@ mockito-inline = { module = "org.mockito:mockito-inline", version = "4.11.0" }
 byte-buddy = { module = "net.bytebuddy:byte-buddy", version = "1.14.3" }
 
 robolectric = { module = "org.robolectric:robolectric", version = "4.14.1" }
+# JVM natives (libjnidispatch) for the Rust core under Robolectric; the AAR variant only ships Android ABIs.
+jna = { module = "net.java.dev.jna:jna", version.ref = "jna" }
 turbine = { module = "app.cash.turbine:turbine", version = "1.0.0" }
 appcompat = { group = "androidx.appcompat", name = "appcompat", version.ref = "appcompat" }
 material = { group = "com.google.android.material", name = "material", version.ref = "material" }

livekit-android-sdk/detekt-baseline-release.xml

@@ -21,7 +21,7 @@
     <ID>CyclomaticComplexMethod:PreconnectAudioBuffer.kt$@Deprecated("Set AudioTrackPublishDefaults.preconnect = true on the RoomOptions instead.") suspend fun &lt;T> Room.withPreconnectAudio( timeout: Duration = TIMEOUT, topic: String = DEFAULT_TOPIC, onError: ((e: Exception) -> Unit)? = null, operation: suspend () -> T, )</ID>
     <ID>CyclomaticComplexMethod:PreconnectAudioBuffer.kt$internal suspend fun Room.startPreconnectAudioJob( roomScope: CoroutineScope, timeout: Duration = TIMEOUT, topic: String = DEFAULT_TOPIC ): () -> Unit</ID>
     <ID>CyclomaticComplexMethod:RTCEngine.kt$RTCEngine$@CheckResult internal suspend fun sendData(dataPacket: LivekitModels.DataPacket): Result&lt;Unit></ID>
-    <ID>CyclomaticComplexMethod:RTCEngine.kt$RTCEngine$@Synchronized @VisibleForTesting(otherwise = VisibleForTesting.PACKAGE_PRIVATE) fun reconnect()</ID>
+    <ID>CyclomaticComplexMethod:RTCEngine.kt$RTCEngine$@Synchronized internal fun reconnect(reason: ReconnectReason)</ID>
     <ID>CyclomaticComplexMethod:RTCEngine.kt$RTCEngine$fun onMessage(dataChannel: DataChannel, buffer: DataChannel.Buffer?)</ID>
     <ID>CyclomaticComplexMethod:RTCEngine.kt$RTCEngine$private fun makeRTCConfig( serverResponse: Either&lt;JoinResponse, ReconnectResponse>, connectOptions: ConnectOptions, ): RTCConfiguration</ID>
     <ID>CyclomaticComplexMethod:Room.kt$Room$@Throws(Exception::class) suspend fun connect(url: String, token: String, options: ConnectOptions = ConnectOptions())</ID>
@@ -36,7 +36,7 @@
     <ID>LargeClass:RTCEngine.kt$RTCEngine : Listener</ID>
     <ID>LargeClass:Room.kt$Room : ListenerParticipantListenerRpcManagerIncomingDataStreamManager</ID>
     <ID>LargeClass:SignalClient.kt$SignalClient : WebSocketListener</ID>
-    <ID>LongMethod:RTCEngine.kt$RTCEngine$@Synchronized @VisibleForTesting(otherwise = VisibleForTesting.PACKAGE_PRIVATE) fun reconnect()</ID>
+    <ID>LongMethod:RTCEngine.kt$RTCEngine$@Synchronized internal fun reconnect(reason: ReconnectReason)</ID>
     <ID>LongMethod:Room.kt$Room$@Throws(Exception::class) suspend fun connect(url: String, token: String, options: ConnectOptions = ConnectOptions())</ID>
     <ID>LongMethod:SignalClient.kt$SignalClient$private fun handleSignalResponseImpl(ws: WebSocket, response: LivekitRtc.SignalResponse, encoded: ByteArray)</ID>
     <ID>LongParameterList:AudioBufferCallbackDispatcher.kt$AudioBufferCallback$(buffer: ByteBuffer, audioFormat: Int, channelCount: Int, sampleRate: Int, bytesRead: Int, captureTimeNs: Long)</ID>

livekit-android-sdk/src/main/java/io/livekit/android/LiveKit.kt

@@ -23,6 +23,7 @@ import io.livekit.android.dagger.DaggerLiveKitComponent
 import io.livekit.android.dagger.RTCModule
 import io.livekit.android.dagger.create
 import io.livekit.android.room.Room
+import io.livekit.android.telemetry.Telemetry
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.LoggingLevel
 
@@ -64,6 +65,17 @@ object LiveKit {
     @JvmStatic
     var enableWebRTCLogging: Boolean = false
 
+    /**
+     * Opts this process out of client telemetry, in effect when this returns. Collection stops,
+     * and everything not yet sent — queued, open or cached on disk — is deleted; Rooms created
+     * afterwards collect nothing, and the first of them deletes what a previous launch left cached.
+     * Call it at every launch, before creating a Room, to collect nothing at all.
+     *
+     * TODO: final shape pending the token/consent discussion.
+     */
+    @JvmStatic
+    fun disableTelemetry() = Telemetry.disable()
+
     /**
      * Certain WebRTC classes need to be initialized prior to use.
      *

livekit-android-sdk/src/main/java/io/livekit/android/dagger/RTCModule.kt

@@ -38,6 +38,8 @@ import io.livekit.android.e2ee.DataPacketCryptorManagerImpl
 import io.livekit.android.memory.CloseableManager
 import io.livekit.android.room.datatrack.LocalDataTrackManagerFactory
 import io.livekit.android.room.datatrack.RemoteDataTrackManagerFactory
+import io.livekit.android.telemetry.Telemetry
+import io.livekit.android.telemetry.telemetryMicrophoneFailed
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.LoggingLevel
 import io.livekit.android.webrtc.CustomAudioProcessingFactory
@@ -113,6 +115,9 @@ internal object RTCModule {
                             .setNativeLibraryName("lkjingle_peerconnection_so")
                             .setInjectableLogger(
                                 { s, severity, s2 ->
+                                    if (severity == Logging.Severity.LS_ERROR) {
+                                        Telemetry.logWebRtc(s2, s)
+                                    }
                                     if (!LiveKit.enableWebRTCLogging) {
                                         return@setInjectableLogger
                                     }
@@ -125,7 +130,8 @@ internal object RTCModule {
                                         else -> LoggingLevel.OFF
                                     }
 
-                                    LKLog.log(loggingLevel, null) { "$s2: $s" }
+                                    // The console only: telemetry already has WebRTC's errors, above.
+                                    if (loggingLevel >= LKLog.loggingLevel) LKLog.logger?.log(loggingLevel, null, "$s2: $s")
                                 },
                                 Logging.Severity.LS_VERBOSE,
                             )
@@ -182,17 +188,20 @@ internal object RTCModule {
         val audioRecordErrorCallback = object : JavaAudioDeviceModule.AudioRecordErrorCallback {
             override fun onWebRtcAudioRecordInitError(errorMessage: String?) {
                 LKLog.e { "onWebRtcAudioRecordInitError: $errorMessage" }
+                telemetryMicrophoneFailed()
             }
 
             override fun onWebRtcAudioRecordStartError(
                 errorCode: JavaAudioDeviceModule.AudioRecordStartErrorCode?,
                 errorMessage: String?,
             ) {
                 LKLog.e { "onWebRtcAudioRecordStartError: $errorCode. $errorMessage" }
+                telemetryMicrophoneFailed()
             }
 
             override fun onWebRtcAudioRecordError(errorMessage: String?) {
                 LKLog.e { "onWebRtcAudioRecordError: $errorMessage" }
+                telemetryMicrophoneFailed()
             }
         }