STORM-1723 Introduce ClusterMetricsConsumer

* ClusterMetricsConsumer publishes cluster-side related metrics into consumers
  * like MetricsConsumer for topology metrics
* Users can implement IClusterMetricsConsumer and configure to cluster conf. file to take effect
  * Please refer conf/storm.yaml.example for more details on configuring
  * Nimbus should be launched with additional jars which are needed for IClusterMetricsConsumer
* Also did some refactor to nimbus.clj

Author: HeartSaVioR

conf/defaults.yaml

@@ -287,6 +287,9 @@ pacemaker.kerberos.users: []
 storm.daemon.metrics.reporter.plugins:
      - "org.apache.storm.daemon.metrics.reporters.JmxPreparableReporter"
 
+# configuration of cluster metrics consumer
+storm.cluster.metrics.consumer.publish.interval.secs: 60
+
 storm.resource.isolation.plugin: "org.apache.storm.container.cgroup.CgroupManager"
 # Also determines whether the unit tests for cgroup runs.  
 # If storm.resource.isolation.plugin.enable is set to false the unit tests for cgroups will not run

conf/storm.yaml.example

@@ -46,3 +46,12 @@
 #     parallelism.hint: 1
 #     argument:
 #       - endpoint: "metrics-collector.mycompany.org"
+
+## Cluster Metrics Consumers
+# storm.cluster.metrics.consumer.register:
+#   - class: "org.apache.storm.metric.LoggingClusterMetricsConsumer"
+#   - class: "org.mycompany.MyMetricsConsumer"
+#     argument:
+#       - endpoint: "metrics-collector.mycompany.org"
+#
+# storm.cluster.metrics.consumer.publish.interval.secs: 60

log4j2/cluster.xml

@@ -54,6 +54,17 @@
         </Policies>
         <DefaultRolloverStrategy max="9"/>
     </RollingFile>
+    <RollingFile name="METRICS"
+                 fileName="${sys:storm.log.dir}/${sys:logfile.name}.metrics"
+                 filePattern="${sys:storm.log.dir}/${sys:logfile.name}.metrics.%i.gz">
+        <PatternLayout>
+            <pattern>${patternMetrics}</pattern>
+        </PatternLayout>
+        <Policies>
+            <SizeBasedTriggeringPolicy size="2 MB"/>
+        </Policies>
+        <DefaultRolloverStrategy max="9"/>
+    </RollingFile>
     <Syslog name="syslog" format="RFC5424" charset="UTF-8" host="localhost" port="514"
             protocol="UDP" appName="[${sys:daemon.name}]" mdcId="mdc" includeMDC="true"
             facility="LOCAL5" enterpriseNumber="18060" newLine="true" exceptionPattern="%rEx{full}"
@@ -69,6 +80,9 @@
         <AppenderRef ref="THRIFT-ACCESS"/>
         <AppenderRef ref="syslog"/>
     </Logger>
+    <Logger name="org.apache.storm.metric.LoggingClusterMetricsConsumer" level="info" additivity="false">
+        <appender-ref ref="METRICS"/>
+    </Logger>
     <root level="info"> <!-- We log everything -->
         <appender-ref ref="A1"/>
         <appender-ref ref="syslog"/>

storm-core/src/clj/org/apache/storm/daemon/nimbus.clj

@@ -54,12 +54,16 @@
   (:import [org.apache.storm.cluster ClusterStateContext DaemonType StormClusterStateImpl ClusterUtils])
   (:use [org.apache.storm util config log converter])
   (:require [org.apache.storm [converter :as converter]])
+  (:require [org.apache.storm.ui.core :as ui])
   (:require [clojure.set :as set])
   (:import [org.apache.storm.daemon.common StormBase Assignment])
   (:import [org.apache.storm.zookeeper Zookeeper])
   (:use [org.apache.storm.daemon common])
   (:use [org.apache.storm config])
   (:import [org.apache.zookeeper data.ACL ZooDefs$Ids ZooDefs$Perms])
