diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index e0e6c0fd..e8628348 100644 --- a/gradle/libs.versions.toml +++ b/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" } diff --git a/livekit-android-sdk/src/main/AndroidManifest.xml b/livekit-android-sdk/src/main/AndroidManifest.xml index 52bbc666..4af969c6 100644 --- a/livekit-android-sdk/src/main/AndroidManifest.xml +++ b/livekit-android-sdk/src/main/AndroidManifest.xml @@ -14,7 +14,11 @@ limitations under the License. --> - + + + + diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/LiveKit.kt b/livekit-android-sdk/src/main/java/io/livekit/android/LiveKit.kt index c281b5aa..760bb00a 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/LiveKit.kt +++ b/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. * diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/dagger/RTCModule.kt b/livekit-android-sdk/src/main/java/io/livekit/android/dagger/RTCModule.kt index bf4a692d..d0b83e5a 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/dagger/RTCModule.kt +++ b/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,6 +189,7 @@ 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( @@ -189,10 +197,12 @@ internal object RTCModule { 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)) } } diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt index bc4c8235..eac8684d 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt +++ b/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) { diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt index 12e0121e..33717413 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt @@ -82,6 +82,14 @@ import io.livekit.android.room.track.Track import io.livekit.android.room.track.TrackPublication import io.livekit.android.room.types.toSDKType import io.livekit.android.room.util.ConnectionWarmer +import io.livekit.android.telemetry.DeviceTelemetry +import io.livekit.android.telemetry.RTCTelemetry +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.telemetry.lowered +import io.livekit.android.telemetry.telemetry import io.livekit.android.util.FlowObservable import io.livekit.android.util.LKLog import io.livekit.android.util.flow @@ -89,11 +97,14 @@ import io.livekit.android.util.flowDelegate import io.livekit.android.util.invoke import io.livekit.android.util.rethrowIfCancellationSignal import io.livekit.android.webrtc.getFilteredStats +import io.livekit.uniffi.TelemetryScope +import io.livekit.uniffi.TelemetrySpan import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asContextElement import kotlinx.coroutines.cancel import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.ensureActive @@ -115,6 +126,10 @@ import livekit.org.webrtc.RendererCommon import livekit.org.webrtc.RtpReceiver import livekit.org.webrtc.SurfaceViewRenderer import livekit.org.webrtc.audio.AudioDeviceModule +import uniffi.livekit_telemetry.ReconnectReason +import uniffi.livekit_telemetry.RoomIdentity +import uniffi.livekit_telemetry.SpanName +import uniffi.livekit_telemetry.SpanStep import java.net.URI import java.util.Date import javax.inject.Named @@ -168,8 +183,51 @@ constructor( private val eventBus = BroadcastEventBus() val events = eventBus.readOnly() + /** + * This Room's scope on the process telemetry pipeline — one trace for the Room's lifetime — + * or null when telemetry is off. Taken at creation, so pre-connect work is part of the call. + */ + internal val telemetryScope: TelemetryScope? = Telemetry.scope() + + /** The scope for this Room's spans, when the `room` instrument is on. */ + internal val traceScope: TelemetryScope? + get() = if (Telemetry.enabled(TelemetryOptions.Instrument.ROOM)) telemetryScope else null + + private val rtcTelemetry: RTCTelemetry? = + telemetryScope?.takeIf { Telemetry.enabled(TelemetryOptions.Instrument.RTC) }?.let { RTCTelemetry(this, it) } + + /** The user-initiated connect, open from [connect] to [onEngineConnected]; the engine stamps its checkpoints. */ + private var connectSpan: TelemetrySpan? = null + set(value) { + field = value + engine.connectSpan = value + } + + /** + * The telemetry trace id of this Room's scope (32 hex characters), or null when telemetry is + * off. Show it to users or attach it to support tickets: it opens the full client-side + * timeline of the call, including connect attempts that never reached a server. + */ + val telemetryTraceId: String? + get() = telemetryScope?.traceId() + + /** + * Record an app-defined telemetry event alongside the SDK's own, in this Room's trace. The + * name is namespaced under `custom.` (`"checkout.started"` ships as `custom.checkout.started`); + * attributes keep their names and types (strings, numbers, booleans). A no-op when telemetry + * is off. + */ + fun emitTelemetryEvent(name: String, attributes: Map = emptyMap()) { + telemetryScope?.emitCustom(name, attributes.lowered()) + } + + /** An app-defined span in this Room's trace; a no-op when telemetry is off. */ + internal fun beginSpan(label: String): TelemetrySpan? = traceScope.begin(SpanName.Custom(label)) + init { engine.listener = this + engine.telemetryScope = telemetryScope + audioSwitchHandler?.let { DeviceTelemetry.observe(it) } // Register SDK-internal text-stream handlers for the RPC v2 transport. These reserve // the topics `lk.rpc_request` and `lk.rpc_response` from user-level handler registration. @@ -365,6 +423,7 @@ constructor( */ val localParticipant: LocalParticipant = localParticipantFactory.create(dynacast = false).apply { internalListener = this@Room + telemetryScope = this@Room.traceScope } private var mutableRemoteParticipants by flowDelegate(emptyMap()) @@ -488,7 +547,10 @@ constructor( state = State.CONNECTING connectOptions = options - coroutineScope = CoroutineScope(defaultDispatcher + SupervisorJob()) + Telemetry.setServer(url, token) + connectSpan = traceScope.begin(SpanName.Connect) + + coroutineScope = CoroutineScope(defaultDispatcher + SupervisorJob() + Telemetry.currentScope.asContextElement(telemetryScope)) roomOptions = getCurrentRoomOptions() @@ -513,7 +575,7 @@ constructor( // rethrow all throwables from the connect job. val emptyCoroutineExceptionHandler = CoroutineExceptionHandler { _, _ -> } val connectJob = coroutineScope.launch( - ioDispatcher + emptyCoroutineExceptionHandler, + ioDispatcher + emptyCoroutineExceptionHandler + Telemetry.currentSpan.asContextElement(connectSpan), ) { if (audioProcessingController is AuthedAudioProcessingController) { audioProcessingController.authenticate(url, token) @@ -592,6 +654,7 @@ constructor( collectMetrics(room = this@Room, rtcEngine = engine) } } + rtcTelemetry?.let { rtc -> coroutineScope.launch { rtc.run() } } } val outerHandler = coroutineContext.job.invokeOnCompletion { cause -> @@ -609,6 +672,7 @@ constructor( connectJob.join() error?.let { + connectSpan?.end(it) if (it !is CancellationException) { handleDisconnect(DisconnectReason.JOIN_FAILURE) } @@ -702,6 +766,7 @@ constructor( localParticipant.updateFromInfo(response.participant) localParticipant.setEnabledPublishCodecs(response.enabledPublishCodecsList) + updateTelemetryRoom() if (response.otherParticipantsList.isNotEmpty()) { response.otherParticipantsList.forEach { info -> @@ -710,6 +775,18 @@ constructor( } } + /** The room and local participant on every telemetry record of this session from now on. */ + private fun updateTelemetryRoom() { + telemetryScope?.setRoom( + RoomIdentity( + sid = sid?.sid?.takeIf { it.isNotEmpty() }, + name = name, + participantSid = localParticipant.sid.value.takeIf { it.isNotEmpty() }, + participantIdentity = localParticipant.identity?.value, + ), + ) + } + private fun setupLocalParticipantEventHandling() { coroutineScope.launch { localParticipant.events.collect { @@ -1053,7 +1130,7 @@ constructor( if (state == State.RECONNECTING) { return } - engine.reconnect() + engine.reconnect(ReconnectReason.NETWORK_CHANGED) } private fun handleDisconnect(reason: DisconnectReason) { @@ -1069,6 +1146,10 @@ constructor( hasLostConnectivity = false state = State.DISCONNECTED + connectSpan?.fail(reason.name) + connectSpan = null + // Once per real session: never on a reconnect. + telemetryScope?.disconnected(reason.telemetry(engine.reconnectFailed)) cleanupRoom() engine.close() @@ -1234,6 +1315,13 @@ constructor( */ override fun onEngineConnected() { state = State.CONNECTED + connectSpan?.run { + step(SpanStep.Engine) + step(SpanStep.PcConnected) + step(SpanStep.RoomConnected) + end() + } + connectSpan = null eventBus.postEvent(RoomEvent.Connected(this), coroutineScope) } @@ -1345,6 +1433,7 @@ constructor( override fun onRoomUpdate(update: LivekitModels.Room) { if (update.sid != null) { sid = Sid(update.sid) + updateTelemetryRoom() } val oldMetadata = metadata metadata = update.metadata diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt index 788ca4da..c653eb4a 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/SignalClient.kt @@ -27,6 +27,7 @@ import io.livekit.android.room.participant.ParticipantTrackPermission import io.livekit.android.room.track.Track import io.livekit.android.stats.NetworkInfo import io.livekit.android.stats.getClientInfo +import io.livekit.android.telemetry.Telemetry import io.livekit.android.util.CloseableCoroutineScope import io.livekit.android.util.Either import io.livekit.android.util.LKLog @@ -36,12 +37,15 @@ import io.livekit.android.util.toHttpUrl import io.livekit.android.util.toWebsocketUrl import io.livekit.android.util.withDeadline import io.livekit.android.webrtc.toProtoSessionDescription +import io.livekit.uniffi.TelemetryScope +import io.livekit.uniffi.TelemetrySpan import kotlinx.coroutines.CancellableContinuation import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asContextElement import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.launch @@ -64,6 +68,7 @@ import okhttp3.WebSocket import okhttp3.WebSocketListener import okio.ByteString import okio.ByteString.Companion.toByteString +import uniffi.livekit_telemetry.SpanStep import java.util.Date import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.atomic.AtomicInteger @@ -108,6 +113,12 @@ constructor( @Volatile private var joinContinuation: CancellableContinuation? = null + + /** The Room's open `lk.connect` span, for the signaling checkpoints; null outside the user-initiated connect. */ + internal var connectSpan: TelemetrySpan? = null + + /** The Room's telemetry scope: signal responses drive the Room's handlers, whose records are filed under it. */ + internal var telemetryScope: TelemetryScope? = null private lateinit var coroutineScope: CloseableCoroutineScope /** @@ -197,7 +208,7 @@ constructor( LKLog.i { "connecting to $wsUrlString" } - coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher) + coroutineScope = CloseableCoroutineScope(SupervisorJob() + ioDispatcher + Telemetry.currentScope.asContextElement(telemetryScope)) lastUrl = wsUrlString lastOptions = options lastRoomOptions = roomOptions @@ -312,6 +323,10 @@ constructor( } // --------------------------------- WebSocket Listener --------------------------------------// + override fun onOpen(webSocket: WebSocket, response: Response) { + connectSpan?.step(SpanStep.WsOpen) + } + override fun onMessage(webSocket: WebSocket, text: String) { if (webSocket != currentWs) { // Possibly message from old websocket, discard. @@ -439,6 +454,7 @@ constructor( } fun sendOffer(offer: SessionDescription, offerId: Int) { + connectSpan?.step(SpanStep.OfferSent) val sd = offer.toProtoSessionDescription(offerId) val request = LivekitRtc.SignalRequest.newBuilder() .setOffer(sd) @@ -448,6 +464,7 @@ constructor( } fun sendAnswer(answer: SessionDescription, offerId: Int) { + connectSpan?.step(SpanStep.AnswerSent) val sd = answer.toProtoSessionDescription(offerId) val request = LivekitRtc.SignalRequest.newBuilder() .setAnswer(sd) diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt index 816ed9b6..7090511d 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/participant/LocalParticipant.kt @@ -65,16 +65,22 @@ import io.livekit.android.room.track.VideoPreset import io.livekit.android.room.track.screencapture.ScreenCaptureParams import io.livekit.android.room.util.EncodingUtils import io.livekit.android.rpc.RpcError +import io.livekit.android.telemetry.begin +import io.livekit.android.telemetry.end +import io.livekit.android.telemetry.setTrack import io.livekit.android.util.LKLog import io.livekit.android.util.flow import io.livekit.android.util.rethrowIfCancellationSignal import io.livekit.android.webrtc.sortVideoCodecPreferences +import io.livekit.uniffi.TelemetryScope import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.Job import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -95,6 +101,7 @@ import livekit.org.webrtc.RtpTransceiver.RtpTransceiverInit import livekit.org.webrtc.SurfaceTextureHelper import livekit.org.webrtc.VideoCapturer import livekit.org.webrtc.VideoProcessor +import uniffi.livekit_telemetry.SpanName import java.nio.ByteBuffer import java.nio.charset.CodingErrorAction import java.util.Collections @@ -152,6 +159,9 @@ internal constructor( internal val enabledPublishVideoCodecs = Collections.synchronizedList(mutableListOf()) + /** The Room's trace scope, for the `lk.publish` span; null when telemetry is off. */ + internal var telemetryScope: TelemetryScope? = null + private var defaultAudioTrack: LocalAudioTrack? = null private var defaultVideoTrack: LocalVideoTrack? = null @@ -668,7 +678,11 @@ internal constructor( return null } + // One publish = one `lk.publish` span, under the connect span for a pre-connect microphone. + val span = telemetryScope.begin(SpanName.Publish) + fun onPublishFailure(e: TrackException.PublishException, triggerEvent: Boolean = true) { + span?.end(e) publishListener?.onPublishFailure(e) if (triggerEvent) { eventBus.postEvent(ParticipantEvent.LocalTrackPublicationFailed(this, track, e), scope) @@ -680,6 +694,7 @@ internal constructor( } val trackSource = Track.Source.fromProto(addTrackRequestBuilder.source ?: LivekitModels.TrackSource.UNRECOGNIZED) + span?.setTrack(track.kind, trackSource) if (!hasPermissionsToPublish(trackSource)) { val exception = TrackException.PublishException("Failed to publish track, insufficient permissions") onPublishFailure(exception) @@ -848,6 +863,8 @@ internal constructor( participant = this, options = options, ) + span?.setTrack(track.kind, trackSource, publication.sid) + span?.end() addTrackPublication(publication) LKLog.v { "add track publication $publication" } @@ -856,6 +873,7 @@ internal constructor( eventBus.postEvent(ParticipantEvent.LocalTrackPublished(this, publication), scope) } } finally { + span?.takeIf { !it.isEnded() }?.run { if (currentCoroutineContext().isActive) fail("PublishException") else cancel() } if (publication == null) { // Negotiation can win the race against a failed or cancelled add track request. // Without a publication there is no unpublish to stop the transceiver, so it diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/video/CameraCapturerUtils.kt b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/video/CameraCapturerUtils.kt index 2996e524..ffbb8e79 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/room/track/video/CameraCapturerUtils.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/room/track/video/CameraCapturerUtils.kt @@ -22,6 +22,7 @@ import android.content.Context import android.hardware.camera2.CameraManager import io.livekit.android.room.track.CameraPosition import io.livekit.android.room.track.LocalVideoTrackOptions +import io.livekit.android.telemetry.TelemetryCameraEvents import io.livekit.android.util.LKLog import livekit.org.webrtc.Camera1Capturer import livekit.org.webrtc.Camera1Enumerator @@ -98,6 +99,7 @@ object CameraCapturerUtils { ): Pair? { val cameraEnumerator = provider.provideEnumerator(context) val cameraEventsDispatchHandler = CameraEventsDispatchHandler() + cameraEventsDispatchHandler.registerHandler(TelemetryCameraEvents) val targetDevice = cameraEnumerator.findCamera(options.deviceId, options.position) ?: return null val targetVideoCapturer = provider.provideCapturer(context, options, cameraEventsDispatchHandler) diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/DeviceTelemetry.kt b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/DeviceTelemetry.kt new file mode 100644 index 00000000..3b9ebb47 --- /dev/null +++ b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/DeviceTelemetry.kt @@ -0,0 +1,225 @@ +/* + * Copyright 2026 LiveKit, 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 io.livekit.android.telemetry + +import android.content.BroadcastReceiver +import android.content.ComponentCallbacks2 +import android.content.Context +import android.content.Intent +import android.content.IntentFilter +import android.content.res.Configuration +import android.net.ConnectivityManager +import android.net.Network +import android.net.NetworkCapabilities +import android.net.NetworkRequest +import android.os.BatteryManager +import android.os.Build +import android.os.PowerManager +import androidx.core.content.ContextCompat +import com.twilio.audioswitch.AudioDevice +import io.livekit.android.audio.AudioSwitchHandler +import io.livekit.uniffi.telemetrySetDeviceState +import livekit.org.webrtc.CameraVideoCapturer +import uniffi.livekit_telemetry.AppState +import uniffi.livekit_telemetry.AudioOutput +import uniffi.livekit_telemetry.AudioRouteReason +import uniffi.livekit_telemetry.CaptureDevice +import uniffi.livekit_telemetry.CaptureFailure +import uniffi.livekit_telemetry.DeviceEvent +import uniffi.livekit_telemetry.DeviceState +import uniffi.livekit_telemetry.MemoryPressure +import uniffi.livekit_telemetry.NetworkType +import uniffi.livekit_telemetry.TelemetryInstrument +import uniffi.livekit_telemetry.ThermalState + +/** + * The Device-area instrument: thermal status, battery saver, memory pressure, network (type, + * metered, Data Saver) and battery, observed process-wide (a device has no room) and pushed to + * the pipeline as [DeviceState] — which stretches the cadence and holds uploads. Callback-driven + * throughout: nothing polls. App state stays `foreground`: the SDK has no lifecycle dependency. + */ +internal class DeviceTelemetry(context: Context) : TelemetryInstrument { + private val app = context.applicationContext + private val powerManager = app.getSystemService(Context.POWER_SERVICE) as? PowerManager + private val connectivityManager = app.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager + + @Volatile + private var memory = MemoryPressure.NORMAL + + @Volatile + private var network: NetworkCapabilities? = null + + @Volatile + private var battery: Intent? = null + private var thermalListener: PowerManager.OnThermalStatusChangedListener? = null + + private val receiver = object : BroadcastReceiver() { + override fun onReceive(context: Context, intent: Intent) { + if (intent.action == Intent.ACTION_BATTERY_CHANGED) battery = intent + push() + } + } + + private val networkCallback = object : ConnectivityManager.NetworkCallback() { + override fun onCapabilitiesChanged(network: Network, capabilities: NetworkCapabilities) { + this@DeviceTelemetry.network = capabilities + push() + } + + override fun onLost(network: Network) { + this@DeviceTelemetry.network = null + push() + } + } + + private val memoryCallbacks = object : ComponentCallbacks2 { + override fun onTrimMemory(level: Int) { + memory = when (level) { + ComponentCallbacks2.TRIM_MEMORY_RUNNING_CRITICAL, ComponentCallbacks2.TRIM_MEMORY_COMPLETE -> MemoryPressure.CRITICAL + ComponentCallbacks2.TRIM_MEMORY_UI_HIDDEN -> return // the UI went away, not memory pressure + else -> MemoryPressure.WARNING + } + push() + } + + override fun onLowMemory() { + memory = MemoryPressure.CRITICAL + push() + } + + override fun onConfigurationChanged(newConfig: Configuration) {} + } + + override fun start() { + network = connectivityManager?.let { cm -> + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) cm.activeNetwork?.let(cm::getNetworkCapabilities) else null + } + val filter = IntentFilter().apply { + addAction(Intent.ACTION_BATTERY_CHANGED) + addAction(PowerManager.ACTION_POWER_SAVE_MODE_CHANGED) + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) addAction(ConnectivityManager.ACTION_RESTRICT_BACKGROUND_CHANGED) + } + // System broadcasts only; the return value is the sticky battery intent. + battery = ContextCompat.registerReceiver(app, receiver, filter, ContextCompat.RECEIVER_NOT_EXPORTED) + connectivityManager?.let { cm -> + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) { + cm.registerDefaultNetworkCallback(networkCallback) + } else { + cm.registerNetworkCallback( + NetworkRequest.Builder().addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET).build(), + networkCallback, + ) + } + } + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) { + thermalListener = PowerManager.OnThermalStatusChangedListener { push() }.also { powerManager?.addThermalStatusListener(it) } + } + app.registerComponentCallbacks(memoryCallbacks) + push() + } + + override fun stop() { + runCatching { app.unregisterReceiver(receiver) } + runCatching { connectivityManager?.unregisterNetworkCallback(networkCallback) } + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) thermalListener?.let { powerManager?.removeThermalStatusListener(it) } + app.unregisterComponentCallbacks(memoryCallbacks) + } + + private fun push() { + val battery = battery + val level = battery?.getIntExtra(BatteryManager.EXTRA_LEVEL, -1)?.takeIf { it >= 0 } + val scale = battery?.getIntExtra(BatteryManager.EXTRA_SCALE, 100)?.takeIf { it > 0 } + val status = battery?.getIntExtra(BatteryManager.EXTRA_STATUS, BatteryManager.BATTERY_STATUS_UNKNOWN) + telemetrySetDeviceState( + DeviceState( + thermal = thermal(), + lowPowerMode = powerManager?.isPowerSaveMode == true, + appState = AppState.FOREGROUND, + memory = memory, + network = network.type, + networkExpensive = network?.hasCapability(NetworkCapabilities.NET_CAPABILITY_NOT_METERED) == false, + networkConstrained = Build.VERSION.SDK_INT >= Build.VERSION_CODES.N && + connectivityManager?.restrictBackgroundStatus == ConnectivityManager.RESTRICT_BACKGROUND_STATUS_ENABLED, + batteryLevel = if (level != null && scale != null) (level * 100 / scale).toUInt() else null, + batteryCharging = status == BatteryManager.BATTERY_STATUS_CHARGING || status == BatteryManager.BATTERY_STATUS_FULL, + ), + ) + } + + private fun thermal(): ThermalState { + if (Build.VERSION.SDK_INT < Build.VERSION_CODES.Q) return ThermalState.NOMINAL + return when (powerManager?.currentThermalStatus) { + PowerManager.THERMAL_STATUS_MODERATE -> ThermalState.FAIR + PowerManager.THERMAL_STATUS_SEVERE -> ThermalState.SERIOUS + PowerManager.THERMAL_STATUS_CRITICAL, PowerManager.THERMAL_STATUS_EMERGENCY, PowerManager.THERMAL_STATUS_SHUTDOWN -> ThermalState.CRITICAL + else -> ThermalState.NOMINAL // NONE, LIGHT + } + } + + private val NetworkCapabilities?.type: NetworkType + get() = when { + this == null -> NetworkType.UNAVAILABLE + hasTransport(NetworkCapabilities.TRANSPORT_VPN) -> NetworkType.VPN + hasTransport(NetworkCapabilities.TRANSPORT_WIFI) -> NetworkType.WIFI + hasTransport(NetworkCapabilities.TRANSPORT_CELLULAR) -> NetworkType.CELL + hasTransport(NetworkCapabilities.TRANSPORT_ETHERNET) -> NetworkType.WIRED + hasTransport(NetworkCapabilities.TRANSPORT_BLUETOOTH) -> NetworkType.BLUETOOTH + else -> NetworkType.OTHER + } + + companion object { + /** + * Route changes and focus loss are events, not state: they explain audio glitches. + * AudioSwitch names no reason for a change; a focus loss is an interruption that ends on + * the matching gain. + */ + fun observe(audio: AudioSwitchHandler) { + audio.registerAudioDeviceChangeListener { _, selected -> + Telemetry.deviceEvent(DeviceEvent.AudioRouteChanged(listOfNotNull(selected?.output), AudioRouteReason.UNKNOWN)) + } + audio.registerOnAudioFocusChangeListener { change -> + Telemetry.deviceEvent(DeviceEvent.AudioInterruption(began = change < 0)) + } + } + + private val AudioDevice.output: AudioOutput + get() = when (this) { + is AudioDevice.BluetoothHeadset -> AudioOutput.BLUETOOTH + is AudioDevice.WiredHeadset -> AudioOutput.WIRED_HEADSET + is AudioDevice.Earpiece -> AudioOutput.RECEIVER + is AudioDevice.Speakerphone -> AudioOutput.SPEAKER + else -> AudioOutput.OTHER + } + } +} + +/** Camera failures from WebRTC's capturer, on every camera track the SDK opens. */ +internal object TelemetryCameraEvents : CameraVideoCapturer.CameraEventsHandler { + override fun onCameraError(message: String?) = + Telemetry.deviceEvent(DeviceEvent.CaptureFailed(CaptureDevice.CAMERA, CaptureFailure.OTHER)) + + override fun onCameraDisconnected() = + Telemetry.deviceEvent(DeviceEvent.CaptureFailed(CaptureDevice.CAMERA, CaptureFailure.DISCONNECTED)) + + override fun onCameraFreezed(message: String?) {} + + override fun onCameraOpening(cameraName: String?) {} + + override fun onFirstFrameAvailable() {} + + override fun onCameraClosed() {} +} diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/RTCTelemetry.kt b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/RTCTelemetry.kt new file mode 100644 index 00000000..69ff320e --- /dev/null +++ b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/RTCTelemetry.kt @@ -0,0 +1,131 @@ +/* + * Copyright 2026 LiveKit, 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 io.livekit.android.telemetry + +import io.livekit.android.events.RoomEvent +import io.livekit.android.room.Room +import io.livekit.android.room.participant.LocalParticipant +import io.livekit.android.room.participant.RemoteParticipant +import io.livekit.android.room.track.RemoteTrackPublication +import io.livekit.android.room.track.Track +import io.livekit.android.room.track.TrackPublication +import io.livekit.uniffi.TelemetryScope +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import livekit.org.webrtc.RTCStatsReport +import uniffi.livekit_telemetry.AttributeValue +import uniffi.livekit_telemetry.RtcStat +import uniffi.livekit_telemetry.StreamDirection +import java.math.BigInteger +import java.util.concurrent.ConcurrentHashMap + +/** + * The RTC-area instrument of one Room. Once a second it hands the raw `getStats()` report of + * every track the Room publishes or subscribes to the Room's scope — the core maps and windows + * it into `lk.rtc.stats.sample` — and it reports the remote tracks' lifecycle so the core can run + * the `lk.subscribe` span (intent → first media, ended by the first inbound reading with bytes). + * No RTC state lives here. + */ +internal class RTCTelemetry(private val room: Room, private val scope: TelemetryScope) { + private class Observed(val track: Track, val direction: StreamDirection) + + /** Published and subscribed tracks by sid. */ + private val observed = ConcurrentHashMap() + + /** Runs for the Room's connected lifetime; cancelled with the Room's scope. */ + suspend fun run() = coroutineScope { + launch { room.events.events.collect { onEvent(it) } } + try { + while (isActive) { + delay(STATS_INTERVAL_MS) + for ((sid, entry) in observed) { + val kind = entry.track.kind.telemetry ?: continue + val report = entry.track.getRTCStats() ?: continue + scope.recordStatsReport(sid, kind, entry.direction, report.telemetryStats, report.telemetryTimestampNs) + } + } + } finally { + observed.clear() + } + } + + private fun onEvent(event: RoomEvent) { + when (event) { + is RoomEvent.TrackPublished -> when (val participant = event.participant) { + is LocalParticipant -> event.publication.track?.let { observed[event.publication.sid] = Observed(it, StreamDirection.OUTBOUND) } + // With autoSubscribe the intent exists the moment the track is known. + is RemoteParticipant -> if ((event.publication as? RemoteTrackPublication)?.isDesired == true) { + spanTrack(event.publication, participant)?.let(scope::subscribeStarted) + } + } + + is RoomEvent.TrackUnpublished -> { + observed.remove(event.publication.sid) + if (event.participant is RemoteParticipant) scope.subscribeCancelled(event.publication.sid) + } + + is RoomEvent.TrackSubscribed -> { + spanTrack(event.publication, event.participant)?.let(scope::subscribed) + observed[event.publication.sid] = Observed(event.track, StreamDirection.INBOUND) + } + + is RoomEvent.TrackUnsubscribed -> { + observed.remove(event.publications.sid) + scope.subscribeCancelled(event.publications.sid) + } + + is RoomEvent.TrackSubscriptionFailed -> scope.subscribeFailed(event.sid, event.exception.errorType()) + else -> {} + } + } + + private fun spanTrack(publication: TrackPublication, participant: RemoteParticipant) = + spanTrack(publication.kind, publication.source, publication.sid, participant.identity?.value) + + companion object { + /** The core windows 1 Hz readings into `statsWindow` samples. */ + private const val STATS_INTERVAL_MS = 1000L + } +} + +/** + * The report as the core takes it: every entry with its standard members, nested maps flattened + * with a dot (`qualityLimitationDurations.cpu`). No field names known here. + */ +internal val RTCStatsReport.telemetryStats: List + get() = statsMap.values.map { stat -> + RtcStat(kind = stat.type, id = stat.id, members = buildMap { flatten(stat.members, "", this) }) + } + +internal val RTCStatsReport.telemetryTimestampNs: ULong + get() = (timestampUs.coerceAtLeast(0.0) * 1000).toULong() + +private fun flatten(values: Map<*, *>, prefix: String, into: MutableMap) { + for ((key, value) in values) { + val name = if (prefix.isEmpty()) key.toString() else "$prefix.$key" + when (value) { + is Boolean -> into[name] = AttributeValue.Bool(value) + is Int, is Long, is Short, is Byte, is BigInteger -> into[name] = AttributeValue.Int((value as Number).toLong()) + is Number -> into[name] = AttributeValue.Double(value.toDouble()) + is String -> into[name] = AttributeValue.Str(value) + is Map<*, *> -> flatten(value, name, into) + else -> {} // sequences carry nothing the core reads + } + } +} diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/Telemetry.kt b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/Telemetry.kt new file mode 100644 index 00000000..c4b83475 --- /dev/null +++ b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/Telemetry.kt @@ -0,0 +1,374 @@ +/* + * Copyright 2026 LiveKit, 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 io.livekit.android.telemetry + +import android.content.Context +import android.os.Build +import io.livekit.android.Version +import io.livekit.android.events.DisconnectReason +import io.livekit.android.room.track.Track +import io.livekit.android.telemetry.TelemetryOptions.Instrument +import io.livekit.android.util.LKLog +import io.livekit.android.util.LoggingLevel +import io.livekit.android.util.executeAsync +import io.livekit.uniffi.LogForwardFilter +import io.livekit.uniffi.LogForwardLevel +import io.livekit.uniffi.TelemetryScope +import io.livekit.uniffi.TelemetrySpan +import io.livekit.uniffi.logForwardBootstrap +import io.livekit.uniffi.logForwardReceive +import io.livekit.uniffi.telemetryConfigure +import io.livekit.uniffi.telemetryDeviceEvent +import io.livekit.uniffi.telemetryDiagnostics +import io.livekit.uniffi.telemetryDisconnectReason +import io.livekit.uniffi.telemetryLog +import io.livekit.uniffi.telemetryScope +import io.livekit.uniffi.telemetrySetAttribute +import io.livekit.uniffi.telemetrySetServer +import io.livekit.uniffi.telemetryShutdown +import kotlinx.coroutines.DelicateCoroutinesApi +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.GlobalScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.launch +import livekit.LivekitModels +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.RequestBody.Companion.toRequestBody +import uniffi.livekit_telemetry.Attribute +import uniffi.livekit_telemetry.AttributeValue +import uniffi.livekit_telemetry.DeviceEvent +import uniffi.livekit_telemetry.ExportException +import uniffi.livekit_telemetry.ExportRequest +import uniffi.livekit_telemetry.ExportResponse +import uniffi.livekit_telemetry.LogRecord +import uniffi.livekit_telemetry.LogSource +import uniffi.livekit_telemetry.Sdk +import uniffi.livekit_telemetry.Severity +import uniffi.livekit_telemetry.SpanName +import uniffi.livekit_telemetry.SpanOutcome +import uniffi.livekit_telemetry.SpanTrack +import uniffi.livekit_telemetry.TelemetryConfig +import uniffi.livekit_telemetry.TelemetryInstrument +import uniffi.livekit_telemetry.TelemetryResource +import uniffi.livekit_telemetry.TelemetryTransport +import uniffi.livekit_telemetry.TrackKind +import uniffi.livekit_telemetry.TrackSource +import java.io.File +import java.io.IOException +import java.util.concurrent.TimeUnit +import kotlin.coroutines.cancellation.CancellationException +import uniffi.livekit_telemetry.DisconnectReason as FfiDisconnectReason +import uniffi.livekit_telemetry.Instrument as FfiInstrument + +/** + * Client telemetry. The pipeline lives in the Rust core, one per process like a logger, and so do + * the instruments it runs: Android only builds the platform ones and hands them over. Configure + * with [io.livekit.android.LiveKit.setTelemetry] before creating Rooms, like the logger. + */ +object Telemetry { + /** What was configured: the instruments a Room starts (`room`, `rtc`) and the log gate. */ + @PublishedApi + @Volatile + internal var options: TelemetryOptions? = null + + /** + * The span the current coroutine is working inside, if any. Bound by the SDK around operations + * (connect, a reconnect cycle) with `asContextElement`, so warn/error records point at it + * without any handle being passed around; does not cross the WebRTC callback boundary. + */ + internal val currentSpan = ThreadLocal() + + /** + * The Room's scope the current coroutine works for, if any: bound on the Room's, the engine's + * and the signal client's coroutine scopes, so a warn/error record from a Room handler is + * filed under that Room's session even with no span in flight. + */ + internal val currentScope = ThreadLocal() + + private var coreLogs: Job? = null + + /** + * Set or change the options. The pipeline starts now, so pre-connect errors are captured; its + * destination waits for the first connect unless the options name an endpoint. Fail-open: the + * app runs without telemetry rather than not at all. + */ + fun configure(appContext: Context, options: TelemetryOptions) { + val context = appContext.applicationContext + this.options = options + try { + val instruments = buildList { + if (Instrument.DEVICE in options.instruments) add(DeviceTelemetry(context)) + } + telemetryConfigure(options.toConfig(context), OkHttpTelemetryTransport(), instruments) + forwardCoreLogs() + } catch (e: Throwable) { + this.options = null + LKLog.w(e) { "Telemetry could not start; running without it." } + } + } + + /** Turn telemetry off after a bounded final flush. */ + suspend fun shutdown() { + if (options == null) return + options = null + telemetryShutdown() + } + + /** + * Attach an attribute to every record of every scope — an `enduser.id`, a tenant, a build + * flavor. Strings, numbers and booleans keep their type; `null` removes the attribute. + */ + fun setAttribute(key: String, value: Any?) { + if (options != null) telemetrySetAttribute(key, value?.lowered()) + } + + /** + * A one-line readout of the pipeline's health for a debug console: status, throughput, + * backlog and losses. + */ + fun diagnostics(): String = if (options != null) telemetryDiagnostics() else "telemetry off" + + /** A Room's scope on the process pipeline; null when telemetry is off (the core is not touched then). */ + internal fun scope(): TelemetryScope? = if (options != null) telemetryScope() else null + + internal fun enabled(instrument: Instrument): Boolean = options?.instruments?.contains(instrument) == true + + internal fun setServer(url: String, token: String) { + if (options != null) telemetrySetServer(url, token) + } + + internal fun deviceEvent(event: DeviceEvent) { + if (enabled(Instrument.DEVICE)) telemetryDeviceEvent(event) + } + + /** Whether [LKLog] hands records at [level] to telemetry, whatever the console level. */ + @PublishedApi + internal fun captures(level: LoggingLevel): Boolean { + val options = options ?: return false + return Instrument.LOGS in options.instruments && level != LoggingLevel.OFF && level >= options.logLevel + } + + /** + * A warn/error record from the SDK logger; the core files it under the ambient span's scope, + * else the ambient Room's, else the process. Telemetry's own lines never feed back into the + * pipeline. + */ + @PublishedApi + internal fun log(level: LoggingLevel, t: Throwable?, message: String) { + if (!captures(level)) return + val caller = Throwable().stackTrace.firstOrNull { frame -> + !frame.className.startsWith(LKLog::class.java.name) && !frame.className.startsWith(Telemetry::class.java.name) + } + if (caller?.className?.startsWith(OWN_PACKAGE) == true) return + val span = currentSpan.get() + val record = LogRecord( + severity = level.severity, + source = LogSource.SDK, + message = listOfNotNull(message.takeIf { it.isNotEmpty() }, t?.toString()).joinToString(": "), + logger = caller?.className?.substringAfterLast('.')?.substringBefore('$'), + function = caller?.methodName, + file = caller?.fileName, + line = caller?.lineNumber?.takeIf { it > 0 }?.toUInt(), + spanId = span?.context()?.spanId, + ) + val scope = currentScope.get() + if (span == null && scope != null) scope.log(record) else telemetryLog(record) + } + + /** WebRTC's native log lines; the core only lets `error` leave the device. */ + internal fun logWebRtc(tag: String, message: String) { + if (enabled(Instrument.LOGS)) telemetryLog(LogRecord(Severity.ERROR, LogSource.WEB_RTC, message, logger = tag)) + } + + /** + * The Rust core's log lines, into the SDK logger like the SDK's own (the pipeline reports its + * health at debug and prints the `describe()` lines there) and, at warn and above, into the + * pipeline as `ffi` records. + */ + @OptIn(DelicateCoroutinesApi::class) + private fun forwardCoreLogs() { + if (coreLogs != null) return + logForwardBootstrap(LogForwardFilter.DEBUG) + coreLogs = GlobalScope.launch(Dispatchers.Default) { + while (true) { + val entry = logForwardReceive() ?: break + val level = entry.level.loggingLevel + if (level >= LKLog.loggingLevel) LKLog.logger?.log(level, null, "${entry.target}: ${entry.message}") + if (level >= LoggingLevel.WARN && enabled(Instrument.LOGS)) { + telemetryLog(LogRecord(level.severity, LogSource.FFI, entry.message, logger = entry.target, file = entry.file, line = entry.line)) + } + } + } + } + + private const val OWN_PACKAGE = "io.livekit.android.telemetry." +} + +// MARK: - Transport + +/** + * The host's half of the pipeline: a dumb bytes mover. The core composed URL, headers and body; + * this only performs the POST and hands back whatever came back, so retry / drop / go-silent is + * decided the same way on every platform. Only a missing response is an error. + */ +internal class OkHttpTelemetryTransport( + private val client: OkHttpClient = OkHttpClient.Builder().callTimeout(10, TimeUnit.SECONDS).build(), +) : TelemetryTransport { + override suspend fun send(request: ExportRequest): ExportResponse { + val httpRequest = try { + Request.Builder() + .url(request.url) + .post(request.body.toRequestBody()) + .apply { request.headers.forEach { (name, value) -> header(name, value) } } + .build() + } catch (e: IllegalArgumentException) { + throw ExportException.Rejected("invalid request: ${e.message}") + } + val response = try { + client.newCall(httpRequest).executeAsync() + } catch (e: IOException) { + throw ExportException.Retryable(e.toString(), null) + } + return response.use { + ExportResponse(it.code.toUShort(), it.headers.toMap(), it.body?.bytes() ?: ByteArray(0)) + } + } +} + +// MARK: - Options lowering + +/** + * The core's record: what the app chose, plus the platform's part (who is reporting, where the + * cache lives). + */ +internal fun TelemetryOptions.toConfig(context: Context) = TelemetryConfig( + endpoint = endpoint, + headers = headers, + sdk = TelemetryResource( + sdk = Sdk.ANDROID, + sdkVersion = Version.CLIENT_VERSION, + osName = "android", + osVersion = Build.VERSION.RELEASE ?: "", + deviceModel = "${Build.MANUFACTURER} ${Build.MODEL}".trim(), + ), + storageDir = storageDirectory?.let { if (it.isAbsolute) it else File(context.cacheDir, it.path) }?.path, + flushIntervalMs = flushInterval.inWholeMilliseconds.coerceAtLeast(0).toULong(), + statsWindowMs = statsWindow.inWholeMilliseconds.coerceAtLeast(0).toULong(), + logSeverity = logLevel.severity, + disabledInstruments = Instrument.entries.filter { it !in instruments }.map { it.ffi }, +) + +private val Instrument.ffi: FfiInstrument + get() = when (this) { + Instrument.ROOM -> FfiInstrument.ROOM + Instrument.RTC -> FfiInstrument.RTC + Instrument.LOGS -> FfiInstrument.LOGS + Instrument.DEVICE -> FfiInstrument.DEVICE + } + +// MARK: - Attributes + +internal fun Any.lowered(): AttributeValue = when (this) { + is String -> AttributeValue.Str(this) + is Boolean -> AttributeValue.Bool(this) + is Int, is Long, is Short, is Byte -> AttributeValue.Int((this as Number).toLong()) + is Number -> AttributeValue.Double(toDouble()) + else -> AttributeValue.Str(toString()) +} + +internal fun Map.lowered(): List = map { (key, value) -> Attribute(key, value.lowered()) } + +// MARK: - Spans + +/** + * An SDK span in this scope's trace, stamped now in the core, nested under the ambient span; + * null when telemetry is off, so a span costs nothing there (call sites chain optionally). + */ +internal fun TelemetryScope?.begin(name: SpanName, parent: TelemetrySpan? = Telemetry.currentSpan.get()): TelemetrySpan? = + this?.start(name, parent) + +/** End successfully. */ +internal fun TelemetrySpan.end() = end(SpanOutcome.OK, null) + +/** + * End on an exception: `cancelled` for a cancellation, `error` otherwise, with the exception's + * type as the status message. + */ +internal fun TelemetrySpan.end(error: Throwable) { + if (error is CancellationException) cancel() else fail(error.errorType()) +} + +/** `error.type` for a span: the exception's class name. */ +internal fun Throwable.errorType(): String = javaClass.simpleName.ifEmpty { javaClass.name } + +/** The track a publish or subscribe span is about; call again once the sid is known. */ +internal fun TelemetrySpan.setTrack(kind: Track.Kind, source: Track.Source, sid: String? = null) { + spanTrack(kind, source, sid)?.let(::setTrack) +} + +internal fun spanTrack(kind: Track.Kind, source: Track.Source, sid: String? = null, remoteIdentity: String? = null): SpanTrack? { + val trackKind = kind.telemetry ?: return null + return SpanTrack(sid, trackKind, source.telemetry, remoteIdentity) +} + +// MARK: - Shared vocabulary + +internal val LoggingLevel.severity: Severity + get() = when (this) { + LoggingLevel.VERBOSE -> Severity.TRACE + LoggingLevel.DEBUG -> Severity.DEBUG + LoggingLevel.INFO -> Severity.INFO + LoggingLevel.WARN -> Severity.WARN + LoggingLevel.ERROR, LoggingLevel.WTF, LoggingLevel.OFF -> Severity.ERROR + } + +internal val LogForwardLevel.loggingLevel: LoggingLevel + get() = when (this) { + LogForwardLevel.ERROR -> LoggingLevel.ERROR + LogForwardLevel.WARN -> LoggingLevel.WARN + LogForwardLevel.INFO -> LoggingLevel.INFO + LogForwardLevel.DEBUG -> LoggingLevel.DEBUG + LogForwardLevel.TRACE -> LoggingLevel.VERBOSE + } + +internal val Track.Kind.telemetry: TrackKind? + get() = when (this) { + Track.Kind.AUDIO -> TrackKind.AUDIO + Track.Kind.VIDEO -> TrackKind.VIDEO + Track.Kind.UNRECOGNIZED -> null + } + +internal val Track.Source.telemetry: TrackSource + get() = when (this) { + Track.Source.CAMERA -> TrackSource.CAMERA + Track.Source.MICROPHONE -> TrackSource.MICROPHONE + Track.Source.SCREEN_SHARE -> TrackSource.SCREEN_SHARE + Track.Source.SCREEN_SHARE_AUDIO -> TrackSource.SCREEN_SHARE_AUDIO + Track.Source.UNKNOWN -> TrackSource.UNKNOWN + } + +/** + * The Room's disconnect reason in the shared vocabulary: the protocol's number (the SDK enum + * mirrors the protocol's names), or the client giving up on a reconnect. + */ +internal fun DisconnectReason.telemetry(reconnectFailed: Boolean): FfiDisconnectReason = + if (this == DisconnectReason.UNKNOWN_REASON && reconnectFailed) { + FfiDisconnectReason.RECONNECT_FAILED + } else { + telemetryDisconnectReason(LivekitModels.DisconnectReason.valueOf(name).number) + } diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/TelemetryOptions.kt b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/TelemetryOptions.kt new file mode 100644 index 00000000..ce7d6761 --- /dev/null +++ b/livekit-android-sdk/src/main/java/io/livekit/android/telemetry/TelemetryOptions.kt @@ -0,0 +1,75 @@ +/* + * Copyright 2026 LiveKit, 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 io.livekit.android.telemetry + +import io.livekit.android.util.LoggingLevel +import java.io.File +import kotlin.time.Duration +import kotlin.time.Duration.Companion.seconds + +/** + * Client telemetry: ships SDK diagnostics (warn/error records, per-track RTC statistics, device + * state, operation spans) out-of-band to an OTLP/HTTP collector. Process-wide: configure once with + * [io.livekit.android.LiveKit.setTelemetry] before creating Rooms. + */ +data class TelemetryOptions( + /** + * Full OTLP/HTTP logs URL, e.g. `http://localhost:4318/v1/logs` for a local collector. + * `null` (the default) derives it from the server the first Room connects to + * (`https:///observability/logs/otlp/v0`) and authenticates with the room token; + * until then everything is buffered on device. + */ + val endpoint: String? = null, + /** Extra request headers, e.g. `Authorization`. */ + val headers: Map = emptyMap(), + /** + * Directory for the on-disk batch cache. A relative path is resolved against the app's cache + * directory (the default, `livekit-telemetry`); `null` keeps batches in memory only. + */ + val storageDirectory: File? = File("livekit-telemetry"), + /** Export cadence. Stretched automatically under thermal / battery-saver pressure. */ + val flushInterval: Duration = 15.seconds, + /** RTC statistics window: one `lk.rtc.stats.sample` per track per window. */ + val statsWindow: Duration = 15.seconds, + /** Which instruments run; all by default. App-defined events and session identity are always on. */ + val instruments: Set = Instrument.ALL, + /** + * Lowest log level that leaves the device (warnings and errors by default). Events are not + * logs and are not subject to it; WebRTC's own logs go from [LoggingLevel.ERROR] regardless. + */ + val logLevel: LoggingLevel = LoggingLevel.WARN, +) { + /** The telemetry instruments, by area. Combine to choose what runs. */ + enum class Instrument { + /** Spans of the Room's operations: `lk.connect`, `lk.reconnect`, `lk.publish`. */ + ROOM, + + /** Track statistics windows and the `lk.subscribe` span (time to media). */ + RTC, + + /** Warning and error log records from the SDK, the Rust core and WebRTC. */ + LOGS, + + /** Device state — thermal, power, memory, network, battery — and audio / capture events. */ + DEVICE, + ; + + companion object { + val ALL: Set = entries.toSet() + } + } +} diff --git a/livekit-android-sdk/src/main/java/io/livekit/android/util/LKLog.kt b/livekit-android-sdk/src/main/java/io/livekit/android/util/LKLog.kt index 9fa634f9..27908af4 100644 --- a/livekit-android-sdk/src/main/java/io/livekit/android/util/LKLog.kt +++ b/livekit-android-sdk/src/main/java/io/livekit/android/util/LKLog.kt @@ -16,6 +16,7 @@ package io.livekit.android.util +import io.livekit.android.telemetry.Telemetry import io.livekit.android.util.LoggingLevel.DEBUG import io.livekit.android.util.LoggingLevel.ERROR import io.livekit.android.util.LoggingLevel.INFO @@ -109,8 +110,14 @@ class LKLog { /** @suppress */ inline fun log(loggingLevel: LoggingLevel, t: Throwable? = null, crossinline message: (() -> String)) { - if (loggingLevel >= LKLog.loggingLevel) { - logger?.log(loggingLevel, t, message()) + val console = loggingLevel >= LKLog.loggingLevel + // Telemetry captures warnings and errors whatever the console level. + if (console || Telemetry.captures(loggingLevel)) { + val text = message() + if (console) { + logger?.log(loggingLevel, t, text) + } + Telemetry.log(loggingLevel, t, text) } } } diff --git a/livekit-android-test/build.gradle b/livekit-android-test/build.gradle index 749b42ce..61280206 100644 --- a/livekit-android-test/build.gradle +++ b/livekit-android-test/build.gradle @@ -35,6 +35,17 @@ android { testOptions { unitTests { includeAndroidResources = true + all { test -> + // Robolectric runs the real Rust core (livekit_uniffi) through JNA; point it at a host + // build of the library: -PlivekitUniffiLibraryPath=… or LIVEKIT_UNIFFI_LIBRARY_PATH. + def uniffiLibraryPath = project.findProperty('livekitUniffiLibraryPath') ?: System.getenv('LIVEKIT_UNIFFI_LIBRARY_PATH') + if (uniffiLibraryPath) { + test.systemProperty 'jna.library.path', uniffiLibraryPath + } + // UniFFI objects register with android.system.SystemCleaner (API 34+), which Robolectric + // implements on top of a JDK-internal cleaner. + test.jvmArgs '--add-exports=java.base/jdk.internal.ref=ALL-UNNAMED' + } } } lint { @@ -129,6 +140,7 @@ dependencies { testImplementation libs.junit testImplementation libs.robolectric + testImplementation libs.jna testImplementation libs.okhttp.mockwebserver testImplementation "org.jetbrains.kotlin:kotlin-reflect:$kotlin_version" kaptTest libs.dagger.compiler diff --git a/livekit-android-test/src/main/AndroidManifest.xml b/livekit-android-test/src/main/AndroidManifest.xml index 8bdb7e14..82ffd941 100644 --- a/livekit-android-test/src/main/AndroidManifest.xml +++ b/livekit-android-test/src/main/AndroidManifest.xml @@ -1,4 +1,8 @@ - + + + + diff --git a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt index 6a51cfeb..539b3127 100644 --- a/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt +++ b/livekit-android-test/src/main/java/io/livekit/android/test/mock/MockPeerConnection.kt @@ -209,6 +209,14 @@ class MockPeerConnection( callback?.onStatsDelivered(RTCStatsReport(0, emptyMap())) } + override fun getStats(sender: RtpSender?, callback: RTCStatsCollectorCallback?) { + callback?.onStatsDelivered(RTCStatsReport(0, emptyMap())) + } + + override fun getStats(receiver: RtpReceiver?, callback: RTCStatsCollectorCallback?) { + callback?.onStatsDelivered(RTCStatsReport(0, emptyMap())) + } + override fun setBitrate(min: Int?, current: Int?, max: Int?): Boolean { return true } diff --git a/livekit-android-test/src/test/java/io/livekit/android/room/RoomTest.kt b/livekit-android-test/src/test/java/io/livekit/android/room/RoomTest.kt index cde24d5f..1ba17718 100644 --- a/livekit-android-test/src/test/java/io/livekit/android/room/RoomTest.kt +++ b/livekit-android-test/src/test/java/io/livekit/android/room/RoomTest.kt @@ -71,6 +71,7 @@ import org.mockito.kotlin.doSuspendableAnswer import org.mockito.kotlin.stub import org.mockito.kotlin.whenever import org.robolectric.RobolectricTestRunner +import uniffi.livekit_telemetry.ReconnectReason @ExperimentalCoroutinesApi @RunWith(RobolectricTestRunner::class) @@ -229,7 +230,7 @@ class RoomTest { callback.onAvailable(network) } - Mockito.verify(rtcEngine).reconnect() + Mockito.verify(rtcEngine).reconnect(ReconnectReason.NETWORK_CHANGED) } @Test diff --git a/livekit-android-test/src/test/java/io/livekit/android/telemetry/TelemetryMockE2ETest.kt b/livekit-android-test/src/test/java/io/livekit/android/telemetry/TelemetryMockE2ETest.kt new file mode 100644 index 00000000..53b714e3 --- /dev/null +++ b/livekit-android-test/src/test/java/io/livekit/android/telemetry/TelemetryMockE2ETest.kt @@ -0,0 +1,275 @@ +/* + * Copyright 2026 LiveKit, 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 io.livekit.android.telemetry + +import androidx.test.core.app.ApplicationProvider +import io.livekit.android.room.ReconnectType +import io.livekit.android.room.SignalClient +import io.livekit.android.test.MockE2ETest +import io.livekit.android.test.mock.TestData +import io.livekit.android.test.mock.room.track.createMockLocalAudioTrack +import io.livekit.android.util.LKLog +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.asContextElement +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.booleanOrNull +import kotlinx.serialization.json.contentOrNull +import kotlinx.serialization.json.intOrNull +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import livekit.LivekitRtc +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertTrue +import org.junit.Assume.assumeTrue +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TestRule +import org.junit.runner.RunWith +import org.junit.runners.model.Statement +import org.robolectric.RobolectricTestRunner +import java.io.File +import java.net.InetSocketAddress +import java.net.Socket +import java.util.UUID +import kotlin.time.Duration.Companion.seconds + +/** + * End to end through the Rust core on the SDK's mocks: mock websocket, mock peer connections and + * a mock local audio track. Needs the collector the Swift tests use — `otelcol-contrib --config + * Tests/LiveKitCoreTests/Telemetry/otelcol-lgtm.yaml`, listening on :4319 and writing OTLP/JSON + * lines to [COLLECTOR_OUTPUT] (the file is shared with other runs, so records are filtered by time, + * never truncated: the collector keeps its write offset) — and a host build of `livekit_uniffi` on + * `jna.library.path` (`-PlivekitUniffiLibraryPath=…`). Skipped when the collector is not there. + */ +@ExperimentalCoroutinesApi +@RunWith(RobolectricTestRunner::class) +class TelemetryMockE2ETest : MockE2ETest() { + + /** Unix nanoseconds when the pipeline started: the device instrument reports its initial state right then. */ + private var startNs = 0L + + /** + * Telemetry is process-wide and a Room takes its scope at creation, so it is configured around + * the whole test, before [mocksSetup] creates the Room, like an app would at launch. + */ + @get:Rule + val telemetryRule = TestRule { base, _ -> + object : Statement() { + override fun evaluate() { + assumeTrue("collector on $COLLECTOR_ENDPOINT", collectorReachable()) + startNs = System.currentTimeMillis() * 1_000_000 + Telemetry.configure( + ApplicationProvider.getApplicationContext(), + TelemetryOptions(endpoint = COLLECTOR_ENDPOINT, storageDirectory = null, flushInterval = 1.seconds, statsWindow = 2.seconds), + ) + assumeTrue("telemetry pipeline started (is livekit_uniffi on jna.library.path?)", Telemetry.options != null) + try { + base.evaluate() + } finally { + runBlocking(Dispatchers.IO) { Telemetry.shutdown() } + } + } + } + } + + @Test + fun sessionReachesTheCollector() = runTest { + val marker = "telemetry e2e ${UUID.randomUUID()}" + Telemetry.setAttribute("acme.tenant", marker) + room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT) + + connect() + val traceId = room.telemetryTraceId + assertNotNull("each Room has a printable session trace id", traceId) + assertEquals(32, traceId!!.length) + + // A quick reconnect on the mocks: the primary (subscriber) peer connection fails, the + // websocket reconnects, ICE reconnects. + disconnectPeerConnection() + testScheduler.advanceTimeBy(1000) + reconnectWebsocket() + connectPeerConnection() + // Bounded: the rtc instrument's 1 s stats loop keeps virtual time from ever going idle. + testScheduler.advanceTimeBy(1000) + + // A published track: the server's TrackPublished answer completes the add-track request. + val publish = launch { room.localParticipant.publishAudioTrack(createMockLocalAudioTrack()) } + simulateMessageFromServer(TestData.LOCAL_TRACK_PUBLISHED) + publish.join() + assertEquals(1, room.localParticipant.audioTrackPublications.size) + + // A warn/error record emitted inside an operation ends up on that operation's span, in + // this Room's trace — the whole path: LKLog → Telemetry → core. From a Room handler with + // no span in flight it is filed under the Room's session; outside any Room context it is + // the process scope. + val op = room.beginSpan("e2e.op") + withContext(Telemetry.currentSpan.asContextElement(op)) { LKLog.e { marker } } + op?.end() + val unknownSid = "TR_unknown_$marker" // the server unpublishes a track we never had: the Room handler warns + simulateMessageFromServer( + LivekitRtc.SignalResponse.newBuilder() + .setTrackUnpublished(LivekitRtc.TrackUnpublishedResponse.newBuilder().setTrackSid(unknownSid)) + .build(), + ) + LKLog.e { "$marker process" } + room.emitTelemetryEvent("e2e.checkpoint", mapOf("e2e.marker" to marker)) + + Thread.sleep(2500) // stats polls (the mocks yield empty reports, so no windows) and a flush + room.disconnect() + Thread.sleep(3000) // the disconnect flush, and the collector's write + val diagnostics = Telemetry.diagnostics() + println("telemetry: trace $traceId — $diagnostics") + + val otlp = OtlpFile(File(COLLECTOR_OUTPUT), since = startNs) + val spans = otlp.spans.filter { it.traceId == traceId } + + // The user-initiated connect, with its steps; the reconnect is its own span. + val connect = spans.filter { it.name == "lk.connect" }.also { assertEquals("one lk.connect per session: $spans", 1, it.size) }.single() + // No answer_sent: the mocks report ICE connected before the launched answer coroutine runs. + val steps = listOf("ws_open", "signal", "join_recv", "pc_created", "engine", "pc_connected", "room_connected") + assertTrue(connect.events.toString(), connect.events.containsAll(steps)) + assertEquals("ok", connect.attributes["lk.outcome"]) + + val reconnect = spans.filter { it.name == "lk.reconnect" }.also { assertEquals("one lk.reconnect: $spans", 1, it.size) }.single() + assertEquals("subscriber_failed", reconnect.attributes["lk.reconnect.reason"]) + assertEquals("quick", reconnect.attributes["lk.reconnect.mode"]) + assertEquals("ok", reconnect.attributes["lk.outcome"]) + assertTrue(reconnect.events.toString(), "attempt 1 quick" in reconnect.events) + + val publishSpan = spans.filter { it.name == "lk.publish" }.also { assertEquals("one lk.publish: $spans", 1, it.size) }.single() + assertEquals("ok", publishSpan.attributes["lk.outcome"]) + assertEquals("audio", publishSpan.attributes["lk.track.kind"]) + assertEquals("microphone", publishSpan.attributes["lk.track.source"]) + assertEquals(TestData.LOCAL_AUDIO_TRACK.sid, publishSpan.attributes["lk.track.sid"]) + + val logs = otlp.logs + val inSpan = logs.firstOrNull { it.body == marker }.also { assertNotNull("error record reached the collector", it) }!! + val opSpan = otlp.spans.firstOrNull { it.name == "e2e.op" }.also { assertNotNull("custom span reached the collector", it) }!! + assertEquals("the record points at the span it was emitted in", opSpan.spanId, inSpan.spanId) + assertEquals("...and therefore lands in this Room's trace", traceId, inSpan.traceId) + assertEquals("roomname", inSpan.attributes["lk.room.name"]) + assertEquals(TestData.LOCAL_PARTICIPANT.identity, inSpan.attributes["lk.participant.identity"]) + val handler = logs.firstOrNull { it.body?.endsWith(unknownSid) == true }.also { assertNotNull("Room-handler warning reached the collector", it) }!! + assertTrue("a Room handler with no span in flight: the Room's session, no span", handler.spanId.isEmpty() && handler.traceId == traceId) + val process = logs.firstOrNull { it.body == "$marker process" }.also { assertNotNull(it) }!! + assertTrue("outside any Room context: the process scope", process.spanId.isEmpty() && process.traceId != traceId) + assertTrue(logs.any { it.eventName == "custom.e2e.checkpoint" && it.attributes["e2e.marker"] == marker }) + for (event in listOf("lk.device.thermal.changed", "lk.device.memory.changed", "lk.device.network.changed", "lk.device.low_power.changed")) { + assertTrue("$event initial value reached the collector", logs.any { it.eventName == event }) + } + // The rule: no info-level log record leaves the device. + assertTrue(logs.filter { it.eventName.isEmpty() }.all { it.severity >= SEVERITY_WARN }) + assertTrue("the pipeline-wide attribute reaches the Room's scope", logs.any { it.traceId == traceId && it.attributes["acme.tenant"] == marker }) + // The Room hung up itself, and said so before the disconnect flush. + val ended = logs.filter { it.eventName == "lk.room.disconnected" && it.traceId == traceId } + assertEquals(ended.map { it.attributes }.toString(), 1, ended.size) + assertEquals("client_initiated", ended.single().attributes["lk.disconnect.reason"]) + // The pipeline's own health: the whole session shipped. + assertTrue(diagnostics, "lost 0" in diagnostics) + } + + private fun reconnectWebsocket() { + wsFactory.listener.onOpen(wsFactory.ws, createOpenResponse(wsFactory.request)) + val softReconnectParam = wsFactory.request.url.queryParameter(SignalClient.CONNECT_QUERY_RECONNECT)?.toIntOrNull() ?: 0 + simulateMessageFromServer(if (softReconnectParam == 0) TestData.JOIN else TestData.RECONNECT) + } + + companion object { + const val COLLECTOR_ENDPOINT = "http://127.0.0.1:4319/v1/logs" + const val COLLECTOR_OUTPUT = "/tmp/livekit-telemetry-otlp.jsonl" + private const val SEVERITY_WARN = 13 + + private fun collectorReachable() = runCatching { Socket().use { it.connect(InetSocketAddress("127.0.0.1", 4319), 500) } }.isSuccess + } +} + +/** What the collector wrote: OTLP/JSON, one export request per line, from [since] (Unix nanoseconds) on. */ +class OtlpFile(file: File, since: Long) { + class Log(val eventName: String, val body: String?, val traceId: String, val spanId: String, val severity: Int, val attributes: Map) + + /** [events] are the span's checkpoints (`ws_open`, `first_media`, `attempt 1 quick`, ...). */ + class Span(val name: String, val traceId: String, val spanId: String, val attributes: Map, val events: List) + + val logs = mutableListOf() + val spans = mutableListOf() + + init { + for (line in file.readLines()) { + val request = runCatching { Json.parseToJsonElement(line).jsonObject }.getOrNull() ?: continue + for (resource in request.array("resourceLogs")) { + for (scope in resource.jsonObject.array("scopeLogs")) { + for (record in scope.jsonObject.array("logRecords")) { + val r = record.jsonObject + if (r.long("timeUnixNano") < since) continue + logs += Log( + eventName = r.string("eventName") ?: "", + body = r["body"]?.jsonObject?.string("stringValue"), + traceId = r.string("traceId") ?: "", + spanId = r.string("spanId") ?: "", + severity = r["severityNumber"]?.jsonPrimitive?.intOrNull ?: 0, + attributes = attributes(r["attributes"]), + ) + } + } + } + for (resource in request.array("resourceSpans")) { + for (scope in resource.jsonObject.array("scopeSpans")) { + for (span in scope.jsonObject.array("spans")) { + val s = span.jsonObject + if (s.long("startTimeUnixNano") < since) continue + spans += Span( + name = s.string("name") ?: "", + traceId = s.string("traceId") ?: "", + spanId = s.string("spanId") ?: "", + attributes = attributes(s["attributes"]), + events = s.array("events").mapNotNull { it.jsonObject.string("name") }, + ) + } + } + } + } + } + + private fun JsonObject.array(key: String): List = this[key]?.jsonArray ?: emptyList() + + private fun JsonObject.string(key: String): String? = (this[key] as? JsonPrimitive)?.contentOrNull + + private fun JsonObject.long(key: String): Long = string(key)?.toLongOrNull() ?: 0L + + /** OTLP/JSON attributes (`[{key, value: {stringValue | intValue | boolValue | doubleValue}}]`) as strings. */ + private fun attributes(value: JsonElement?): Map = + (value?.jsonArray ?: emptyList()).mapNotNull { pair -> + val entry = pair.jsonObject + val key = entry.string("key") ?: return@mapNotNull null + val any = entry["value"]?.jsonObject ?: return@mapNotNull null + val text = any.string("stringValue") + ?: any.string("intValue") + ?: any["boolValue"]?.jsonPrimitive?.booleanOrNull?.toString() + ?: any.string("doubleValue") + ?: return@mapNotNull null + key to text + }.toMap() +} diff --git a/settings.gradle b/settings.gradle index b5882cbb..b2b98054 100644 --- a/settings.gradle +++ b/settings.gradle @@ -14,6 +14,7 @@ pluginManagement { dependencyResolutionManagement { repositoriesMode.set(RepositoriesMode.FAIL_ON_PROJECT_REPOS) repositories { + mavenLocal() // prototype: local build of livekit-uniffi-android google() mavenCentral() maven { url 'https://jitpack.io' }