From 6d695bdbf4e8ddbc02680bcfc38299f6afac9db3 Mon Sep 17 00:00:00 2001 From: Alexander Bakker Date: Thu, 13 Apr 2023 19:41:42 +0200 Subject: [PATCH 01/12] Rename received_messages metric to be more conventional This patch renames the ``received_messages`` metric to ``mqtt2prometheus_received_messages_total``, making it a bit more in line with conventional Prometheus metric naming. I also slightly adjusted the descriptions. --- pkg/metrics/instrumentation.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/metrics/instrumentation.go b/pkg/metrics/instrumentation.go index 5dd75eb..6f64ff8 100644 --- a/pkg/metrics/instrumentation.go +++ b/pkg/metrics/instrumentation.go @@ -13,14 +13,14 @@ const ( var defaultInstrumentation = instrumentation{ messageMetric: prometheus.NewCounterVec( prometheus.CounterOpts{ - Name: "received_messages", - Help: "received messages per topic and status", + Name: "mqtt2prometheus_received_messages_total", + Help: "Total number of messages received per topic and status", }, []string{"status", "topic"}, ), connectedMetric: prometheus.NewGauge( prometheus.GaugeOpts{ Name: "mqtt2prometheus_connected", - Help: "is the mqtt2prometheus exporter connected to the broker", + Help: "Whether the mqtt2prometheus exporter is connected to the broker", }, ), } From d562b927f8a097e24ea21141786bbf4dcec672ba Mon Sep 17 00:00:00 2001 From: dmolle <35704791+dmolle@users.noreply.github.com> Date: Sat, 1 Apr 2023 01:32:24 +0200 Subject: [PATCH 02/12] Update Readme.md JSON needs to be upper case, otherwiese "could not setup a metric extractor {"error": "unsupported object format: json"} " is thrown. --- Readme.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Readme.md b/Readme.md index 8c19f20..c38348a 100644 --- a/Readme.md +++ b/Readme.md @@ -170,7 +170,7 @@ mqtt: # This is the default. object_per_topic_config: # The encoding of the object, currently only json is supported - encoding: json + encoding: JSON 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 From 9da7b1b3794931f5d592feb5fead878889910f57 Mon Sep 17 00:00:00 2001 From: yan Date: Wed, 19 Apr 2023 01:06:13 +0200 Subject: [PATCH 03/12] Handle json payload in metric_per_topic mode --- Readme.md | 6 ++++++ hack/shellyplusht.yaml | 22 ++++++++++++++++++++++ pkg/config/config.go | 1 + pkg/metrics/extractor.go | 14 +++++++++++++- 4 files changed, 42 insertions(+), 1 deletion(-) create mode 100644 hack/shellyplusht.yaml diff --git a/Readme.md b/Readme.md index c38348a..99f7d09 100644 --- a/Readme.md +++ b/Readme.md @@ -71,6 +71,12 @@ E.g let's assume the following MQTT JSON message: We can now set `json_parsing.seperator` to `/`. This allows us to specify `mqtt_name` as `computed/heat.index`. Keep in mind, `json_parsing.seperator` is a global setting. This affects all `mqtt_name` fields in your configuration. +Some devices like Shelly Plus H&T publish one metric per-topic in a JSON format: +``` +shellies/shellyplusht-xxx/status/humidity:0 {"id": 0,"rh":51.9} +``` +You can use PayloadField to extract the desired value. + ### Tasmota An example configuration for the tasmota based Gosund SP111 device is given in [examples/gosund_sp111.yaml](examples/gosund_sp111.yaml). diff --git a/hack/shellyplusht.yaml b/hack/shellyplusht.yaml new file mode 100644 index 0000000..0fbff98 --- /dev/null +++ b/hack/shellyplusht.yaml @@ -0,0 +1,22 @@ +mqtt: + server: tcp://mosquitto:1883 + topic_path: shellies/+/sensor/+ + device_id_regex: "shellies/(?P.*)/sensor" + metric_per_topic_config: + metric_name_regex: "shellies/(?P.*)/sensor/(?P.*)" + qos: 0 +cache: + timeout: 24h +metrics: + - prom_name: temperature + # The name of the metric in a MQTT JSON message + mqtt_name: status/temperature:0 + # The field to extract in JSON payload + PayloadField: rh + # The prometheus help text for this metric + help: shelly temperature reading + # The prometheus type for this metric. Valid values are: "gauge" and "counter" + type: gauge + # A map of string to string for constant labels. This labels will be attached to every prometheus metric + const_labels: + sensor_type: shellyplusht diff --git a/pkg/config/config.go b/pkg/config/config.go index c28a40e..376de29 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -130,6 +130,7 @@ type MetricPerTopicConfig struct { type MetricConfig struct { PrometheusName string `yaml:"prom_name"` MQTTName string `yaml:"mqtt_name"` + PayloadField string `yaml:"payload_field"` SensorNameFilter Regexp `yaml:"sensor_name_filter"` Help string `yaml:"help"` ValueType string `yaml:"type"` diff --git a/pkg/metrics/extractor.go b/pkg/metrics/extractor.go index 1f2ed47..0476cd1 100644 --- a/pkg/metrics/extractor.go +++ b/pkg/metrics/extractor.go @@ -51,7 +51,19 @@ func NewMetricPerTopicExtractor(p Parser, metricNameRegex *config.Regexp) Extrac return nil, nil } - m, err := p.parseMetric(config, string(payload)) + 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) + } + } else { + rawValue = string(payload) + } + + m, err := p.parseMetric(config, rawValue) if err != nil { return nil, fmt.Errorf("failed to parse metric: %w", err) } From 1f018bf1087d0848948030d68e71be8c7ac772d8 Mon Sep 17 00:00:00 2001 From: Renovate Bot Date: Wed, 9 Aug 2023 14:31:54 +0000 Subject: [PATCH 04/12] Update dependency golang to v1.21 --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 3622bb9..330a905 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.20 as builder +FROM golang:1.21 as builder COPY . /build/mqtt2prometheus WORKDIR /build/mqtt2prometheus From a963576416996cb29d010ef17d79d1c4f80c2f25 Mon Sep 17 00:00:00 2001 From: Renovate Bot Date: Sat, 9 Dec 2023 14:27:46 +0000 Subject: [PATCH 05/12] Update dependency prom/prometheus to v2.48.1 --- hack/docker-compose.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hack/docker-compose.yml b/hack/docker-compose.yml index 808622e..ce5965d 100644 --- a/hack/docker-compose.yml +++ b/hack/docker-compose.yml @@ -22,7 +22,7 @@ services: - 1883:1883 - 9001:9001 prometheus: - image: prom/prometheus:v2.42.0 + image: prom/prometheus:v2.48.1 ports: - 9090:9090 volumes: From 70de9d66d75517907b08d960bb351017d172a39b Mon Sep 17 00:00:00 2001 From: Christoph Petrausch <263448+hikhvar@users.noreply.github.com> Date: Thu, 14 Dec 2023 23:06:01 +0100 Subject: [PATCH 06/12] Bump newest version to go 1.21 in tests --- .github/workflows/tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 7cd5338..26b59c2 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -8,7 +8,7 @@ jobs: strategy: matrix: # Test oldest and newest supported version - go-version: [1.14.x, 1.19.x] + go-version: [1.14.x, 1.21.x] platform: [ubuntu-latest, macos-latest, windows-latest] runs-on: ${{ matrix.platform }} steps: From 7419ffac11d3fd74d156841435b36f67441e3167 Mon Sep 17 00:00:00 2001 From: Christoph Petrausch <263448+hikhvar@users.noreply.github.com> Date: Thu, 14 Dec 2023 23:09:17 +0100 Subject: [PATCH 07/12] Bump to Go 1.21 in tests in relase --- .github/workflows/release.yml | 2 +- .github/workflows/tests.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 69aa9e8..85593ec 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -16,7 +16,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v3 with: - go-version: 1.19.x + go-version: 1.21.x - name: Test run: go test -cover ./... - name: Vet diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 26b59c2..dc16e4f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -45,7 +45,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v3 with: - go-version: 1.19.x + go-version: 1.21.x - name: Test run: go test -cover ./... - name: Vet From 6e9f1a21af1a207124caf97d3ec1d78e0de8952b Mon Sep 17 00:00:00 2001 From: Christoph Petrausch <263448+hikhvar@users.noreply.github.com> Date: Thu, 14 Dec 2023 23:17:40 +0100 Subject: [PATCH 08/12] Fix goreleaser deprecation --- .goreleaser.yml | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/.goreleaser.yml b/.goreleaser.yml index 1d61803..5ec7787 100644 --- a/.goreleaser.yml +++ b/.goreleaser.yml @@ -45,12 +45,15 @@ builds: - hardfloat - softfloat archives: -- replacements: - darwin: Darwin - linux: Linux - windows: Windows - 386: i386 - amd64: x86_64 +- name_template: + name_template: >- + {{- .ProjectName }}_ + {{- .Version }}_ + {{- title .Os }}_ + {{- if eq .Arch "amd64" }}x86_64 + {{- else if eq .Arch "386" }}i386 + {{- else }}{{ .Arch }}{{ end }} + {{- if .Arm }}v{{ .Arm }}{{ end -}} checksum: name_template: 'checksums.txt' snapshot: From c071e922230e47a0936da51d651b081aaf290e2d Mon Sep 17 00:00:00 2001 From: Christoph Petrausch <263448+hikhvar@users.noreply.github.com> Date: Thu, 14 Dec 2023 23:22:25 +0100 Subject: [PATCH 09/12] Fix go releaser config --- .goreleaser.yml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.goreleaser.yml b/.goreleaser.yml index 5ec7787..21b4645 100644 --- a/.goreleaser.yml +++ b/.goreleaser.yml @@ -45,8 +45,7 @@ builds: - hardfloat - softfloat archives: -- name_template: - name_template: >- +- name_template: >- {{- .ProjectName }}_ {{- .Version }}_ {{- title .Os }}_ From 5c1917ff683a911b44abf7c225b55191d59c89cb Mon Sep 17 00:00:00 2001 From: Christoph Petrausch <263448+hikhvar@users.noreply.github.com> Date: Thu, 14 Dec 2023 23:22:37 +0100 Subject: [PATCH 10/12] Bump action versions --- .github/workflows/codeql-analysis.yml | 2 +- .github/workflows/release.yml | 2 +- .github/workflows/renovate.yaml | 2 +- .github/workflows/tests.yml | 16 ++++++++-------- 4 files changed, 11 insertions(+), 11 deletions(-) diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml index 03ed688..ec61a0f 100644 --- a/.github/workflows/codeql-analysis.yml +++ b/.github/workflows/codeql-analysis.yml @@ -35,7 +35,7 @@ jobs: steps: - name: Checkout repository - uses: actions/checkout@v2 + uses: actions/checkout@v4 # Initializes the CodeQL tools for scanning. - name: Initialize CodeQL diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 85593ec..9461498 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -10,7 +10,7 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Unshallow run: git fetch --prune --unshallow - name: Set up Go diff --git a/.github/workflows/renovate.yaml b/.github/workflows/renovate.yaml index e7e8ce5..24221ca 100644 --- a/.github/workflows/renovate.yaml +++ b/.github/workflows/renovate.yaml @@ -16,7 +16,7 @@ jobs: APP_ID: ${{ secrets.APP_ID }} - name: Checkout - uses: actions/checkout@v2.0.0 + uses: actions/checkout@v4 - name: Self-hosted Renovate uses: renovatebot/github-action@v32.118.0 diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index dc16e4f..0352f9f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -17,7 +17,7 @@ jobs: with: go-version: ${{ matrix.go-version }} - name: Checkout code - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Test run: go test ./... - name: Vet @@ -26,9 +26,13 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout code - uses: actions/checkout@v2 + uses: actions/checkout@v4 + - name: Set up Go + uses: actions/setup-go@v3 + with: + go-version: 1.21.x - name: Run golangci-lint - uses: golangci/golangci-lint-action@v2 + uses: golangci/golangci-lint-action@v3 with: only-new-issues: true @@ -39,17 +43,13 @@ jobs: runs-on: ubuntu-latest steps: - name: Checkout - uses: actions/checkout@v2 + uses: actions/checkout@v4 - name: Unshallow run: git fetch --prune --unshallow - name: Set up Go uses: actions/setup-go@v3 with: go-version: 1.21.x - - name: Test - run: go test -cover ./... - - name: Vet - run: go vet ./... - name: Set up QEMU uses: docker/setup-qemu-action@v1 - name: Set up Docker Buildx From 4943c97963be25fe8498bf1ecc8bedc361c6b905 Mon Sep 17 00:00:00 2001 From: Christian Schneider Date: Wed, 6 Dec 2023 10:12:23 +0100 Subject: [PATCH 11/12] Add option to enforce strict monotonicy in metrics. This change adds the new option `force_monotonicy` to metric configurations. It is intended for almost-but-not-really monotinic sources, such as counters which reset when the sensor is restarted. When this option is set to `true`, the source metric value is regularly written to disk. This allows us to detect and compensate counter resets even between restarts. When a reset is detected, the last value before the reset becomes the new offset, which is added to the metric value going forth. The result is a strictly monotonic time series, like an ever increasing counter. --- Readme.md | 14 +++++ cmd/mqtt2prometheus.go | 2 +- pkg/config/config.go | 24 +++++++- pkg/metrics/extractor.go | 17 +++++- pkg/metrics/parser.go | 94 +++++++++++++++++++++++++++- pkg/metrics/parser_test.go | 121 +++++++++++++++++++++++++++++++++++-- 6 files changed, 261 insertions(+), 11 deletions(-) diff --git a/Readme.md b/Readme.md index 99f7d09..b79484d 100644 --- a/Readme.md +++ b/Readme.md @@ -182,6 +182,8 @@ cache: # Set the timeout to -1 to disable the deletion of metrics from the cache. The exporter presents the ingest timestamp # to prometheus. timeout: 24h + # Path to the directory to keep the state for monotonic metrics. + state_directory: "/var/lib/mqtt2prometheus" json_parsing: # Separator. Used to split path to elements when accessing json fields. # You can access json fields with dots in it. F.E. {"key.name": {"nested": "value"}} @@ -248,6 +250,18 @@ metrics: # Metric value to use if a match cannot be found in the map above. # If not specified, parsing error will occur. error_value: 1 + # 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 + # Regular expression to only match sensors with the given name pattern + sensor_name_filter: "^shellyplus1pm-.*$" + # The prometheus help text for this metric + 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. + force_monotonicy: true ``` diff --git a/cmd/mqtt2prometheus.go b/cmd/mqtt2prometheus.go index 02d1776..6adcb00 100644 --- a/cmd/mqtt2prometheus.go +++ b/cmd/mqtt2prometheus.go @@ -227,7 +227,7 @@ func setupGoKitLogger(l *zap.Logger) log.Logger { } func setupExtractor(cfg config.Config) (metrics.Extractor, error) { - parser := metrics.NewParser(cfg.Metrics, cfg.JsonParsing.Separator) + parser := metrics.NewParser(cfg.Metrics, cfg.JsonParsing.Separator, cfg.Cache.StateDir) if cfg.MQTT.ObjectPerTopicConfig != nil { switch cfg.MQTT.ObjectPerTopicConfig.Encoding { case config.EncodingJSON: diff --git a/pkg/config/config.go b/pkg/config/config.go index 376de29..a14b510 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -3,6 +3,7 @@ package config import ( "fmt" "io/ioutil" + "os" "regexp" "time" @@ -26,7 +27,8 @@ var MQTTConfigDefaults = MQTTConfig{ } var CacheConfigDefaults = CacheConfig{ - Timeout: 2 * time.Minute, + Timeout: 2 * time.Minute, + StateDir: "/var/lib/mqtt2prometheus", } var JsonParsingConfigDefaults = JsonParsingConfig{ @@ -94,7 +96,8 @@ type Config struct { } type CacheConfig struct { - Timeout time.Duration `yaml:"timeout"` + Timeout time.Duration `yaml:"timeout"` + StateDir string `yaml:"state_directory"` } type JsonParsingConfig struct { @@ -135,6 +138,7 @@ type MetricConfig struct { Help string `yaml:"help"` ValueType string `yaml:"type"` OmitTimestamp bool `yaml:"omit_timestamp"` + ForceMonotonicy bool `yaml:"force_monotonicy"` ConstantLabels map[string]string `yaml:"const_labels"` StringValueMapping *StringValueMappingConfig `yaml:"string_value_mapping"` MQTTValueScale float64 `yaml:"mqtt_value_scale"` @@ -179,6 +183,9 @@ func LoadConfig(configFile string) (Config, error) { if cfg.Cache == nil { cfg.Cache = &CacheConfigDefaults } + if cfg.Cache.StateDir == "" { + cfg.Cache.StateDir = CacheConfigDefaults.StateDir + } if cfg.JsonParsing == nil { cfg.JsonParsing = &JsonParsingConfigDefaults } @@ -217,5 +224,18 @@ func LoadConfig(configFile string) (Config, error) { } } + // If any metric forces monotonicy, we need a state directory. + forcesMonotonicy := false + for _, m := range cfg.Metrics { + if m.ForceMonotonicy { + forcesMonotonicy = true + } + } + if forcesMonotonicy { + if err := os.MkdirAll(cfg.Cache.StateDir, 0755); err != nil { + return Config{}, err + } + } + return cfg, nil } diff --git a/pkg/metrics/extractor.go b/pkg/metrics/extractor.go index 0476cd1..ea91e38 100644 --- a/pkg/metrics/extractor.go +++ b/pkg/metrics/extractor.go @@ -2,6 +2,7 @@ package metrics import ( "fmt" + "regexp" "github.com/hikhvar/mqtt2prometheus/pkg/config" gojsonq "github.com/thedevsaddam/gojsonq/v2" @@ -9,6 +10,16 @@ import ( type Extractor func(topic string, payload []byte, deviceID string) (MetricCollection, error) +// metricID returns a deterministic identifier per metic config which is safe to use in a file path. +func metricID(topic, metric, deviceID, promName string) string { + re := regexp.MustCompile(`[^a-zA-Z0-9]`) + deviceID = re.ReplaceAllString(deviceID, "_") + topic = re.ReplaceAllString(topic, "_") + metric = re.ReplaceAllString(metric, "_") + promName = re.ReplaceAllString(promName, "_") + return fmt.Sprintf("%s-%s-%s-%s", deviceID, topic, metric, promName) +} + func NewJSONObjectExtractor(p Parser) Extractor { return func(topic string, payload []byte, deviceID string) (MetricCollection, error) { var mc MetricCollection @@ -27,7 +38,8 @@ func NewJSONObjectExtractor(p Parser) Extractor { continue } - m, err := p.parseMetric(config, rawValue) + 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) } @@ -63,7 +75,8 @@ func NewMetricPerTopicExtractor(p Parser, metricNameRegex *config.Regexp) Extrac rawValue = string(payload) } - m, err := p.parseMetric(config, rawValue) + 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) } diff --git a/pkg/metrics/parser.go b/pkg/metrics/parser.go index 2482239..9efbd66 100644 --- a/pkg/metrics/parser.go +++ b/pkg/metrics/parser.go @@ -2,22 +2,44 @@ package metrics import ( "fmt" + "os" "strconv" + "strings" "time" "github.com/hikhvar/mqtt2prometheus/pkg/config" + "gopkg.in/yaml.v2" ) +// monotonicState holds the runtime information to realize a monotonic increasing value. +type monotonicState 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"` +} + +// metricState holds runtime information per metric configuration. +type metricState struct { + monotonic monotonicState + // The last time the state file was written + lastWritten time.Time +} + 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 + // Directory holding state files + stateDir string + // Per-metric state + states map[string]*metricState } var now = time.Now -func NewParser(metrics []config.MetricConfig, separator string) Parser { +func NewParser(metrics []config.MetricConfig, separator, stateDir string) Parser { cfgs := make(map[string][]config.MetricConfig) for i := range metrics { key := metrics[i].MQTTName @@ -26,6 +48,8 @@ func NewParser(metrics []config.MetricConfig, separator string) Parser { return Parser{ separator: separator, metricConfigs: cfgs, + stateDir: strings.TrimRight(stateDir, "/"), + states: make(map[string]*metricState), } } @@ -47,7 +71,7 @@ func (p *Parser) findMetricConfig(metric string, deviceID string) (config.Metric // 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, value interface{}) (Metric, error) { +func (p *Parser) parseMetric(cfg config.MetricConfig, metricID string, value interface{}) (Metric, error) { var metricValue float64 if boolValue, ok := value.(bool); ok { @@ -87,6 +111,22 @@ func (p *Parser) parseMetric(cfg config.MetricConfig, value interface{}) (Metric return Metric{}, fmt.Errorf("got data with unexpectd type: %T ('%s')", value, value) } + if cfg.ForceMonotonicy { + ms, err := p.getMetricState(metricID) + if 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 { metricValue = metricValue * cfg.MQTTValueScale } @@ -103,3 +143,53 @@ func (p *Parser) parseMetric(cfg config.MetricConfig, value interface{}) (Metric IngestTime: ingestTime, }, nil } + +func (p *Parser) stateFileName(metricID string) string { + return fmt.Sprintf("%s/%s.yaml", p.stateDir, metricID) +} + +// readMetricState parses the metric state from the configured path. +// If the file does not exist, an empty state is returned. +func (p *Parser) readMetricState(metricID string) (*metricState, error) { + data, err := os.ReadFile(p.stateFileName(metricID)) + state := &metricState{} + if err != nil { + // The file does not exist for new metrics. + if os.IsNotExist(err) { + return state, nil + } + return state, err + } + err = yaml.UnmarshalStrict(data, &state.monotonic) + 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) + if err != nil { + return err + } + return os.WriteFile(p.stateFileName(metricID), out, 0644) +} + +// getMetricState returns the state of the given metric. +// The state is read from and written back to disk as needed. +func (p *Parser) getMetricState(metricID string) (*metricState, error) { + var err error + state, found := p.states[metricID] + if !found { + if state, err = p.readMetricState(metricID); err != nil { + return nil, err + } + p.states[metricID] = state + } + // Write the state back to disc every minute. + if now().Sub(state.lastWritten) >= time.Minute { + if err = p.writeMetricState(metricID, state); err == nil { + state.lastWritten = now() + } + } + return state, err +} diff --git a/pkg/metrics/parser_test.go b/pkg/metrics/parser_test.go index d23df61..31d3b72 100644 --- a/pkg/metrics/parser_test.go +++ b/pkg/metrics/parser_test.go @@ -1,6 +1,7 @@ package metrics import ( + "os" "reflect" "testing" "time" @@ -10,6 +11,12 @@ import ( ) func TestParser_parseMetric(t *testing.T) { + stateDir, err := os.MkdirTemp("", "parser_test") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(stateDir) + now = testNow type fields struct { metricConfigs map[string][]config.MetricConfig @@ -415,12 +422,111 @@ func TestParser_parseMetric(t *testing.T) { }, wantErr: true, }, + { + name: "monotonic gauge, step 1: initial value", + fields: fields{ + map[string][]config.MetricConfig{ + "aenergy.total": []config.MetricConfig{ + { + PrometheusName: "total_energy", + ValueType: "gauge", + OmitTimestamp: true, + ForceMonotonicy: true, + }, + }, + }, + }, + args: args{ + metricPath: "aenergy.total", + deviceID: "shellyplus1pm-foo", + value: 1.0, + }, + want: Metric{ + Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil), + ValueType: prometheus.GaugeValue, + Value: 1.0, + }, + }, + { + name: "monotonic gauge, step 2: monotonic increase does not add offset", + fields: fields{ + map[string][]config.MetricConfig{ + "aenergy.total": []config.MetricConfig{ + { + PrometheusName: "total_energy", + ValueType: "gauge", + OmitTimestamp: true, + ForceMonotonicy: true, + }, + }, + }, + }, + args: args{ + metricPath: "aenergy.total", + deviceID: "shellyplus1pm-foo", + value: 2.0, + }, + want: Metric{ + Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil), + ValueType: prometheus.GaugeValue, + Value: 2.0, + }, + }, + { + 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{ + { + PrometheusName: "total_energy", + ValueType: "gauge", + OmitTimestamp: true, + ForceMonotonicy: true, + }, + }, + }, + }, + args: args{ + metricPath: "aenergy.total", + deviceID: "shellyplus1pm-foo", + value: 0.0, + }, + want: Metric{ + Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil), + ValueType: prometheus.GaugeValue, + Value: 2.0, + }, + }, + { + name: "monotonic gauge, step 4: monotonic increase with offset", + fields: fields{ + map[string][]config.MetricConfig{ + "aenergy.total": []config.MetricConfig{ + { + PrometheusName: "total_energy", + ValueType: "gauge", + OmitTimestamp: true, + ForceMonotonicy: true, + }, + }, + }, + }, + args: args{ + metricPath: "aenergy.total", + deviceID: "shellyplus1pm-foo", + value: 1.0, + }, + want: Metric{ + Description: prometheus.NewDesc("total_energy", "", []string{"sensor", "topic"}, nil), + ValueType: prometheus.GaugeValue, + Value: 3.0, + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - p := &Parser{ - metricConfigs: tt.fields.metricConfigs, - } + 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) @@ -431,7 +537,8 @@ func TestParser_parseMetric(t *testing.T) { return } - got, err := p.parseMetric(config, tt.args.value) + id := metricID("", tt.args.metricPath, tt.args.deviceID, config.PrometheusName) + got, err := p.parseMetric(config, id, tt.args.value) if (err != nil) != tt.wantErr { t.Errorf("parseMetric() error = %v, wantErr %v", err, tt.wantErr) return @@ -439,6 +546,12 @@ func TestParser_parseMetric(t *testing.T) { if !reflect.DeepEqual(got, tt.want) { t.Errorf("parseMetric() got = %v, want %v", got, tt.want) } + + if config.ForceMonotonicy { + if err = p.writeMetricState(id, p.states[id]); err != nil { + t.Errorf("failed to write metric state: %v", err) + } + } }) } } From 84ed960831ca3153dc75a25b2a52a9c158d9fc65 Mon Sep 17 00:00:00 2001 From: Christian Schneider Date: Mon, 5 Feb 2024 21:56:40 +0100 Subject: [PATCH 12/12] Restrict to Go 1.14 functionality --- pkg/metrics/parser.go | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/pkg/metrics/parser.go b/pkg/metrics/parser.go index 9efbd66..d016b18 100644 --- a/pkg/metrics/parser.go +++ b/pkg/metrics/parser.go @@ -2,6 +2,7 @@ package metrics import ( "fmt" + "io" "os" "strconv" "strings" @@ -151,8 +152,8 @@ func (p *Parser) stateFileName(metricID string) string { // readMetricState parses the metric state from the configured path. // If the file does not exist, an empty state is returned. func (p *Parser) readMetricState(metricID string) (*metricState, error) { - data, err := os.ReadFile(p.stateFileName(metricID)) state := &metricState{} + f, err := os.Open(p.stateFileName(metricID)) if err != nil { // The file does not exist for new metrics. if os.IsNotExist(err) { @@ -160,6 +161,16 @@ func (p *Parser) readMetricState(metricID string) (*metricState, error) { } return state, err } + defer f.Close() + + var data []byte + if info, err := f.Stat(); err == nil { + data = make([]byte, int(info.Size())) + } + if _, err := f.Read(data); err != nil && err != io.EOF { + return state, err + } + err = yaml.UnmarshalStrict(data, &state.monotonic) state.lastWritten = now() return state, err @@ -171,7 +182,13 @@ func (p *Parser) writeMetricState(metricID string, state *metricState) error { if err != nil { return err } - return os.WriteFile(p.stateFileName(metricID), out, 0644) + f, err := os.OpenFile(p.stateFileName(metricID), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return err + } + _, err = f.Write(out) + f.Close() + return err } // getMetricState returns the state of the given metric.