Make more consistent tedge mqtt logging

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

Author: didier-wenzek

crates/common/mqtt_channel/src/config.rs

@@ -267,7 +267,7 @@ impl Config {
         &mut self,
         ca_file: impl AsRef<Path>,
     ) -> Result<&mut Self, certificate::CertificateError> {
-        debug!("Using CA certificate: {}", ca_file.as_ref().display());
+        debug!(target: "MQTT", "Using CA certificate: {}", ca_file.as_ref().display());
         let authentication_config = self.broker.authentication.get_or_insert(Default::default());
         let cert_store = &mut authentication_config.cert_store;
 
@@ -282,7 +282,7 @@ impl Config {
         &mut self,
         ca_dir: impl AsRef<Path>,
     ) -> Result<&mut Self, certificate::CertificateError> {
-        debug!("Using CA directory: {}", ca_dir.as_ref().display());
+        debug!(target: "MQTT", "Using CA directory: {}", ca_dir.as_ref().display());
         let authentication_config = self.broker.authentication.get_or_insert(Default::default());
         let cert_store = &mut authentication_config.cert_store;
 
@@ -300,8 +300,8 @@ impl Config {
         cert_file: P,
         key_file: P,
     ) -> Result<&mut Self, CertificateError> {
-        debug!("Using client certificate: {}", cert_file.as_ref().display());
-        debug!("Using client private key: {}", key_file.as_ref().display());
+        debug!(target: "MQTT", "Using client certificate: {}", cert_file.as_ref().display());
+        debug!(target: "MQTT", "Using client private key: {}", key_file.as_ref().display());
         let cert_chain = parse_root_certificate::read_cert_chain(cert_file)?;
         let key = parse_root_certificate::read_pvt_key(key_file)?;
 

crates/common/mqtt_channel/src/connection.rs

@@ -10,6 +10,7 @@ use futures::SinkExt;
 use futures::StreamExt;
 use log::error;
 use log::info;
+use log::warn;
 use rumqttc::AsyncClient;
 use rumqttc::Event;
 use rumqttc::EventLoop;
@@ -132,17 +133,17 @@ impl Connection {
         const SECURE_MQTT_PORT: u16 = 8883;
 
         if config.broker.port == INSECURE_MQTT_PORT && config.broker.authentication.is_some() {
-            eprintln!("WARNING: Connecting on port 1883 for insecure MQTT using a TLS connection");
+            warn!(target: "MQTT", "Connecting on port 1883 for insecure MQTT using a TLS connection");
         }
         if config.broker.port == SECURE_MQTT_PORT && config.broker.authentication.is_none() {
-            eprintln!("WARNING: Connecting on port 8883 for secure MQTT without a CA file");
+            warn!(target: "MQTT", "Connecting on port 8883 for secure MQTT without a CA file");
         }
 
         let mqtt_options = config.rumqttc_options()?;
         let (mqtt_client, mut event_loop) = AsyncClient::new(mqtt_options, config.queue_capacity);
 
-        info!(
-            "MQTT connecting to broker: host={}:{}, session_name={:?}",
+        info!(target: "MQTT",
+            "Connecting to broker: host={}:{}, session_name={:?}",
             config.broker.host, config.broker.port, config.session_name
         );
 
@@ -152,7 +153,7 @@ impl Connection {
                     if let Some(err) = MqttError::maybe_connection_error(&ack) {
                         return Err(err);
                     };
-                    info!("MQTT connection established");
+                    info!(target: "MQTT", "Connection established");
 
                     let subscriptions = config.subscriptions.filters();
 
@@ -175,16 +176,16 @@ impl Connection {
                     // Messages can be received before a sub ack
                     // Errors on send are ignored: it just means the client has closed the receiving channel.
                     if msg.payload.len() > config.max_packet_size {
-                        error!("Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}", 
+                        error!(target: "MQTT", "Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}",
                             msg.topic, msg.payload.len(), config.max_packet_size);
                         continue;
                     }
                     let _ = message_sender.send(msg.into()).await;
                 }
 
                 Err(err) => {
-                    error!(
-                        "MQTT: failed to connect to broker at '{host}:{port}': {err}",
+                    error!(target: "MQTT",
+                        "Failed to connect to broker at '{host}:{port}': {err}",
                         host = config.broker.host,
                         port = config.broker.port
                     );
@@ -244,7 +245,7 @@ impl Connection {
             match event {
                 Ok(Event::Incoming(Packet::Publish(msg))) => {
                     if msg.payload.len() > config.max_packet_size {
-                        error!("Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}", 
+                        error!(target: "MQTT", "Dropping message received on topic {} with payload size {} that exceeds the maximum packet size of {}",
                             msg.topic, msg.payload.len(), config.max_packet_size);
                         continue;
                     }
@@ -255,9 +256,9 @@ impl Connection {
 
                 Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                     if let Some(err) = MqttError::maybe_connection_error(&ack) {
-                        error!("MQTT connection Error {err}");
+                        error!(target: "MQTT", "Connection Error {err}");
                     } else {
-                        info!("MQTT connection re-established");
+                        info!(target: "MQTT", "Connection re-established");
                         if let Some(ref imsg_fn) = config.initial_message {
                             // publish the initial message on connect
                             let message = imsg_fn.new_init_message();
@@ -292,7 +293,7 @@ impl Connection {
                 }
 
                 Err(err) => {
-                    error!("MQTT connection error: {err}");
+                    error!(target: "MQTT", "Connection error: {err}");
 
                     // Errors on send are ignored: it just means the client has closed the receiving channel.
                     let _ = error_sender.send(err.into()).await;
@@ -307,7 +308,7 @@ impl Connection {
         // to make sure the disconnect is effective
         loop {
             if (event_loop.poll().await).is_err() {
-                info!("MQTT connection closed");
+                info!(target: "MQTT", "Connection closed");
                 break;
             }
         }

crates/common/mqtt_channel/src/session.rs

@@ -1,5 +1,6 @@
 use crate::Config;
 use crate::MqttError;
+use log::error;
 use log::warn;
 use rumqttc::AsyncClient;
 use rumqttc::ConnectReturnCode;
@@ -44,7 +45,7 @@ pub async fn init_session(config: &Config) -> Result<(), MqttError> {
             Err(err) => match err {
                 rumqttc::ConnectionError::ConnectionRefused(ConnectReturnCode::Success) => {}
                 _ => {
-                    warn!("{}", MqttError::from_connection_error(err));
+                    warn!(target: "MQTT", "{}", MqttError::from_connection_error(err));
                     break;
                 }
             },
@@ -85,7 +86,7 @@ pub async fn clear_session(config: &Config) -> Result<(), MqttError> {
             }
 
             Err(err) => {
-                eprintln!("Connection Error {}", err);
+                error!(target: "MQTT", "Connection Error {}", err);
                 break;
             }
             _ => (),

crates/core/tedge/src/cli/mqtt/publish.rs

@@ -71,7 +71,7 @@ async fn publish(cmd: &MqttPublishCommand) -> Result<(), anyhow::Error> {
     {
         Ok(Ok(())) => (),
         Ok(err) => err?,
-        Err(signal) => info!("{signal:?}"),
+        Err(signal) => info!(target: "MQTT", "{signal:?}"),
     }
     mqtt.close().await;
 

crates/core/tedge/src/cli/mqtt/subscribe.rs

@@ -7,11 +7,11 @@ use mqtt_channel::TopicFilter;
 use std::time::Duration;
 use tedge_config::tedge_toml::MqttAuthClientConfig;
 use tokio::io::AsyncWriteExt;
+use tracing::error;
 use tracing::info;
 
 const DEFAULT_QUEUE_CAPACITY: usize = 10;
 use super::MAX_PACKET_SIZE;
-use crate::error;
 
 pub struct MqttSubscribeCommand {
     pub host: String,
@@ -76,7 +76,7 @@ async fn subscribe(cmd: &MqttSubscribeCommand) -> Result<(), anyhow::Error> {
             Ok(Some(message)) => message,
             Ok(None) => break,
             Err(signal) => {
-                info!("{signal:?}");
+                info!(target: "MQTT", "{signal:?}");
                 break;
             }
         };
@@ -92,11 +92,11 @@ async fn subscribe(cmd: &MqttSubscribeCommand) -> Result<(), anyhow::Error> {
                 let _ = stdout.flush().await;
                 n_messages += 1;
                 if matches!(cmd.count, Some(count) if count > 0 && n_messages >= count) {
-                    info!("Received {n_messages} message/s");
+                    info!(target: "MQTT", "Received {n_messages} message/s");
                     break;
                 }
             }
-            Err(err) => error!("{err}"),
+            Err(err) => error!(target: "MQTT", "{err}"),
         }
     }