+  (:import [org.apache.storm.metric ClusterMetricsConsumerExecutor]
+           [org.apache.storm.metric.api IClusterMetricsConsumer$ClusterInfo DataPoint IClusterMetricsConsumer$SupervisorInfo]
+           [org.apache.storm Config])
   (:import [org.apache.storm.utils VersionInfo LocalState]
            [org.json.simple JSONValue])
   (:require [clj-time.core :as time])
@@ -167,6 +171,13 @@
         (catch Exception e
           (log-warn-error e "Ingoring exception, Could not initialize " (conf NIMBUS-TOPOLOGY-ACTION-NOTIFIER-PLUGIN)))))))
 
+(defn mk-cluster-metrics-consumer-executors [storm-conf]
+  (map
+    (fn [consumer]
+      (ClusterMetricsConsumerExecutor. (get consumer "class")
+                                       (get consumer "argument")))
+    (get storm-conf STORM-CLUSTER-METRICS-CONSUMER-REGISTER)))
+
 (defn nimbus-data [conf inimbus]
   (let [forced-scheduler (.getForcedScheduler inimbus)]
     {:conf conf
@@ -209,6 +220,7 @@
      :topo-history-state (ConfigUtils/nimbusTopoHistoryState conf)
      :nimbus-autocred-plugins (AuthUtils/getNimbusAutoCredPlugins conf)
      :nimbus-topology-action-notifier (create-tology-action-notifier conf)
+     :cluster-consumer-executors (mk-cluster-metrics-consumer-executors conf)
      }))
 
 (defn inbox [nimbus]
@@ -1358,13 +1370,42 @@
   (and (>= val lower)
     (<= val upper)))
 
