From ede28370899c6cec07ab9f5752808a0cc0e19db9 Mon Sep 17 00:00:00 2001 From: Armen Petrosyan Date: Tue, 13 Oct 2026 10:07:40 +0300 Subject: [PATCH] Keep the current sampler when the Jaeger sampling strategy is unchanged JaegerRemoteSampler built a new sampler from every poll response, even when the strategy had not changed. For a RATE_LIMITING strategy that means a new RateLimiter with a full balance on every poll, so under load each poll lets through up to maxTracesPerSecond extra traces. Remember the last applied response and keep the current sampler when the response bytes are the same. Keeping the limiter state when only the rate changes is left for a follow-up. Part of #2465 --- .../jaeger/sampler/JaegerRemoteSampler.java | 22 +++++++++++-- .../sampler/JaegerRemoteSamplerTest.java | 32 +++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/sdk-extensions/jaeger-remote-sampler/src/main/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSampler.java b/sdk-extensions/jaeger-remote-sampler/src/main/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSampler.java index ac8a6d56033..8d23a45d205 100644 --- a/sdk-extensions/jaeger-remote-sampler/src/main/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSampler.java +++ b/sdk-extensions/jaeger-remote-sampler/src/main/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSampler.java @@ -6,6 +6,7 @@ package io.opentelemetry.sdk.extension.trace.jaeger.sampler; import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.internal.GuardedBy; import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.context.Context; import io.opentelemetry.sdk.common.CompletableResultCode; @@ -19,6 +20,7 @@ import io.opentelemetry.sdk.trace.samplers.SamplingResult; import java.io.IOException; import java.net.URI; +import java.util.Arrays; import java.util.List; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -48,6 +50,11 @@ public final class JaegerRemoteSampler implements Sampler { private final GrpcSender grpcSender; private final URI endpoint; private final int pollingIntervalMs; + private final Object lock = new Object(); + + @GuardedBy("lock") + @Nullable + private byte[] lastResponseMessage; JaegerRemoteSampler( GrpcSender grpcSender, @@ -95,10 +102,19 @@ private void onResponse(GrpcResponse grpcResponse) { if (isShutdown.get()) { return; } + byte[] responseMessage = grpcResponse.getResponseMessage(); try { - SamplingStrategyResponse strategyResponse = - SamplingStrategyResponseUnMarshaler.read(grpcResponse.getResponseMessage()); - sampler = updateSampler(strategyResponse); + synchronized (lock) { + // Keep the current sampler when the strategy hasn't changed, so stateful samplers such + // as RateLimitingSampler don't lose their state on every poll. + if (Arrays.equals(responseMessage, lastResponseMessage)) { + return; + } + SamplingStrategyResponse strategyResponse = + SamplingStrategyResponseUnMarshaler.read(responseMessage); + sampler = updateSampler(strategyResponse); + lastResponseMessage = responseMessage; + } } catch (IOException e) { logger.log(Level.WARNING, "Failed to unmarshal strategy response", e); } diff --git a/sdk-extensions/jaeger-remote-sampler/src/test/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSamplerTest.java b/sdk-extensions/jaeger-remote-sampler/src/test/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSamplerTest.java index 5b42ca9f0d8..46ff40a0cd3 100644 --- a/sdk-extensions/jaeger-remote-sampler/src/test/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSamplerTest.java +++ b/sdk-extensions/jaeger-remote-sampler/src/test/java/io/opentelemetry/sdk/extension/trace/jaeger/sampler/JaegerRemoteSamplerTest.java @@ -38,6 +38,7 @@ import java.util.Collections; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Stream; import javax.annotation.Nullable; import javax.net.ssl.KeyManager; @@ -71,6 +72,8 @@ class JaegerRemoteSamplerTest { private static final ConcurrentLinkedQueue responses = new ConcurrentLinkedQueue<>(); + private static final AtomicInteger requestCount = new AtomicInteger(); + private static void addGrpcError(GrpcStatusCode code, @Nullable String message) { grpcErrors.add(new ArmeriaStatusException(code.getValue(), message)); } @@ -104,6 +107,7 @@ public void getSamplingStrategy( io.opentelemetry.sdk.extension.trace.jaeger.proto.api_v2 .Sampling.SamplingStrategyResponse> responseObserver) { + requestCount.incrementAndGet(); ArmeriaStatusException grpcError = grpcErrors.poll(); if (grpcError != null) { responseObserver.onError(grpcError); @@ -142,6 +146,7 @@ public void getSamplingStrategy( public void before() { grpcErrors.clear(); responses.clear(); + requestCount.set(0); } @Test @@ -449,6 +454,33 @@ void perOperationSampling_zeroRateLimiting() { } } + @Test + void unchangedStrategy_keepsSampler() { + try (JaegerRemoteSampler sampler = + JaegerRemoteSampler.builder() + .setEndpoint(server.httpUri().toString()) + .setServiceName(SERVICE_NAME) + .setPollingInterval(50, TimeUnit.MILLISECONDS) + .build()) { + await().untilAsserted(samplerIsType(sampler, RateLimitingSampler.class)); + Sampler initial = sampler.getSampler(); + + // Further polls return the same strategy, so the stateful sampler must be retained. + int polls = requestCount.get(); + await().untilAsserted(() -> assertThat(requestCount.get()).isGreaterThan(polls + 2)); + assertThat(sampler.getSampler()).isSameAs(initial); + + // A changed strategy still replaces the sampler. + responses.add( + Sampling.SamplingStrategyResponse.newBuilder() + .setStrategyType(SamplingStrategyType.RATE_LIMITING) + .setRateLimitingSampling( + RateLimitingSamplingStrategy.newBuilder().setMaxTracesPerSecond(5).build()) + .build()); + await().untilAsserted(() -> assertThat(sampler.getSampler()).isNotSameAs(initial)); + } + } + @Test void internal_error_server_response() { addGrpcError(GrpcStatusCode.INTERNAL, "internal error");