Reconnect to the broker if all devices do not receive any message within the threshold.

2025-11-01 12:33:44 +00:00 · 2024-07-15 22:14:02 +03:00
parent caa786a816
commit 79a84c4720
6 changed files with 63 additions and 35 deletions
--- a/docs/prometheus.md
+++ b/docs/prometheus.md
@@ -7,13 +7,13 @@
      is `rest`.

   If  `EXPORTER_TYPE=rest` is selected, then provide values for the following parameters:
-      - `ECOFLOW_ACCESS_KEY` - the access key from the Ecoflow development website
-      - `ECOFLOW_SECRET_KEY` - the secret key from the Ecoflow development website
-   
+    - `ECOFLOW_ACCESS_KEY` - the access key from the Ecoflow development website
+    - `ECOFLOW_SECRET_KEY` - the secret key from the Ecoflow development website
+
   If  `EXPORTER_TYPE=mqtt` is selected, then provide values for the following parameters:
-      - `ECOFLOW_EMAIL` - your email address that you use to log in to the Ecoflow mobile app
-      - `ECOFLOW_PASSWORD` - your ecoflow password
-      - `ECOFLOW_DEVICES` - the list of devices serial numbers separated by comma. For instance: `SN1,SN2,SN3`
+    - `ECOFLOW_EMAIL` - your email address that you use to log in to the Ecoflow mobile app
+    - `ECOFLOW_PASSWORD` - your ecoflow password
+    - `ECOFLOW_DEVICES` - the list of devices serial numbers separated by comma. For instance: `SN1,SN2,SN3`

 3. (OPTIONALLY) Update other variables if you need to:
    - `METRIC_PREFIX`: the prefix that will be added to all metrics. Default value is `ecoflow`. For instance
@@ -22,6 +22,11 @@
    - `SCRAPING_INTERVAL` - scrapping interval in seconds. How often should the exporter execute requests to Ecoflow
      Rest API in order to get the data. Default value is 30 seconds. Align this value
      with `docker-compose/prometheus/prometheus.yml`
+    - `MQTT_DEVICE_OFFLINE_THRESHOLD_SECONDS` - the threshold in seconds which indicates how long we should way for a
+      metric message from MQTT broker. Default value: 60 seconds. If we don't receive message within 60 seconds we
+      consider that device is offline. If we don't receive messages within the threshold for all devices, we'll try to
+      reconnect to the MQTT broker (there is a strange behavior that MQTT stop sends messages if you open Ecoflow mobile
+      app and then close it).
    - `DEBUG_ENABLED` - enable debug log messages. Default value is "false". To enable use values `true` or `1`
    - `GRAFANA_USERNAME` - admin username in Grafana. Default value: `grafana`. Can be changed later in Grafana UI
    - `GRAFANA_PASSWORD` - admin password in Grafana. Default value: `grafana`. Can be changed later in Grafana UI
--- a/docs/redis.md
+++ b/docs/redis.md
@@ -18,18 +18,18 @@ This can be useful when declaring variables in Grafana to fetch all devices you

 1. Go to docker-compose folder: `cd docker-compose`
 2. Update `.env` file with two mandatory parameters:
-   - `REDIS_ENABLED` - true (or 1) if you want to enable integration with Redis. Default value is false
-   - `EXPORTER_TYPE` - the type of exporter you'd like to use. Possible values: `rest` and `mqtt`. Default value
-     is `rest`.
+    - `REDIS_ENABLED` - true (or 1) if you want to enable integration with Redis. Default value is false
+    - `EXPORTER_TYPE` - the type of exporter you'd like to use. Possible values: `rest` and `mqtt`. Default value
+      is `rest`.

   If  `EXPORTER_TYPE=rest` is selected, then provide values for the following parameters:
-   - `ECOFLOW_ACCESS_KEY` - the access key from the Ecoflow development website
-   - `ECOFLOW_SECRET_KEY` - the secret key from the Ecoflow development website
+    - `ECOFLOW_ACCESS_KEY` - the access key from the Ecoflow development website
+    - `ECOFLOW_SECRET_KEY` - the secret key from the Ecoflow development website

   If  `EXPORTER_TYPE=mqtt` is selected, then provide values for the following parameters:
-   - `ECOFLOW_EMAIL` - your email address that you use to log in to the Ecoflow mobile app
-   - `ECOFLOW_PASSWORD` - your ecoflow password
-   - `ECOFLOW_DEVICES` - the list of devices serial numbers separated by comma. For instance: `SN1,SN2,SN3`
+    - `ECOFLOW_EMAIL` - your email address that you use to log in to the Ecoflow mobile app
+    - `ECOFLOW_PASSWORD` - your ecoflow password
+    - `ECOFLOW_DEVICES` - the list of devices serial numbers separated by comma. For instance: `SN1,SN2,SN3`

 3. (OPTIONALLY) Update other variables if you need to:
    - `REDIS_URL` - Redis url. Default value: `localhost:6379`
