Merge pull request #2033 from rina23q/improve/1918/support-configurab…

…le-tedge-mapper-mqtt-topics

Support configurable tedge-mapper MQTT topics

crates/common/tedge_config/src/tedge_config_cli/new.rs

@@ -352,6 +352,11 @@ define_tedge_config! {
         #[tedge_config(default(from_optional_key = "c8y.url"))]
         mqtt: HostPort<MQTT_TLS_PORT>,
 
+        /// Set of MQTT topics the Cumulocity mapper should subscribe to
+        #[tedge_config(example = "tedge/alarms/#,tedge/measurements/+")]
+        #[tedge_config(default(value = "tedge/measurements,tedge/measurements/+,tedge/alarms/+/+,tedge/alarms/+/+/+,tedge/events/+,tedge/events/+/+,tedge/health/+,tedge/health/+/+"))]
+        topics: TemplatesSet,
+
     },
 
     #[tedge_config(deprecated_name = "azure")] // for 0.1.0 compatibility
@@ -371,7 +376,12 @@ define_tedge_config! {
             #[tedge_config(example = "true")]
             #[tedge_config(default(value = true))]
             timestamp: bool,
-        }
+        },
+
+        /// Set of MQTT topics the Azure IoT mapper should subscribe to
+        #[tedge_config(example = "tedge/measurements,tedge/measurements/+")]
+        #[tedge_config(default(value = "tedge/measurements,tedge/measurements/+,tedge/health/+,tedge/health/+/+"))]
+        topics: TemplatesSet,
     },
 
     aws: {
@@ -390,7 +400,12 @@ define_tedge_config! {
             #[tedge_config(example = "true")]
             #[tedge_config(default(value = true))]
             timestamp: bool,
-        }
+        },
+
+        /// Set of MQTT topics the AWS IoT mapper should subscribe to
+        #[tedge_config(example = "tedge/measurements,tedge/measurements/+")]
+        #[tedge_config(default(value = "tedge/measurements,tedge/measurements/+,tedge/alarms/+/+,tedge/alarms/+/+/+,tedge/events/+,tedge/events/+/+,tedge/health/+,tedge/health/+/+"))]
+        topics: TemplatesSet,
     },
 
     mqtt: {

crates/core/tedge_mapper/src/aws/mapper.rs

@@ -3,12 +3,14 @@ use crate::core::mapper::start_basic_actors;
 use async_trait::async_trait;
 use aws_mapper_ext::converter::AwsConverter;
 use clock::WallClock;
+use mqtt_channel::TopicFilter;
 use std::path::Path;
 use tedge_actors::ConvertingActor;
 use tedge_actors::MessageSink;
 use tedge_actors::MessageSource;
 use tedge_actors::NoConfig;
 use tedge_config::new::TEdgeConfig;
+use tracing::warn;
 
 const AWS_MAPPER_NAME: &str = "tedge-mapper-aws";
 
@@ -32,7 +34,7 @@ impl TEdgeComponent for AwsMapper {
         let mut aws_converting_actor = ConvertingActor::builder(
             "AwsConverter",
             aws_converter,
-            AwsConverter::in_topic_filter(),
+            get_topic_filter(&tedge_config),
         );
 
         aws_converting_actor.add_input(&mut mqtt_actor);
@@ -44,3 +46,13 @@ impl TEdgeComponent for AwsMapper {
         Ok(())
     }
 }
+
+fn get_topic_filter(tedge_config: &TEdgeConfig) -> TopicFilter {
+    let mut topics = TopicFilter::empty();
+    for topic in tedge_config.aws.topics.0.clone() {
+        if topics.add(&topic).is_err() {
+            warn!("The configured topic '{topic}' is invalid and ignored.");
+        }
+    }
+    topics
+}

crates/core/tedge_mapper/src/az/mapper.rs

@@ -3,12 +3,14 @@ use crate::core::mapper::start_basic_actors;
 use async_trait::async_trait;
 use az_mapper_ext::converter::AzureConverter;
 use clock::WallClock;
+use mqtt_channel::TopicFilter;
 use std::path::Path;
 use tedge_actors::ConvertingActor;
 use tedge_actors::MessageSink;
 use tedge_actors::MessageSource;
 use tedge_actors::NoConfig;
 use tedge_config::new::TEdgeConfig;
+use tracing::warn;
 
 const AZURE_MAPPER_NAME: &str = "tedge-mapper-az";
 
@@ -36,11 +38,8 @@ impl TEdgeComponent for AzureMapper {
 
         let az_converter =
             AzureConverter::new(tedge_config.az.mapper.timestamp, Box::new(WallClock));
-        let mut az_converting_actor = ConvertingActor::builder(
-            "AzConverter",
-            az_converter,
-            AzureConverter::in_topic_filter(),
-        );
+        let mut az_converting_actor =
+            ConvertingActor::builder("AzConverter", az_converter, get_topic_filter(&tedge_config));
         az_converting_actor.add_input(&mut mqtt_actor);
 
         az_converting_actor.register_peer(NoConfig, mqtt_actor.get_sender());
@@ -51,3 +50,13 @@ impl TEdgeComponent for AzureMapper {
         Ok(())
     }
 }
+
+fn get_topic_filter(tedge_config: &TEdgeConfig) -> TopicFilter {
+    let mut topics = TopicFilter::empty();
+    for topic in tedge_config.az.topics.0.clone() {
+        if topics.add(&topic).is_err() {
+            warn!("The configured topic '{topic}' is invalid and ignored.");
+        }
+    }
+    topics
+}

crates/extensions/aws_mapper_ext/src/converter.rs

@@ -7,7 +7,6 @@ use tedge_actors::Converter;
 use tedge_api::serialize::ThinEdgeJsonSerializer;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::Topic;
-use tedge_mqtt_ext::TopicFilter;
 
 use crate::error::ConversionError;
 use crate::size_threshold::SizeThreshold;
@@ -37,21 +36,6 @@ impl AwsConverter {
         }
     }
 
