ZooKeeper: more instrumentation [#CLICKHOUSE-2]

dbms/src/Common/CurrentMetrics.cpp

@@ -28,7 +28,9 @@
     M(MemoryTrackingForMerges) \
     M(LeaderElection) \
     M(EphemeralNode) \
+    M(ZooKeeperSession) \
     M(ZooKeeperWatch) \
+    M(ZooKeeperRequest) \
     M(DelayedInserts) \
     M(ContextLockWait) \
     M(StorageBufferRows) \

dbms/src/Common/ProfileEvents.cpp

@@ -66,6 +66,7 @@
     M(ZooKeeperCheck) \
     M(ZooKeeperClose) \
     M(ZooKeeperExceptions) \
+    M(ZooKeeperWaitMicroseconds) \
     \
     M(DistributedConnectionFailTry) \
     M(DistributedConnectionMissingTable) \

dbms/src/Common/ZooKeeper/ZooKeeperImpl.cpp

@@ -36,6 +36,13 @@ namespace ProfileEvents
     extern const Event ZooKeeperList;
     extern const Event ZooKeeperCheck;
     extern const Event ZooKeeperClose;
+    extern const Event ZooKeeperWaitMicroseconds;
+}
+
+namespace CurrentMetrics
+{
+    extern const Metric ZooKeeperRequest;
+    extern const Metric ZooKeeperWatch;
 }
 
 
@@ -904,6 +911,7 @@ void ZooKeeper::receiveEvent()
                     if (callback)
                         callback(watch_response);   /// NOTE We may process callbacks not under mutex.
 
+                CurrentMetrics::sub(CurrentMetrics::ZooKeeperWatch, it->second.size());
                 watches.erase(it);
             }
         };
@@ -919,6 +927,7 @@ void ZooKeeper::receiveEvent()
 
             request_info = std::move(it->second);
             operations.erase(it);
+            CurrentMetrics::sub(CurrentMetrics::ZooKeeperRequest);
         }
 
         response = request_info.request->makeResponse();
@@ -936,6 +945,9 @@ void ZooKeeper::receiveEvent()
     if (length != actual_length)
         throw Exception("Response length doesn't match. Expected: " + toString(length) + ", actual: " + toString(actual_length), ZMARSHALLINGERROR);
 
+    auto elapsed_microseconds = std::chrono::duration_cast<std::chrono::microseconds>(clock::now() - request_info.time).count();
+    ProfileEvents::increment(ProfileEvents::ZooKeeperWaitMicroseconds, elapsed_microseconds);
+
     if (request_info.callback)
         request_info.callback(*response);
 }
@@ -973,6 +985,7 @@ void ZooKeeper::finalize(bool error_send, bool error_receive)
                         request_info.callback(*response);
                 }
 
+                CurrentMetrics::sub(CurrentMetrics::ZooKeeperRequest, operations.size());
                 operations.clear();
             }
 
@@ -991,6 +1004,7 @@ void ZooKeeper::finalize(bool error_send, bool error_receive)
                             callback(response);
                 }
 
+                CurrentMetrics::sub(CurrentMetrics::ZooKeeperWatch, watches.size());
                 watches.clear();
             }
         }
@@ -1221,6 +1235,7 @@ void ZooKeeper::pushRequest(RequestInfo && info)
     ProfileEvents::increment(ProfileEvents::ZooKeeperTransactions);
 
     {
+        CurrentMetrics::add(CurrentMetrics::ZooKeeperRequest);
         std::lock_guard lock(operations_mutex);
         operations[info.request->xid] = info;
     }
@@ -1228,6 +1243,7 @@ void ZooKeeper::pushRequest(RequestInfo && info)
     if (info.watch)
     {
         info.request->has_watch = true;
+        CurrentMetrics::add(CurrentMetrics::ZooKeeperWatch);
         std::lock_guard lock(watches_mutex);
         watches[info.request->getPath()].emplace_back(std::move(info.watch));
     }

dbms/src/Common/ZooKeeper/ZooKeeperImpl.h

@@ -2,6 +2,7 @@
 
 #include <Core/Types.h>
 #include <Common/ConcurrentBoundedQueue.h>
+#include <Common/CurrentMetrics.h>
 
 #include <IO/ReadBuffer.h>
 #include <IO/WriteBuffer.h>
@@ -23,6 +24,12 @@
 #include <functional>
 
 
+namespace CurrentMetrics
+{
+    extern const Metric ZooKeeperSession;
+}
+
+
 namespace ZooKeeperImpl
 {
 
@@ -563,6 +570,8 @@ private:
 
     template <typename T>
     void read(T &);
+
+    CurrentMetrics::Increment metric_increment{CurrentMetrics::ZooKeeperSession};
 };
 
 };