Skip to content

RUM-17613: timeseries [3/5] Pass the live RUM context to timeseries pipelines #3717

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
satween wants to merge 1 commit into
base: tvaleev/feature/RUM-17613-2-timeseries-schema
Choose a base branch
from
Open

RUM-17613: timeseries [3/5] Pass the live RUM context to timeseries pipelines #3717

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...roid-rum/src/main/kotlin/com/datadog/android/rum/internal/domain/scope/RumSessionScope.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,11 @@ internal class RumSessionScope(
}
}

timeseriesCollector.onViewTypeUpdate(getActiveRumContext().viewType)
// getActiveRumContext() copies the whole context chain, so it is only built when there is a
// collector to feed: timeseries collection is opt-in and this runs on every RUM event.
if (timeseriesCollector !is NoOpTimeseriesCollector) {
timeseriesCollector.onRumContextUpdate(getActiveRumContext())
}

return if (isSessionComplete()) {
null
Expand Down
30 changes: 16 additions & 14 deletions ...src/main/kotlin/com/datadog/android/rum/internal/timeseries/DefaultTimeseriesCollector.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ package com.datadog.android.rum.internal.timeseries
import androidx.annotation.WorkerThread
import com.datadog.android.api.InternalLogger
import com.datadog.android.core.internal.utils.scheduleSafe
import com.datadog.android.rum.internal.domain.RumContext
import com.datadog.android.rum.internal.domain.scope.RumViewType
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
Expand All @@ -24,11 +25,11 @@ import java.util.concurrent.atomic.AtomicReference
* - After [onSessionStop], the instance must not be restarted; create a new instance.
*
* Background sampling:
* - When [collectInBackground] is `false`, [onViewTypeUpdate] pauses sampling on leaving foreground
* - When [collectInBackground] is `false`, [onRumContextUpdate] pauses sampling on leaving foreground
* and resumes it on returning to foreground.
*
* Threading:
* - [onSessionStart] / [onSessionStop] / [onViewTypeUpdate] are called from the RUM event-handler thread.
* - [onSessionStart] / [onSessionStop] / [onRumContextUpdate] are called from the RUM event-handler thread.
* - Sampling tasks run on a [scheduledExecutorService] shared with other RUM components
* (owned by the SDK core); the instance neither owns nor shuts it down.
* - [Pipeline] is responsible for its own thread safety via internal synchronization.
Expand All @@ -40,7 +41,10 @@ internal class DefaultTimeseriesCollector(
private val internalLogger: InternalLogger,
internal val pipelines: List<Pipeline<*>>,
private val collectInBackground: Boolean,
internal val scheduledExecutorService: ScheduledExecutorService
internal val scheduledExecutorService: ScheduledExecutorService,
// Read by the sampling ticks so every batch is attributed to the view that was active when it
// was drained, not to the one active when the collector was created.
@Volatile private var rumContext: RumContext
) : TimeseriesCollector {

private enum class State { IDLE, RUNNING, SUSPENDED, STOPPED }
Expand All @@ -49,9 +53,6 @@ internal class DefaultTimeseriesCollector(
// Incremented on every start/resume. Ticks carrying a stale generation self-terminate.
private val currentGeneration = AtomicInteger(0)

@Volatile
private var currentViewType: RumViewType? = null

@WorkerThread
override fun onSessionStart() {
if (state.compareAndSet(State.IDLE, State.RUNNING)) {
Expand All @@ -64,7 +65,7 @@ internal class DefaultTimeseriesCollector(
if (state.getAndSet(State.STOPPED) != State.STOPPED) {
pipelines.forEach { pipeline ->
try {
synchronized(pipeline, pipeline::flush)
pipeline.flush(rumContext)
} catch (@Suppress("TooGenericExceptionCaught") t: Throwable) {
internalLogger.log(
level = InternalLogger.Level.ERROR,
Expand All @@ -78,10 +79,13 @@ internal class DefaultTimeseriesCollector(
}

@WorkerThread
override fun onViewTypeUpdate(newViewType: RumViewType) {
if (newViewType == currentViewType) return
val isEnterForeground = !currentViewType.isForeground && newViewType.isForeground
currentViewType = newViewType
override fun onRumContextUpdate(newRumContext: RumContext) {
val oldViewType = rumContext.viewType
val newViewType = newRumContext.viewType
val isEnterForeground = !oldViewType.isForeground && newViewType.isForeground

rumContext = newRumContext

if (!collectInBackground) {
if (isEnterForeground && state.compareAndSet(State.SUSPENDED, State.RUNNING)) {
startSampling()
Expand All @@ -105,9 +109,7 @@ internal class DefaultTimeseriesCollector(
) {
if (!isActive(generation)) return@scheduleSafe
try {
synchronized(pipeline) {
if (isActive(generation)) pipeline.execute()
}
pipeline.execute(rumContext)

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Snapshot the RUM context when the batch is drained

When a view transition is handled while reader.read() is in progress, this argument has already captured the previous rumContext; after the read completes, Pipeline.execute() can drain the buffer using that stale view ID even though the new view is active. This contradicts the intended drain-time attribution and can compute session.hasReplay from the wrong view, so the live context should be obtained after sampling, at the point the buffer is drained.

Useful? React with 👍 / 👎.

} catch (@Suppress("TooGenericExceptionCaught") t: Throwable) {
internalLogger.log(
level = InternalLogger.Level.ERROR,
Expand Down
19 changes: 5 additions & 14 deletions ...n/kotlin/com/datadog/android/rum/internal/timeseries/DefaultTimeseriesCollectorFactory.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,12 +45,12 @@ internal class DefaultTimeseriesCollectorFactory(
val pipelines = mutableListOf<Pipeline<*>>()

if (TimeseriesType.CPU in configuration.enabledTypes) {
pipelines += createCpuPipeline(sessionType, rumContext, customAttributes)
pipelines += createCpuPipeline(sessionType, customAttributes)
}

if (TimeseriesType.MEMORY in configuration.enabledTypes) {
if (totalRamBytes > 0L) {
pipelines += createMemoryPipeline(sessionType, rumContext, customAttributes)
pipelines += createMemoryPipeline(sessionType, customAttributes)
} else {
sdkCore.internalLogger.log(
InternalLogger.Level.WARN,
Expand All @@ -65,15 +65,12 @@ internal class DefaultTimeseriesCollectorFactory(
internalLogger = sdkCore.internalLogger,
collectInBackground = configuration.collectInBackground,
scheduledExecutorService = scheduledExecutorService,
rumContext = rumContext,
pipelines = pipelines
)
}

private fun createMemoryPipeline(
sessionType: RumSessionType,
rumContext: RumContext,
customAttributes: () -> Map<String, Any?>
) = Pipeline(
private fun createMemoryPipeline(sessionType: RumSessionType, customAttributes: () -> Map<String, Any?>) = Pipeline(
sdkCore = sdkCore,
reader = VitalReaderWrapper(
vitalReader = MemoryVitalReader(internalLogger = sdkCore.internalLogger),
Expand All @@ -90,16 +87,11 @@ internal class DefaultTimeseriesCollectorFactory(
internalLogger = sdkCore.internalLogger
),
dataWriter = dataWriter,
rumContext = rumContext,
customAttributes = customAttributes,
insightsCollector = insightsCollector
)

private fun createCpuPipeline(
sessionType: RumSessionType,
rumContext: RumContext,
customAttributes: () -> Map<String, Any?>
) = Pipeline(
private fun createCpuPipeline(sessionType: RumSessionType, customAttributes: () -> Map<String, Any?>) = Pipeline(
sdkCore = sdkCore,
reader = CpuDatapointReader(
cpuStatReader = CpuStatReader(internalLogger = sdkCore.internalLogger),
Expand All @@ -115,7 +107,6 @@ internal class DefaultTimeseriesCollectorFactory(
internalLogger = sdkCore.internalLogger
),
dataWriter = dataWriter,
rumContext = rumContext,
customAttributes = customAttributes,
insightsCollector = insightsCollector
)
Expand Down
16 changes: 10 additions & 6 deletions ...d-sdk-android-rum/src/main/kotlin/com/datadog/android/rum/internal/timeseries/Pipeline.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,22 +27,26 @@ internal class Pipeline<T : Any>(
private val buffer: Buffer<T>,
private val eventFactory: EventFactory<T, *>,
private val dataWriter: DataWriter<Any>,
private val rumContext: RumContext,
private val customAttributes: () -> Map<String, Any?>,
private val insightsCollector: InsightsCollector = NoOpInsightsCollector()
) {
val intervalMs: Long get() = reader.intervalMs

@WorkerThread
fun execute() {
reader.read()?.let(buffer::add)
if (buffer.isFull()) drainAndWrite()
fun execute(rumContext: RumContext) {
// reader.read() may hit the filesystem (/proc); kept outside the lock so a concurrent
// flush() waits only for the buffer and never for I/O.
val dataPoint = reader.read()
synchronized(this) {
Comment on lines +39 to +40

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Keep terminal flush ordered after in-flight reads

When a session or the RUM feature stops while a scheduled tick is inside reader.read(), onSessionStop() can flush the buffer before this block acquires the lock. The tick then adds its point after the final flush, does not reschedule because its generation is inactive, and the collector is discarded, so that point is permanently lost. The terminal flush needs to wait for an in-flight read/add operation or arrange another flush after the point is added.

Useful? React with 👍 / 👎.

dataPoint?.let(buffer::add)
if (buffer.isFull()) drainAndWrite(rumContext)
}
}

@WorkerThread
fun flush() = drainAndWrite()
fun flush(rumContext: RumContext) = synchronized(this) { drainAndWrite(rumContext) }

private fun drainAndWrite() {
private fun drainAndWrite(rumContext: RumContext) {
val dataPoints = buffer.drain().ifEmpty { return }
// Snapshotted here and not at serialization time: serialization runs later on the core
// context thread, by then the RUM monitor owning those attributes may already be
Expand Down
3 changes: 1 addition & 2 deletions ...id-rum/src/main/kotlin/com/datadog/android/rum/internal/timeseries/TimeseriesCollector.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,13 @@ package com.datadog.android.rum.internal.timeseries

import com.datadog.android.rum.RumSessionType
import com.datadog.android.rum.internal.domain.RumContext
import com.datadog.android.rum.internal.domain.scope.RumViewType
import com.datadog.tools.annotation.NoOpImplementation

@NoOpImplementation
internal interface TimeseriesCollector {
fun onSessionStart()
fun onSessionStop()
fun onViewTypeUpdate(newViewType: RumViewType)
fun onRumContextUpdate(newRumContext: RumContext)

@NoOpImplementation
interface Factory {
Expand Down
23 changes: 13 additions & 10 deletions ...-rum/src/test/kotlin/com/datadog/android/rum/internal/domain/scope/RumSessionScopeTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1974,9 +1974,9 @@ internal class RumSessionScopeTest {
// Then — one timeseries per tracked session, updates go to the live one only
verify(mockTimeseriesCollectorFactory, times(2)).create(any(), any(), any())
verify(firstTimeseriesCollector).onSessionStop()
verify(firstTimeseriesCollector, times(1)).onViewTypeUpdate(fakeViewContext.viewType)
verify(firstTimeseriesCollector, times(1)).onRumContextUpdate(fakeViewContext)
verify(secondTimeseriesCollector).onSessionStart()
verify(secondTimeseriesCollector).onViewTypeUpdate(fakeViewContext.viewType)
verify(secondTimeseriesCollector).onRumContextUpdate(fakeViewContext)
}

@Test
Expand Down Expand Up @@ -2034,14 +2034,14 @@ internal class RumSessionScopeTest {
// When
testedScope.handleEvent(forge.addErrorEvent(), fakeDatadogContext, mockEventWriteScope, mockWriter)

// Then — the session scope view type, once per handled event
val viewTypeCaptor = argumentCaptor<RumViewType>()
verify(mockTimeseriesCollector, times(2)).onViewTypeUpdate(viewTypeCaptor.capture())
assertThat(viewTypeCaptor.allValues).containsOnly(testedScope.getRumContext().viewType)
// Then — the session scope context, once per handled event
val rumContextCaptor = argumentCaptor<RumContext>()
verify(mockTimeseriesCollector, times(2)).onRumContextUpdate(rumContextCaptor.capture())
assertThat(rumContextCaptor.allValues).containsOnly(testedScope.getRumContext())
}

@Test
fun `M pass active view type W onViewTypeUpdate() { StartView creates a foreground view }`(forge: Forge) {
fun `M pass active view context W onRumContextUpdate() { StartView creates a foreground view }`(forge: Forge) {
// Given — a real view manager child scope, so the context comes from an actual RumViewScope
initializeTestedScope(
withMockChildScope = false,
Expand All @@ -2058,9 +2058,12 @@ internal class RumSessionScopeTest {
)

// Then
val viewTypeCaptor = argumentCaptor<RumViewType>()
verify(mockTimeseriesCollector).onViewTypeUpdate(viewTypeCaptor.capture())
assertThat(viewTypeCaptor.firstValue).isEqualTo(RumViewType.FOREGROUND)
val rumContextCaptor = argumentCaptor<RumContext>()
verify(mockTimeseriesCollector).onRumContextUpdate(rumContextCaptor.capture())
assertThat(rumContextCaptor.firstValue.viewType).isEqualTo(RumViewType.FOREGROUND)
assertThat(rumContextCaptor.firstValue.viewName).isEqualTo(fakeStartViewEvent.key.name)
assertThat(rumContextCaptor.firstValue.viewId).isNotNull
assertThat(rumContextCaptor.firstValue.sessionId).isEqualTo(testedScope.sessionId)
}

@Test
Expand Down
17 changes: 17 additions & 0 deletions ...oid-rum/src/test/kotlin/com/datadog/android/rum/internal/monitor/DatadogRumMonitorTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2971,6 +2971,23 @@ internal class DatadogRumMonitorTest {

// endregion

// region timeseries

@Test
fun `M stop the active session timeseries W stopTimeseries()`() {
// Given
val mockSessionScope = mock<RumSessionScope>()
whenever(mockApplicationScope.activeSession) doReturn mockSessionScope

// When
testedMonitor.stopTimeseries()

// Then
verify(mockSessionScope).stopTimeseries()
}

// endregion

@OptIn(ExperimentalRumApi::class)
@Test
fun `M produce StartOperation event W startOperation`(
Expand Down
6 changes: 3 additions & 3 deletions ...tlin/com/datadog/android/rum/internal/timeseries/DefaultTimeseriesCollectorFactoryTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,7 +223,7 @@ internal class DefaultTimeseriesCollectorFactoryTest {
}

@Test
fun `M propagate executor to the collector and rum context to the pipelines W create()`(
fun `M propagate executor and rum context to the collector W create()`(
@LongForgery(min = 1L) fakeTotalRamBytes: Long
) {
// Given
Expand All @@ -236,8 +236,8 @@ internal class DefaultTimeseriesCollectorFactoryTest {
check(timeseries is DefaultTimeseriesCollector)
assertThat(timeseries.scheduledExecutorService).isSameAs(mockExecutor)
assertThat(
timeseries.pipelines.map { it.getFieldValue<RumContext, Pipeline<*>>("rumContext") }
).containsOnly(fakeRumContext)
timeseries.getFieldValue<RumContext, DefaultTimeseriesCollector>("rumContext")
).isSameAs(fakeRumContext)
}

@ParameterizedTest
Expand Down
Loading
Loading