diff --git a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java b/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java index e69280e3..b5a155bd 100644 --- a/evcache-client/test/com/netflix/evcache/test/EVCacheTestDI.java +++ b/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> 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_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 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> 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 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 eeb7941d..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 @@ -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"; 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 3a0dbcd9..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 @@ -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 @@ public class EVCacheClient { private final Property ignoreTouch; 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; @@ -133,6 +140,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 +150,17 @@ 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); + 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); @@ -1342,6 +1358,11 @@ public boolean shutdown(long timeout, TimeUnit unit) { 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,6 +1701,25 @@ 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 (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).
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..dd66aad5 --- /dev/null +++ b/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. + * + *

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); + 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 cpuWallTimeRatio 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 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; + } + } +} 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..b8cfe97c 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); } @@ -384,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/com/netflix/evcache/pool/EVCacheLoopProbeTest.java b/evcache-core/src/test/java/com/netflix/evcache/pool/EVCacheLoopProbeTest.java new file mode 100644 index 00000000..ca77e326 --- /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.sampleCpuWallTimeRatio())); + } + + @Test + public void reportsCpuWallTimeRatioForCurrentThread() throws Exception { + assumeThreadCpuTimeAvailable(); + EVCacheLoopProbe probe = new EVCacheLoopProbe(); + + probe.tick(); + assertTrue(Double.isNaN(probe.sampleCpuWallTimeRatio())); + + 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 + public void returnsZeroWhenNoNewSampleWasPublished() throws Exception { + assumeThreadCpuTimeAvailable(); + EVCacheLoopProbe probe = new EVCacheLoopProbe(); + + probe.tick(); + assertTrue(Double.isNaN(probe.sampleCpuWallTimeRatio())); + + assertEquals(probe.sampleCpuWallTimeRatio(), 0.0); + } + + @Test + public void polledMeterPublishesProbeRatio() throws Exception { + assumeThreadCpuTimeAvailable(); + Registry registry = new DefaultRegistry(); + 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::sampleCpuWallTimeRatio); + + probe.tick(); + PolledMeter.update(registry); + Gauge gauge = registry.gauge(id); + assertTrue(Double.isNaN(gauge.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 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(); + ratio = probe.sampleCpuWallTimeRatio(); + } + return ratio; + } + + 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); + ratio = gauge.value(); + } + return ratio; + } + + 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..2112cbff 100644 --- a/evcache-core/src/test/java/test-suite.xml +++ b/evcache-core/src/test/java/test-suite.xml @@ -3,6 +3,9 @@ + + +