Added ZooKeeper instrumentation for enhanced stats

all/pom.xml

@@ -69,6 +69,12 @@
           <artifactId>netty-common</artifactId></exclusion>
       </exclusions>
     </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>pulsar-zookeeper</artifactId>
+      <version>${project.version}</version>
+    </dependency>
   </dependencies>
 
   <build>

all/src/assemble/bin.xml

@@ -72,6 +72,9 @@
         <exclude>io.netty:netty-handler</exclude>
         <exclude>io.netty:netty-transport-native-epoll</exclude>
         <exclude>io.netty:netty-codec-http</exclude>
+
+        <!-- Already included in pulsar-zookeeper instrumented jar -->
+        <exclude>org.apache.zookeeper:zookeeper</exclude>
       </excludes>
     </dependencySet>
   </dependencySets>

bin/pulsar

@@ -192,12 +192,19 @@ elif [ $COMMAND == "bookie" ]; then
     exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.bookkeeper.proto.BookieServer --conf $PULSAR_BOOKKEEPER_CONF $@
 elif [ $COMMAND == "zookeeper" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"zookeeper.log"}
-    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.zookeeper.server.quorum.QuorumPeerMain $PULSAR_ZK_CONF $@
+
+    # Add instrumentation
+    WEAVER_JAR=`find $PULSAR_HOME/lib -name 'aspectjweaver-*.jar'`
+    if [ -n "$WEAVER_JAR" ]; then OPTS="$OPTS -javaagent:$WEAVER_JAR"; fi
+    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE com.yahoo.pulsar.zookeeper.ZooKeeperStarter $PULSAR_ZK_CONF $@
 elif [ $COMMAND == "global-zookeeper" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"global-zookeeper.log"}
     # Allow global ZK to turn into read-only mode when it cannot reach the quorum
     OPTS="${OPTS} -Dreadonlymode.enabled=true"
-    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.zookeeper.server.quorum.QuorumPeerMain $PULSAR_GLOBAL_ZK_CONF $@
+    # Add instrumentation
+    WEAVER_JAR=`find $PULSAR_HOME/lib -name 'aspectjweaver-*.jar'`
+    if [ -n "$WEAVER_JAR" ]; then OPTS="$OPTS -javaagent:$WEAVER_JAR"; fi
+    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE com.yahoo.pulsar.zookeeper.ZooKeeperStarter $PULSAR_GLOBAL_ZK_CONF $@
 elif [ $COMMAND == "discovery" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"discovery.log"}
     exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE com.yahoo.pulsar.discovery.service.server.DiscoveryServiceStarter $PULSAR_DISCOVERY_CONF $@

conf/log4j.properties

@@ -23,6 +23,8 @@ pulsar.log.dir=logs
 pulsar.log.file=pulsar.log
 
 log4j.rootLogger=${pulsar.root.logger}
+log4j.logger.org.apache.zookeeper.server.ZooKeeperServer=WARN
+log4j.logger.org.apache.zookeeper.ZooKeeper=WARN
 
 # Example with rolling log file
 #log4j.rootLogger=DEBUG, CONSOLE, ROLLINGFILE

pom.xml

@@ -84,6 +84,7 @@ flexible messaging model and an intuitive client API.</description>
     <module>pulsar-testclient</module>
     <module>pulsar-broker-auth-athenz</module>
     <module>pulsar-client-auth-athenz</module>
+    <module>pulsar-zookeeper</module>
     <module>all</module>
   </modules>
 
@@ -103,6 +104,7 @@ flexible messaging model and an intuitive client API.</description>
     <jetty.version>9.3.11.v20160721</jetty.version>
     <athenz.version>1.1.8</athenz.version>
     <prometheus.version>0.0.21</prometheus.version>
+    <aspectj.version>1.8.9</aspectj.version>
   </properties>
 
   <dependencyManagement>
