feat: tedge connect to c8y mqtt service endpoint

Author: albinsuresh

crates/common/tedge_config/src/tedge_toml/models/mod.rs

@@ -24,6 +24,7 @@ use strum::Display;
 
 pub const HTTPS_PORT: u16 = 443;
 pub const MQTT_TLS_PORT: u16 = 8883;
+pub const MQTT_SVC_TLS_PORT: u16 = 9883;
 
 pub use self::apt_config::*;
 pub use self::auto::*;

crates/common/tedge_config/src/tedge_toml/tedge_config.rs

@@ -20,6 +20,7 @@ use super::models::SoftwareManagementApiFlag;
 use super::models::TemplatesSet;
 use super::models::TopicPrefix;
 use super::models::HTTPS_PORT;
+use super::models::MQTT_SVC_TLS_PORT;
 use super::models::MQTT_TLS_PORT;
 use super::tedge_config_location::TEdgeConfigLocation;
 use crate::models::AbsolutePath;
@@ -226,6 +227,10 @@ define_tedge_config! {
         #[tedge_config(reader(private))]
         url: ConnectUrl,
 
+        /// Cumulocity tenant ID
+        #[tedge_config(example = "t12345678")]
+        tenant_id: String,
+
         /// The path where Cumulocity root certificate(s) are stored
         #[tedge_config(note = "The value can be a directory path as well as the path of the certificate file.")]
         #[tedge_config(example = "/etc/tedge/c8y-trusted-root-certificates.pem", default(function = "default_root_cert_path"))]
@@ -433,6 +438,23 @@ define_tedge_config! {
             #[tedge_config(example = "60m", default(from_str = "60m"))]
             interval: SecondsOrHumanTime,
         },
+
+        mqtt_service: {
+            /// Wheather to connect to the MQTT service endpoint or not
+            #[tedge_config(example = "true", default(value = false))]
+            enabled: bool,
+
+            /// MQTT service endpoint for the Cumulocity tenant, with optional port.
+            #[tedge_config(example = "mqtt.your-tenant.cumulocity.com:1234")]
+            #[tedge_config(default(from_optional_key = "c8y.url"))]
+            url: HostPort<MQTT_SVC_TLS_PORT>,
+
+            /// The topic prefix that will be used for the Cumulocity MQTT service endpoint connection.
+            /// For instance, if set to "c8y-mqtt", then messages published to `c8y-mqtt/xyz`
+            /// will be forwarded to the MQTT service endpoint on the `xyz` topic
+            #[tedge_config(example = "c8y-mqtt", default(function = "c8y_mqtt_service_topic_prefix"))]
+            topic_prefix: TopicPrefix,
+        }
     },
 
     #[tedge_config(deprecated_name = "azure")] // for 0.1.0 compatibility
@@ -1114,6 +1136,10 @@ fn c8y_topic_prefix() -> TopicPrefix {
     TopicPrefix::try_new("c8y").unwrap()
 }
 
+fn c8y_mqtt_service_topic_prefix() -> TopicPrefix {
+    TopicPrefix::try_new("c8y-mqtt").unwrap()
+}
+
 fn az_topic_prefix() -> TopicPrefix {
     TopicPrefix::try_new("az").unwrap()
 }

crates/common/tedge_config/src/tedge_toml/tedge_config/append_remove.rs

@@ -40,6 +40,7 @@ impl_append_remove_for_single_value!(
     ConnectUrl,
     HostPort<HTTPS_PORT>,
     HostPort<MQTT_TLS_PORT>,
+    HostPort<MQTT_SVC_TLS_PORT>,
     bool,
     IpAddr,
     u16,

crates/core/tedge/src/bridge/c8y.rs

@@ -1,6 +1,8 @@
 use super::config::ProxyWrapper;
 use super::BridgeConfig;
 use crate::bridge::config::BridgeLocation;
+use crate::ConfigError;
+use c8y_api::http_proxy::read_c8y_credentials;
 use camino::Utf8PathBuf;
 use std::borrow::Cow;
 use std::process::Command;
@@ -13,8 +15,10 @@ use tedge_config::models::AutoFlag;
 use tedge_config::models::HostPort;
 use tedge_config::models::TemplatesSet;
 use tedge_config::models::TopicPrefix;
+use tedge_config::models::MQTT_SVC_TLS_PORT;
 use tedge_config::models::MQTT_TLS_PORT;
 use tedge_config::tedge_toml::ProfileName;
+use tedge_config::TEdgeConfig;
 use which::which;
 
 #[derive(Debug)]
@@ -201,6 +205,171 @@ impl From<BridgeConfigC8yParams> for BridgeConfig {
     }
 }
 
