Append the Storm component's task ID to the metric name, too

README.md

@@ -85,6 +85,8 @@ System wide deployment requires three steps:
 
 ### Notes
 
+#### Topology Name
+
 You can override the topology name used when reporting to statsd by calling:
 
     statsdConfig.put(Config.TOPOLOGY_NAME, "myTopologyName");
@@ -93,6 +95,20 @@ You can override the topology name used when reporting to statsd by calling:
 
 This will be useful if you use versioned topology names (.e.g. appending a timestamp or a version string), but only care to track them as one in statsd.    
 
+#### Statsd Metric Type
+
+You can configure the Statsd metric type to be sent to Statsd with the following property:
+
+    metrics.statsd.metric_type
+
+Allowed values are `counter` (the default) and `gauge`.
+
+#### Statsd Metric Name
+
+You can include the Storm Task ID in the Statsd metric name by setting the following property to `true`:
+
+    metrics.statsd.include_task_id
+
 ## License
 
 storm-metrics-statsd

src/main/java/com/endgame/storm/metrics/statsd/StatsdMetricConsumer.java

@@ -45,11 +45,41 @@ public class StatsdMetricConsumer implements IMetricsConsumer {
 	public static final String STATSD_HOST = "metrics.statsd.host";
 	public static final String STATSD_PORT = "metrics.statsd.port";
 	public static final String STATSD_PREFIX = "metrics.statsd.prefix";
-
+	public static final String STATSD_METRIC_TYPE = "metrics.statsd.metric_type";
+	public static final String STATSD_INCLUDE_TASKID = "metrics.statsd.include_task_id";
+
+	enum StatsdMetricType {
+		COUNTER("counter"),
+		GAUGE("gauge");
+
+		private String metricType;
+
+		StatsdMetricType(String metricType) {
+			this.metricType = metricType;
+		}
+
+		public String getMetricType() {
+			return metricType;
+		}
+
+		public static StatsdMetricType fromString(String metricType) {
+			if (metricType != null) {
+				for (StatsdMetricType type : StatsdMetricType.values()) {
+					if (metricType.equalsIgnoreCase(type.metricType)) {
+						return type;
+					}
+				}
+			}
+			throw new IllegalArgumentException("No metric type " + metricType + " found");
+		}
+	}
+
 	String topologyName;
 	String statsdHost;
 	int statsdPort = 8125;
 	String statsdPrefix = "storm.metrics.";
+	StatsdMetricType statsdMetricType = StatsdMetricType.COUNTER;
+	boolean includeTaskId = false;
 
 	transient StatsDClient statsd;
 
@@ -84,10 +114,18 @@ void parseConfig(Map conf) {
 				statsdPrefix += ".";
 			}
 		}
+
+		if (conf.containsKey(STATSD_METRIC_TYPE)) {
+			statsdMetricType = StatsdMetricType.fromString((String) conf.get(STATSD_METRIC_TYPE));
+		}
+
+		if (conf.containsKey(STATSD_INCLUDE_TASKID)) {
+			includeTaskId = Boolean.parseBoolean((String) conf.get(STATSD_INCLUDE_TASKID));
+		}
 	}
 
 	String clean(String s) {
-		return s.replace('.', '_').replace('/', '_');
+		return s.replace('.', '_').replace('/', '_').replace(':', '_');
 	}
 
 	@Override
@@ -141,6 +179,10 @@ List<Metric> dataPointsToMetrics(TaskInfo taskInfo,
 				.append(taskInfo.srcWorkerPort).append(".")
 				.append(clean(taskInfo.srcComponentId)).append(".");
 
+		if (includeTaskId) {
+			sb.append(taskInfo.srcTaskId).append(".");
+		}
+
 		int hdrLength = sb.length();
 
 		for (DataPoint p : dataPoints) {
@@ -168,8 +210,17 @@ List<Metric> dataPointsToMetrics(TaskInfo taskInfo,
 	}
 
 	public void report(String s, int number) {
-		LOG.debug("reporting: {}={}", s, number);
-		statsd.count(s, number);
+		LOG.debug("reporting " + statsdMetricType + ": {}={}", s, number);
+
+		switch (statsdMetricType) {
+		case COUNTER:
+			statsd.count(s, number);
+			break;
+		case GAUGE:
+			statsd.recordGaugeValue(s, number);
+			break;
+		default:
+		}
 	}
 
 	@Override