@@ -423,6 +425,12 @@ flexible messaging model and an intuitive client API.</description>
         <version>${prometheus.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>io.prometheus</groupId>
+        <artifactId>simpleclient_servlet</artifactId>
+        <version>${prometheus.version}</version>
+      </dependency>
+
       <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming_2.10</artifactId>
@@ -451,6 +459,18 @@ flexible messaging model and an intuitive client API.</description>
         </exclusions>
       </dependency>
 
+      <dependency>
+        <groupId>org.aspectj</groupId>
+        <artifactId>aspectjrt</artifactId>
+        <version>${aspectj.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>org.aspectj</groupId>
+        <artifactId>aspectjweaver</artifactId>
+        <version>${aspectj.version}</version>
+      </dependency>
+
     </dependencies>
   </dependencyManagement>
 

pulsar-zookeeper/pom.xml

@@ -0,0 +1,129 @@
+<!--
+
+    Copyright 2016 Yahoo Inc.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.yahoo.pulsar</groupId>
+    <artifactId>pulsar</artifactId>
+    <version>1.18-SNAPSHOT</version>
+    <relativePath>..</relativePath>
+  </parent>
+
+  <artifactId>pulsar-zookeeper</artifactId>
+  <packaging>jar</packaging>
+  <name>pulsar-zookeeper</name>
+  <description>Instrumented ZooKeeper with detailed stats</description>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.zookeeper</groupId>
+      <artifactId>zookeeper</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.aspectj</groupId>
+      <artifactId>aspectjrt</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.aspectj</groupId>
+      <artifactId>aspectjweaver</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>io.prometheus</groupId>
+      <artifactId>simpleclient</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>io.prometheus</groupId>
+      <artifactId>simpleclient_hotspot</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>io.prometheus</groupId>
+      <artifactId>simpleclient_servlet</artifactId>
+    </dependency>
+
+     <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-servlet</artifactId>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>aspectj-maven-plugin</artifactId>
+        <version>1.10</version>
+        <configuration>
+          <complianceLevel>1.8</complianceLevel>
+          <source>1.8</source>
+          <target>1.8</target>
+          <showWeaveInfo>true</showWeaveInfo>
+          <weaveDependencies>
+            <weaveDependency>
+              <groupId>org.apache.zookeeper</groupId>
+              <artifactId>zookeeper</artifactId>
+            </weaveDependency>
+          </weaveDependencies>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>process-sources</phase>
+            <goals>
+              <goal>compile</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+    <pluginManagement>
+      <plugins>
+        <!--This plugin's configuration is used to store Eclipse m2e settings only. It has no influence on the Maven build itself.-->
+        <plugin>
+          <groupId>org.eclipse.m2e</groupId>
+          <artifactId>lifecycle-mapping</artifactId>
+          <version>1.0.0</version>
+          <configuration>
+            <lifecycleMappingMetadata>
+              <pluginExecutions>
+                <pluginExecution>
+                  <pluginExecutionFilter>
+                    <groupId>org.codehaus.mojo</groupId>
+                    <artifactId>aspectj-maven-plugin</artifactId>
+                    <versionRange>[1.10,)</versionRange>
+                    <goals>
+                      <goal>compile</goal>
+                    </goals>
+                  </pluginExecutionFilter>
+                  <action>
+                    <ignore></ignore>
+                  </action>
+                </pluginExecution>
+              </pluginExecutions>
+            </lifecycleMappingMetadata>
+          </configuration>
+        </plugin>
+      </plugins>
+    </pluginManagement>
+  </build>
+</project>

pulsar-zookeeper/src/main/java/com/yahoo/pulsar/zookeeper/FinalRequestProcessorAspect.java

@@ -0,0 +1,123 @@
+/**
+ * Copyright 2016 Yahoo Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.yahoo.pulsar.zookeeper;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.server.Request;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.annotation.Pointcut;
+
+import io.prometheus.client.Counter;
+import io.prometheus.client.Summary;
+
+@Aspect
+public class FinalRequestProcessorAspect {
+
+    private static final Map<Integer, String> requestTypeMap = new HashMap<>();
+
+    static {
+        // Prefill the map
+        requestTypeMap.put(OpCode.notification, "notification");
+        requestTypeMap.put(OpCode.create, "create");
+        requestTypeMap.put(OpCode.delete, "delete");
+        requestTypeMap.put(OpCode.exists, "exists");
+        requestTypeMap.put(OpCode.getData, "getData");
+        requestTypeMap.put(OpCode.setData, "setData");
+        requestTypeMap.put(OpCode.getACL, "getACL");
+        requestTypeMap.put(OpCode.setACL, "setACL");
+        requestTypeMap.put(OpCode.getChildren, "getChildren");
+        requestTypeMap.put(OpCode.sync, "sync");
+        requestTypeMap.put(OpCode.ping, "ping");
+        requestTypeMap.put(OpCode.getChildren2, "getChildren2");
+        requestTypeMap.put(OpCode.check, "check");
+        requestTypeMap.put(OpCode.multi, "multi");
+        requestTypeMap.put(OpCode.auth, "auth");
+        requestTypeMap.put(OpCode.setWatches, "setWatches");
+        requestTypeMap.put(OpCode.sasl, "sasl");
+        requestTypeMap.put(OpCode.createSession, "createSession");
+        requestTypeMap.put(OpCode.closeSession, "closeSession");
+        requestTypeMap.put(OpCode.error, "error");
+    }
+
+    private static final Counter requests = Counter
+            .build("zookeeper_server_requests", "Requests issued to a particular server").labelNames("type").create()
+            .register();
+
+    private static final Summary requestsLatency = Summary.build().name("zookeeper_server_requests_latency_ms")
+            .help("Requests latency in millis") //
+            .quantile(0.50, 0.01) //
+            .quantile(0.75, 0.01) //
+            .quantile(0.95, 0.01) //
+            .quantile(0.99, 0.01) //
+            .quantile(0.999, 0.01) //
+            .quantile(0.9999, 0.01) //
+            .quantile(1.0, 0.01) //
+            .maxAgeSeconds(60) //
+            .labelNames("type") //
+            .create().register();
+
+    @Pointcut("execution(void org.apache.zookeeper.server.FinalRequestProcessor.processRequest(..))")
+    public void processRequest() {
+    }
+
+    @Around("processRequest()")
+    public void timedProcessRequest(ProceedingJoinPoint joinPoint) throws Throwable {
+        joinPoint.proceed();
+
+        Request request = (Request) joinPoint.getArgs()[0];
+
+        String type = requestTypeMap.getOrDefault(request.type, "unknown");
+        requests.labels(type).inc();
+
+        long latencyMs = System.currentTimeMillis() - request.createTime;
+        String latencyLabel = isWriteRequest(request.type) ? "write" : "read";
+        requestsLatency.labels(latencyLabel).observe(latencyMs);
+    }
+
+    private static boolean isWriteRequest(int opCode) {
+        switch (opCode) {
+        case OpCode.create:
+        case OpCode.delete:
+        case OpCode.setData:
+        case OpCode.setACL:
+        case OpCode.sync:
+        case OpCode.createSession:
+        case OpCode.closeSession:
+            return true;
+
+        case OpCode.notification:
+        case OpCode.exists:
+        case OpCode.getData:
+        case OpCode.getACL:
+        case OpCode.getChildren:
+        case OpCode.ping:
+        case OpCode.getChildren2:
+        case OpCode.check:
+        case OpCode.multi:
+        case OpCode.auth:
+        case OpCode.setWatches:
+        case OpCode.sasl:
+        case OpCode.error:
+        default:
+            return false;
+        }
+    }
+}