From de8a376c41b75282b7a79ce5f87b95c9a741e16f Mon Sep 17 00:00:00 2001 From: udaysagar2177 Date: Tue, 7 Apr 2026 01:12:56 -0700 Subject: [PATCH 1/5] Add FilteringSpanExporter with composable SpanFilter and TraceFilter interfaces Adds a trace-level filtering span exporter to the processors module. FilteringSpanExporter wraps a delegate SpanExporter and only forwards spans belonging to traces that match at least one configured filter. Filtering is trace-aware: if any filter matches, all spans sharing that trace ID are exported together. Two filter interfaces enable composable filtering: - SpanFilter: per-span evaluation (e.g., error status, slow duration) - TraceFilter: batch-level evaluation (e.g., overall trace wall-clock duration) Built-in implementations: - ErrorSpanFilter: keeps traces with error spans - DurationSpanFilter: keeps traces with slow individual spans - TraceDurationFilter: keeps traces with long wall-clock duration Optional Meter parameter emits a dropped-span counter with reason attribute. --- processors/README.md | 41 +++ .../contrib/filter/DurationSpanFilter.java | 34 ++ .../contrib/filter/ErrorSpanFilter.java | 18 ++ .../contrib/filter/FilteringSpanExporter.java | 164 ++++++++++ .../contrib/filter/SpanFilter.java | 26 ++ .../contrib/filter/TraceDurationFilter.java | 44 +++ .../contrib/filter/TraceFilter.java | 30 ++ .../filter/DurationSpanFilterTest.java | 48 +++ .../contrib/filter/ErrorSpanFilterTest.java | 46 +++ .../filter/FilteringSpanExporterTest.java | 293 ++++++++++++++++++ .../filter/TraceDurationFilterTest.java | 77 +++++ 11 files changed, 821 insertions(+) create mode 100644 processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java create mode 100644 processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java create mode 100644 processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java create mode 100644 processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java create mode 100644 processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java create mode 100644 processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java create mode 100644 processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java create mode 100644 processors/src/test/java/io/opentelemetry/contrib/filter/ErrorSpanFilterTest.java create mode 100644 processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java create mode 100644 processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java diff --git a/processors/README.md b/processors/README.md index 795fe61ced..a8de0a1e5d 100644 --- a/processors/README.md +++ b/processors/README.md @@ -32,6 +32,47 @@ logger_provider: `FilteringLogRecordProcessor` is a `LogRecordProcessor` that only keep logs based on a predicate +## Filtering Span Exporter + +`FilteringSpanExporter` is a `SpanExporter` wrapper that filters spans at the trace level before delegating to the underlying exporter. Filtering is composable via two interfaces: + +- `SpanFilter` - evaluates individual spans (e.g., error status, slow duration) +- `TraceFilter` - evaluates all spans belonging to a trace within the batch (e.g., overall trace wall-clock duration) + +A trace is kept if any `SpanFilter` matches any of its spans, OR any `TraceFilter` matches the trace's span group. All spans sharing a trace ID are exported together. + +Built-in filters: +- `ErrorSpanFilter` - keeps traces containing any span with error status +- `DurationSpanFilter` - keeps traces containing any span exceeding a duration threshold +- `TraceDurationFilter` - keeps traces whose wall-clock duration (max end - min start) exceeds a threshold + +Usage: + +```java +SpanExporter delegate = OtlpGrpcSpanExporter.getDefault(); + +// Export traces with errors, individual spans > 2s, or trace duration > 10s +SpanExporter filtering = new FilteringSpanExporter( + delegate, + Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(2000)), + Collections.singletonList(new TraceDurationFilter(10000))); + +// Custom filters +SpanFilter nameFilter = span -> span.getName().contains("important"); +SpanExporter custom = new FilteringSpanExporter( + delegate, + Collections.singletonList(nameFilter), + Collections.emptyList()); + +// Optionally pass a Meter to emit dropped-span metrics +Meter meter = openTelemetry.getMeter("my-service"); +SpanExporter withMetrics = new FilteringSpanExporter( + delegate, + Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(2000)), + Collections.singletonList(new TraceDurationFilter(10000)), + meter); +``` + ## Component owners - [Cesar Munoz](https://github.com/LikeTheSalad), Elastic diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java new file mode 100644 index 0000000000..9ff3b374d7 --- /dev/null +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java @@ -0,0 +1,34 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.concurrent.TimeUnit; + +/** + * A {@link SpanFilter} that keeps traces containing any span whose duration exceeds a configurable + * threshold. + */ +public final class DurationSpanFilter implements SpanFilter { + + private final long thresholdNanos; + + /** + * Creates a new {@code DurationSpanFilter}. + * + * @param thresholdMs the duration threshold in milliseconds; spans with duration strictly greater + * than this are considered interesting + */ + public DurationSpanFilter(long thresholdMs) { + this.thresholdNanos = TimeUnit.MILLISECONDS.toNanos(thresholdMs); + } + + @Override + public boolean shouldKeep(SpanData spanData) { + long durationNanos = spanData.getEndEpochNanos() - spanData.getStartEpochNanos(); + return durationNanos > thresholdNanos; + } +} diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java new file mode 100644 index 0000000000..6e28be90f0 --- /dev/null +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java @@ -0,0 +1,18 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.sdk.trace.data.SpanData; + +/** A {@link SpanFilter} that keeps traces containing any span with {@link StatusCode#ERROR}. */ +public final class ErrorSpanFilter implements SpanFilter { + + @Override + public boolean shouldKeep(SpanData spanData) { + return spanData.getStatus().getStatusCode() == StatusCode.ERROR; + } +} diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java new file mode 100644 index 0000000000..7a8394a155 --- /dev/null +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java @@ -0,0 +1,164 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nullable; + +/** + * A {@link SpanExporter} wrapper that filters spans before delegating to the underlying exporter. + * Filtering operates at the trace level: if any filter matches, all spans sharing that trace ID are + * exported together. + * + *

Two types of filters are supported: + * + *

+ * + *