+#[derive(Debug)]
+pub struct BridgeConfigC8yMqttServiceParams {
+    pub mqtt_host: HostPort<MQTT_SVC_TLS_PORT>,
+    pub config_file: Cow<'static, str>,
+    pub tenant_id: String,
+    pub remote_clientid: String,
+    pub remote_username: Option<String>,
+    pub remote_password: Option<String>,
+    pub bridge_root_cert_path: Utf8PathBuf,
+    pub bridge_certfile: Utf8PathBuf,
+    pub bridge_keyfile: Utf8PathBuf,
+    pub include_local_clean_session: AutoFlag,
+    pub bridge_location: BridgeLocation,
+    pub topic_prefix: TopicPrefix,
+    pub profile_name: Option<ProfileName>,
+    pub mqtt_schema: MqttSchema,
+    pub keepalive_interval: Duration,
+}
+
+impl TryFrom<(&TEdgeConfig, Option<&ProfileName>)> for BridgeConfigC8yMqttServiceParams {
+    type Error = ConfigError;
+
+    fn try_from(value: (&TEdgeConfig, Option<&ProfileName>)) -> Result<Self, Self::Error> {
+        let (config, profile) = value;
+
+        let bridge_location = match config.mqtt.bridge.built_in {
+            true => BridgeLocation::BuiltIn,
+            false => BridgeLocation::Mosquitto,
+        };
+        let mqtt_schema = MqttSchema::with_root(config.mqtt.topic_root.clone());
+        let c8y_config = config.c8y.try_get(profile)?;
+
+        let (remote_username, remote_password) =
+            match c8y_config.auth_method.to_type(&c8y_config.credentials_path) {
+                AuthType::Certificate => (None, None),
+                AuthType::Basic => {
+                    let (username, password) = read_c8y_credentials(&c8y_config.credentials_path)?;
+                    (Some(username), Some(password))
+                }
+            };
+
+        let config_file = if let Some(profile_name) = &profile {
+            format!("c8y-mqtt-svc@{profile_name}-bridge.conf")
+        } else {
+            "c8y-mqtt-svc-bridge.conf".to_string()
+        };
+        let params = BridgeConfigC8yMqttServiceParams {
+            mqtt_host: c8y_config.mqtt_service.url.or_config_not_set()?.clone(),
+            config_file: config_file.into(),
+            tenant_id: c8y_config.tenant_id.or_config_not_set()?.clone(),
+            bridge_root_cert_path: c8y_config.root_cert_path.clone().into(),
+            remote_clientid: c8y_config.device.id()?.clone(),
+            remote_username,
+            remote_password,
+            bridge_certfile: c8y_config.device.cert_path.clone().into(),
+            bridge_keyfile: c8y_config.device.key_path.clone().into(),
+            include_local_clean_session: c8y_config.bridge.include.local_cleansession.clone(),
+            bridge_location,
+            topic_prefix: c8y_config.mqtt_service.topic_prefix.clone(),
+            profile_name: profile.cloned(),
+            mqtt_schema,
+            keepalive_interval: c8y_config.bridge.keepalive_interval.duration(),
+        };
+
+        Ok(params)
+    }
+}
+
+impl From<BridgeConfigC8yMqttServiceParams> for BridgeConfig {
+    fn from(params: BridgeConfigC8yMqttServiceParams) -> Self {
+        let BridgeConfigC8yMqttServiceParams {
+            mqtt_host,
+            config_file,
+            bridge_root_cert_path,
+            tenant_id,
+            mut remote_username,
+            remote_password,
+            remote_clientid,
+            bridge_certfile,
+            bridge_keyfile,
+            include_local_clean_session,
+            bridge_location,
+            topic_prefix,
+            profile_name,
+            mqtt_schema,
+            keepalive_interval,
+        } = params;
+
+        let address = mqtt_host
+            .to_string()
+            .parse::<HostPort<MQTT_TLS_PORT>>()
+            .expect("MQTT service address must be in the expected format");
+
+        let topics: Vec<String> = vec![
+            // Outgoing
+            format!(r#"# out 1 {topic_prefix}/"#),
+            // Incoming
+            format!(r#"$debug/$error in 1 {topic_prefix}/"#),
+        ];
+
+        let auth_type = if remote_password.is_some() {
+            AuthType::Basic
+        } else {
+            AuthType::Certificate
+        };
+
+        // When cert based auth is used, provide the tenant id as the username
+        if auth_type == AuthType::Certificate {
+            remote_username = Some(tenant_id.clone());
+        }
+
+        let (include_local_clean_session, mosquitto_version) = match include_local_clean_session {
+            AutoFlag::True => (true, None),
+            AutoFlag::False => (false, None),
+            AutoFlag::Auto => is_mosquitto_version_above_2(),
+        };
+
+        let service_name = format!("mosquitto-{topic_prefix}-bridge");
+        let health = mqtt_schema.topic_for(
+            &EntityTopicId::default_main_service(&service_name).unwrap(),
+            &Channel::Health,
+        );
+
+        Self {
+            cloud_name: "c8y-mqtt".into(),
+            config_file,
+            connection: if let Some(profile) = &profile_name {
+                format!("edge_to_c8y_mqtt_service@{profile}")
+            } else {
+                "edge_to_c8y_mqtt_service".into()
+            },
+            address,
+            remote_username,
+            remote_password,
+            bridge_root_cert_path,
+            remote_clientid,
+            local_clientid: if let Some(profile) = &profile_name {
+                format!("CumulocityMqttService@{profile}")
+            } else {
+                "CumulocityMqttService".into()
+            },
+            bridge_certfile,
+            bridge_keyfile,
+            use_mapper: true,
+            use_agent: true,
+            try_private: false,
+            start_type: "automatic".into(),
+            clean_session: true,
+            include_local_clean_session,
+            local_clean_session: false,
+            notifications: true,
+            notifications_local_only: true,
+            notification_topic: health.name,
+            bridge_attempt_unsubscribe: false,
+            topics,
+            bridge_location,
+            connection_check_attempts: 3,
+            auth_type,
+            mosquitto_version,
+            keepalive_interval,
+            proxy: None,
+        }
+    }
+}
+
 /// Return whether or not mosquitto version is >= 2.0.0
 ///
 /// As mosquitto doesn't provide a `--version` flag, this is a guess.

crates/core/tedge/src/cli/connect/command.rs

@@ -2,6 +2,7 @@
 use crate::bridge::aws::BridgeConfigAwsParams;
 #[cfg(feature = "azure")]
 use crate::bridge::azure::BridgeConfigAzureParams;
+use crate::bridge::c8y::BridgeConfigC8yMqttServiceParams;
 #[cfg(feature = "c8y")]
 use crate::bridge::c8y::BridgeConfigC8yParams;
 use crate::bridge::BridgeConfig;
@@ -746,6 +747,10 @@ pub fn bridge_config(
                     }
                 };
 
+            if c8y_config.mqtt_service.enabled && remote_password.is_none() {
+                // If the MQTT service connection is enabled and cert based auth is used, then tenant_id must be set
+                c8y_config.tenant_id.or_config_not_set()?;
+            }
             let params = BridgeConfigC8yParams {
                 mqtt_host: c8y_config.mqtt.or_config_not_set()?.clone(),
                 config_file: cloud.bridge_config_filename(),
@@ -841,10 +846,19 @@ impl ConnectCommand {
         }
 
         if bridge_config.bridge_location == BridgeLocation::Mosquitto {
-            if let Err(err) = write_bridge_config_to_file(tedge_config, bridge_config).await {
-                // We want to preserve previous errors and therefore discard result of this function.
-                let _ = clean_up(tedge_config, bridge_config);
-                return Err(err.into());
+            write_mosquitto_bridge_config_file(tedge_config, bridge_config).await?;
+
+            if let Cloud::C8y(profile_name) = &self.cloud {
+                let c8y_config = tedge_config.c8y.try_get(profile_name.as_deref())?;
+                if c8y_config.mqtt_service.enabled {
+                    let config_params = BridgeConfigC8yMqttServiceParams::try_from((
+                        tedge_config,
+                        profile_name.as_deref(),
+                    ))?;
+                    let mqtt_svc_bridge_config = BridgeConfig::from(config_params);
+                    write_mosquitto_bridge_config_file(tedge_config, &mqtt_svc_bridge_config)
+                        .await?;
+                }
             }
         } else {
             use_built_in_bridge(tedge_config, bridge_config).await?;
@@ -1040,6 +1054,19 @@ async fn write_generic_mosquitto_config_to_file(
     Ok(())
 }
 
+async fn write_mosquitto_bridge_config_file(
+    tedge_config: &TEdgeConfig,
+    bridge_config: &BridgeConfig,
+) -> Result<(), ConnectError> {
+    if let Err(err) = write_bridge_config_to_file(tedge_config, bridge_config).await {
+        // We want to preserve previous errors and therefore discard result of this function.
+        // let _ = clean_up(tedge_config, bridge_config);
+        return Err(err);
+    }
+
+    Ok(())
+}
+
 async fn write_bridge_config_to_file(
     config: &TEdgeConfig,
     bridge_config: &BridgeConfig,