From d13765013b5436e6458401485c0b0b8b828c7848 Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Wed, 28 Oct 2026 16:26:23 -0700 Subject: [PATCH 1/2] feat(java): add setAttributes and custom span/log processors Co-authored-by: Cursor --- .../observability-java/README.md | 75 ++++++- .../observability-java/lib/build.gradle | 11 +- .../launchdarkly/observability/LDObserve.java | 43 +++- .../observability/ObservabilityOptions.java | 75 +++++++ .../internal/GlobalAttributes.java | 96 ++++++++ .../GlobalAttributesLogProcessor.java | 28 +++ .../GlobalAttributesSpanProcessor.java | 41 ++++ .../observability/internal/OtelManager.java | 25 ++- .../internal/SafeLogRecordProcessor.java | 70 ++++++ .../internal/SafeSpanProcessor.java | 100 +++++++++ .../ObservabilityPluginTest.java | 47 ++++ .../internal/GlobalAttributesTest.java | 169 ++++++++++++++ .../internal/PluginUserFlowTest.java | 210 +++++++++++++++++- .../internal/SafeProcessorTest.java | 202 +++++++++++++++++ 14 files changed, 1179 insertions(+), 13 deletions(-) create mode 100644 sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributes.java create mode 100644 sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesLogProcessor.java create mode 100644 sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesSpanProcessor.java create mode 100644 sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeLogRecordProcessor.java create mode 100644 sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeSpanProcessor.java create mode 100644 sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/GlobalAttributesTest.java create mode 100644 sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/SafeProcessorTest.java diff --git a/sdk/@launchdarkly/observability-java/README.md b/sdk/@launchdarkly/observability-java/README.md index 1f784a55ad..a2017ac639 100644 --- a/sdk/@launchdarkly/observability-java/README.md +++ b/sdk/@launchdarkly/observability-java/README.md @@ -42,9 +42,9 @@ The plugin does not bring the LaunchDarkly SDK with it, so you choose its versio ### Dependencies you get -- `io.opentelemetry:opentelemetry-api` is a compile-scoped (`api`) dependency, because `LDObserve` and `ObservabilityOptions` take OpenTelemetry API types such as `Attributes` and `Severity`. You do not need to declare it yourself. +- `io.opentelemetry:opentelemetry-api` is a compile-scoped (`api`) dependency, because `LDObserve` and `ObservabilityOptions` take OpenTelemetry API types such as `Attributes` and `Severity`. You do not need to declare it yourself. `io.opentelemetry:opentelemetry-sdk` is compile-scoped too, because `ObservabilityOptions` accepts SDK `SpanProcessor` and `LogRecordProcessor` instances. - OTLP is sent over HTTP with OpenTelemetry's JDK sender, which uses `java.net.http.HttpClient`. The plugin does **not** pull OkHttp, Okio or the Kotlin standard library onto your classpath. The build fails if any of those leak in. -- The OpenTelemetry SDK and exporters are runtime dependencies, and they are the only ones: JSON is handled with the LaunchDarkly SDK's own `LDValue`, so no Gson or other JSON library is added. The plugin records flag evaluations with its own hook and does not depend on `launchdarkly-java-server-sdk-otel` or on the LaunchDarkly SDK itself. The build fails if the POM names anything outside `io.opentelemetry`. +- The OpenTelemetry exporters are runtime dependencies, and together with the OpenTelemetry API and SDK they are the only ones: JSON is handled with the LaunchDarkly SDK's own `LDValue`, so no Gson or other JSON library is added. The plugin records flag evaluations with its own hook and does not depend on `launchdarkly-java-server-sdk-otel` or on the LaunchDarkly SDK itself. The build fails if the POM names anything outside `io.opentelemetry`. ### OpenTelemetry versions @@ -259,6 +259,73 @@ LDObserve.shutdown(); Every `LDObserve` method is safe to call before the plugin is initialized; such calls are local no-ops and do not install the OpenTelemetry no-op global. +### Global attributes + +`LDObserve.setAttributes` adds attributes to everything recorded after the call: every span and log record the plugin exports (flag-evaluation spans, error spans, and spans or records from instrumentation wired to `LDObserve.getOpenTelemetry()` included) and every metric recorded through `LDObserve`. Calling it again merges by key, and `LDObserve.removeAttributes` drops keys. + +```java +import com.launchdarkly.observability.LDObserve; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; + +import java.util.List; + +LDObserve.setAttributes(Attributes.of( + AttributeKey.stringKey("tenant.id"), "acme", + AttributeKey.stringKey("deployment.region"), "us-east-1")); + +// Later: replace one value and drop another +LDObserve.setAttributes(Attributes.of(AttributeKey.stringKey("deployment.region"), "eu-west-1")); +LDObserve.removeAttributes(List.of("tenant.id")); +``` + +- Attributes passed to an `LDObserve` call, or set on a span builder, win over global attributes of the same key; global attributes only fill keys that are missing. Resource attributes are separate and unchanged. +- A span gets the global attributes current when it starts. +- Keys are matched by name, whatever the attribute type. +- The attributes are held in memory for the life of the process (they survive `LDObserve.shutdown()`), are shared by every thread, and can be set before the plugin starts. +- Metrics recorded directly on a meter from `LDObserve.getOpenTelemetry()` do not get them; record through `LDObserve` instead. + +### Custom span and log processors + +`spanProcessors` and `logProcessors` add your own OpenTelemetry `SpanProcessor` and `LogRecordProcessor` instances to the plugin's pipeline, for example to enrich or redact telemetry before it is exported. `spanProcessor` and `logProcessor` add one at a time. + +```java +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.logs.LogRecordProcessor; +import io.opentelemetry.sdk.trace.ReadWriteSpan; +import io.opentelemetry.sdk.trace.ReadableSpan; +import io.opentelemetry.sdk.trace.SpanProcessor; + +AttributeKey email = AttributeKey.stringKey("user.email"); + +SpanProcessor addPool = new SpanProcessor() { + @Override + public void onStart(Context parentContext, ReadWriteSpan span) { + span.setAttribute("host.pool", "blue"); + } + @Override public boolean isStartRequired() { return true; } + @Override public void onEnd(ReadableSpan span) {} + @Override public boolean isEndRequired() { return false; } +}; +LogRecordProcessor redactEmail = (context, logRecord) -> { + if (logRecord.toLogRecordData().getAttributes().get(email) != null) { + logRecord.setAttribute(email, "[redacted]"); + } +}; + +ObservabilityOptions options = new ObservabilityOptions.Builder() + .serviceName("my-service") + .spanProcessors(List.of(addPool)) + .logProcessors(List.of(redactEmail)) + .build(); +``` + +- Processors run in list order, after the plugin adds the global attributes and before its own processing: the `logsApiLevel` filter, the LaunchDarkly sampling rules and export. Changes a processor makes (in `onStart` for spans) are exported, and sampling still applies. +- A processor that throws is logged and skipped for that call; the span or record is still exported. `forceFlush` and `shutdown` are forwarded when the plugin flushes or shuts down. +- They are read when the providers start. A second `LDClient` with another SDK key does not add its processors (see [Lifecycle and multiple clients](#lifecycle-and-multiple-clients)). +- The interfaces come from the OpenTelemetry SDK (`opentelemetry-sdk`), which the plugin exposes as a compile-scoped dependency for this reason. + ## Configuration Options | Option | Default | Description | @@ -275,6 +342,8 @@ Every `LDObserve` method is safe to call before the plugin is initialized; such | `logsApiLevel` | `INFO` | Minimum severity for exported logs. `NONE` disables log export. | | `debug` | `false` | Enable debug logging | | `manualStart` | `false` | Delay OTLP provider startup until `LDObserve.start(sdkKey, options)` | +| `spanProcessors` / `spanProcessor` | none | Your own span processors, run in order before sampling and export. See [Custom span and log processors](#custom-span-and-log-processors) | +| `logProcessors` / `logProcessor` | none | Your own log record processors, run in order before the level filter, sampling and export | | `shutdownHook` | `true` | Flush and stop the providers from a JVM shutdown hook. Set `false` under Spring Boot or another framework that owns shutdown, and call `LDObserve.shutdown()` from its close callback, so telemetry from requests drained during graceful shutdown is not dropped | ### Proxies and custom headers @@ -307,7 +376,7 @@ Levels are `TRACE`, `DEBUG`, `INFO`, `WARN`, `ERROR`, `FATAL` and `NONE`, matchi ./gradlew :lib:publishToMavenLocal # then see ../../e2e/java-plugin for consumer and server examples ``` -`:lib:build` fails if OkHttp, Okio or Kotlin appear on the runtime classpath, if the generated POM does not expose `opentelemetry-api` at compile scope, or if it names anything outside `io.opentelemetry`. `./gradlew :lib:test -PldServerSdkVersion=7.10.1` runs the tests against the oldest supported SDK. +`:lib:build` fails if OkHttp, Okio or Kotlin appear on the runtime classpath, if the generated POM does not expose `opentelemetry-api` and `opentelemetry-sdk` at compile scope, or if it names anything outside `io.opentelemetry`. `./gradlew :lib:test -PldServerSdkVersion=7.10.1` runs the tests against the oldest supported SDK. The unit tests run the sampling scenarios shared with the Go, .NET, Python and Node SDKs (`lib/src/test/resources/sampling`, copied from `go/internal/otel`). `:lib:build` also fails if those copies drift from Go's; copy the files over when the scenarios change. diff --git a/sdk/@launchdarkly/observability-java/lib/build.gradle b/sdk/@launchdarkly/observability-java/lib/build.gradle index 4fc25a0fb6..b45f3d55dc 100644 --- a/sdk/@launchdarkly/observability-java/lib/build.gradle +++ b/sdk/@launchdarkly/observability-java/lib/build.gradle @@ -41,7 +41,9 @@ dependencies { // types such as Attributes and Severity, so consumers must compile against // it. 'api' publishes it with compile scope in the POM. api "io.opentelemetry:opentelemetry-api:${otelVersion}" - implementation "io.opentelemetry:opentelemetry-sdk:${otelVersion}" + // ObservabilityOptions takes SDK SpanProcessor and LogRecordProcessor + // instances, so the SDK is part of the public API too. + api "io.opentelemetry:opentelemetry-sdk:${otelVersion}" implementation "io.opentelemetry:opentelemetry-sdk-logs:${otelVersion}" implementation "io.opentelemetry:opentelemetry-sdk-metrics:${otelVersion}" @@ -102,7 +104,7 @@ tasks.register('checkRuntimeClasspath') { tasks.register('checkPublishedPom') { group = 'verification' - description = 'Checks the generated POM exposes opentelemetry-api at compile scope and depends on nothing but OpenTelemetry.' + description = 'Checks the generated POM exposes opentelemetry-api and opentelemetry-sdk at compile scope and depends on nothing but OpenTelemetry.' dependsOn 'generatePomFileForReleasePublication' def pomFile = layout.buildDirectory.file('publications/release/pom-default.xml') doLast { @@ -114,6 +116,11 @@ tasks.register('checkPublishedPom') { throw new GradleException("opentelemetry-api must be a compile-scoped dependency in the POM, found scope '" + (otelApi.size() == 0 ? 'missing' : otelApi.scope.text()) + "'") } + def otelSdk = deps.find { it.artifactId.text() == 'opentelemetry-sdk' } + if (otelSdk.size() == 0 || otelSdk.scope.text() != 'compile') { + throw new GradleException("opentelemetry-sdk must be a compile-scoped dependency in the POM, found scope '" + + (otelSdk.size() == 0 ? 'missing' : otelSdk.scope.text()) + "'") + } def okhttp = deps.find { it.artifactId.text() == 'opentelemetry-exporter-sender-okhttp' } if (okhttp.size() != 0) { throw new GradleException('The POM must not depend on opentelemetry-exporter-sender-okhttp') diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/LDObserve.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/LDObserve.java index babfb5a030..9561646fcd 100644 --- a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/LDObserve.java +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/LDObserve.java @@ -1,6 +1,7 @@ package com.launchdarkly.observability; import com.launchdarkly.observability.internal.Constants; +import com.launchdarkly.observability.internal.GlobalAttributes; import com.launchdarkly.observability.internal.OtelManager; import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.common.Attributes; @@ -10,6 +11,8 @@ import io.opentelemetry.api.trace.StatusCode; import io.opentelemetry.api.trace.Tracer; +import java.util.Collection; + /** * Static singleton public API for LaunchDarkly Observability. * @@ -22,12 +25,44 @@ public final class LDObserve { private LDObserve() {} + /** + * Merges attributes into the process-wide global attributes, which are + * added to every span and log record the plugin exports (flag + * evaluations, error spans and spans or records from instrumentation + * using {@link #getOpenTelemetry()} included) and to every metric + * recorded through {@code LDObserve}. A key already in the global + * attributes takes the new value. + * + *

