[STORM-3565] add support for adding dimensions to metric data points

storm-client/src/jvm/org/apache/storm/executor/Executor.java

@@ -309,10 +309,12 @@ public void metricsTick(Task task, TupleImpl tuple) {
             List<IMetricsConsumer.DataPoint> dataPoints = new ArrayList<>();
             if (nameToRegistry != null) {
                 for (Map.Entry<String, IMetric> entry : nameToRegistry.entrySet()) {
+                    String name = entry.getKey();
                     IMetric metric = entry.getValue();
                     Object value = metric.getValueAndReset();
+                    Map<String, String> dimensions = metric.getDimensions();
                     if (value != null) {
-                        IMetricsConsumer.DataPoint dataPoint = new IMetricsConsumer.DataPoint(entry.getKey(), value);
+                        IMetricsConsumer.DataPoint dataPoint = new IMetricsConsumer.DataPoint(name, value, dimensions);
                         dataPoints.add(dataPoint);
                     }
                 }

storm-client/src/jvm/org/apache/storm/metric/api/DimensionalCountMetric.java

@@ -0,0 +1,56 @@
+/**
+ * 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.api;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class DimensionalCountMetric implements IDimensionalMetric {
+
+    Map<Map<String, String>, CountMetric> dimToMetrics;     // dimension map : submetric
+
+    public void incr(Map<String, String> dimensions) {
+        incrBy(dimensions, 1);
+    }
+
+    public void incrBy(Map<String, String> dimensions, long incrementBy) {
+        CountMetric countMetric = dimToMetrics.get(dimensions);
+        if (countMetric == null) {
+            countMetric = new CountMetric();
+        }
+        countMetric.incrBy(incrementBy);
+    }
+
+
+    @Override
+    public Map<Map<String, String>, IMetric> getDimToMetric() {
+        return Collections.unmodifiableMap(dimToMetrics);
+    }
+
+    @Override
+    public Map<Map<String, String>, Object> getValueAndReset() {
+        Map<Map<String, String>, Object> ret = new HashMap<>();
+        for (Map.Entry<Map<String, String>, CountMetric> e : dimToMetrics.entrySet()) {
+            Map<String, String> dimensions = e.getKey();
+            ret.put(dimensions, getValueAndResetByDimensions(dimensions));
+        }
+        return ret;
+    }
+
+    @Override
+    public Object getValueAndResetByDimensions(Map<String, String> dimensions) {
+        CountMetric subMetric = dimToMetrics.get(dimensions);
+        return subMetric.getValueAndReset();
+    }
+}

storm-client/src/jvm/org/apache/storm/metric/api/IDimensionalMetric.java

@@ -0,0 +1,32 @@
+/**
+ * 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.api;
+
+import java.util.Map;
+
+public interface IDimensionalMetric extends IMetric {
+
+    /**
+     * Dimensions is a collection of additional key-value metadata which is optional.
+     * It could be useful for further fine-grained measurement for a metric.
+     * It may also help better grouping in downstream metric monitoring system.
+     *
+     * @return a map of dimension name to a sub-metric
+     */
+    Map<Map<String, String>, IMetric> getDimToMetric();   // dimensions : sub-metric
+
+    @Override
+    Map<Map<String, String>, Object> getValueAndReset();      // dimensions : sub-metric value
+
+    Object getValueAndResetByDimensions(Map<String, String> dimensions);
+}

storm-client/src/jvm/org/apache/storm/metric/api/IMetric.java

@@ -12,16 +12,39 @@
 
 package org.apache.storm.metric.api;
 
+import java.util.Collections;
+import java.util.Map;
+
 /**
  * Produces metrics.
+ * Usually, metric is a measurement identified by a name string.
+ * Dimensions are a collection of additional key-value metadata map containing extra information of this measurement.
+ * It is optional when customizing your metric by implementing this interface
  */
 public interface IMetric {
     /**
      * Get value and reset.
-     * @return an object that will be sent sent to {@link IMetricsConsumer#handleDataPoints(org.apache.storm.metric.api.IMetricsConsumer
-     * .TaskInfo,
-     *     java.util.Collection)}. If null is returned nothing will be sent. If this value can be reset, like with a counter, a side effect
+     * @return an object that will be sent to
+     *     {@link IMetricsConsumer#handleDataPoints(org.apache.storm.metric.api.IMetricsConsumer.TaskInfo, java.util.Collection)}.
+     *     If null is returned nothing will be sent. If this value can be reset, like with a counter, a side effect
      *     of calling this should be that the value is reset.
      */
     Object getValueAndReset();
+
+
+    /**
+     * Check whether this metric is carrying additional dimension map.
+     * @return a boolean value.
+     */
+    default boolean isDimensional() {
+        return false;
+    }
+
+    /**
+     * Get dimension map. An empty map will be returned if metric is not dimensional.
+     * @return a K-V map of the additional metadata.
+     */
+    default Map<String, String> getDimensions() {
+        return Collections.emptyMap();
+    }
 }

storm-client/src/jvm/org/apache/storm/metric/api/IMetricsConsumer.java

@@ -13,6 +13,7 @@
 package org.apache.storm.metric.api;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 import org.apache.storm.task.IErrorReporter;
@@ -59,18 +60,34 @@ public String toString() {
     class DataPoint {
         public String name;
         public Object value;
+        public Map<String, String> dimensions;
 
         public DataPoint() {
         }
 
         public DataPoint(String name, Object value) {
+            this(name, value, Collections.emptyMap());
+        }
+
+        public DataPoint(String name, Object value, Map<String, String> dimensions) {
             this.name = name;
             this.value = value;
+            this.dimensions = dimensions;
         }
 
         @Override
         public String toString() {
-            return "[" + name + " = " + value + "]";
+            StringBuilder sb = new StringBuilder();
+            sb.append("[");
+            sb.append(name);
+            sb.append("=");
+            sb.append(value);
+            if (!dimensions.isEmpty()) {
+                sb.append(", ");
+                sb.append(dimensions.toString());
+            }
+            sb.append("]");
+            return sb.toString();
         }
 
         @Override
@@ -84,13 +101,16 @@ public boolean equals(Object o) {
 
             DataPoint dataPoint = (DataPoint) o;
 
-            return Objects.equals(name, dataPoint.name) && Objects.deepEquals(value, dataPoint.value);
+            return Objects.equals(name, dataPoint.name)
+                && Objects.deepEquals(value, dataPoint.value)
+                && Objects.deepEquals(dimensions, dataPoint.dimensions);
         }
 
         @Override
         public int hashCode() {
             int result = name != null ? name.hashCode() : 0;
             result = 31 * result + (value != null ? value.hashCode() : 0);
+            result = 31 * result + (dimensions != null ? dimensions.hashCode() : 0);
             return result;
         }
     }