Telemetry

gradle/libs.versions.toml

@@ -32,7 +32,9 @@ noise = "2.0.0"
 lifecycleProcess = "2.8.7"
 agp = "8.7.2"
 kotlin = "1.9.25"
-livekit-uniffi = "0.1.12"
+# prototype: local build — `cargo make android-package-local` in rust-sdks/livekit-uniffi publishes 0.0.1 to Maven Local
+livekit-uniffi = "0.0.1"
+jna = "5.16.0"
 
 [libraries]
 livekit-uniffi = { module = "io.livekit:livekit-uniffi-android", version.ref = "livekit-uniffi" }
@@ -103,6 +105,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/src/main/AndroidManifest.xml

@@ -14,7 +14,11 @@
  limitations under the License.
 -->
 
-<manifest xmlns:android="http://schemas.android.com/apk/res/android">
+<manifest xmlns:android="http://schemas.android.com/apk/res/android"
+    xmlns:tools="http://schemas.android.com/tools">
+
+    <!-- prototype: the local livekit-uniffi-android build declares minSdk 24; the published artifact is minSdk 21. -->
+    <uses-sdk tools:overrideLibrary="io.livekit.uniffi" />
 
     <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
     <uses-permission android:name="android.permission.INTERNET" />

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

@@ -23,8 +23,13 @@ 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.telemetry.TelemetryOptions
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.LoggingLevel
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.launch
 
 /**
  * The main entry point into using LiveKit.
@@ -64,6 +69,22 @@ object LiveKit {
     @JvmStatic
     var enableWebRTCLogging: Boolean = false
 
+    /**
+     * Turn client telemetry on: warn/error records, RTC statistics, operation spans and device
+     * state, shipped out-of-band to an OTLP collector. Process-wide, like [loggingLevel]: call it
+     * before creating Rooms — each Room gets its own scope (see [Room.telemetryTraceId]).
+     * `null` turns telemetry off (the default) after a bounded final flush.
+     */
+    @OptIn(DelicateCoroutinesApi::class)
+    @JvmStatic
+    fun setTelemetry(appContext: Context, options: TelemetryOptions?) {
+        if (options != null) {
+            Telemetry.configure(appContext, options)
+        } else {
+            GlobalScope.launch { Telemetry.shutdown() }
+        }
+    }
+
     /**
      * 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,7 @@ 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.util.LKLog
 import io.livekit.android.util.LoggingLevel
 import io.livekit.android.webrtc.CustomAudioProcessingFactory
@@ -62,6 +63,9 @@ import livekit.org.webrtc.VideoDecoderFactory
 import livekit.org.webrtc.VideoEncoderFactory
 import livekit.org.webrtc.audio.AudioDeviceModule
 import livekit.org.webrtc.audio.JavaAudioDeviceModule
+import uniffi.livekit_telemetry.CaptureDevice
+import uniffi.livekit_telemetry.CaptureFailure
+import uniffi.livekit_telemetry.DeviceEvent
 import javax.inject.Named
 import javax.inject.Singleton
 
@@ -113,6 +117,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
                                     }
@@ -182,17 +189,20 @@ internal object RTCModule {
         val audioRecordErrorCallback = object : JavaAudioDeviceModule.AudioRecordErrorCallback {
             override fun onWebRtcAudioRecordInitError(errorMessage: String?) {
                 LKLog.e { "onWebRtcAudioRecordInitError: $errorMessage" }
+                Telemetry.deviceEvent(DeviceEvent.CaptureFailed(CaptureDevice.MICROPHONE, CaptureFailure.OTHER))
             }
 
             override fun onWebRtcAudioRecordStartError(
                 errorCode: JavaAudioDeviceModule.AudioRecordStartErrorCode?,
                 errorMessage: String?,
             ) {
                 LKLog.e { "onWebRtcAudioRecordStartError: $errorCode. $errorMessage" }
+                Telemetry.deviceEvent(DeviceEvent.CaptureFailed(CaptureDevice.MICROPHONE, CaptureFailure.OTHER))
             }
 
             override fun onWebRtcAudioRecordError(errorMessage: String?) {
                 LKLog.e { "onWebRtcAudioRecordError: $errorMessage" }
+                Telemetry.deviceEvent(DeviceEvent.CaptureFailed(CaptureDevice.MICROPHONE, CaptureFailure.OTHER))
             }
         }
 

livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt

@@ -43,6 +43,10 @@ import io.livekit.android.room.util.MediaConstraintKeys
 import io.livekit.android.room.util.createAnswer
 import io.livekit.android.room.util.setLocalDescription
 import io.livekit.android.room.util.waitUntilConnected
+import io.livekit.android.telemetry.Telemetry
+import io.livekit.android.telemetry.TelemetryOptions
+import io.livekit.android.telemetry.begin
+import io.livekit.android.telemetry.end
 import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.Either
 import io.livekit.android.util.FlowObservable
@@ -65,9 +69,12 @@ import io.livekit.android.webrtc.peerconnection.RTCThreadToken
 import io.livekit.android.webrtc.peerconnection.executeBlockingOnRTCThread
 import io.livekit.android.webrtc.peerconnection.launchBlockingOnRTCThread
 import io.livekit.android.webrtc.toProtoSessionDescription
+import io.livekit.uniffi.TelemetryScope
+import io.livekit.uniffi.TelemetrySpan
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.asContextElement
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.ensureActive
@@ -100,6 +107,9 @@ import livekit.org.webrtc.RtpSender
 import livekit.org.webrtc.RtpTransceiver
 import livekit.org.webrtc.RtpTransceiver.RtpTransceiverInit
 import livekit.org.webrtc.SessionDescription
+import uniffi.livekit_telemetry.ReconnectReason
+import uniffi.livekit_telemetry.SpanName
+import uniffi.livekit_telemetry.SpanStep
 import java.nio.ByteBuffer
 import javax.inject.Inject
 import javax.inject.Named
@@ -128,6 +138,34 @@ internal constructor(
 ) : SignalClient.Listener {
     internal var listener: Listener? = null
 
+    /**
+     * The Room's telemetry scope; null when telemetry is off. Bound on the engine's and the signal
+     * client's coroutines, so the Room handlers they drive log under the Room's session.
+     */
+    internal var telemetryScope: TelemetryScope? = null
+        set(value) {
+            field = value
+            client.telemetryScope = value
+        }
+
+    /** The scope for the `lk.reconnect` span, when the `room` instrument is on. */
+    private val traceScope: TelemetryScope?
+        get() = telemetryScope?.takeIf { Telemetry.enabled(TelemetryOptions.Instrument.ROOM) }
+
+    /**
+     * The Room's open `lk.connect` span while the user-initiated connect runs; the checkpoints
+     * are stamped here and in [SignalClient].
+     */
+    internal var connectSpan: TelemetrySpan? = null
+        set(value) {
+            field = value
+            client.connectSpan = value
+        }
+
+    /** Whether the last disconnect was the reconnect policy running out of attempts. */
+    @Volatile
+    internal var reconnectFailed = false
+
     /**
      * Reflects the combined connection state of SignalClient and primary PeerConnection.
      */