@@ -40,6 +40,11 @@ This can be useful when declaring variables in Grafana to fetch all devices you
      metric `bms_bmsStatus.minCellTemp` will be exported to prometheus as `ecoflow.bms_bmsStatus.minCellTemp`.
    - `SCRAPING_INTERVAL` - scrapping interval in seconds. How often should the exporter execute requests to Ecoflow
      Rest API in order to get the data. Default value is 30 seconds.
+    - `MQTT_DEVICE_OFFLINE_THRESHOLD_SECONDS` - the threshold in seconds which indicates how long we should way for a
+      metric message from MQTT broker. Default value: 60 seconds. If we don't receive message within 60 seconds we
+      consider that device is offline. If we don't receive messages within the threshold for all devices, we'll try to
+      reconnect to the MQTT broker (there is a strange behavior that MQTT stop sends messages if you open Ecoflow mobile
+      app and then close it).
    - `DEBUG_ENABLED` - enable debug log messages. Default value is "false". To enable use values `true` or `1`
    - `GRAFANA_USERNAME` - admin username in Grafana. Default value: `grafana`. Can be changed later in Grafana UI
    - `GRAFANA_PASSWORD` - admin password in Grafana. Default value: `grafana`. Can be changed later in Grafana UI
@@ -52,21 +57,22 @@ This can be useful when declaring variables in Grafana to fetch all devices you
    - http://localhost:3000 - Grafana
    - Redis is available at the value of `REDIS_URL` variable
 9. Install Redis plugin: Navigate to http://localhost:3000/plugins/redis-datasource and click on `Install` button
-![img.png](images/redis_plugin.png)
+   ![img.png](images/redis_plugin.png)
 10. Create Redis datasource: Navigate to http://localhost:3000/connections/datasources/new and search for `Redis`.
-![img.png](images/redis_datasource.png)
+    ![img.png](images/redis_datasource.png)
 11. Create your dashboard.

-
 ## Dashboard example
+
 ![img.png](images/dashboard_example.png)

 ### Grafana dashboard tips

 - I suggest to add new Variable "Device" to get the dropdown list of devices. Example:
-![img_1.png](images/redis_add_variable.png)
+  ![img_1.png](images/redis_add_variable.png)

- If you have Prometheus query defined like `ecoflow_bms_bms_status_cycles{device="$device"}` you can implement the same using Redis:
+- If you have Prometheus query defined like `ecoflow_bms_bms_status_cycles{device="$device"}` you can implement the same
+  using Redis:
    - *Datasource*: Redis
    - *Type*: RedisTimeSeries
    - *Command*: TS.GET
--- a/docs/timescaledb.md
+++ b/docs/timescaledb.md
@@ -26,18 +26,18 @@ There is no cleanup procedure implemented at the moment, so you might want to cl

 1. Go to docker-compose folder: `cd docker-compose`
 2. Update `.env` file with two mandatory parameters:
-   - `TIMESCALE_ENABLED` - true (or 1) if you want to enable integration with TimescaleDB. Default value is false
-   - `EXPORTER_TYPE` - the type of exporter you'd like to use. Possible values: `rest` and `mqtt`. Default value
-     is `rest`.
+    - `TIMESCALE_ENABLED` - true (or 1) if you want to enable integration with TimescaleDB. Default value is false
+    - `EXPORTER_TYPE` - the type of exporter you'd like to use. Possible values: `rest` and `mqtt`. Default value
+      is `rest`.

   If  `EXPORTER_TYPE=rest` is selected, then provide values for the following parameters:
-   - `ECOFLOW_ACCESS_KEY` - the access key from the Ecoflow development website
-   - `ECOFLOW_SECRET_KEY` - the secret key from the Ecoflow development website
+    - `ECOFLOW_ACCESS_KEY` - the access key from the Ecoflow development website
+    - `ECOFLOW_SECRET_KEY` - the secret key from the Ecoflow development website

   If  `EXPORTER_TYPE=mqtt` is selected, then provide values for the following parameters:
-   - `ECOFLOW_EMAIL` - your email address that you use to log in to the Ecoflow mobile app
-   - `ECOFLOW_PASSWORD` - your ecoflow password
-   - `ECOFLOW_DEVICES` - the list of devices serial numbers separated by comma. For instance: `SN1,SN2,SN3`
+    - `ECOFLOW_EMAIL` - your email address that you use to log in to the Ecoflow mobile app
+    - `ECOFLOW_PASSWORD` - your ecoflow password
+    - `ECOFLOW_DEVICES` - the list of devices serial numbers separated by comma. For instance: `SN1,SN2,SN3`

 3. (OPTIONALLY) Update other variables if you need to:
    - `TIMESCALE_USERNAME` - TimescaleDB username. Default value: `postgres`
@@ -49,14 +49,21 @@ There is no cleanup procedure implemented at the moment, so you might want to cl
      metric `bms_bmsStatus.minCellTemp` will be exported to prometheus as `ecoflow.bms_bmsStatus.minCellTemp`.
    - `SCRAPING_INTERVAL` - scrapping interval in seconds. How often should the exporter execute requests to Ecoflow
      Rest API in order to get the data. Default value is 30 seconds.
