Move lock metrics to MetricsReporter

Adds two new metrics:

 - The load of acquiring each lock path: The average number of threads waiting
   to acquire the lock within the last minute (or unit of time).  Aka the lock
   queue (depth).

 - The load of the lock for each lock path: The average number of threads
   holding the lock within the last minute (or unit of time).  This is always
   <= 1.  Aka the lock utilization.

Changes the LockCounters to LockMetrics, and exporting those once every minute
through MetricReporter which is designed for this.

5 files changed, 196 insertions, 29 deletions

diff --git a/zkfacade/src/test/java/com/yahoo/vespa/curator/CuratorTest.java b/zkfacade/src/test/java/com/yahoo/vespa/curator/CuratorTest.java
index 06440098c52..2bf40c4e2bb 100644
--- a/zkfacade/src/test/java/com/yahoo/vespa/curator/CuratorTest.java
+++ b/zkfacade/src/test/java/com/yahoo/vespa/curator/CuratorTest.java
@@ -99,7 +99,7 @@ public class CuratorTest {
     }
 
     private Curator createCurator(ConfigserverConfig configserverConfig)  {
-        return new Curator(configserverConfig, Optional.empty(), Optional.empty());
+        return new Curator(configserverConfig, Optional.empty());
     }
 
     private static class PortAllocator {
diff --git a/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/AtomicDurationSumTest.java b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/AtomicDurationSumTest.java
new file mode 100644
index 00000000000..f1f9ce5950b
--- /dev/null
+++ b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/AtomicDurationSumTest.java
@@ -0,0 +1,80 @@
+package com.yahoo.vespa.curator.stats;// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+import org.junit.Test;
+
+import java.time.Duration;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author hakon
+ */
+public class AtomicDurationSumTest {
+    private final AtomicDurationSum atomicDurationSum = new AtomicDurationSum();
+
+    @Test
+    public void test() {
+        assertAtomicDurationSum(Duration.ZERO, 0);
+        atomicDurationSum.add(Duration.ofMillis(3));
+        assertAtomicDurationSum(Duration.ofMillis(3), 1);
+        atomicDurationSum.add(Duration.ofMillis(5));
+        assertAtomicDurationSum(Duration.ofMillis(8), 2);
+        assertEquals(0.004, atomicDurationSum.get().averageDuration().get().toMillis() / 1000., 0.00001);
+
+        DurationSum durationSum = atomicDurationSum.getAndReset();
+        assertEquals(Duration.ofMillis(8), durationSum.duration());
+        assertEquals(2, durationSum.count());
+        assertAtomicDurationSum(Duration.ZERO, 0);
+    }
+
+    @Test
+    public void testNegatives() {
+        atomicDurationSum.add(Duration.ofMillis(-1));
+        assertAtomicDurationSum(Duration.ofMillis(-1), 1);
+    }
+
+    private void assertAtomicDurationSum(Duration expectedDuration, int expectedCount) {
+        DurationSum durationSum = atomicDurationSum.get();
+        assertEquals(expectedDuration, durationSum.duration());
+        assertEquals(expectedCount, durationSum.count());
+    }
+
+    @Test
+    public void encoding() {
+        assertEquals(40, AtomicDurationSum.DURATION_BITS);
+        assertEquals(24, AtomicDurationSum.COUNT_BITS);
+
+        assertEquals(0xFFFFFFFFFF000000L, AtomicDurationSum.DURATION_MASK);
+        assertEquals(0x0000000000FFFFFFL, AtomicDurationSum.COUNT_MASK);
+
+        // duration is signed
+        assertEquals(0xFFFFFF8000000000L, AtomicDurationSum.MIN_DURATION);
+        assertEquals(0x0000007FFFFFFFFFL, AtomicDurationSum.MAX_DURATION);
+
+        // count is unsigned
+        assertEquals(0x0000000000000000L, AtomicDurationSum.MIN_COUNT);
+        assertEquals(0x0000000000FFFFFFL, AtomicDurationSum.MAX_COUNT);
+
+        assertDurationEncoding(Duration.ZERO);
+        assertDurationEncoding(Duration.ofMillis(1));
+        assertDurationEncoding(Duration.ofMillis(-1));
+        assertDurationEncoding(Duration.ofMillis(AtomicDurationSum.MIN_DURATION));
+        assertDurationEncoding(Duration.ofMillis(AtomicDurationSum.MAX_DURATION));
+
+        assertCountEncoding(1L);
+        assertCountEncoding(AtomicDurationSum.MIN_COUNT);
+        assertCountEncoding(AtomicDurationSum.MAX_COUNT);
+        assertEquals(0L, AtomicDurationSum.decodeCount(AtomicDurationSum.MAX_COUNT + 1));
+    }
+
+    private void assertDurationEncoding(Duration duration) {
+        long encoded = AtomicDurationSum.encodeDuration(duration);
+        Duration decodedDuration = AtomicDurationSum.decodeDuration(encoded);
+        assertEquals(duration, decodedDuration);
+    }
+
+    private void assertCountEncoding(long count) {
+        int actualCount = AtomicDurationSum.decodeCount(count);
+        assertEquals(count, actualCount);
+    }
+}
\ No newline at end of file
diff --git a/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LatencyStoreTest.java b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LatencyStoreTest.java
new file mode 100644
index 00000000000..119dccca229
--- /dev/null
+++ b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LatencyStoreTest.java
@@ -0,0 +1,52 @@
+package com.yahoo.vespa.curator.stats;// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+import com.yahoo.test.ManualClock;
+import org.junit.Test;
+
+import java.time.Duration;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author hakon
+ */
+public class LatencyStoreTest {
+    private final ManualClock clock = new ManualClock();
+    private final LatencyStore store = new LatencyStore(clock);
+
+    @Test
+    public void verifyDefaultAndEmpty() {
+        assertGetLatencyMetrics(0, 0.000f, 0f);
+        assertGetAndResetLatencyMetrics(0, 0.000f, 0f);
+        assertGetLatencyMetrics(0, 0.000f, 0f);
+    }
+
+    @Test
+    public void commonCase() {
+        store.reportLatency(Duration.ofMillis(2));
+        store.reportLatency(Duration.ofMillis(6));
+        clock.advance(Duration.ofMillis(2));
+        assertGetLatencyMetrics(2, 0.004f, 4f);
+        clock.advance(Duration.ofMillis(14));
+        assertGetAndResetLatencyMetrics(2, 0.004f, 0.5f);
+        assertGetLatencyMetrics(0, 0.000f, 0f);
+    }
+
+    private void assertGetLatencyMetrics(int count, float average, float load) {
+        LatencyMetrics latencyMetrics = store.getLatencyMetrics();
+        assertEquals(count, latencyMetrics.count());
+        assertDoubleEquals(average, latencyMetrics.averageInSeconds());
+        assertDoubleEquals(load, latencyMetrics.load());
+    }
+
+    private void assertGetAndResetLatencyMetrics(int count, float average, float load) {
+        LatencyMetrics latencyMetrics = store.getAndResetLatencyMetrics();
+        assertEquals(count, latencyMetrics.count());
+        assertDoubleEquals(average, latencyMetrics.averageInSeconds());
+        assertDoubleEquals(load, latencyMetrics.load());
+    }
+
+    private static void assertDoubleEquals(float expected, float actual) {
+        assertEquals(expected, actual, 1e-6);
+    }
+}
\ No newline at end of file
diff --git a/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockAttemptSamplesTest.java b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockAttemptSamplesTest.java
index bc731169f96..252a90f8bb4 100644
--- a/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockAttemptSamplesTest.java
+++ b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockAttemptSamplesTest.java
@@ -6,7 +6,6 @@ import org.junit.Test;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Map;
-import java.util.Optional;
 import java.util.stream.Collectors;
 
 import static org.junit.Assert.assertEquals;
@@ -54,8 +53,8 @@ public class LockAttemptSamplesTest {
     }
 
     private boolean maybeSample(String lockPath, int secondsDuration) {
-        LockAttempt lockAttempt = LockAttempt.invokingAcquire(threadLockStats, lockPath, Duration.ofSeconds(1),
-                Optional.empty(), Optional.empty());
+        LockAttempt lockAttempt = LockAttempt.invokingAcquire(threadLockStats, lockPath,
+                Duration.ofSeconds(1), new LockMetrics());
         Instant instant = lockAttempt.getTimeAcquiredWasInvoked().plus(Duration.ofSeconds(secondsDuration));
         lockAttempt.setTerminalState(LockAttempt.LockState.RELEASED, instant);
         return samples.maybeSample(lockAttempt);
diff --git a/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockTest.java b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockTest.java
index a1f95dc0735..a7715eb9756 100644
--- a/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockTest.java
+++ b/zkfacade/src/test/java/com/yahoo/vespa/curator/stats/LockTest.java
@@ -8,10 +8,10 @@ import org.junit.Test;
 
 import java.time.Duration;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -27,7 +27,7 @@ public class LockTest {
     private final InterProcessLock mutex = mock(InterProcessLock.class);
     private final String lockPath = "/lock/path";
     private final Duration acquireTimeout = Duration.ofSeconds(10);
-    private final Lock lock = new Lock(lockPath, mutex, Optional.empty());
+    private final Lock lock = new Lock(lockPath, mutex);
 
     @Before
     public void setUp() {
@@ -46,10 +46,12 @@ public class LockTest {
             assertSame(e.getCause(), exception);
         }
 
-        var expectedCounters = new LockCounters();
-        expectedCounters.invokeAcquireCount.set(1);
-        expectedCounters.acquireFailedCount.set(1);
-        assertEquals(Map.of(lockPath, expectedCounters), LockStats.getGlobal().getLockCountersByPath());
+        var expectedMetrics = new LockMetrics();
+        expectedMetrics.setAcquireCount(1);
+        expectedMetrics.setCumulativeAcquireCount(1);
+        expectedMetrics.setAcquireFailedCount(1);
+        expectedMetrics.setCumulativeAcquireFailedCount(1);
+        assertLockMetrics(expectedMetrics);
 
         List<LockAttempt> slowLockAttempts = LockStats.getGlobal().getLockAttemptSamples();
         assertEquals(1, slowLockAttempts.size());
@@ -67,6 +69,28 @@ public class LockTest {
         assertEquals(0, threadLockStats.getOngoingLockAttempts().size());
     }
 
+    private void assertLockMetrics(LockMetrics expected) {
+        LockMetrics actual = LockStats.getGlobal().getLockMetricsByPath().get(lockPath);
+        assertNotNull(actual);
+
+        assertEquals(expected.getCumulativeAcquireCount(), actual.getCumulativeAcquireCount());
+        assertEquals(expected.getCumulativeAcquireFailedCount(), actual.getCumulativeAcquireFailedCount());
+        assertEquals(expected.getCumulativeAcquireTimedOutCount(), actual.getCumulativeAcquireTimedOutCount());
+        assertEquals(expected.getCumulativeAcquireSucceededCount(), actual.getCumulativeAcquireSucceededCount());
+        assertEquals(expected.getCumulativeReleaseCount(), actual.getCumulativeReleaseCount());
+        assertEquals(expected.getCumulativeReleaseFailedCount(), actual.getCumulativeReleaseFailedCount());
+
+        assertEquals(expected.getAndResetAcquireCount(), actual.getAndResetAcquireCount());
+        assertEquals(expected.getAndResetAcquireFailedCount(), actual.getAndResetAcquireFailedCount());
+        assertEquals(expected.getAndResetAcquireTimedOutCount(), actual.getAndResetAcquireTimedOutCount());
+        assertEquals(expected.getAndResetAcquireSucceededCount(), actual.getAndResetAcquireSucceededCount());
+        assertEquals(expected.getAndResetReleaseCount(), actual.getAndResetReleaseCount());
+        assertEquals(expected.getAndResetReleaseFailedCount(), actual.getAndResetReleaseFailedCount());
+
+        assertEquals(expected.getAcquiringNow(), actual.getAcquiringNow());
+        assertEquals(expected.getLockedNow(), actual.getLockedNow());
+    }
+
     @Test
     public void acquireTimesOut() throws Exception {
         when(mutex.acquire(anyLong(), any())).thenReturn(false);
@@ -78,10 +102,12 @@ public class LockTest {
             assertTrue("unexpected exception: " + e.getMessage(), e.getMessage().contains("Timed out"));
         }
 
-        var expectedCounters = new LockCounters();
-        expectedCounters.invokeAcquireCount.set(1);
-        expectedCounters.acquireTimedOutCount.set(1);
-        assertEquals(Map.of(lockPath, expectedCounters), LockStats.getGlobal().getLockCountersByPath());
+        var expectedMetrics = new LockMetrics();
+        expectedMetrics.setAcquireCount(1);
+        expectedMetrics.setCumulativeAcquireCount(1);
+        expectedMetrics.setAcquireTimedOutCount(1);
+        expectedMetrics.setCumulativeAcquireTimedOutCount(1);
+        assertLockMetrics(expectedMetrics);
     }
 
     @Test
@@ -90,29 +116,39 @@ public class LockTest {
 
         lock.acquire(acquireTimeout);
 
-        var expectedCounters = new LockCounters();
-        expectedCounters.invokeAcquireCount.set(1);
-        expectedCounters.lockAcquiredCount.set(1);
-        expectedCounters.inCriticalRegionCount.set(1);
-        assertEquals(Map.of(lockPath, expectedCounters), LockStats.getGlobal().getLockCountersByPath());
+        var expectedMetrics = new LockMetrics();
+        expectedMetrics.setAcquireCount(1);
+        expectedMetrics.setCumulativeAcquireCount(1);
+        expectedMetrics.setAcquireSucceededCount(1);
+        expectedMetrics.setCumulativeAcquireSucceededCount(1);
+        expectedMetrics.setLockedNow(1);
+        assertLockMetrics(expectedMetrics);
 
         // reenter
+        // NB: non-cumulative counters are reset on fetch
         lock.acquire(acquireTimeout);
-        expectedCounters.invokeAcquireCount.set(2);
-        expectedCounters.lockAcquiredCount.set(2);
-        expectedCounters.inCriticalRegionCount.set(2);
+        expectedMetrics.setAcquireCount(1);  // reset to 0 above, + 1
+        expectedMetrics.setCumulativeAcquireCount(2);
+        expectedMetrics.setAcquireSucceededCount(1); // reset to 0 above, +1
+        expectedMetrics.setCumulativeAcquireSucceededCount(2);
+        expectedMetrics.setLockedNow(2);
+        assertLockMetrics(expectedMetrics);
 
         // inner-most closes
         lock.close();
-        expectedCounters.inCriticalRegionCount.set(1);
-        expectedCounters.locksReleasedCount.set(1);
-        assertEquals(Map.of(lockPath, expectedCounters), LockStats.getGlobal().getLockCountersByPath());
+        expectedMetrics.setAcquireCount(0);  // reset to 0 above
+        expectedMetrics.setAcquireSucceededCount(0); // reset to 0 above
+        expectedMetrics.setReleaseCount(1);
+        expectedMetrics.setCumulativeReleaseCount(1);
+        expectedMetrics.setLockedNow(1);
+        assertLockMetrics(expectedMetrics);
 
         // outer-most closes
         lock.close();
-        expectedCounters.inCriticalRegionCount.set(0);
-        expectedCounters.locksReleasedCount.set(2);
-        assertEquals(Map.of(lockPath, expectedCounters), LockStats.getGlobal().getLockCountersByPath());
+        expectedMetrics.setReleaseCount(1);  // reset to 0 above, +1
+        expectedMetrics.setCumulativeReleaseCount(2);
+        expectedMetrics.setLockedNow(0);
+        assertLockMetrics(expectedMetrics);
     }
 
     @Test
@@ -120,7 +156,7 @@ public class LockTest {
         when(mutex.acquire(anyLong(), any())).thenReturn(true);
 
         String lockPath2 = "/lock/path/2";
-        Lock lock2 = new Lock(lockPath2, mutex, Optional.empty());
+        Lock lock2 = new Lock(lockPath2, mutex);
 
         lock.acquire(acquireTimeout);
         lock2.acquire(acquireTimeout);