-    pub fn in_topic_filter() -> TopicFilter {
-        vec![
-            "tedge/measurements",
-            "tedge/measurements/+",
-            "tedge/health",
-            "tedge/health/+",
-            "tedge/events/+",
-            "tedge/events/+/+",
-            "tedge/alarms/+/+",
-            "tedge/alarms/+/+/+",
-        ]
-        .try_into()
-        .unwrap()
-    }
-
     fn try_convert(&mut self, input: &MqttMessage) -> Result<Vec<MqttMessage>, ConversionError> {
         let default_timestamp = self.add_timestamp.then(|| self.clock.now());
 

crates/extensions/az_mapper_ext/src/converter.rs

@@ -7,13 +7,11 @@ use tedge_actors::Converter;
 use tedge_api::serialize::ThinEdgeJsonSerializer;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::Topic;
-use tedge_mqtt_ext::TopicFilter;
 
 const AZ_MQTT_THRESHOLD: usize = 1024 * 128;
 
 #[derive(Debug)]
 pub struct MapperConfig {
-    pub in_topic_filter: TopicFilter,
     pub out_topic: Topic,
     pub errors_topic: Topic,
 }
@@ -28,7 +26,6 @@ pub struct AzureConverter {
 impl AzureConverter {
     pub fn new(add_timestamp: bool, clock: Box<dyn Clock>) -> Self {
         let mapper_config = MapperConfig {
-            in_topic_filter: Self::in_topic_filter(),
             out_topic: Topic::new_unchecked("az/messages/events/"),
             errors_topic: Topic::new_unchecked("tedge/errors"),
         };
@@ -49,12 +46,6 @@ impl AzureConverter {
         }
     }
 
-    pub fn in_topic_filter() -> TopicFilter {
-        vec!["tedge/measurements", "tedge/measurements/+"]
-            .try_into()
-            .unwrap()
-    }
-
     fn try_convert(&mut self, input: &MqttMessage) -> Result<Vec<MqttMessage>, ConversionError> {
         self.size_threshold.validate(input)?;
         let default_timestamp = self.add_timestamp.then(|| self.clock.now());

crates/extensions/c8y_mapper_ext/src/actor.rs

@@ -180,10 +180,8 @@ impl C8yMapperBuilder {
 
         let box_builder = SimpleMessageBoxBuilder::new("CumulocityMapper", 16);
 
-        let mqtt_publisher = mqtt.connect_consumer(
-            C8yMapperConfig::subscriptions(&config.config_dir).unwrap(),
-            adapt(&box_builder.get_sender()),
-        );
+        let mqtt_publisher =
+            mqtt.connect_consumer(config.topics.clone(), adapt(&box_builder.get_sender()));
         let http_proxy = C8YHttpProxy::new("C8yMapper => C8YHttpProxy", http);
         let timer_sender = timer.connect_consumer(NoConfig, adapt(&box_builder.get_sender()));
         fs_watcher.register_peer(config.ops_dir.clone(), adapt(&box_builder.get_sender()));

crates/extensions/c8y_mapper_ext/src/config.rs

@@ -9,6 +9,7 @@ use tedge_config::new::ConfigNotSet;
 use tedge_config::new::ReadError;
 use tedge_config::new::TEdgeConfig;
 use tedge_mqtt_ext::TopicFilter;
+use tracing::log::warn;
 
 pub const MQTT_MESSAGE_SIZE_THRESHOLD: usize = 16184;
 
@@ -20,6 +21,7 @@ pub struct C8yMapperConfig {
     pub service_type: String,
     pub ops_dir: PathBuf,
     pub c8y_host: String,
+    pub topics: TopicFilter,
 }
 
 impl C8yMapperConfig {
@@ -30,6 +32,7 @@ impl C8yMapperConfig {
         device_type: String,
         service_type: String,
         c8y_host: String,
+        topics: TopicFilter,
     ) -> Self {
         let ops_dir = config_dir.join("operations").join("c8y");
 
@@ -41,6 +44,7 @@ impl C8yMapperConfig {
             service_type,
             ops_dir,
             c8y_host,
+            topics,
         }
     }
 
@@ -56,29 +60,29 @@ impl C8yMapperConfig {
         let service_type = tedge_config.service.ty.clone();
         let c8y_host = tedge_config.c8y_url().or_config_not_set()?.to_string();
 
+        // The topics to subscribe = default internal topics + user configurable external topics
+        let mut topics = Self::internal_topic_filter(&config_dir)?;
+        for topic in tedge_config.c8y.topics.0.clone() {
+            if topics.add(&topic).is_err() {
+                warn!("The configured topic '{topic}' is invalid and ignored.");
+            }
+        }
+
         Ok(C8yMapperConfig::new(
             config_dir,
             logs_path,
             device_id,
             device_type,
             service_type,
             c8y_host,
+            topics,
         ))
     }
 
-    pub fn subscriptions(config_dir: &Path) -> Result<TopicFilter, C8yMapperConfigError> {
-        let operations = Operations::try_new(config_dir.join("operations/c8y"))?;
+    pub fn internal_topic_filter(config_dir: &Path) -> Result<TopicFilter, C8yMapperConfigError> {
         let mut topic_filter: TopicFilter = vec![
-            "tedge/measurements",
-            "tedge/measurements/+",
-            "tedge/alarms/+/+",
-            "tedge/alarms/+/+/+",
             "c8y-internal/alarms/+/+",
             "c8y-internal/alarms/+/+/+",
-            "tedge/events/+",
-            "tedge/events/+/+",
-            "tedge/health/+",
-            "tedge/health/+/+",
             C8yTopic::SmartRestRequest.to_string().as_str(),
             ResponseTopic::SoftwareListResponse.as_str(),
             ResponseTopic::SoftwareUpdateResponse.as_str(),
@@ -87,12 +91,30 @@ impl C8yMapperConfig {
         .try_into()
         .expect("topics that mapper should subscribe to");
 
-        for topic in operations.topics_for_operations() {
-            topic_filter.add(&topic)?;
+        if let Ok(operations) = Operations::try_new(config_dir.join("operations").join("c8y")) {
+            for topic in operations.topics_for_operations() {
+                topic_filter.add(&topic)?;
+            }
         }
 
         Ok(topic_filter)
     }
+
+    /// List of all possible external topics that Cumulocity mapper addresses. For testing purpose.
+    pub fn default_external_topic_filter() -> TopicFilter {
+        vec![
+            "tedge/measurements",
+            "tedge/measurements/+",
+            "tedge/alarms/+/+",
+            "tedge/alarms/+/+/+",
+            "tedge/events/+",
+            "tedge/events/+/+",
+            "tedge/health/+",
+            "tedge/health/+/+",
+        ]
+        .try_into()
+        .unwrap()
+    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -102,6 +124,9 @@ pub enum C8yMapperConfigBuildError {
 
     #[error(transparent)]
     FromConfigNotSet(#[from] ConfigNotSet),
+
+    #[error(transparent)]
+    FromC8yMapperConfigError(#[from] C8yMapperConfigError),
 }
 
 #[derive(thiserror::Error, Debug)]

crates/extensions/c8y_mapper_ext/src/converter.rs

@@ -1707,13 +1707,15 @@ mod tests {
         CumulocityConverter,
         SimpleMessageBox<C8YRestRequest, C8YRestResult>,
     ) {
+        tmp_dir.dir("operations").dir("c8y");
+        tmp_dir.dir("tedge").dir("agent");
+
         let device_id = "test-device".into();
         let device_type = "test-device-type".into();
         let service_type = "service".into();
         let c8y_host = "test.c8y.io".into();
-
-        tmp_dir.dir("operations").dir("c8y");
-        tmp_dir.dir("tedge").dir("agent");
+        let mut topics = C8yMapperConfig::internal_topic_filter(&tmp_dir.to_path_buf()).unwrap();
+        topics.add_all(C8yMapperConfig::default_external_topic_filter());
 
         let config = C8yMapperConfig::new(
             tmp_dir.to_path_buf(),
@@ -1722,6 +1724,7 @@ mod tests {
             device_type,
             service_type,
             c8y_host,
+            topics,
         );
 
         let mqtt_builder: SimpleMessageBoxBuilder<MqttMessage, MqttMessage> =

crates/extensions/c8y_mapper_ext/src/tests.rs

@@ -1013,7 +1013,6 @@ EOF
 }
 
 #[tokio::test]
-
 async fn custom_operation_timeout_sigterm() {
     // The test assures SM Mapper correctly receives custom operation on `c8y/s/ds`
     // and executes the custom operation, it will timeout because it will not complete before given timeout
@@ -1086,7 +1085,6 @@ operation failed due to timeout: duration=1sEOF";
 }
 
 #[tokio::test]
-
 async fn custom_operation_timeout_sigkill() {
     // The test assures SM Mapper correctly receives custom operation on `c8y/s/ds`
     // and executes the custom operation, it will timeout because it will not complete before given timeout
@@ -1243,14 +1241,16 @@ async fn spawn_c8y_mapper_actor(
     SimpleMessageBox<NoMessage, FsWatchEvent>,
     SimpleMessageBox<SyncStart, SyncComplete>,
 ) {
+    if init {
+        config_dir.dir("operations").dir("c8y");
+    }
+
     let device_name = "test-device".into();
     let device_type = "test-device-type".into();
     let service_type = "service".into();
     let c8y_host = "test.c8y.io".into();
-
-    if init {
-        config_dir.dir("operations").dir("c8y");
-    }
+    let mut topics = C8yMapperConfig::internal_topic_filter(config_dir.path()).unwrap();
+    topics.add_all(C8yMapperConfig::default_external_topic_filter());
 
     let config = C8yMapperConfig::new(
         config_dir.to_path_buf(),
@@ -1259,6 +1259,7 @@ async fn spawn_c8y_mapper_actor(
         device_type,
         service_type,
         c8y_host,
+        topics,
     );
 
     let mut mqtt_builder: SimpleMessageBoxBuilder<MqttMessage, MqttMessage> =