Merge pull request #3817 from albinsuresh/feat/update-supported-log-list-on-software-update

feat: publish supported log types when software_update or config_update finishes

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,15 @@ impl LogManagerConfig {
             ChannelFilter::Command(OperationType::LogUpload),
         );
 
+        let mut log_metadata_sync_topics = mqtt_schema.topics(
+            EntityFilter::Entity(&mqtt_device_topic_id),
+            ChannelFilter::Command(OperationType::SoftwareUpdate),
+        );
+        log_metadata_sync_topics.add_all(mqtt_schema.topics(
+            EntityFilter::Entity(&mqtt_device_topic_id),
+            ChannelFilter::Command(OperationType::ConfigUpdate),
+        ));
+
         Ok(Self {
             mqtt_schema,
             config_dir,
@@ -72,6 +82,7 @@ impl LogManagerConfig {
             plugin_config_path,
             logtype_reload_topic,
             logfile_request_topic,
+            log_metadata_sync_topics,
             sudo_enabled: true,
         })
     }