Emqtt batch output

etc/telegraf.conf

@@ -543,6 +543,13 @@
 #   ## Use SSL but skip chain & host verification
 #   # insecure_skip_verify = false
 #
+#   ## Batch messages in a topic
+#   ## batch = false
+#   ## Flag to determine if messages sent in a topic in a flush interval,
+#   ## need to be batched into one message.
+#   ## batch = true, batches the messages in a topic to one messages
+#   ## batch = false, default behaviour
+#
 #   ## Data format to output.
 #   ## Each data format has its own unique set of configuration options, read
 #   ## more about them here:
@@ -1740,19 +1747,19 @@
 #   ## List of metrics collected on above servers
 #   ## Each metric consists in a name, a jmx path and either
 #   ## a pass or drop slice attribute.
-#   ## This collect all heap memory usage metrics.
+#   ## This collect all heap memory usage metrics.
 #   [[inputs.jolokia.metrics]]
 #     name = "heap_memory_usage"
 #     mbean  = "java.lang:type=Memory"
 #     attribute = "HeapMemoryUsage"
 #
-#   ## This collect thread counts metrics.
+#   ## This collect thread counts metrics.
 #   [[inputs.jolokia.metrics]]
 #     name = "thread_count"
 #     mbean  = "java.lang:type=Threading"
 #     attribute = "TotalStartedThreadCount,ThreadCount,DaemonThreadCount,PeakThreadCount"
 #
-#   ## This collect number of class loaded/unloaded counts metrics.
+#   ## This collect number of class loaded/unloaded counts metrics.
 #   [[inputs.jolokia.metrics]]
 #     name = "class_count"
 #     mbean  = "java.lang:type=ClassLoading"
@@ -2201,7 +2208,7 @@
 #   reverse_metric_names = true
 
 
-# # A plugin to collect stats from Opensmtpd - a validating, recursive, and caching DNS resolver 
+# # A plugin to collect stats from Opensmtpd - a validating, recursive, and caching DNS resolver
 # [[inputs.opensmtpd]]
 #   ## If running as a restricted user you can prepend sudo for additional access:
 #   #use_sudo = false
@@ -3456,4 +3463,3 @@
 # [[inputs.zipkin]]
 #   # path = "/api/v1/spans" # URL path for span data
 #   # port = 9411            # Port on which Telegraf listens
-

plugins/outputs/mqtt/README.md

@@ -30,6 +30,9 @@ This plugin writes to a [MQTT Broker](http://http://mqtt.org/) acting as a mqtt
   ## Use SSL but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Batch messages sent on a topic in a flush interval
+  # batch = false
+
   ## Data format to output.
   data_format = "influx"
 ```
@@ -49,4 +52,5 @@ This plugin writes to a [MQTT Broker](http://http://mqtt.org/) acting as a mqtt
 * `ssl_cert`: SSL CERT
 * `ssl_key`: SSL key
 * `insecure_skip_verify`: Use SSL but skip chain & host verification (default: false)
+* `batch`: Batch messages sent on a topic within a flush interval (default: false)
 * `data_format`: [About Telegraf data formats](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md)

plugins/outputs/mqtt/mqtt.go

@@ -47,14 +47,15 @@ var sampleConfig = `
 `
 
 type MQTT struct {
-	Servers     []string `toml:"servers"`
-	Username    string
-	Password    string
-	Database    string
-	Timeout     internal.Duration
-	TopicPrefix string
-	QoS         int    `toml:"qos"`
-	ClientID    string `toml:"client_id"`
+	Servers      []string `toml:"servers"`
+	Username     string
+	Password     string
+	Database     string
+	Timeout      internal.Duration
+	TopicPrefix  string
+	QoS          int    `toml:"qos"`
+	ClientID     string `toml:"client_id"`
+	BatchMessage bool   `toml:"batch"`
 
 	// Path to CA file
 	SSLCA string `toml:"ssl_ca"`
@@ -124,6 +125,8 @@ func (m *MQTT) Write(metrics []telegraf.Metric) error {
 		hostname = ""
 	}
 
+	metricsmap := make(map[string][]byte)
+
 	for _, metric := range metrics {
 		var t []string
 		if m.TopicPrefix != "" {
@@ -141,7 +144,18 @@ func (m *MQTT) Write(metrics []telegraf.Metric) error {
 			return err
 		}
 
-		err = m.publish(topic, buf)
+		if m.BatchMessage {
+			metricsmap[topic] = append(metricsmap[topic], buf...)
+		} else {
+			err = m.publish(topic, buf)
+			if err != nil {
+				return fmt.Errorf("Could not write to MQTT server, %s", err)
+			}
+		}
+	}
+
+	for key := range metricsmap {
+		err := m.publish(key, metricsmap[key])
 		if err != nil {
 			return fmt.Errorf("Could not write to MQTT server, %s", err)
 		}