Add metrics to lock attempts

9 files changed, 112 insertions, 40 deletions

diff --git a/zkfacade/abi-spec.json b/zkfacade/abi-spec.json
index e026559b283..0f421621523 100644
--- a/zkfacade/abi-spec.json
+++ b/zkfacade/abi-spec.json
@@ -68,8 +68,8 @@
     "methods": [
       "public static com.yahoo.vespa.curator.Curator create(java.lang.String)",
       "public static com.yahoo.vespa.curator.Curator create(java.lang.String, java.util.Optional)",
-      "public void <init>(com.yahoo.cloud.config.ConfigserverConfig, com.yahoo.vespa.zookeeper.VespaZooKeeperServer)",
-      "protected void <init>(java.lang.String, java.lang.String, java.util.function.Function)",
+      "public void <init>(com.yahoo.cloud.config.ConfigserverConfig, com.yahoo.jdisc.Metric, com.yahoo.vespa.zookeeper.VespaZooKeeperServer)",
+      "protected void <init>(java.lang.String, java.lang.String, java.util.function.Function, java.util.Optional)",
       "public java.lang.String connectionSpec()",
       "public org.apache.curator.framework.recipes.atomic.DistributedAtomicLong createAtomicCounter(java.lang.String)",
       "public org.apache.curator.framework.recipes.locks.InterProcessLock createMutex(java.lang.String)",
@@ -104,8 +104,8 @@
       "public"
     ],
     "methods": [
-      "public void <init>(java.lang.String, com.yahoo.vespa.curator.Curator)",
-      "public void <init>(java.lang.String, org.apache.curator.framework.recipes.locks.InterProcessLock)",
+      "public void <init>(java.lang.String, com.yahoo.vespa.curator.Curator, java.util.Optional)",
+      "public void <init>(java.lang.String, org.apache.curator.framework.recipes.locks.InterProcessLock, java.util.Optional)",
       "public void acquire(java.time.Duration)",
       "public void close()"
     ],
diff --git a/zkfacade/src/main/java/com/yahoo/vespa/curator/Curator.java b/zkfacade/src/main/java/com/yahoo/vespa/curator/Curator.java
index 37b1fa1c9fb..82c7f6175ce 100644
--- a/zkfacade/src/main/java/com/yahoo/vespa/curator/Curator.java
+++ b/zkfacade/src/main/java/com/yahoo/vespa/curator/Curator.java
@@ -4,6 +4,7 @@ package com.yahoo.vespa.curator;
 import com.google.inject.Inject;
 import com.yahoo.cloud.config.ConfigserverConfig;
 import com.yahoo.io.IOUtils;
+import com.yahoo.jdisc.Metric;
 import com.yahoo.net.HostName;
 import com.yahoo.path.Path;
 import com.yahoo.text.Utf8;
