Merge pull request #3834 from albinsuresh/feat/tedge-agent-sync-signal

feat: sync signal support for agent

Author: albinsuresh

crates/core/tedge_agent/src/agent.rs

@@ -47,7 +47,9 @@ use tedge_api::entity_store::EntityRegistrationMessage;
 use tedge_api::mqtt_topics::DeviceTopicId;
 use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::MqttSchema;
+use tedge_api::mqtt_topics::OperationType;
 use tedge_api::mqtt_topics::Service;
+use tedge_api::mqtt_topics::ServiceTopicId;
 use tedge_api::path::DataDir;
 use tedge_api::EntityStore;
 use tedge_config::tedge_toml::TEdgeConfigReaderService;
@@ -90,6 +92,7 @@ pub(crate) struct AgentConfig {
     pub state_dir: Utf8PathBuf,
     pub operations_dir: Utf8PathBuf,
     pub mqtt_device_topic_id: EntityTopicId,
+    pub service_topic_id: ServiceTopicId,
     pub mqtt_topic_root: Arc<str>,
     pub tedge_http_host: Arc<str>,
     pub service: TEdgeConfigReaderService,
@@ -116,11 +119,13 @@ impl AgentConfig {
             .mqtt_topic_root
             .unwrap_or(tedge_config.mqtt.topic_root.clone().into());
 
-        let mqtt_device_topic_id = cliopts
+        let mqtt_device_topic_id: EntityTopicId = cliopts
             .mqtt_device_topic_id
             .unwrap_or(tedge_config.mqtt.device_topic_id.clone().into())
             .parse()
             .context("Could not parse the device MQTT topic")?;
+        let service_topic_id = mqtt_device_topic_id.to_default_service_topic_id("tedge-agent")
+            .with_context(|| format!("Device topic id {} currently needs default scheme, e.g: 'device/DEVICE_NAME//'", mqtt_device_topic_id))?;
 
         let mqtt_session_name = format!("{TEDGE_AGENT}#{mqtt_topic_root}/{mqtt_device_topic_id}");
 
@@ -157,6 +162,7 @@ impl AgentConfig {
         let operation_config = OperationConfig::from_tedge_config(
             mqtt_topic_root.to_string(),
             &mqtt_device_topic_id,
+            service_topic_id.clone().into(),
             &tedge_config,
         )
         .await?;
@@ -213,6 +219,7 @@ impl AgentConfig {
             state_dir,
             mqtt_topic_root,
             mqtt_device_topic_id,
+            service_topic_id,
             tedge_http_host,
             identity,
             cloud_root_certs,
@@ -307,8 +314,7 @@ impl Agent {
         // Health actor
         // TODO: take a user-configurable service topic id
         let device_topic_id = self.config.mqtt_device_topic_id.clone();
-        let service_topic_id = device_topic_id.to_default_service_topic_id("tedge-agent")
-            .with_context(|| format!("Device topic id {} currently needs default scheme, e.g: 'device/DEVICE_NAME//'", device_topic_id))?;
+        let service_topic_id = self.config.service_topic_id.clone();
         let service = Service {
             service_topic_id: service_topic_id.clone(),
             device_topic_id: DeviceTopicId::new(device_topic_id.clone()),
@@ -384,7 +390,7 @@ impl Agent {
             )
             .await?;
             converter_actor_builder.register_builtin_operation(&mut log_actor);
-            converter_actor_builder.register_sync_signal_sink(&log_actor);
+            converter_actor_builder.register_sync_signal_sink(OperationType::LogUpload, &log_actor);
             Some(log_actor)
         } else {
             None

crates/core/tedge_agent/src/operation_workflows/actor.rs

@@ -22,6 +22,7 @@ use tedge_api::mqtt_topics::EntityTopicError;
 use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::MqttSchema;
 use tedge_api::mqtt_topics::OperationType;
+use tedge_api::mqtt_topics::SignalType;
 use tedge_api::workflow::extract_json_output;
 use tedge_api::workflow::CommandBoard;
 use tedge_api::workflow::CommandId;
@@ -139,11 +140,55 @@ impl WorkflowActor {
     /// but also from *this* actor as all its state transitions are published over MQTT.
     /// Only the former will be actually processed with [Self::process_command_update].
     async fn process_mqtt_message(&mut self, message: MqttMessage) -> Result<(), RuntimeError> {
-        let Ok((operation, cmd_id)) = self.extract_command_identifiers(&message.topic.name) else {
-            log::error!("Unknown command channel: {}", &message.topic.name);
+        let Ok((_, channel)) = self.mqtt_schema.entity_channel_of(&message.topic) else {
+            log::error!("Unknown topic: {}", &message.topic.name);
             return Ok(());
         };
+        match channel {
+            Channel::Command { operation, cmd_id } => {
+                self.process_command_message(message, operation, cmd_id)
+                    .await
+            }
+            Channel::Signal { signal_type } => {
+                self.process_signal_message(message, signal_type).await
+            }
+            _ => {
+                error!("Unsupported channel: {}", channel);
+                Ok(())
+            }
+        }
+    }
 
+    async fn process_signal_message(
+        &mut self,
+        _message: MqttMessage,
+        signal_type: SignalType,
+    ) -> Result<(), RuntimeError> {
+        match signal_type {
+            SignalType::Sync => {
+                info!("Received sync signal, requesting all builtin actors to sync");
+                self.sync_signal_dispatcher.sync_all().await?;
+            }
+            SignalType::SyncOperation(operation) => {
+                info!(
+                    "Received sync signal for {}, requesting the corresponding actor to sync",
+                    operation
+                );
+                self.sync_signal_dispatcher.sync(operation).await?;
+            }
+            SignalType::Custom(_) => {
+                // Custom signal types are not handled yet
+            }
+        }
+        Ok(())
+    }
+
+    async fn process_command_message(
+        &mut self,
+        message: MqttMessage,
+        operation: OperationType,
+        cmd_id: String,
+    ) -> Result<(), RuntimeError> {
         let Ok(state) = GenericCommandState::from_command_message(&message) else {
             log::error!("Invalid command payload: {}", &message.topic.name);
             return Ok(());
@@ -430,7 +475,7 @@ impl WorkflowActor {
         new_state: GenericCommandState,
     ) -> Result<(), RuntimeError> {
         if new_state.is_finished() {
-            self.sync_dependent_actors(&new_state).await?;
+            self.sync_listener_actors(&new_state).await?;
             self.finalize_builtin_command_update(new_state).await?;
 
             Ok(())
@@ -463,13 +508,13 @@ impl WorkflowActor {
         self.process_command_update(adapted_state).await
     }
 
-    async fn sync_dependent_actors(
+    async fn sync_listener_actors(
         &mut self,
         command: &GenericCommandState,
     ) -> Result<(), RuntimeError> {
         if let Some(command) = command.operation() {
             self.sync_signal_dispatcher
-                .send(command.as_str().into())
+                .sync_listener(command.as_str().into())
                 .await?;
         }
         Ok(())

crates/core/tedge_agent/src/operation_workflows/builder.rs

@@ -24,10 +24,11 @@ use tedge_actors::RuntimeRequestSink;
 use tedge_actors::Service;
 use tedge_actors::UnboundedLoggingReceiver;
 use tedge_api::commands::CmdMetaSyncSignal;
-use tedge_api::mqtt_topics::ChannelFilter::AnyCommand;
+use tedge_api::mqtt_topics::ChannelFilter;
 use tedge_api::mqtt_topics::EntityFilter;
 use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::MqttSchema;
+use tedge_api::mqtt_topics::OperationType;
 use tedge_api::workflow::GenericCommandData;
 use tedge_api::workflow::GenericCommandState;
 use tedge_api::workflow::OperationName;
@@ -73,7 +74,11 @@ impl WorkflowActorBuilder {
 
         let mqtt_publisher = mqtt_actor.get_sender();
         mqtt_actor.connect_sink(
-            Self::subscriptions(&config.mqtt_schema, &config.device_topic_id),
+            Self::subscriptions(
+                &config.mqtt_schema,
+                &config.device_topic_id,
+                &config.service_topic_id,
+            ),
             &input_sender,
         );
         let mqtt_publisher = LoggingSender::new("MqttPublisher".into(), mqtt_publisher);
@@ -110,19 +115,36 @@ impl WorkflowActorBuilder {
     }
 
     /// Register an actor to receive sync signals on completion of other commands
-    pub fn register_sync_signal_sink<OperationActor>(&mut self, actor: &OperationActor)
-    where
+    pub fn register_sync_signal_sink<OperationActor>(
+        &mut self,
+        op_type: OperationType,
+        actor: &OperationActor,
+    ) where
         OperationActor: MessageSink<CmdMetaSyncSignal> + SyncOnCommand,
     {
         let sender = actor.get_sender();
+        self.sync_signal_dispatcher
+            .register_operation_handler(op_type, sender.sender_clone());
         for operation in actor.sync_on_commands() {
             self.sync_signal_dispatcher
-                .register_sync_signal_sender(operation, sender.sender_clone());
+                .register_operation_listener(operation, sender.sender_clone());
         }
     }
 
-    pub fn subscriptions(mqtt_schema: &MqttSchema, device_topic_id: &EntityTopicId) -> TopicFilter {
-        mqtt_schema.topics(EntityFilter::Entity(device_topic_id), AnyCommand)
+    pub fn subscriptions(
+        mqtt_schema: &MqttSchema,
+        device_topic_id: &EntityTopicId,
+        service_topic_id: &EntityTopicId,
+    ) -> TopicFilter {
+        let mut topics = mqtt_schema.topics(
+            EntityFilter::Entity(device_topic_id),
+            ChannelFilter::AnyCommand,
+        );
+        topics.add_all(mqtt_schema.topics(
+            EntityFilter::Entity(service_topic_id),
+            ChannelFilter::AnySignal,
+        ));
+        topics
     }
 }
 

crates/core/tedge_agent/src/operation_workflows/config.rs

@@ -7,6 +7,7 @@ use tedge_config::TEdgeConfig;
 pub struct OperationConfig {
     pub mqtt_schema: MqttSchema,
     pub device_topic_id: EntityTopicId,
+    pub service_topic_id: EntityTopicId,
     pub log_dir: Utf8PathBuf,
     pub config_dir: Utf8PathBuf,
     pub state_dir: Utf8PathBuf,
@@ -17,13 +18,15 @@ impl OperationConfig {
     pub async fn from_tedge_config(
         topic_root: String,
         device_topic_id: &EntityTopicId,
+        service_topic_id: EntityTopicId,
         tedge_config: &TEdgeConfig,
     ) -> Result<OperationConfig, tedge_config::TEdgeConfigError> {
         let config_dir = tedge_config.root_dir();
 
         Ok(OperationConfig {
             mqtt_schema: MqttSchema::with_root(topic_root),
             device_topic_id: device_topic_id.clone(),
+            service_topic_id,
             log_dir: tedge_config.logs.path.join("agent"),
             config_dir: config_dir.to_owned(),
             state_dir: tedge_config.agent.state.path.clone().into(),

crates/core/tedge_agent/src/operation_workflows/message_box.rs

@@ -49,26 +49,59 @@ impl CommandDispatcher {
 
 #[derive(Default)]
 pub(crate) struct SyncSignalDispatcher {
-    senders: HashMap<OperationType, Vec<DynSender<CmdMetaSyncSignal>>>,
+    op_handler_senders: HashMap<OperationType, DynSender<CmdMetaSyncSignal>>,
+    op_listener_senders: HashMap<OperationType, Vec<DynSender<CmdMetaSyncSignal>>>,
 }
 
 impl SyncSignalDispatcher {
-    /// Register where to send sync signals for the given command type
-    pub(crate) fn register_sync_signal_sender(
+    /// Register sender for the operation handler to receive sync signals for that operation
+    pub(crate) fn register_operation_handler(
         &mut self,
         operation: OperationType,
         sender: DynSender<CmdMetaSyncSignal>,
     ) {
-        self.senders.entry(operation).or_default().push(sender);
+        self.op_handler_senders.insert(operation, sender);
     }
 
-    pub(crate) async fn send(&mut self, operation: OperationType) -> Result<(), ChannelError> {
-        let Some(senders) = self.senders.get_mut(&operation) else {
+    /// Register senders for operation listeners that must be notified when the given operation completes
+    pub(crate) fn register_operation_listener(
+        &mut self,
+        operation: OperationType,
+        sender: DynSender<CmdMetaSyncSignal>,
+    ) {
+        self.op_listener_senders
+            .entry(operation)
+            .or_default()
+            .push(sender);
+    }
+
+    /// Send sync signal to all registered listeners for the given operation
+    pub(crate) async fn sync_listener(
+        &mut self,
+        operation: OperationType,
+    ) -> Result<(), ChannelError> {
+        let Some(senders) = self.op_listener_senders.get_mut(&operation) else {
             return Ok(());
         };
         for sender in senders {
             sender.send(()).await?;
         }
         Ok(())
     }
+
+    /// Send sync signal to the operation handler actor registered for the given operation
+    pub(crate) async fn sync(&mut self, operation: OperationType) -> Result<(), ChannelError> {
+        if let Some(sender) = self.op_handler_senders.get_mut(&operation) {
+            sender.send(()).await?;
+        };
+        Ok(())
+    }
+
+    /// Send sync signal to all registered operation handler actors
+    pub(crate) async fn sync_all(&mut self) -> Result<(), ChannelError> {
+        for sender in self.op_handler_senders.values_mut() {
+            sender.send(()).await?;
+        }
+        Ok(())
+    }
 }

crates/core/tedge_agent/src/operation_workflows/tests.rs

@@ -361,9 +361,16 @@ async fn spawn_mqtt_operation_converter(device_topic_id: &str) -> Result<TestHan
 
     let tmp_dir = tempfile::TempDir::new().unwrap();
     let tmp_path = Utf8Path::from_path(tmp_dir.path()).unwrap();
+    let device_topic_id = device_topic_id
+        .parse::<EntityTopicId>()
+        .expect("Invalid topic id");
+    let service_topic_id = device_topic_id
+        .default_service_for_device("tedge-agent")
+        .expect("Invalid service topic id");
     let config = OperationConfig {
         mqtt_schema: MqttSchema::new(),
-        device_topic_id: device_topic_id.parse().expect("Invalid topic id"),
+        device_topic_id,
+        service_topic_id,
         log_dir: tmp_path.into(),
         config_dir: tmp_path.into(),
         state_dir: tmp_path.join("running-operations"),

crates/core/tedge_api/src/mqtt_topics.rs

@@ -798,6 +798,7 @@ impl OperationType {
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum SignalType {
     Sync,
+    SyncOperation(OperationType),
     Custom(String),
 }
 
@@ -833,6 +834,14 @@ impl<'a> From<&'a str> for SignalType {
     fn from(s: &'a str) -> SignalType {
         match s {
             "sync" => SignalType::Sync,
+            "sync_restart" => SignalType::SyncOperation(OperationType::Restart),
+            "sync_firmware_update" => SignalType::SyncOperation(OperationType::FirmwareUpdate),
+            "sync_software_update" => SignalType::SyncOperation(OperationType::SoftwareUpdate),
+            "sync_software_list" => SignalType::SyncOperation(OperationType::SoftwareList),
+            "sync_config_update" => SignalType::SyncOperation(OperationType::ConfigUpdate),
+            "sync_config_snapshot" => SignalType::SyncOperation(OperationType::ConfigSnapshot),
+            "sync_log_upload" => SignalType::SyncOperation(OperationType::LogUpload),
+            "sync_device_profile" => SignalType::SyncOperation(OperationType::DeviceProfile),
             custom => SignalType::Custom(custom.to_string()),
         }
     }
@@ -842,6 +851,7 @@ impl Display for SignalType {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
             SignalType::Sync => write!(f, "sync"),
+            SignalType::SyncOperation(operation) => write!(f, "sync_{}", operation),
             SignalType::Custom(custom) => write!(f, "{custom}"),
         }
     }

crates/extensions/c8y_mapper_ext/src/signals.rs

@@ -26,7 +26,7 @@ impl CumulocityConverter {
                     }
                 }
             }
-            SignalType::Custom(_) => {}
+            SignalType::SyncOperation(_) | SignalType::Custom(_) => {}
         }
 
         Ok(messages)