Merge pull request #14 from hikhvar/support-tasmota

Support tasmota

Readme.md

@@ -8,28 +8,27 @@ publish the received messages as prometheus metrics. I wrote this exporter to pu
 metrics from small embedded sensors based on the NodeMCU to prometheus. The used arduino scetch can be found in the [dht22tomqtt](https://github.com/hikhvar/dht22tomqtt) repository.
 
 ## Assumptions about Messages and Topics
-This exporter makes some assumptions about the message format and MQTT topics. This exporter assumes that each
-client publish the metrics into a dedicated topic. The last level topic becomes the `sensor` label in prometheus.
-This exporter assume that the message are JSON objects with only float fields. The golang type for the messages is: 
-
-```go
-type MQTTPayload map[string]float64
-```
-
-For example the message
+This exporter makes some assumptions about the MQTT topics. This exporter assumes that each
+client publish the metrics into a dedicated topic. The regular expression ìn the configuration field `mqtt.device_id_regex`
+defines how to extract the device ID from the MQTT topic. This allow an arbitrary place of the device ID in the mqtt topic.
+For example the [tasmota](https://github.com/arendst/Tasmota) firmware pushes the telemetry data to the topics `tele/<deviceid>/SENSOR`.
 
+Let us assume the default configuration from [#ConfigFile]. A sensor publishes the following message
 ```json
-{"temperature":23.20,"humidity":51.60,"heat_index":22.92}
+{"temperature":23.20,"humidity":51.60, "computed": {"heat_index":22.92} }
 ```
 
-published to the MQTT topic `devices/me/livingroom` becomes the following prometheus metrics:
+to the MQTT topic `devices/me/livingroom`. This message becomes the following prometheus metrics:
 
 ```text
-temperature{sensor="livingroom"} 23.2
-heat_index{sensor="livingroom"} 22.92
-humidity{sensor="livingroom"} 51.6
+temperature{sensor="livingroom",topic="devices/me/livingroom"} 23.2
+heat_index{sensor="livingroom",topic="devices/me/livingroom"} 22.92
+humidity{sensor="livingroom",topic="devices/me/livingroom"} 51.6
 ```
 
+### Tasmota
+An example configuration for the tasmota based Gosund SP111 device is given in [examples/gosund_sp111.yaml](examples/gosund_sp111.yaml).
+
 ## Build
 
 To build the exporter run:
@@ -85,18 +84,23 @@ The config file can look like this:
 mqtt:
   # The MQTT broker to connect to
   server: tcp://127.0.0.1:1883
-  # The Topic path to subscripe to. Actually this will become `$topic_path/+`
-  topic_path: v1/devices/me
+  # The Topic path to subscripe to. Be aware that you have to specify the wildcard. 
+  topic_path: v1/devices/me/+
+  # Optional: Regular expression to extract the device ID from the topic path. The default regular expression, assumes
+  # that the last "element" of the topic_path is the device id.
+  # The regular expression must contain a named capture group with the name deviceid
+  # For example the expression for tasamota based sensors is "tele/(?P<deviceid>.*)/.*"
+  device_id_regex: "(.*/)?(?P<deviceid>.*)"
   # The MQTT QoS level
   qos: 0
 cache:
   # Timeout. Each received metric will be presented for this time if no update is send via MQTT
-  timeout: 2min
+  timeout: 24h
 # This is a list of valid metrics. Only metrics listed here will be exported
 metrics:
     # The name of the metric in prometheus
   - prom_name: temperature_celsius
-    # The name of the metric in a MQTT JSON message
+    # The name of the metric in a MQTT JSON message. This can be an arbitrary gojsonq path.
     mqtt_name: temperature
     # The prometheus help text for this metric
     help: DHT22 temperature reading
@@ -118,8 +122,8 @@ metrics:
       sensor_type: dht22
     # The name of the metric in prometheus
   - prom_name: heat_index
-    # The name of the metric in a MQTT JSON message
-    mqtt_name: heat_index
+    # The name of the metric in a MQTT JSON message. Here a nested field.
+    mqtt_name: computed.heat_index
     # The prometheus help text for this metric
     help: DHT22 heatIndex calculation
     # The prometheus type for this metric. Valid values are: "gauge" and "counter"

cmd/mqtt2prometheus.go

@@ -51,13 +51,13 @@ func main() {
 	mqttClientOptions.SetPassword(cfg.MQTT.Password)
 
 	collector := metrics.NewCollector(cfg.Cache.Timeout, cfg.Metrics)
-	ingest := metrics.NewIngest(collector, cfg.Metrics)
+	ingest := metrics.NewIngest(collector, cfg.Metrics, cfg.MQTT.DeviceIDRegex)
 
 	errorChan := make(chan error, 1)
 
 	for {
 		err = mqttclient.Subscribe(mqttClientOptions, mqttclient.SubscribeOptions{
-			Topic:             cfg.MQTT.TopicPath + "/+",
+			Topic:             cfg.MQTT.TopicPath,
 			QoS:               cfg.MQTT.QoS,
 			OnMessageReceived: ingest.SetupSubscriptionHandler(errorChan),
 		})

config.yaml.dist

@@ -5,8 +5,13 @@ mqtt:
   # Optional: Username and Password for authenticating with the MQTT Server
   # user: bob
   # password: happylittleclouds
-  # The Topic path to subscripe to. Actually this will become `$topic_path/+`
-  topic_path: v1/devices/me
+  # The Topic path to subscripe to.
+  topic_path: v1/devices/me/+
+  # Optional: Regular expression to extract the device ID from the topic path. The default regular expression, assumes
+  # that the last "element" of the topic_path is the device id.
+  # The regular expression must contain a named capture group with the name deviceid
+  # For example the expression for tasamota based sensors is "tele/(?P<deviceid>.*)/.*"
+  # device_id_regex: "(.*/)?(?P<deviceid>.*)"
   # The MQTT QoS level
   qos: 0
 cache:

examples/gosund_sp111.yaml

@@ -0,0 +1,48 @@
+# Settings for the MQTT Client. Currently only these three are supported
+mqtt:
+  # The MQTT broker to connect to
+  server: tcp://192.168.1.11:1883
+  # Optional: Username and Password for authenticating with the MQTT Server
+  # user: bob
+  # password: happylittleclouds
+  # The Topic path to subscripe to. Actually this will become `$topic_path/+`
+  topic_path: tele/+/SENSOR
+  # Optional: Regular expression to extract the device ID from the topic path. The default regular expression, assumes
+  # that the last "element" of the topic_path is the device id.
+  # The regular expression must contain a named capture group with the name deviceid
+  # For example the expression for tasamota based sensors is "tele/(?P<deviceid>.*)/.*"
+  device_id_regex: "tele/(?P<deviceid>.*)/SENSOR"
+  # The MQTT QoS level
+  qos: 0
+cache:
+  # Timeout. Each received metric will be presented for this time if no update is send via MQTT.
+  # Set the timeout to -1 to disable the deletion of metrics from the cache. The exporter presents the ingest timestamp
+  # to prometheus.
+  timeout: 24h
+# This is a list of valid metrics. Only metrics listed here will be exported
+metrics:
+  # The name of the metric in prometheus
+  - prom_name: consumed_energy_kilowatthours_total
+    mqtt_name: "ENERGY.Total"
+    help: "total measured kilowatthours since flash"
+    type: counter
+  - prom_name: voltage_volts
+    mqtt_name: "ENERGY.Voltage"
+    help: "Currently measured voltage"
+    type: gauge
+  - prom_name: current_amperes
+    mqtt_name: "ENERGY.Current"
+    help: "Currently measured current"
+    type: gauge
+  - prom_name: power_watts
+    mqtt_name: "ENERGY.Power"
+    help: "Currently measured power"
+    type: gauge
+  - prom_name: apparent_power_watt
+    mqtt_name: "ENERGY.ApparentPower"
+    help: "Currently apparent power"
+    type: gauge
+  - prom_name: reactive_power_watt
+    mqtt_name: "ENERGY.ReactivePower"
+    help: "Currently reactive power"
+    type: gauge

go.mod

@@ -12,6 +12,8 @@ require (
 	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/common v0.9.1
 	github.com/prometheus/procfs v0.0.11
+	github.com/thedevsaddam/gojsonq v2.3.0+incompatible // indirect
+	github.com/thedevsaddam/gojsonq/v2 v2.5.2
 	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980
 	gopkg.in/yaml.v2 v2.2.5
 )

go.sum

@@ -82,6 +82,11 @@ github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/thedevsaddam/gojsonq v1.9.1 h1:zQulEP43nwmq5EKrNWyIgJVbqDeMdC1qzXM/f5O15a0=
+github.com/thedevsaddam/gojsonq v2.3.0+incompatible h1:i2lFTvGY4LvoZ2VUzedsFlRiyaWcJm3Uh6cQ9+HyQA8=
+github.com/thedevsaddam/gojsonq v2.3.0+incompatible/go.mod h1:RBcQaITThgJAAYKH7FNp2onYodRz8URfsuEGpAch0NA=
+github.com/thedevsaddam/gojsonq/v2 v2.5.2 h1:CoMVaYyKFsVj6TjU6APqAhAvC07hTI6IQen8PHzHYY0=
+github.com/thedevsaddam/gojsonq/v2 v2.5.2/go.mod h1:bv6Xa7kWy82uT0LnXPE2SzGqTj33TAEeR560MdJkiXs=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=

pkg/config/config.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"fmt"
 	"io/ioutil"
 	"regexp"
 	"time"
@@ -12,22 +13,25 @@ import (
 const GaugeValueType = "gauge"
 const CounterValueType = "counter"
 
+const DeviceIDRegexGroup = "deviceid"
+
 var MQTTConfigDefaults = MQTTConfig{
-	Server:    "tcp://127.0.0.1:1883",
-	TopicPath: "v1/devices/me",
-	QoS:       0,
+	Server:        "tcp://127.0.0.1:1883",
+	TopicPath:     "v1/devices/me",
+	DeviceIDRegex: mustNewRegexp(fmt.Sprintf("(.*/)?(?P<%s>.*)", DeviceIDRegexGroup)),
+	QoS:           0,
 }
 
 var CacheConfigDefaults = CacheConfig{
 	Timeout: 2 * time.Minute,
 }
 
-type RegexpFilter struct {
+type Regexp struct {
 	r       *regexp.Regexp
 	pattern string
 }
 
-func (rf *RegexpFilter) UnmarshalYAML(unmarshal func(interface{}) error) error {
+func (rf *Regexp) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	var pattern string
 	if err := unmarshal(&pattern); err != nil {
 		return err
@@ -41,14 +45,37 @@ func (rf *RegexpFilter) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	return nil
 }
 
-func (rf *RegexpFilter) MarshalYAML() (interface{}, error) {
+func (rf *Regexp) MarshalYAML() (interface{}, error) {
 	return rf.pattern, nil
 }
 
-func (rf *RegexpFilter) Match(s string) bool {
+func (rf *Regexp) Match(s string) bool {
 	return rf.r == nil || rf.r.MatchString(s)
 }
 
+// GroupValue returns the value of the given group. If the group is not part of the underlying regexp, returns the empty string.
+func (rf *Regexp) GroupValue(s string, groupName string) string {
+	match := rf.r.FindStringSubmatch(s)
+	groupValues := make(map[string]string)
+	for i, name := range rf.r.SubexpNames() {
+		if name != "" {
+			groupValues[name] = match[i]
+		}
+	}
+	return groupValues[groupName]
+}
+
+func (rf *Regexp) RegEx() *regexp.Regexp {
+	return rf.r
+}
+
+func mustNewRegexp(pattern string) *Regexp {
+	return &Regexp{
+		pattern: pattern,
+		r:       regexp.MustCompile(pattern),
+	}
+}
+
 type Config struct {
 	Metrics []MetricConfig `yaml:"metrics"`
 	MQTT    *MQTTConfig    `yaml:"mqtt,omitempty"`
@@ -60,18 +87,19 @@ type CacheConfig struct {
 }
 
 type MQTTConfig struct {
-	Server    string `yaml:"server"`
-	TopicPath string `yaml:"topic_path"`
-	User      string `yaml:"user"`
-	Password  string `yaml:"password"`
-	QoS       byte   `yaml:"qos"`
+	Server        string  `yaml:"server"`
+	TopicPath     string  `yaml:"topic_path"`
+	DeviceIDRegex *Regexp `yaml:"device_id_regex"`
+	User          string  `yaml:"user"`
+	Password      string  `yaml:"password"`
+	QoS           byte    `yaml:"qos"`
 }
 
 // Metrics Config is a mapping between a metric send on mqtt to a prometheus metric
 type MetricConfig struct {
 	PrometheusName     string                    `yaml:"prom_name"`
 	MQTTName           string                    `yaml:"mqtt_name"`
-	SensorNameFilter   RegexpFilter              `yaml:"sensor_name_filter"`
+	SensorNameFilter   Regexp                    `yaml:"sensor_name_filter"`
 	Help               string                    `yaml:"help"`
 	ValueType          string                    `yaml:"type"`
 	ConstantLabels     map[string]string         `yaml:"const_labels"`
@@ -87,7 +115,7 @@ type StringValueMappingConfig struct {
 
 func (mc *MetricConfig) PrometheusDescription() *prometheus.Desc {
 	return prometheus.NewDesc(
-		mc.PrometheusName, mc.Help, []string{"sensor"}, mc.ConstantLabels,
+		mc.PrometheusName, mc.Help, []string{"sensor", "topic"}, mc.ConstantLabels,
 	)
 }
 
@@ -117,5 +145,17 @@ func LoadConfig(configFile string) (Config, error) {
 	if cfg.Cache == nil {
 		cfg.Cache = &CacheConfigDefaults
 	}
+	if cfg.MQTT.DeviceIDRegex == nil {
+		cfg.MQTT.DeviceIDRegex = MQTTConfigDefaults.DeviceIDRegex
+	}
+	var validRegex bool
+	for _, name := range cfg.MQTT.DeviceIDRegex.RegEx().SubexpNames() {
+		if name == DeviceIDRegexGroup {
+			validRegex = true
+		}
+	}
+	if !validRegex {
+		return Config{}, fmt.Errorf("device id regex %q does not contain required regex group %q", cfg.MQTT.DeviceIDRegex.pattern, DeviceIDRegexGroup)
+	}
 	return cfg, nil
 }

pkg/metrics/collector.go

@@ -25,6 +25,7 @@ type Metric struct {
 	Value       float64
 	ValueType   prometheus.ValueType
 	IngestTime  time.Time
+	Topic       string
 }
 
 type MetricCollection []Metric
@@ -59,6 +60,7 @@ func (c *MemoryCachedCollector) Collect(mc chan<- prometheus.Metric) {
 				metric.ValueType,
 				metric.Value,
 				device,
+				metric.Topic,
 			)
 			if err != nil {
 				panic(err)