From ab2b842f5642115d9e9cd834b13392a62751f60e Mon Sep 17 00:00:00 2001 From: Jason Koch Date: Wed, 13 May 2026 23:44:48 +0000 Subject: [PATCH 1/4] Add a metric that exposes IO thread utilization This allows to spot issues where evcache is overloaded and supports some basic prediction of whether an instance might become overloaded with a significant traffic increase or whether the thread pool could be tuned. --- .../netflix/evcache/test/EVCacheTestDI.java | 35 +++++ .../metrics/EVCacheMetricsFactory.java | 1 + .../netflix/evcache/pool/EVCacheClient.java | 15 ++ .../evcache/pool/EVCacheLoopProbe.java | 133 ++++++++++++++++++ .../net/spy/memcached/EVCacheConnection.java | 20 +++ .../spy/memcached/EVCacheMemcachedClient.java | 5 + .../evcache/pool/EVCacheLoopProbeTest.java | 120 ++++++++++++++++ evcache-core/src/test/java/test-suite.xml | 1 + 8 files changed, 330 insertions(+) create mode 100644 evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java create mode 100644 evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java diff --git a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java index e69280e3..63cc7677 100644 --- a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java +++ b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java @@ -11,12 +11,17 @@ import com.netflix.evcache.EVCacheException; import com.netflix.evcache.EVCacheGetOperationListener; import com.netflix.evcache.EVCacheLatch; +import com.netflix.evcache.metrics.EVCacheMetricsFactory; import com.netflix.evcache.operation.EVCacheOperationFuture; import com.netflix.evcache.pool.EVCacheClient; import com.netflix.evcache.pool.ServerGroup; import com.netflix.evcache.test.transcoder.Movie; import com.netflix.evcache.test.transcoder.MovieTranscoder; import com.netflix.evcache.util.KeyHasher; +import com.netflix.spectator.api.Gauge; +import com.netflix.spectator.api.Id; +import com.netflix.spectator.api.Registry; +import com.netflix.spectator.api.patterns.PolledMeter; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -76,6 +81,36 @@ public void testEVCache() { assertNotNull(evCache); } + @Test(dependsOnMethods = { "testGet" }) + public void testLoopCpuUtilizationMetricRegistered() throws Exception { + final Registry registry = EVCacheMetricsFactory.getInstance().getRegistry(); + final Map> clientsByServerGroup = manager.getEVCacheClientPool(appName).getAllInstancesByServerGroup(); + assertFalse(clientsByServerGroup.isEmpty(), "expected EVCache clients for " + appName); + + PolledMeter.update(registry); + for (List clients : clientsByServerGroup.values()) { + for (EVCacheClient client : clients) { + final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, client.getTagList()); + assertTrue(registry.state().containsKey(id), "expected loop CPU utilization meter for client " + client); + } + } + + boolean nonZero = false; + for (int attempt = 0; attempt < 10 && !nonZero; attempt++) { + get(0, evCache); + Thread.sleep(1_100); + PolledMeter.update(registry); + for (List clients : clientsByServerGroup.values()) { + for (EVCacheClient client : clients) { + final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, client.getTagList()); + final Gauge gauge = registry.gauge(id); + nonZero |= gauge.value() > 0.0; + } + } + } + assertTrue(nonZero, "expected loop CPU utilization meter to report a non-zero value"); + } + @Test(dependsOnMethods = { "testEVCache" }) public void testKeySizeCheck() throws Exception { final String key = "This is an invalid key"; diff --git a/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java b/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java index eeb7941d..2de0adda 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java +++ b/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java @@ -284,6 +284,7 @@ public String getStatusCode(StatusCode sc) { public static final String INTERNAL_EXECUTOR = "internal.evc.client.executor"; public static final String INTERNAL_EXECUTOR_SCHEDULED = "internal.evc.client.scheduledExecutor"; public static final String INTERNAL_POOL_INIT_ERROR = "internal.evc.client.init.error"; + public static final String INTERNAL_LOOP_CPU_UTILIZATION = "internal.evc.client.loop.cpuUtilization"; public static final String INTERNAL_NUM_CHUNK_SIZE = "internal.evc.client.chunking.numOfChunks"; public static final String INTERNAL_CHUNK_DATA_SIZE = "internal.evc.client.chunking.dataSize"; diff --git a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java index 3a0dbcd9..6895389f 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java +++ b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java @@ -20,7 +20,10 @@ import com.netflix.evcache.util.KeyHasher.HashingAlgorithm; import com.netflix.spectator.api.BasicTag; import com.netflix.spectator.api.Counter; +import com.netflix.spectator.api.Id; +import com.netflix.spectator.api.Registry; import com.netflix.spectator.api.Tag; +import com.netflix.spectator.api.patterns.PolledMeter; import java.io.BufferedInputStream; import java.io.IOException; import java.io.PrintWriter; @@ -102,6 +105,7 @@ public class EVCacheClient { private final Property ignoreTouch; private List tags; private final Map counterMap = new ConcurrentHashMap(); + private final Id loopCpuUtilizationId; private final Property hashingAlgo; protected final Counter operationsCounter; private final boolean isDuetClient; @@ -133,6 +137,8 @@ public class EVCacheClient { tagList.add(new BasicTag(EVCacheMetricsFactory.STAT_NAME, EVCacheMetricsFactory.POOL_OPERATIONS)); operationsCounter = EVCacheMetricsFactory.getInstance().getCounter(EVCacheMetricsFactory.INTERNAL_STATS, tagList); + final Registry registry = EVCacheMetricsFactory.getInstance().getRegistry(); + this.enableChunking = EVCacheConfig.getInstance().getPropertyRepository().get(this.serverGroup.getName()+ ".chunk.data", Boolean.class).orElseGet(appName + ".chunk.data").orElse(false); this.chunkSize = EVCacheConfig.getInstance().getPropertyRepository().get(this.serverGroup.getName() + ".chunk.size", Integer.class).orElseGet(appName + ".chunk.size").orElse(1180); this.writeBlock = EVCacheConfig.getInstance().getPropertyRepository().get(appName + "." + this.serverGroup.getName() + ".write.block.duration", Integer.class).orElseGet(appName + ".write.block.duration").orElse(25); @@ -141,10 +147,14 @@ public class EVCacheClient { this.ignoreTouch = EVCacheConfig.getInstance().getPropertyRepository().get(appName + "." + this.serverGroup.getName() + ".ignore.touch", Boolean.class).orElseGet(appName + ".ignore.touch").orElse(false); this.connectionFactory = pool.getEVCacheClientPoolManager().getConnectionFactoryProvider().getConnectionFactory(this); + loopCpuUtilizationId = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, this.tags); this.connectionObserver = new EVCacheConnectionObserver(this); this.ignoreInactiveNodes = EVCacheConfig.getInstance().getPropertyRepository().get(appName + ".ignore.inactive.nodes", Boolean.class).orElse(true); this.evcacheMemcachedClient = new EVCacheMemcachedClient(connectionFactory, memcachedNodesInZone, readTimeout, this); + PolledMeter.using(registry) + .withId(loopCpuUtilizationId) + .monitorValue(this.evcacheMemcachedClient.getLoopProbe(), EVCacheLoopProbe::sampleUtilization); this.evcacheMemcachedClient.addObserver(connectionObserver); this.decodingTranscoder = new EVCacheSerializingTranscoder(Integer.MAX_VALUE); @@ -1342,6 +1352,11 @@ public boolean shutdown(long timeout, TimeUnit unit) { if(shutdown) return true; shutdown = true; + try { + PolledMeter.remove(EVCacheMetricsFactory.getInstance().getRegistry(), loopCpuUtilizationId); + } catch(Throwable t) { + log.warn("Exception while removing loop CPU utilization meter", t); + } try { evcacheMemcachedClient.shutdown(timeout, unit); } catch(Throwable t) { diff --git a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java new file mode 100644 index 00000000..9b19c7d4 --- /dev/null +++ b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java @@ -0,0 +1,133 @@ +package com.netflix.evcache.pool; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadMXBean; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Publishes event-loop CPU utilization from the loop thread itself. + * + *

The loop thread periodically publishes an immutable {@code long[]} snapshot + * containing {@code {threadCpuNs, wallNs}}. Spectator's polling thread reads the + * latest snapshot and computes the delta ratio without performing cross-thread + * ThreadMXBean lookups.

+ */ +public final class EVCacheLoopProbe { + private static final Logger log = LoggerFactory.getLogger(EVCacheLoopProbe.class); + private static final ThreadMXBean THREAD_MX_BEAN = ManagementFactory.getThreadMXBean(); + private static final long PUBLISH_INTERVAL_NS = TimeUnit.MILLISECONDS.toNanos(1_000); + private static final int CPU_UTILIZATION_WARNING_THRESHOLD = 3; + + private final AtomicReference snapshot = new AtomicReference(new long[] { 0L, 0L }); + private final boolean cpuTimeAvailable; + + // Loop-thread-private throttle state. + private long nextPublishNs; + private boolean tickFailureLogged; + private boolean negativeCpuTimeLogged; + + // PolledMeter-reader-private state. + private long prevCpuNs; + private long prevWallNs; + private int aboveOneSamples; + private boolean aboveOneLogged; + + public EVCacheLoopProbe() { + this.cpuTimeAvailable = isCurrentThreadCpuTimeAvailable(); + if (!cpuTimeAvailable) { + log.warn("Thread CPU time is not available; EVCache loop CPU utilization will report NaN"); + } + } + + /** + * Publish the current thread's CPU time and wall time at most every second. + * + *

This method is intentionally no-throw: it is called from the EVCache IO + * loop in a finally block and must never terminate the loop.

+ */ + public void tick() { + try { + tickInternal(); + } catch (Throwable t) { + if (!tickFailureLogged) { + tickFailureLogged = true; + try { + log.warn("EVCache loop CPU utilization probe failed; suppressing future probe errors", t); + } catch (Throwable ignored) { + // Keep the event loop alive even if logging fails. + } + } + } + } + + private void tickInternal() { + if (!cpuTimeAvailable) return; + + final long now = System.nanoTime(); + if (nextPublishNs != 0L && now - nextPublishNs < 0L) return; + nextPublishNs = now + PUBLISH_INTERVAL_NS; + + final long cpuNs = THREAD_MX_BEAN.getCurrentThreadCpuTime(); + if (cpuNs < 0L) { + if (!negativeCpuTimeLogged) { + negativeCpuTimeLogged = true; + log.warn("Thread CPU time returned a negative value; skipping EVCache loop CPU utilization publish"); + } + return; + } + + snapshot.lazySet(new long[] { cpuNs, now }); + } + + /** + * Return loop-thread CPU utilization over the interval since the previous poll. + */ + public double sampleUtilization() { + if (!cpuTimeAvailable) return Double.NaN; + + final long[] s = snapshot.get(); + final long cpuNs = s[0]; + final long wallNs = s[1]; + if (prevWallNs == 0L) { + prevCpuNs = cpuNs; + prevWallNs = wallNs; + return Double.NaN; + } + + final long dWall = wallNs - prevWallNs; + if (dWall <= 0L) return 0.0; + + final long dCpu = cpuNs - prevCpuNs; + prevCpuNs = cpuNs; + prevWallNs = wallNs; + + double utilization = (double) dCpu / (double) dWall; + if (utilization < 0.0) return 0.0; + + if (utilization > 1.0) { + aboveOneSamples++; + if (aboveOneSamples >= CPU_UTILIZATION_WARNING_THRESHOLD && !aboveOneLogged) { + aboveOneLogged = true; + log.warn("EVCache loop CPU utilization exceeded 1.0 for {} consecutive samples; latest value={}", + CPU_UTILIZATION_WARNING_THRESHOLD, utilization); + } + } else { + aboveOneSamples = 0; + } + + return Math.min(utilization, 1.05); + } + + private static boolean isCurrentThreadCpuTimeAvailable() { + try { + return THREAD_MX_BEAN.isThreadCpuTimeSupported() && THREAD_MX_BEAN.isThreadCpuTimeEnabled(); + } catch (Throwable t) { + log.warn("Unable to determine ThreadMXBean CPU-time capability", t); + return false; + } + } +} diff --git a/evcache-core/src/main/java/net/spy/memcached/EVCacheConnection.java b/evcache-core/src/main/java/net/spy/memcached/EVCacheConnection.java index 19784073..ed5489db 100644 --- a/evcache-core/src/main/java/net/spy/memcached/EVCacheConnection.java +++ b/evcache-core/src/main/java/net/spy/memcached/EVCacheConnection.java @@ -11,6 +11,8 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; +import com.netflix.evcache.pool.EVCacheLoopProbe; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,6 +20,7 @@ public class EVCacheConnection extends MemcachedConnection { private static final Logger log = LoggerFactory.getLogger(EVCacheConnection.class); + private EVCacheLoopProbe probe; private final net.spy.memcached.compat.log.Logger spyLogger; public EVCacheConnection(String name, int bufSize, ConnectionFactory f, @@ -28,6 +31,21 @@ public EVCacheConnection(String name, int bufSize, ConnectionFactory f, spyLogger = super.getLogger(); } + @Override + public synchronized void start() { + // MemcachedConnection starts the thread from its constructor. Initialize + // the probe before super.start() so Thread.start() safely publishes it + // to run() without requiring a volatile read on the event-loop path. + if (probe == null) { + probe = new EVCacheLoopProbe(); + } + super.start(); + } + + public EVCacheLoopProbe getProbe() { + return probe; + } + @Override public void shutdown() throws IOException { try { @@ -63,6 +81,8 @@ public void run() { } catch (Throwable e) { log.error("SEVERE EVCACHE ISSUE.", e);// This ensures the thread // doesn't die + } finally { + probe.tick(); } } if (log.isDebugEnabled()) log.debug(toString() + " : Shutdown"); diff --git a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java index 8272f0cb..2d196dea 100644 --- a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java +++ b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java @@ -15,6 +15,7 @@ import com.netflix.evcache.operation.EVCacheOperationFuture; import com.netflix.evcache.pool.EVCacheClient; import com.netflix.evcache.pool.EVCacheClientUtil; +import com.netflix.evcache.pool.EVCacheLoopProbe; import com.netflix.evcache.pool.EVCacheValue; import com.netflix.evcache.util.EVCacheConfig; import com.netflix.spectator.api.BasicTag; @@ -123,6 +124,10 @@ public NodeLocator getNodeLocator() { return this.mconn.getLocator(); } + public EVCacheLoopProbe getLoopProbe() { + return ((EVCacheConnection) this.mconn).getProbe(); + } + public MemcachedNode getEVCacheNode(String key) { return this.mconn.getLocator().getPrimary(key); } diff --git a/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java b/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java new file mode 100644 index 00000000..20e1d134 --- /dev/null +++ b/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java @@ -0,0 +1,120 @@ +package com.netflix.evcache.pool; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadMXBean; + +import com.netflix.evcache.metrics.EVCacheMetricsFactory; +import com.netflix.spectator.api.DefaultRegistry; +import com.netflix.spectator.api.Gauge; +import com.netflix.spectator.api.Id; +import com.netflix.spectator.api.Registry; +import com.netflix.spectator.api.patterns.PolledMeter; + +import org.testng.SkipException; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +public class EVCacheLoopProbeTest { + + @Test + public void firstSampleEstablishesBaseline() throws Exception { + assumeThreadCpuTimeAvailable(); + EVCacheLoopProbe probe = new EVCacheLoopProbe(); + + probe.tick(); + + assertTrue(Double.isNaN(probe.sampleUtilization())); + } + + @Test + public void reportsCpuUtilizationForCurrentThread() throws Exception { + assumeThreadCpuTimeAvailable(); + EVCacheLoopProbe probe = new EVCacheLoopProbe(); + + probe.tick(); + assertTrue(Double.isNaN(probe.sampleUtilization())); + + double utilization = samplePositiveUtilization(probe); + assertTrue(utilization > 0.0, "expected positive utilization, got " + utilization); + assertTrue(utilization <= 1.05, "expected utilization to be clamped, got " + utilization); + } + + @Test + public void returnsZeroWhenNoNewSampleWasPublished() throws Exception { + assumeThreadCpuTimeAvailable(); + EVCacheLoopProbe probe = new EVCacheLoopProbe(); + + probe.tick(); + assertTrue(Double.isNaN(probe.sampleUtilization())); + + assertEquals(probe.sampleUtilization(), 0.0); + } + + @Test + public void polledMeterPublishesProbeUtilization() throws Exception { + assumeThreadCpuTimeAvailable(); + Registry registry = new DefaultRegistry(); + Id id = registry.createId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, "evc.connection.id", "0", "ipc.server.asg", "test"); + EVCacheLoopProbe probe = new EVCacheLoopProbe(); + + try { + PolledMeter.using(registry) + .withId(id) + .monitorValue(probe, EVCacheLoopProbe::sampleUtilization); + + probe.tick(); + PolledMeter.update(registry); + Gauge gauge = registry.gauge(id); + assertTrue(Double.isNaN(gauge.value())); + + double value = pollPositiveUtilization(registry, probe, gauge); + assertTrue(value > 0.0, "expected polled meter to publish positive utilization, got " + value); + assertTrue(value <= 1.05, "expected utilization to be clamped, got " + value); + } finally { + PolledMeter.remove(registry, id); + } + } + + private static double samplePositiveUtilization(EVCacheLoopProbe probe) throws Exception { + double utilization = 0.0; + for (int i = 0; i < 10 && utilization <= 0.0; i++) { + Thread.sleep(1_100); + busySpinForAtLeastMillis(50); + probe.tick(); + utilization = probe.sampleUtilization(); + } + return utilization; + } + + private static double pollPositiveUtilization(Registry registry, EVCacheLoopProbe probe, Gauge gauge) throws Exception { + double utilization = 0.0; + for (int i = 0; i < 10 && utilization <= 0.0; i++) { + Thread.sleep(1_100); + busySpinForAtLeastMillis(50); + probe.tick(); + PolledMeter.update(registry); + utilization = gauge.value(); + } + return utilization; + } + + private static void assumeThreadCpuTimeAvailable() { + ThreadMXBean tmx = ManagementFactory.getThreadMXBean(); + if (!tmx.isThreadCpuTimeSupported() || !tmx.isThreadCpuTimeEnabled()) { + throw new SkipException("thread CPU time is not available on this JVM"); + } + } + + private static void busySpinForAtLeastMillis(long millis) { + long deadline = System.nanoTime() + millis * 1_000_000L; + long value = 0L; + while (System.nanoTime() < deadline) { + value += System.nanoTime(); + } + if (value == 42L) { + throw new AssertionError("unreachable"); + } + } +} diff --git a/evcache-core/src/test/java/test-suite.xml b/evcache-core/src/test/java/test-suite.xml index f031a615..7d7568a6 100644 --- a/evcache-core/src/test/java/test-suite.xml +++ b/evcache-core/src/test/java/test-suite.xml @@ -3,6 +3,7 @@ + From 5c099268e9157bb53e6715f47cef5f4d36724edd Mon Sep 17 00:00:00 2001 From: Jason Koch Date: Fri, 26 Jun 2026 21:04:03 +0000 Subject: [PATCH 2/4] Rename loop CPU utilization metric to cpuWallTimeRatio The previous name implied a percent (0-100) and suggested it measured the loop's own runtime. The value is actually a fraction (0-1) of thread CPU time over wall time, which excludes time the thread is parked in selector.select() and time it was runnable but descheduled. Renaming makes the units and the underlying measurement explicit so consumers don't accidentally treat it as a percent or as a direct measure of loop load. --- .../netflix/evcache/test/EVCacheTestDI.java | 10 ++-- .../metrics/EVCacheMetricsFactory.java | 2 +- .../netflix/evcache/pool/EVCacheClient.java | 12 ++--- .../evcache/pool/EVCacheLoopProbe.java | 30 +++++++----- .../evcache/pool/EVCacheLoopProbeTest.java | 48 +++++++++---------- 5 files changed, 54 insertions(+), 48 deletions(-) diff --git a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java index 63cc7677..b7fc9c76 100644 --- a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java +++ b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java @@ -82,7 +82,7 @@ public void testEVCache() { } @Test(dependsOnMethods = { "testGet" }) - public void testLoopCpuUtilizationMetricRegistered() throws Exception { + public void testLoopCpuWallTimeRatioMetricRegistered() throws Exception { final Registry registry = EVCacheMetricsFactory.getInstance().getRegistry(); final Map> clientsByServerGroup = manager.getEVCacheClientPool(appName).getAllInstancesByServerGroup(); assertFalse(clientsByServerGroup.isEmpty(), "expected EVCache clients for " + appName); @@ -90,8 +90,8 @@ public void testLoopCpuUtilizationMetricRegistered() throws Exception { PolledMeter.update(registry); for (List clients : clientsByServerGroup.values()) { for (EVCacheClient client : clients) { - final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, client.getTagList()); - assertTrue(registry.state().containsKey(id), "expected loop CPU utilization meter for client " + client); + final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_WALL_TIME_RATIO, client.getTagList()); + assertTrue(registry.state().containsKey(id), "expected loop cpuWallTimeRatio meter for client " + client); } } @@ -102,13 +102,13 @@ public void testLoopCpuUtilizationMetricRegistered() throws Exception { PolledMeter.update(registry); for (List clients : clientsByServerGroup.values()) { for (EVCacheClient client : clients) { - final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, client.getTagList()); + final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_WALL_TIME_RATIO, client.getTagList()); final Gauge gauge = registry.gauge(id); nonZero |= gauge.value() > 0.0; } } } - assertTrue(nonZero, "expected loop CPU utilization meter to report a non-zero value"); + assertTrue(nonZero, "expected loop cpuWallTimeRatio meter to report a non-zero value"); } @Test(dependsOnMethods = { "testEVCache" }) diff --git a/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java b/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java index 2de0adda..69375c8e 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java +++ b/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java @@ -284,7 +284,7 @@ public String getStatusCode(StatusCode sc) { public static final String INTERNAL_EXECUTOR = "internal.evc.client.executor"; public static final String INTERNAL_EXECUTOR_SCHEDULED = "internal.evc.client.scheduledExecutor"; public static final String INTERNAL_POOL_INIT_ERROR = "internal.evc.client.init.error"; - public static final String INTERNAL_LOOP_CPU_UTILIZATION = "internal.evc.client.loop.cpuUtilization"; + public static final String INTERNAL_LOOP_CPU_WALL_TIME_RATIO = "internal.evc.client.loop.cpuWallTimeRatio"; public static final String INTERNAL_NUM_CHUNK_SIZE = "internal.evc.client.chunking.numOfChunks"; public static final String INTERNAL_CHUNK_DATA_SIZE = "internal.evc.client.chunking.dataSize"; diff --git a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java index 6895389f..6587bfe5 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java +++ b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java @@ -105,7 +105,7 @@ public class EVCacheClient { private final Property ignoreTouch; private List tags; private final Map counterMap = new ConcurrentHashMap(); - private final Id loopCpuUtilizationId; + private final Id loopCpuWallTimeRatioId; private final Property hashingAlgo; protected final Counter operationsCounter; private final boolean isDuetClient; @@ -147,14 +147,14 @@ public class EVCacheClient { this.ignoreTouch = EVCacheConfig.getInstance().getPropertyRepository().get(appName + "." + this.serverGroup.getName() + ".ignore.touch", Boolean.class).orElseGet(appName + ".ignore.touch").orElse(false); this.connectionFactory = pool.getEVCacheClientPoolManager().getConnectionFactoryProvider().getConnectionFactory(this); - loopCpuUtilizationId = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, this.tags); + loopCpuWallTimeRatioId = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_WALL_TIME_RATIO, this.tags); this.connectionObserver = new EVCacheConnectionObserver(this); this.ignoreInactiveNodes = EVCacheConfig.getInstance().getPropertyRepository().get(appName + ".ignore.inactive.nodes", Boolean.class).orElse(true); this.evcacheMemcachedClient = new EVCacheMemcachedClient(connectionFactory, memcachedNodesInZone, readTimeout, this); PolledMeter.using(registry) - .withId(loopCpuUtilizationId) - .monitorValue(this.evcacheMemcachedClient.getLoopProbe(), EVCacheLoopProbe::sampleUtilization); + .withId(loopCpuWallTimeRatioId) + .monitorValue(this.evcacheMemcachedClient.getLoopProbe(), EVCacheLoopProbe::sampleCpuWallTimeRatio); this.evcacheMemcachedClient.addObserver(connectionObserver); this.decodingTranscoder = new EVCacheSerializingTranscoder(Integer.MAX_VALUE); @@ -1353,9 +1353,9 @@ public boolean shutdown(long timeout, TimeUnit unit) { shutdown = true; try { - PolledMeter.remove(EVCacheMetricsFactory.getInstance().getRegistry(), loopCpuUtilizationId); + PolledMeter.remove(EVCacheMetricsFactory.getInstance().getRegistry(), loopCpuWallTimeRatioId); } catch(Throwable t) { - log.warn("Exception while removing loop CPU utilization meter", t); + log.warn("Exception while removing loop cpuWallTimeRatio meter", t); } try { evcacheMemcachedClient.shutdown(timeout, unit); diff --git a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java index 9b19c7d4..dd66aad5 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java +++ b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java @@ -9,12 +9,18 @@ import org.slf4j.LoggerFactory; /** - * Publishes event-loop CPU utilization from the loop thread itself. + * Publishes the loop thread's CPU-time to wall-time ratio from the loop thread itself. * *

The loop thread periodically publishes an immutable {@code long[]} snapshot * containing {@code {threadCpuNs, wallNs}}. Spectator's polling thread reads the * latest snapshot and computes the delta ratio without performing cross-thread * ThreadMXBean lookups.

+ * + *

The reported value is {@code dCpu/dWall} in [0, 1.05]: the fraction of wall + * time the loop thread was on a CPU. Time parked in {@code selector.select()} + * is correctly excluded, but time the thread was runnable-but-descheduled + * (CPU contention) is also excluded, so this metric is a lower bound on true + * loop demand under CPU pressure.

*/ public final class EVCacheLoopProbe { private static final Logger log = LoggerFactory.getLogger(EVCacheLoopProbe.class); @@ -39,7 +45,7 @@ public final class EVCacheLoopProbe { public EVCacheLoopProbe() { this.cpuTimeAvailable = isCurrentThreadCpuTimeAvailable(); if (!cpuTimeAvailable) { - log.warn("Thread CPU time is not available; EVCache loop CPU utilization will report NaN"); + log.warn("Thread CPU time is not available; EVCache loop cpuWallTimeRatio will report NaN"); } } @@ -56,7 +62,7 @@ public void tick() { if (!tickFailureLogged) { tickFailureLogged = true; try { - log.warn("EVCache loop CPU utilization probe failed; suppressing future probe errors", t); + log.warn("EVCache loop cpuWallTimeRatio probe failed; suppressing future probe errors", t); } catch (Throwable ignored) { // Keep the event loop alive even if logging fails. } @@ -75,7 +81,7 @@ private void tickInternal() { if (cpuNs < 0L) { if (!negativeCpuTimeLogged) { negativeCpuTimeLogged = true; - log.warn("Thread CPU time returned a negative value; skipping EVCache loop CPU utilization publish"); + log.warn("Thread CPU time returned a negative value; skipping EVCache loop cpuWallTimeRatio publish"); } return; } @@ -84,9 +90,9 @@ private void tickInternal() { } /** - * Return loop-thread CPU utilization over the interval since the previous poll. + * Return loop-thread cpu-time / wall-time ratio over the interval since the previous poll. */ - public double sampleUtilization() { + public double sampleCpuWallTimeRatio() { if (!cpuTimeAvailable) return Double.NaN; final long[] s = snapshot.get(); @@ -105,21 +111,21 @@ public double sampleUtilization() { prevCpuNs = cpuNs; prevWallNs = wallNs; - double utilization = (double) dCpu / (double) dWall; - if (utilization < 0.0) return 0.0; + double ratio = (double) dCpu / (double) dWall; + if (ratio < 0.0) return 0.0; - if (utilization > 1.0) { + if (ratio > 1.0) { aboveOneSamples++; if (aboveOneSamples >= CPU_UTILIZATION_WARNING_THRESHOLD && !aboveOneLogged) { aboveOneLogged = true; - log.warn("EVCache loop CPU utilization exceeded 1.0 for {} consecutive samples; latest value={}", - CPU_UTILIZATION_WARNING_THRESHOLD, utilization); + log.warn("EVCache loop cpuWallTimeRatio exceeded 1.0 for {} consecutive samples; latest value={}", + CPU_UTILIZATION_WARNING_THRESHOLD, ratio); } } else { aboveOneSamples = 0; } - return Math.min(utilization, 1.05); + return Math.min(ratio, 1.05); } private static boolean isCurrentThreadCpuTimeAvailable() { diff --git a/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java b/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java index 20e1d134..ca77e326 100644 --- a/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java +++ b/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java @@ -25,20 +25,20 @@ public void firstSampleEstablishesBaseline() throws Exception { probe.tick(); - assertTrue(Double.isNaN(probe.sampleUtilization())); + assertTrue(Double.isNaN(probe.sampleCpuWallTimeRatio())); } @Test - public void reportsCpuUtilizationForCurrentThread() throws Exception { + public void reportsCpuWallTimeRatioForCurrentThread() throws Exception { assumeThreadCpuTimeAvailable(); EVCacheLoopProbe probe = new EVCacheLoopProbe(); probe.tick(); - assertTrue(Double.isNaN(probe.sampleUtilization())); + assertTrue(Double.isNaN(probe.sampleCpuWallTimeRatio())); - double utilization = samplePositiveUtilization(probe); - assertTrue(utilization > 0.0, "expected positive utilization, got " + utilization); - assertTrue(utilization <= 1.05, "expected utilization to be clamped, got " + utilization); + double ratio = samplePositiveRatio(probe); + assertTrue(ratio > 0.0, "expected positive ratio, got " + ratio); + assertTrue(ratio <= 1.05, "expected ratio to be clamped, got " + ratio); } @Test @@ -47,57 +47,57 @@ public void returnsZeroWhenNoNewSampleWasPublished() throws Exception { EVCacheLoopProbe probe = new EVCacheLoopProbe(); probe.tick(); - assertTrue(Double.isNaN(probe.sampleUtilization())); + assertTrue(Double.isNaN(probe.sampleCpuWallTimeRatio())); - assertEquals(probe.sampleUtilization(), 0.0); + assertEquals(probe.sampleCpuWallTimeRatio(), 0.0); } @Test - public void polledMeterPublishesProbeUtilization() throws Exception { + public void polledMeterPublishesProbeRatio() throws Exception { assumeThreadCpuTimeAvailable(); Registry registry = new DefaultRegistry(); - Id id = registry.createId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_UTILIZATION, "evc.connection.id", "0", "ipc.server.asg", "test"); + Id id = registry.createId(EVCacheMetricsFactory.INTERNAL_LOOP_CPU_WALL_TIME_RATIO, "evc.connection.id", "0", "ipc.server.asg", "test"); EVCacheLoopProbe probe = new EVCacheLoopProbe(); try { PolledMeter.using(registry) .withId(id) - .monitorValue(probe, EVCacheLoopProbe::sampleUtilization); + .monitorValue(probe, EVCacheLoopProbe::sampleCpuWallTimeRatio); probe.tick(); PolledMeter.update(registry); Gauge gauge = registry.gauge(id); assertTrue(Double.isNaN(gauge.value())); - double value = pollPositiveUtilization(registry, probe, gauge); - assertTrue(value > 0.0, "expected polled meter to publish positive utilization, got " + value); - assertTrue(value <= 1.05, "expected utilization to be clamped, got " + value); + double value = pollPositiveRatio(registry, probe, gauge); + assertTrue(value > 0.0, "expected polled meter to publish positive ratio, got " + value); + assertTrue(value <= 1.05, "expected ratio to be clamped, got " + value); } finally { PolledMeter.remove(registry, id); } } - private static double samplePositiveUtilization(EVCacheLoopProbe probe) throws Exception { - double utilization = 0.0; - for (int i = 0; i < 10 && utilization <= 0.0; i++) { + private static double samplePositiveRatio(EVCacheLoopProbe probe) throws Exception { + double ratio = 0.0; + for (int i = 0; i < 10 && ratio <= 0.0; i++) { Thread.sleep(1_100); busySpinForAtLeastMillis(50); probe.tick(); - utilization = probe.sampleUtilization(); + ratio = probe.sampleCpuWallTimeRatio(); } - return utilization; + return ratio; } - private static double pollPositiveUtilization(Registry registry, EVCacheLoopProbe probe, Gauge gauge) throws Exception { - double utilization = 0.0; - for (int i = 0; i < 10 && utilization <= 0.0; i++) { + private static double pollPositiveRatio(Registry registry, EVCacheLoopProbe probe, Gauge gauge) throws Exception { + double ratio = 0.0; + for (int i = 0; i < 10 && ratio <= 0.0; i++) { Thread.sleep(1_100); busySpinForAtLeastMillis(50); probe.tick(); PolledMeter.update(registry); - utilization = gauge.value(); + ratio = gauge.value(); } - return utilization; + return ratio; } private static void assumeThreadCpuTimeAvailable() { From df88ad1bd3c3f4bfb6757d252b43fdce92588065 Mon Sep 17 00:00:00 2001 From: Jason Koch Date: Thu, 14 May 2026 17:14:46 +0000 Subject: [PATCH 3/4] Add a metric to capture write enqueue latency This captures from when the caller tries to send a request until the request is finally sent on the wire, primarily so we can see when the enqueue time is increasing (a signal for pressure on the IO loop). --- .../netflix/evcache/test/EVCacheTestDI.java | 26 +++++++++ .../metrics/EVCacheMetricsFactory.java | 1 + .../operation/EVCacheBulkGetFuture.java | 7 +++ .../operation/EVCacheOperationFuture.java | 8 ++- .../netflix/evcache/pool/EVCacheClient.java | 27 +++++++++ .../spy/memcached/EVCacheMemcachedClient.java | 2 +- .../EVCacheBulkGetFutureLatencyTest.java | 41 +++++++++++++ .../EVCacheOperationFutureLatencyTest.java | 57 +++++++++++++++++++ evcache-core/src/test/java/test-suite.xml | 2 + 9 files changed, 169 insertions(+), 2 deletions(-) create mode 100644 evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheBulkGetFutureLatencyTest.java create mode 100644 evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheOperationFutureLatencyTest.java diff --git a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java index b7fc9c76..1a4007ba 100644 --- a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java +++ b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java @@ -21,6 +21,7 @@ import com.netflix.spectator.api.Gauge; import com.netflix.spectator.api.Id; import com.netflix.spectator.api.Registry; +import com.netflix.spectator.api.Timer; import com.netflix.spectator.api.patterns.PolledMeter; import java.util.HashMap; import java.util.List; @@ -111,6 +112,31 @@ public void testLoopCpuWallTimeRatioMetricRegistered() throws Exception { assertTrue(nonZero, "expected loop cpuWallTimeRatio meter to report a non-zero value"); } + @Test(dependsOnMethods = { "testLoopCpuUtilizationMetricRegistered" }) + public void testLoopEnqueueToWriteLatencyMetricRecords() throws Exception { + final Registry registry = EVCacheMetricsFactory.getInstance().getRegistry(); + final Map> clientsByServerGroup = manager.getEVCacheClientPool(appName).getAllInstancesByServerGroup(); + assertFalse(clientsByServerGroup.isEmpty(), "expected EVCache clients for " + appName); + + boolean recorded = false; + for (int attempt = 0; attempt < 10 && !recorded; attempt++) { + get(attempt, evCache); + Thread.sleep(50); + for (List clients : clientsByServerGroup.values()) { + for (EVCacheClient client : clients) { + final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_ENQUEUE_TO_WRITE_LATENCY, client.getTagList()); + final Timer timer = registry.timer(id); + if (timer.count() > 0L) { + recorded = true; + break; + } + } + if (recorded) break; + } + } + assertTrue(recorded, "expected enqueueToWriteLatency timer to record at least one sample"); + } + @Test(dependsOnMethods = { "testEVCache" }) public void testKeySizeCheck() throws Exception { final String key = "This is an invalid key"; diff --git a/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java b/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java index 69375c8e..173d1b7a 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java +++ b/evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java @@ -285,6 +285,7 @@ public String getStatusCode(StatusCode sc) { public static final String INTERNAL_EXECUTOR_SCHEDULED = "internal.evc.client.scheduledExecutor"; public static final String INTERNAL_POOL_INIT_ERROR = "internal.evc.client.init.error"; public static final String INTERNAL_LOOP_CPU_WALL_TIME_RATIO = "internal.evc.client.loop.cpuWallTimeRatio"; + public static final String INTERNAL_LOOP_ENQUEUE_TO_WRITE_LATENCY = "internal.evc.client.loop.enqueueToWriteLatency"; public static final String INTERNAL_NUM_CHUNK_SIZE = "internal.evc.client.chunking.numOfChunks"; public static final String INTERNAL_CHUNK_DATA_SIZE = "internal.evc.client.chunking.dataSize"; diff --git a/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheBulkGetFuture.java b/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheBulkGetFuture.java index b8379136..d2f7b7bf 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheBulkGetFuture.java +++ b/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheBulkGetFuture.java @@ -53,15 +53,21 @@ public class EVCacheBulkGetFuture extends BulkGetFuture { private final Collection ops; private final CountDownLatch latch; private final long start; + private final long operationAttachedNs; private final EVCacheClient client; private AtomicReferenceArray operationStates; public EVCacheBulkGetFuture(Map> m, Collection getOps, CountDownLatch l, ExecutorService service, EVCacheClient client) { + this(m, getOps, l, service, client, System.nanoTime()); + } + + public EVCacheBulkGetFuture(Map> m, Collection getOps, CountDownLatch l, ExecutorService service, EVCacheClient client, long operationAttachedNs) { super(m, getOps, l, service); rvMap = m; ops = getOps; latch = l; this.start = System.currentTimeMillis(); + this.operationAttachedNs = operationAttachedNs; this.client = client; this.operationStates = null; } @@ -345,6 +351,7 @@ public void signalComplete() { public void signalSingleOpComplete(int sequenceNo, GetOperation op) { this.operationStates.set(sequenceNo, new SingleOperationState(op)); + client.recordLoopEnqueueToWriteLatency(op, operationAttachedNs); } public boolean cancel(boolean ign) { diff --git a/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheOperationFuture.java b/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheOperationFuture.java index 2b9941cd..44ae462a 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheOperationFuture.java +++ b/evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheOperationFuture.java @@ -74,6 +74,7 @@ private static final class LazySharedExecutor { private final CountDownLatch latch; private final AtomicReference objRef; private Operation op; + private long operationAttachedNs; private final String key; private final long start; private final EVCacheClient client; @@ -93,6 +94,7 @@ public Operation getOperation() { public void setOperation(Operation to) { this.op = to; + this.operationAttachedNs = System.nanoTime(); super.setOperation(to); } @@ -380,7 +382,11 @@ public void call() { } public void signalComplete() { - super.signalComplete(); + try { + client.recordLoopEnqueueToWriteLatency(op, operationAttachedNs); + } finally { + super.signalComplete(); + } } /** diff --git a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java index 6587bfe5..71234990 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java +++ b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java @@ -23,6 +23,7 @@ import com.netflix.spectator.api.Id; import com.netflix.spectator.api.Registry; import com.netflix.spectator.api.Tag; +import com.netflix.spectator.api.Timer; import com.netflix.spectator.api.patterns.PolledMeter; import java.io.BufferedInputStream; import java.io.IOException; @@ -32,6 +33,7 @@ import java.net.SocketAddress; import java.net.URLDecoder; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.AbstractMap.SimpleEntry; import java.util.ArrayList; import java.util.Collection; @@ -106,6 +108,7 @@ public class EVCacheClient { private List tags; private final Map counterMap = new ConcurrentHashMap(); private final Id loopCpuWallTimeRatioId; + private final Timer loopEnqueueToWriteLatency; private final Property hashingAlgo; protected final Counter operationsCounter; private final boolean isDuetClient; @@ -155,6 +158,9 @@ public class EVCacheClient { PolledMeter.using(registry) .withId(loopCpuWallTimeRatioId) .monitorValue(this.evcacheMemcachedClient.getLoopProbe(), EVCacheLoopProbe::sampleCpuWallTimeRatio); + this.loopEnqueueToWriteLatency = EVCacheMetricsFactory.getInstance() + .getPercentileTimer(EVCacheMetricsFactory.INTERNAL_LOOP_ENQUEUE_TO_WRITE_LATENCY, + this.tags, Duration.ofMillis(100)); this.evcacheMemcachedClient.addObserver(connectionObserver); this.decodingTranscoder = new EVCacheSerializingTranscoder(Integer.MAX_VALUE); @@ -1695,6 +1701,27 @@ public Counter getOperationCounter() { return operationsCounter; } + /** + * Record per-operation in-process latency from when an EVCache future attached + * a spymemcached {@link net.spy.memcached.ops.Operation} (immediately before + * enqueue into the memcached connection) to when the loop thread finished + * writing the operation to the socket. + * + *

This is the lagging knee-detector that complements the phase 1 loop CPU + * utilization gauge. It is no-throw on purpose so a bad measurement cannot + * break an operation completion callback. + */ + public void recordLoopEnqueueToWriteLatency(net.spy.memcached.ops.Operation op, long operationAttachedNs) { + if (op == null || operationAttachedNs <= 0L) return; + try { + final long wc = op.getWriteCompleteTimestamp(); + if (wc <= 0L || wc < operationAttachedNs) return; + loopEnqueueToWriteLatency.record(wc - operationAttachedNs, TimeUnit.NANOSECONDS); + } catch (Throwable t) { + if (log.isDebugEnabled()) log.debug("recordLoopEnqueueToWriteLatency failed", t); + } + } + /** * Return the keys upto the limit. The key will be cannoicalized key( or hashed Key).
diff --git a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java index 2d196dea..b8cfe97c 100644 --- a/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java +++ b/evcache-core/src/main/java/net/spy/memcached/EVCacheMemcachedClient.java @@ -389,7 +389,7 @@ public EVCacheBulkGetFuture asyncGetBulk(Collection plainKeys, int initialLatchCount = chunks.isEmpty() ? 0 : 1; final CountDownLatch latch = new CountDownLatch(initialLatchCount); final Collection ops = new ArrayList(chunks.size()); - final EVCacheBulkGetFuture rv = new EVCacheBulkGetFuture(m, ops, latch, executorService, client); + final EVCacheBulkGetFuture rv = new EVCacheBulkGetFuture(m, ops, latch, executorService, client, System.nanoTime()); rv.setExpectedCount(chunks.size()); final DistributionSummary dataSizeDS = getDataSizeDistributionSummary( diff --git a/evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheBulkGetFutureLatencyTest.java b/evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheBulkGetFutureLatencyTest.java new file mode 100644 index 00000000..01605637 --- /dev/null +++ b/evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheBulkGetFutureLatencyTest.java @@ -0,0 +1,41 @@ +package com.netflix.evcache.operation; + +import com.netflix.evcache.pool.EVCacheClient; +import net.spy.memcached.ops.GetOperation; +import net.spy.memcached.ops.Operation; +import org.testng.annotations.Test; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +public class EVCacheBulkGetFutureLatencyTest { + + @Test + public void recordsLatencyAgainstSharedBulkTimestampForEachChunk() { + EVCacheClient client = mock(EVCacheClient.class); + long bulkAttachedNs = System.nanoTime(); + + Map> rvMap = new HashMap>(); + Collection ops = new ArrayList(); + + EVCacheBulkGetFuture future = new EVCacheBulkGetFuture( + rvMap, ops, new CountDownLatch(1), null, client, bulkAttachedNs); + future.setExpectedCount(2); + + GetOperation chunkOne = mock(GetOperation.class); + GetOperation chunkTwo = mock(GetOperation.class); + + future.signalSingleOpComplete(0, chunkOne); + future.signalSingleOpComplete(1, chunkTwo); + + verify(client).recordLoopEnqueueToWriteLatency(chunkOne, bulkAttachedNs); + verify(client).recordLoopEnqueueToWriteLatency(chunkTwo, bulkAttachedNs); + } +} diff --git a/evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheOperationFutureLatencyTest.java b/evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheOperationFutureLatencyTest.java new file mode 100644 index 00000000..9205f84f --- /dev/null +++ b/evcache-core/src/test/java/com/netflix/evcache/operation/EVCacheOperationFutureLatencyTest.java @@ -0,0 +1,57 @@ +package com.netflix.evcache.operation; + +import com.netflix.evcache.pool.EVCacheClient; +import net.spy.memcached.ops.Operation; +import org.mockito.ArgumentCaptor; +import org.testng.annotations.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertTrue; + +public class EVCacheOperationFutureLatencyTest { + + @Test + public void recordsLatencyOnSignalCompleteForSingleOp() { + EVCacheClient client = mock(EVCacheClient.class); + Operation op = mock(Operation.class); + // pretend the write completed shortly after setOperation() + when(op.getWriteCompleteTimestamp()).thenReturn(System.nanoTime() + 1_000_000L); + + EVCacheOperationFuture future = new EVCacheOperationFuture( + "key", new CountDownLatch(1), new AtomicReference(null), 1000L, null, client); + + future.setOperation(op); + future.signalComplete(); + + ArgumentCaptor startCaptor = ArgumentCaptor.forClass(Long.class); + verify(client).recordLoopEnqueueToWriteLatency(org.mockito.Matchers.eq(op), startCaptor.capture()); + assertTrue(startCaptor.getValue() > 0L, + "expected operationAttachedNs to be captured > 0, got " + startCaptor.getValue()); + } + + @Test + public void recordsLatencyWithLatestAttachedTimestampOnRetry() throws Exception { + EVCacheClient client = mock(EVCacheClient.class); + Operation firstOp = mock(Operation.class); + Operation retryOp = mock(Operation.class); + + EVCacheOperationFuture future = new EVCacheOperationFuture( + "key", new CountDownLatch(1), new AtomicReference(null), 1000L, null, client); + + future.setOperation(firstOp); + long firstStart = System.nanoTime(); + Thread.sleep(2); + future.setOperation(retryOp); + future.signalComplete(); + + ArgumentCaptor startCaptor = ArgumentCaptor.forClass(Long.class); + verify(client).recordLoopEnqueueToWriteLatency(org.mockito.Matchers.eq(retryOp), startCaptor.capture()); + assertTrue(startCaptor.getValue() >= firstStart, + "expected retry timestamp to be >= first attach time"); + } +} diff --git a/evcache-core/src/test/java/test-suite.xml b/evcache-core/src/test/java/test-suite.xml index 7d7568a6..2112cbff 100644 --- a/evcache-core/src/test/java/test-suite.xml +++ b/evcache-core/src/test/java/test-suite.xml @@ -4,6 +4,8 @@ + + From e6f60737a00b966361152b755712af6abd33e3b2 Mon Sep 17 00:00:00 2001 From: Jason Koch Date: Mon, 18 May 2026 19:06:32 +0000 Subject: [PATCH 4/4] fix: flakey test and improve comments --- .../netflix/evcache/test/EVCacheTestDI.java | 13 +++++++++--- .../netflix/evcache/pool/EVCacheClient.java | 20 +++++++++---------- 2 files changed, 19 insertions(+), 14 deletions(-) diff --git a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java index 1a4007ba..b5a155bd 100644 --- a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java +++ b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java @@ -118,10 +118,17 @@ public void testLoopEnqueueToWriteLatencyMetricRecords() throws Exception { final Map> clientsByServerGroup = manager.getEVCacheClientPool(appName).getAllInstancesByServerGroup(); assertFalse(clientsByServerGroup.isEmpty(), "expected EVCache clients for " + appName); + // recordLoopEnqueueToWriteLatency is invoked from EVCacheOperationFuture.signalComplete, + // which runs on the spymemcached IO loop *after* OperationFuture's latch is decremented. + // That means evCache.get() can return before the metric has been recorded. Issue gets + // and poll the timer until one sample shows up, with a generous total budget so slow + // CI hosts can't lose the race. + final long deadlineNs = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); boolean recorded = false; - for (int attempt = 0; attempt < 10 && !recorded; attempt++) { - get(attempt, evCache); - Thread.sleep(50); + int attempt = 0; + while (!recorded && System.nanoTime() < deadlineNs) { + get(attempt++, evCache); + Thread.sleep(100); for (List clients : clientsByServerGroup.values()) { for (EVCacheClient client : clients) { final Id id = EVCacheMetricsFactory.getInstance().getId(EVCacheMetricsFactory.INTERNAL_LOOP_ENQUEUE_TO_WRITE_LATENCY, client.getTagList()); diff --git a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java index 71234990..e8c7859c 100644 --- a/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java +++ b/evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java @@ -1705,21 +1705,19 @@ public Counter getOperationCounter() { * Record per-operation in-process latency from when an EVCache future attached * a spymemcached {@link net.spy.memcached.ops.Operation} (immediately before * enqueue into the memcached connection) to when the loop thread finished - * writing the operation to the socket. + * writing the operation to the socket (may be queued at socket). * - *

This is the lagging knee-detector that complements the phase 1 loop CPU - * utilization gauge. It is no-throw on purpose so a bad measurement cannot - * break an operation completion callback. + * Use this to identify if the evcache IO thread is getting busy enough that it + * is impacting transaction latency. This does not completely capture the socket + * to network packet time as there may still be queueing on the socket and NIC. */ public void recordLoopEnqueueToWriteLatency(net.spy.memcached.ops.Operation op, long operationAttachedNs) { if (op == null || operationAttachedNs <= 0L) return; - try { - final long wc = op.getWriteCompleteTimestamp(); - if (wc <= 0L || wc < operationAttachedNs) return; - loopEnqueueToWriteLatency.record(wc - operationAttachedNs, TimeUnit.NANOSECONDS); - } catch (Throwable t) { - if (log.isDebugEnabled()) log.debug("recordLoopEnqueueToWriteLatency failed", t); - } + + final long writeComplete = op.getWriteCompleteTimestamp(); + if (writeComplete <= 0L || writeComplete < operationAttachedNs) return; + + loopEnqueueToWriteLatency.record(writeComplete - operationAttachedNs, TimeUnit.NANOSECONDS); }