feat: sync signal support for agent

Author: albinsuresh

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,49 @@ 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);
+                return 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?;
+                Ok(())
+            }
+            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(());

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::SignalType;
 use tedge_api::workflow::GenericCommandData;
 use tedge_api::workflow::GenericCommandState;
 use tedge_api::workflow::OperationName;
@@ -122,7 +123,15 @@ impl WorkflowActorBuilder {
     }
 
     pub fn subscriptions(mqtt_schema: &MqttSchema, device_topic_id: &EntityTopicId) -> TopicFilter {
-        mqtt_schema.topics(EntityFilter::Entity(device_topic_id), AnyCommand)
+        let mut topics = mqtt_schema.topics(
+            EntityFilter::Entity(device_topic_id),
+            ChannelFilter::AnyCommand,
+        );
+        topics.add_all(mqtt_schema.topics(
+            EntityFilter::Entity(device_topic_id),
+            ChannelFilter::Signal(SignalType::Sync),
+        ));
+        topics
     }
 }
 

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

@@ -71,4 +71,14 @@ impl SyncSignalDispatcher {
         }
         Ok(())
     }
+
+    /// Send sync signal to all registered actors
+    pub(crate) async fn sync_all(&mut self) -> Result<(), ChannelError> {
+        for senders in self.senders.values_mut() {
+            for sender in senders {
+                sender.send(()).await?;
+            }
+        }
+        Ok(())
+    }
 }

tests/RobotFramework/tests/cumulocity/log/log_operation_plugins.robot

@@ -52,6 +52,17 @@ Supported log types updated on config update
     ...    date_from=${start_time}
     ...    message_contains=software-management
 
+Supported log types updated on sync signal
+    Install Package Using APT    haveged
+
+    ${start_time}=    Get Unix Timestamp
+    Execute Command    tedge mqtt pub te/device/main///signal/sync '{}'
+
+    Should Have MQTT Messages
+    ...    topic=te/device/main///cmd/log_upload
+    ...    date_from=${start_time}
+    ...    message_contains=haveged::journald
+
 Log operation journald plugin can return logs for all units
     Should Contain Supported Log Types    all-units::journald
     ${start_timestamp}=    Get Current Date    UTC    -1 hours    result_format=%Y-%m-%dT%H:%M:%S+0000