Add a metric to capture write enqueue latency

evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java

@@ -11,12 +11,18 @@
 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.Timer;
+import com.netflix.spectator.api.patterns.PolledMeter;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -76,6 +82,68 @@ public void testEVCache() {
         assertNotNull(evCache);
     }
 
+    @Test(dependsOnMethods = { "testGet" })
+    public void testLoopCpuWallTimeRatioMetricRegistered() throws Exception {
+        final Registry registry = EVCacheMetricsFactory.getInstance().getRegistry();
+        final Map<ServerGroup, List<EVCacheClient>> clientsByServerGroup = manager.getEVCacheClientPool(appName).getAllInstancesByServerGroup();
+        assertFalse(clientsByServerGroup.isEmpty(), "expected EVCache clients for " + appName);
+
+        PolledMeter.update(registry);
+        for (List<EVCacheClient> clients : clientsByServerGroup.values()) {
+            for (EVCacheClient client : clients) {
+                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);
+            }
+        }
+
+        boolean nonZero = false;
+        for (int attempt = 0; attempt < 10 && !nonZero; attempt++) {
+            get(0, evCache);
+            Thread.sleep(1_100);
+            PolledMeter.update(registry);
+            for (List<EVCacheClient> clients : clientsByServerGroup.values()) {
+                for (EVCacheClient client : clients) {
+                    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 cpuWallTimeRatio meter to report a non-zero value");
+    }
+
+    @Test(dependsOnMethods = { "testLoopCpuUtilizationMetricRegistered" })
+    public void testLoopEnqueueToWriteLatencyMetricRecords() throws Exception {
+        final Registry registry = EVCacheMetricsFactory.getInstance().getRegistry();
+        final Map<ServerGroup, List<EVCacheClient>> 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;
+        int attempt = 0;
+        while (!recorded && System.nanoTime() < deadlineNs) {
+            get(attempt++, evCache);
+            Thread.sleep(100);
+            for (List<EVCacheClient> 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";

evcache-core/src/main/java/com/netflix/evcache/metrics/EVCacheMetricsFactory.java

@@ -284,6 +284,8 @@ 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_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";

evcache-core/src/main/java/com/netflix/evcache/operation/EVCacheBulkGetFuture.java

@@ -53,15 +53,21 @@ public class EVCacheBulkGetFuture<T> extends BulkGetFuture<T> {
     private final Collection<Operation> ops;
     private final CountDownLatch latch;
     private final long start;
+    private final long operationAttachedNs;
     private final EVCacheClient client;
     private AtomicReferenceArray<SingleOperationState> operationStates;
 
     public EVCacheBulkGetFuture(Map<String, Future<T>> m, Collection<Operation> getOps, CountDownLatch l, ExecutorService service, EVCacheClient client) {
+        this(m, getOps, l, service, client, System.nanoTime());
+    }
+
+    public EVCacheBulkGetFuture(Map<String, Future<T>> m, Collection<Operation> 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) {

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<T> 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();
+        }
     }
 
     /**

evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheClient.java

@@ -20,7 +20,11 @@
 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.Timer;
+import com.netflix.spectator.api.patterns.PolledMeter;
 import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.PrintWriter;
@@ -29,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;
@@ -102,6 +107,8 @@
     private final Property<Boolean> ignoreTouch;
     private List<Tag> tags;
     private final Map<String, Counter> counterMap = new ConcurrentHashMap<String, Counter>();
+    private final Id loopCpuWallTimeRatioId;
+    private final Timer loopEnqueueToWriteLatency;
     private final Property<String> hashingAlgo;
     protected final Counter operationsCounter;
     private final boolean isDuetClient;
@@ -133,6 +140,8 @@
         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 +150,17 @@
         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);
+        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(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);
@@ -999,7 +1015,7 @@
     }
 
     /**
      * @Deprecated This method does NOT support a mix of plain and hashed keys in {@code keys}. All keys are
      * decoded exactly using the given transcoder (note that hashed keys require two step decoding).
      * For supporting a mix of hashed and plain keys in the {@code keys} collection,
      * use {@link #getAsyncBulk(Collection, Set, Transcoder, EVCacheTranscoder, String, boolean, BiPredicate)}.
@@ -1342,6 +1358,11 @@
         if(shutdown) return true;
 
         shutdown = true;
+        try {
+            PolledMeter.remove(EVCacheMetricsFactory.getInstance().getRegistry(), loopCpuWallTimeRatioId);
+        } catch(Throwable t) {
+            log.warn("Exception while removing loop cpuWallTimeRatio meter", t);
+        }
         try {
             evcacheMemcachedClient.shutdown(timeout, unit);
         } catch(Throwable t) {
@@ -1680,10 +1701,29 @@
         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 (may be queued at socket).
+     *
+     * 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;
+
+        final long writeComplete = op.getWriteCompleteTimestamp();
+        if (writeComplete <= 0L || writeComplete < operationAttachedNs) return;
+
+        loopEnqueueToWriteLatency.record(writeComplete - operationAttachedNs, TimeUnit.NANOSECONDS);
+    }
+
 
     /**
      * Return the keys upto the limit. The key will be cannoicalized key( or hashed Key).<br>
      * <B> The keys are read into memory so make sure you have enough memory to read the specified number of keys<b>
      * @param limit - The number of keys that need to fetched from each memcached clients.
      * @return - the List of keys.
      */

evcache-core/src/main/java/com/netflix/evcache/pool/EVCacheLoopProbe.java

@@ -0,0 +1,139 @@
+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 the loop thread's CPU-time to wall-time ratio from the loop thread itself.
+ *
+ * <p>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.</p>
+ *
+ * <p>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.</p>
+ */
+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<long[]> snapshot = new AtomicReference<long[]>(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 cpuWallTimeRatio will report NaN");
+        }
+    }
+
+    /**
+     * Publish the current thread's CPU time and wall time at most every second.
+     *
+     * <p>This method is intentionally no-throw: it is called from the EVCache IO
+     * loop in a finally block and must never terminate the loop.</p>
+     */
+    public void tick() {
+        try {
+            tickInternal();
+        } catch (Throwable t) {
+            if (!tickFailureLogged) {
+                tickFailureLogged = true;
+                try {
+                    log.warn("EVCache loop cpuWallTimeRatio 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 cpuWallTimeRatio publish");
+            }
+            return;
+        }
+
+        snapshot.lazySet(new long[] { cpuNs, now });
+    }
+
+    /**
+     * Return loop-thread cpu-time / wall-time ratio over the interval since the previous poll.
+     */
+    public double sampleCpuWallTimeRatio() {
+        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 ratio = (double) dCpu / (double) dWall;
+        if (ratio < 0.0) return 0.0;
+
+        if (ratio > 1.0) {
+            aboveOneSamples++;
+            if (aboveOneSamples >= CPU_UTILIZATION_WARNING_THRESHOLD && !aboveOneLogged) {
+                aboveOneLogged = true;
+                log.warn("EVCache loop cpuWallTimeRatio exceeded 1.0 for {} consecutive samples; latest value={}",
+                        CPU_UTILIZATION_WARNING_THRESHOLD, ratio);
+            }
+        } else {
+            aboveOneSamples = 0;
+        }
+
+        return Math.min(ratio, 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;
+        }
+    }
+}