ZOOKEEPER-4284: Add metrics for observer sync time

Motivation

- With enabling the feature of followers hosting observers, we would need a metric to measure the observer sync time just like what we have for the follower sync time.

Changes

- Added the "observer_sync_time" metrics
- Added unit test on the metrics

Author: liwang <liwang@apple.com>

Reviewers: Enrico Olivelli <eolivelli@apache.org>, Michael Han <hanm@apache.org>, Damien Diederen <ddiederen@apache.org>

Closes #1691 from li4wang/ZOOKEEPER-4284

zookeeper-server/src/main/java/org/apache/zookeeper/server/ServerMetrics.java

@@ -75,6 +75,7 @@ public final class ServerMetrics {
         UPDATE_LATENCY = metricsContext.getSummary("updatelatency", DetailLevel.ADVANCED);
         PROPAGATION_LATENCY = metricsContext.getSummary("propagation_latency", DetailLevel.ADVANCED);
         FOLLOWER_SYNC_TIME = metricsContext.getSummary("follower_sync_time", DetailLevel.BASIC);
+        OBSERVER_SYNC_TIME = metricsContext.getSummary("observer_sync_time", DetailLevel.BASIC);
         ELECTION_TIME = metricsContext.getSummary("election_time", DetailLevel.BASIC);
         LOOKING_COUNT = metricsContext.getCounter("looking_count");
         DIFF_COUNT = metricsContext.getCounter("diff_count");
@@ -296,6 +297,7 @@ public final class ServerMetrics {
     public final Summary PROPAGATION_LATENCY;
 
     public final Summary FOLLOWER_SYNC_TIME;
+    public final Summary OBSERVER_SYNC_TIME;
 
     public final Summary ELECTION_TIME;
 

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/Observer.java

@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.jute.Record;
+import org.apache.zookeeper.common.Time;
 import org.apache.zookeeper.server.ObserverBean;
 import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ServerMetrics;
@@ -114,11 +115,14 @@ public class Observer extends Learner {
                     throw new Exception("learned about role change");
                 }
 
+                final long startTime = Time.currentElapsedTime();
                 self.setLeaderAddressAndId(master.addr, master.getId());
                 self.setZabState(QuorumPeer.ZabState.SYNCHRONIZATION);
                 syncWithLeader(newLeaderZxid);
                 self.setZabState(QuorumPeer.ZabState.BROADCAST);
                 completedSync = true;
+                final long syncTime = Time.currentElapsedTime() - startTime;
+                ServerMetrics.getMetrics().OBSERVER_SYNC_TIME.add(syncTime);
                 QuorumPacket qp = new QuorumPacket();
                 while (this.isRunning() && nextLearnerMaster.get() == null) {
                     readPacket(qp);

zookeeper-server/src/test/java/org/apache/zookeeper/test/ObserverMasterTest.java

@@ -55,6 +55,7 @@ import org.apache.zookeeper.ZooKeeper.States;
 import org.apache.zookeeper.admin.ZooKeeperAdmin;
 import org.apache.zookeeper.jmx.MBeanRegistry;
 import org.apache.zookeeper.jmx.ZKMBeanInfo;
+import org.apache.zookeeper.metrics.MetricsUtils;
 import org.apache.zookeeper.server.admin.Commands;
 import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
 import org.apache.zookeeper.server.util.PortForwarder;
@@ -85,6 +86,8 @@ public class ObserverMasterTest extends ObserverMasterTestBase {
         assertTrue(ClientBase.waitForServerUp("127.0.0.1:" + CLIENT_PORT_OBS, CONNECTION_TIMEOUT),
                 "waiting for server 3 being up");
 
+        validateObserverSyncTimeMetrics();
+
         if (testObserverMaster) {
             int masterPort = q3.getQuorumPeer().observer.getSocket().getPort();
             LOG.info("port {} {}", masterPort, OM_PORT);
@@ -522,4 +525,15 @@ public class ObserverMasterTest extends ObserverMasterTestBase {
 
     }
 
+    private void validateObserverSyncTimeMetrics() {
+        final String name = "observer_sync_time";
+        final Map<String, Object> metrics = MetricsUtils.currentServerMetrics();
+
+        assertEquals(5, metrics.keySet().stream().filter(key -> key.contains(name)).count());
+        assertNotNull(metrics.get(String.format("avg_%s", name)));
+        assertNotNull(metrics.get(String.format("min_%s", name)));
+        assertNotNull(metrics.get(String.format("max_%s", name)));
+        assertNotNull(metrics.get(String.format("cnt_%s", name)));
+        assertNotNull(metrics.get(String.format("sum_%s", name)));
+    }
 }