groupResponseWaitingTimeAvgMap =
- new VeniceConcurrentHashMap<>();
+ static final long GROUP_RESPONSE_WAITING_TIME_WINDOW_MS = TimeUnit.SECONDS.toMillis(30);
+
+ /**
+ * Per-Helix-group OTel metric entity states and per-group response-waiting-time averages,
+ * keyed by group ID. Each map grows lazily via {@code computeIfAbsent} and is bounded by the
+ * number of Helix groups configured for the store (typically 3–5). Entries are not evicted —
+ * the maps persist for the lifetime of this stats instance.
+ *
+ * Exactly one of {@link #groupResponseWaitingTimeTehutiAvgMap} (legacy Tehuti
+ * {@link io.tehuti.Metric} references) and {@link #groupResponseWaitingTimeIndependentAvgMap}
+ * (independent {@link SlidingWindowAverage}) is non-null, selected by {@link #useSelfContainedStats}
+ * at construction time. The remaining maps hold OTel {@link MetricEntityStateBase} instances and
+ * Tehuti-joined recording state and are always populated.
+ */
+ private final VeniceConcurrentHashMap groupResponseWaitingTimeTehutiAvgMap;
+ private final VeniceConcurrentHashMap groupResponseWaitingTimeIndependentAvgMap;
private final VeniceConcurrentHashMap groupRequestCountMap =
new VeniceConcurrentHashMap<>();
private final VeniceConcurrentHashMap groupPendingRequestMap =
@@ -39,6 +53,7 @@ public class HelixGroupStats extends AbstractVeniceStats {
private final VeniceConcurrentHashMap groupResponseWaitingTimeMap =
new VeniceConcurrentHashMap<>();
private final String storeName;
+ private final boolean useSelfContainedStats;
// OTel metrics
private final MetricEntityStateBase helixGroupCount;
@@ -49,12 +64,30 @@ public class HelixGroupStats extends AbstractVeniceStats {
private final Attributes baseAttributes;
public HelixGroupStats(MetricsRepository metricsRepository) {
- this(metricsRepository, "");
+ this(metricsRepository, "", false);
+ }
+
+ public HelixGroupStats(MetricsRepository metricsRepository, boolean useSelfContainedStats) {
+ this(metricsRepository, "", useSelfContainedStats);
}
public HelixGroupStats(MetricsRepository metricsRepository, String prefix) {
+ this(metricsRepository, prefix, false);
+ }
+
+ /**
+ * @param useSelfContainedStats {@code false} (default) reads the per-group response-waiting-time
+ * average from the Tehuti {@link io.tehuti.metrics.stats.Avg}
+ * metric; {@code true} reads from an independent
+ * {@link SlidingWindowAverage} owned by this class so the routing
+ * decision remains correct even when the Tehuti dependency is removed.
+ */
+ public HelixGroupStats(MetricsRepository metricsRepository, String prefix, boolean useSelfContainedStats) {
super(metricsRepository, prefix.isEmpty() ? "HelixGroupStats" : prefix + "_HelixGroupStats");
this.storeName = prefix;
+ this.useSelfContainedStats = useSelfContainedStats;
+ this.groupResponseWaitingTimeTehutiAvgMap = useSelfContainedStats ? null : new VeniceConcurrentHashMap<>();
+ this.groupResponseWaitingTimeIndependentAvgMap = useSelfContainedStats ? new VeniceConcurrentHashMap<>() : null;
// When storeName is empty, it means the stats is used for Venice Router.
if (storeName.isEmpty()) {
this.otelRepository = null;
@@ -135,22 +168,43 @@ private MetricEntityStateBase buildHelixGroupResponseWaitingTime(int groupId, Me
}
public void recordGroupResponseWaitingTime(int groupId, double responseWaitingTime) {
+ // Tehuti+OTel joint recording — always active, regardless of which read path is selected.
MetricEntityStateBase groupResponseWaitingTime = groupResponseWaitingTimeMap.computeIfAbsent(groupId, id -> {
MeasurableStat avgStat = new Avg();
MetricEntityStateBase waitTime = buildHelixGroupResponseWaitingTime(groupId, avgStat);
- groupResponseWaitingTimeAvgMap
- .put(groupId, getMetricsRepository().getMetric(getMetricFullName(waitTime.getTehutiSensor(), avgStat)));
+ if (!useSelfContainedStats) {
+ // Legacy path: cache the Tehuti Metric reference for routing logic to read.
+ groupResponseWaitingTimeTehutiAvgMap
+ .put(groupId, getMetricsRepository().getMetric(getMetricFullName(waitTime.getTehutiSensor(), avgStat)));
+ }
return waitTime;
});
groupResponseWaitingTime.record(responseWaitingTime);
+
+ if (useSelfContainedStats) {
+ // Independent sliding-window average owned by this class — routing decision stays correct
+ // even when Tehuti is disabled.
+ groupResponseWaitingTimeIndependentAvgMap
+ .computeIfAbsent(groupId, id -> new SlidingWindowAverage(GROUP_RESPONSE_WAITING_TIME_WINDOW_MS))
+ .record(responseWaitingTime);
+ }
}
public double getGroupResponseWaitingTimeAvg(int groupId) {
- Metric groupResponseWaitingTimeAvgMetric = groupResponseWaitingTimeAvgMap.get(groupId);
- if (groupResponseWaitingTimeAvgMetric == null) {
- return -1;
+ double avgLatency;
+ if (useSelfContainedStats) {
+ SlidingWindowAverage counter = groupResponseWaitingTimeIndependentAvgMap.get(groupId);
+ if (counter == null) {
+ return -1;
+ }
+ avgLatency = counter.average();
+ } else {
+ Metric metric = groupResponseWaitingTimeTehutiAvgMap.get(groupId);
+ if (metric == null) {
+ return -1;
+ }
+ avgLatency = metric.value();
}
- double avgLatency = groupResponseWaitingTimeAvgMetric.value();
if (Double.isNaN(avgLatency)) {
return -1;
}
diff --git a/internal/venice-client-common/src/main/java/com/linkedin/venice/utils/concurrent/LatencyPercentileProvider.java b/internal/venice-client-common/src/main/java/com/linkedin/venice/utils/concurrent/LatencyPercentileProvider.java
new file mode 100644
index 00000000000..4e95d6cb66a
--- /dev/null
+++ b/internal/venice-client-common/src/main/java/com/linkedin/venice/utils/concurrent/LatencyPercentileProvider.java
@@ -0,0 +1,105 @@
+package com.linkedin.venice.utils.concurrent;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongArray;
+
+
+/**
+ * Independent percentile provider for read-path latencies. Owns its backing state so
+ * signal services can produce a throttling signal without depending on Tehuti's windowed
+ * {@link io.tehuti.metrics.stats.Percentiles} computation — Tehuti {@code Percentiles} can
+ * only be read via {@code MetricsRepository.getMetric()}, making the signal unavailable when
+ * Tehuti is removed. This class provides the same p99 signal with no metrics-library dependency.
+ *
+ * Each {@link LatencyType} keeps a fixed-size sample reservoir (ring buffer). Observations
+ * overwrite the oldest sample once the buffer is full — this approximates a moving window for
+ * high-throughput workloads where the buffer fills quickly relative to the throttler signal
+ * refresh interval. Percentile reads snapshot the reservoir, sort, and index.
+ *
+ *
Write path — {@link #observe}: O(1), lock-free
+ * One {@link AtomicLong#getAndIncrement()} to claim the next slot + one
+ * {@link AtomicLongArray#set} to store the value (encoded via
+ * {@link Double#doubleToRawLongBits} — {@code AtomicLongArray} has no {@code double} variant
+ * in the JDK). No locks, no CAS retry loop. Safe for concurrent writers at any request rate.
+ *
+ * Read path — {@link #getP99}: O(n log n), called only periodically
+ * Snapshots all {@code capacity} slots (default {@value #DEFAULT_RESERVOIR_CAPACITY}), sorts
+ * them, and returns the value at the 99th-percentile index. At the default capacity this takes
+ * ~50–100 µs on modern hardware (fits in L2 cache). Crucially, {@code getP99} is only ever
+ * called on a fixed periodic schedule (default every 30 s), so the sort cost is amortised
+ * across millions of writes and is negligible in practice.
+ *
+ * Concurrent writes during a read are tolerated — the signal is an approximation and a few
+ * samples changing underfoot does not meaningfully skew a p99 over thousands of observations.
+ */
+public class LatencyPercentileProvider {
+ /** Read-path latency categories observed by this provider. */
+ public enum LatencyType {
+ SINGLE_GET, MULTI_GET, READ_COMPUTE
+ }
+
+ /** Default reservoir capacity per {@link LatencyType} — large enough to stabilize p99 under load. */
+ public static final int DEFAULT_RESERVOIR_CAPACITY = 4096;
+
+ private final Reservoir[] reservoirs;
+
+ public LatencyPercentileProvider() {
+ this(DEFAULT_RESERVOIR_CAPACITY);
+ }
+
+ public LatencyPercentileProvider(int reservoirCapacity) {
+ if (reservoirCapacity <= 0) {
+ throw new IllegalArgumentException("reservoirCapacity must be > 0, got " + reservoirCapacity);
+ }
+ LatencyType[] types = LatencyType.values();
+ this.reservoirs = new Reservoir[types.length];
+ for (int i = 0; i < types.length; i++) {
+ this.reservoirs[i] = new Reservoir(reservoirCapacity);
+ }
+ }
+
+ public void observe(LatencyType type, double latencyMs) {
+ reservoirs[type.ordinal()].observe(latencyMs);
+ }
+
+ /**
+ * @return 99th percentile of observations in the reservoir for {@code type}, or 0 when no
+ * samples have been recorded yet. Returning 0 (rather than NaN) keeps the threshold
+ * comparison semantics of callers: signal stays inactive when the reservoir is empty.
+ */
+ public double getP99(LatencyType type) {
+ return reservoirs[type.ordinal()].percentile(99.0);
+ }
+
+ private static final class Reservoir {
+ private final int capacity;
+ private final AtomicLongArray samples;
+ private final AtomicLong nextIndex = new AtomicLong();
+
+ Reservoir(int capacity) {
+ this.capacity = capacity;
+ this.samples = new AtomicLongArray(capacity);
+ }
+
+ void observe(double value) {
+ long idx = nextIndex.getAndIncrement();
+ samples.set((int) Math.floorMod(idx, capacity), Double.doubleToRawLongBits(value));
+ }
+
+ double percentile(double p) {
+ long total = nextIndex.get();
+ if (total <= 0) {
+ return 0.0;
+ }
+ int size = (int) Math.min(total, (long) capacity);
+ double[] copy = new double[size];
+ for (int i = 0; i < size; i++) {
+ copy[i] = Double.longBitsToDouble(samples.get(i));
+ }
+ Arrays.sort(copy);
+ int idx = (int) Math.min(size - 1L, Math.max(0L, Math.round((p / 100.0) * (size - 1))));
+ return copy[idx];
+ }
+ }
+}
diff --git a/internal/venice-client-common/src/main/java/com/linkedin/venice/utils/concurrent/SlidingWindowAverage.java b/internal/venice-client-common/src/main/java/com/linkedin/venice/utils/concurrent/SlidingWindowAverage.java
new file mode 100644
index 00000000000..d66c3913d7e
--- /dev/null
+++ b/internal/venice-client-common/src/main/java/com/linkedin/venice/utils/concurrent/SlidingWindowAverage.java
@@ -0,0 +1,143 @@
+package com.linkedin.venice.utils.concurrent;
+
+import com.linkedin.venice.utils.SystemTime;
+import com.linkedin.venice.utils.Time;
+import java.util.concurrent.atomic.DoubleAdder;
+import java.util.concurrent.atomic.LongAdder;
+
+
+/**
+ * Sliding-window arithmetic mean over two tumbling sub-buckets.
+ *
+ *
Why two buckets?
+ * A single tumbling bucket resets abruptly at every boundary — the average drops to NaN even
+ * under steady load. Two buckets eliminate this cliff edge: at any instant you see between one
+ * and two full buckets of data. {@code windowMs} is tunable; shorter = more responsive but
+ * noisier, longer = smoother but lagging.
+ *
+ * Write path — {@link #record}: O(1), lock-free
+ * One {@link LongAdder#increment()} and one {@link DoubleAdder#add(double)} into the active
+ * bucket. Rotation fires at most once per {@code windowMs / 2} under a short
+ * {@code synchronized} block — never on the per-record hot path.
+ *
+ * Read path — {@link #average}: O(cells), cached
+ * A full recompute merges two pairs of {@link LongAdder}/{@link DoubleAdder} cells (O(cells),
+ * typically ≤ 32 per adder on a busy JVM). Because {@code average()} is called on the
+ * per-request routing hot path, the result is cached for {@value #CACHE_MS} ms via a single
+ * volatile reference to an immutable {@link CachedAverage} object — one volatile load on the
+ * fast path. Staleness is negligible relative to the sub-bucket duration ({@code windowMs / 2}).
+ */
+public class SlidingWindowAverage {
+ static final long CACHE_MS = 100;
+
+ private final long bucketMs;
+ private final Time time;
+ private final long cacheMs;
+ private final Object rotateLock = new Object();
+ private volatile Bucket current;
+ private volatile Bucket previous;
+
+ /* Single volatile reference — one load on the hot path, atomic update via reference swap.
+ * NaN avg signals "no data": the cache is bypassed so new records are seen immediately. */
+ private volatile CachedAverage cache = CachedAverage.EMPTY;
+
+ private static final class CachedAverage {
+ static final CachedAverage EMPTY = new CachedAverage(Double.NaN, 0);
+
+ final double avg;
+ final long refreshMs;
+
+ CachedAverage(double avg, long refreshMs) {
+ this.avg = avg;
+ this.refreshMs = refreshMs;
+ }
+
+ boolean isValid(long now, long ttlMs) {
+ return !Double.isNaN(avg) && now - refreshMs < ttlMs;
+ }
+ }
+
+ private static final class Bucket {
+ final long startMs;
+ final LongAdder count = new LongAdder();
+ final DoubleAdder sum = new DoubleAdder();
+
+ Bucket(long startMs) {
+ this.startMs = startMs;
+ }
+ }
+
+ public SlidingWindowAverage(long windowMs) {
+ this(windowMs, new SystemTime(), CACHE_MS);
+ }
+
+ public SlidingWindowAverage(long windowMs, Time time) {
+ this(windowMs, time, CACHE_MS);
+ }
+
+ public SlidingWindowAverage(long windowMs, Time time, long cacheMs) {
+ if (windowMs <= 0) {
+ throw new IllegalArgumentException("windowMs must be > 0, got " + windowMs);
+ }
+ if (time == null) {
+ throw new IllegalArgumentException("time must not be null");
+ }
+ this.bucketMs = Math.max(1, windowMs / 2);
+ this.time = time;
+ this.cacheMs = cacheMs;
+ long now = time.getMilliseconds();
+ this.current = new Bucket(now);
+ this.previous = new Bucket(now - bucketMs);
+ }
+
+ public void record(double value) {
+ Bucket b = maybeRotate();
+ b.count.increment();
+ b.sum.add(value);
+ }
+
+ /**
+ * @return arithmetic mean across the current sliding window, or {@link Double#NaN} when no
+ * observations have been recorded. Result is cached for {@value #CACHE_MS} ms; NaN
+ * is never cached so new records arriving in an empty window are seen immediately.
+ */
+ public double average() {
+ long now = time.getMilliseconds();
+ CachedAverage c = cache; // single volatile load
+ if (c.isValid(now, cacheMs)) {
+ return c.avg;
+ }
+ maybeRotate();
+ double count = current.count.sum() + previous.count.sum();
+ double avg = count == 0 ? Double.NaN : (current.sum.sum() + previous.sum.sum()) / count;
+ // Atomic reference swap — cache update is always consistent (readers see full CachedAverage).
+ // NaN is not cached: leave cache as EMPTY so the next call recomputes and picks up new records.
+ if (!Double.isNaN(avg)) {
+ cache = new CachedAverage(avg, now);
+ }
+ return avg;
+ }
+
+ private Bucket maybeRotate() {
+ long now = time.getMilliseconds();
+ Bucket cur = current;
+ if (now - cur.startMs < bucketMs) {
+ return cur;
+ }
+ synchronized (rotateLock) {
+ cur = current;
+ long age = now - cur.startMs;
+ if (age < bucketMs) {
+ return cur;
+ }
+ if (age >= 2 * bucketMs) {
+ previous = new Bucket(now - bucketMs);
+ current = new Bucket(now);
+ } else {
+ previous = cur;
+ current = new Bucket(now);
+ }
+ return current;
+ }
+ }
+}
diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/reliability/LoadControllerTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/reliability/LoadControllerTest.java
index 4102059bbd1..58c58f9725e 100644
--- a/internal/venice-client-common/src/test/java/com/linkedin/venice/reliability/LoadControllerTest.java
+++ b/internal/venice-client-common/src/test/java/com/linkedin/venice/reliability/LoadControllerTest.java
@@ -4,18 +4,27 @@
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
+import com.linkedin.venice.utils.TestMockTime;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class LoadControllerTest {
- @Test
- public void testRequestRejectionWhenOverload() {
+ @DataProvider(name = "counterPaths")
+ public Object[][] counterPaths() {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "counterPaths")
+ public void testRequestRejectionWhenOverload(boolean useIndependentCounter) {
LoadController loadController = LoadController.newBuilder()
.setWindowSizeInSec(5)
.setRejectionRatioUpdateIntervalInSec(1)
.setMaxRejectionRatio(0.5)
.setAcceptMultiplier(2.0)
+ .setUseIndependentCounter(useIndependentCounter)
.build();
for (int i = 0; i < 100; i++) {
loadController.recordRequest();
@@ -36,13 +45,14 @@ public void testRequestRejectionWhenOverload() {
assertTrue(rejectCount > 300);
}
- @Test
- public void testRejectionRatioReset() throws InterruptedException {
+ @Test(dataProvider = "counterPaths")
+ public void testRejectionRatioReset(boolean useIndependentCounter) throws InterruptedException {
LoadController loadController = LoadController.newBuilder()
.setWindowSizeInSec(3)
.setRejectionRatioUpdateIntervalInSec(1)
.setMaxRejectionRatio(0.5)
.setAcceptMultiplier(1.0)
+ .setUseIndependentCounter(useIndependentCounter)
.build();
for (int i = 0; i < 100; i++) {
loadController.recordRequest();
@@ -64,6 +74,129 @@ public void testRejectionRatioReset() throws InterruptedException {
assertFalse(loadController.isOverloaded());
}
+ @Test(dataProvider = "counterPaths")
+ public void testNotOverloadedWhenAcceptKeepsUp(boolean useIndependentCounter) {
+ LoadController loadController = LoadController.newBuilder()
+ .setWindowSizeInSec(5)
+ .setRejectionRatioUpdateIntervalInSec(1)
+ .setMaxRejectionRatio(0.5)
+ .setAcceptMultiplier(1.0)
+ .setUseIndependentCounter(useIndependentCounter)
+ .build();
+ for (int i = 0; i < 100; i++) {
+ loadController.recordRequest();
+ loadController.recordAccept();
+ }
+ assertEquals(loadController.getRejectionRatio(), 0.0d);
+ assertFalse(loadController.isOverloaded());
+ assertFalse(loadController.shouldRejectRequest());
+ }
+
+ @Test(dataProvider = "counterPaths")
+ public void testMaxRejectionRatioIsRespected(boolean useIndependentCounter) {
+ LoadController loadController = LoadController.newBuilder()
+ .setWindowSizeInSec(5)
+ .setRejectionRatioUpdateIntervalInSec(1)
+ .setMaxRejectionRatio(0.25)
+ .setAcceptMultiplier(2.0)
+ .setUseIndependentCounter(useIndependentCounter)
+ .build();
+ // 1000 requests, 0 accepts — raw ratio would approach 1.0 but must be capped at 0.25.
+ for (int i = 0; i < 1000; i++) {
+ loadController.recordRequest();
+ }
+ assertEquals(loadController.getRejectionRatio(), 0.25d, 1e-9);
+ }
+
+ @Test(dataProvider = "counterPaths")
+ public void testZeroRequestsMeansZeroRatio(boolean useIndependentCounter) {
+ LoadController loadController = LoadController.newBuilder()
+ .setWindowSizeInSec(5)
+ .setRejectionRatioUpdateIntervalInSec(1)
+ .setMaxRejectionRatio(0.5)
+ .setAcceptMultiplier(2.0)
+ .setUseIndependentCounter(useIndependentCounter)
+ .build();
+ assertEquals(loadController.getRejectionRatio(), 0.0d);
+ assertFalse(loadController.isOverloaded());
+ assertFalse(loadController.shouldRejectRequest());
+ }
+
+ @Test(dataProvider = "counterPaths")
+ public void testRejectionRatioIsCachedBetweenUpdates(boolean useIndependentCounter) {
+ TestMockTime mockTime = new TestMockTime(0);
+ LoadController loadController = LoadController.newBuilder()
+ .setWindowSizeInSec(10)
+ .setRejectionRatioUpdateIntervalInSec(5)
+ .setMaxRejectionRatio(1.0)
+ .setAcceptMultiplier(2.0)
+ .setTime(mockTime)
+ .setUseIndependentCounter(useIndependentCounter)
+ .build();
+ for (int i = 0; i < 100; i++) {
+ loadController.recordRequest();
+ }
+ double first = loadController.getRejectionRatio();
+ assertTrue(first > 0);
+
+ // Record a flood of accepts, but without advancing time past the update interval the cached
+ // ratio must be returned unchanged.
+ for (int i = 0; i < 200; i++) {
+ loadController.recordAccept();
+ }
+ assertEquals(loadController.getRejectionRatio(), first, 1e-9);
+
+ // Advance past the update interval; the ratio must refresh and drop toward 0 because accepts
+ // now outpace the accept-multiplier adjusted requests.
+ mockTime.addMilliseconds(TimeUnit.SECONDS.toMillis(6));
+ double refreshed = loadController.getRejectionRatio();
+ assertTrue(refreshed < first);
+ }
+
+ /**
+ * Parity: feed identical sequences to both the Tehuti-backed and the independent-counter
+ * instances and assert their rejection ratios are equivalent for the scenarios the production
+ * code actually hits. Ensures flipping the flag does not change externally observable behavior.
+ */
+ @Test
+ public void testBothCounterPathsProduceEquivalentRatios() {
+ LoadController tehuti = LoadController.newBuilder()
+ .setWindowSizeInSec(5)
+ .setRejectionRatioUpdateIntervalInSec(1)
+ .setMaxRejectionRatio(1.0)
+ .setAcceptMultiplier(2.0)
+ .setUseIndependentCounter(false)
+ .build();
+ LoadController independent = LoadController.newBuilder()
+ .setWindowSizeInSec(5)
+ .setRejectionRatioUpdateIntervalInSec(1)
+ .setMaxRejectionRatio(1.0)
+ .setAcceptMultiplier(2.0)
+ .setUseIndependentCounter(true)
+ .build();
+
+ // Overload — 100 requests, 30 accepts.
+ for (int i = 0; i < 100; i++) {
+ tehuti.recordRequest();
+ independent.recordRequest();
+ }
+ for (int i = 0; i < 30; i++) {
+ tehuti.recordAccept();
+ independent.recordAccept();
+ }
+ assertEquals(tehuti.getRejectionRatio(), independent.getRejectionRatio(), 1e-6);
+
+ // Steady state on top — heavy balanced traffic. Allow a small tolerance: the two counters use
+ // independent bucket structures, so cached ratios may differ by at most one update cycle.
+ for (int i = 0; i < 500; i++) {
+ tehuti.recordRequest();
+ tehuti.recordAccept();
+ independent.recordRequest();
+ independent.recordAccept();
+ }
+ assertEquals(tehuti.getRejectionRatio(), independent.getRejectionRatio(), 0.02);
+ }
+
@Test
public void testCompletableAllOf() {
CompletableFuture future1 = new CompletableFuture<>();
diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/routing/HelixGroupStatsTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/routing/HelixGroupStatsTest.java
index fd2036c00bf..28515c2bae1 100644
--- a/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/routing/HelixGroupStatsTest.java
+++ b/internal/venice-client-common/src/test/java/com/linkedin/venice/stats/routing/HelixGroupStatsTest.java
@@ -18,16 +18,22 @@
import io.tehuti.Metric;
import io.tehuti.metrics.MetricsRepository;
import java.util.Map;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
public class HelixGroupStatsTest {
- @Test
- public void testMetrics() {
+ @DataProvider(name = "groupAvgPaths")
+ public Object[][] groupAvgPaths() {
+ return new Object[][] { { false }, { true } };
+ }
+
+ @Test(dataProvider = "groupAvgPaths")
+ public void testMetrics(boolean useSelfContainedStats) {
MetricsRepository metricsRepository = MetricsRepositoryUtils.createSingleThreadedMetricsRepository();
String storeName = "test_store";
- HelixGroupStats stats = new HelixGroupStats(metricsRepository, storeName);
+ HelixGroupStats stats = new HelixGroupStats(metricsRepository, storeName, useSelfContainedStats);
// No data points
assertEquals(stats.getGroupResponseWaitingTimeAvg(0), -1d);
@@ -282,8 +288,8 @@ public void testGroupPendingRequestMetric() {
otelMetricPrefix);
}
- @Test
- public void testGroupResponseWaitingTimeMetric() {
+ @Test(dataProvider = "groupAvgPaths")
+ public void testGroupResponseWaitingTimeMetric(boolean useSelfContainedStats) {
// Set up Venice metrics repository with both Tehuti and OpenTelemetry support
InMemoryMetricReader inMemoryMetricReader = InMemoryMetricReader.create();
VeniceMetricsRepository metricsRepository =
@@ -291,7 +297,7 @@ public void testGroupResponseWaitingTimeMetric() {
String otelMetricPrefix = FAST_CLIENT.getMetricsPrefix();
String storeName = "test_store";
- HelixGroupStats stats = new HelixGroupStats(metricsRepository, storeName);
+ HelixGroupStats stats = new HelixGroupStats(metricsRepository, storeName, useSelfContainedStats);
// Record response waiting times for different groups
int groupId0 = 0;
@@ -329,7 +335,8 @@ public void testGroupResponseWaitingTimeMetric() {
double avgGroup2 = stats.getGroupResponseWaitingTimeAvg(groupId2);
assertEquals(avgGroup2, 55.0, 0.01, "Group 2 average response waiting time should be 55.0ms");
- // Verify Tehuti metric names exist
+ // Verify Tehuti metric names exist — both paths keep Tehuti joint recording active for
+ // dashboards, so these must be present regardless of flag.
String expectedTehutiMetricNameGroup0 = "." + storeName + "_HelixGroupStats--group_0_response_waiting_time.Avg";
assertNotNull(
metrics.get(expectedTehutiMetricNameGroup0),
@@ -395,4 +402,44 @@ public void testGroupResponseWaitingTimeMetric() {
otelMetricPrefix);
}
+ /**
+ * Parity: two HelixGroupStats instances (one with Tehuti read path, one with independent read
+ * path) receive the identical recording sequence; asserts both report the same per-group avg.
+ * Guards against flipping the flag changing the routing decision.
+ */
+ @Test
+ public void testBothGroupAvgPathsAgree() {
+ MetricsRepository metricsRepositoryTehuti = new MetricsRepository();
+ MetricsRepository metricsRepositoryIndependent = new MetricsRepository();
+ String storeName = "parity_store";
+
+ HelixGroupStats tehuti = new HelixGroupStats(metricsRepositoryTehuti, storeName, false);
+ HelixGroupStats independent = new HelixGroupStats(metricsRepositoryIndependent, storeName, true);
+
+ double[] group0 = { 50, 100, 75 };
+ double[] group1 = { 120, 80 };
+ double[] group2 = { 30, 90, 60, 40 };
+
+ for (double v: group0) {
+ tehuti.recordGroupResponseWaitingTime(0, v);
+ independent.recordGroupResponseWaitingTime(0, v);
+ }
+ for (double v: group1) {
+ tehuti.recordGroupResponseWaitingTime(1, v);
+ independent.recordGroupResponseWaitingTime(1, v);
+ }
+ for (double v: group2) {
+ tehuti.recordGroupResponseWaitingTime(2, v);
+ independent.recordGroupResponseWaitingTime(2, v);
+ }
+
+ assertEquals(tehuti.getGroupResponseWaitingTimeAvg(0), independent.getGroupResponseWaitingTimeAvg(0), 0.01);
+ assertEquals(tehuti.getGroupResponseWaitingTimeAvg(1), independent.getGroupResponseWaitingTimeAvg(1), 0.01);
+ assertEquals(tehuti.getGroupResponseWaitingTimeAvg(2), independent.getGroupResponseWaitingTimeAvg(2), 0.01);
+
+ // Absent group returns -1 on both paths.
+ assertEquals(tehuti.getGroupResponseWaitingTimeAvg(99), -1d);
+ assertEquals(independent.getGroupResponseWaitingTimeAvg(99), -1d);
+ }
+
}
diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/utils/concurrent/SlidingWindowAverageTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/utils/concurrent/SlidingWindowAverageTest.java
new file mode 100644
index 00000000000..ced70f45f09
--- /dev/null
+++ b/internal/venice-client-common/src/test/java/com/linkedin/venice/utils/concurrent/SlidingWindowAverageTest.java
@@ -0,0 +1,105 @@
+package com.linkedin.venice.utils.concurrent;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+import com.linkedin.venice.utils.TestMockTime;
+import io.tehuti.metrics.MetricConfig;
+import io.tehuti.metrics.MetricsRepository;
+import io.tehuti.metrics.Sensor;
+import io.tehuti.metrics.stats.Avg;
+import io.tehuti.utils.SystemTime;
+import java.util.concurrent.TimeUnit;
+import org.testng.annotations.Test;
+
+
+public class SlidingWindowAverageTest {
+ @Test
+ public void testInvalidArgs() {
+ assertThrows(IllegalArgumentException.class, () -> new SlidingWindowAverage(0));
+ assertThrows(IllegalArgumentException.class, () -> new SlidingWindowAverage(-1));
+ assertThrows(IllegalArgumentException.class, () -> new SlidingWindowAverage(1000, null));
+ }
+
+ @Test
+ public void testBasicAverage() {
+ TestMockTime time = new TestMockTime(0);
+ SlidingWindowAverage avg = new SlidingWindowAverage(10_000, time);
+
+ assertTrue(Double.isNaN(avg.average()));
+ avg.record(50);
+ avg.record(100);
+ avg.record(75);
+ assertEquals(avg.average(), 75.0, 1e-9);
+ }
+
+ @Test
+ public void testAverageCarriesAcrossOneBucketRotation() {
+ TestMockTime time = new TestMockTime(0);
+ SlidingWindowAverage avg = new SlidingWindowAverage(10_000, time);
+ avg.record(10);
+ avg.record(20);
+ time.addMilliseconds(6_000);
+ avg.record(30);
+ // All three observations still visible across both buckets.
+ assertEquals(avg.average(), 20.0, 1e-9);
+ }
+
+ @Test
+ public void testWholeWindowExpiry() {
+ TestMockTime time = new TestMockTime(0);
+ SlidingWindowAverage avg = new SlidingWindowAverage(3_000, time);
+ avg.record(100);
+ time.addMilliseconds(10_000);
+ assertTrue(Double.isNaN(avg.average()));
+ }
+
+ @Test
+ public void testCacheServesStaleDuringTtl() {
+ TestMockTime time = new TestMockTime(0);
+ SlidingWindowAverage avg = new SlidingWindowAverage(10_000, time, SlidingWindowAverage.CACHE_MS);
+ avg.record(100);
+ assertEquals(avg.average(), 100.0, 1e-9); // primes cache
+
+ avg.record(300); // new record; cache still live
+ time.addMilliseconds(50); // within CACHE_MS
+ assertEquals(avg.average(), 100.0, 1e-9, "should return cached value within TTL");
+
+ time.addMilliseconds(60); // past CACHE_MS
+ assertEquals(avg.average(), 200.0, 1e-9, "should recompute after TTL expires");
+ }
+
+ @Test
+ public void testCacheRefreshesAfterWindowExpiry() {
+ TestMockTime time = new TestMockTime(0);
+ SlidingWindowAverage avg = new SlidingWindowAverage(10_000, time);
+ avg.record(50);
+ assertEquals(avg.average(), 50.0, 1e-9); // prime cache
+ time.addMilliseconds(30_000); // both buckets expire + TTL expired
+ assertTrue(Double.isNaN(avg.average()));
+ }
+
+ /** Parity: SlidingWindowAverage.average() matches Tehuti Avg for the same values. */
+ @Test
+ public void testAverageMatchesTehutiAvg() {
+ TestMockTime time = new TestMockTime(0);
+ SlidingWindowAverage avg = new SlidingWindowAverage(30_000, time);
+
+ MetricConfig config = new MetricConfig().timeWindow(30, TimeUnit.SECONDS);
+ MetricsRepository repo = new MetricsRepository(config);
+ Avg tehutiAvg = new Avg();
+ Sensor sensor = repo.sensor("avg_parity");
+ sensor.add("avg_parity", tehutiAvg);
+
+ double[] values = { 10, 20, 30, 40, 50, 60, 70, 80, 90, 100 };
+ for (double v: values) {
+ avg.record(v);
+ sensor.record(v);
+ }
+
+ double expected = 55.0;
+ assertEquals(avg.average(), expected, 1e-9);
+ assertEquals(tehutiAvg.measure(config, new SystemTime().milliseconds()), expected, 1e-9);
+ }
+}
diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java
index f6404adc76b..88b2e158a17 100644
--- a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java
+++ b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java
@@ -1200,6 +1200,8 @@ private ConfigKeys() {
public static final String SERVER_ADAPTIVE_THROTTLER_SIGNAL_REFRESH_INTERVAL_IN_SECONDS =
"server.adaptive.throttler.signal.refresh.interval.in.seconds";
+ // use.self.contained.stats is defined in VeniceMetricsConfig as a cross-cutting config.
+
/**
* A list of JVM arguments for forked child process, separated by semicolon.
*/
diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/TestVeniceServer.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/TestVeniceServer.java
index b5322ec7cc3..4d110d6d59f 100644
--- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/TestVeniceServer.java
+++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/TestVeniceServer.java
@@ -17,6 +17,7 @@
import com.linkedin.venice.security.SSLFactory;
import com.linkedin.venice.server.VeniceServer;
import com.linkedin.venice.server.VeniceServerContext;
+import com.linkedin.venice.utils.concurrent.LatencyPercentileProvider;
import io.netty.channel.ChannelHandlerContext;
import io.tehuti.metrics.MetricsRepository;
import java.util.Optional;
@@ -51,7 +52,8 @@ protected ListenerService createListenerService(
Optional storeAccessController,
DiskHealthCheckService diskHealthService,
StorageEngineBackedCompressorFactory compressorFactory,
- Optional resourceReadUsageTracker) {
+ Optional resourceReadUsageTracker,
+ LatencyPercentileProvider latencyPercentileProvider) {
return new ListenerService(
storageEngineRepository,
diff --git a/services/venice-router/src/main/java/com/linkedin/venice/router/api/routing/helix/HelixGroupSelector.java b/services/venice-router/src/main/java/com/linkedin/venice/router/api/routing/helix/HelixGroupSelector.java
index 1831c57d7d5..311a622dc79 100644
--- a/services/venice-router/src/main/java/com/linkedin/venice/router/api/routing/helix/HelixGroupSelector.java
+++ b/services/venice-router/src/main/java/com/linkedin/venice/router/api/routing/helix/HelixGroupSelector.java
@@ -3,6 +3,7 @@
import com.linkedin.alpini.base.concurrency.TimeoutProcessor;
import com.linkedin.venice.exceptions.VeniceException;
import com.linkedin.venice.helix.HelixInstanceConfigRepository;
+import com.linkedin.venice.stats.VeniceMetricsConfig;
import com.linkedin.venice.stats.routing.HelixGroupStats;
import io.tehuti.metrics.MetricsRepository;
import java.util.concurrent.TimeUnit;
@@ -30,7 +31,8 @@ public HelixGroupSelector(
HelixInstanceConfigRepository instanceConfigRepository,
HelixGroupSelectionStrategyEnum strategyEnum,
TimeoutProcessor timeoutProcessor) {
- this.helixGroupStats = new HelixGroupStats(metricsRepository);
+ boolean selfContained = VeniceMetricsConfig.useSelfContainedStats(metricsRepository);
+ this.helixGroupStats = new HelixGroupStats(metricsRepository, selfContained);
this.instanceConfigRepository = instanceConfigRepository;
Class strategyClass = strategyEnum.getStrategyClass();
if (strategyClass.equals(HelixGroupLeastLoadedStrategy.class)) {
diff --git a/services/venice-server/src/main/java/com/linkedin/venice/listener/HttpChannelInitializer.java b/services/venice-server/src/main/java/com/linkedin/venice/listener/HttpChannelInitializer.java
index de57ea5ea0e..c8a6af28eb6 100644
--- a/services/venice-server/src/main/java/com/linkedin/venice/listener/HttpChannelInitializer.java
+++ b/services/venice-server/src/main/java/com/linkedin/venice/listener/HttpChannelInitializer.java
@@ -29,6 +29,7 @@
import com.linkedin.venice.utils.ReflectUtils;
import com.linkedin.venice.utils.SslUtils;
import com.linkedin.venice.utils.Utils;
+import com.linkedin.venice.utils.concurrent.LatencyPercentileProvider;
import io.grpc.ServerInterceptor;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
@@ -82,6 +83,32 @@ public HttpChannelInitializer(
Optional storeAccessController,
StorageReadRequestHandler requestHandler,
StorageEngineRepository storageEngineRepository) {
+ this(
+ storeMetadataRepository,
+ customizedViewRepository,
+ metricsRepository,
+ sslFactory,
+ sslHandshakeExecutor,
+ serverConfig,
+ routerAccessController,
+ storeAccessController,
+ requestHandler,
+ storageEngineRepository,
+ null);
+ }
+
+ public HttpChannelInitializer(
+ ReadOnlyStoreRepository storeMetadataRepository,
+ CompletableFuture customizedViewRepository,
+ MetricsRepository metricsRepository,
+ Optional sslFactory,
+ ThreadPoolExecutor sslHandshakeExecutor,
+ VeniceServerConfig serverConfig,
+ Optional routerAccessController,
+ Optional storeAccessController,
+ StorageReadRequestHandler requestHandler,
+ StorageEngineRepository storageEngineRepository,
+ LatencyPercentileProvider latencyPercentileProvider) {
this.serverConfig = serverConfig;
this.requestHandler = requestHandler;
this.isDaVinciClient = serverConfig.isDaVinciClient();
@@ -98,7 +125,8 @@ public HttpChannelInitializer(
storeMetadataRepository,
isUnregisterMetricForDeletedStoreEnabled,
isDaVinciClient,
- readOtelStatsEnabled);
+ readOtelStatsEnabled,
+ latencyPercentileProvider);
this.multiGetStats = new AggServerHttpRequestStats(
serverConfig.getClusterName(),
metricsRepository,
@@ -107,7 +135,8 @@ public HttpChannelInitializer(
storeMetadataRepository,
isUnregisterMetricForDeletedStoreEnabled,
isDaVinciClient,
- readOtelStatsEnabled);
+ readOtelStatsEnabled,
+ latencyPercentileProvider);
this.computeStats = new AggServerHttpRequestStats(
serverConfig.getClusterName(),
metricsRepository,
@@ -116,7 +145,8 @@ public HttpChannelInitializer(
storeMetadataRepository,
isUnregisterMetricForDeletedStoreEnabled,
isDaVinciClient,
- readOtelStatsEnabled);
+ readOtelStatsEnabled,
+ latencyPercentileProvider);
if (serverConfig.isComputeFastAvroEnabled()) {
LOGGER.info("Fast avro for compute is enabled");
@@ -184,7 +214,8 @@ public HttpChannelInitializer(
if (serverConfig.isLoadControllerEnabled()) {
this.loadControllerHandler = new ServerLoadControllerHandler(
serverConfig,
- new ServerLoadStats(metricsRepository, "server_load", serverConfig.getClusterName()));
+ new ServerLoadStats(metricsRepository, "server_load", serverConfig.getClusterName()),
+ metricsRepository);
LOGGER.info("Server load controller is enabled");
} else {
this.loadControllerHandler = null;
diff --git a/services/venice-server/src/main/java/com/linkedin/venice/listener/ListenerService.java b/services/venice-server/src/main/java/com/linkedin/venice/listener/ListenerService.java
index ce3ddd3a97b..236266d8759 100644
--- a/services/venice-server/src/main/java/com/linkedin/venice/listener/ListenerService.java
+++ b/services/venice-server/src/main/java/com/linkedin/venice/listener/ListenerService.java
@@ -19,6 +19,7 @@
import com.linkedin.venice.security.SSLFactory;
import com.linkedin.venice.service.AbstractVeniceService;
import com.linkedin.venice.stats.ThreadPoolStats;
+import com.linkedin.venice.utils.concurrent.LatencyPercentileProvider;
import com.linkedin.venice.utils.concurrent.ThreadPoolFactory;
import io.grpc.ServerInterceptor;
import io.netty.bootstrap.ServerBootstrap;
@@ -79,6 +80,40 @@ public ListenerService(
DiskHealthCheckService diskHealthService,
StorageEngineBackedCompressorFactory compressorFactory,
Optional resourceReadUsageTracker) {
+ this(
+ storageEngineRepository,
+ storeMetadataRepository,
+ schemaRepository,
+ customizedViewRepository,
+ ingestionMetadataRetriever,
+ readMetadataRetriever,
+ serverConfig,
+ metricsRepository,
+ sslFactory,
+ routerAccessController,
+ storeAccessController,
+ diskHealthService,
+ compressorFactory,
+ resourceReadUsageTracker,
+ null);
+ }
+
+ public ListenerService(
+ StorageEngineRepository storageEngineRepository,
+ ReadOnlyStoreRepository storeMetadataRepository,
+ ReadOnlySchemaRepository schemaRepository,
+ CompletableFuture customizedViewRepository,
+ IngestionMetadataRetriever ingestionMetadataRetriever,
+ ReadMetadataRetriever readMetadataRetriever,
+ VeniceServerConfig serverConfig,
+ MetricsRepository metricsRepository,
+ Optional sslFactory,
+ Optional routerAccessController,
+ Optional storeAccessController,
+ DiskHealthCheckService diskHealthService,
+ StorageEngineBackedCompressorFactory compressorFactory,
+ Optional resourceReadUsageTracker,
+ LatencyPercentileProvider latencyPercentileProvider) {
this.serverConfig = serverConfig;
this.port = serverConfig.getListenerPort();
@@ -126,7 +161,8 @@ public ListenerService(
routerAccessController,
storeAccessController,
requestHandler,
- storageEngineRepository);
+ storageEngineRepository,
+ latencyPercentileProvider);
Class serverSocketChannelClass = NioServerSocketChannel.class;
boolean epollEnabled = serverConfig.isRestServiceEpollEnabled();
diff --git a/services/venice-server/src/main/java/com/linkedin/venice/listener/ServerLoadControllerHandler.java b/services/venice-server/src/main/java/com/linkedin/venice/listener/ServerLoadControllerHandler.java
index 18ca9624c07..34707d14f95 100644
--- a/services/venice-server/src/main/java/com/linkedin/venice/listener/ServerLoadControllerHandler.java
+++ b/services/venice-server/src/main/java/com/linkedin/venice/listener/ServerLoadControllerHandler.java
@@ -7,12 +7,14 @@
import com.linkedin.venice.read.RequestType;
import com.linkedin.venice.reliability.LoadController;
import com.linkedin.venice.stats.ServerLoadStats;
+import com.linkedin.venice.stats.VeniceMetricsConfig;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.ReferenceCountUtil;
+import io.tehuti.metrics.MetricsRepository;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -36,13 +38,18 @@ public class ServerLoadControllerHandler extends SimpleChannelInboundHandler createServices() {
storeAccessController,
diskHealthCheckService,
compressorFactory,
- resourceReadUsageTracker);
+ resourceReadUsageTracker,
+ adaptiveThrottlerSignalService != null ? adaptiveThrottlerSignalService.getLatencyPercentileProvider() : null);
services.add(listenerService);
/**
@@ -851,7 +853,8 @@ protected ListenerService createListenerService(
Optional storeAccessController,
DiskHealthCheckService diskHealthService,
StorageEngineBackedCompressorFactory compressorFactory,
- Optional resourceReadUsageTracker) {
+ Optional resourceReadUsageTracker,
+ LatencyPercentileProvider latencyPercentileProvider) {
return new ListenerService(
storageEngineRepository,
storeMetadataRepository,
@@ -866,7 +869,8 @@ protected ListenerService createListenerService(
storeAccessController,
diskHealthService,
compressorFactory,
- resourceReadUsageTracker);
+ resourceReadUsageTracker,
+ latencyPercentileProvider);
}
public static void main(String args[]) throws Exception {
diff --git a/services/venice-server/src/main/java/com/linkedin/venice/stats/AggServerHttpRequestStats.java b/services/venice-server/src/main/java/com/linkedin/venice/stats/AggServerHttpRequestStats.java
index d058bf9b97e..4744a06849d 100644
--- a/services/venice-server/src/main/java/com/linkedin/venice/stats/AggServerHttpRequestStats.java
+++ b/services/venice-server/src/main/java/com/linkedin/venice/stats/AggServerHttpRequestStats.java
@@ -6,6 +6,7 @@
import com.linkedin.venice.stats.dimensions.HttpResponseStatusCodeCategory;
import com.linkedin.venice.stats.dimensions.HttpResponseStatusEnum;
import com.linkedin.venice.stats.dimensions.VeniceResponseStatusCategory;
+import com.linkedin.venice.utils.concurrent.LatencyPercentileProvider;
import io.tehuti.metrics.MetricsRepository;
@@ -23,6 +24,28 @@ public AggServerHttpRequestStats(
boolean unregisterMetricForDeletedStoreEnabled,
boolean isDaVinciClient,
boolean readOtelStatsEnabled) {
+ this(
+ clusterName,
+ metricsRepository,
+ requestType,
+ isKeyValueProfilingEnabled,
+ metadataRepository,
+ unregisterMetricForDeletedStoreEnabled,
+ isDaVinciClient,
+ readOtelStatsEnabled,
+ null);
+ }
+
+ public AggServerHttpRequestStats(
+ String clusterName,
+ MetricsRepository metricsRepository,
+ RequestType requestType,
+ boolean isKeyValueProfilingEnabled,
+ ReadOnlyStoreRepository metadataRepository,
+ boolean unregisterMetricForDeletedStoreEnabled,
+ boolean isDaVinciClient,
+ boolean readOtelStatsEnabled,
+ LatencyPercentileProvider latencyPercentileProvider) {
super(
clusterName,
metricsRepository,
@@ -30,7 +53,8 @@ public AggServerHttpRequestStats(
requestType,
isKeyValueProfilingEnabled,
isDaVinciClient,
- readOtelStatsEnabled),
+ readOtelStatsEnabled,
+ latencyPercentileProvider),
metadataRepository,
unregisterMetricForDeletedStoreEnabled,
false);
@@ -41,16 +65,19 @@ static class ServerHttpRequestStatsSupplier implements StatsSupplier