allow authentication against mqtt server

Cargo.lock

@@ -304,6 +304,17 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "hostname"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3c731c3e10504cc8ed35cfe2f1db4c9274c3d35fa486e3b31df46f068ef3e867"
+dependencies = [
+ "libc",
+ "match_cfg",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "http"
 version = "0.2.1"
@@ -480,6 +491,12 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
 
+[[package]]
+name = "match_cfg"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ffbee8634e0d45d258acb448e7eaab3fce7a0a467395d4d9f228e3c1f01fb2e4"
+
 [[package]]
 name = "matches"
 version = "0.1.8"
@@ -751,6 +768,7 @@ dependencies = [
  "env_logger",
  "err-derive",
  "futures-util",
+ "hostname",
  "log",
  "main_error",
  "maplit",

Cargo.toml

@@ -18,6 +18,7 @@ env_logger = "0.7"
 err-derive = "0.2.1"
 serde_json = "1.0.45"
 rumqttc = "0.2.0"
+hostname = "0.3.1"
 
 [dev-dependencies]
 maplit = "1.0.2"

src/config.rs

@@ -97,7 +97,9 @@ pub struct PrometheusConfig {
 #[derive(Debug, Clone, Deserialize)]
 pub struct MqttConfig {
     pub host: String,
-    pub port: Option<u16>
+    pub port: Option<u16>,
+    pub username: Option<String>,
+    pub password: Option<String>,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -114,7 +116,7 @@ pub enum Method {
     Get,
     Put,
     Post,
-    Mqtt
+    Mqtt,
 }
 
 #[derive(Debug, Clone, Deserialize)]

src/trigger.rs

@@ -1,14 +1,16 @@
-use crate::config::{Action, Condition, Config, Method, Parameter, ParameterError, Trigger, MqttConfig};
+use crate::config::{
+    Action, Condition, Config, Method, MqttConfig, Parameter, ParameterError, Trigger,
+};
 use err_derive::Error;
 use futures_util::future::try_join_all;
 use log::{error, info};
 use main_error::MainError;
 use prometheus_edge_detector::EdgeDetector;
 use reqwest::Client;
+use rumqttc::{AsyncClient, ClientError, Event, MqttOptions, Outgoing, QoS};
 use std::collections::HashMap;
 use std::time::{Duration, SystemTime};
 use tokio::time::delay_for;
-use rumqttc::{MqttOptions, AsyncClient, QoS};
 
 pub struct TriggerManager {
     http_client: Client,
@@ -85,7 +87,13 @@ impl TriggerManager {
                     match self.get_edge(&trigger.condition, delay).await {
                         Ok(Some(new_edge)) if new_edge == edge => {
                             info!("[{}] Edge still valid, triggering", trigger.name);
-                            if let Err(e) = run_action(&trigger.action, &self.http_client, self.mqtt_config.as_ref()).await {
+                            if let Err(e) = run_action(
+                                &trigger.action,
+                                &self.http_client,
+                                self.mqtt_config.as_ref(),
+                            )
+                            .await
+                            {
                                 error!("[{}]: {}", trigger.name, e);
                             }
                             delay_for(delay_duration).await;
@@ -138,7 +146,7 @@ async fn interpolate_option_params(
 ) -> Result<Option<String>, ParameterError> {
     match input.as_ref() {
         Some(input) => Ok(Some(interpolate_params(input, params).await?)),
-        None => Ok(None)
+        None => Ok(None),
     }
 }
 
@@ -174,7 +182,11 @@ async fn test_interpolate() {
     assert_eq!("foo_bar".to_string(), result.unwrap());
 }
 
-async fn run_action(action: &Action, client: &Client, mqtt_config: Option<&MqttConfig>) -> Result<(), TriggerError> {
+async fn run_action(
+    action: &Action,
+    client: &Client,
+    mqtt_config: Option<&MqttConfig>,
+) -> Result<(), TriggerError> {
     let url = interpolate_option_params(&action.url, &action.params).await?;
     let topic = interpolate_option_params(&action.topic, &action.params).await?;
     let payload = interpolate_option_params(&action.payload, &action.params).await?;
@@ -191,18 +203,11 @@ async fn run_action(action: &Action, client: &Client, mqtt_config: Option<&MqttC
         }
         (Method::Mqtt, _, Some(topic), Some(payload)) => {
             if let Some(mqtt_config) = mqtt_config {
-                let mqtt_options = MqttOptions::new("rumqtt-async", mqtt_config.host.as_str(), mqtt_config.port.unwrap_or(1883));
+                send_mqtt_message(mqtt_config, topic, payload).await?;
-
-                let (mqtt_client, mut event_loop) = AsyncClient::new(mqtt_options, 10);
-                mqtt_client.publish(topic, QoS::AtMostOnce, false, payload).await?;
-
-                let _ = tokio::time::timeout(Duration::from_secs(1), async move {
-                    loop {
-                        let _ = event_loop.poll().await;
-                    }
-                }).await;
             } else {
-                return Err(TriggerError::Configuration("mqtt trigger configured, but no mqtt server configured".to_string()));
+                return Err(TriggerError::Configuration(
+                    "mqtt trigger configured, but no mqtt server configured".to_string(),
+                ));
             }
         }
         _ => {}
@@ -210,3 +215,37 @@ async fn run_action(action: &Action, client: &Client, mqtt_config: Option<&MqttC
 
     Ok(())
 }
+
+async fn send_mqtt_message(
+    config: &MqttConfig,
+    topic: String,
+    payload: String,
+) -> Result<(), ClientError> {
+    let hostname = hostname::get()
+        .map(|os_str| os_str.to_string_lossy().to_string())
+        .unwrap_or_default();
+    let mut mqtt_options = MqttOptions::new(
+        format!("prometheus-edge-trigger-{}", hostname),
+        config.host.as_str(),
+        config.port.unwrap_or(1883),
+    );
+    if let (Some(username), Some(password)) = (&config.username, &config.password) {
+        mqtt_options.set_credentials(username, password);
+    }
+
+    let (mqtt_client, mut event_loop) = AsyncClient::new(mqtt_options, 10);
+    mqtt_client
+        .publish(topic, QoS::AtMostOnce, false, payload)
+        .await?;
+    mqtt_client.disconnect().await?;
+
+    let _ = tokio::time::timeout(Duration::from_secs(1), async move {
+        while let Ok(event) = event_loop.poll().await {
+            if matches!(event, Event::Outgoing(Outgoing::Disconnect)) {
+                break;
+            }
+        }
+    })
+    .await;
+    Ok(())
+}