mirror of
https://github.com/hikhvar/mqtt2prometheus.git
synced 2026-02-14 09:59:52 +00:00
Christoph Petrausch
GitHub
65
Readme.md
65
Readme.md
@@ -251,6 +251,32 @@ metrics:
|
||||
# If not specified, parsing error will occur.
|
||||
error_value: 1
|
||||
# The name of the metric in prometheus
|
||||
- prom_name: total_light_usage_seconds
|
||||
# The name of the metric in a MQTT JSON message
|
||||
mqtt_name: state
|
||||
# Regular expression to only match sensors with the given name pattern
|
||||
sensor_name_filter: "^.*-light$"
|
||||
# The prometheus help text for this metric
|
||||
help: Total time the light was on, in seconds
|
||||
# The prometheus type for this metric. Valid values are: "gauge" and "counter"
|
||||
type: counter
|
||||
# according to prometheus exposition format timestamp is not mandatory, we can omit it if the reporting from the sensor is sporadic
|
||||
omit_timestamp: true
|
||||
# A map of string to string for constant labels. This labels will be attached to every prometheus metric
|
||||
const_labels:
|
||||
sensor_type: ikea
|
||||
# When specified, enables mapping between string values to metric values.
|
||||
string_value_mapping:
|
||||
# A map of string to metric value.
|
||||
map:
|
||||
off: 0
|
||||
low: 0
|
||||
# Metric value to use if a match cannot be found in the map above.
|
||||
# If not specified, parsing error will occur.
|
||||
error_value: 1
|
||||
# Sum up the time the light is on, see the section "Expressions" below.
|
||||
expression: "value > 0 ? last_result + elapsed.Seconds() : last_result"
|
||||
# The name of the metric in prometheus
|
||||
- prom_name: total_energy
|
||||
# The name of the metric in a MQTT JSON message
|
||||
mqtt_name: aenergy.total
|
||||
@@ -260,9 +286,8 @@ metrics:
|
||||
help: Total energy used
|
||||
# The prometheus type for this metric. Valid values are: "gauge" and "counter"
|
||||
type: counter
|
||||
# This setting requires an almost monotonic counter as the source. When monotonicy is enforced, the metric value is regularly written to disk. Thus, resets in the source counter can be detected and corrected by adding an offset as if the reset did not happen. The result is a strict monotonic increasing time series, like an ever growing counter.
|
||||
# This setting requires an almost monotonic counter as the source. When monotonicy is enforced, the metric value is regularly written to disk. Thus, resets in the source counter can be detected and corrected by adding an offset as if the reset did not happen. The result is a true monotonic increasing time series, like an ever growing counter.
|
||||
force_monotonicy: true
|
||||
|
||||
```
|
||||
|
||||
### Environment Variables
|
||||
@@ -308,7 +333,43 @@ Create a docker secret to store the password(`mqtt-credential` in the example be
|
||||
- config-tasmota.yml:/config.yaml:ro
|
||||
```
|
||||
|
||||
### Expressions
|
||||
|
||||
Metric values can be derived from sensor inputs using complex expressions. Set the metric config option `expression` to the desired formular to calculate the result from the input. Here's an example which integrates all positive values over time:
|
||||
|
||||
```yaml
|
||||
expression: "value > 0 ? last_result + value * elapsed.Seconds() : last_result"
|
||||
```
|
||||
|
||||
During the evaluation, the following variables are available to the expression:
|
||||
* `value` - the current sensor value (after string-value mapping, if configured)
|
||||
* `last_value` - the `value` during the previous expression evaluation
|
||||
* `last_result` - the result from the previous expression evaluation
|
||||
* `elapsed` - the time that passed since the previous evaluation, as a [Duration](https://pkg.go.dev/time#Duration) value
|
||||
|
||||
The [language definition](https://expr-lang.org/docs/v1.9/Language-Definition) describes the expression syntax. In addition, the following functions are available:
|
||||
* `now()` - the current time as a [Time](https://pkg.go.dev/time#Time) value
|
||||
* `int(x)` - convert `x` to an integer value
|
||||
* `float(x)` - convert `x` to a floating point value
|
||||
* `round(x)` - rounds value `x` to the nearest integer
|
||||
* `ceil(x)` - rounds value `x` up to the next higher integer
|
||||
* `floor(x)` - rounds value `x` down to the next lower integer
|
||||
* `abs(x)` - returns the `x` as a positive number
|
||||
* `min(x, y)` - returns the minimum of `x` and `y`
|
||||
* `max(x, y)` - returns the maximum of `x` and `y`
|
||||
|
||||
[Time](https://pkg.go.dev/time#Time) and [Duration](https://pkg.go.dev/time#Duration) values come with their own methods which can be used in expressions. For example, `elapsed.Milliseconds()` yields the number of milliseconds that passed since the last evaluation, while `now().Sub(elapsed).Weekday()` returns the day of the week during the previous evaluation.
|
||||
|
||||
The `last_value`, `last_result`, and the timestamp of the last evaluation are regularly stored on disk. When mqtt2prometheus is restarted, the data is read back for the next evaluation. This means that you can calculate stable, long-running time serious which depend on the previous result.
|
||||
|
||||
#### Evaluation Order
|
||||
|
||||
It is important to understand the sequence of transformations from a sensor input to the final output which is exported to Prometheus. The steps are as follows:
|
||||
|
||||
1. The sensor input is converted to a number. If a `string_value_mapping` is configured, it is consulted for the conversion.
|
||||
1. If an `expression` is configured, it is evaluated using the converted number. The result of the evaluation replaces the converted sensor value.
|
||||
1. If `force_monotonicy` is set to `true`, any new value that is smaller than the previous one is considered to be a counter reset. When a reset is detected, the previous value becomes the value offset which is automatically added to each consecutive value. The offset is persistet between restarts of mqtt2prometheus.
|
||||
1. If `mqtt_value_scale` is set to a non-zero value, it is applied to the the value to yield the final metric value.
|
||||
|
||||
## Frequently Asked Questions
|
||||
|
||||
|
||||
1
go.mod
1
go.mod
@@ -3,6 +3,7 @@ module github.com/hikhvar/mqtt2prometheus
|
||||
go 1.14
|
||||
|
||||
require (
|
||||
github.com/antonmedv/expr v1.8.9
|
||||
github.com/eclipse/paho.mqtt.golang v1.3.5
|
||||
github.com/go-kit/kit v0.10.0
|
||||
github.com/patrickmn/go-cache v2.1.0+incompatible
|
||||
|
||||
18
go.sum
18
go.sum
@@ -34,6 +34,7 @@ dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7
|
||||
github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
|
||||
github.com/DATA-DOG/go-sqlmock v1.3.3/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
|
||||
github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
|
||||
github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
|
||||
github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
|
||||
@@ -44,6 +45,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
|
||||
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
|
||||
github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
|
||||
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
|
||||
github.com/antonmedv/expr v1.8.9 h1:O9stiHmHHww9b4ozhPx7T6BK7fXfOCHJ8ybxf0833zw=
|
||||
github.com/antonmedv/expr v1.8.9/go.mod h1:5qsM3oLGDND7sDmQGDXHkYfkjYMUX14qsgqmHhwGEk8=
|
||||
github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
|
||||
github.com/apache/thrift v0.13.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
|
||||
github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
|
||||
@@ -76,6 +79,7 @@ github.com/coreos/go-systemd v0.0.0-20180511133405-39ca1b05acc7/go.mod h1:F5haX7
|
||||
github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
|
||||
github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
|
||||
github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
|
||||
github.com/davecgh/go-spew v0.0.0-20161028175848-04cdfd42973b/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
@@ -96,6 +100,8 @@ github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5Kwzbycv
|
||||
github.com/franela/goblin v0.0.0-20200105215937-c9ffbefa60db/go.mod h1:7dvUGVsVBjqR7JHJk0brhHOZYGmfBYOrK0ZhYMEtBr4=
|
||||
github.com/franela/goreq v0.0.0-20171204163338-bcd34c9993f8/go.mod h1:ZhphrRTfi2rbfLwlschooIH4+wKKDR4Pdxhh+TRoA20=
|
||||
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
|
||||
github.com/gdamore/encoding v1.0.0/go.mod h1:alR0ol34c49FCSBLjhosxzcPHQbf2trDkoo5dl+VrEg=
|
||||
github.com/gdamore/tcell v1.3.0/go.mod h1:Hjvr+Ofd+gLglo7RYKxxnzCBmev3BzsS67MebKS4zMM=
|
||||
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
|
||||
github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
@@ -232,11 +238,15 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/lightstep/lightstep-tracer-common/golang/gogo v0.0.0-20190605223551-bc2310a04743/go.mod h1:qklhhLq1aX+mtWk9cPHPzaBjWImj5ULL6C7HFJtXQMM=
|
||||
github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4=
|
||||
github.com/lucasb-eyer/go-colorful v1.0.2/go.mod h1:0MS4r+7BZKSJ5mw4/S5MPN+qHFF1fYclkSPilDOKW0s=
|
||||
github.com/lucasb-eyer/go-colorful v1.0.3/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
|
||||
github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ=
|
||||
github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
|
||||
github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
|
||||
github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
|
||||
github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
|
||||
github.com/mattn/go-runewidth v0.0.4/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
|
||||
github.com/mattn/go-runewidth v0.0.8/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
|
||||
github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
|
||||
@@ -289,6 +299,7 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
|
||||
github.com/pmezard/go-difflib v0.0.0-20151028094244-d8ed2627bdf0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
|
||||
@@ -324,11 +335,14 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
|
||||
github.com/prometheus/procfs v0.6.0 h1:mxy4L2jP6qMonqmq+aTtOx1ifVWUgG/TAmntgbh3xv4=
|
||||
github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
|
||||
github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
|
||||
github.com/rivo/tview v0.0.0-20200219210816-cd38d7432498/go.mod h1:6lkG1x+13OShEf0EaOCaTQYyB7d5nSbb181KtjlS+84=
|
||||
github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
|
||||
github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
|
||||
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
|
||||
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
|
||||
github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
|
||||
github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
|
||||
github.com/sanity-io/litter v1.2.0/go.mod h1:JF6pZUFgu2Q0sBZ+HSV35P8TVPI1TTzEwyu9FXAw2W4=
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
|
||||
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
|
||||
github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
|
||||
@@ -345,10 +359,13 @@ github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3
|
||||
github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v0.0.0-20161117074351-18a02ba4a312/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
|
||||
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 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
|
||||
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
|
||||
github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
|
||||
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
|
||||
github.com/thedevsaddam/gojsonq/v2 v2.5.2 h1:CoMVaYyKFsVj6TjU6APqAhAvC07hTI6IQen8PHzHYY0=
|
||||
github.com/thedevsaddam/gojsonq/v2 v2.5.2/go.mod h1:bv6Xa7kWy82uT0LnXPE2SzGqTj33TAEeR560MdJkiXs=
|
||||
github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
|
||||
@@ -493,6 +510,7 @@ golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7w
|
||||
golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190626150813-e07cf5db2756/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
|
||||
@@ -138,6 +138,7 @@ type MetricConfig struct {
|
||||
Help string `yaml:"help"`
|
||||
ValueType string `yaml:"type"`
|
||||
OmitTimestamp bool `yaml:"omit_timestamp"`
|
||||
Expression string `yaml:"expression"`
|
||||
ForceMonotonicy bool `yaml:"force_monotonicy"`
|
||||
ConstantLabels map[string]string `yaml:"const_labels"`
|
||||
StringValueMapping *StringValueMappingConfig `yaml:"string_value_mapping"`
|
||||
@@ -233,7 +234,7 @@ func LoadConfig(configFile string) (Config, error) {
|
||||
}
|
||||
if forcesMonotonicy {
|
||||
if err := os.MkdirAll(cfg.Cache.StateDir, 0755); err != nil {
|
||||
return Config{}, err
|
||||
return Config{}, fmt.Errorf("failed to create directory %q: %w", cfg.Cache.StateDir, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -32,19 +32,16 @@ func NewJSONObjectExtractor(p Parser) Extractor {
|
||||
continue
|
||||
}
|
||||
|
||||
// Find a valid metrics config
|
||||
config, found := p.findMetricConfig(path, deviceID)
|
||||
if !found {
|
||||
continue
|
||||
// Find all valid metric configs
|
||||
for _, config := range p.findMetricConfigs(path, deviceID) {
|
||||
id := metricID(topic, path, deviceID, config.PrometheusName)
|
||||
m, err := p.parseMetric(config, id, rawValue)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse valid value from '%v' for metric %q: %w", rawValue, config.PrometheusName, err)
|
||||
}
|
||||
m.Topic = topic
|
||||
mc = append(mc, m)
|
||||
}
|
||||
|
||||
id := metricID(topic, path, deviceID, config.PrometheusName)
|
||||
m, err := p.parseMetric(config, id, rawValue)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse valid metric value: %w", err)
|
||||
}
|
||||
m.Topic = topic
|
||||
mc = append(mc, m)
|
||||
}
|
||||
return mc, nil
|
||||
}
|
||||
@@ -52,35 +49,34 @@ func NewJSONObjectExtractor(p Parser) Extractor {
|
||||
|
||||
func NewMetricPerTopicExtractor(p Parser, metricNameRegex *config.Regexp) Extractor {
|
||||
return func(topic string, payload []byte, deviceID string) (MetricCollection, error) {
|
||||
var mc MetricCollection
|
||||
metricName := metricNameRegex.GroupValue(topic, config.MetricNameRegexGroup)
|
||||
if metricName == "" {
|
||||
return nil, fmt.Errorf("failed to find valid metric in topic path")
|
||||
}
|
||||
|
||||
// Find a valid metrics config
|
||||
config, found := p.findMetricConfig(metricName, deviceID)
|
||||
if !found {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var rawValue interface{}
|
||||
if config.PayloadField != "" {
|
||||
parsed := gojsonq.New(gojsonq.SetSeparator(p.separator)).FromString(string(payload))
|
||||
rawValue = parsed.Find(config.PayloadField)
|
||||
parsed.Reset()
|
||||
if rawValue == nil {
|
||||
return nil, fmt.Errorf("failed to extract field %s from payload %s", config.PayloadField, payload)
|
||||
// Find all valid metric configs
|
||||
for _, config := range p.findMetricConfigs(metricName, deviceID) {
|
||||
var rawValue interface{}
|
||||
if config.PayloadField != "" {
|
||||
parsed := gojsonq.New(gojsonq.SetSeparator(p.separator)).FromString(string(payload))
|
||||
rawValue = parsed.Find(config.PayloadField)
|
||||
parsed.Reset()
|
||||
if rawValue == nil {
|
||||
return nil, fmt.Errorf("failed to extract field %q from payload %q for metric %q", config.PayloadField, payload, metricName)
|
||||
}
|
||||
} else {
|
||||
rawValue = string(payload)
|
||||
}
|
||||
} else {
|
||||
rawValue = string(payload)
|
||||
}
|
||||
|
||||
id := metricID(topic, metricName, deviceID, config.PrometheusName)
|
||||
m, err := p.parseMetric(config, id, rawValue)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse metric: %w", err)
|
||||
id := metricID(topic, metricName, deviceID, config.PrometheusName)
|
||||
m, err := p.parseMetric(config, id, rawValue)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to parse valid value from '%v' for metric %q: %w", rawValue, config.PrometheusName, err)
|
||||
}
|
||||
m.Topic = topic
|
||||
mc = append(mc, m)
|
||||
}
|
||||
m.Topic = topic
|
||||
return MetricCollection{m}, nil
|
||||
return mc, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,7 +11,7 @@ import (
|
||||
func TestNewJSONObjectExtractor_parseMetric(t *testing.T) {
|
||||
now = testNow
|
||||
type fields struct {
|
||||
metricConfigs map[string][]config.MetricConfig
|
||||
metricConfigs map[string][]*config.MetricConfig
|
||||
}
|
||||
type args struct {
|
||||
metricPath string
|
||||
@@ -31,8 +31,8 @@ func TestNewJSONObjectExtractor_parseMetric(t *testing.T) {
|
||||
name: "string value",
|
||||
separator: "->",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"SDS0X1->PM2->5": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"SDS0X1->PM2->5": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
MQTTName: "SDS0X1.PM2.5",
|
||||
@@ -57,8 +57,8 @@ func TestNewJSONObjectExtractor_parseMetric(t *testing.T) {
|
||||
name: "string value with dots in path",
|
||||
separator: "->",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"SDS0X1->PM2.5": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"SDS0X1->PM2.5": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
MQTTName: "SDS0X1->PM2.5",
|
||||
@@ -83,8 +83,8 @@ func TestNewJSONObjectExtractor_parseMetric(t *testing.T) {
|
||||
name: "metric matching SensorNameFilter",
|
||||
separator: ".",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"temperature": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"temperature": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
MQTTName: "temperature",
|
||||
@@ -110,8 +110,8 @@ func TestNewJSONObjectExtractor_parseMetric(t *testing.T) {
|
||||
name: "metric not matching SensorNameFilter",
|
||||
separator: ".",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"temperature": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"temperature": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
MQTTName: "temperature",
|
||||
|
||||
@@ -3,48 +3,137 @@ package metrics
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/antonmedv/expr"
|
||||
"github.com/antonmedv/expr/vm"
|
||||
"github.com/hikhvar/mqtt2prometheus/pkg/config"
|
||||
"gopkg.in/yaml.v2"
|
||||
)
|
||||
|
||||
// monotonicState holds the runtime information to realize a monotonic increasing value.
|
||||
type monotonicState struct {
|
||||
// dynamicState holds the runtime information for dynamic metric configs.
|
||||
type dynamicState struct {
|
||||
// Basline value to add to each parsed metric value to maintain monotonicy
|
||||
Offset float64 `yaml:"value_offset"`
|
||||
// Last value that was parsed before the offset was added
|
||||
LastRawValue float64 `yaml:"last_raw_value"`
|
||||
// Last value that was used for evaluating the given expression
|
||||
LastExprValue float64 `yaml:"last_expr_value"`
|
||||
// Last result returned from evaluating the given expression
|
||||
LastExprResult float64 `yaml:"last_expr_result"`
|
||||
// Last result returned from evaluating the given expression
|
||||
LastExprTimestamp time.Time `yaml:"last_expr_timestamp"`
|
||||
}
|
||||
|
||||
// metricState holds runtime information per metric configuration.
|
||||
type metricState struct {
|
||||
monotonic monotonicState
|
||||
dynamic dynamicState
|
||||
// The last time the state file was written
|
||||
lastWritten time.Time
|
||||
// Compiled evaluation expression
|
||||
program *vm.Program
|
||||
// Environment in which the expression is evaluated
|
||||
env map[string]interface{}
|
||||
}
|
||||
|
||||
type Parser struct {
|
||||
separator string
|
||||
// Maps the mqtt metric name to a list of configs
|
||||
// The first that matches SensorNameFilter will be used
|
||||
metricConfigs map[string][]config.MetricConfig
|
||||
metricConfigs map[string][]*config.MetricConfig
|
||||
// Directory holding state files
|
||||
stateDir string
|
||||
// Per-metric state
|
||||
states map[string]*metricState
|
||||
}
|
||||
|
||||
// Identifiers within the expression evaluation environment.
|
||||
const (
|
||||
env_value = "value"
|
||||
env_last_value = "last_value"
|
||||
env_last_result = "last_result"
|
||||
env_elapsed = "elapsed"
|
||||
env_now = "now"
|
||||
env_int = "int"
|
||||
env_float = "float"
|
||||
env_round = "round"
|
||||
env_ceil = "ceil"
|
||||
env_floor = "floor"
|
||||
env_abs = "abs"
|
||||
env_min = "min"
|
||||
env_max = "max"
|
||||
)
|
||||
|
||||
var now = time.Now
|
||||
|
||||
func toInt64(i interface{}) int64 {
|
||||
switch v := i.(type) {
|
||||
case float32:
|
||||
return int64(v)
|
||||
case float64:
|
||||
return int64(v)
|
||||
case int:
|
||||
return int64(v)
|
||||
case int32:
|
||||
return int64(v)
|
||||
case int64:
|
||||
return v
|
||||
case time.Duration:
|
||||
return int64(v)
|
||||
default:
|
||||
return v.(int64) // Hope for the best
|
||||
}
|
||||
}
|
||||
|
||||
func toFloat64(i interface{}) float64 {
|
||||
switch v := i.(type) {
|
||||
case float32:
|
||||
return float64(v)
|
||||
case float64:
|
||||
return v
|
||||
case int:
|
||||
return float64(v)
|
||||
case int32:
|
||||
return float64(v)
|
||||
case int64:
|
||||
return float64(v)
|
||||
case time.Duration:
|
||||
return float64(v)
|
||||
default:
|
||||
return v.(float64) // Hope for the best
|
||||
}
|
||||
}
|
||||
|
||||
// defaultExprEnv returns the default environment for expression evaluation.
|
||||
func defaultExprEnv() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
// Variables
|
||||
env_value: 0.0,
|
||||
env_last_value: 0.0,
|
||||
env_last_result: 0.0,
|
||||
env_elapsed: time.Duration(0),
|
||||
// Functions
|
||||
env_now: now,
|
||||
env_int: toInt64,
|
||||
env_float: toFloat64,
|
||||
env_round: math.Round,
|
||||
env_ceil: math.Ceil,
|
||||
env_floor: math.Floor,
|
||||
env_abs: math.Abs,
|
||||
env_min: math.Min,
|
||||
env_max: math.Max,
|
||||
}
|
||||
}
|
||||
|
||||
func NewParser(metrics []config.MetricConfig, separator, stateDir string) Parser {
|
||||
cfgs := make(map[string][]config.MetricConfig)
|
||||
cfgs := make(map[string][]*config.MetricConfig)
|
||||
for i := range metrics {
|
||||
key := metrics[i].MQTTName
|
||||
cfgs[key] = append(cfgs[key], metrics[i])
|
||||
cfgs[key] = append(cfgs[key], &metrics[i])
|
||||
}
|
||||
return Parser{
|
||||
separator: separator,
|
||||
@@ -55,25 +144,26 @@ func NewParser(metrics []config.MetricConfig, separator, stateDir string) Parser
|
||||
}
|
||||
|
||||
// Config returns the underlying metrics config
|
||||
func (p *Parser) config() map[string][]config.MetricConfig {
|
||||
func (p *Parser) config() map[string][]*config.MetricConfig {
|
||||
return p.metricConfigs
|
||||
}
|
||||
|
||||
// validMetric returns config matching the metric and deviceID
|
||||
// Second return value indicates if config was found.
|
||||
func (p *Parser) findMetricConfig(metric string, deviceID string) (config.MetricConfig, bool) {
|
||||
// validMetric returns all configs matching the metric and deviceID.
|
||||
func (p *Parser) findMetricConfigs(metric string, deviceID string) []*config.MetricConfig {
|
||||
configs := []*config.MetricConfig{}
|
||||
for _, c := range p.metricConfigs[metric] {
|
||||
if c.SensorNameFilter.Match(deviceID) {
|
||||
return c, true
|
||||
configs = append(configs, c)
|
||||
}
|
||||
}
|
||||
return config.MetricConfig{}, false
|
||||
return configs
|
||||
}
|
||||
|
||||
// parseMetric parses the given value according to the given deviceID and metricPath. The config allows to
|
||||
// parse a metric value according to the device ID.
|
||||
func (p *Parser) parseMetric(cfg config.MetricConfig, metricID string, value interface{}) (Metric, error) {
|
||||
func (p *Parser) parseMetric(cfg *config.MetricConfig, metricID string, value interface{}) (Metric, error) {
|
||||
var metricValue float64
|
||||
var err error
|
||||
|
||||
if boolValue, ok := value.(bool); ok {
|
||||
if boolValue {
|
||||
@@ -100,7 +190,7 @@ func (p *Parser) parseMetric(cfg config.MetricConfig, metricID string, value int
|
||||
// otherwise try to parse float
|
||||
floatValue, err := strconv.ParseFloat(strValue, 64)
|
||||
if err != nil {
|
||||
return Metric{}, fmt.Errorf("got data with unexpectd type: %T ('%s') and failed to parse to float", value, value)
|
||||
return Metric{}, fmt.Errorf("got data with unexpectd type: %T ('%v') and failed to parse to float", value, value)
|
||||
}
|
||||
metricValue = floatValue
|
||||
|
||||
@@ -109,23 +199,19 @@ func (p *Parser) parseMetric(cfg config.MetricConfig, metricID string, value int
|
||||
} else if floatValue, ok := value.(float64); ok {
|
||||
metricValue = floatValue
|
||||
} else {
|
||||
return Metric{}, fmt.Errorf("got data with unexpectd type: %T ('%s')", value, value)
|
||||
return Metric{}, fmt.Errorf("got data with unexpectd type: %T ('%v')", value, value)
|
||||
}
|
||||
|
||||
if cfg.Expression != "" {
|
||||
if metricValue, err = p.evalExpression(metricID, cfg.Expression, metricValue); err != nil {
|
||||
return Metric{}, err
|
||||
}
|
||||
}
|
||||
|
||||
if cfg.ForceMonotonicy {
|
||||
ms, err := p.getMetricState(metricID)
|
||||
if err != nil {
|
||||
if metricValue, err = p.enforceMonotonicy(metricID, metricValue); err != nil {
|
||||
return Metric{}, err
|
||||
}
|
||||
// When the source metric is reset, the last adjusted value becomes the new offset.
|
||||
if metricValue < ms.monotonic.LastRawValue {
|
||||
ms.monotonic.Offset += ms.monotonic.LastRawValue
|
||||
// Trigger flushing the new state to disk.
|
||||
ms.lastWritten = time.Time{}
|
||||
}
|
||||
|
||||
ms.monotonic.LastRawValue = metricValue
|
||||
metricValue += ms.monotonic.Offset
|
||||
}
|
||||
|
||||
if cfg.MQTTValueScale != 0 {
|
||||
@@ -159,7 +245,7 @@ func (p *Parser) readMetricState(metricID string) (*metricState, error) {
|
||||
if os.IsNotExist(err) {
|
||||
return state, nil
|
||||
}
|
||||
return state, err
|
||||
return state, fmt.Errorf("failed to read file %q: %v", f.Name(), err)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
@@ -171,14 +257,14 @@ func (p *Parser) readMetricState(metricID string) (*metricState, error) {
|
||||
return state, err
|
||||
}
|
||||
|
||||
err = yaml.UnmarshalStrict(data, &state.monotonic)
|
||||
err = yaml.UnmarshalStrict(data, &state.dynamic)
|
||||
state.lastWritten = now()
|
||||
return state, err
|
||||
}
|
||||
|
||||
// writeMetricState writes back the metric's current state to the configured path.
|
||||
func (p *Parser) writeMetricState(metricID string, state *metricState) error {
|
||||
out, err := yaml.Marshal(state.monotonic)
|
||||
out, err := yaml.Marshal(state.dynamic)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -186,9 +272,11 @@ func (p *Parser) writeMetricState(metricID string, state *metricState) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = f.Write(out)
|
||||
f.Close()
|
||||
return err
|
||||
defer f.Close()
|
||||
if _, err = f.Write(out); err != nil {
|
||||
return fmt.Errorf("failed to write file %q: %v", f.Name(), err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// getMetricState returns the state of the given metric.
|
||||
@@ -210,3 +298,63 @@ func (p *Parser) getMetricState(metricID string) (*metricState, error) {
|
||||
}
|
||||
return state, err
|
||||
}
|
||||
|
||||
// enforceMonotonicy makes sure the given values never decrease from one call to the next.
|
||||
// If the current value is smaller than the last one, a consistent offset is added.
|
||||
func (p *Parser) enforceMonotonicy(metricID string, value float64) (float64, error) {
|
||||
ms, err := p.getMetricState(metricID)
|
||||
if err != nil {
|
||||
return value, err
|
||||
}
|
||||
// When the source metric is reset, the last adjusted value becomes the new offset.
|
||||
if value < ms.dynamic.LastRawValue {
|
||||
ms.dynamic.Offset += ms.dynamic.LastRawValue
|
||||
// Trigger flushing the new state to disk.
|
||||
ms.lastWritten = time.Time{}
|
||||
}
|
||||
|
||||
ms.dynamic.LastRawValue = value
|
||||
return value + ms.dynamic.Offset, nil
|
||||
}
|
||||
|
||||
// evalExpression runs the given code in the metric's environment and returns the result.
|
||||
// In case of an error, the original value is returned.
|
||||
func (p *Parser) evalExpression(metricID, code string, value float64) (float64, error) {
|
||||
ms, err := p.getMetricState(metricID)
|
||||
if err != nil {
|
||||
return value, err
|
||||
}
|
||||
if ms.program == nil {
|
||||
ms.env = defaultExprEnv()
|
||||
ms.program, err = expr.Compile(code, expr.Env(ms.env), expr.AsFloat64())
|
||||
if err != nil {
|
||||
return value, fmt.Errorf("failed to compile expression %q: %w", code, err)
|
||||
}
|
||||
// Trigger flushing the new state to disk.
|
||||
ms.lastWritten = time.Time{}
|
||||
}
|
||||
|
||||
// Update the environment
|
||||
ms.env[env_value] = value
|
||||
ms.env[env_last_value] = ms.dynamic.LastExprValue
|
||||
ms.env[env_last_result] = ms.dynamic.LastExprResult
|
||||
if ms.dynamic.LastExprTimestamp.IsZero() {
|
||||
ms.env[env_elapsed] = time.Duration(0)
|
||||
} else {
|
||||
ms.env[env_elapsed] = now().Sub(ms.dynamic.LastExprTimestamp)
|
||||
}
|
||||
|
||||
result, err := expr.Run(ms.program, ms.env)
|
||||
if err != nil {
|
||||
return value, fmt.Errorf("failed to evaluate expression %q: %w", code, err)
|
||||
}
|
||||
// Type was statically checked above.
|
||||
ret := result.(float64)
|
||||
|
||||
// Update the dynamic state
|
||||
ms.dynamic.LastExprResult = ret
|
||||
ms.dynamic.LastExprValue = value
|
||||
ms.dynamic.LastExprTimestamp = now()
|
||||
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
@@ -10,6 +10,19 @@ import (
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
var testNowElapsed time.Duration
|
||||
|
||||
func testNow() time.Time {
|
||||
now, err := time.Parse(
|
||||
time.RFC3339,
|
||||
"2020-11-01T22:08:41+00:00")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
now = now.Add(testNowElapsed)
|
||||
return now
|
||||
}
|
||||
|
||||
func TestParser_parseMetric(t *testing.T) {
|
||||
stateDir, err := os.MkdirTemp("", "parser_test")
|
||||
if err != nil {
|
||||
@@ -19,7 +32,7 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
|
||||
now = testNow
|
||||
type fields struct {
|
||||
metricConfigs map[string][]config.MetricConfig
|
||||
metricConfigs map[string][]*config.MetricConfig
|
||||
}
|
||||
type args struct {
|
||||
metricPath string
|
||||
@@ -27,17 +40,18 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
value interface{}
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
fields fields
|
||||
args args
|
||||
want Metric
|
||||
wantErr bool
|
||||
name string
|
||||
fields fields
|
||||
args args
|
||||
want Metric
|
||||
wantErr bool
|
||||
elapseNow time.Duration
|
||||
}{
|
||||
{
|
||||
name: "value without timestamp",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"temperature": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"temperature": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
ValueType: "gauge",
|
||||
@@ -62,8 +76,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "string value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"temperature": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"temperature": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
ValueType: "gauge",
|
||||
@@ -87,8 +101,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "scaled string value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"temperature": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"temperature": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
ValueType: "gauge",
|
||||
@@ -113,8 +127,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "string value failure",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"temperature": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"temperature": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
ValueType: "gauge",
|
||||
@@ -132,8 +146,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "float value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"temperature": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"temperature": {
|
||||
{
|
||||
PrometheusName: "temperature",
|
||||
ValueType: "gauge",
|
||||
@@ -157,8 +171,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "scaled float value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"humidity": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"humidity": {
|
||||
{
|
||||
PrometheusName: "humidity",
|
||||
ValueType: "gauge",
|
||||
@@ -183,8 +197,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "negative scaled float value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"humidity": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"humidity": {
|
||||
{
|
||||
PrometheusName: "humidity",
|
||||
ValueType: "gauge",
|
||||
@@ -209,8 +223,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "bool value true",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -234,8 +248,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "scaled bool value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -260,8 +274,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "bool value false",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -285,8 +299,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "string mapping value success",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -316,8 +330,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "string mapping value failure default to error value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -348,8 +362,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "string mapping value failure no error value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -373,8 +387,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "metric not configured",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -399,8 +413,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "unexpected type",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"enabled": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"enabled": {
|
||||
{
|
||||
PrometheusName: "enabled",
|
||||
ValueType: "gauge",
|
||||
@@ -425,8 +439,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "monotonic gauge, step 1: initial value",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"aenergy.total": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"aenergy.total": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
@@ -450,8 +464,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "monotonic gauge, step 2: monotonic increase does not add offset",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"aenergy.total": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"aenergy.total": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
@@ -475,8 +489,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "monotonic gauge, step 3: raw metric is reset, last value becomes the new offset",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"aenergy.total": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"aenergy.total": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
@@ -500,8 +514,8 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
{
|
||||
name: "monotonic gauge, step 4: monotonic increase with offset",
|
||||
fields: fields{
|
||||
map[string][]config.MetricConfig{
|
||||
"aenergy.total": []config.MetricConfig{
|
||||
map[string][]*config.MetricConfig{
|
||||
"aenergy.total": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
@@ -522,20 +536,127 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
Value: 3.0,
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "integrate positive values using expressions, step 1",
|
||||
fields: fields{
|
||||
map[string][]*config.MetricConfig{
|
||||
"apower": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
OmitTimestamp: true,
|
||||
Expression: "value > 0 ? last_result + value * elapsed.Hours() : last_result",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
args: args{
|
||||
metricPath: "apower",
|
||||
deviceID: "shellyplus1pm-foo",
|
||||
value: 60.0,
|
||||
},
|
||||
want: Metric{
|
||||
Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil),
|
||||
ValueType: prometheus.GaugeValue,
|
||||
Value: 0.0, // No elapsed time yet, hence no integration
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "integrate positive values using expressions, step 2",
|
||||
fields: fields{
|
||||
map[string][]*config.MetricConfig{
|
||||
"apower": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
OmitTimestamp: true,
|
||||
Expression: "value > 0 ? last_result + value * elapsed.Hours() : last_result",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
elapseNow: time.Minute,
|
||||
args: args{
|
||||
metricPath: "apower",
|
||||
deviceID: "shellyplus1pm-foo",
|
||||
value: 60.0,
|
||||
},
|
||||
want: Metric{
|
||||
Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil),
|
||||
ValueType: prometheus.GaugeValue,
|
||||
Value: 1.0, // 60 watts for 1 minute = 1 Wh
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "integrate positive values using expressions, step 3",
|
||||
fields: fields{
|
||||
map[string][]*config.MetricConfig{
|
||||
"apower": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
OmitTimestamp: true,
|
||||
Expression: "value > 0 ? last_result + value * elapsed.Hours() : last_result",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
elapseNow: 2 * time.Minute,
|
||||
args: args{
|
||||
metricPath: "apower",
|
||||
deviceID: "shellyplus1pm-foo",
|
||||
value: -60.0,
|
||||
},
|
||||
want: Metric{
|
||||
Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil),
|
||||
ValueType: prometheus.GaugeValue,
|
||||
Value: 1.0, // negative input is ignored
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "integrate positive values using expressions, step 4",
|
||||
fields: fields{
|
||||
map[string][]*config.MetricConfig{
|
||||
"apower": {
|
||||
{
|
||||
PrometheusName: "total_energy",
|
||||
ValueType: "gauge",
|
||||
OmitTimestamp: true,
|
||||
Expression: "value > 0 ? last_result + value * elapsed.Hours() : last_result",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
elapseNow: 3 * time.Minute,
|
||||
args: args{
|
||||
metricPath: "apower",
|
||||
deviceID: "shellyplus1pm-foo",
|
||||
value: 600.0,
|
||||
},
|
||||
want: Metric{
|
||||
Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil),
|
||||
ValueType: prometheus.GaugeValue,
|
||||
Value: 11.0, // 600 watts for 1 minute = 10 Wh
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
testNowElapsed = tt.elapseNow
|
||||
defer func() { testNowElapsed = time.Duration(0) }()
|
||||
|
||||
p := NewParser(nil, config.JsonParsingConfigDefaults.Separator, stateDir)
|
||||
p.metricConfigs = tt.fields.metricConfigs
|
||||
|
||||
// Find a valid metrics config
|
||||
config, found := p.findMetricConfig(tt.args.metricPath, tt.args.deviceID)
|
||||
if !found {
|
||||
configs := p.findMetricConfigs(tt.args.metricPath, tt.args.deviceID)
|
||||
if len(configs) != 1 {
|
||||
if !tt.wantErr {
|
||||
t.Errorf("MetricConfig not found")
|
||||
}
|
||||
return
|
||||
}
|
||||
config := configs[0]
|
||||
|
||||
id := metricID("", tt.args.metricPath, tt.args.deviceID, config.PrometheusName)
|
||||
got, err := p.parseMetric(config, id, tt.args.value)
|
||||
@@ -547,7 +668,7 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
t.Errorf("parseMetric() got = %v, want %v", got, tt.want)
|
||||
}
|
||||
|
||||
if config.ForceMonotonicy {
|
||||
if config.ForceMonotonicy || config.Expression != "" {
|
||||
if err = p.writeMetricState(id, p.states[id]); err != nil {
|
||||
t.Errorf("failed to write metric state: %v", err)
|
||||
}
|
||||
@@ -556,16 +677,109 @@ func TestParser_parseMetric(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func testNow() time.Time {
|
||||
now, err := time.Parse(
|
||||
time.RFC3339,
|
||||
"2020-11-01T22:08:41+00:00")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return now
|
||||
}
|
||||
|
||||
func floatP(f float64) *float64 {
|
||||
return &f
|
||||
}
|
||||
|
||||
func TestParser_evalExpression(t *testing.T) {
|
||||
now = testNow
|
||||
testNowElapsed = time.Duration(0)
|
||||
id := "metric"
|
||||
|
||||
tests := []struct {
|
||||
expression string
|
||||
values []float64
|
||||
results []float64
|
||||
}{
|
||||
{
|
||||
expression: "value + value",
|
||||
values: []float64{1, 0, -4},
|
||||
results: []float64{2, 0, -8},
|
||||
},
|
||||
{
|
||||
expression: "value - last_value",
|
||||
values: []float64{1, 2, 5, 7},
|
||||
results: []float64{1, 1, 3, 2},
|
||||
},
|
||||
{
|
||||
expression: "last_result + value",
|
||||
values: []float64{1, 2, 3, 4},
|
||||
results: []float64{1, 3, 6, 10},
|
||||
},
|
||||
{
|
||||
expression: "last_result + elapsed.Milliseconds()",
|
||||
values: []float64{0, 0, 0, 0},
|
||||
results: []float64{0, 1000, 2000, 3000},
|
||||
},
|
||||
{
|
||||
expression: "now().Unix()",
|
||||
values: []float64{0, 0},
|
||||
results: []float64{float64(testNow().Unix()), float64(testNow().Unix() + 1)},
|
||||
},
|
||||
{
|
||||
expression: "int(1.1) + int(1.9)",
|
||||
values: []float64{0},
|
||||
results: []float64{2},
|
||||
},
|
||||
{
|
||||
expression: "float(elapsed)",
|
||||
values: []float64{0, 0},
|
||||
results: []float64{0, float64(time.Second)},
|
||||
},
|
||||
{
|
||||
expression: "round(value)",
|
||||
values: []float64{1.1, 2.5, 3.9},
|
||||
results: []float64{1, 3, 4},
|
||||
},
|
||||
{
|
||||
expression: "ceil(value)",
|
||||
values: []float64{1.1, 2.9, 4.0},
|
||||
results: []float64{2, 3, 4},
|
||||
},
|
||||
{
|
||||
expression: "floor(value)",
|
||||
values: []float64{1.1, 2.9, 4.0},
|
||||
results: []float64{1, 2, 4},
|
||||
},
|
||||
{
|
||||
expression: "abs(value)",
|
||||
values: []float64{0, 1, -2},
|
||||
results: []float64{0, 1, 2},
|
||||
},
|
||||
{
|
||||
expression: "min(value, 0)",
|
||||
values: []float64{1, -2, 3, -4},
|
||||
results: []float64{0, -2, 0, -4},
|
||||
},
|
||||
{
|
||||
expression: "max(value, 0)",
|
||||
values: []float64{1, -2, 3, -4},
|
||||
results: []float64{1, 0, 3, 0},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.expression, func(t *testing.T) {
|
||||
stateDir, err := os.MkdirTemp("", "parser_test")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer os.RemoveAll(stateDir)
|
||||
defer func() { testNowElapsed = time.Duration(0) }()
|
||||
|
||||
p := NewParser(nil, ".", stateDir)
|
||||
for i, value := range tt.values {
|
||||
got, err := p.evalExpression(id, tt.expression, value)
|
||||
want := tt.results[i]
|
||||
if err != nil {
|
||||
t.Errorf("evaluating the %dth value '%v' failed: %v", i, value, err)
|
||||
}
|
||||
if got != want {
|
||||
t.Errorf("unexpected result for %dth value, got %v, want %v", i, got, want)
|
||||
}
|
||||
// Advance the clock by one second for every sample
|
||||
testNowElapsed = testNowElapsed + time.Second
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user