Global attributes never replace an attribute the span, record or + * metric call already has: the attributes passed to an {@code LDObserve} + * method, or set on a span builder, win; resource attributes are + * separate and stay as they are. A span gets the values current when it + * starts; changing them later does not touch spans already started.

+ * + *

Kept in memory for the life of the process, including across + * {@link #shutdown()}. Safe to call before the plugin starts, and from + * any thread. Keys are matched by name, whatever the attribute type. + * Null or empty attributes are ignored.

+ */ + public static void setAttributes(Attributes attributes) { + GlobalAttributes.set(attributes); + } + + /** + * Removes the global attributes with these key names. See + * {@link #setAttributes(Attributes)}. Names that are not set are + * ignored, as is a null collection. + */ + public static void removeAttributes(Collection keys) { + GlobalAttributes.remove(keys); + } + /** * Records a gauge metric. */ public static void recordMetric(String name, double value, Attributes attributes) { try { - OtelManager.getGauge(name).set(value, orEmpty(attributes)); + OtelManager.getGauge(name).set(value, GlobalAttributes.merge(attributes)); } catch (LinkageError e) { // The synchronous gauge API arrived in OpenTelemetry 1.38. On an // older API (Spring Boot 3.2 pins 1.31) the plugin is already off @@ -39,7 +74,7 @@ public static void recordMetric(String name, double value, Attributes attributes * Records a counter metric. */ public static void recordCount(String name, long value, Attributes attributes) { - OtelManager.getCounter(name).add(value, orEmpty(attributes)); + OtelManager.getCounter(name).add(value, GlobalAttributes.merge(attributes)); } /** @@ -53,14 +88,14 @@ public static void recordIncr(String name, Attributes attributes) { * Records a histogram metric. */ public static void recordHistogram(String name, double value, Attributes attributes) { - OtelManager.getHistogram(name).record(value, orEmpty(attributes)); + OtelManager.getHistogram(name).record(value, GlobalAttributes.merge(attributes)); } /** * Records an up/down counter metric. */ public static void recordUpDownCounter(String name, long value, Attributes attributes) { - OtelManager.getUpDownCounter(name).add(value, orEmpty(attributes)); + OtelManager.getUpDownCounter(name).add(value, GlobalAttributes.merge(attributes)); } /** diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/ObservabilityOptions.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/ObservabilityOptions.java index 1ab432c9db..638e3fff2c 100644 --- a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/ObservabilityOptions.java +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/ObservabilityOptions.java @@ -3,12 +3,15 @@ import com.launchdarkly.observability.internal.Constants; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.logs.Severity; +import io.opentelemetry.sdk.logs.LogRecordProcessor; +import io.opentelemetry.sdk.trace.SpanProcessor; import java.net.InetSocketAddress; import java.net.Proxy; import java.net.ProxySelector; import java.net.SocketAddress; import java.net.URI; +import java.util.ArrayList; import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; @@ -34,6 +37,8 @@ public final class ObservabilityOptions { private final boolean debug; private final boolean manualStart; private final boolean shutdownHook; + private final List spanProcessors; + private final List logProcessors; private ObservabilityOptions(Builder builder) { this.enabled = builder.enabled; @@ -49,6 +54,8 @@ private ObservabilityOptions(Builder builder) { this.debug = builder.debug; this.manualStart = builder.manualStart; this.shutdownHook = builder.shutdownHook; + this.spanProcessors = Collections.unmodifiableList(new ArrayList<>(builder.spanProcessors)); + this.logProcessors = Collections.unmodifiableList(new ArrayList<>(builder.logProcessors)); } /** Whether the plugin exports telemetry at all. See {@link Builder#enabled(boolean)}. */ @@ -70,6 +77,10 @@ private ObservabilityOptions(Builder builder) { public boolean isManualStart() { return manualStart; } /** Whether the plugin flushes and stops on JVM exit by itself. See {@link Builder#shutdownHook(boolean)}. */ public boolean isShutdownHook() { return shutdownHook; } + /** Span processors run on every span before export, in order. See {@link Builder#spanProcessors(List)}. Never null. */ + public List getSpanProcessors() { return spanProcessors; } + /** Log record processors run on every record before export, in order. See {@link Builder#logProcessors(List)}. Never null. */ + public List getLogProcessors() { return logProcessors; } /** * Minimum severity for exported log records. Mirrors the OpenTelemetry @@ -121,6 +132,8 @@ public static class Builder { private boolean debug = false; private boolean manualStart = false; private boolean shutdownHook = true; + private List spanProcessors = new ArrayList<>(); + private List logProcessors = new ArrayList<>(); public Builder() {} @@ -268,6 +281,68 @@ public Builder shutdownHook(boolean shutdownHook) { return this; } + /** + * Sets span processors that run on every span the plugin's tracer + * provider records, before it is sampled and exported. Replaces any + * previously set span processors. + * + *

They run in list order, after the plugin has added the + * {@linkplain LDObserve#setAttributes global attributes}, so changes + * they make in {@code onStart} are exported and the LaunchDarkly + * sampling rules still apply. An exception a processor throws is + * logged and does not stop the span from being exported. + * {@code forceFlush} and {@code shutdown} are forwarded when the + * plugin flushes or shuts down.

+ * + *

Read when the providers start; later changes have no effect.

+ */ + public Builder spanProcessors(List spanProcessors) { + List copy = new ArrayList<>(Objects.requireNonNull(spanProcessors, "spanProcessors")); + copy.forEach(processor -> Objects.requireNonNull(processor, "spanProcessors element")); + this.spanProcessors = copy; + return this; + } + + /** + * Adds one span processor after any already set. See + * {@link #spanProcessors(List)}. + */ + public Builder spanProcessor(SpanProcessor spanProcessor) { + this.spanProcessors.add(Objects.requireNonNull(spanProcessor, "spanProcessor")); + return this; + } + + /** + * Sets log record processors that run on every log record the + * plugin's logger provider receives, before the + * {@link #logsApiLevel(LogLevel) level filter}, sampling and export. + * Replaces any previously set log processors. + * + *

They run in list order, after the plugin has added the + * {@linkplain LDObserve#setAttributes global attributes}, so changes + * they make are exported. An exception a processor throws is logged + * and does not stop the record from being exported. + * {@code forceFlush} and {@code shutdown} are forwarded when the + * plugin flushes or shuts down.

+ * + *

Read when the providers start; later changes have no effect.

+ */ + public Builder logProcessors(List logProcessors) { + List copy = new ArrayList<>(Objects.requireNonNull(logProcessors, "logProcessors")); + copy.forEach(processor -> Objects.requireNonNull(processor, "logProcessors element")); + this.logProcessors = copy; + return this; + } + + /** + * Adds one log record processor after any already set. See + * {@link #logProcessors(List)}. + */ + public Builder logProcessor(LogRecordProcessor logProcessor) { + this.logProcessors.add(Objects.requireNonNull(logProcessor, "logProcessor")); + return this; + } + public ObservabilityOptions build() { return new ObservabilityOptions(this); } diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributes.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributes.java new file mode 100644 index 0000000000..a1d8c8853e --- /dev/null +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributes.java @@ -0,0 +1,96 @@ +package com.launchdarkly.observability.internal; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; + +/** + * The process-wide attributes added to every span, log record and + * {@code LDObserve} metric. Held as one immutable snapshot, so readers on the + * hot path pay a volatile read and nothing else. It lives outside + * {@link OtelManager} so values set before the plugin starts apply once it + * does. + * + *

Keys are matched by name: setting {@code "tier"} as a long replaces a + * {@code "tier"} string, and a span or record that already has a {@code "tier"} + * of any type keeps it.

+ */ +public final class GlobalAttributes { + + private static final AtomicReference CURRENT = new AtomicReference<>(Attributes.empty()); + + private GlobalAttributes() {} + + public static Attributes get() { + return CURRENT.get(); + } + + /** Merges {@code attributes} into the global set; their values replace those of keys already present. */ + public static void set(Attributes attributes) { + if (attributes == null || attributes.isEmpty()) { + return; + } + Set names = names(attributes); + CURRENT.updateAndGet(current -> { + AttributesBuilder builder = current.toBuilder(); + builder.removeIf(key -> names.contains(key.getKey())); + return builder.putAll(attributes).build(); + }); + } + + /** Removes the attributes with these key names. */ + public static void remove(Collection keys) { + if (keys == null || keys.isEmpty()) { + return; + } + Set names = new HashSet<>(keys); + CURRENT.updateAndGet(current -> { + AttributesBuilder builder = current.toBuilder(); + builder.removeIf(key -> names.contains(key.getKey())); + return builder.build(); + }); + } + + /** + * Returns {@code attributes} with the global attributes added for every key + * name it does not already have. + */ + public static Attributes merge(Attributes attributes) { + Attributes globals = CURRENT.get(); + if (globals.isEmpty()) { + return attributes != null ? attributes : Attributes.empty(); + } + if (attributes == null || attributes.isEmpty()) { + return globals; + } + return attributes.toBuilder().putAll(missingFrom(globals, attributes)).build(); + } + + /** The entries of {@code globals} whose key name is not in {@code existing}. */ + static Attributes missingFrom(Attributes globals, Attributes existing) { + if (existing.isEmpty()) { + return globals; + } + Set present = names(existing); + AttributesBuilder builder = globals.toBuilder(); + builder.removeIf(key -> present.contains(key.getKey())); + return builder.build(); + } + + private static Set names(Attributes attributes) { + Set names = new HashSet<>(); + for (AttributeKey key : attributes.asMap().keySet()) { + names.add(key.getKey()); + } + return names; + } + + static void resetForTest() { + CURRENT.set(Attributes.empty()); + } +} diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesLogProcessor.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesLogProcessor.java new file mode 100644 index 0000000000..f7ca105322 --- /dev/null +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesLogProcessor.java @@ -0,0 +1,28 @@ +package com.launchdarkly.observability.internal; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.logs.LogRecordProcessor; +import io.opentelemetry.sdk.logs.ReadWriteLogRecord; + +/** + * Adds the {@link GlobalAttributes} to every log record emitted through the + * plugin's logger provider, for each key the record does not already have. + * Registered first, so user processors see them. + */ +public final class GlobalAttributesLogProcessor implements LogRecordProcessor { + + @Override + public void onEmit(Context context, ReadWriteLogRecord logRecord) { + Attributes globals = GlobalAttributes.get(); + if (globals.isEmpty()) { + return; + } + // ReadWriteLogRecord#getAttributes is newer than the 1.38 floor. + Attributes missing = GlobalAttributes.missingFrom( + globals, logRecord.toLogRecordData().getAttributes()); + if (!missing.isEmpty()) { + logRecord.setAllAttributes(missing); + } + } +} diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesSpanProcessor.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesSpanProcessor.java new file mode 100644 index 0000000000..166f146cb4 --- /dev/null +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/GlobalAttributesSpanProcessor.java @@ -0,0 +1,41 @@ +package com.launchdarkly.observability.internal; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.trace.ReadWriteSpan; +import io.opentelemetry.sdk.trace.ReadableSpan; +import io.opentelemetry.sdk.trace.SpanProcessor; + +/** + * Adds the {@link GlobalAttributes} current at span start to every span the + * plugin's tracer provider starts, for each key the span was not started with. + * Registered first, so user processors see them. + */ +public final class GlobalAttributesSpanProcessor implements SpanProcessor { + + @Override + public void onStart(Context parentContext, ReadWriteSpan span) { + Attributes globals = GlobalAttributes.get(); + if (globals.isEmpty()) { + return; + } + Attributes missing = GlobalAttributes.missingFrom(globals, span.getAttributes()); + if (!missing.isEmpty()) { + span.setAllAttributes(missing); + } + } + + @Override + public boolean isStartRequired() { + return true; + } + + @Override + public void onEnd(ReadableSpan span) { + } + + @Override + public boolean isEndRequired() { + return false; + } +} diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/OtelManager.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/OtelManager.java index 7704f15117..2b933d02b3 100644 --- a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/OtelManager.java +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/OtelManager.java @@ -32,13 +32,17 @@ import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.common.CompletableResultCode; import io.opentelemetry.sdk.common.export.ProxyOptions; +import io.opentelemetry.sdk.logs.LogRecordProcessor; import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.SdkLoggerProviderBuilder; import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor; import io.opentelemetry.sdk.metrics.SdkMeterProvider; import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; import io.opentelemetry.sdk.resources.Resource; import io.opentelemetry.sdk.trace.ReadableSpan; import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder; +import io.opentelemetry.sdk.trace.SpanProcessor; import io.opentelemetry.sdk.trace.export.BatchSpanProcessor; import io.opentelemetry.sdk.trace.samplers.Sampler; @@ -236,8 +240,17 @@ private static OtelManager create(String sdkKey, ObservabilityOptions options) { // current parent comes unsampled from the Java agent, another SDK or // an upstream service. Volume is governed by the backend rules that // SamplingTraceExporter applies at export. - SdkTracerProvider tracerProvider = SdkTracerProvider.builder() + // + // Processors run in registration order on the same span: global + // attributes first, then the user's, so the exporter (and the + // sampling rules it applies) sees what they set. + SdkTracerProviderBuilder tracerProviderBuilder = SdkTracerProvider.builder() .setSampler(Sampler.alwaysOn()) + .addSpanProcessor(new GlobalAttributesSpanProcessor()); + for (SpanProcessor processor : options.getSpanProcessors()) { + tracerProviderBuilder.addSpanProcessor(new SafeSpanProcessor(processor)); + } + SdkTracerProvider tracerProvider = tracerProviderBuilder .addSpanProcessor(BatchSpanProcessor.builder(samplingTraceExporter) .setScheduleDelay(Duration.ofSeconds(1)) .setExporterTimeout(Duration.ofSeconds(30)) @@ -263,7 +276,14 @@ private static OtelManager create(String sdkKey, ObservabilityOptions options) { LogLevelFilterProcessor levelFilter = new LogLevelFilterProcessor( samplingLogProcessor, options.getLogsApiLevel().getSeverityNumber()); - SdkLoggerProvider loggerProvider = SdkLoggerProvider.builder() + // As for spans: global attributes, then the user's processors, which + // see records the level filter later drops. + SdkLoggerProviderBuilder loggerProviderBuilder = SdkLoggerProvider.builder() + .addLogRecordProcessor(new GlobalAttributesLogProcessor()); + for (LogRecordProcessor processor : options.getLogProcessors()) { + loggerProviderBuilder.addLogRecordProcessor(new SafeLogRecordProcessor(processor)); + } + SdkLoggerProvider loggerProvider = loggerProviderBuilder .addLogRecordProcessor(levelFilter) .setResource(resource) .build(); @@ -567,6 +587,7 @@ static void resetForTest() { synchronized (INIT_LOCK) { manager = INSTANCE.getAndSet(null); PENDING_SAMPLING_CONFIG.set(null); + GlobalAttributes.resetForTest(); registeredSdkKey = null; samplingFetchSdkKey = null; if (shutdownHook != null) { diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeLogRecordProcessor.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeLogRecordProcessor.java new file mode 100644 index 0000000000..d274c25bad --- /dev/null +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeLogRecordProcessor.java @@ -0,0 +1,70 @@ +package com.launchdarkly.observability.internal; + +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.LogRecordProcessor; +import io.opentelemetry.sdk.logs.ReadWriteLogRecord; + +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Wraps a log record processor from {@code ObservabilityOptions} so an + * exception it throws is logged instead of reaching the code that logged, or + * stopping the processors after it, the exporter included. + */ +final class SafeLogRecordProcessor implements LogRecordProcessor { + + private static final Logger log = Logger.getLogger(SafeLogRecordProcessor.class.getName()); + + private final LogRecordProcessor delegate; + private final AtomicBoolean failureLogged = new AtomicBoolean(); + + SafeLogRecordProcessor(LogRecordProcessor delegate) { + this.delegate = delegate; + } + + @Override + public void onEmit(Context context, ReadWriteLogRecord logRecord) { + try { + delegate.onEmit(context, logRecord); + } catch (RuntimeException | LinkageError e) { + logFailure("onEmit", e); + } + } + + @Override + public CompletableResultCode forceFlush() { + try { + CompletableResultCode result = delegate.forceFlush(); + return result != null ? result : CompletableResultCode.ofSuccess(); + } catch (RuntimeException | LinkageError e) { + logFailure("forceFlush", e); + return CompletableResultCode.ofFailure(); + } + } + + @Override + public CompletableResultCode shutdown() { + try { + CompletableResultCode result = delegate.shutdown(); + return result != null ? result : CompletableResultCode.ofSuccess(); + } catch (RuntimeException | LinkageError e) { + logFailure("shutdown", e); + return CompletableResultCode.ofFailure(); + } + } + + /** See {@link SafeSpanProcessor}: only the first failure is logged at WARNING. */ + private void logFailure(String method, Throwable e) { + Level level = failureLogged.compareAndSet(false, true) ? Level.WARNING : Level.FINE; + log.log(level, "Log record processor " + delegate.getClass().getName() + " threw from " + method + + "; the record is still exported", e); + } + + @Override + public String toString() { + return "SafeLogRecordProcessor{" + delegate + "}"; + } +} diff --git a/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeSpanProcessor.java b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeSpanProcessor.java new file mode 100644 index 0000000000..fa86db4a0f --- /dev/null +++ b/sdk/@launchdarkly/observability-java/lib/src/main/java/com/launchdarkly/observability/internal/SafeSpanProcessor.java @@ -0,0 +1,100 @@ +package com.launchdarkly.observability.internal; + +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; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * Wraps a span processor from {@code ObservabilityOptions} so an exception it + * throws is logged instead of reaching the code that started or ended the + * span, or stopping the processors after it, the exporter included. + */ +final class SafeSpanProcessor implements SpanProcessor { + + private static final Logger log = Logger.getLogger(SafeSpanProcessor.class.getName()); + + private final SpanProcessor delegate; + private final AtomicBoolean failureLogged = new AtomicBoolean(); + + SafeSpanProcessor(SpanProcessor delegate) { + this.delegate = delegate; + } + + @Override + public void onStart(Context parentContext, ReadWriteSpan span) { + try { + delegate.onStart(parentContext, span); + } catch (RuntimeException | LinkageError e) { + logFailure("onStart", e); + } + } + + @Override + public boolean isStartRequired() { + try { + return delegate.isStartRequired(); + } catch (RuntimeException | LinkageError e) { + logFailure("isStartRequired", e); + return true; + } + } + + @Override + public void onEnd(ReadableSpan span) { + try { + delegate.onEnd(span); + } catch (RuntimeException | LinkageError e) { + logFailure("onEnd", e); + } + } + + @Override + public boolean isEndRequired() { + try { + return delegate.isEndRequired(); + } catch (RuntimeException | LinkageError e) { + logFailure("isEndRequired", e); + return true; + } + } + + @Override + public CompletableResultCode forceFlush() { + try { + CompletableResultCode result = delegate.forceFlush(); + return result != null ? result : CompletableResultCode.ofSuccess(); + } catch (RuntimeException | LinkageError e) { + logFailure("forceFlush", e); + return CompletableResultCode.ofFailure(); + } + } + + @Override + public CompletableResultCode shutdown() { + try { + CompletableResultCode result = delegate.shutdown(); + return result != null ? result : CompletableResultCode.ofSuccess(); + } catch (RuntimeException | LinkageError e) { + logFailure("shutdown", e); + return CompletableResultCode.ofFailure(); + } + } + + /** The first failure is logged with its stack trace; later ones only at FINE, so a processor failing on every span does not flood the log. */ + private void logFailure(String method, Throwable e) { + Level level = failureLogged.compareAndSet(false, true) ? Level.WARNING : Level.FINE; + log.log(level, "Span processor " + delegate.getClass().getName() + " threw from " + method + + "; the span is still exported", e); + } + + @Override + public String toString() { + return "SafeSpanProcessor{" + delegate + "}"; + } +} diff --git a/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/ObservabilityPluginTest.java b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/ObservabilityPluginTest.java index 7199109802..a3df5220e9 100644 --- a/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/ObservabilityPluginTest.java +++ b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/ObservabilityPluginTest.java @@ -5,10 +5,18 @@ import com.launchdarkly.sdk.server.integrations.PluginMetadata; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.sdk.logs.LogRecordProcessor; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.SpanProcessor; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; import org.junit.jupiter.api.Test; import java.net.Proxy; import java.net.URI; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Map; @@ -54,6 +62,45 @@ void optionsBuilderDefaults() { assertTrue(options.getCustomHeaders().isEmpty()); assertNull(options.getProxySelector()); assertEquals(ObservabilityOptions.LogLevel.INFO, options.getLogsApiLevel()); + assertTrue(options.getSpanProcessors().isEmpty()); + assertTrue(options.getLogProcessors().isEmpty()); + } + + @Test + void processorsAccumulateInOrderAndAreCopied() { + SpanProcessor spanA = SimpleSpanProcessor.create(InMemorySpanExporter.create()); + SpanProcessor spanB = SimpleSpanProcessor.create(InMemorySpanExporter.create()); + LogRecordProcessor logA = SimpleLogRecordProcessor.create(InMemoryLogRecordExporter.create()); + LogRecordProcessor logB = SimpleLogRecordProcessor.create(InMemoryLogRecordExporter.create()); + List spanList = new ArrayList<>(List.of(spanA)); + ObservabilityOptions.Builder builder = new ObservabilityOptions.Builder() + .spanProcessors(spanList) + .spanProcessor(spanB) + .logProcessors(List.of(logA)) + .logProcessor(logB); + spanList.add(SimpleSpanProcessor.create(InMemorySpanExporter.create())); + ObservabilityOptions options = builder.build(); + + assertEquals(List.of(spanA, spanB), options.getSpanProcessors()); + assertEquals(List.of(logA, logB), options.getLogProcessors()); + assertThrows(UnsupportedOperationException.class, () -> options.getSpanProcessors().add(spanA)); + assertThrows(UnsupportedOperationException.class, () -> options.getLogProcessors().add(logA)); + + ObservabilityOptions replaced = builder.spanProcessors(List.of(spanB)).logProcessors(List.of()).build(); + assertEquals(List.of(spanB), replaced.getSpanProcessors()); + assertTrue(replaced.getLogProcessors().isEmpty()); + assertEquals(List.of(spanA, spanB), options.getSpanProcessors(), "built options must not change"); + } + + @Test + void processorsRejectNull() { + ObservabilityOptions.Builder builder = new ObservabilityOptions.Builder(); + assertThrows(NullPointerException.class, () -> builder.spanProcessors(null)); + assertThrows(NullPointerException.class, () -> builder.spanProcessors(Arrays.asList((SpanProcessor) null))); + assertThrows(NullPointerException.class, () -> builder.spanProcessor(null)); + assertThrows(NullPointerException.class, () -> builder.logProcessors(null)); + assertThrows(NullPointerException.class, () -> builder.logProcessors(Arrays.asList((LogRecordProcessor) null))); + assertThrows(NullPointerException.class, () -> builder.logProcessor(null)); } @Test diff --git a/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/GlobalAttributesTest.java b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/GlobalAttributesTest.java new file mode 100644 index 0000000000..5da959f959 --- /dev/null +++ b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/GlobalAttributesTest.java @@ -0,0 +1,169 @@ +package com.launchdarkly.observability.internal; + +import com.launchdarkly.observability.LDObserve; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class GlobalAttributesTest { + + private static final AttributeKey TENANT = AttributeKey.stringKey("tenant.id"); + private static final AttributeKey REGION = AttributeKey.stringKey("region"); + private static final AttributeKey TIER = AttributeKey.longKey("tier"); + private static final AttributeKey TIER_STRING = AttributeKey.stringKey("tier"); + + @BeforeEach + @AfterEach + void reset() { + GlobalAttributes.resetForTest(); + } + + @Test + void setMergesByKeyAndRemoveDropsKeys() { + LDObserve.setAttributes(Attributes.of(TENANT, "acme", REGION, "us-east-1")); + LDObserve.setAttributes(Attributes.of(REGION, "eu-west-1", TIER, 3L)); + + assertEquals(Attributes.of(TENANT, "acme", REGION, "eu-west-1", TIER, 3L), GlobalAttributes.get()); + + LDObserve.removeAttributes(List.of("region", "not-set")); + assertEquals(Attributes.of(TENANT, "acme", TIER, 3L), GlobalAttributes.get()); + } + + @Test + void keysMatchByNameWhateverTheType() { + LDObserve.setAttributes(Attributes.of(TIER, 3L)); + LDObserve.setAttributes(Attributes.of(TIER_STRING, "gold")); + assertEquals(Attributes.of(TIER_STRING, "gold"), GlobalAttributes.get()); + + LDObserve.removeAttributes(List.of("tier")); + assertTrue(GlobalAttributes.get().isEmpty()); + } + + @Test + void nullAndEmptyArgumentsAreIgnored() { + LDObserve.setAttributes(Attributes.of(TENANT, "acme")); + Attributes before = GlobalAttributes.get(); + + LDObserve.setAttributes(null); + LDObserve.setAttributes(Attributes.empty()); + LDObserve.removeAttributes(null); + LDObserve.removeAttributes(List.of()); + + assertSame(before, GlobalAttributes.get()); + } + + @Test + void mergeKeepsPerCallValues() { + Attributes perCall = Attributes.of(REGION, "per-call"); + assertSame(perCall, GlobalAttributes.merge(perCall), "no globals must not copy the attributes"); + assertEquals(Attributes.empty(), GlobalAttributes.merge(null)); + + LDObserve.setAttributes(Attributes.of(TENANT, "acme", REGION, "global", TIER, 3L)); + + assertEquals(Attributes.of(TENANT, "acme", REGION, "per-call", TIER, 3L), GlobalAttributes.merge(perCall)); + assertEquals(Attributes.of(TENANT, "acme", REGION, "global", TIER_STRING, "gold"), + GlobalAttributes.merge(Attributes.of(TIER_STRING, "gold")), + "a per-call key of another type still wins over the global of the same name"); + assertEquals(GlobalAttributes.get(), GlobalAttributes.merge(null)); + } + + @Test + void concurrentSetsAreAllKept() throws Exception { + int threads = 8; + int perThread = 50; + ExecutorService pool = Executors.newFixedThreadPool(threads); + CountDownLatch start = new CountDownLatch(1); + try { + List> futures = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + int thread = t; + futures.add(pool.submit(() -> { + start.await(); + for (int i = 0; i < perThread; i++) { + LDObserve.setAttributes(Attributes.of(AttributeKey.longKey("k." + thread + "." + i), (long) i)); + } + return null; + })); + } + start.countDown(); + for (Future future : futures) { + future.get(10, TimeUnit.SECONDS); + } + } finally { + pool.shutdownNow(); + } + assertEquals(threads * perThread, GlobalAttributes.get().size()); + } + + @Test + void spanProcessorFillsOnlyKeysTheSpanStartedWithout() { + InMemorySpanExporter exporter = InMemorySpanExporter.create(); + SdkTracerProvider provider = SdkTracerProvider.builder() + .addSpanProcessor(new GlobalAttributesSpanProcessor()) + .addSpanProcessor(SimpleSpanProcessor.create(exporter)) + .build(); + try { + LDObserve.setAttributes(Attributes.of(TENANT, "acme", REGION, "global", TIER, 3L)); + Span early = provider.get("test").spanBuilder("early") + .setAllAttributes(Attributes.of(REGION, "builder", TIER_STRING, "gold")) + .startSpan(); + LDObserve.setAttributes(Attributes.of(TENANT, "changed-later")); + early.end(); + + SpanData span = exporter.getFinishedSpanItems().get(0); + assertEquals("acme", span.getAttributes().get(TENANT), "a span keeps the values current at start"); + assertEquals("builder", span.getAttributes().get(REGION)); + assertEquals("gold", span.getAttributes().get(TIER_STRING)); + assertNull(span.getAttributes().get(TIER), "a key of another type with the same name is not added"); + } finally { + provider.shutdown(); + } + } + + @Test + void logProcessorFillsOnlyKeysTheRecordIsMissing() { + InMemoryLogRecordExporter exporter = InMemoryLogRecordExporter.create(); + SdkLoggerProvider provider = SdkLoggerProvider.builder() + .addLogRecordProcessor(new GlobalAttributesLogProcessor()) + .addLogRecordProcessor(SimpleLogRecordProcessor.create(exporter)) + .build(); + try { + provider.get("test").logRecordBuilder().setBody("no globals yet").emit(); + LDObserve.setAttributes(Attributes.of(TENANT, "acme", REGION, "global")); + provider.get("test").logRecordBuilder() + .setBody("with globals") + .setAllAttributes(Attributes.of(REGION, "per-call")) + .emit(); + + List records = exporter.getFinishedLogRecordItems(); + assertTrue(records.get(0).getAttributes().isEmpty()); + assertEquals(Attributes.of(TENANT, "acme", REGION, "per-call"), records.get(1).getAttributes()); + } finally { + provider.shutdown(); + } + } +} diff --git a/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/PluginUserFlowTest.java b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/PluginUserFlowTest.java index 7c01717ac8..3b79796bd8 100644 --- a/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/PluginUserFlowTest.java +++ b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/PluginUserFlowTest.java @@ -15,10 +15,18 @@ import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.logs.Severity; import io.opentelemetry.api.trace.Span; +import io.opentelemetry.context.Context; import io.opentelemetry.context.Scope; import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.logs.LogRecordProcessor; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter; import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.ReadWriteSpan; +import io.opentelemetry.sdk.trace.ReadableSpan; import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.SpanProcessor; import io.opentelemetry.sdk.trace.data.SpanData; import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; import io.opentelemetry.sdk.trace.samplers.Sampler; @@ -34,11 +42,14 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; import java.util.zip.GZIPInputStream; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -51,6 +62,11 @@ class PluginUserFlowTest { "{\"data\":{\"sampling\":{\"spans\":[{\"name\":{\"matchValue\":\"health-check\"}," + "\"attributes\":[],\"events\":[],\"samplingRatio\":0}],\"logs\":[]}}}"; + private static final AttributeKey TENANT = AttributeKey.stringKey("tenant.id"); + private static final AttributeKey TIER = AttributeKey.longKey("tenant.tier"); + private static final AttributeKey REGION = AttributeKey.stringKey("region"); + private static final AttributeKey STALE = AttributeKey.stringKey("stale"); + private HttpServer server; private final List traces = new CopyOnWriteArrayList<>(); private final List rawTraces = new CopyOnWriteArrayList<>(); @@ -376,7 +392,198 @@ void flagEvaluationExportsAfterManualStart() throws Exception { } } + @Test + void globalAttributesSetBeforeStartReachEverySignal() throws Exception { + LDObserve.setAttributes(Attributes.of(TENANT, "tenant-global-7", TIER, 3L)); + + startCollector(); + InMemorySpanExporter spans = InMemorySpanExporter.create(); + InMemoryLogRecordExporter logRecords = InMemoryLogRecordExporter.create(); + ObservabilityOptions options = optionsBuilder(false) + .spanProcessor(SimpleSpanProcessor.create(spans)) + .logProcessor(SimpleLogRecordProcessor.create(logRecords)) + .build(); + LDConfig config = new LDConfig.Builder() + .offline(true) + .plugins(Components.plugins().setPlugins(List.of(new ObservabilityPlugin(options)))) + .build(); + + try (LDClient client = new LDClient("sdk-test", config)) { + client.boolVariation("checkout-enabled", LDContext.create("user-key-test"), false); + LDObserve.startSpan("user-checkout", Attributes.empty()).end(); + LDObserve.recordError(new IllegalStateException("payment declined"), Attributes.empty()); + LDObserve.recordLog("user signed in", Severity.INFO, Attributes.empty()); + LDObserve.recordCount("orders.placed", 1, Attributes.empty()); + LDObserve.recordMetric("queue.depth", 4.0, Attributes.empty()); + LDObserve.flush(); + } + + List finished = spans.getFinishedSpanItems(); + List names = new ArrayList<>(); + for (SpanData span : finished) { + names.add(span.getName()); + assertEquals("tenant-global-7", span.getAttributes().get(TENANT), span.getName() + " has no global"); + assertEquals(3L, span.getAttributes().get(TIER), span.getName() + " has no global"); + } + assertTrue(names.contains("user-checkout"), "manual span missing: " + names); + assertTrue(names.contains(Constants.ERROR_SPAN_NAME), "error span missing: " + names); + assertTrue(finished.stream().anyMatch(s -> "checkout-enabled".equals( + s.getAttributes().get(AttributeKey.stringKey("feature_flag.key")))), + "flag evaluation span missing: " + names); + + List records = logRecords.getFinishedLogRecordItems(); + assertEquals(1, records.size()); + assertEquals("tenant-global-7", records.get(0).getAttributes().get(TENANT)); + + assertTrue(payloadContains(traces, "tenant-global-7"), "globals were not exported on spans"); + assertTrue(payloadContains(logs, "tenant-global-7"), "globals were not exported on logs"); + assertTrue(payloadContains(metrics, "orders.placed") && payloadContains(metrics, "tenant-global-7"), + "globals were not exported on metrics: " + summarize(metrics)); + } + + @Test + void perCallAttributesWinAndSpansKeepTheGlobalsCurrentAtStart() throws Exception { + startCollector(); + InMemorySpanExporter spans = InMemorySpanExporter.create(); + InMemoryLogRecordExporter logRecords = InMemoryLogRecordExporter.create(); + LDObserve.start("sdk-test", optionsBuilder(true) + .spanProcessor(SimpleSpanProcessor.create(spans)) + .logProcessor(SimpleLogRecordProcessor.create(logRecords)) + .build()); + + LDObserve.setAttributes(Attributes.of(REGION, "region-global", STALE, "stale-global")); + Span early = LDObserve.startSpan("early", Attributes.empty()); + LDObserve.setAttributes(Attributes.of(REGION, "region-updated")); + LDObserve.removeAttributes(List.of("stale")); + early.end(); + LDObserve.startSpan("explicit", Attributes.of(REGION, "region-call")).end(); + LDObserve.startSpan("late", Attributes.empty()).end(); + LDObserve.recordLog("explicit log", Severity.INFO, Attributes.of(REGION, "region-call")); + LDObserve.recordCount("orders.region", 1, Attributes.of(REGION, "region-call")); + LDObserve.flush(); + + SpanData earlySpan = span(spans, "early"); + assertEquals("region-global", earlySpan.getAttributes().get(REGION)); + assertEquals("stale-global", earlySpan.getAttributes().get(STALE)); + SpanData explicitSpan = span(spans, "explicit"); + assertEquals("region-call", explicitSpan.getAttributes().get(REGION)); + assertNull(explicitSpan.getAttributes().get(STALE), "a removed global must not be applied"); + assertEquals("region-updated", span(spans, "late").getAttributes().get(REGION)); + + assertEquals("region-call", logRecords.getFinishedLogRecordItems().get(0).getAttributes().get(REGION)); + + assertTrue(payloadContains(metrics, "region-call"), "per-call metric attribute missing: " + summarize(metrics)); + assertFalse(payloadContains(metrics, "region-updated"), "a global overwrote a per-call metric attribute"); + } + + @Test + void userProcessorsRunInOrderAndTheirChangesAreExported() throws Exception { + List calls = new CopyOnWriteArrayList<>(); + AttributeKey enriched = AttributeKey.stringKey("enriched.by"); + AttributeKey saw = AttributeKey.stringKey("second.saw"); + SpanProcessor first = new OnStartProcessor(span -> { + calls.add("first-span"); + span.setAttribute(enriched, "first-processor"); + }); + SpanProcessor second = new OnStartProcessor(span -> { + calls.add("second-span"); + span.setAttribute(saw, span.getAttribute(enriched) + "|" + span.getAttribute(TENANT)); + }); + LogRecordProcessor firstLog = (context, record) -> { + calls.add("first-log"); + record.setAttribute(enriched, "first-log-processor"); + }; + LogRecordProcessor secondLog = (context, record) -> { + calls.add("second-log"); + Attributes seen = record.toLogRecordData().getAttributes(); + record.setAttribute(saw, seen.get(enriched) + "|" + seen.get(TENANT)); + }; + LDObserve.setAttributes(Attributes.of(TENANT, "tenant-global-7")); + + startCollector(); + LDObserve.start("sdk-test", optionsBuilder(true) + .spanProcessors(List.of(first, second)) + .logProcessors(List.of(firstLog, secondLog)) + .build()); + + LDObserve.startSpan("ordered", Attributes.empty()).end(); + LDObserve.recordLog("ordered log", Severity.INFO, Attributes.empty()); + LDObserve.flush(); + + assertEquals(List.of("first-span", "second-span", "first-log", "second-log"), calls); + assertTrue(payloadContains(traces, "first-processor|tenant-global-7"), + "user span processor changes were not exported: " + summarize(traces)); + assertTrue(payloadContains(logs, "first-log-processor|tenant-global-7"), + "user log processor changes were not exported: " + summarize(logs)); + + // The sampling rules still apply after the user's processors. + assertTrue(waitForSamplingApplied(), "sampling config was not applied"); + assertTrue(waitUntilHealthCheckDropped(), "sampling rule did not drop health-check spans"); + } + + @Test + void throwingUserProcessorsDoNotBreakExport() throws Exception { + InMemorySpanExporter after = InMemorySpanExporter.create(); + startCollector(); + LDObserve.start("sdk-test", optionsBuilder(true) + .spanProcessor(new SafeProcessorTest.ThrowingSpanProcessor()) + .spanProcessor(SimpleSpanProcessor.create(after)) + .logProcessor(new SafeProcessorTest.ThrowingLogProcessor()) + .build()); + + assertDoesNotThrow(() -> LDObserve.startSpan("after-throw", Attributes.empty()).end()); + assertDoesNotThrow(() -> LDObserve.recordLog("log after throw", Severity.INFO, Attributes.empty())); + assertDoesNotThrow(() -> LDObserve.recordError(new IllegalStateException("still recorded"), Attributes.empty())); + assertDoesNotThrow(LDObserve::flush); + + assertTrue(payloadContains(traces, "after-throw"), "span was not exported: " + summarize(traces)); + assertTrue(payloadContains(traces, "still recorded"), "error span was not exported"); + assertTrue(payloadContains(logs, "log after throw"), "log was not exported: " + summarize(logs)); + assertEquals(2, after.getFinishedSpanItems().size(), "processors after the throwing one must still run"); + + assertDoesNotThrow(LDObserve::shutdown); + } + + private static SpanData span(InMemorySpanExporter exporter, String name) { + return exporter.getFinishedSpanItems().stream() + .filter(s -> s.getName().equals(name)) + .findFirst() + .orElseThrow(() -> new AssertionError("no span named " + name)); + } + + /** A span processor that only acts in {@code onStart}. */ + private static final class OnStartProcessor implements SpanProcessor { + private final Consumer onStart; + + OnStartProcessor(Consumer onStart) { + this.onStart = onStart; + } + + @Override + public void onStart(Context parentContext, ReadWriteSpan span) { + onStart.accept(span); + } + + @Override + public boolean isStartRequired() { + return true; + } + + @Override + public void onEnd(ReadableSpan span) { + } + + @Override + public boolean isEndRequired() { + return false; + } + } + private ObservabilityOptions options(boolean manualStart) { + return optionsBuilder(manualStart).build(); + } + + private ObservabilityOptions.Builder optionsBuilder(boolean manualStart) { String base = "http".concat("://") + server.getAddress().getHostString() + ":" + server.getAddress().getPort(); return new ObservabilityOptions.Builder() .serviceName("orders-service") @@ -384,8 +591,7 @@ private ObservabilityOptions options(boolean manualStart) { .environment("test") .otlpEndpoint(base) .backendUrl(base + "/sampling") - .manualStart(manualStart) - .build(); + .manualStart(manualStart); } private void startCollector() throws IOException { diff --git a/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/SafeProcessorTest.java b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/SafeProcessorTest.java new file mode 100644 index 0000000000..c1e486d7e2 --- /dev/null +++ b/sdk/@launchdarkly/observability-java/lib/src/test/java/com/launchdarkly/observability/internal/SafeProcessorTest.java @@ -0,0 +1,202 @@ +package com.launchdarkly.observability.internal; + +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.LogRecordProcessor; +import io.opentelemetry.sdk.logs.ReadWriteLogRecord; +import io.opentelemetry.sdk.logs.SdkLoggerProvider; +import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor; +import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.ReadWriteSpan; +import io.opentelemetry.sdk.trace.ReadableSpan; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.SpanProcessor; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SafeProcessorTest { + + @Test + void throwingSpanProcessorDoesNotStopTheSpanOrLaterProcessors() { + InMemorySpanExporter exporter = InMemorySpanExporter.create(); + SdkTracerProvider provider = SdkTracerProvider.builder() + .addSpanProcessor(new SafeSpanProcessor(new ThrowingSpanProcessor())) + .addSpanProcessor(SimpleSpanProcessor.create(exporter)) + .build(); + + assertDoesNotThrow(() -> provider.get("test").spanBuilder("survives").startSpan().end()); + assertEquals("survives", exporter.getFinishedSpanItems().get(0).getName()); + + CompletableResultCode flush = assertDoesNotThrow(provider::forceFlush); + assertFalse(flush.join(5, TimeUnit.SECONDS).isSuccess()); + CompletableResultCode shutdown = assertDoesNotThrow(provider::shutdown); + assertFalse(shutdown.join(5, TimeUnit.SECONDS).isSuccess()); + } + + @Test + void spanProcessorFlushAndShutdownAreForwarded() { + CountingSpanProcessor counting = new CountingSpanProcessor(); + SdkTracerProvider provider = SdkTracerProvider.builder() + .addSpanProcessor(new SafeSpanProcessor(counting)) + .build(); + + provider.get("test").spanBuilder("counted").startSpan().end(); + assertTrue(provider.forceFlush().join(5, TimeUnit.SECONDS).isSuccess()); + assertTrue(provider.shutdown().join(5, TimeUnit.SECONDS).isSuccess()); + + assertEquals(1, counting.starts.get()); + assertEquals(1, counting.ends.get()); + assertEquals(1, counting.flushes.get()); + assertEquals(1, counting.shutdowns.get()); + } + + @Test + void throwingLogProcessorDoesNotStopTheRecordOrLaterProcessors() { + InMemoryLogRecordExporter exporter = InMemoryLogRecordExporter.create(); + SdkLoggerProvider provider = SdkLoggerProvider.builder() + .addLogRecordProcessor(new SafeLogRecordProcessor(new ThrowingLogProcessor())) + .addLogRecordProcessor(SimpleLogRecordProcessor.create(exporter)) + .build(); + + assertDoesNotThrow(() -> provider.get("test").logRecordBuilder().setBody("survives").emit()); + assertEquals("survives", exporter.getFinishedLogRecordItems().get(0).getBodyValue().asString()); + + CompletableResultCode flush = assertDoesNotThrow(provider::forceFlush); + assertFalse(flush.join(5, TimeUnit.SECONDS).isSuccess()); + CompletableResultCode shutdown = assertDoesNotThrow(provider::shutdown); + assertFalse(shutdown.join(5, TimeUnit.SECONDS).isSuccess()); + } + + @Test + void logProcessorFlushAndShutdownAreForwarded() { + AtomicInteger emits = new AtomicInteger(); + AtomicInteger flushes = new AtomicInteger(); + AtomicInteger shutdowns = new AtomicInteger(); + LogRecordProcessor counting = new LogRecordProcessor() { + @Override + public void onEmit(Context context, ReadWriteLogRecord logRecord) { + emits.incrementAndGet(); + } + + @Override + public CompletableResultCode forceFlush() { + flushes.incrementAndGet(); + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + shutdowns.incrementAndGet(); + return CompletableResultCode.ofSuccess(); + } + }; + SdkLoggerProvider provider = SdkLoggerProvider.builder() + .addLogRecordProcessor(new SafeLogRecordProcessor(counting)) + .build(); + + provider.get("test").logRecordBuilder().setBody("counted").emit(); + assertTrue(provider.forceFlush().join(5, TimeUnit.SECONDS).isSuccess()); + assertTrue(provider.shutdown().join(5, TimeUnit.SECONDS).isSuccess()); + + assertEquals(1, emits.get()); + assertEquals(1, flushes.get()); + assertEquals(1, shutdowns.get()); + } + + static final class ThrowingSpanProcessor implements SpanProcessor { + @Override + public void onStart(Context parentContext, ReadWriteSpan span) { + throw new IllegalStateException("onStart failed"); + } + + @Override + public boolean isStartRequired() { + return true; + } + + @Override + public void onEnd(ReadableSpan span) { + throw new IllegalStateException("onEnd failed"); + } + + @Override + public boolean isEndRequired() { + return true; + } + + @Override + public CompletableResultCode forceFlush() { + throw new IllegalStateException("forceFlush failed"); + } + + @Override + public CompletableResultCode shutdown() { + throw new IllegalStateException("shutdown failed"); + } + } + + static final class ThrowingLogProcessor implements LogRecordProcessor { + @Override + public void onEmit(Context context, ReadWriteLogRecord logRecord) { + throw new IllegalStateException("onEmit failed"); + } + + @Override + public CompletableResultCode forceFlush() { + throw new IllegalStateException("forceFlush failed"); + } + + @Override + public CompletableResultCode shutdown() { + throw new IllegalStateException("shutdown failed"); + } + } + + private static final class CountingSpanProcessor implements SpanProcessor { + final AtomicInteger starts = new AtomicInteger(); + final AtomicInteger ends = new AtomicInteger(); + final AtomicInteger flushes = new AtomicInteger(); + final AtomicInteger shutdowns = new AtomicInteger(); + + @Override + public void onStart(Context parentContext, ReadWriteSpan span) { + starts.incrementAndGet(); + } + + @Override + public boolean isStartRequired() { + return true; + } + + @Override + public void onEnd(ReadableSpan span) { + ends.incrementAndGet(); + } + + @Override + public boolean isEndRequired() { + return true; + } + + @Override + public CompletableResultCode forceFlush() { + flushes.incrementAndGet(); + return CompletableResultCode.ofSuccess(); + } + + @Override + public CompletableResultCode shutdown() { + shutdowns.incrementAndGet(); + return CompletableResultCode.ofSuccess(); + } + } +} From be24b27ffd7ab650f1946d3bd6a3602dbf4ddec8 Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Wed, 28 Oct 2026 17:17:24 -0700 Subject: [PATCH 2/2] chore(java): add global attributes repro to e2e app Co-authored-by: Cursor --- e2e/java-plugin-spring-boot/README.md | 20 ++++++++++ .../example/springboot/Application.java | 1 + .../springboot/GlobalAttributesRepro.java | 39 +++++++++++++++++++ .../springboot/RequestTelemetryFilter.java | 9 ++++- 4 files changed, 67 insertions(+), 2 deletions(-) create mode 100644 e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/GlobalAttributesRepro.java diff --git a/e2e/java-plugin-spring-boot/README.md b/e2e/java-plugin-spring-boot/README.md index 493c479115..57615fd4cd 100644 --- a/e2e/java-plugin-spring-boot/README.md +++ b/e2e/java-plugin-spring-boot/README.md @@ -45,6 +45,26 @@ Environment the app honours: | `SERVICE_VERSION` | Reported as `service.version`, default `dev`. | | `LD_BASE_URI`, `LD_STREAM_URI`, `LD_EVENTS_URI` | Flag SDK polling, streaming and events URLs for a non-default LaunchDarkly instance. Used only when all three are set. | +### Global attributes repro + +`Application.main` calls `GlobalAttributesRepro.applyGlobalAttributes()` +before Spring starts, so `LDObserve.setAttributes` runs before the +`LDClient` and the plugin exist. It sets `utm_campaign=spring_promo` and +`utm_source=email` on all telemetry. For every request, `RequestTelemetryFilter` +prints what to look for in LaunchDarkly: + +``` +[#1] expect utm_campaign=spring_promo on this row (span GET /rolldice) +[#1] expect utm_campaign=spring_promo on this row (metric http.server.requests, route /rolldice) +[#1] expect utm_campaign=spring_promo on this row (log "GET /rolldice -> 200") +[#1] expect utm_source=per_call (per-call wins over global) (log "GET /rolldice -> 200") +``` + +The request log line passes `utm_source=per_call` itself, so it must show +`per_call` rather than the global `email`. Flag-evaluation spans and errors +from `/error` carry the globals too. The tests boot the context without +`main`, so they run without the globals. + `make build` produces the executable jar in `build/libs/`. `make test` boots the application against a local OTLP collector and sampling endpoint (`ApplicationTelemetryTest`), drives every route, and checks the exported diff --git a/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/Application.java b/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/Application.java index 3a242a293b..0fb3e82df7 100644 --- a/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/Application.java +++ b/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/Application.java @@ -24,6 +24,7 @@ @SpringBootApplication public class Application { public static void main(String[] args) { + GlobalAttributesRepro.applyGlobalAttributes(); SpringApplication.run(Application.class, args); } } diff --git a/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/GlobalAttributesRepro.java b/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/GlobalAttributesRepro.java new file mode 100644 index 0000000000..77a40c743e --- /dev/null +++ b/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/GlobalAttributesRepro.java @@ -0,0 +1,39 @@ +package com.launchdarkly.observability.example.springboot; + +import com.launchdarkly.observability.LDObserve; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; + +/** + * Reproduces "global attributes (e.g. UTM params) on all telemetry": the UTM + * attributes are set once at startup, before the plugin exists, and every + * request's span, log and metrics must carry them. The {@code [#1]} lines on + * stdout say what to look for on the matching row in LaunchDarkly. + */ +final class GlobalAttributesRepro { + static final AttributeKey UTM_CAMPAIGN = AttributeKey.stringKey("utm_campaign"); + static final AttributeKey UTM_SOURCE = AttributeKey.stringKey("utm_source"); + + static final Attributes GLOBAL_ATTRIBUTES = Attributes.of( + UTM_CAMPAIGN, "spring_promo", + UTM_SOURCE, "email"); + + /** Passed on the request log line only; it must win over the global {@code utm_source}. */ + static final Attributes PER_CALL_ATTRIBUTES = Attributes.of(UTM_SOURCE, "per_call"); + + private GlobalAttributesRepro() {} + + /** Call before the Spring context (and so the LDClient and plugin) starts. */ + static void applyGlobalAttributes() { + LDObserve.setAttributes(GLOBAL_ATTRIBUTES); + System.out.println("[#1] global attributes set before the LaunchDarkly client starts: " + GLOBAL_ATTRIBUTES); + } + + static void expectGlobal(String row) { + System.out.println("[#1] expect utm_campaign=spring_promo on this row (" + row + ")"); + } + + static void expectPerCallWins(String row) { + System.out.println("[#1] expect utm_source=per_call (per-call wins over global) (" + row + ")"); + } +} diff --git a/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/RequestTelemetryFilter.java b/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/RequestTelemetryFilter.java index 19b7646996..e6d8a61cbe 100644 --- a/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/RequestTelemetryFilter.java +++ b/e2e/java-plugin-spring-boot/src/main/java/com/launchdarkly/observability/example/springboot/RequestTelemetryFilter.java @@ -47,13 +47,18 @@ protected void doFilterInternal(HttpServletRequest request, HttpServletResponse } span.setAttribute(STATUS, (long) status); span.end(); + GlobalAttributesRepro.expectGlobal("span " + request.getMethod() + " " + route); double millis = TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - started) / 1000.0; Attributes attrs = Attributes.of(ROUTE, route, STATUS, (long) status); LDObserve.recordHistogram("http.server.request.duration_ms", millis, attrs); LDObserve.recordIncr("http.server.requests", attrs); - LDObserve.recordLog(request.getMethod() + " " + route + " -> " + status, - status >= 500 ? Severity.ERROR : Severity.INFO, attrs); + GlobalAttributesRepro.expectGlobal("metric http.server.requests, route " + route); + String message = request.getMethod() + " " + route + " -> " + status; + LDObserve.recordLog(message, status >= 500 ? Severity.ERROR : Severity.INFO, + attrs.toBuilder().putAll(GlobalAttributesRepro.PER_CALL_ATTRIBUTES).build()); + GlobalAttributesRepro.expectGlobal("log \"" + message + "\""); + GlobalAttributesRepro.expectPerCallWins("log \"" + message + "\""); } } }