@@ -154,7 +192,7 @@ internal constructor(
             ConnectionState.DISCONNECTED -> {
                 LKLog.d { "primary ICE disconnected" }
                 if (oldVal == ConnectionState.CONNECTED) {
-                    reconnect()
+                    reconnect(if (isSubscriberPrimary) ReconnectReason.SUBSCRIBER_FAILED else ReconnectReason.PUBLISHER_FAILED)
                 }
             }
 
@@ -258,9 +296,10 @@ internal constructor(
         roomOptions: RoomOptions,
     ): JoinResponse {
         coroutineScope.close()
-        coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher)
+        coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher + Telemetry.currentScope.asContextElement(telemetryScope))
         sessionUrl = url
         sessionToken = token
+        reconnectFailed = false
         connectOptions = options
         lastRoomOptions = roomOptions
         return joinImpl(url, token, options, roomOptions)
@@ -276,6 +315,8 @@ internal constructor(
             connectionState = ConnectionState.CONNECTING
         }
         val joinResponse = client.join(url, token, options, roomOptions)
+        connectSpan?.step(SpanStep.Signal)
+        connectSpan?.step(SpanStep.JoinRecv)
         ensureActive()
 
         if (joinResponse.hasParticipant()) {
@@ -298,6 +339,7 @@ internal constructor(
         isSubscriberPrimary = joinResponse.subscriberPrimary
 
         configure(joinResponse, options)
+        connectSpan?.step(SpanStep.PcCreated)
 
         // Subscriber-primary defers the publisher PC until something is published. After a full
         // reconnect `hasPublished` is still set, so re-negotiate here — otherwise the ICE wait
@@ -368,7 +410,7 @@ internal constructor(
                     // Also reconnect on publisher disconnect
                     publisherObserver.connectionChangeListener = { newState ->
                         if (newState.isDisconnected()) {
-                            reconnect()
+                            reconnect(ReconnectReason.PUBLISHER_FAILED)
                         }
                     }
                 } else {
@@ -575,9 +617,12 @@ internal constructor(
     /**
      * reconnect Signal and PeerConnections
      */
-    @Synchronized
     @VisibleForTesting(otherwise = VisibleForTesting.PACKAGE_PRIVATE)
-    fun reconnect() {
+    fun reconnect() = reconnect(ReconnectReason.UNKNOWN)
+
+    /** One reconnect cycle = one `lk.reconnect` span; attempts are its checkpoints. */
+    @Synchronized
+    internal fun reconnect(reason: ReconnectReason) {
         if (reconnectingJob?.isActive == true) {
             LKLog.d { "Reconnection is already in progress" }
             return
@@ -595,7 +640,8 @@ internal constructor(
         val forceFullReconnect = fullReconnectOnNext
         fullReconnectOnNext = false
         endSignalSession()
-        val job = coroutineScope.launch {
+        val reconnectSpan = traceScope.begin(SpanName.Reconnect(reason))
+        val job = coroutineScope.launch(Telemetry.currentSpan.asContextElement(reconnectSpan)) {
             var hasResumedOnce = false
             var hasReconnectedOnce = false
 
@@ -642,6 +688,7 @@ internal constructor(
                     ReconnectType.FORCE_SOFT_RECONNECT -> false
                     ReconnectType.FORCE_FULL_RECONNECT -> true
                 }
+                reconnectSpan?.step(SpanStep.Attempt((retries + 1).toUInt(), isFullReconnect))
 
                 var lastMessageSeq: Int? = null
                 val connectOptions = connectOptions ?: ConnectOptions()
@@ -754,6 +801,7 @@ internal constructor(
                         outgoingDataTrackManager.republishTracks()
                     }
                     incomingDataTrackManager.resendSubscriptionUpdates()
+                    reconnectSpan?.end()
                     listener?.onPostReconnect(isFullReconnect)
                     return@launch
                 }
@@ -765,12 +813,19 @@ internal constructor(
                 }
             }
 
+            if (isClosed) {
+                reconnectSpan?.cancel() // disconnect() or a newer reconnect won
+            } else {
+                reconnectFailed = true
+                reconnectSpan?.fail("ReconnectFailed")
+            }
             close("Failed reconnecting")
             listener?.onEngineDisconnected(DisconnectReason.UNKNOWN_REASON)
         }
 
         reconnectingJob = job
         job.invokeOnCompletion {
+            reconnectSpan?.takeIf { !it.isEnded() }?.cancel()
             if (reconnectingJob == job) {
                 reconnectingJob = null
             }
@@ -1340,7 +1395,7 @@ internal constructor(
         LKLog.i { "received close event: $reason, code: $code" }
         endSignalSession()
         abortPendingPublishTracks()
-        reconnect()
+        reconnect(ReconnectReason.SIGNAL_DISCONNECTED)
     }
 
     override fun onRemoteMuteChanged(trackSid: String, muted: Boolean) {