Add client-side metrics to the ZooKeeper Java client

zookeeper-docs/src/main/resources/markdown/zookeeperMonitor.md

@@ -18,6 +18,7 @@ limitations under the License.
 
 * [New Metrics System](#Metrics-System)
     * [Metrics](#Metrics)
+    * [Client Metrics](#Client-Metrics)
     * [Prometheus](#Prometheus)
     * [Grafana](#Grafana)
 
@@ -37,6 +38,44 @@ client, security, failures, watch/session, requestProcessor, and so forth.
 ### Metrics
 All the metrics are included in the `ServerMetrics.java`.
 
+<a name="Client-Metrics"></a>
+
+### Client Metrics
+In addition to the server-side metrics above, the Java client can emit **client-side metrics** that
+give visibility into behavior observed from inside the application process (per-operation counts and
+latency, error counts by result code, ping round-trip time, and connection/session lifecycle events).
+All client metrics are defined in `ClientMetrics.java`.
+
+Client metrics are **opt-in** and use the same pluggable `MetricsProvider` system as the server. By
+default the client uses the `NullMetricsProvider`, so there is no overhead unless the feature is
+explicitly enabled. To enable it, set the metrics provider class name (and any provider-specific
+configuration) via system properties:
+
+```
+# in-memory provider; values can be read programmatically via MetricsProvider#dump
+-Dzookeeper.clientMetricsProvider.className=org.apache.zookeeper.metrics.impl.DefaultMetricsProvider
+
+# any property under this prefix is passed to the provider with the prefix stripped
+-Dzookeeper.clientMetricsProvider.<providerConfigKey>=<value>
+```
+
+The metrics provider is a per-JVM singleton shared by all `ZooKeeper` client instances in the
+process: the first client to configure a real provider wins, so multiple clients do not, for
+example, attempt to bind the same Prometheus HTTP port twice.
+
+The main client metrics include:
+
+| Metric | Type | Description |
+|---|---|---|
+| `<op>_request_latency` | Summary (per operation) | Latency of completed requests; `cnt_<op>_request_latency` also gives the per-operation request count |
+| `<code>_unsuccessful_request` | Summary (per error code) | Count of failed requests, labelled by `KeeperException.Code` |
+| `ping_latency` | Summary | Client-observed ping round-trip time (ms) |
+| `connect_latency` | Summary | Time to establish a connection to a server (ms) |
+| `connection_established_count` | Counter | Number of connections successfully established |
+| `connection_loss_count` | Counter | Number of times an established connection was lost |
+| `reconnect_count` | Counter | Number of reconnection attempts |
+| `session_expired_count` | Counter | Number of session expirations detected by the client |
+
 <a name="Prometheus"></a>
 
 ### Prometheus

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

@@ -66,6 +66,7 @@
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.ZooKeeper.States;
 import org.apache.zookeeper.ZooKeeper.WatchRegistration;
+import org.apache.zookeeper.client.ClientMetrics;
 import org.apache.zookeeper.client.HostProvider;
 import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.client.ZooKeeperSaslClient;
@@ -279,6 +280,9 @@ static class Packet {
 
         boolean finished;
 
+        /** Client elapsed time (ms) when the request was submitted; used to compute latency. */
+        long requestSubmitTimeMs;
+
         AsyncCallback cb;
 
         Object ctx;
@@ -313,6 +317,7 @@ static class Packet {
             this.response = response;
             this.readOnly = readOnly;
             this.watchRegistration = watchRegistration;
+            this.requestSubmitTimeMs = Time.currentElapsedTime();
         }
 
         public void createBB() {
@@ -445,6 +450,8 @@ public ClientCnxn(
         sendThread = new SendThread(clientCnxnSocket);
         eventThread = new EventThread();
         this.clientConfig = zooKeeper.getClientConfig();
+        ClientMetrics.initialize(clientConfig.getMetricsProviderClassName(),
+                                 clientConfig.getMetricsProviderConfiguration());
         initRequestTimeout();
     }
 
@@ -746,6 +753,7 @@ private void processEvent(Object event) {
     // @VisibleForTesting
     protected void finishPacket(Packet p) {
         int err = p.replyHeader.getErr();
+        recordRequestMetrics(p, err);
         if (p.watchRegistration != null) {
             p.watchRegistration.register(err);
         }
@@ -782,6 +790,26 @@ protected void finishPacket(Packet p) {
         }
     }
 
+    /**
+     * Emit client-side request metrics for a completed packet: per-operation latency (and therefore
+     * per-operation count) and, for failed requests, a count labelled by the error code.
+     */
+    private void recordRequestMetrics(Packet p, int err) {
+        RequestHeader h = p.requestHeader;
+        if (h == null) {
+            return;
+        }
+        ClientMetrics metrics = ClientMetrics.getMetrics();
+        String op = ClientMetrics.opName(h.getType());
+        if (p.requestSubmitTimeMs > 0) {
+            metrics.REQUEST_LATENCY.add(op, Time.currentElapsedTime() - p.requestSubmitTimeMs);
+        }
+        if (err != 0) {
+            Code code = Code.get(err);
+            metrics.UNSUCCESSFUL_REQUEST.add(code != null ? code.name() : String.valueOf(err), 1);
+        }
+    }
+
     void queueEvent(String clientPath, int err, Set<Watcher> materializedWatchers, EventType eventType) {
         KeeperState sessionState = KeeperState.SyncConnected;
         if (KeeperException.Code.SESSIONEXPIRED.intValue() == err
@@ -876,6 +904,7 @@ public RWServerFoundException(String msg) {
     class SendThread extends ZooKeeperThread {
 
         private long lastPingSentNs;
+        private long connectStartTimeMs;
         private final ClientCnxnSocket clientCnxnSocket;
         private boolean isFirstConnect = true;
 
@@ -887,9 +916,11 @@ void readResponse(ByteBuffer incomingBuffer) throws IOException {
             replyHdr.deserialize(bbia, "header");
             switch (replyHdr.getXid()) {
             case PING_XID:
+                long pingRttMs = (System.nanoTime() - lastPingSentNs) / 1000000;
                 LOG.debug("Got ping response for session id: 0x{} after {}ms.",
                     Long.toHexString(sessionId),
-                    ((System.nanoTime() - lastPingSentNs) / 1000000));
+                    pingRttMs);
+                ClientMetrics.getMetrics().PING_LATENCY.add(pingRttMs);
                 return;
               case AUTHPACKET_XID:
                 LOG.debug("Got auth session id: 0x{}", Long.toHexString(sessionId));
@@ -1141,7 +1172,9 @@ private void sendPing() {
         private void startConnect(InetSocketAddress addr) throws IOException {
             // initializing it for new connection
             saslLoginFailed = false;
+            connectStartTimeMs = Time.currentElapsedTime();
             if (!isFirstConnect) {
+                ClientMetrics.getMetrics().RECONNECT_COUNT.inc();
                 try {
                     Thread.sleep(ThreadLocalRandom.current().nextLong(1000));
                 } catch (InterruptedException e) {
@@ -1334,6 +1367,7 @@ public void run() {
         private void cleanAndNotifyState() {
             cleanup();
             if (state.isAlive()) {
+                ClientMetrics.getMetrics().CONNECTION_LOSS_COUNT.inc();
                 eventThread.queueEvent(new WatchedEvent(Event.EventType.None, Event.KeeperState.Disconnected, null));
             }
             clientCnxnSocket.updateNow();
@@ -1435,6 +1469,7 @@ void onConnected(
                     "Unable to reconnect to ZooKeeper service, session 0x%s has expired",
                     Long.toHexString(sessionId));
                 LOG.warn(warnInfo);
+                ClientMetrics.getMetrics().SESSION_EXPIRED_COUNT.inc();
                 throw new SessionExpiredException(warnInfo);
             }
 
@@ -1449,6 +1484,11 @@ void onConnected(
             sessionPasswd = _sessionPasswd;
             changeZkState((isRO) ? States.CONNECTEDREADONLY : States.CONNECTED);
             seenRwServerBefore |= !isRO;
+            ClientMetrics establishedMetrics = ClientMetrics.getMetrics();
+            establishedMetrics.CONNECTION_ESTABLISHED_COUNT.inc();
+            if (connectStartTimeMs > 0) {
+                establishedMetrics.CONNECT_LATENCY.add(Time.currentElapsedTime() - connectStartTimeMs);
+            }
             LOG.info(
                 "Session establishment complete on server {}, session id = 0x{}, negotiated timeout = {}{}",
                 clientCnxnSocket.getRemoteSocketAddress(),