diff --git a/.changeset/fiery-maps-obey.md b/.changeset/fiery-maps-obey.md new file mode 100644 index 000000000..b8f576a99 --- /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. 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 bf4956c8b..82960687e 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,32 @@ 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 + +`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. + +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() + +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. + ## 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..51421e12d 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/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; @@ -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/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; @@ -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/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; } @@ -509,3 +515,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 (Lcom/posthog/server/PostHogInterface;)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 a56043325..c1f79c0b0 100644 --- a/posthog-server/build.gradle.kts +++ b/posthog-server/build.gradle.kts @@ -107,9 +107,11 @@ dependencies { signature("org.codehaus.mojo.signature:java18:${PosthogBuildConfig.Plugins.SIGNATURE_JAVA18}@signature") compileOnly("io.opentelemetry:opentelemetry-api:1.32.0") + compileOnly("io.opentelemetry:opentelemetry-sdk-trace:1.32.0") // tests testImplementation("io.opentelemetry:opentelemetry-api:1.32.0") + 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 e27a69930..3008beb8a 100644 --- a/posthog-server/gradle.lockfile +++ b/posthog-server/gradle.lockfile @@ -21,6 +21,8 @@ com.squareup.okio:okio:3.11.0=jvmTestRuntimeClasspathForLint,runtimeClasspath,te 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..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,5 +1,6 @@ package com.posthog.server +import com.posthog.server.internal.enrichCurrentSpanSafely import java.util.UUID /** @@ -23,6 +24,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(): PostHogInterface? = currentOwner.get() /** * Returns the context active on the current thread, or null when no scope is active. @@ -54,11 +58,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: PostHogInterface, + 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: PostHogInterface, + data: PostHogRequestContextData, + fresh: Boolean, + ): Scope = beginOwnedScope(owner, data, fresh) + + private fun beginOwnedScope( + owner: PostHogInterface?, + 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)) - return Scope(previous) + currentOwner.set(owner) + enrichCurrentSpanSafely() + return Scope(previous, previousOwner) } /** @@ -91,6 +127,29 @@ public class PostHogRequestContext private constructor() { } } + /** + * Runs [block] in an owner-bound scope and restores the previous scope afterwards. + */ + public fun withContext( + owner: PostHogInterface, + 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: PostHogInterface, + data: PostHogRequestContextData, + fresh: Boolean, + block: () -> T, + ): T { + beginScope(owner, data, fresh).use { + return block() + } + } + /** * Extracts PostHog tracing headers into context data. * @@ -244,7 +303,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: PostHogInterface?, + ) : AutoCloseable { private var closed: Boolean = false override fun close() { @@ -257,6 +319,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 new file mode 100644 index 000000000..8dc47dcea --- /dev/null +++ b/posthog-server/src/main/java/com/posthog/server/internal/ContextSpanBridge.kt @@ -0,0 +1,45 @@ +package com.posthog.server.internal + +import com.posthog.server.PostHogInterface +import com.posthog.server.PostHogRequestContext +import java.lang.ref.WeakReference +import java.util.WeakHashMap + +private val registeredProcessors = WeakHashMap>() + +internal fun registerSpanContextOwner(owner: PostHogInterface): 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 { + Class.forName("io.opentelemetry.sdk.trace.ReadWriteSpan", false, PostHogRequestContext::class.java.classLoader) + true + } catch (_: Exception) { + false + } catch (_: LinkageError) { + false + } +} + +internal fun enrichCurrentSpanSafely() { + val owner = PostHogRequestContext.owner() ?: return + val registered = synchronized(registeredProcessors) { registeredProcessors.values.any { it.get() === owner } } + if (!registered || !openTelemetrySdkAvailable) { + return + } + try { + 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 new file mode 100644 index 000000000..7a3a56208 --- /dev/null +++ b/posthog-server/src/main/java/com/posthog/server/internal/OpenTelemetrySpanContext.kt @@ -0,0 +1,40 @@ +package com.posthog.server.internal + +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: 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: PostHogInterface, + span: ReadWriteSpan, +) { + try { + // 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 + 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..37938ac14 --- /dev/null +++ b/posthog-server/src/main/java/com/posthog/server/otel/PostHogContextSpanProcessor.kt @@ -0,0 +1,65 @@ +package com.posthog.server.otel + +import com.posthog.server.PostHogInterface +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.lang.ref.WeakReference +import java.util.concurrent.atomic.AtomicBoolean + +/** + * 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. + * + * 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(owner: PostHogInterface) : SpanProcessor { + private val ownerRef = WeakReference(owner) + private val stopped = AtomicBoolean(false) + private val registration = registerSpanContextOwner(owner) + + override fun onStart( + parentContext: Context, + span: ReadWriteSpan, + ) { + if (stopped.get()) { + return + } + val owner = ownerRef.get() ?: return + enrichSpanWithRequestContext(owner, span) + } + + override fun isStartRequired(): Boolean = true + + override fun onEnd(span: ReadableSpan): Unit = Unit + + override fun isEndRequired(): Boolean = false + + override fun shutdown(): CompletableResultCode { + if (stopped.compareAndSet(false, true)) { + unregisterSpanContextOwner(registration) + } + 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..97879b8ea --- /dev/null +++ b/posthog-server/src/test/java/com/posthog/server/PostHogIncomingSpanContextTest.kt @@ -0,0 +1,308 @@ +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.AfterTest +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 { + private val owner: PostHogInterface = PostHog.with(PostHogConfig("phc_test", preloadFeatureFlags = false)) + + @AfterTest + fun closeClient() { + owner.close() + } + + @Test + fun `request scope enriches an existing application SDK span`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).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(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"]) + } + } + } finally { + span.end() + } + } + } + + @Test + fun `child spans keep their request snapshot after the scope ends`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> + val tracer = provider.get("posthog-test") + val child = + PostHogRequestContext.withContext(owner, PostHogRequestContextData("person-one", "session-one"), fresh = true) { + tracer.spanBuilder("child").startSpan() as ReadWriteSpan + } + 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"]) + 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(owner)).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( + owner, + 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(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(owner, 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(owner)).build().use { provider -> + for (value in listOf(null, "", " ")) { + 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) + } + } + } + } + + @Test + fun `failure closes inner scope and the next span sees its parent context`() { + SdkTracerProvider.builder().addSpanProcessor(PostHogContextSpanProcessor(owner)).build().use { provider -> + val tracer = provider.get("posthog-test") + PostHogRequestContext.withContext(owner, PostHogRequestContextData("outer-person", "outer-session"), fresh = true) { + assertFailsWith { + 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"))) + 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(owner)).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( + owner, + 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(owner)).build().use { provider -> + val executor = Executors.newSingleThreadExecutor() + val release = CountDownLatch(1) + try { + val work = + 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 + 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(owner) + SdkTracerProvider.builder().addSpanProcessor(processor).build().use { provider -> + val incoming = provider.get("posthog-test").spanBuilder("incoming").startSpan() as ReadWriteSpan + assertTrue(processor.shutdown().isSuccess) + 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) + } + } + } + + @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(owner)).build().use { provider -> + val span = provider.get("posthog-test").spanBuilder("sampled-out").startSpan() + span.makeCurrent().use { + PostHogRequestContext.beginScope(owner, 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() + } + } + } +} 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() + } + } +}