@@ -66,28 +67,31 @@ public class Curator implements AutoCloseable {
     // All lock keys, to allow re-entrancy. This will grow forever, but this should be too slow to be a problem
     private final ConcurrentHashMap<Path, Lock> locks = new ConcurrentHashMap<>();
 
+    private final Optional<Metric> metric;
+
     /** Creates a curator instance from a comma-separated string of ZooKeeper host:port strings */
     public static Curator create(String connectionSpec) {
-        return new Curator(connectionSpec, connectionSpec, Optional.of(ZK_CLIENT_CONFIG_FILE));
+        return new Curator(connectionSpec, connectionSpec, Optional.empty(), Optional.of(ZK_CLIENT_CONFIG_FILE));
     }
 
     // For testing only, use Optional.empty for clientConfigFile parameter to create default zookeeper client config
     public static Curator create(String connectionSpec, Optional<File> clientConfigFile) {
-        return new Curator(connectionSpec, connectionSpec, clientConfigFile);
+        return new Curator(connectionSpec, connectionSpec, Optional.empty(), clientConfigFile);
     }
 
     // Depend on ZooKeeperServer to make sure it is started first
     // TODO: Move zookeeperserver config out of configserverconfig (requires update of controller services.xml as well)
     @Inject
-    public Curator(ConfigserverConfig configserverConfig, VespaZooKeeperServer server) {
-        this(configserverConfig, Optional.of(ZK_CLIENT_CONFIG_FILE));
+    public Curator(ConfigserverConfig configserverConfig, Metric metric, VespaZooKeeperServer server) {
+        this(configserverConfig, Optional.of(metric), Optional.of(ZK_CLIENT_CONFIG_FILE));
     }
 
-    Curator(ConfigserverConfig configserverConfig, Optional<File> clientConfigFile) {
-        this(createConnectionSpec(configserverConfig), createEnsembleConnectionSpec(configserverConfig), clientConfigFile);
+    Curator(ConfigserverConfig configserverConfig, Optional<Metric> metric, Optional<File> clientConfigFile) {
+        this(createConnectionSpec(configserverConfig), createEnsembleConnectionSpec(configserverConfig), metric, clientConfigFile);
     }
 
-    private Curator(String connectionSpec, String zooKeeperEnsembleConnectionSpec, Optional<File> clientConfigFile) {
+    private Curator(String connectionSpec, String zooKeeperEnsembleConnectionSpec, Optional<Metric> metric,
+                    Optional<File> clientConfigFile) {
         this(connectionSpec,
                 zooKeeperEnsembleConnectionSpec,
                 (retryPolicy) -> CuratorFrameworkFactory
@@ -98,20 +102,24 @@ public class Curator implements AutoCloseable {
                         .connectString(connectionSpec)
                         .zookeeperFactory(new VespaZooKeeperFactory(createClientConfig(clientConfigFile)))
                         .dontUseContainerParents() // TODO: Remove when we know ZooKeeper 3.5 works fine, consider waiting until Vespa 8
-                        .build());
+                        .build(),
+                metric);
     }
 
     protected Curator(String connectionSpec,
                       String zooKeeperEnsembleConnectionSpec,
-                      Function<RetryPolicy, CuratorFramework> curatorFactory) {
+                      Function<RetryPolicy, CuratorFramework> curatorFactory,
+                      Optional<Metric> metric) {
         this(connectionSpec, zooKeeperEnsembleConnectionSpec, curatorFactory,
-                new ExponentialBackoffRetry((int) BASE_SLEEP_TIME.toMillis(), MAX_RETRIES));
+                new ExponentialBackoffRetry((int) BASE_SLEEP_TIME.toMillis(), MAX_RETRIES),
+                metric);
     }
 
     private Curator(String connectionSpec,
                     String zooKeeperEnsembleConnectionSpec,
                     Function<RetryPolicy, CuratorFramework> curatorFactory,
-                    RetryPolicy retryPolicy) {
+                    RetryPolicy retryPolicy,
+                    Optional<Metric> metric) {
         this.connectionSpec = connectionSpec;
         this.retryPolicy = retryPolicy;
         this.curatorFramework = curatorFactory.apply(retryPolicy);
@@ -124,6 +132,7 @@ public class Curator implements AutoCloseable {
 
         this.zooKeeperEnsembleConnectionSpec = zooKeeperEnsembleConnectionSpec;
         this.zooKeeperEnsembleCount = zooKeeperEnsembleConnectionSpec.split(",").length;
+        this.metric = metric;
     }
 
     private static String createConnectionSpec(ConfigserverConfig configserverConfig) {
@@ -354,7 +363,7 @@ public class Curator implements AutoCloseable {
     /** Create and acquire a re-entrant lock in given path */
     public Lock lock(Path path, Duration timeout) {
         create(path);
-        Lock lock = locks.computeIfAbsent(path, (pathArg) -> new Lock(pathArg.getAbsolute(), this));
+        Lock lock = locks.computeIfAbsent(path, (pathArg) -> new Lock(pathArg.getAbsolute(), this, metric));
         lock.acquire(timeout);
         return lock;
     }
diff --git a/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java b/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java
index c9a55e40da1..305e5d459fc 100644
--- a/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java
+++ b/zkfacade/src/main/java/com/yahoo/vespa/curator/Lock.java
@@ -2,6 +2,7 @@
 package com.yahoo.vespa.curator;
 
 import com.google.common.util.concurrent.UncheckedTimeoutException;
+import com.yahoo.jdisc.Metric;
 import com.yahoo.path.Path;
 import com.yahoo.transaction.Mutex;
 import com.yahoo.vespa.curator.stats.LockStats;
@@ -9,6 +10,8 @@ import com.yahoo.vespa.curator.stats.ThreadLockStats;
 import org.apache.curator.framework.recipes.locks.InterProcessLock;
 
 import java.time.Duration;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -23,21 +26,25 @@ public class Lock implements Mutex {
 
     private final InterProcessLock mutex;
     private final String lockPath;
+    private final Optional<Metric> metric;
+    private final Optional<Metric.Context> metricContext;
 
-    public Lock(String lockPath, Curator curator) {
-        this(lockPath, curator.createMutex(lockPath));
+    public Lock(String lockPath, Curator curator, Optional<Metric> metric) {
+        this(lockPath, curator.createMutex(lockPath), metric);
     }
 
     /** Public for testing only */
-    public Lock(String lockPath, InterProcessLock mutex) {
+    public Lock(String lockPath, InterProcessLock mutex, Optional<Metric> metric) {
         this.lockPath = lockPath;
         this.mutex = mutex;
+        this.metric = metric;
+        this.metricContext = metric.map(aMetric -> aMetric.createContext(Map.of("lockPath", lockPath)));
     }
 
     /** Take the lock with the given timeout. This may be called multiple times from the same thread - each matched by a close */
     public void acquire(Duration timeout) throws UncheckedTimeoutException {
         ThreadLockStats threadLockStats = LockStats.getForCurrentThread();
-        threadLockStats.invokingAcquire(lockPath, timeout);
+        threadLockStats.invokingAcquire(lockPath, timeout, metric, metricContext);
 
         final boolean acquired;
         try {
diff --git a/zkfacade/src/main/java/com/yahoo/vespa/curator/mock/MockCurator.java b/zkfacade/src/main/java/com/yahoo/vespa/curator/mock/MockCurator.java
index 3da7678c44e..2ef42f539d2 100644
--- a/zkfacade/src/main/java/com/yahoo/vespa/curator/mock/MockCurator.java
+++ b/zkfacade/src/main/java/com/yahoo/vespa/curator/mock/MockCurator.java
@@ -137,7 +137,7 @@ public class MockCurator extends Curator {
      *                       This is not what ZooKeeper does.
      */
     public MockCurator(boolean stableOrdering) {
-        super("", "", (retryPolicy) -> null);
+        super("", "", (retryPolicy) -> null, Optional.empty());
         this.stableOrdering = stableOrdering;
         curatorFramework = new MockCuratorFramework();
         curatorFramework.start();
diff --git a/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/LockAttempt.java b/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/LockAttempt.java
index d57a4b7d9dc..bec7c08b9b3 100644
--- a/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/LockAttempt.java
+++ b/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/LockAttempt.java
@@ -1,6 +1,8 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.curator.stats;
 
+import com.yahoo.jdisc.Metric;
+
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -21,14 +23,17 @@ public class LockAttempt {
     private final String lockPath;
     private final Instant callAcquireInstant;
     private final Duration timeout;
+    private final Optional<Metric> metric;
+    private final Optional<Metric.Context> metricContext;
     private final List<LockAttempt> nestedLockAttempts = new ArrayList<>();
 
     private volatile Optional<Instant> lockAcquiredInstant = Optional.empty();
     private volatile Optional<Instant> terminalStateInstant = Optional.empty();
     private volatile Optional<String> stackTrace = Optional.empty();
 
-    public static LockAttempt invokingAcquire(ThreadLockStats threadLockStats, String lockPath, Duration timeout) {
-        return new LockAttempt(threadLockStats, lockPath, timeout, Instant.now());
+    public static LockAttempt invokingAcquire(ThreadLockStats threadLockStats, String lockPath, Duration timeout,
+                                              Optional<Metric> metric, Optional<Metric.Context> metricContext) {
+        return new LockAttempt(threadLockStats, lockPath, timeout, Instant.now(), metric, metricContext);
     }
 
     public enum LockState {
@@ -44,11 +49,16 @@ public class LockAttempt {
 
     private volatile LockState lockState = LockState.ACQUIRING;
 
-    private LockAttempt(ThreadLockStats threadLockStats, String lockPath, Duration timeout, Instant callAcquireInstant) {
+    private LockAttempt(ThreadLockStats threadLockStats, String lockPath, Duration timeout, Instant callAcquireInstant,
+                        Optional<Metric> metric, Optional<Metric.Context> metricContext) {
         this.threadLockStats = threadLockStats;
         this.lockPath = lockPath;
         this.callAcquireInstant = callAcquireInstant;
         this.timeout = timeout;
+        this.metric = metric;
+        this.metricContext = metricContext;
+
+        addToMetric("lockAttempt.acquiring", 1);
     }
 
     public String getThreadName() { return threadLockStats.getThreadName(); }
@@ -57,13 +67,16 @@ public class LockAttempt {
     public Duration getAcquireTimeout() { return timeout; }
     public LockState getLockState() { return lockState; }
     public Optional<Instant> getTimeLockWasAcquired() { return lockAcquiredInstant; }
+    public Instant getTimeAcquireEndedOrNow() {
+        return lockAcquiredInstant.orElseGet(() -> getTimeTerminalStateWasReached().orElseGet(Instant::now));
+    }
     public Optional<Instant> getTimeTerminalStateWasReached() { return terminalStateInstant; }
     public Optional<String> getStackTrace() { return stackTrace; }
     public List<LockAttempt> getNestedLockAttempts() { return List.copyOf(nestedLockAttempts); }
+    public Optional<Metric> metric() { return metric; }
+    public Optional<Metric.Context> metricContext() { return metricContext; }
 
-    public Duration getDurationOfAcquire() {
-        return Duration.between(callAcquireInstant, lockAcquiredInstant.orElseGet(Instant::now));
-    }
+    public Duration getDurationOfAcquire() { return Duration.between(callAcquireInstant, getTimeAcquireEndedOrNow()); }
 
     public Duration getDurationWithLock() {
         return lockAcquiredInstant
@@ -90,14 +103,41 @@ public class LockAttempt {
         nestedLockAttempts.add(nestedLockAttempt);
     }
 
-    void acquireFailed() { setTerminalState(LockState.ACQUIRE_FAILED); }
-    void timedOut() { setTerminalState(LockState.TIMED_OUT); }
-    void released() { setTerminalState(LockState.RELEASED); }
-    void releasedWithError() { setTerminalState(LockState.RELEASED_WITH_ERROR); }
+    void acquireFailed() {
+        setTerminalState(LockState.ACQUIRE_FAILED);
+        addToMetric("lockAttempt.acquiring", -1);
+        addToMetric("lockAttempt.acquireFailed", 1);
+        setMetricTo("lockAttempt.acquiringLatency", getDurationOfAcquire().toMillis() / 1000.);
+    }
+
+    void timedOut() {
+        setTerminalState(LockState.TIMED_OUT);
+        addToMetric("lockAttempt.acquiring", -1);
+        addToMetric("lockAttempt.acquireTimedOut", 1);
+        setMetricTo("lockAttempt.acquiringLatency", getDurationOfAcquire().toMillis() / 1000.);
+    }
 
     void lockAcquired() {
         lockState = LockState.ACQUIRED;
         lockAcquiredInstant = Optional.of(Instant.now());
+        addToMetric("lockAttempt.acquiring", -1);
+        addToMetric("lockAttempt.acquired", 1);
+        setMetricTo("lockAttempt.locked", 1);
+        setMetricTo("lockAttempt.acquiringLatency", getDurationOfAcquire().toMillis() / 1000.);
+    }
+
+    void released() {
+        setTerminalState(LockState.RELEASED);
+        setMetricTo("lockAttempt.locked", 0);
+        addToMetric("lockAttempt.released", 1);
+        setMetricTo("lockAttempt.lockedLatency", getDurationWithLock().toMillis() / 1000.);
+    }
+
+    void releasedWithError() {
+        setTerminalState(LockState.RELEASED_WITH_ERROR);
+        setMetricTo("lockAttempt.locked", 0);
+        addToMetric("lockAttempt.releaseError", 1);
+        setMetricTo("lockAttempt.lockedLatency", getDurationWithLock().toMillis() / 1000.);
     }
 
     void setTerminalState(LockState terminalState) { setTerminalState(terminalState, Instant.now()); }
@@ -106,4 +146,16 @@ public class LockAttempt {
         lockState = terminalState;
         terminalStateInstant = Optional.of(instant);
     }
+
+    private void addToMetric(String key, Number value) {
+        if (metric.isPresent() && metricContext.isPresent()) {
+            metric.get().add(key, value, metricContext.get());
+        }
+    }
+
+    private void setMetricTo(String key, Number value) {
+        if (metric.isPresent() && metricContext.isPresent()) {
+            metric.get().set(key, value, metricContext.get());
+        }
+    }
 }
diff --git a/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/ThreadLockStats.java b/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/ThreadLockStats.java
index c994121441a..90a2c2f8d25 100644
--- a/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/ThreadLockStats.java
+++ b/zkfacade/src/main/java/com/yahoo/vespa/curator/stats/ThreadLockStats.java
@@ -1,6 +1,7 @@
 // Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.curator.stats;
 
+import com.yahoo.jdisc.Metric;
 import com.yahoo.vespa.curator.Lock;
 
 import java.time.Duration;
@@ -60,11 +61,12 @@ public class ThreadLockStats {
     public Optional<RecordedLockAttempts> getOngoingRecording() { return ongoingRecording; }
 
     /** Mutable method (see class doc) */
-    public void invokingAcquire(String lockPath, Duration timeout) {
+    public void invokingAcquire(String lockPath, Duration timeout,
+                                Optional<Metric> metric, Optional<Metric.Context> metricContext) {
         LockCounters lockCounters = getGlobalLockCounters(lockPath);
         lockCounters.invokeAcquireCount.incrementAndGet();
         lockCounters.inCriticalRegionCount.incrementAndGet();
-        LockAttempt lockAttempt = LockAttempt.invokingAcquire(this, lockPath, timeout);
+        LockAttempt lockAttempt = LockAttempt.invokingAcquire(this, lockPath, timeout, metric, metricContext);
 
         LockAttempt lastLockAttempt = lockAttemptsStack.peekLast();
         if (lastLockAttempt == null) {
@@ -93,11 +95,11 @@ public class ThreadLockStats {
     /** Mutable method (see class doc) */
     public void lockAcquired(String lockPath) {
         getGlobalLockCounters(lockPath).lockAcquiredCount.incrementAndGet();
-        LockAttempt lastLockAttempt = lockAttemptsStack.peekLast();
-        if (lastLockAttempt == null) {
+        LockAttempt lockAttempt = lockAttemptsStack.peekLast();
+        if (lockAttempt == null) {
             throw new IllegalStateException("lockAcquired invoked without lockAttempts");
         }
-        lastLockAttempt.lockAcquired();
+        lockAttempt.lockAcquired();
     }
 
     /** Mutable method (see class doc) */
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 2bf40c4e2bb..06440098c52 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());
+        return new Curator(configserverConfig, Optional.empty(), Optional.empty());
     }
 
     private static class PortAllocator {
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 14dbbad56ba..bc731169f96 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,6 +6,7 @@ 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;
@@ -53,7 +54,8 @@ public class LockAttemptSamplesTest {
     }
 
     private boolean maybeSample(String lockPath, int secondsDuration) {
-        LockAttempt lockAttempt = LockAttempt.invokingAcquire(threadLockStats, lockPath, Duration.ofSeconds(1));
+        LockAttempt lockAttempt = LockAttempt.invokingAcquire(threadLockStats, lockPath, Duration.ofSeconds(1),
+                Optional.empty(), Optional.empty());
         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 a221975c120..a1f95dc0735 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
@@ -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);
+    private final Lock lock = new Lock(lockPath, mutex, Optional.empty());
 
     @Before
     public void setUp() {
@@ -120,7 +120,7 @@ public class LockTest {
         when(mutex.acquire(anyLong(), any())).thenReturn(true);
 
         String lockPath2 = "/lock/path/2";
-        Lock lock2 = new Lock(lockPath2, mutex);
+        Lock lock2 = new Lock(lockPath2, mutex, Optional.empty());
 
         lock.acquire(acquireTimeout);
         lock2.acquire(acquireTimeout);