readme

Cargo.lock

@@ -983,7 +983,7 @@ dependencies = [
 
 [[package]]
 name = "mqtt-exporter"
-version = "0.1.0"
+version = "0.1.2"
 dependencies = [
  "backoff",
  "bytes",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mqtt-exporter"
-version = "0.1.0"
+version = "0.1.2"
 edition = "2021"
 
 [dependencies]
@@ -20,3 +20,6 @@ serde = { version = "1.0.218", features = ["derive"] }
 serde_regex = "1.1.0"
 tokio = { version = "1.44.0", features = ["net", "rt-multi-thread"] }
 tokio-util = { version = "0.7.13", features = ["codec"] }
+
+[profile.release]
+lto = "fat"

Dockerfile

@@ -1,8 +1,12 @@
-FROM rust:1.85
+FROM rust:1 AS builder
 
 WORKDIR /usr/src/mqtt-exporter
-COPY . .
+COPY src/ ./src
+COPY Cargo.* .
 
-RUN cargo install --path .
+RUN cargo install --path . --root /app
+
+FROM debian:bookworm-slim
+COPY --from=builder /app/bin/mqtt-exporter /usr/local/bin/mqtt-exporter
 
 CMD ["mqtt-exporter"]

README.md

@@ -0,0 +1,5 @@
+# mqtt-exporter-rs
+
+A reimplementation of [mqtt-exporter](https://github.com/kpetremann/mqtt-exporter) in Rust. That's pretty much the gist of it. I was having issues with memory leaks in the Python implementation so I spun this together. It works for my purposes but is far from tested.
+
+I slightly modified the logic, since I don't understand why some of the topic munging is done and would prefer to keep the topics as-is.

src/main.rs

@@ -2,25 +2,24 @@ use backoff::backoff::Backoff;
 use config::Config;
 use core::str;
 use json::JsonValue;
-use log::{debug, error, info};
+use log::{debug, error, info, warn};
 use metrics::gauge;
 use metrics_exporter_prometheus::PrometheusBuilder;
 use metrics_util::MetricKindMask;
-use mqtt_exporter::transforms::*;
 use regex::Regex;
 use rumqttc::{
-    AsyncClient,
+    AsyncClient, ConnAck,
     Event::{self},
-    MqttOptions, Packet, Publish, QoS,
+    EventLoop, MqttOptions, Packet, Publish, QoS, SubscribeFilter,
 };
 use serde::Deserialize;
 use std::{
     net::{IpAddr, Ipv6Addr},
     process::exit,
-    time::Duration,
+    time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
-use std::time::{SystemTime, UNIX_EPOCH};
+use mqtt_exporter::transforms::*;
 
 struct ConfigDefaults;
 impl ConfigDefaults {
@@ -111,6 +110,7 @@ struct AppConfig {
 }
 struct App {
     config: AppConfig,
+    client: AsyncClient,
     transformers: Vec<Box<dyn Transform>>,
 }
 
@@ -178,7 +178,7 @@ impl App {
                     let metric_s = metric_path.join("_");
                     let mut labels = labels.clone();
                     labels.push((self.config.topic_label.to_owned(), topic.to_owned()));
-                    debug!("metric: {metric_s} value: {v} labels: {labels:?}");
+                    debug!(target: "mqtt-exporter::prom", "metric: {metric_s} value: {v} labels: {labels:?}");
 
                     if self.config.expose_last_seen {
                         let metric_ts = format!("{}_ts", metric_s);
@@ -210,18 +210,34 @@ impl App {
         Ok(json::parse(str::from_utf8(payload)?)?)
     }
 
-    fn handle_event(&self, event: &Event) {
+    async fn handle_event(&self, event: &Event) {
         match event {
             Event::Incoming(Packet::Publish(notification)) => self.handle_publish(notification),
+            Event::Incoming(Packet::ConnAck(ack)) => self.handle_connack(ack).await,
+            // Event::Incoming(Packet::SubAck(ack)) => self.handle_suback(ack),
             Event::Outgoing(_) => {}
-            e => debug!("Unhandled event: {:?}", e),
+            e => debug!(target: "mqtt-exporter::mqtt", "Unhandled event: {:?}", e),
         }
     }
 
     fn handle_publish(&self, event: &Publish) {
         match Self::deserialize_payload(&event.payload) {
             Ok(parsed) => self.update_metrics(&event.topic, parsed),
-            Err(e) => error!("Error deserializing JSON payload: {:?}", e),
+            Err(e) => {
+                error!(target: "mqtt-exporter::mqtt", "Error deserializing JSON payload: {:?}", e)
+            }
+        }
+    }
+
+    async fn handle_connack(&self, ack: &ConnAck) {
+        match ack.code {
+            rumqttc::ConnectReturnCode::Success => {
+                info!(target: "mqtt-exporter::mqtt", "Successfully connected to MQTT broker");
+                self.subscribe().await;
+            }
+            code => {
+                warn!(target: "mqtt-exporter::mqtt", "Non success connection return code {:?}", code)
+            }
         }
     }
 
@@ -231,7 +247,7 @@ impl App {
         if !&self.topic_filter(raw_topic) {
             return;
         }
-        debug!("topic {} payload {}", raw_topic, data.pretty(2));
+        debug!(target: "mqtt-exporter::mqtt", "topic {} payload {}", raw_topic, data.pretty(2));
         let mut topic = raw_topic.to_owned();
 
         (topic, data) = self
@@ -249,45 +265,43 @@ impl App {
         self.parse_metrics(data, &topic, &topic, metric_path, &labels);
     }
 
-    async fn mqtt_task(self) {
-        let mut opts = MqttOptions::new(
-            &self.config.mqtt_client_id,
-            &self.config.mqtt_address,
-            self.config.mqtt_port,
-        );
-        opts.set_keep_alive(Duration::from_secs(self.config.mqtt_keepalive))
-            .set_max_packet_size(150000, 150000);
-
-        if let (Some(username), Some(password)) =
-            (&self.config.mqtt_username, &self.config.mqtt_password)
+    async fn subscribe(&self) {
+        match self
+            .client
+            .subscribe_many(self.config.mqtt_topic.iter().map(|topic| SubscribeFilter {
+                path: topic.to_owned(),
+                qos: QoS::AtLeastOnce,
+            }))
+            .await
         {
-            opts.set_credentials(username, password);
-        }
-
-        let (client, mut eventloop) = AsyncClient::new(opts, 10);
-
-        for topic in &self.config.mqtt_topic {
-            client.subscribe(topic, QoS::AtLeastOnce).await.unwrap();
+            Ok(_) => {
+                info!(target: "mqtt-exporter::mqtt", "Successfully subscribed to topics `{}`", &self.config.mqtt_topic.join(" "))
+            }
+            Err(e) => {
+                error!(target: "mqtt-exporter::mqtt", "Failed to subscribe to topics `{}` - {}", &self.config.mqtt_topic.join(" "), e)
+            }
+        }
     }
 
+    async fn mqtt_task(self, mut eventloop: EventLoop) {
         let mut bo = backoff::ExponentialBackoff::default();
 
         loop {
             match eventloop.poll().await {
                 Ok(event) => {
                     bo.reset();
-                    self.handle_event(&event)
+                    self.handle_event(&event).await
                 }
                 Err(e) => {
-                    error!("Connection error: {}", e);
+                    error!(target: "mqtt-exporter::mqtt", "Connection error: {}", e);
                     let delay = bo.next_backoff();
                     match delay {
                         Some(delay) => {
-                            debug!("Backing off for {}s", delay.as_secs_f32());
+                            debug!(target: "mqtt-exporter::mqtt", "Backing off for {}s", delay.as_secs_f32());
                             tokio::time::sleep(delay).await
                         }
                         None => {
-                            error!("Connection timed out, giving up");
+                            error!(target: "mqtt-exporter::mqtt", "Connection timed out, giving up");
                             exit(-1)
                         }
                     }
@@ -307,7 +321,7 @@ impl App {
         builder.install().unwrap();
     }
 
-    fn new(config: AppConfig) -> Self {
+    fn new(config: AppConfig, client: AsyncClient) -> Self {
         let mut transformers: Vec<Box<dyn Transform>> = Vec::new();
         if !config.zwave_topic_prefix.is_empty() {
             transformers.push(Box::new(ZwaveNormalize::new(&config.zwave_topic_prefix)))
@@ -317,7 +331,8 @@ impl App {
                 &config.esphome_topic_prefixes,
             )))
         }
-        if !config.hubitat_topic_prefixes.is_empty() {
+        if !config.hubitat_topic_prefixes.is_empty() && !config.hubitat_topic_prefixes[0].is_empty()
+        {
             transformers.push(Box::new(HubitatNormalize::new(
                 &config.hubitat_topic_prefixes,
             )))
@@ -325,9 +340,11 @@ impl App {
         if config.zigbee2mqtt_availability {
             transformers.push(Box::new(Zigbee2MqttAvailability::default()))
         }
-        debug!("Transformers enabled: {:?}", transformers);
+        debug!(target: "mqtt-exporter::config", "Transformers enabled: {:?}", transformers);
+
         Self {
             config,
+            client,
             transformers,
         }
     }
@@ -351,10 +368,25 @@ async fn main() {
         )
         .build()
         .unwrap();
-    let app_config = config.try_deserialize().unwrap();
-    info!("Loaded config: {app_config:?}");
+    let app_config: AppConfig = config.try_deserialize().unwrap();
+    info!(target: "mqtt-exporter::config", "Loaded config: {app_config:?}");
 
-    let app = App::new(app_config);
-    app.prom_task().await;
-    app.mqtt_task().await;
+    let mut opts = MqttOptions::new(
+        &app_config.mqtt_client_id,
+        &app_config.mqtt_address,
+        app_config.mqtt_port,
+    );
+    opts.set_keep_alive(Duration::from_secs(app_config.mqtt_keepalive))
+        .set_max_packet_size(150000, 150000);
+
+    if let (Some(username), Some(password)) = (&app_config.mqtt_username, &app_config.mqtt_password)
+    {
+        opts.set_credentials(username, password);
+    }
+
+    let (client, mut eventloop) = AsyncClient::new(opts, 10);
+
+    let app = App::new(app_config, client);
+    app.prom_task().await;
+    app.mqtt_task(eventloop).await;
 }