diff --git a/thresh/pom.xml b/thresh/pom.xml
index b8d6268..08205ca 100644
--- a/thresh/pom.xml
+++ b/thresh/pom.xml
@@ -122,7 +122,11 @@
mysql-connector-java
5.1.26
-
+
+ com.timgroup
+ java-statsd-client
+ 3.1.0
+
org.apache.kafka
kafka_2.9.2
diff --git a/thresh/src/main/config/thresh-sample-config.yml b/thresh/src/main/config/thresh-sample-config.yml
index f753877..af89a03 100644
--- a/thresh/src/main/config/thresh-sample-config.yml
+++ b/thresh/src/main/config/thresh-sample-config.yml
@@ -1,6 +1,14 @@
metricSpoutThreads: 2
metricSpoutTasks: 2
+statsdConfig:
+ host: localhost
+ port: 8125
+ prefix: monasca.storm.
+ dimensions: !!map
+ service : monitoring
+ component : storm
+
metricSpoutConfig:
maxWaitTime: 500
diff --git a/thresh/src/main/java/monasca/thresh/ThresholdingConfiguration.java b/thresh/src/main/java/monasca/thresh/ThresholdingConfiguration.java
index 6f23e05..e8b4c60 100644
--- a/thresh/src/main/java/monasca/thresh/ThresholdingConfiguration.java
+++ b/thresh/src/main/java/monasca/thresh/ThresholdingConfiguration.java
@@ -19,6 +19,7 @@ package monasca.thresh;
import monasca.common.configuration.KafkaProducerConfiguration;
import monasca.thresh.infrastructure.thresholding.DataSourceFactory;
+import monasca.thresh.utils.StatsdConfig;
import java.io.Serializable;
import java.util.Set;
@@ -73,4 +74,8 @@ public class ThresholdingConfiguration implements Serializable {
/** Database configuration. */
@Valid @NotNull public DataSourceFactory database = new DataSourceFactory();
+
+ /** StatsD configuration. */
+ @Valid @NotNull public StatsdConfig statsdConfig = new StatsdConfig();
+
}
diff --git a/thresh/src/main/java/monasca/thresh/TopologyModule.java b/thresh/src/main/java/monasca/thresh/TopologyModule.java
index 563a9c2..f7eba1f 100644
--- a/thresh/src/main/java/monasca/thresh/TopologyModule.java
+++ b/thresh/src/main/java/monasca/thresh/TopologyModule.java
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2014 Hewlett-Packard Development Company, L.P.
+ * Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@ import monasca.thresh.infrastructure.thresholding.MetricAggregationBolt;
import monasca.thresh.infrastructure.thresholding.MetricFilteringBolt;
import monasca.thresh.infrastructure.thresholding.MetricSpout;
import monasca.thresh.infrastructure.thresholding.deserializer.EventDeserializer;
+import monasca.thresh.utils.StatsdMetricConsumer;
import monasca.common.util.Injector;
@@ -69,6 +70,30 @@ public class TopologyModule extends AbstractModule {
stormConfig = new Config();
stormConfig.setNumWorkers(config.numWorkerProcesses);
stormConfig.setNumAckers(config.numAckerThreads);
+
+ /* Configure the StatsdMetricConsumer */
+ java.util.Map