-;TODO: when translating this function, you should replace the map-val with a proper for loop HERE
-(defserverfn service-handler [conf inimbus]
-  (.prepare inimbus conf (ConfigUtils/masterInimbusDir conf))
-  (log-message "Starting Nimbus with conf " conf)
-  (let [nimbus (nimbus-data conf inimbus)
-        blob-store (:blob-store nimbus)
-        principal-to-local (AuthUtils/GetPrincipalToLocalPlugin conf)
+(defn extract-cluster-metrics [^ClusterSummary summ]
+  (let [cluster-summ (ui/cluster-summary summ "nimbus")]
+    {:cluster-info (IClusterMetricsConsumer$ClusterInfo. (System/currentTimeMillis))
+     :data-points  (map
+                     (fn [[k v]] (DataPoint. k v))
+                     (select-keys cluster-summ ["supervisors" "topologies" "slotsTotal" "slotsUsed" "slotsFree"
+                                                "executorsTotal" "tasksTotal"]))}))
+(defn extract-supervisors-metrics [^ClusterSummary summ]
+  (let [sups (.get_supervisors summ)
+        supervisors-summ ((ui/supervisor-summary sups) "supervisors")]
+    (map (fn [supervisor-summ]
+           {:supervisor-info (IClusterMetricsConsumer$SupervisorInfo.
+                               (supervisor-summ "host")
+                               (supervisor-summ "id")
+                               (System/currentTimeMillis))
+            :data-points     (map
+                               (fn [[k v]] (DataPoint. k v))
+                               (select-keys supervisor-summ ["slotsTotal" "slotsUsed" "totalMem" "totalCpu"
+                                                             "usedMem" "usedCpu"]))})
+         supervisors-summ)))
+
+(defn send-cluster-metrics-to-executors [nimbus-service nimbus]
+  (let [cluster-summary (.getClusterInfo nimbus-service)
+        cluster-metrics (extract-cluster-metrics cluster-summary)
+        supervisors-metrics (extract-supervisors-metrics cluster-summary)]
+    (dofor
+      [consumer-executor (:cluster-consumer-executors nimbus)]
+      (do
+        (.handleDataPoints consumer-executor (:cluster-info cluster-metrics) (:data-points cluster-metrics))
+        (dofor
+          [supervisor-metrics supervisors-metrics]
+          (do
+            (.handleDataPoints consumer-executor (:supervisor-info supervisor-metrics) (:data-points supervisor-metrics))))))))
+
+(defn mk-reified-nimbus [nimbus conf blob-store]
+  (let [principal-to-local (AuthUtils/GetPrincipalToLocalPlugin conf)
         admin-users (or (.get conf NIMBUS-ADMINS) [])
         get-common-topo-info
           (fn [^String storm-id operation]
@@ -1400,64 +1441,6 @@
                            (doto (ErrorInfo. (:error e) (:time-secs e))
                              (.set_host (:host e))
                              (.set_port (:port e)))))]
-    (.prepare ^org.apache.storm.nimbus.ITopologyValidator (:validator nimbus) conf)
-
-    ;add to nimbuses
-    (.addNimbusHost (:storm-cluster-state nimbus) (.toHostPortString (:nimbus-host-port-info nimbus))
-      (NimbusSummary.
-        (.getHost (:nimbus-host-port-info nimbus))
-        (.getPort (:nimbus-host-port-info nimbus))
-        (Time/currentTimeSecs)
-        false ;is-leader
-        STORM-VERSION))
-
-    (.addToLeaderLockQueue (:leader-elector nimbus))
-    (cleanup-corrupt-topologies! nimbus)
-    (when (instance? LocalFsBlobStore blob-store)
-      ;register call back for blob-store
-      (.blobstore (:storm-cluster-state nimbus) (fn [] (blob-sync conf nimbus)))
-      (setup-blobstore nimbus))
-
-    (when (is-leader nimbus :throw-exception false)
-      (doseq [storm-id (.activeStorms (:storm-cluster-state nimbus))]
-        (transition! nimbus storm-id :startup)))
-
-    (.scheduleRecurring (:timer nimbus)
-      0
-      (conf NIMBUS-MONITOR-FREQ-SECS)
-      (fn []
-        (when-not (conf ConfigUtils/NIMBUS_DO_NOT_REASSIGN)
-          (locking (:submit-lock nimbus)
-            (mk-assignments nimbus)))
-        (do-cleanup nimbus)))
-    ;; Schedule Nimbus inbox cleaner
-    (.scheduleRecurring (:timer nimbus)
-      0
-      (conf NIMBUS-CLEANUP-INBOX-FREQ-SECS)
-      (fn [] (clean-inbox (inbox nimbus) (conf NIMBUS-INBOX-JAR-EXPIRATION-SECS))))
-    ;; Schedule nimbus code sync thread to sync code from other nimbuses.
-    (if (instance? LocalFsBlobStore blob-store)
-      (.scheduleRecurring (:timer nimbus)
-        0
-        (conf NIMBUS-CODE-SYNC-FREQ-SECS)
-        (fn [] (blob-sync conf nimbus))))
-    ;; Schedule topology history cleaner
-    (when-let [interval (conf LOGVIEWER-CLEANUP-INTERVAL-SECS)]
-      (.scheduleRecurring (:timer nimbus)
-        0
-        (conf LOGVIEWER-CLEANUP-INTERVAL-SECS)
-        (fn [] (clean-topology-history (conf LOGVIEWER-CLEANUP-AGE-MINS) nimbus))))
-    (.scheduleRecurring (:timer nimbus)
-      0
-      (conf NIMBUS-CREDENTIAL-RENEW-FREQ-SECS)
-      (fn []
-        (renew-credentials nimbus)))
-
-    (def nimbus:num-supervisors (StormMetricsRegistry/registerGauge "nimbus:num-supervisors"
-      (fn [] (.size (.supervisors (:storm-cluster-state nimbus) nil)))))
-
-    (StormMetricsRegistry/startMetricsReporters conf)
-    
     (reify Nimbus$Iface
       (^void submitTopologyWithOpts
         [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology
@@ -2195,6 +2178,85 @@
       (isWaiting [this]
         (.isTimerWaiting (:timer nimbus))))))
 
+;TODO: when translating this function, you should replace the map-val with a proper for loop HERE
+(defserverfn service-handler [conf inimbus]
+  (.prepare inimbus conf (ConfigUtils/masterInimbusDir conf))
+  (log-message "Starting Nimbus with conf " conf)
+  (let [nimbus (nimbus-data conf inimbus)
+        blob-store (:blob-store nimbus)]
+    (.prepare ^org.apache.storm.nimbus.ITopologyValidator (:validator nimbus) conf)
+
+    ;add to nimbuses
+    (.addNimbusHost (:storm-cluster-state nimbus) (.toHostPortString (:nimbus-host-port-info nimbus))
+      (NimbusSummary.
+        (.getHost (:nimbus-host-port-info nimbus))
+        (.getPort (:nimbus-host-port-info nimbus))
+        (Time/currentTimeSecs)
+        false ;is-leader
+        STORM-VERSION))
+
+    (.addToLeaderLockQueue (:leader-elector nimbus))
+    (cleanup-corrupt-topologies! nimbus)
+    (when (instance? LocalFsBlobStore blob-store)
+      ;register call back for blob-store
+      (.blobstore (:storm-cluster-state nimbus) (fn [] (blob-sync conf nimbus)))
+      (setup-blobstore nimbus))
+
+    (doseq [consumer (:cluster-consumer-executors nimbus)]
+      (.prepare consumer))
+
+    (when (is-leader nimbus :throw-exception false)
+      (doseq [storm-id (.activeStorms (:storm-cluster-state nimbus))]
+        (transition! nimbus storm-id :startup)))
+
+    (.scheduleRecurring (:timer nimbus)
+      0
+      (conf NIMBUS-MONITOR-FREQ-SECS)
+      (fn []
+        (when-not (conf ConfigUtils/NIMBUS_DO_NOT_REASSIGN)
+          (locking (:submit-lock nimbus)
+            (mk-assignments nimbus)))
+        (do-cleanup nimbus)))
+    ;; Schedule Nimbus inbox cleaner
+    (.scheduleRecurring (:timer nimbus)
+      0
+      (conf NIMBUS-CLEANUP-INBOX-FREQ-SECS)
+      (fn [] (clean-inbox (inbox nimbus) (conf NIMBUS-INBOX-JAR-EXPIRATION-SECS))))
+    ;; Schedule nimbus code sync thread to sync code from other nimbuses.
+    (if (instance? LocalFsBlobStore blob-store)
+      (.scheduleRecurring (:timer nimbus)
+        0
+        (conf NIMBUS-CODE-SYNC-FREQ-SECS)
+        (fn [] (blob-sync conf nimbus))))
+    ;; Schedule topology history cleaner
+    (when-let [interval (conf LOGVIEWER-CLEANUP-INTERVAL-SECS)]
+      (.scheduleRecurring (:timer nimbus)
+        0
+        (conf LOGVIEWER-CLEANUP-INTERVAL-SECS)
+        (fn [] (clean-topology-history (conf LOGVIEWER-CLEANUP-AGE-MINS) nimbus))))
+    (.scheduleRecurring (:timer nimbus)
+      0
+      (conf NIMBUS-CREDENTIAL-RENEW-FREQ-SECS)
+      (fn []
+        (renew-credentials nimbus)))
+
+    (def nimbus:num-supervisors (StormMetricsRegistry/registerGauge "nimbus:num-supervisors"
+      (fn [] (.size (.supervisors (:storm-cluster-state nimbus) nil)))))
+
+    (StormMetricsRegistry/startMetricsReporters conf)
+
+    (let [reified-inimbus (mk-reified-nimbus nimbus conf blob-store)]
+      (do
+        (if (:cluster-consumer-executors nimbus)
+          (.scheduleRecurring (:timer nimbus)
+            0
+            (conf STORM-CLUSTER-METRICS-CONSUMER-PUBLISH-INTERVAL-SECS)
+            (fn []
+              (when (is-leader nimbus :throw-exception false)
+                (send-cluster-metrics-to-executors reified-inimbus nimbus))))))
+      reified-inimbus)))
+
+
 (defn validate-port-available[conf]
   (try
     (let [socket (ServerSocket. (conf NIMBUS-THRIFT-PORT))]

storm-core/src/jvm/org/apache/storm/Config.java

@@ -1519,6 +1519,22 @@ public class Config extends HashMap<String, Object> {
     @isPositiveNumber
     public static final String BACKPRESSURE_DISRUPTOR_LOW_WATERMARK="backpressure.disruptor.low.watermark";
 
+    /**
+     * A list of classes implementing IClusterMetricsConsumer (See storm.yaml.example for exact config format).
+     * Each listed class will be routed cluster related metrics data.
+     * Each listed class maps 1:1 to a ClusterMetricsConsumerExecutor and they're executed in Nimbus.
+     * Only consumers which run in leader Nimbus receives metrics data.
+     */
+    @isListEntryCustom(entryValidatorClasses = {ClusterMetricRegistryValidator.class})
+    public static final String STORM_CLUSTER_METRICS_CONSUMER_REGISTER = "storm.cluster.metrics.consumer.register";
+
+    /**
+     * How often cluster metrics data is published to metrics consumer.
+     */
+    @NotNull
+    @isPositiveNumber
+    public static final String STORM_CLUSTER_METRICS_CONSUMER_PUBLISH_INTERVAL_SECS = "storm.cluster.metrics.consumer.publish.interval.secs";
+
     /**
      * A list of users that are allowed to interact with the topology.  To use this set
      * nimbus.authorizer to org.apache.storm.security.auth.authorizer.SimpleACLAuthorizer

storm-core/src/jvm/org/apache/storm/metric/ClusterMetricsConsumerExecutor.java

@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.storm.metric;
+
+import org.apache.storm.Config;
+import org.apache.storm.metric.api.DataPoint;
+import org.apache.storm.metric.api.IClusterMetricsConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+
+public class ClusterMetricsConsumerExecutor {
+    public static final Logger LOG = LoggerFactory.getLogger(ClusterMetricsConsumerExecutor.class);
+
+    private IClusterMetricsConsumer metricsConsumer;
+    private String consumerClassName;
+    private Object registrationArgument;
+
+    public ClusterMetricsConsumerExecutor(String consumerClassName, Object registrationArgument) {
+        this.consumerClassName = consumerClassName;
+        this.registrationArgument = registrationArgument;
+    }
+
+    public void prepare() {
+        try {
+            metricsConsumer = (IClusterMetricsConsumer)Class.forName(consumerClassName).newInstance();
+        } catch (Exception e) {
+            throw new RuntimeException("Could not instantiate a class listed in config under section " +
+                    Config.STORM_CLUSTER_METRICS_CONSUMER_REGISTER + " with fully qualified name " + consumerClassName, e);
+        }
+
+        metricsConsumer.prepare(registrationArgument);
+    }
+
+    public void handleDataPoints(final IClusterMetricsConsumer.ClusterInfo clusterInfo, final Collection<DataPoint> dataPoints) {
+        try {
+            metricsConsumer.handleDataPoints(clusterInfo, dataPoints);
+        } catch (Throwable e) {
+            LOG.error("Error while handling cluster data points, consumer class: " + consumerClassName, e);
+        }
+    }
+
+    public void handleDataPoints(final IClusterMetricsConsumer.SupervisorInfo supervisorInfo, final Collection<DataPoint> dataPoints) {
+        try {
+            metricsConsumer.handleDataPoints(supervisorInfo, dataPoints);
+        } catch (Throwable e) {
+            LOG.error("Error while handling cluster data points, consumer class: " + consumerClassName, e);
+        }
+    }
+
+    public void cleanup() {
+        metricsConsumer.cleanup();
+    }
+}