A trace is kept if any {@code SpanFilter} matches any of its spans, OR any {@code TraceFilter} + * matches the trace's span group. + */ +public final class FilteringSpanExporter implements SpanExporter { + + private static final AttributeKey REASON_KEY = AttributeKey.stringKey("reason"); + + private final SpanExporter delegate; + private final List spanFilters; + private final List traceFilters; + @Nullable private final LongCounter droppedSpansCounter; + + /** + * Creates a new {@code FilteringSpanExporter}. + * + * @param delegate the exporter to delegate to for spans that pass filtering + * @param spanFilters per-span filters; a trace is kept if any filter matches any span + * @param traceFilters batch-level filters; a trace is kept if any filter matches the trace group + * @param meter optional {@link Meter} for emitting dropped-span metrics; pass {@code null} to + * disable metrics + */ + public FilteringSpanExporter( + SpanExporter delegate, + List spanFilters, + List traceFilters, + @Nullable Meter meter) { + this.delegate = delegate; + this.spanFilters = Collections.unmodifiableList(new ArrayList<>(spanFilters)); + this.traceFilters = Collections.unmodifiableList(new ArrayList<>(traceFilters)); + if (meter != null) { + this.droppedSpansCounter = + meter + .counterBuilder("filtering.span.exporter.dropped") + .setDescription("Number of spans dropped by the filtering span exporter") + .build(); + } else { + this.droppedSpansCounter = null; + } + } + + /** + * Creates a new {@code FilteringSpanExporter} without metrics. + * + * @param delegate the exporter to delegate to for spans that pass filtering + * @param spanFilters per-span filters + * @param traceFilters batch-level filters + */ + public FilteringSpanExporter( + SpanExporter delegate, List spanFilters, List traceFilters) { + this(delegate, spanFilters, traceFilters, null); + } + + @Override + public CompletableResultCode export(Collection spans) { + // Group spans by trace ID and evaluate span-level filters in a single pass + Set interestingTraceIds = new HashSet<>(); + Map> spansByTrace = new HashMap<>(); + + for (SpanData span : spans) { + String traceId = span.getSpanContext().getTraceId(); + + List traceSpans = spansByTrace.get(traceId); + if (traceSpans == null) { + traceSpans = new ArrayList<>(); + spansByTrace.put(traceId, traceSpans); + } + traceSpans.add(span); + + // Check span-level filters + if (!interestingTraceIds.contains(traceId)) { + for (SpanFilter filter : spanFilters) { + if (filter.shouldKeep(span)) { + interestingTraceIds.add(traceId); + break; + } + } + } + } + + // Evaluate trace-level filters + if (!traceFilters.isEmpty()) { + for (Map.Entry> entry : spansByTrace.entrySet()) { + String traceId = entry.getKey(); + if (!interestingTraceIds.contains(traceId)) { + for (TraceFilter filter : traceFilters) { + if (filter.shouldKeep(traceId, entry.getValue())) { + interestingTraceIds.add(traceId); + break; + } + } + } + } + } + + // Collect filtered spans + List filtered = new ArrayList<>(); + long droppedCount = 0; + + for (SpanData span : spans) { + if (interestingTraceIds.contains(span.getSpanContext().getTraceId())) { + filtered.add(span); + } else { + droppedCount++; + } + } + + if (droppedSpansCounter != null && droppedCount > 0) { + droppedSpansCounter.add(droppedCount, Attributes.of(REASON_KEY, "not_interesting")); + } + + if (filtered.isEmpty()) { + return CompletableResultCode.ofSuccess(); + } + + return delegate.export(filtered); + } + + @Override + public CompletableResultCode flush() { + return delegate.flush(); + } + + @Override + public CompletableResultCode shutdown() { + return delegate.shutdown(); + } +} diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java new file mode 100644 index 0000000000..272a09a8ba --- /dev/null +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java @@ -0,0 +1,26 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import io.opentelemetry.sdk.trace.data.SpanData; + +/** + * A filter that evaluates individual spans to determine if their containing trace should be + * exported. Used by {@link FilteringSpanExporter} to make per-span keep/drop decisions. + * + *