+    - `MQTT_DEVICE_OFFLINE_THRESHOLD_SECONDS` - the threshold in seconds which indicates how long we should way for a
+      metric message from MQTT broker. Default value: 60 seconds. If we don't receive message within 60 seconds we
+      consider that device is offline. If we don't receive messages within the threshold for all devices, we'll try to
+      reconnect to the MQTT broker (there is a strange behavior that MQTT stop sends messages if you open Ecoflow mobile
+      app and then close it).
    - `DEBUG_ENABLED` - enable debug log messages. Default value is "false". To enable use values `true` or `1`
    - `GRAFANA_USERNAME` - admin username in Grafana. Default value: `grafana`. Can be changed later in Grafana UI
    - `GRAFANA_PASSWORD` - admin password in Grafana. Default value: `grafana`. Can be changed later in Grafana UI

 4. Save `.env` file with your changes.
 5. Start timescaledb container: `docker-compose -f docker-compose/timescale-compose.yml up -d`.\
-   :exclamation: *NOTE*: The exporter does not wait until the TimescaleDB is UP, because TimescaleDB is an optional dependency for the
-   exporter. Thus, it's important to start the DB first and then the exporter. There is a retry mechanism to wait for the DB
+   :exclamation: *NOTE*: The exporter does not wait until the TimescaleDB is UP, because TimescaleDB is an optional
+   dependency for the
+   exporter. Thus, it's important to start the DB first and then the exporter. There is a retry mechanism to wait for
+   the DB
   to be operational, however it's better to do not rely on it and start the DB before the exporter.
 6. Start the exporter and
   grafana: `docker-compose -f docker-compose/grafana-compose.yml -f docker-compose/exporter-remote-compose.yml up -d`
--- a/main.go
+++ b/main.go
@@ -25,7 +25,7 @@ const (
 // prometheus
 const (
 	defaultMetricsPort             = "2112"
-	defaultOfflineThresholdSeconds = 30
+	defaultOfflineThresholdSeconds = 60
 )

 // timescaledb
--- a/mqtt_exporter.go
+++ b/mqtt_exporter.go
@@ -105,21 +105,21 @@ func (e *MqttMetricsExporter) OnConnectionLost(_ mqtt.Client, err error) {
 	slog.Error("Lost connection to the broker", "error", err)
 }

-func (e *MqttMetricsExporter) OnReconnect(client mqtt.Client, options *mqtt.ClientOptions) {
+func (e *MqttMetricsExporter) OnReconnect(_ mqtt.Client, _ *mqtt.ClientOptions) {
 	slog.Info("Trying to reconnect to the broker...")
 }

 func (e *MqttMetricsExporter) monitorDeviceStatus() {
 	for {
-		time.Sleep(e.offlineThreshold / 2) // Check twice as often as the threshold
-		mu.Lock()
+		time.Sleep(e.offlineThreshold)
 		if !e.c.Client.IsConnected() {
 			slog.Debug("MQTT client is not connected to the broker, we don't know the devices statuses...")
 			continue
 		}
+		var offlineDevicesCount = 0
 		for sn, status := range deviceStatuses {
 			if time.Since(status.LastReceived) > e.offlineThreshold {
-				slog.Debug("Device is offline", "serial_number", sn)
+				offlineDevicesCount = offlineDevicesCount + 1
 				for _, handler := range e.handlers {
 					hh := handler
 					go hh.Handle(context.Background(), ecoflow.DeviceInfo{
@@ -129,7 +129,14 @@ func (e *MqttMetricsExporter) monitorDeviceStatus() {
 				}
 			}
 		}
-		mu.Unlock()
+		//if true it means that either all devices are offline or we don't receive any messages from MQTT broker.
+		//either way we need to reconnect the client
+		if len(e.devices) == offlineDevicesCount {
+			slog.Error("All devices are either offline or we don't receive messages from MQTT topic, we'll try to reconnect")
+			e.c.Client.Disconnect(250)
+			time.Sleep(5 * time.Second)
+			e.c.Client.Connect()
+		}
 	}
 }

@@ -142,6 +149,7 @@ func (e *MqttMetricsExporter) initDeviceStatuses() {
 	}
 }

+// assuming that SN is the last part of the topic ("/app/device/property/${sn}")
 func getSnFromTopic(topic string) string {
 	topicStr := strings.Split(topic, "/")
 	return topicStr[len(topicStr)-1]
--- a/prometheus.go
+++ b/prometheus.go
@@ -88,7 +88,9 @@ func (p *PrometheusExporter) handleOneMetric(device ecoflow.DeviceInfo, field st
 		slog.Error("Unable to generate metric name", "metric", field)
 		return
 	}
+	p.mu.Lock()
 	gauge, ok := p.metrics[deviceMetricName]
+	p.mu.Unlock()
 	if !ok {
 		slog.Debug("Adding new metric", "metric", metricName, "device", device.SN)
 		gauge = prometheus.NewGauge(prometheus.GaugeOpts{