From 6ea9d10d0f0b77aae97fc41b73c1670414b47475 Mon Sep 17 00:00:00 2001 From: pauldambra Date: Sat, 17 Oct 2026 14:40:11 +0100 Subject: [PATCH 1/3] feat(server): add request context to OpenTelemetry spans Generated-By: PostHog Desktop Task-Id: 5681be44-4986-48db-b140-1bda88f75f96 --- .changeset/fiery-maps-obey.md | 5 + README.md | 8 + posthog-server/api/posthog-server.api | 10 + posthog-server/build.gradle.kts | 3 + posthog-server/gradle.lockfile | 4 + .../posthog/server/PostHogRequestContext.kt | 2 + .../server/internal/ContextSpanBridge.kt | 25 ++ .../internal/OpenTelemetrySpanContext.kt | 29 ++ .../otel/PostHogContextSpanProcessor.kt | 51 ++++ .../server/NoOpenTelemetryScopeFixture.kt | 21 ++ .../server/PostHogIncomingSpanContextTest.kt | 272 ++++++++++++++++++ 11 files changed, 430 insertions(+) create mode 100644 .changeset/fiery-maps-obey.md create mode 100644 posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt create mode 100644 posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt create mode 100644 posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt create mode 100644 posthog-server/src/test/java/com/posthog/server/NoOpenTelemetryScopeFixture.kt create mode 100644 posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt diff --git a/.changeset/fiery-maps-obey.md b/.changeset/fiery-maps-obey.md new file mode 100644 index 000000000..bda91152e --- /dev/null +++ b/.changeset/fiery-maps-obey.md @@ -0,0 +1,5 @@ +--- +"posthog-server": minor +--- + +Add an optional OpenTelemetry span processor and request-scope bridge for session and person context on application spans. diff --git a/README.md b/README.md index 7efc69445..6923fad2c 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,14 @@ SDK usage examples and code snippets live in the official documentation so they Do not exclude that artifact from your build. Without it, session replay, interaction autocapture and touch tracking stay off for the whole process. The SDK writes a warning to Logcat when this happens. +## Server span context + +The `posthog-server` request context enriches an active recording OpenTelemetry SDK span with available `sessionId` and `posthogDistinctId` attributes. Register `com.posthog.server.otel.PostHogContextSpanProcessor` on the application's existing provider to include those IDs on new child spans. + +OpenTelemetry SDK 1.32.0 or later remains an optional application dependency. Existing canonical attributes take precedence, including attributes with other value types. The bridge preserves the application's sampling and export setup. + +Use a fresh request scope and close it on the same thread. Work on another thread needs its own context scope. Automatic coroutine, Reactor, and framework propagation require application setup. See the [Java tracing guide](https://posthog.com/docs/distributed-tracing/installation/java) for setup. + ## Questions? ### [Check out our community page.](https://posthog.com/posts) diff --git a/posthog-server/api/posthog-server.api b/posthog-server/api/posthog-server.api index 2a8716456..d8edec626 100644 --- a/posthog-server/api/posthog-server.api +++ b/posthog-server/api/posthog-server.api @@ -509,3 +509,13 @@ public final class com/posthog/server/PostHogRequestContextData { public final fun getSessionId ()Ljava/lang/String; } +public final class com/posthog/server/otel/PostHogContextSpanProcessor : io/opentelemetry/sdk/trace/SpanProcessor { + public fun ()V + public fun forceFlush ()Lio/opentelemetry/sdk/common/CompletableResultCode; + public fun isEndRequired ()Z + public fun isStartRequired ()Z + public fun onEnd (Lio/opentelemetry/sdk/trace/ReadableSpan;)V + public fun onStart (Lio/opentelemetry/context/Context;Lio/opentelemetry/sdk/trace/ReadWriteSpan;)V + public fun shutdown ()Lio/opentelemetry/sdk/common/CompletableResultCode; +} + diff --git a/posthog-server/build.gradle.kts b/posthog-server/build.gradle.kts index 8b00fa208..244e580c5 100644 --- a/posthog-server/build.gradle.kts +++ b/posthog-server/build.gradle.kts @@ -106,7 +106,10 @@ dependencies { // compatibility signature("org.codehaus.mojo.signature:java18:${PosthogBuildConfig.Plugins.SIGNATURE_JAVA18}@signature") + compileOnly("io.opentelemetry:opentelemetry-sdk-trace:1.32.0") + // tests + testImplementation("io.opentelemetry:opentelemetry-sdk-trace:1.32.0") testImplementation("org.mockito.kotlin:mockito-kotlin:${PosthogBuildConfig.Dependencies.MOCKITO}") testImplementation("org.mockito:mockito-inline:${PosthogBuildConfig.Dependencies.MOCKITO_INLINE}") testImplementation("org.jetbrains.kotlin:kotlin-test-junit:${PosthogBuildConfig.Kotlin.KOTLIN}") diff --git a/posthog-server/gradle.lockfile b/posthog-server/gradle.lockfile index 5112a6c69..3008beb8a 100644 --- a/posthog-server/gradle.lockfile +++ b/posthog-server/gradle.lockfile @@ -19,6 +19,10 @@ com.squareup.okio:okio-jvm:3.11.0=jvmTestRuntimeClasspathForLint,runtimeClasspat com.squareup.okio:okio-jvm:3.6.0=compileClasspath,jvmTestCompileClasspathForLint,testCompileClasspath com.squareup.okio:okio:3.11.0=jvmTestRuntimeClasspathForLint,runtimeClasspath,testRuntimeClasspath com.squareup.okio:okio:3.6.0=compileClasspath,implementationDependenciesMetadata,jvmTestCompileClasspathForLint,testCompileClasspath,testImplementationDependenciesMetadata +io.opentelemetry:opentelemetry-api:1.32.0=compileClasspath,jvmTestCompileClasspathForLint,jvmTestRuntimeClasspathForLint,testCompileClasspath,testRuntimeClasspath +io.opentelemetry:opentelemetry-context:1.32.0=compileClasspath,jvmTestCompileClasspathForLint,jvmTestRuntimeClasspathForLint,testCompileClasspath,testRuntimeClasspath +io.opentelemetry:opentelemetry-sdk-common:1.32.0=compileClasspath,jvmTestCompileClasspathForLint,jvmTestRuntimeClasspathForLint,testCompileClasspath,testRuntimeClasspath +io.opentelemetry:opentelemetry-sdk-trace:1.32.0=compileClasspath,jvmTestCompileClasspathForLint,jvmTestRuntimeClasspathForLint,testCompileClasspath,testRuntimeClasspath it.unimi.dsi:fastutil-core:8.5.12=dokkaGfmPartialPlugin,dokkaGfmPlugin,dokkaHtmlPartialPlugin,dokkaHtmlPlugin,dokkaJavadocPartialPlugin,dokkaJavadocPlugin,dokkaJekyllPartialPlugin,dokkaJekyllPlugin jakarta.activation:jakarta.activation-api:1.2.1=dokkaGfmPartialRuntime,dokkaGfmRuntime,dokkaHtmlPartialRuntime,dokkaHtmlRuntime,dokkaJavadocPartialRuntime,dokkaJavadocRuntime,dokkaJekyllPartialRuntime,dokkaJekyllRuntime jakarta.xml.bind:jakarta.xml.bind-api:2.3.2=dokkaGfmPartialRuntime,dokkaGfmRuntime,dokkaHtmlPartialRuntime,dokkaHtmlRuntime,dokkaJavadocPartialRuntime,dokkaJavadocRuntime,dokkaJekyllPartialRuntime,dokkaJekyllRuntime diff --git a/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt b/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt index edcc5fd01..1ac1d1043 100644 --- a/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt +++ b/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt @@ -1,5 +1,6 @@ package com.posthog.server +import com.posthog.server.internal.enrichCurrentSpanSafely import java.util.UUID /** @@ -58,6 +59,7 @@ public class PostHogRequestContext private constructor() { val previous = currentContext.get() val base = if (fresh) null else previous currentContext.set(resolveContext(data, base)) + enrichCurrentSpanSafely() return Scope(previous) } diff --git a/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt b/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt new file mode 100644 index 000000000..ad5158fc3 --- /dev/null +++ b/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt @@ -0,0 +1,25 @@ +package com.posthog.server.internal + +import com.posthog.server.PostHogRequestContext + +private val openTelemetrySdkAvailable: Boolean by lazy { + try { + Class.forName("io.opentelemetry.sdk.trace.ReadWriteSpan", false, PostHogRequestContext::class.java.classLoader) + true + } catch (_: Exception) { + false + } catch (_: LinkageError) { + false + } +} + +internal fun enrichCurrentSpanSafely() { + if (!openTelemetrySdkAvailable) { + return + } + try { + enrichActiveOpenTelemetrySpan() + } catch (_: Exception) { + } catch (_: LinkageError) { + } +} diff --git a/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt b/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt new file mode 100644 index 000000000..7f98d6be7 --- /dev/null +++ b/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt @@ -0,0 +1,29 @@ +package com.posthog.server.internal + +import com.posthog.server.PostHogRequestContext +import io.opentelemetry.api.trace.Span +import io.opentelemetry.sdk.trace.ReadWriteSpan + +internal fun enrichActiveOpenTelemetrySpan() { + val span = Span.current() + if (span is ReadWriteSpan) { + enrichSpanWithRequestContext(span) + } +} + +internal fun enrichSpanWithRequestContext(span: ReadWriteSpan) { + try { + if (!span.isRecording) { + return + } + val context = PostHogRequestContext.current() ?: return + val existingNames = span.toSpanData().attributes.asMap().keys.map { it.key }.toSet() + listOf("sessionId" to context.sessionId, "posthogDistinctId" to context.distinctId).forEach { (key, value) -> + if (!value.isNullOrBlank() && key !in existingNames) { + span.setAttribute(key, value) + } + } + } catch (_: Exception) { + } catch (_: LinkageError) { + } +} diff --git a/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt b/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt new file mode 100644 index 000000000..bb9c30e04 --- /dev/null +++ b/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt @@ -0,0 +1,51 @@ +package com.posthog.server.otel + +import com.posthog.server.PostHogRequestContext +import com.posthog.server.internal.enrichSpanWithRequestContext +import io.opentelemetry.context.Context +import io.opentelemetry.sdk.common.CompletableResultCode +import io.opentelemetry.sdk.trace.ReadWriteSpan +import io.opentelemetry.sdk.trace.ReadableSpan +import io.opentelemetry.sdk.trace.SpanProcessor + +/** + * Adds available PostHog request IDs when an application span starts. + * + * Register this processor on the application's existing OpenTelemetry SDK provider. + * OpenTelemetry SDK 1.32.0 or later must be supplied by the application. This processor + * does not export spans or change providers, exporters, or sampling. + * + * The processor reads [PostHogRequestContext] on the current thread and snapshots + * `sessionId` and `posthogDistinctId`. Existing attributes with either name take + * precedence regardless of their type. Missing or blank IDs remain absent. + * + * Start and close a fresh PostHog context scope for each request. Work on another + * thread requires its own explicit context scope. Close each scope on the thread + * that opened it. This processor does not add + * coroutine, Reactor, or other framework context propagation. + */ +public class PostHogContextSpanProcessor public constructor() : SpanProcessor { + @Volatile private var stopped: Boolean = false + + override fun onStart( + parentContext: Context, + span: ReadWriteSpan, + ) { + if (!stopped) { + enrichSpanWithRequestContext(span) + } + } + + override fun isStartRequired(): Boolean = true + + override fun onEnd(span: ReadableSpan): Unit = Unit + + override fun isEndRequired(): Boolean = false + + override fun shutdown(): CompletableResultCode { + stopped = true + return CompletableResultCode.ofSuccess() + } + + override fun forceFlush(): CompletableResultCode = CompletableResultCode.ofSuccess() +} diff --git a/posthog-server/src/test/java/com/posthog/server/NoOpenTelemetryScopeFixture.kt b/posthog-server/src/test/java/com/posthog/server/NoOpenTelemetryScopeFixture.kt new file mode 100644 index 000000000..74ad75303 --- /dev/null +++ b/posthog-server/src/test/java/com/posthog/server/NoOpenTelemetryScopeFixture.kt @@ -0,0 +1,21 @@ +package com.posthog.server + +public object NoOpenTelemetryScopeFixture { + @JvmStatic + public fun main(args: Array) { + check(runCatching { Class.forName("io.opentelemetry.sdk.trace.ReadWriteSpan") }.isFailure) + val apiPresent = runCatching { Class.forName("io.opentelemetry.api.trace.Span") }.isSuccess + check(apiPresent == (args.firstOrNull() == "api-only")) + PostHogRequestContext.withContext(PostHogRequestContextData("example-person", "example-session"), fresh = true) { + check(PostHogRequestContext.current()?.sessionId == "example-session") + try { + PostHogRequestContext.withContext(PostHogRequestContextData(sessionId = "nested")) { + throw IllegalStateException("synthetic failure") + } + } catch (_: IllegalStateException) { + check(PostHogRequestContext.current()?.sessionId == "example-session") + } + } + check(PostHogRequestContext.current() == null) + } +} diff --git a/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt b/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt new file mode 100644 index 000000000..68b0ce4be --- /dev/null +++ b/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt @@ -0,0 +1,272 @@ +package com.posthog.server + +import com.posthog.server.otel.PostHogContextSpanProcessor +import io.opentelemetry.api.common.AttributeKey +import io.opentelemetry.api.common.Attributes +import io.opentelemetry.sdk.trace.ReadWriteSpan +import io.opentelemetry.sdk.trace.SdkTracerProvider +import io.opentelemetry.sdk.trace.SpanLimits +import io.opentelemetry.sdk.trace.samplers.Sampler +import java.io.File +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertFalse +import kotlin.test.assertNull +import kotlin.test.assertTrue + +internal class PostHogIncomingSpanContextTest { + @Test + fun `request scope enriches an existing application SDK span`() { + SdkTracerProvider.builder().build().use { provider -> + val span = provider.get("posthog-test").spanBuilder("incoming").startSpan() as ReadWriteSpan + try { + span.makeCurrent().use { + val context = + PostHogRequestContext.fromHeaders( + mapOf( + PostHogRequestContext.SESSION_ID_HEADER to "browser-session", + PostHogRequestContext.DISTINCT_ID_HEADER to "browser-person", + ), + ) + PostHogRequestContext.beginScope(context, fresh = true).use { + val attributes = span.toSpanData().attributes.asMap().mapKeys { it.key.key } + assertEquals("browser-session", attributes["sessionId"]) + assertEquals("browser-person", attributes["posthogDistinctId"]) + } + } + } finally { + span.end() + } + } + } + + @Test + fun `child spans keep their request snapshot after the scope ends`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + val tracer = provider.get("posthog-test") + val child = + PostHogRequestContext.withContext(PostHogRequestContextData("person-one", "session-one"), fresh = true) { + tracer.spanBuilder("child").startSpan() as ReadWriteSpan + } + PostHogRequestContext.withContext(PostHogRequestContextData("person-two", "session-two"), fresh = true) { + child.end() + val attrs = child.toSpanData().attributes.asMap().mapKeys { it.key.key } + assertEquals("person-one", attrs["posthogDistinctId"]) + assertEquals("session-one", attrs["sessionId"]) + } + val outside = tracer.spanBuilder("outside").startSpan() as ReadWriteSpan + outside.end() + assertTrue(outside.toSpanData().attributes.isEmpty) + } + } + + @Test + fun `canonical attributes of every supported type take precedence`() { + val cases = + listOf( + Attributes.builder().put("sessionId", "explicit").put("posthogDistinctId", "explicit").build(), + Attributes.builder().put("sessionId", true).put("posthogDistinctId", 42L).build(), + Attributes.builder() + .put(AttributeKey.stringArrayKey("sessionId"), listOf("explicit")) + .put("posthogDistinctId", false).build(), + ) + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + for (explicit in cases) { + val span = + provider.get("posthog-test").spanBuilder("explicit") + .setAllAttributes(explicit).startSpan() as ReadWriteSpan + try { + span.makeCurrent().use { + PostHogRequestContext.beginScope( + PostHogRequestContextData("header-person", "header-session"), + fresh = true, + ).use { + assertEquals(explicit, span.toSpanData().attributes) + } + } + } finally { + span.end() + } + } + } + } + + @Test + fun `enrichment does not evict existing attributes at the SDK limit`() { + SdkTracerProvider.builder() + .setSpanLimits(SpanLimits.builder().setMaxNumberOfAttributes(1).build()) + .addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + PostHogRequestContext.withContext(PostHogRequestContextData("person", "session"), fresh = true) { + val span = + provider.get("posthog-test").spanBuilder("full") + .setAttribute("keep", "application").startSpan() as ReadWriteSpan + span.makeCurrent().use { + PostHogRequestContext.beginScope(PostHogRequestContextData("other-person", "other-session")).use { + assertEquals(Attributes.of(AttributeKey.stringKey("keep"), "application"), span.toSpanData().attributes) + } + } + span.end() + } + } + } + + @Test + fun `missing and blank IDs remain absent`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + for (value in listOf(null, "", " ")) { + PostHogRequestContext.withContext(PostHogRequestContextData(value, value), fresh = true) { + val span = provider.get("posthog-test").spanBuilder("missing").startSpan() as ReadWriteSpan + span.end() + assertTrue(span.toSpanData().attributes.isEmpty) + } + } + } + } + + @Test + fun `failure closes inner scope and the next span sees its parent context`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + val tracer = provider.get("posthog-test") + PostHogRequestContext.withContext(PostHogRequestContextData("outer-person", "outer-session"), fresh = true) { + assertFailsWith { + PostHogRequestContext.withContext(PostHogRequestContextData("inner-person", "inner-session")) { + val inner = tracer.spanBuilder("inner").startSpan() as ReadWriteSpan + inner.end() + assertEquals("inner-session", inner.toSpanData().attributes.get(AttributeKey.stringKey("sessionId"))) + throw IllegalStateException("synthetic failure") + } + } + val restored = tracer.spanBuilder("restored").startSpan() as ReadWriteSpan + restored.end() + assertEquals("outer-session", restored.toSpanData().attributes.get(AttributeKey.stringKey("sessionId"))) + } + assertNull(PostHogRequestContext.current()) + } + } + + @Test + fun `concurrent request threads and reused workers stay isolated`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + val executor = Executors.newFixedThreadPool(2) + try { + val ready = CountDownLatch(2) + val futures = + listOf("one", "two").map { id -> + executor.submit { + val attributes = + PostHogRequestContext.withContext( + PostHogRequestContextData("person-$id", "session-$id"), + fresh = true, + ) { + ready.countDown() + check(ready.await(5, TimeUnit.SECONDS)) + val span = provider.get("posthog-test").spanBuilder("request-$id").startSpan() as ReadWriteSpan + span.end() + span.toSpanData().attributes + } + check(PostHogRequestContext.current() == null) + attributes + } + } + futures.forEachIndexed { index, future -> + val id = listOf("one", "two")[index] + val attrs = future.get(5, TimeUnit.SECONDS) + assertEquals("session-$id", attrs.get(AttributeKey.stringKey("sessionId"))) + assertEquals("person-$id", attrs.get(AttributeKey.stringKey("posthogDistinctId"))) + } + val outside = + executor.submit { + val span = provider.get("posthog-test").spanBuilder("reused-worker").startSpan() as ReadWriteSpan + span.end() + span.toSpanData().attributes.isEmpty + } + assertTrue(outside.get(5, TimeUnit.SECONDS)) + } finally { + executor.shutdownNow() + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)) + } + } + } + + @Test + fun `detached work receives no implicit context after request completion`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + val executor = Executors.newSingleThreadExecutor() + val release = CountDownLatch(1) + try { + val work = + PostHogRequestContext.withContext(PostHogRequestContextData("person", "session"), fresh = true) { + executor.submit { + check(release.await(5, TimeUnit.SECONDS)) + val span = provider.get("posthog-test").spanBuilder("detached").startSpan() as ReadWriteSpan + span.end() + span.toSpanData().attributes + } + } + release.countDown() + assertTrue(work.get(5, TimeUnit.SECONDS).isEmpty) + } finally { + release.countDown() + executor.shutdownNow() + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)) + } + } + } + + @Test + fun `processor shutdown stops future child enrichment`() { + val processor = PostHogContextSpanProcessor() + SdkTracerProvider.builder().addSpanProcessor(processor).build().use { provider -> + assertTrue(processor.shutdown().isSuccess) + PostHogRequestContext.withContext(PostHogRequestContextData("person", "session"), fresh = true) { + val span = provider.get("posthog-test").spanBuilder("after-shutdown").startSpan() as ReadWriteSpan + span.end() + assertTrue(span.toSpanData().attributes.isEmpty) + } + } + } + + @Test + fun `application sampling remains unchanged`() { + SdkTracerProvider.builder().setSampler(Sampler.alwaysOff()) + .addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + val span = provider.get("posthog-test").spanBuilder("sampled-out").startSpan() + span.makeCurrent().use { + PostHogRequestContext.beginScope(PostHogRequestContextData("person", "session"), fresh = true).use { + assertFalse(span.isRecording) + } + } + span.end() + } + } + + @Test + fun `request scopes work with no tracing SDK or with API only`() { + for (mode in listOf("none", "api-only")) { + val classpath = + System.getProperty("java.class.path").split(File.pathSeparator) + .filterNot { path -> + val name = File(path).name + if (mode == "none") name.startsWith("opentelemetry-") else name.startsWith("opentelemetry-sdk") + }.joinToString(File.pathSeparator) + val process = + ProcessBuilder( + File(System.getProperty("java.home"), "bin/java").absolutePath, + "-cp", + classpath, + NoOpenTelemetryScopeFixture::class.java.name, + mode, + ).redirectErrorStream(true).start() + try { + assertTrue(process.waitFor(30, TimeUnit.SECONDS), "Fixture JVM did not exit: $mode") + assertEquals(0, process.exitValue(), process.inputStream.bufferedReader().readText()) + } finally { + process.destroyForcibly() + } + } + } +} From 1e066d6be3488e24115acfef57cddde81c26a9b4 Mon Sep 17 00:00:00 2001 From: pauldambra Date: Sun, 18 Oct 2026 11:13:11 +0100 Subject: [PATCH 2/3] fix(server): bind span enrichment to its client Generated-By: PostHog Desktop Task-Id: bfd18848-e498-4352-a58a-b9ba85de4198 --- README.md | 12 +++ posthog-server/api/posthog-server.api | 8 +- .../posthog/server/PostHogRequestContext.kt | 72 ++++++++++++++++- .../server/internal/ContextSpanBridge.kt | 24 +++++- .../internal/OpenTelemetrySpanContext.kt | 12 ++- .../otel/PostHogContextSpanProcessor.kt | 18 +++-- .../server/PostHogIncomingSpanContextTest.kt | 78 ++++++++++++++----- .../posthog/server/PostHogSpanPrivacyTest.kt | 75 ++++++++++++++++++ 8 files changed, 263 insertions(+), 36 deletions(-) create mode 100644 posthog-server/src/test/java/com/posthog/server/PostHogSpanPrivacyTest.kt diff --git a/README.md b/README.md index 6923fad2c..5ab3c556e 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,18 @@ Do not exclude that artifact from your build. Without it, session replay, intera The `posthog-server` request context enriches an active recording OpenTelemetry SDK span with available `sessionId` and `posthogDistinctId` attributes. Register `com.posthog.server.otel.PostHogContextSpanProcessor` on the application's existing provider to include those IDs on new child spans. +Pass the same PostHog client to the processor and to each request scope. Only that client's processor adds IDs, and a disabled or opted-out client adds none. + +```kotlin +val tracerProvider = SdkTracerProvider.builder() + .addSpanProcessor(PostHogContextSpanProcessor(posthog)) + .build() + +PostHogRequestContext.withContext(posthog, PostHogRequestContext.fromHeaders(headers), fresh = true) { + handleRequest() +} +``` + OpenTelemetry SDK 1.32.0 or later remains an optional application dependency. Existing canonical attributes take precedence, including attributes with other value types. The bridge preserves the application's sampling and export setup. Use a fresh request scope and close it on the same thread. Work on another thread needs its own context scope. Automatic coroutine, Reactor, and framework propagation require application setup. See the [Java tracing guide](https://posthog.com/docs/distributed-tracing/installation/java) for setup. diff --git a/posthog-server/api/posthog-server.api b/posthog-server/api/posthog-server.api index d8edec626..7409d5989 100644 --- a/posthog-server/api/posthog-server.api +++ b/posthog-server/api/posthog-server.api @@ -476,6 +476,8 @@ public final class com/posthog/server/PostHogRequestContext { public static final field Companion Lcom/posthog/server/PostHogRequestContext$Companion; public static final field DISTINCT_ID_HEADER Ljava/lang/String; public static final field SESSION_ID_HEADER Ljava/lang/String; + public static final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; + public static final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public static final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; public static final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public static final fun current ()Lcom/posthog/server/PostHogRequestContextData; @@ -485,6 +487,8 @@ public final class com/posthog/server/PostHogRequestContext { } public final class com/posthog/server/PostHogRequestContext$Companion { + public final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; + public final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; public final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public final fun current ()Lcom/posthog/server/PostHogRequestContextData; @@ -492,6 +496,8 @@ public final class com/posthog/server/PostHogRequestContext$Companion { public final fun fromHeaders (Ljava/util/Map;Z)Lcom/posthog/server/PostHogRequestContextData; public final fun fromHeaders (Ljava/util/Map;ZLjava/util/Map;)Lcom/posthog/server/PostHogRequestContextData; public static synthetic fun fromHeaders$default (Lcom/posthog/server/PostHogRequestContext$Companion;Ljava/util/Map;ZLjava/util/Map;ILjava/lang/Object;)Lcom/posthog/server/PostHogRequestContextData; + public final fun withContext (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object; + public final fun withContext (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;ZLkotlin/jvm/functions/Function0;)Ljava/lang/Object; public final fun withContext (Lcom/posthog/server/PostHogRequestContextData;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object; public final fun withContext (Lcom/posthog/server/PostHogRequestContextData;ZLkotlin/jvm/functions/Function0;)Ljava/lang/Object; } @@ -510,7 +516,7 @@ public final class com/posthog/server/PostHogRequestContextData { } public final class com/posthog/server/otel/PostHogContextSpanProcessor : io/opentelemetry/sdk/trace/SpanProcessor { - public fun ()V + public fun (Lcom/posthog/PostHogStateless;)V public fun forceFlush ()Lio/opentelemetry/sdk/common/CompletableResultCode; public fun isEndRequired ()Z public fun isStartRequired ()Z diff --git a/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt b/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt index 1ac1d1043..c2f5966f1 100644 --- a/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt +++ b/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt @@ -1,5 +1,6 @@ package com.posthog.server +import com.posthog.PostHogStateless import com.posthog.server.internal.enrichCurrentSpanSafely import java.util.UUID @@ -24,6 +25,9 @@ public class PostHogRequestContext private constructor() { private const val MAX_HEADER_VALUE_LENGTH = 1000 private val currentContext = ThreadLocal() + private val currentOwner = ThreadLocal() + + internal fun owner(): PostHogStateless? = currentOwner.get() /** * Returns the context active on the current thread, or null when no scope is active. @@ -55,12 +59,43 @@ public class PostHogRequestContext private constructor() { public fun beginScope( data: PostHogRequestContextData, fresh: Boolean, + ): Scope { + return beginOwnedScope(if (fresh) null else currentOwner.get(), data, fresh) + } + + /** + * Starts a scope owned by [owner] for capture and optional span enrichment. + * A different owner does not inherit the previous client's IDs or properties. + */ + @JvmStatic + public fun beginScope( + owner: PostHogStateless, + data: PostHogRequestContextData, + ): Scope = beginScope(owner, data, fresh = false) + + /** + * Starts an owner-bound scope. Only this client's registered processor may enrich spans. + * Opted-out clients add no attributes. [fresh] ignores any previous context. + */ + @JvmStatic + public fun beginScope( + owner: PostHogStateless, + data: PostHogRequestContextData, + fresh: Boolean, + ): Scope = beginOwnedScope(owner, data, fresh) + + private fun beginOwnedScope( + owner: PostHogStateless?, + data: PostHogRequestContextData, + fresh: Boolean, ): Scope { val previous = currentContext.get() - val base = if (fresh) null else previous + val previousOwner = currentOwner.get() + val base = if (fresh || previousOwner !== owner) null else previous currentContext.set(resolveContext(data, base)) + currentOwner.set(owner) enrichCurrentSpanSafely() - return Scope(previous) + return Scope(previous, previousOwner) } /** @@ -93,6 +128,29 @@ public class PostHogRequestContext private constructor() { } } + /** + * Runs [block] in an owner-bound scope and restores the previous scope afterwards. + */ + public fun withContext( + owner: PostHogStateless, + data: PostHogRequestContextData, + block: () -> T, + ): T = withContext(owner, data, fresh = false, block) + + /** + * Runs [block] in an owner-bound scope, optionally ignoring previous context with [fresh]. + */ + public fun withContext( + owner: PostHogStateless, + data: PostHogRequestContextData, + fresh: Boolean, + block: () -> T, + ): T { + beginScope(owner, data, fresh).use { + return block() + } + } + /** * Extracts PostHog tracing headers into context data. * @@ -246,7 +304,10 @@ public class PostHogRequestContext private constructor() { /** * Active request context scope. Close it to restore the previous context. */ - public class Scope internal constructor(private val previous: PostHogRequestContextData?) : AutoCloseable { + public class Scope internal constructor( + private val previous: PostHogRequestContextData?, + private val previousOwner: PostHogStateless?, + ) : AutoCloseable { private var closed: Boolean = false override fun close() { @@ -259,6 +320,11 @@ public class PostHogRequestContext private constructor() { } else { currentContext.set(previous) } + if (previousOwner == null) { + currentOwner.remove() + } else { + currentOwner.set(previousOwner) + } closed = true } } diff --git a/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt b/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt index ad5158fc3..5012e8589 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt @@ -1,6 +1,24 @@ package com.posthog.server.internal +import com.posthog.PostHogStateless import com.posthog.server.PostHogRequestContext +import java.lang.ref.WeakReference +import java.util.WeakHashMap + +private val registeredProcessors = WeakHashMap>() + +internal fun registerSpanContextOwner(owner: PostHogStateless): Any = + Any().also { token -> + synchronized(registeredProcessors) { + registeredProcessors[token] = WeakReference(owner) + } + } + +internal fun unregisterSpanContextOwner(token: Any) { + synchronized(registeredProcessors) { + registeredProcessors.remove(token) + } +} private val openTelemetrySdkAvailable: Boolean by lazy { try { @@ -14,11 +32,13 @@ private val openTelemetrySdkAvailable: Boolean by lazy { } internal fun enrichCurrentSpanSafely() { - if (!openTelemetrySdkAvailable) { + val owner = PostHogRequestContext.owner() ?: return + val registered = synchronized(registeredProcessors) { registeredProcessors.values.any { it.get() === owner } } + if (!registered || !openTelemetrySdkAvailable) { return } try { - enrichActiveOpenTelemetrySpan() + enrichActiveOpenTelemetrySpan(owner) } catch (_: Exception) { } catch (_: LinkageError) { } diff --git a/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt b/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt index 7f98d6be7..c98e51cd4 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt @@ -1,19 +1,23 @@ package com.posthog.server.internal +import com.posthog.PostHogStateless import com.posthog.server.PostHogRequestContext import io.opentelemetry.api.trace.Span import io.opentelemetry.sdk.trace.ReadWriteSpan -internal fun enrichActiveOpenTelemetrySpan() { +internal fun enrichActiveOpenTelemetrySpan(owner: PostHogStateless) { val span = Span.current() if (span is ReadWriteSpan) { - enrichSpanWithRequestContext(span) + enrichSpanWithRequestContext(owner, span) } } -internal fun enrichSpanWithRequestContext(span: ReadWriteSpan) { +internal fun enrichSpanWithRequestContext( + owner: PostHogStateless, + span: ReadWriteSpan, +) { try { - if (!span.isRecording) { + if (owner.isOptOut() || PostHogRequestContext.owner() !== owner || !span.isRecording) { return } val context = PostHogRequestContext.current() ?: return diff --git a/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt b/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt index bb9c30e04..c6e3ee60f 100644 --- a/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt +++ b/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt @@ -1,12 +1,16 @@ package com.posthog.server.otel +import com.posthog.PostHogStateless import com.posthog.server.PostHogRequestContext import com.posthog.server.internal.enrichSpanWithRequestContext +import com.posthog.server.internal.registerSpanContextOwner +import com.posthog.server.internal.unregisterSpanContextOwner import io.opentelemetry.context.Context import io.opentelemetry.sdk.common.CompletableResultCode import io.opentelemetry.sdk.trace.ReadWriteSpan import io.opentelemetry.sdk.trace.ReadableSpan import io.opentelemetry.sdk.trace.SpanProcessor +import java.util.concurrent.atomic.AtomicBoolean /** * Adds available PostHog request IDs when an application span starts. @@ -19,20 +23,22 @@ import io.opentelemetry.sdk.trace.SpanProcessor * `sessionId` and `posthogDistinctId`. Existing attributes with either name take * precedence regardless of their type. Missing or blank IDs remain absent. * + * Bind each request scope to the same [owner]. Opted-out clients add no attributes. * Start and close a fresh PostHog context scope for each request. Work on another * thread requires its own explicit context scope. Close each scope on the thread * that opened it. This processor does not add * coroutine, Reactor, or other framework context propagation. */ -public class PostHogContextSpanProcessor public constructor() : SpanProcessor { - @Volatile private var stopped: Boolean = false +public class PostHogContextSpanProcessor public constructor(private val owner: PostHogStateless) : SpanProcessor { + private val stopped = AtomicBoolean(false) + private val registration = registerSpanContextOwner(owner) override fun onStart( parentContext: Context, span: ReadWriteSpan, ) { - if (!stopped) { - enrichSpanWithRequestContext(span) + if (!stopped.get()) { + enrichSpanWithRequestContext(owner, span) } } @@ -43,7 +49,9 @@ public class PostHogContextSpanProcessor public constructor() : SpanProcessor { override fun isEndRequired(): Boolean = false override fun shutdown(): CompletableResultCode { - stopped = true + if (stopped.compareAndSet(false, true)) { + unregisterSpanContextOwner(registration) + } return CompletableResultCode.ofSuccess() } diff --git a/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt b/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt index 68b0ce4be..a521acef1 100644 --- a/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt +++ b/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt @@ -11,6 +11,7 @@ import java.io.File import java.util.concurrent.CountDownLatch import java.util.concurrent.Executors import java.util.concurrent.TimeUnit +import kotlin.test.AfterTest import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertFailsWith @@ -19,9 +20,16 @@ import kotlin.test.assertNull import kotlin.test.assertTrue internal class PostHogIncomingSpanContextTest { + private val owner = PostHog().apply { setup(PostHogConfig("phc_test", preloadFeatureFlags = false)) } + + @AfterTest + fun closeClient() { + owner.close() + } + @Test fun `request scope enriches an existing application SDK span`() { - SdkTracerProvider.builder().build().use { provider -> + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> val span = provider.get("posthog-test").spanBuilder("incoming").startSpan() as ReadWriteSpan try { span.makeCurrent().use { @@ -32,7 +40,7 @@ internal class PostHogIncomingSpanContextTest { PostHogRequestContext.DISTINCT_ID_HEADER to "browser-person", ), ) - PostHogRequestContext.beginScope(context, fresh = true).use { + PostHogRequestContext.beginScope(owner, context, fresh = true).use { val attributes = span.toSpanData().attributes.asMap().mapKeys { it.key.key } assertEquals("browser-session", attributes["sessionId"]) assertEquals("browser-person", attributes["posthogDistinctId"]) @@ -46,13 +54,13 @@ internal class PostHogIncomingSpanContextTest { @Test fun `child spans keep their request snapshot after the scope ends`() { - SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> val tracer = provider.get("posthog-test") val child = - PostHogRequestContext.withContext(PostHogRequestContextData("person-one", "session-one"), fresh = true) { + PostHogRequestContext.withContext(owner, PostHogRequestContextData("person-one", "session-one"), fresh = true) { tracer.spanBuilder("child").startSpan() as ReadWriteSpan } - PostHogRequestContext.withContext(PostHogRequestContextData("person-two", "session-two"), fresh = true) { + PostHogRequestContext.withContext(owner, PostHogRequestContextData("person-two", "session-two"), fresh = true) { child.end() val attrs = child.toSpanData().attributes.asMap().mapKeys { it.key.key } assertEquals("person-one", attrs["posthogDistinctId"]) @@ -74,7 +82,7 @@ internal class PostHogIncomingSpanContextTest { .put(AttributeKey.stringArrayKey("sessionId"), listOf("explicit")) .put("posthogDistinctId", false).build(), ) - SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> for (explicit in cases) { val span = provider.get("posthog-test").spanBuilder("explicit") @@ -82,6 +90,7 @@ internal class PostHogIncomingSpanContextTest { try { span.makeCurrent().use { PostHogRequestContext.beginScope( + owner, PostHogRequestContextData("header-person", "header-session"), fresh = true, ).use { @@ -99,13 +108,13 @@ internal class PostHogIncomingSpanContextTest { fun `enrichment does not evict existing attributes at the SDK limit`() { SdkTracerProvider.builder() .setSpanLimits(SpanLimits.builder().setMaxNumberOfAttributes(1).build()) - .addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> - PostHogRequestContext.withContext(PostHogRequestContextData("person", "session"), fresh = true) { + .addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> + PostHogRequestContext.withContext(owner, PostHogRequestContextData("person", "session"), fresh = true) { val span = provider.get("posthog-test").spanBuilder("full") .setAttribute("keep", "application").startSpan() as ReadWriteSpan span.makeCurrent().use { - PostHogRequestContext.beginScope(PostHogRequestContextData("other-person", "other-session")).use { + PostHogRequestContext.beginScope(owner, PostHogRequestContextData("other-person", "other-session")).use { assertEquals(Attributes.of(AttributeKey.stringKey("keep"), "application"), span.toSpanData().attributes) } } @@ -116,9 +125,9 @@ internal class PostHogIncomingSpanContextTest { @Test fun `missing and blank IDs remain absent`() { - SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> for (value in listOf(null, "", " ")) { - PostHogRequestContext.withContext(PostHogRequestContextData(value, value), fresh = true) { + PostHogRequestContext.withContext(owner, PostHogRequestContextData(value, value), fresh = true) { val span = provider.get("posthog-test").spanBuilder("missing").startSpan() as ReadWriteSpan span.end() assertTrue(span.toSpanData().attributes.isEmpty) @@ -129,11 +138,11 @@ internal class PostHogIncomingSpanContextTest { @Test fun `failure closes inner scope and the next span sees its parent context`() { - SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> val tracer = provider.get("posthog-test") - PostHogRequestContext.withContext(PostHogRequestContextData("outer-person", "outer-session"), fresh = true) { + PostHogRequestContext.withContext(owner, PostHogRequestContextData("outer-person", "outer-session"), fresh = true) { assertFailsWith { - PostHogRequestContext.withContext(PostHogRequestContextData("inner-person", "inner-session")) { + PostHogRequestContext.withContext(owner, PostHogRequestContextData("inner-person", "inner-session")) { val inner = tracer.spanBuilder("inner").startSpan() as ReadWriteSpan inner.end() assertEquals("inner-session", inner.toSpanData().attributes.get(AttributeKey.stringKey("sessionId"))) @@ -150,7 +159,7 @@ internal class PostHogIncomingSpanContextTest { @Test fun `concurrent request threads and reused workers stay isolated`() { - SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> val executor = Executors.newFixedThreadPool(2) try { val ready = CountDownLatch(2) @@ -159,6 +168,7 @@ internal class PostHogIncomingSpanContextTest { executor.submit { val attributes = PostHogRequestContext.withContext( + owner, PostHogRequestContextData("person-$id", "session-$id"), fresh = true, ) { @@ -194,12 +204,12 @@ internal class PostHogIncomingSpanContextTest { @Test fun `detached work receives no implicit context after request completion`() { - SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> val executor = Executors.newSingleThreadExecutor() val release = CountDownLatch(1) try { val work = - PostHogRequestContext.withContext(PostHogRequestContextData("person", "session"), fresh = true) { + PostHogRequestContext.withContext(owner, PostHogRequestContextData("person", "session"), fresh = true) { executor.submit { check(release.await(5, TimeUnit.SECONDS)) val span = provider.get("posthog-test").spanBuilder("detached").startSpan() as ReadWriteSpan @@ -219,10 +229,18 @@ internal class PostHogIncomingSpanContextTest { @Test fun `processor shutdown stops future child enrichment`() { - val processor = PostHogContextSpanProcessor() + val processor = PostHogContextSpanProcessor(owner) SdkTracerProvider.builder().addSpanProcessor(processor).build().use { provider -> + val incoming = provider.get("posthog-test").spanBuilder("incoming").startSpan() as ReadWriteSpan assertTrue(processor.shutdown().isSuccess) - PostHogRequestContext.withContext(PostHogRequestContextData("person", "session"), fresh = true) { + assertTrue(processor.shutdown().isSuccess) + incoming.makeCurrent().use { + PostHogRequestContext.beginScope(owner, PostHogRequestContextData("person", "session"), fresh = true).use { + assertTrue(incoming.toSpanData().attributes.isEmpty) + } + } + incoming.end() + PostHogRequestContext.withContext(owner, PostHogRequestContextData("person", "session"), fresh = true) { val span = provider.get("posthog-test").spanBuilder("after-shutdown").startSpan() as ReadWriteSpan span.end() assertTrue(span.toSpanData().attributes.isEmpty) @@ -230,13 +248,31 @@ internal class PostHogIncomingSpanContextTest { } } + @Test + fun `ownerless scopes keep capture context without enriching spans`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> + val tracer = provider.get("ownerless") + val incoming = tracer.spanBuilder("incoming").startSpan() as ReadWriteSpan + incoming.makeCurrent().use { + PostHogRequestContext.withContext(PostHogRequestContextData("person", "session"), fresh = true) { + assertEquals("person", PostHogRequestContext.current()?.distinctId) + val child = tracer.spanBuilder("child").startSpan() as ReadWriteSpan + child.end() + assertTrue(child.toSpanData().attributes.isEmpty) + assertTrue(incoming.toSpanData().attributes.isEmpty) + } + } + incoming.end() + } + } + @Test fun `application sampling remains unchanged`() { SdkTracerProvider.builder().setSampler(Sampler.alwaysOff()) - .addSpanProcessor(PostHogContextSpanProcessor()).build().use { provider -> + .addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> val span = provider.get("posthog-test").spanBuilder("sampled-out").startSpan() span.makeCurrent().use { - PostHogRequestContext.beginScope(PostHogRequestContextData("person", "session"), fresh = true).use { + PostHogRequestContext.beginScope(owner, PostHogRequestContextData("person", "session"), fresh = true).use { assertFalse(span.isRecording) } } diff --git a/posthog-server/src/test/java/com/posthog/server/PostHogSpanPrivacyTest.kt b/posthog-server/src/test/java/com/posthog/server/PostHogSpanPrivacyTest.kt new file mode 100644 index 000000000..312cc89de --- /dev/null +++ b/posthog-server/src/test/java/com/posthog/server/PostHogSpanPrivacyTest.kt @@ -0,0 +1,75 @@ +package com.posthog.server + +import com.posthog.server.otel.PostHogContextSpanProcessor +import io.opentelemetry.api.common.AttributeKey +import io.opentelemetry.sdk.trace.ReadWriteSpan +import io.opentelemetry.sdk.trace.SdkTracerProvider +import org.junit.Test +import org.junit.runner.RunWith +import org.junit.runners.Parameterized +import kotlin.test.assertEquals +import kotlin.test.assertNull + +@RunWith(Parameterized::class) +internal class PostHogSpanPrivacyTest(private val state: String) { + companion object { + @JvmStatic + @Parameterized.Parameters(name = "{0}") + fun states(): List> = listOf("enabled", "before", "during", "uninitialized", "closed").map { arrayOf(it) } + } + + @Test + fun `scope owner controls request and child enrichment independently`() { + val owner = PostHog() + val other = PostHog() + val config = PostHogConfig("phc_test", preloadFeatureFlags = false) + try { + if (state != "uninitialized") owner.setup(config) + other.setup(PostHogConfig("phc_other", preloadFeatureFlags = false)) + if (state == "before") owner.optOut() + if (state == "closed") owner.close() + SdkTracerProvider.builder() + .addSpanProcessor(PostHogContextSpanProcessor(owner)) + .addSpanProcessor(PostHogContextSpanProcessor(other)).build().use { provider -> + val tracer = provider.get("privacy") + val incoming = tracer.spanBuilder("incoming").setAttribute("sessionId", false).startSpan() as ReadWriteSpan + incoming.makeCurrent().use { + PostHogRequestContext.withContext(owner, PostHogRequestContextData("owner-person", "owner-session"), fresh = true) { + val initial = if (state == "enabled" || state == "during") "owner-person" else null + assertEquals(initial, incoming.toSpanData().attributes.get(AttributeKey.stringKey("posthogDistinctId"))) + assertEquals(false, incoming.toSpanData().attributes.get(AttributeKey.booleanKey("sessionId"))) + if (state == "during") owner.optOut() + val child = tracer.spanBuilder("child").setAttribute("sessionId", false).startSpan() as ReadWriteSpan + child.makeCurrent().use { + PostHogRequestContext.beginScope(PostHogRequestContextData("later-person", "later-session")).use { + val expected = if (state == "enabled") "owner-person" else null + assertEquals(expected, child.toSpanData().attributes.get(AttributeKey.stringKey("posthogDistinctId"))) + assertEquals(false, child.toSpanData().attributes.get(AttributeKey.booleanKey("sessionId"))) + } + } + child.end() + PostHogRequestContext.withContext(other, PostHogRequestContextData(distinctId = "other-person")) { + val separate = tracer.spanBuilder("other-owner").startSpan() as ReadWriteSpan + separate.end() + assertEquals( + "other-person", + separate.toSpanData().attributes.get(AttributeKey.stringKey("posthogDistinctId")), + ) + assertNull(separate.toSpanData().attributes.get(AttributeKey.stringKey("sessionId"))) + } + assertEquals("owner-session", PostHogRequestContext.current()?.sessionId) + val restored = tracer.spanBuilder("restored-owner").startSpan() as ReadWriteSpan + restored.end() + val expected = if (state == "enabled") "owner-session" else null + assertEquals(expected, restored.toSpanData().attributes.get(AttributeKey.stringKey("sessionId"))) + } + } + incoming.end() + assertNull(PostHogRequestContext.current()) + } + } finally { + owner.close() + other.close() + } + } +} From 1048c0b436d633d4270c9941e1c4fd57712e8769 Mon Sep 17 00:00:00 2001 From: pauldambra Date: Sun, 18 Oct 2026 11:20:33 +0100 Subject: [PATCH 3/3] fix(server): accept the public client type for span enrichment Generated-By: PostHog Desktop Task-Id: bfd18848-e498-4352-a58a-b9ba85de4198 --- .changeset/fiery-maps-obey.md | 2 +- README.md | 10 ++++++++-- posthog-server/api/posthog-server.api | 14 +++++++------- .../com/posthog/server/PostHogRequestContext.kt | 17 ++++++++--------- .../server/internal/ContextSpanBridge.kt | 6 +++--- .../server/internal/OpenTelemetrySpanContext.kt | 15 +++++++++++---- .../server/otel/PostHogContextSpanProcessor.kt | 16 +++++++++++----- .../server/PostHogIncomingSpanContextTest.kt | 2 +- 8 files changed, 50 insertions(+), 32 deletions(-) diff --git a/.changeset/fiery-maps-obey.md b/.changeset/fiery-maps-obey.md index bda91152e..b8f576a99 100644 --- a/.changeset/fiery-maps-obey.md +++ b/.changeset/fiery-maps-obey.md @@ -2,4 +2,4 @@ "posthog-server": minor --- -Add an optional OpenTelemetry span processor and request-scope bridge for session and person context on application spans. +Add an optional OpenTelemetry span processor and request-scope bridge for session and person context on application spans. The processor is bound to one PostHog client. Only scopes owned by that client add IDs, and a disabled or opted-out client adds none. A nested scope without a client keeps the outer scope's owner. If the application replaces its client, build a new processor. diff --git a/README.md b/README.md index 5ab3c556e..7234646da 100644 --- a/README.md +++ b/README.md @@ -29,11 +29,17 @@ Do not exclude that artifact from your build. Without it, session replay, intera ## Server span context -The `posthog-server` request context enriches an active recording OpenTelemetry SDK span with available `sessionId` and `posthogDistinctId` attributes. Register `com.posthog.server.otel.PostHogContextSpanProcessor` on the application's existing provider to include those IDs on new child spans. +`posthog-server` can add `sessionId` and `posthogDistinctId` attributes to recording OpenTelemetry SDK spans. To turn this on, register `com.posthog.server.otel.PostHogContextSpanProcessor` on the application's existing provider. Then open each request scope with the same PostHog client. -Pass the same PostHog client to the processor and to each request scope. Only that client's processor adds IDs, and a disabled or opted-out client adds none. +Spans get IDs only from a scope owned by that client. A disabled or opted-out client adds no IDs. An outermost scope opened without a client adds no span IDs. A nested scope opened without a client keeps the owner of the outer scope. + +When a client-owned scope opens, it also adds IDs to the current active span. This happens when that client has a registered processor on any provider. + +The processor holds its client weakly. If the application closes its client and makes a new one, build a new processor and provider for the new client. ```kotlin +val posthog = PostHog.with(config) + val tracerProvider = SdkTracerProvider.builder() .addSpanProcessor(PostHogContextSpanProcessor(posthog)) .build() diff --git a/posthog-server/api/posthog-server.api b/posthog-server/api/posthog-server.api index 7409d5989..51421e12d 100644 --- a/posthog-server/api/posthog-server.api +++ b/posthog-server/api/posthog-server.api @@ -476,8 +476,8 @@ public final class com/posthog/server/PostHogRequestContext { public static final field Companion Lcom/posthog/server/PostHogRequestContext$Companion; public static final field DISTINCT_ID_HEADER Ljava/lang/String; public static final field SESSION_ID_HEADER Ljava/lang/String; - public static final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; - public static final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; + public static final fun beginScope (Lcom/posthog/server/PostHogInterface;Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; + public static final fun beginScope (Lcom/posthog/server/PostHogInterface;Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public static final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; public static final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public static final fun current ()Lcom/posthog/server/PostHogRequestContextData; @@ -487,8 +487,8 @@ public final class com/posthog/server/PostHogRequestContext { } public final class com/posthog/server/PostHogRequestContext$Companion { - public final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; - public final fun beginScope (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; + public final fun beginScope (Lcom/posthog/server/PostHogInterface;Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; + public final fun beginScope (Lcom/posthog/server/PostHogInterface;Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;)Lcom/posthog/server/PostHogRequestContext$Scope; public final fun beginScope (Lcom/posthog/server/PostHogRequestContextData;Z)Lcom/posthog/server/PostHogRequestContext$Scope; public final fun current ()Lcom/posthog/server/PostHogRequestContextData; @@ -496,8 +496,8 @@ public final class com/posthog/server/PostHogRequestContext$Companion { public final fun fromHeaders (Ljava/util/Map;Z)Lcom/posthog/server/PostHogRequestContextData; public final fun fromHeaders (Ljava/util/Map;ZLjava/util/Map;)Lcom/posthog/server/PostHogRequestContextData; public static synthetic fun fromHeaders$default (Lcom/posthog/server/PostHogRequestContext$Companion;Ljava/util/Map;ZLjava/util/Map;ILjava/lang/Object;)Lcom/posthog/server/PostHogRequestContextData; - public final fun withContext (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object; - public final fun withContext (Lcom/posthog/PostHogStateless;Lcom/posthog/server/PostHogRequestContextData;ZLkotlin/jvm/functions/Function0;)Ljava/lang/Object; + public final fun withContext (Lcom/posthog/server/PostHogInterface;Lcom/posthog/server/PostHogRequestContextData;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object; + public final fun withContext (Lcom/posthog/server/PostHogInterface;Lcom/posthog/server/PostHogRequestContextData;ZLkotlin/jvm/functions/Function0;)Ljava/lang/Object; public final fun withContext (Lcom/posthog/server/PostHogRequestContextData;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object; public final fun withContext (Lcom/posthog/server/PostHogRequestContextData;ZLkotlin/jvm/functions/Function0;)Ljava/lang/Object; } @@ -516,7 +516,7 @@ public final class com/posthog/server/PostHogRequestContextData { } public final class com/posthog/server/otel/PostHogContextSpanProcessor : io/opentelemetry/sdk/trace/SpanProcessor { - public fun (Lcom/posthog/PostHogStateless;)V + public fun (Lcom/posthog/server/PostHogInterface;)V public fun forceFlush ()Lio/opentelemetry/sdk/common/CompletableResultCode; public fun isEndRequired ()Z public fun isStartRequired ()Z diff --git a/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt b/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt index c2f5966f1..8b2f3f4a6 100644 --- a/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt +++ b/posthog-server/src/main/java/com/posthog/server/PostHogRequestContext.kt @@ -1,6 +1,5 @@ package com.posthog.server -import com.posthog.PostHogStateless import com.posthog.server.internal.enrichCurrentSpanSafely import java.util.UUID @@ -25,9 +24,9 @@ public class PostHogRequestContext private constructor() { private const val MAX_HEADER_VALUE_LENGTH = 1000 private val currentContext = ThreadLocal() - private val currentOwner = ThreadLocal() + private val currentOwner = ThreadLocal() - internal fun owner(): PostHogStateless? = currentOwner.get() + internal fun owner(): PostHogInterface? = currentOwner.get() /** * Returns the context active on the current thread, or null when no scope is active. @@ -69,7 +68,7 @@ public class PostHogRequestContext private constructor() { */ @JvmStatic public fun beginScope( - owner: PostHogStateless, + owner: PostHogInterface, data: PostHogRequestContextData, ): Scope = beginScope(owner, data, fresh = false) @@ -79,13 +78,13 @@ public class PostHogRequestContext private constructor() { */ @JvmStatic public fun beginScope( - owner: PostHogStateless, + owner: PostHogInterface, data: PostHogRequestContextData, fresh: Boolean, ): Scope = beginOwnedScope(owner, data, fresh) private fun beginOwnedScope( - owner: PostHogStateless?, + owner: PostHogInterface?, data: PostHogRequestContextData, fresh: Boolean, ): Scope { @@ -132,7 +131,7 @@ public class PostHogRequestContext private constructor() { * Runs [block] in an owner-bound scope and restores the previous scope afterwards. */ public fun withContext( - owner: PostHogStateless, + owner: PostHogInterface, data: PostHogRequestContextData, block: () -> T, ): T = withContext(owner, data, fresh = false, block) @@ -141,7 +140,7 @@ public class PostHogRequestContext private constructor() { * Runs [block] in an owner-bound scope, optionally ignoring previous context with [fresh]. */ public fun withContext( - owner: PostHogStateless, + owner: PostHogInterface, data: PostHogRequestContextData, fresh: Boolean, block: () -> T, @@ -306,7 +305,7 @@ public class PostHogRequestContext private constructor() { */ public class Scope internal constructor( private val previous: PostHogRequestContextData?, - private val previousOwner: PostHogStateless?, + private val previousOwner: PostHogInterface?, ) : AutoCloseable { private var closed: Boolean = false diff --git a/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt b/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt index 5012e8589..8dc47dcea 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt @@ -1,13 +1,13 @@ package com.posthog.server.internal -import com.posthog.PostHogStateless +import com.posthog.server.PostHogInterface import com.posthog.server.PostHogRequestContext import java.lang.ref.WeakReference import java.util.WeakHashMap -private val registeredProcessors = WeakHashMap>() +private val registeredProcessors = WeakHashMap>() -internal fun registerSpanContextOwner(owner: PostHogStateless): Any = +internal fun registerSpanContextOwner(owner: PostHogInterface): Any = Any().also { token -> synchronized(registeredProcessors) { registeredProcessors[token] = WeakReference(owner) diff --git a/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt b/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt index c98e51cd4..7a3a56208 100644 --- a/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt +++ b/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt @@ -1,23 +1,30 @@ package com.posthog.server.internal -import com.posthog.PostHogStateless +import com.posthog.server.PostHog +import com.posthog.server.PostHogInterface import com.posthog.server.PostHogRequestContext import io.opentelemetry.api.trace.Span import io.opentelemetry.sdk.trace.ReadWriteSpan -internal fun enrichActiveOpenTelemetrySpan(owner: PostHogStateless) { +internal fun enrichActiveOpenTelemetrySpan(owner: PostHogInterface) { val span = Span.current() if (span is ReadWriteSpan) { enrichSpanWithRequestContext(owner, span) } } +private fun PostHogInterface.optedOut(): Boolean = + when (this) { + is PostHog -> isOptOut() + } + internal fun enrichSpanWithRequestContext( - owner: PostHogStateless, + owner: PostHogInterface, span: ReadWriteSpan, ) { try { - if (owner.isOptOut() || PostHogRequestContext.owner() !== owner || !span.isRecording) { + // Check the owner first because a closed client logs on each isOptOut() call. + if (PostHogRequestContext.owner() !== owner || !span.isRecording || owner.optedOut()) { return } val context = PostHogRequestContext.current() ?: return diff --git a/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt b/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt index c6e3ee60f..37938ac14 100644 --- a/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt +++ b/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt @@ -1,6 +1,6 @@ package com.posthog.server.otel -import com.posthog.PostHogStateless +import com.posthog.server.PostHogInterface import com.posthog.server.PostHogRequestContext import com.posthog.server.internal.enrichSpanWithRequestContext import com.posthog.server.internal.registerSpanContextOwner @@ -10,6 +10,7 @@ import io.opentelemetry.sdk.common.CompletableResultCode import io.opentelemetry.sdk.trace.ReadWriteSpan import io.opentelemetry.sdk.trace.ReadableSpan import io.opentelemetry.sdk.trace.SpanProcessor +import java.lang.ref.WeakReference import java.util.concurrent.atomic.AtomicBoolean /** @@ -23,13 +24,16 @@ import java.util.concurrent.atomic.AtomicBoolean * `sessionId` and `posthogDistinctId`. Existing attributes with either name take * precedence regardless of their type. Missing or blank IDs remain absent. * - * Bind each request scope to the same [owner]. Opted-out clients add no attributes. + * Bind each request scope to the same owner client. Opted-out clients add no attributes. + * The processor holds the owner weakly. If the application replaces its client, + * build a new processor and provider for the new client. * Start and close a fresh PostHog context scope for each request. Work on another * thread requires its own explicit context scope. Close each scope on the thread * that opened it. This processor does not add * coroutine, Reactor, or other framework context propagation. */ -public class PostHogContextSpanProcessor public constructor(private val owner: PostHogStateless) : SpanProcessor { +public class PostHogContextSpanProcessor public constructor(owner: PostHogInterface) : SpanProcessor { + private val ownerRef = WeakReference(owner) private val stopped = AtomicBoolean(false) private val registration = registerSpanContextOwner(owner) @@ -37,9 +41,11 @@ public class PostHogContextSpanProcessor public constructor(private val owner: P parentContext: Context, span: ReadWriteSpan, ) { - if (!stopped.get()) { - enrichSpanWithRequestContext(owner, span) + if (stopped.get()) { + return } + val owner = ownerRef.get() ?: return + enrichSpanWithRequestContext(owner, span) } override fun isStartRequired(): Boolean = true diff --git a/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt b/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt index a521acef1..97879b8ea 100644 --- a/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt +++ b/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt @@ -20,7 +20,7 @@ import kotlin.test.assertNull import kotlin.test.assertTrue internal class PostHogIncomingSpanContextTest { - private val owner = PostHog().apply { setup(PostHogConfig("phc_test", preloadFeatureFlags = false)) } + private val owner: PostHogInterface = PostHog.with(PostHogConfig("phc_test", preloadFeatureFlags = false)) @AfterTest fun closeClient() {