If any {@code SpanFilter} returns {@code true} for any span in a trace, all spans sharing that + * trace ID are exported. + */ +public interface SpanFilter { + + /** + * Evaluates whether the given span is interesting enough to keep its entire trace. + * + * @param spanData the span to evaluate + * @return {@code true} if this span should cause its trace to be exported + */ + boolean shouldKeep(SpanData spanData); +} diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java new file mode 100644 index 0000000000..37cfa6f646 --- /dev/null +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java @@ -0,0 +1,44 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.Collection; +import java.util.concurrent.TimeUnit; + +/** + * A {@link TraceFilter} that keeps traces whose overall wall-clock duration (max end - min start + * across all spans in the batch sharing a trace ID) exceeds a configurable threshold. + */ +public final class TraceDurationFilter implements TraceFilter { + + private final long thresholdNanos; + + /** + * Creates a new {@code TraceDurationFilter}. + * + * @param thresholdMs the trace duration threshold in milliseconds; traces with wall-clock + * duration strictly greater than this are considered interesting + */ + public TraceDurationFilter(long thresholdMs) { + this.thresholdNanos = TimeUnit.MILLISECONDS.toNanos(thresholdMs); + } + + @Override + public boolean shouldKeep(String traceId, Collection spans) { + long minStart = Long.MAX_VALUE; + long maxEnd = Long.MIN_VALUE; + for (SpanData span : spans) { + if (span.getStartEpochNanos() < minStart) { + minStart = span.getStartEpochNanos(); + } + if (span.getEndEpochNanos() > maxEnd) { + maxEnd = span.getEndEpochNanos(); + } + } + return (maxEnd - minStart) > thresholdNanos; + } +} diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java new file mode 100644 index 0000000000..a646df9b46 --- /dev/null +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java @@ -0,0 +1,30 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.Collection; + +/** + * A filter that evaluates all spans belonging to a single trace within an export batch to determine + * if the trace should be exported. Used by {@link FilteringSpanExporter} for decisions that require + * batch-level context (e.g., overall trace wall-clock duration). + * + *

If any {@code TraceFilter} returns {@code true} for a trace, all spans sharing that trace ID + * are exported. + */ +public interface TraceFilter { + + /** + * Evaluates whether the given trace (represented as all spans sharing a trace ID within the + * current batch) should be exported. + * + * @param traceId the trace ID + * @param spans all spans in the current batch belonging to this trace + * @return {@code true} if this trace should be exported + */ + boolean shouldKeep(String traceId, Collection spans); +} diff --git a/processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java b/processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java new file mode 100644 index 0000000000..a068d13733 --- /dev/null +++ b/processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java @@ -0,0 +1,48 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +class DurationSpanFilterTest { + + private static final long THRESHOLD_MS = 2000L; + private static final long THRESHOLD_NANOS = TimeUnit.MILLISECONDS.toNanos(THRESHOLD_MS); + + private final DurationSpanFilter filter = new DurationSpanFilter(THRESHOLD_MS); + + @Test + void spanOverThresholdIsKept() { + SpanData span = spanWithDurationNanos(THRESHOLD_NANOS + 1); + assertThat(filter.shouldKeep(span)).isTrue(); + } + + @Test + void spanExactlyAtThresholdIsDropped() { + SpanData span = spanWithDurationNanos(THRESHOLD_NANOS); + assertThat(filter.shouldKeep(span)).isFalse(); + } + + @Test + void spanUnderThresholdIsDropped() { + SpanData span = spanWithDurationNanos(TimeUnit.MILLISECONDS.toNanos(500)); + assertThat(filter.shouldKeep(span)).isFalse(); + } + + private static SpanData spanWithDurationNanos(long durationNanos) { + SpanData span = mock(SpanData.class); + long startNanos = TimeUnit.MILLISECONDS.toNanos(1_000_000_000L); + when(span.getStartEpochNanos()).thenReturn(startNanos); + when(span.getEndEpochNanos()).thenReturn(startNanos + durationNanos); + return span; + } +} diff --git a/processors/src/test/java/io/opentelemetry/contrib/filter/ErrorSpanFilterTest.java b/processors/src/test/java/io/opentelemetry/contrib/filter/ErrorSpanFilterTest.java new file mode 100644 index 0000000000..c7b8864a91 --- /dev/null +++ b/processors/src/test/java/io/opentelemetry/contrib/filter/ErrorSpanFilterTest.java @@ -0,0 +1,46 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.data.StatusData; +import org.junit.jupiter.api.Test; + +class ErrorSpanFilterTest { + + private final ErrorSpanFilter filter = new ErrorSpanFilter(); + + @Test + void errorSpanIsKept() { + SpanData span = spanWithStatus(StatusCode.ERROR); + assertThat(filter.shouldKeep(span)).isTrue(); + } + + @Test + void okSpanIsDropped() { + SpanData span = spanWithStatus(StatusCode.OK); + assertThat(filter.shouldKeep(span)).isFalse(); + } + + @Test + void unsetSpanIsDropped() { + SpanData span = spanWithStatus(StatusCode.UNSET); + assertThat(filter.shouldKeep(span)).isFalse(); + } + + private static SpanData spanWithStatus(StatusCode statusCode) { + SpanData span = mock(SpanData.class); + StatusData status = mock(StatusData.class); + when(status.getStatusCode()).thenReturn(statusCode); + when(span.getStatus()).thenReturn(status); + return span; + } +} diff --git a/processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java b/processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java new file mode 100644 index 0000000000..118019eace --- /dev/null +++ b/processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java @@ -0,0 +1,293 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyCollection; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.LongCounterBuilder; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.data.StatusData; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class FilteringSpanExporterTest { + + private static final long SPAN_THRESHOLD_MS = 2000L; + private static final long TRACE_THRESHOLD_MS = 10000L; + + private SpanExporter delegate; + private FilteringSpanExporter exporter; + + @BeforeEach + void setUp() { + delegate = mock(SpanExporter.class); + when(delegate.export(anyCollection())).thenReturn(CompletableResultCode.ofSuccess()); + exporter = + new FilteringSpanExporter( + delegate, + Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(SPAN_THRESHOLD_MS)), + Collections.singletonList(new TraceDurationFilter(TRACE_THRESHOLD_MS))); + } + + // --- Trace grouping: if one span is interesting, all siblings are kept --- + + @Test + void uninterestingSpansAreDropped() { + SpanData fastSpan = createSpan("trace-1", "span-1", StatusCode.OK, 0, 500); + + CompletableResultCode result = exporter.export(Collections.singletonList(fastSpan)); + + assertThat(result.isSuccess()).isTrue(); + verify(delegate, never()).export(anyCollection()); + } + + @Test + void traceGrouping_interestingSpanKeepsSiblings() { + String traceId = "trace-1"; + SpanData slowSpan = createSpan(traceId, "span-1", StatusCode.OK, 0, 3000); + SpanData fastSibling = createSpan(traceId, "span-2", StatusCode.OK, 0, 100); + + exporter.export(Arrays.asList(slowSpan, fastSibling)); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(Collection.class); + verify(delegate).export(captor.capture()); + assertThat(captor.getValue()).containsExactlyInAnyOrder(slowSpan, fastSibling); + } + + @Test + void mixedTraces_onlyInterestingTracesKept() { + SpanData slowSpan = createSpan("trace-slow", "span-1", StatusCode.OK, 0, 3000); + SpanData slowSibling = createSpan("trace-slow", "span-2", StatusCode.OK, 0, 200); + SpanData fastSpan = createSpan("trace-fast", "span-3", StatusCode.OK, 0, 500); + + exporter.export(Arrays.asList(slowSpan, slowSibling, fastSpan)); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(Collection.class); + verify(delegate).export(captor.capture()); + assertThat(captor.getValue()).containsExactlyInAnyOrder(slowSpan, slowSibling); + } + + @Test + void traceKeptByTraceDurationFilter() { + SpanData a1 = createSpan("trace-a", "span-1", StatusCode.OK, 0, 200); + SpanData a2 = createSpan("trace-a", "span-2", StatusCode.OK, 14800, 200); + SpanData b1 = createSpan("trace-b", "span-3", StatusCode.OK, 0, 300); + + exporter.export(Arrays.asList(a1, a2, b1)); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(Collection.class); + verify(delegate).export(captor.capture()); + assertThat(captor.getValue()).containsExactlyInAnyOrder(a1, a2); + } + + // --- Composability: different filter combinations --- + + @Test + void onlySpanFilters_noTraceFilters() { + FilteringSpanExporter spanOnly = + new FilteringSpanExporter( + delegate, + Collections.singletonList(new ErrorSpanFilter()), + Collections.emptyList()); + + SpanData errorSpan = createSpan("trace-1", "span-1", StatusCode.ERROR, 0, 100); + SpanData fastSpan = createSpan("trace-2", "span-2", StatusCode.OK, 0, 100); + + spanOnly.export(Arrays.asList(errorSpan, fastSpan)); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(Collection.class); + verify(delegate).export(captor.capture()); + assertThat(captor.getValue()).containsExactly(errorSpan); + } + + @Test + void onlyTraceFilters_noSpanFilters() { + FilteringSpanExporter traceOnly = + new FilteringSpanExporter( + delegate, + Collections.emptyList(), + Collections.singletonList(new TraceDurationFilter(TRACE_THRESHOLD_MS))); + + SpanData earlySpan = createSpan("trace-1", "span-1", StatusCode.OK, 0, 500); + SpanData lateSpan = createSpan("trace-1", "span-2", StatusCode.OK, 11500, 500); + SpanData errorSpan = createSpan("trace-2", "span-3", StatusCode.ERROR, 0, 100); + + traceOnly.export(Arrays.asList(earlySpan, lateSpan, errorSpan)); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(Collection.class); + verify(delegate).export(captor.capture()); + assertThat(captor.getValue()).containsExactlyInAnyOrder(earlySpan, lateSpan); + } + + @Test + void customSpanFilter() { + SpanFilter nameFilter = span -> span.getName().contains("important"); + FilteringSpanExporter custom = + new FilteringSpanExporter( + delegate, Collections.singletonList(nameFilter), Collections.emptyList()); + + SpanData important = createNamedSpan("trace-1", "important-operation", StatusCode.OK, 0, 100); + SpanData boring = createNamedSpan("trace-2", "boring-operation", StatusCode.OK, 0, 100); + + custom.export(Arrays.asList(important, boring)); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(Collection.class); + verify(delegate).export(captor.capture()); + assertThat(captor.getValue()).containsExactly(important); + } + + // --- Metrics --- + + @Test + void droppedSpanMetricIsEmittedWhenMeterProvided() { + Meter meter = mock(Meter.class); + LongCounterBuilder counterBuilder = mock(LongCounterBuilder.class); + LongCounter counter = mock(LongCounter.class); + when(meter.counterBuilder("filtering.span.exporter.dropped")).thenReturn(counterBuilder); + when(counterBuilder.setDescription("Number of spans dropped by the filtering span exporter")) + .thenReturn(counterBuilder); + when(counterBuilder.build()).thenReturn(counter); + + FilteringSpanExporter exporterWithMetrics = + new FilteringSpanExporter( + delegate, + Collections.singletonList(new DurationSpanFilter(SPAN_THRESHOLD_MS)), + Collections.emptyList(), + meter); + + SpanData fastSpan = createSpan("trace-1", "span-1", StatusCode.OK, 0, 500); + exporterWithMetrics.export(Collections.singletonList(fastSpan)); + + verify(counter).add(eq(1L), any()); + } + + @Test + void noMetricEmittedWhenNoSpansDropped() { + Meter meter = mock(Meter.class); + LongCounterBuilder counterBuilder = mock(LongCounterBuilder.class); + LongCounter counter = mock(LongCounter.class); + when(meter.counterBuilder("filtering.span.exporter.dropped")).thenReturn(counterBuilder); + when(counterBuilder.setDescription("Number of spans dropped by the filtering span exporter")) + .thenReturn(counterBuilder); + when(counterBuilder.build()).thenReturn(counter); + + FilteringSpanExporter exporterWithMetrics = + new FilteringSpanExporter( + delegate, + Collections.singletonList(new DurationSpanFilter(SPAN_THRESHOLD_MS)), + Collections.emptyList(), + meter); + + SpanData slowSpan = createSpan("trace-1", "span-1", StatusCode.OK, 0, 3000); + exporterWithMetrics.export(Collections.singletonList(slowSpan)); + + verify(counter, never()).add(anyLong(), any()); + } + + // --- Delegation --- + + @Test + void flushDelegatesToWrapped() { + when(delegate.flush()).thenReturn(CompletableResultCode.ofSuccess()); + + CompletableResultCode result = exporter.flush(); + + assertThat(result.isSuccess()).isTrue(); + verify(delegate).flush(); + } + + @Test + void shutdownDelegatesToWrapped() { + when(delegate.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + + CompletableResultCode result = exporter.shutdown(); + + assertThat(result.isSuccess()).isTrue(); + verify(delegate).shutdown(); + } + + @Test + void emptyBatchReturnsSuccess() { + CompletableResultCode result = exporter.export(Collections.emptyList()); + + assertThat(result.isSuccess()).isTrue(); + verify(delegate, never()).export(anyCollection()); + } + + // --- Helpers --- + + private static SpanData createSpan( + String traceId, String spanId, StatusCode statusCode, long startOffsetMs, long durationMs) { + return createSpanNanos( + traceId, + spanId, + statusCode, + TimeUnit.MILLISECONDS.toNanos(startOffsetMs), + TimeUnit.MILLISECONDS.toNanos(durationMs)); + } + + private static SpanData createNamedSpan( + String traceId, String name, StatusCode statusCode, long startOffsetMs, long durationMs) { + SpanData span = + createSpanNanos( + traceId, + "span-id", + statusCode, + TimeUnit.MILLISECONDS.toNanos(startOffsetMs), + TimeUnit.MILLISECONDS.toNanos(durationMs)); + when(span.getName()).thenReturn(name); + return span; + } + + private static SpanData createSpanNanos( + String traceId, + String spanId, + StatusCode statusCode, + long startOffsetNanos, + long durationNanos) { + SpanData span = mock(SpanData.class); + SpanContext context = mock(SpanContext.class); + StatusData status = mock(StatusData.class); + + when(context.getTraceId()).thenReturn(traceId); + when(context.getSpanId()).thenReturn(spanId); + when(span.getSpanContext()).thenReturn(context); + when(status.getStatusCode()).thenReturn(statusCode); + when(span.getStatus()).thenReturn(status); + + long baseNanos = TimeUnit.MILLISECONDS.toNanos(1_000_000_000L); + long startNanos = baseNanos + startOffsetNanos; + when(span.getStartEpochNanos()).thenReturn(startNanos); + when(span.getEndEpochNanos()).thenReturn(startNanos + durationNanos); + + return span; + } +} diff --git a/processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java b/processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java new file mode 100644 index 0000000000..50c94361f3 --- /dev/null +++ b/processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java @@ -0,0 +1,77 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.contrib.filter; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.opentelemetry.sdk.trace.data.SpanData; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +class TraceDurationFilterTest { + + private static final long THRESHOLD_MS = 10000L; + private static final long THRESHOLD_NANOS = TimeUnit.MILLISECONDS.toNanos(THRESHOLD_MS); + + private final TraceDurationFilter filter = new TraceDurationFilter(THRESHOLD_MS); + + @Test + void traceOverThresholdIsKept() { + SpanData early = spanAt(0, 500); + SpanData late = spanAt(11500, 500); + + assertThat(filter.shouldKeep("trace-1", Arrays.asList(early, late))).isTrue(); + } + + @Test + void traceExactlyAtThresholdIsDropped() { + SpanData span1 = spanAtNanos(0, 100_000_000L); + SpanData span2 = spanAtNanos(THRESHOLD_NANOS - 100_000_000L, 100_000_000L); + + assertThat(filter.shouldKeep("trace-1", Arrays.asList(span1, span2))).isFalse(); + } + + @Test + void traceJustOverThresholdIsKept() { + SpanData span1 = spanAtNanos(0, 100_000_000L); + SpanData span2 = spanAtNanos(THRESHOLD_NANOS - 100_000_000L + 1, 100_000_000L); + + assertThat(filter.shouldKeep("trace-1", Arrays.asList(span1, span2))).isTrue(); + } + + @Test + void traceUnderThresholdIsDropped() { + SpanData span1 = spanAt(0, 500); + SpanData span2 = spanAt(4000, 500); + + assertThat(filter.shouldKeep("trace-1", Arrays.asList(span1, span2))).isFalse(); + } + + @Test + void singleSpanTrace() { + SpanData span = spanAt(0, 500); + + assertThat(filter.shouldKeep("trace-1", Collections.singletonList(span))).isFalse(); + } + + private static SpanData spanAt(long startOffsetMs, long durationMs) { + return spanAtNanos( + TimeUnit.MILLISECONDS.toNanos(startOffsetMs), TimeUnit.MILLISECONDS.toNanos(durationMs)); + } + + private static SpanData spanAtNanos(long startOffsetNanos, long durationNanos) { + SpanData span = mock(SpanData.class); + long baseNanos = TimeUnit.MILLISECONDS.toNanos(1_000_000_000L); + long startNanos = baseNanos + startOffsetNanos; + when(span.getStartEpochNanos()).thenReturn(startNanos); + when(span.getEndEpochNanos()).thenReturn(startNanos + durationNanos); + return span; + } +} From fa9e515acea291c896e103ca26cd54eb73886bd7 Mon Sep 17 00:00:00 2001 From: udaysagar2177 Date: Tue, 7 Apr 2026 13:28:04 -0700 Subject: [PATCH 2/5] Fix README linter: add blank line before built-in filters list --- processors/README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/processors/README.md b/processors/README.md index a8de0a1e5d..8e3ece968a 100644 --- a/processors/README.md +++ b/processors/README.md @@ -42,6 +42,7 @@ logger_provider: A trace is kept if any `SpanFilter` matches any of its spans, OR any `TraceFilter` matches the trace's span group. All spans sharing a trace ID are exported together. Built-in filters: + - `ErrorSpanFilter` - keeps traces containing any span with error status - `DurationSpanFilter` - keeps traces containing any span exceeding a duration threshold - `TraceDurationFilter` - keeps traces whose wall-clock duration (max end - min start) exceeds a threshold From 542418b8f947eefb7ca8102464db2fd45dd78f9f Mon Sep 17 00:00:00 2001 From: udaysagar2177 Date: Mon, 13 Apr 2026 17:03:54 -0700 Subject: [PATCH 3/5] Address PR review feedback on FilteringSpanExporter - Accept Duration instead of long millis in DurationSpanFilter and TraceDurationFilter - Add negative threshold validation to both duration filters - Add early return for empty span list in TraceDurationFilter (bug: overflow caused false positive) - Add Objects.requireNonNull validation in FilteringSpanExporter constructor - Rename metric to otel.contrib.processor.span.filtered and add unit {span} - Add tests for empty span list and negative threshold validation --- .../contrib/filter/DurationSpanFilter.java | 13 +++++++----- .../contrib/filter/FilteringSpanExporter.java | 8 +++++-- .../contrib/filter/TraceDurationFilter.java | 16 +++++++++----- .../filter/DurationSpanFilterTest.java | 15 ++++++++++--- .../filter/FilteringSpanExporterTest.java | 21 +++++++++++-------- .../filter/TraceDurationFilterTest.java | 20 +++++++++++++++--- 6 files changed, 66 insertions(+), 27 deletions(-) diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java index 9ff3b374d7..dc14af465b 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java @@ -6,7 +6,7 @@ package io.opentelemetry.contrib.filter; import io.opentelemetry.sdk.trace.data.SpanData; -import java.util.concurrent.TimeUnit; +import java.time.Duration; /** * A {@link SpanFilter} that keeps traces containing any span whose duration exceeds a configurable @@ -19,11 +19,14 @@ public final class DurationSpanFilter implements SpanFilter { /** * Creates a new {@code DurationSpanFilter}. * - * @param thresholdMs the duration threshold in milliseconds; spans with duration strictly greater - * than this are considered interesting + * @param threshold the duration threshold; spans with duration strictly greater than this are + * considered interesting */ - public DurationSpanFilter(long thresholdMs) { - this.thresholdNanos = TimeUnit.MILLISECONDS.toNanos(thresholdMs); + public DurationSpanFilter(Duration threshold) { + if (threshold.isNegative()) { + throw new IllegalArgumentException("threshold must be non-negative, got: " + threshold); + } + this.thresholdNanos = threshold.toNanos(); } @Override diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java index 7a8394a155..48d4b60d1e 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java @@ -19,6 +19,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import javax.annotation.Nullable; @@ -61,14 +62,17 @@ public FilteringSpanExporter( List spanFilters, List traceFilters, @Nullable Meter meter) { - this.delegate = delegate; + this.delegate = Objects.requireNonNull(delegate, "delegate"); + Objects.requireNonNull(spanFilters, "spanFilters"); + Objects.requireNonNull(traceFilters, "traceFilters"); this.spanFilters = Collections.unmodifiableList(new ArrayList<>(spanFilters)); this.traceFilters = Collections.unmodifiableList(new ArrayList<>(traceFilters)); if (meter != null) { this.droppedSpansCounter = meter - .counterBuilder("filtering.span.exporter.dropped") + .counterBuilder("otel.contrib.processor.span.filtered") .setDescription("Number of spans dropped by the filtering span exporter") + .setUnit("{span}") .build(); } else { this.droppedSpansCounter = null; diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java index 37cfa6f646..844eb2f33e 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java @@ -6,8 +6,8 @@ package io.opentelemetry.contrib.filter; import io.opentelemetry.sdk.trace.data.SpanData; +import java.time.Duration; import java.util.Collection; -import java.util.concurrent.TimeUnit; /** * A {@link TraceFilter} that keeps traces whose overall wall-clock duration (max end - min start @@ -20,15 +20,21 @@ public final class TraceDurationFilter implements TraceFilter { /** * Creates a new {@code TraceDurationFilter}. * - * @param thresholdMs the trace duration threshold in milliseconds; traces with wall-clock - * duration strictly greater than this are considered interesting + * @param threshold the trace duration threshold; traces with wall-clock duration strictly greater + * than this are considered interesting */ - public TraceDurationFilter(long thresholdMs) { - this.thresholdNanos = TimeUnit.MILLISECONDS.toNanos(thresholdMs); + public TraceDurationFilter(Duration threshold) { + if (threshold.isNegative()) { + throw new IllegalArgumentException("threshold must be non-negative, got: " + threshold); + } + this.thresholdNanos = threshold.toNanos(); } @Override public boolean shouldKeep(String traceId, Collection spans) { + if (spans.isEmpty()) { + return false; + } long minStart = Long.MAX_VALUE; long maxEnd = Long.MIN_VALUE; for (SpanData span : spans) { diff --git a/processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java b/processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java index a068d13733..b9e13716d4 100644 --- a/processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java +++ b/processors/src/test/java/io/opentelemetry/contrib/filter/DurationSpanFilterTest.java @@ -6,19 +6,21 @@ package io.opentelemetry.contrib.filter; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import io.opentelemetry.sdk.trace.data.SpanData; +import java.time.Duration; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; class DurationSpanFilterTest { - private static final long THRESHOLD_MS = 2000L; - private static final long THRESHOLD_NANOS = TimeUnit.MILLISECONDS.toNanos(THRESHOLD_MS); + private static final Duration THRESHOLD = Duration.ofSeconds(2); + private static final long THRESHOLD_NANOS = THRESHOLD.toNanos(); - private final DurationSpanFilter filter = new DurationSpanFilter(THRESHOLD_MS); + private final DurationSpanFilter filter = new DurationSpanFilter(THRESHOLD); @Test void spanOverThresholdIsKept() { @@ -38,6 +40,13 @@ void spanUnderThresholdIsDropped() { assertThat(filter.shouldKeep(span)).isFalse(); } + @Test + void negativeThresholdThrows() { + assertThatThrownBy(() -> new DurationSpanFilter(Duration.ofMillis(-1))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("threshold must be non-negative"); + } + private static SpanData spanWithDurationNanos(long durationNanos) { SpanData span = mock(SpanData.class); long startNanos = TimeUnit.MILLISECONDS.toNanos(1_000_000_000L); diff --git a/processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java b/processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java index 118019eace..6b104bcd50 100644 --- a/processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java +++ b/processors/src/test/java/io/opentelemetry/contrib/filter/FilteringSpanExporterTest.java @@ -24,6 +24,7 @@ import io.opentelemetry.sdk.trace.data.SpanData; import io.opentelemetry.sdk.trace.data.StatusData; import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.time.Duration; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -34,8 +35,8 @@ class FilteringSpanExporterTest { - private static final long SPAN_THRESHOLD_MS = 2000L; - private static final long TRACE_THRESHOLD_MS = 10000L; + private static final Duration SPAN_THRESHOLD = Duration.ofSeconds(2); + private static final Duration TRACE_THRESHOLD = Duration.ofSeconds(10); private SpanExporter delegate; private FilteringSpanExporter exporter; @@ -47,8 +48,8 @@ void setUp() { exporter = new FilteringSpanExporter( delegate, - Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(SPAN_THRESHOLD_MS)), - Collections.singletonList(new TraceDurationFilter(TRACE_THRESHOLD_MS))); + Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(SPAN_THRESHOLD)), + Collections.singletonList(new TraceDurationFilter(TRACE_THRESHOLD))); } // --- Trace grouping: if one span is interesting, all siblings are kept --- @@ -132,7 +133,7 @@ void onlyTraceFilters_noSpanFilters() { new FilteringSpanExporter( delegate, Collections.emptyList(), - Collections.singletonList(new TraceDurationFilter(TRACE_THRESHOLD_MS))); + Collections.singletonList(new TraceDurationFilter(TRACE_THRESHOLD))); SpanData earlySpan = createSpan("trace-1", "span-1", StatusCode.OK, 0, 500); SpanData lateSpan = createSpan("trace-1", "span-2", StatusCode.OK, 11500, 500); @@ -171,15 +172,16 @@ void droppedSpanMetricIsEmittedWhenMeterProvided() { Meter meter = mock(Meter.class); LongCounterBuilder counterBuilder = mock(LongCounterBuilder.class); LongCounter counter = mock(LongCounter.class); - when(meter.counterBuilder("filtering.span.exporter.dropped")).thenReturn(counterBuilder); + when(meter.counterBuilder("otel.contrib.processor.span.filtered")).thenReturn(counterBuilder); when(counterBuilder.setDescription("Number of spans dropped by the filtering span exporter")) .thenReturn(counterBuilder); + when(counterBuilder.setUnit("{span}")).thenReturn(counterBuilder); when(counterBuilder.build()).thenReturn(counter); FilteringSpanExporter exporterWithMetrics = new FilteringSpanExporter( delegate, - Collections.singletonList(new DurationSpanFilter(SPAN_THRESHOLD_MS)), + Collections.singletonList(new DurationSpanFilter(SPAN_THRESHOLD)), Collections.emptyList(), meter); @@ -194,15 +196,16 @@ void noMetricEmittedWhenNoSpansDropped() { Meter meter = mock(Meter.class); LongCounterBuilder counterBuilder = mock(LongCounterBuilder.class); LongCounter counter = mock(LongCounter.class); - when(meter.counterBuilder("filtering.span.exporter.dropped")).thenReturn(counterBuilder); + when(meter.counterBuilder("otel.contrib.processor.span.filtered")).thenReturn(counterBuilder); when(counterBuilder.setDescription("Number of spans dropped by the filtering span exporter")) .thenReturn(counterBuilder); + when(counterBuilder.setUnit("{span}")).thenReturn(counterBuilder); when(counterBuilder.build()).thenReturn(counter); FilteringSpanExporter exporterWithMetrics = new FilteringSpanExporter( delegate, - Collections.singletonList(new DurationSpanFilter(SPAN_THRESHOLD_MS)), + Collections.singletonList(new DurationSpanFilter(SPAN_THRESHOLD)), Collections.emptyList(), meter); diff --git a/processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java b/processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java index 50c94361f3..6e56ca4d8a 100644 --- a/processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java +++ b/processors/src/test/java/io/opentelemetry/contrib/filter/TraceDurationFilterTest.java @@ -6,10 +6,12 @@ package io.opentelemetry.contrib.filter; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import io.opentelemetry.sdk.trace.data.SpanData; +import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.concurrent.TimeUnit; @@ -17,10 +19,10 @@ class TraceDurationFilterTest { - private static final long THRESHOLD_MS = 10000L; - private static final long THRESHOLD_NANOS = TimeUnit.MILLISECONDS.toNanos(THRESHOLD_MS); + private static final Duration THRESHOLD = Duration.ofSeconds(10); + private static final long THRESHOLD_NANOS = THRESHOLD.toNanos(); - private final TraceDurationFilter filter = new TraceDurationFilter(THRESHOLD_MS); + private final TraceDurationFilter filter = new TraceDurationFilter(THRESHOLD); @Test void traceOverThresholdIsKept() { @@ -61,6 +63,18 @@ void singleSpanTrace() { assertThat(filter.shouldKeep("trace-1", Collections.singletonList(span))).isFalse(); } + @Test + void emptySpanListReturnsFalse() { + assertThat(filter.shouldKeep("trace-1", Collections.emptyList())).isFalse(); + } + + @Test + void negativeThresholdThrows() { + assertThatThrownBy(() -> new TraceDurationFilter(Duration.ofMillis(-1))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("threshold must be non-negative"); + } + private static SpanData spanAt(long startOffsetMs, long durationMs) { return spanAtNanos( TimeUnit.MILLISECONDS.toNanos(startOffsetMs), TimeUnit.MILLISECONDS.toNanos(durationMs)); From b2fb23297355d3eb196ec449cdc88c829804e044 Mon Sep 17 00:00:00 2001 From: udaysagar2177 Date: Mon, 13 Apr 2026 17:20:37 -0700 Subject: [PATCH 4/5] Clarify that filtering is scoped to a single export batch Javadoc and README previously said "keeps traces" which implies full trace-level guarantees. Reworded to make clear that filtering decisions apply only to spans within the same export() call, and that a trace split across batches may be partially exported. Also updated README code examples to use Duration API. --- processors/README.md | 22 ++++++++++--------- .../contrib/filter/DurationSpanFilter.java | 4 ++-- .../contrib/filter/ErrorSpanFilter.java | 5 ++++- .../contrib/filter/FilteringSpanExporter.java | 17 +++++++++----- .../contrib/filter/SpanFilter.java | 13 ++++++----- .../contrib/filter/TraceDurationFilter.java | 4 ++-- .../contrib/filter/TraceFilter.java | 6 ++--- 7 files changed, 41 insertions(+), 30 deletions(-) diff --git a/processors/README.md b/processors/README.md index 8e3ece968a..42f17aa55f 100644 --- a/processors/README.md +++ b/processors/README.md @@ -34,29 +34,31 @@ logger_provider: ## Filtering Span Exporter -`FilteringSpanExporter` is a `SpanExporter` wrapper that filters spans at the trace level before delegating to the underlying exporter. Filtering is composable via two interfaces: +`FilteringSpanExporter` is a `SpanExporter` wrapper that filters spans within each export batch before delegating to the underlying exporter. Filtering is composable via two interfaces: - `SpanFilter` - evaluates individual spans (e.g., error status, slow duration) - `TraceFilter` - evaluates all spans belonging to a trace within the batch (e.g., overall trace wall-clock duration) -A trace is kept if any `SpanFilter` matches any of its spans, OR any `TraceFilter` matches the trace's span group. All spans sharing a trace ID are exported together. +Within a batch, if any `SpanFilter` matches any span or any `TraceFilter` matches a trace's span group, all spans sharing that trace ID in the batch are exported together. + +**Note:** Filtering decisions are scoped to a single `export()` call. Spans from the same trace arriving in different batches are evaluated independently, so a trace split across batches may be partially exported. Built-in filters: -- `ErrorSpanFilter` - keeps traces containing any span with error status -- `DurationSpanFilter` - keeps traces containing any span exceeding a duration threshold -- `TraceDurationFilter` - keeps traces whose wall-clock duration (max end - min start) exceeds a threshold +- `ErrorSpanFilter` - matches spans with error status +- `DurationSpanFilter` - matches spans exceeding a duration threshold +- `TraceDurationFilter` - matches when a trace's wall-clock duration (max end - min start) in the batch exceeds a threshold Usage: ```java SpanExporter delegate = OtlpGrpcSpanExporter.getDefault(); -// Export traces with errors, individual spans > 2s, or trace duration > 10s +// Export spans whose batch-colocated trace has errors, individual spans > 2s, or trace duration > 10s SpanExporter filtering = new FilteringSpanExporter( delegate, - Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(2000)), - Collections.singletonList(new TraceDurationFilter(10000))); + Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(Duration.ofSeconds(2))), + Collections.singletonList(new TraceDurationFilter(Duration.ofSeconds(10)))); // Custom filters SpanFilter nameFilter = span -> span.getName().contains("important"); @@ -69,8 +71,8 @@ SpanExporter custom = new FilteringSpanExporter( Meter meter = openTelemetry.getMeter("my-service"); SpanExporter withMetrics = new FilteringSpanExporter( delegate, - Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(2000)), - Collections.singletonList(new TraceDurationFilter(10000)), + Arrays.asList(new ErrorSpanFilter(), new DurationSpanFilter(Duration.ofSeconds(2))), + Collections.singletonList(new TraceDurationFilter(Duration.ofSeconds(10))), meter); ``` diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java index dc14af465b..037fd82ef1 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/DurationSpanFilter.java @@ -9,8 +9,8 @@ import java.time.Duration; /** - * A {@link SpanFilter} that keeps traces containing any span whose duration exceeds a configurable - * threshold. + * A {@link SpanFilter} that matches spans whose duration exceeds a configurable threshold, causing + * all batch-colocated spans sharing the same trace ID to be exported. */ public final class DurationSpanFilter implements SpanFilter { diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java index 6e28be90f0..34e5f59081 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/ErrorSpanFilter.java @@ -8,7 +8,10 @@ import io.opentelemetry.api.trace.StatusCode; import io.opentelemetry.sdk.trace.data.SpanData; -/** A {@link SpanFilter} that keeps traces containing any span with {@link StatusCode#ERROR}. */ +/** + * A {@link SpanFilter} that matches spans with {@link StatusCode#ERROR}, causing all + * batch-colocated spans sharing the same trace ID to be exported. + */ public final class ErrorSpanFilter implements SpanFilter { @Override diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java index 48d4b60d1e..a003a8eaf1 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java @@ -25,8 +25,8 @@ /** * A {@link SpanExporter} wrapper that filters spans before delegating to the underlying exporter. - * Filtering operates at the trace level: if any filter matches, all spans sharing that trace ID are - * exported together. + * Filtering operates at the trace level within each export batch: if any filter matches, all spans + * sharing that trace ID in that batch are exported together. * *

