feat: publish supported log types on software_update completion

Author: albinsuresh

crates/core/tedge_agent/src/agent.rs

@@ -384,6 +384,7 @@ impl Agent {
             )
             .await?;
             converter_actor_builder.register_builtin_operation(&mut log_actor);
+            converter_actor_builder.register_sync_signal_sink(&log_actor);
             Some(log_actor)
         } else {
             None

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

@@ -1,4 +1,5 @@
 use crate::operation_workflows::message_box::CommandDispatcher;
+use crate::operation_workflows::message_box::SyncSignalDispatcher;
 use crate::operation_workflows::persist::WorkflowRepository;
 use crate::state_repository::state::AgentStateRepository;
 use async_trait::async_trait;
@@ -53,6 +54,7 @@ pub struct WorkflowActor {
     pub(crate) log_dir: Utf8PathBuf,
     pub(crate) input_receiver: UnboundedLoggingReceiver<AgentInput>,
     pub(crate) builtin_command_dispatcher: CommandDispatcher,
+    pub(crate) sync_signal_dispatcher: SyncSignalDispatcher,
     pub(crate) command_sender: DynSender<InternalCommandState>,
     pub(crate) mqtt_publisher: LoggingSender<MqttMessage>,
     pub(crate) script_runner: ClientMessageBox<Execute, std::io::Result<Output>>,
@@ -428,7 +430,10 @@ impl WorkflowActor {
         new_state: GenericCommandState,
     ) -> Result<(), RuntimeError> {
         if new_state.is_finished() {
-            self.finalize_builtin_command_update(new_state).await
+            self.sync_dependent_actors(&new_state).await?;
+            self.finalize_builtin_command_update(new_state).await?;
+
+            Ok(())
         } else {
             // As not finalized, the builtin state is sent back
             // to the builtin operation actor for further processing.
@@ -458,6 +463,18 @@ impl WorkflowActor {
         self.process_command_update(adapted_state).await
     }
 
+    async fn sync_dependent_actors(
+        &mut self,
+        command: &GenericCommandState,
+    ) -> Result<(), RuntimeError> {
+        if let Some(command) = command.operation() {
+            self.sync_signal_dispatcher
+                .send(command.as_str().into())
+                .await?;
+        }
+        Ok(())
+    }
+
     fn open_command_log(
         &mut self,
         state: &GenericCommandState,

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

@@ -3,6 +3,7 @@ use crate::operation_workflows::actor::InternalCommandState;
 use crate::operation_workflows::actor::WorkflowActor;
 use crate::operation_workflows::config::OperationConfig;
 use crate::operation_workflows::message_box::CommandDispatcher;
+use crate::operation_workflows::message_box::SyncSignalDispatcher;
 use crate::operation_workflows::persist::WorkflowRepository;
 use crate::state_repository::state::agent_state_dir;
 use crate::state_repository::state::AgentStateRepository;
@@ -22,13 +23,15 @@ use tedge_actors::RuntimeRequest;
 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::EntityFilter;
 use tedge_api::mqtt_topics::EntityTopicId;
 use tedge_api::mqtt_topics::MqttSchema;
 use tedge_api::workflow::GenericCommandData;
 use tedge_api::workflow::GenericCommandState;
 use tedge_api::workflow::OperationName;
+use tedge_api::workflow::SyncOnCommand;
 use tedge_file_system_ext::FsWatchEvent;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::TopicFilter;
@@ -39,6 +42,7 @@ pub struct WorkflowActorBuilder {
     input_sender: DynSender<AgentInput>,
     input_receiver: UnboundedLoggingReceiver<AgentInput>,
     command_dispatcher: CommandDispatcher,
+    sync_signal_dispatcher: SyncSignalDispatcher,
     command_sender: DynSender<InternalCommandState>,
     mqtt_publisher: LoggingSender<MqttMessage>,
     script_runner: ClientMessageBox<Execute, std::io::Result<Output>>,
@@ -65,6 +69,8 @@ impl WorkflowActorBuilder {
         let command_dispatcher = CommandDispatcher::default();
         let command_sender = input_sender.sender_clone();
 
+        let sync_signal_dispatcher = SyncSignalDispatcher::default();
+
         let mqtt_publisher = mqtt_actor.get_sender();
         mqtt_actor.connect_sink(
             Self::subscriptions(&config.mqtt_schema, &config.device_topic_id),
@@ -81,6 +87,7 @@ impl WorkflowActorBuilder {
             input_sender,
             input_receiver,
             command_dispatcher,
+            sync_signal_dispatcher,
             command_sender,
             mqtt_publisher,
             signal_sender,
@@ -98,7 +105,19 @@ impl WorkflowActorBuilder {
         actor.connect_sink(NoConfig, &self.input_sender);
         for (operation, sender) in actor.into_iter() {
             self.command_dispatcher
-                .register_operation_handler(operation, sender)
+                .register_operation_handler(operation, sender);
+        }
+    }
+
+    /// Register an actor to receive sync signals on completion of other commands
+    pub fn register_sync_signal_sink<OperationActor>(&mut self, actor: &OperationActor)
+    where
+        OperationActor: MessageSink<CmdMetaSyncSignal> + SyncOnCommand,
+    {
+        let sender = actor.get_sender();
+        for operation in actor.sync_on_commands() {
+            self.sync_signal_dispatcher
+                .register_sync_signal_sender(operation, sender.sender_clone());
         }
     }
 
@@ -136,6 +155,7 @@ impl Builder<WorkflowActor> for WorkflowActorBuilder {
             log_dir: self.config.log_dir,
             input_receiver: self.input_receiver,
             builtin_command_dispatcher: self.command_dispatcher,
+            sync_signal_dispatcher: self.sync_signal_dispatcher,
             mqtt_publisher: self.mqtt_publisher,
             command_sender: self.command_sender,
             script_runner: self.script_runner,

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

@@ -4,6 +4,8 @@ use std::collections::HashMap;
 use tedge_actors::ChannelError;
 use tedge_actors::DynSender;
 use tedge_actors::Sender;
+use tedge_api::commands::CmdMetaSyncSignal;
+use tedge_api::mqtt_topics::OperationType;
 use tedge_api::workflow::GenericCommandState;
 use tedge_api::workflow::OperationName;
 
@@ -44,3 +46,29 @@ impl CommandDispatcher {
         self.senders.keys().cloned().collect()
     }
 }
+
+#[derive(Default)]
+pub(crate) struct SyncSignalDispatcher {
+    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(
+        &mut self,
+        operation: OperationType,
+        sender: DynSender<CmdMetaSyncSignal>,
+    ) {
+        self.senders.entry(operation).or_default().push(sender);
+    }
+
+    pub(crate) async fn send(&mut self, operation: OperationType) -> Result<(), ChannelError> {
+        let Some(senders) = self.senders.get_mut(&operation) else {
+            return Ok(());
+        };
+        for sender in senders {
+            sender.send(()).await?;
+        }
+        Ok(())
+    }
+}

crates/core/tedge_api/src/commands.rs

@@ -298,6 +298,8 @@ pub trait CommandPayload {
     }
 }
 
+pub type CmdMetaSyncSignal = ();
+
 /// All the messages are serialized using json.
 pub trait Jsonify {
     fn from_json(json_str: &str) -> Result<Self, serde_json::Error>

crates/core/tedge_api/src/workflow/mod.rs

@@ -506,6 +506,12 @@ pub enum IterationError {
     IndexOutOfBounds(usize),
 }
 
+/// An actor builder must implement this if it to receive sync signals on completion of other commands
+pub trait SyncOnCommand {
+    /// Return the list of operations for which this actor wants to receive sync signals
+    fn sync_on_commands(&self) -> Vec<OperationType>;
+}
+
 #[cfg(test)]
 mod tests {
     use super::GenericCommandState;

crates/extensions/tedge_log_manager/src/actor.rs

@@ -12,6 +12,7 @@ use tedge_actors::MessageReceiver;
 use tedge_actors::RuntimeError;
 use tedge_actors::Sender;
 use tedge_actors::SimpleMessageBox;
+use tedge_api::commands::CmdMetaSyncSignal;
 use tedge_api::commands::CommandStatus;
 use tedge_api::commands::LogUploadCmd;
 use tedge_api::commands::LogUploadCmdMetadata;
@@ -34,7 +35,7 @@ type MqttTopic = String;
 pub type LogUploadRequest = (MqttTopic, UploadRequest);
 pub type LogUploadResult = (MqttTopic, UploadResult);
 
-fan_in_message_type!(LogInput[LogUploadCmd, FsWatchEvent, LogUploadResult] : Debug);
+fan_in_message_type!(LogInput[LogUploadCmd, CmdMetaSyncSignal, FsWatchEvent, LogUploadResult] : Debug);
 fan_in_message_type!(LogOutput[LogUploadCmd, LogUploadCmdMetadata] : Debug);
 
 impl LogOutput {
@@ -80,6 +81,9 @@ impl Actor for LogManagerActor {
                 LogInput::LogUploadResult((topic, result)) => {
                     self.process_uploaded_log(&topic, result).await?;
                 }
+                LogInput::CmdMetaSyncSignal(_) => {
+                    self.reload_supported_log_types().await?;
+                }
             }
         }
         Ok(())

crates/extensions/tedge_log_manager/src/config.rs

@@ -27,6 +27,7 @@ pub struct LogManagerConfig {
     pub plugin_config_path: PathBuf,
     pub logtype_reload_topic: Topic,
     pub logfile_request_topic: TopicFilter,
+    pub log_metadata_sync_topics: TopicFilter,
     pub sudo_enabled: bool,
 }
 
@@ -62,6 +63,11 @@ impl LogManagerConfig {
             ChannelFilter::Command(OperationType::LogUpload),
         );
 
+        let log_metadata_sync_topics = mqtt_schema.topics(
+            EntityFilter::Entity(&mqtt_device_topic_id),
+            ChannelFilter::Command(OperationType::SoftwareUpdate),
+        );
+
         Ok(Self {
             mqtt_schema,
             config_dir,
@@ -72,6 +78,7 @@ impl LogManagerConfig {
             plugin_config_path,
             logtype_reload_topic,
             logfile_request_topic,
+            log_metadata_sync_topics,
             sudo_enabled: true,
         })
     }

crates/extensions/tedge_log_manager/src/lib.rs

@@ -24,11 +24,13 @@ use tedge_actors::RuntimeRequest;
 use tedge_actors::RuntimeRequestSink;
 use tedge_actors::Service;
 use tedge_actors::SimpleMessageBoxBuilder;
+use tedge_api::commands::CmdMetaSyncSignal;
 use tedge_api::commands::LogUploadCmd;
 use tedge_api::mqtt_topics::OperationType;
 use tedge_api::workflow::GenericCommandData;
 use tedge_api::workflow::GenericCommandState;
 use tedge_api::workflow::OperationName;
+use tedge_api::workflow::SyncOnCommand;
 use tedge_file_system_ext::FsWatchEvent;
 use tedge_utils::file::create_directory_with_defaults;
 use tedge_utils::file::move_file;
@@ -144,31 +146,41 @@ impl LogManagerBuilder {
 
     /// List of MQTT topic filters the log actor has to subscribe to
     fn subscriptions(config: &LogManagerConfig) -> TopicFilter {
-        config.logfile_request_topic.clone()
+        let mut topics = config.logfile_request_topic.clone();
+        topics.add_all(config.log_metadata_sync_topics.clone());
+        topics
     }
 
     /// Extract a log actor request from an MQTT message
     fn mqtt_message_parser(config: &LogManagerConfig) -> impl Fn(MqttMessage) -> Option<LogInput> {
         let logfile_request_topic = config.logfile_request_topic.clone();
+        let log_metadata_sync_topics = config.log_metadata_sync_topics.clone();
         let mqtt_schema = config.mqtt_schema.clone();
         move |message| {
-            if !logfile_request_topic.accept(&message) {
+            if logfile_request_topic.accept(&message) {
+                LogUploadCmd::parse(&mqtt_schema, message)
+                    .map_err(|err| {
+                        error!(
+                            target: "log plugins",
+                            "Incorrect log request payload: {}", err
+                        )
+                    })
+                    .unwrap_or(None)
+                    .map(|cmd| cmd.into())
+            } else if log_metadata_sync_topics.accept(&message) {
+                if let Ok(cmd) = GenericCommandState::from_command_message(&message) {
+                    if cmd.is_finished() {
+                        return Some(LogInput::CmdMetaSyncSignal(()));
+                    }
+                }
+                None
+            } else {
                 error!(
                     target: "log plugins",
                     "Received unexpected message on topic: {}", message.topic.name
                 );
-                return None;
+                None
             }
-
-            LogUploadCmd::parse(&mqtt_schema, message)
-                .map_err(|err| {
-                    error!(
-                        target: "log plugins",
-                        "Incorrect log request payload: {}", err
-                    )
-                })
-                .unwrap_or(None)
-                .map(|cmd| cmd.into())
         }
     }
 
@@ -240,3 +252,16 @@ impl IntoIterator for &LogManagerBuilder {
         vec![(OperationType::LogUpload.to_string(), sender.into())].into_iter()
     }
 }
+
+impl MessageSink<CmdMetaSyncSignal> for LogManagerBuilder {
+    fn get_sender(&self) -> DynSender<CmdMetaSyncSignal> {
+        self.box_builder.get_sender().sender_clone()
+    }
+}
+
+impl SyncOnCommand for LogManagerBuilder {
+    /// Return the list of operations for which this actor wants to receive sync signals
+    fn sync_on_commands(&self) -> Vec<OperationType> {
+        vec![OperationType::SoftwareUpdate]
+    }
+}