Two types of filters are supported: * @@ -36,8 +36,12 @@ * overall trace wall-clock duration) * * - *

A trace is kept if any {@code SpanFilter} matches any of its spans, OR any {@code TraceFilter} - * matches the trace's span group. + *

A trace's spans are kept if any {@code SpanFilter} matches any span in the batch, OR any + * {@code TraceFilter} matches the trace's span group within the batch. + * + *

Important: Filtering decisions are scoped to a single {@link + * #export(Collection)} call. Spans from the same trace that arrive in different batches are + * evaluated independently, so a trace split across batches may be partially exported. */ public final class FilteringSpanExporter implements SpanExporter { @@ -52,8 +56,9 @@ public final class FilteringSpanExporter implements SpanExporter { * Creates a new {@code FilteringSpanExporter}. * * @param delegate the exporter to delegate to for spans that pass filtering - * @param spanFilters per-span filters; a trace is kept if any filter matches any span - * @param traceFilters batch-level filters; a trace is kept if any filter matches the trace group + * @param spanFilters per-span filters; a trace's spans in the batch are kept if any filter matches + * @param traceFilters batch-level filters; a trace's spans in the batch are kept if any filter + * matches * @param meter optional {@link Meter} for emitting dropped-span metrics; pass {@code null} to * disable metrics */ diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java index 272a09a8ba..7268f94005 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/SpanFilter.java @@ -8,19 +8,20 @@ import io.opentelemetry.sdk.trace.data.SpanData; /** - * A filter that evaluates individual spans to determine if their containing trace should be - * exported. Used by {@link FilteringSpanExporter} to make per-span keep/drop decisions. + * A filter that evaluates individual spans to determine if their containing trace's spans (within + * the same export batch) should be exported. Used by {@link FilteringSpanExporter} to make per-span + * keep/drop decisions. * - *

If any {@code SpanFilter} returns {@code true} for any span in a trace, all spans sharing that - * trace ID are exported. + *

If any {@code SpanFilter} returns {@code true} for any span in a batch, all spans sharing that + * trace ID within the same batch are exported. */ public interface SpanFilter { /** - * Evaluates whether the given span is interesting enough to keep its entire trace. + * Evaluates whether the given span is interesting enough to keep its trace's spans in the batch. * * @param spanData the span to evaluate - * @return {@code true} if this span should cause its trace to be exported + * @return {@code true} if this span should cause its trace's batch-colocated spans to be exported */ boolean shouldKeep(SpanData spanData); } diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java index 844eb2f33e..b2ccb7c8ed 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceDurationFilter.java @@ -10,8 +10,8 @@ import java.util.Collection; /** - * A {@link TraceFilter} that keeps traces whose overall wall-clock duration (max end - min start - * across all spans in the batch sharing a trace ID) exceeds a configurable threshold. + * A {@link TraceFilter} that matches when the wall-clock duration (max end - min start across all + * spans in the batch sharing a trace ID) exceeds a configurable threshold. */ public final class TraceDurationFilter implements TraceFilter { diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java index a646df9b46..55dbad6be0 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/TraceFilter.java @@ -10,11 +10,11 @@ /** * A filter that evaluates all spans belonging to a single trace within an export batch to determine - * if the trace should be exported. Used by {@link FilteringSpanExporter} for decisions that require - * batch-level context (e.g., overall trace wall-clock duration). + * if those spans should be exported. Used by {@link FilteringSpanExporter} for decisions that + * require batch-level context (e.g., overall trace wall-clock duration). * *

If any {@code TraceFilter} returns {@code true} for a trace, all spans sharing that trace ID - * are exported. + * within the same batch are exported. */ public interface TraceFilter { From 0a0c9d3018f4fa60278952b427c1016ec4c7925c Mon Sep 17 00:00:00 2001 From: udaysagar2177 Date: Thu, 23 Apr 2026 00:52:21 -0700 Subject: [PATCH 5/5] Apply Spotless formatting: wrap long Javadoc line in FilteringSpanExporter --- .../io/opentelemetry/contrib/filter/FilteringSpanExporter.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java index a003a8eaf1..ba5d359221 100644 --- a/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java +++ b/processors/src/main/java/io/opentelemetry/contrib/filter/FilteringSpanExporter.java @@ -56,7 +56,8 @@ public final class FilteringSpanExporter implements SpanExporter { * Creates a new {@code FilteringSpanExporter}. * * @param delegate the exporter to delegate to for spans that pass filtering - * @param spanFilters per-span filters; a trace's spans in the batch are kept if any filter matches + * @param spanFilters per-span filters; a trace's spans in the batch are kept if any filter + * matches * @param traceFilters batch-level filters; a trace's spans in the batch are kept if any filter * matches * @param meter optional {@link Meter} for emitting dropped-span metrics; pass {@code null} to