Merge pull request #2108 from albinsuresh/c8y-mapper-ops-handling

C8y mapper handling c8y_LogfileRequest

crates/common/mqtt_channel/src/topics.rs

@@ -157,6 +157,16 @@ impl TryInto<TopicFilter> for &str {
     }
 }
 
+impl FromIterator<TopicFilter> for TopicFilter {
+    fn from_iter<T: IntoIterator<Item = TopicFilter>>(filters: T) -> Self {
+        let mut combined_filters = TopicFilter::empty();
+        for filter in filters.into_iter() {
+            combined_filters.add_all(filter)
+        }
+        combined_filters
+    }
+}
+
 impl TryInto<TopicFilter> for Vec<&str> {
     type Error = MqttError;
 

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

@@ -385,6 +385,13 @@ define_tedge_config! {
         #[tedge_config(default(value = "te/+/+/+/+/m/+,tedge/alarms/+/+,tedge/alarms/+/+/+,tedge/events/+,tedge/events/+/+,tedge/health/+,tedge/health/+/+"))]
         topics: TemplatesSet,
 
+        enable: {
+            /// Enable log management
+            // TODO turn the default to true, when c8y-log-plugin will be deprecated
+            #[tedge_config(example = "true", default(value = false))]
+            log_management: bool,
+        }
+
     },
 
     #[tedge_config(deprecated_name = "azure")] // for 0.1.0 compatibility

crates/core/c8y_api/src/smartrest/topic.rs

@@ -1,6 +1,8 @@
 use mqtt_channel::MqttError;
 use mqtt_channel::Topic;
 use mqtt_channel::TopicFilter;
+use tedge_api::entity_store::EntityMetadata;
+use tedge_api::entity_store::EntityType;
 use tedge_api::topic::ResponseTopic;
 use tedge_api::TopicError;
 
@@ -88,6 +90,17 @@ impl From<C8yTopic> for TopicFilter {
     }
 }
 
+// FIXME this From conversion is error prone as this can only be used for responses.
+impl From<&EntityMetadata> for C8yTopic {
+    fn from(value: &EntityMetadata) -> Self {
+        match value.r#type {
+            EntityType::MainDevice => Self::SmartRestResponse,
+            EntityType::ChildDevice => Self::ChildSmartRestResponse(value.entity_id.clone()),
+            EntityType::Service => Self::SmartRestResponse, // TODO how services are handled by c8y?
+        }
+    }
+}
+
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum MapperSubscribeTopic {
     C8yTopic(C8yTopic),

crates/core/tedge_api/src/entity_store.rs

@@ -11,6 +11,7 @@ use std::collections::HashMap;
 
 use crate::entity_store;
 use crate::mqtt_topics::EntityTopicId;
+use crate::mqtt_topics::TopicIdError;
 use mqtt_channel::Message;
 use mqtt_channel::Topic;
 
@@ -70,6 +71,7 @@ impl EntityStore {
 
         let entity_id = main_device.entity_id?;
         let metadata = EntityMetadata {
+            topic_id: main_device.topic_id.clone(),
             entity_id: entity_id.clone(),
             r#type: main_device.r#type,
             parent: None,
@@ -170,6 +172,7 @@ impl EntityStore {
             .entity_id
             .unwrap_or_else(|| self.derive_entity_id(&message.topic_id));
         let entity_metadata = EntityMetadata {
+            topic_id: message.topic_id.clone(),
             r#type: message.r#type,
             entity_id: entity_id.clone(),
             parent,
@@ -179,7 +182,7 @@ impl EntityStore {
         // device is affected if it was previously registered and was updated
         let previous = self
             .entities
-            .insert(message.topic_id.clone(), entity_metadata);
+            .insert(entity_metadata.topic_id.clone(), entity_metadata);
 
         if previous.is_some() {
             affected_entities.push(message.topic_id);
@@ -283,6 +286,7 @@ impl EntityStore {
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct EntityMetadata {
+    pub topic_id: EntityTopicId,
     pub parent: Option<EntityTopicId>,
     pub r#type: EntityType,
     pub entity_id: String,
@@ -297,9 +301,10 @@ pub enum EntityType {
 }
 
 impl EntityMetadata {
-    /// Creates a entity metadata for a child device.
+    /// Creates a entity metadata for the main device.
     pub fn main_device(device_id: String) -> Self {
         Self {
+            topic_id: EntityTopicId::default_main_device(),
             entity_id: device_id,
             r#type: EntityType::MainDevice,
             parent: None,
@@ -308,13 +313,14 @@ impl EntityMetadata {
     }
 
     /// Creates a entity metadata for a child device.
-    pub fn child_device(child_device_id: String) -> Self {
-        Self {
+    pub fn child_device(child_device_id: String) -> Result<Self, TopicIdError> {
+        Ok(Self {
+            topic_id: EntityTopicId::default_child_device(&child_device_id)?,
             entity_id: child_device_id,
             r#type: EntityType::ChildDevice,
             parent: Some(EntityTopicId::default_main_device()),
             other: serde_json::json!({}),
-        }
+        })
     }
 }
 

crates/core/tedge_api/src/lib.rs

@@ -1,19 +1,18 @@
-pub mod error;
-pub mod messages;
-mod software;
-pub mod topic;
-
 pub mod alarm;
 pub mod builder;
 pub mod data;
 pub mod entity_store;
+pub mod error;
 pub mod event;
 pub mod group;
 pub mod health;
 pub mod measurement;
+pub mod messages;
 pub mod mqtt_topics;
 pub mod parser;
 pub mod serialize;
+mod software;
+pub mod topic;
 pub mod utils;
 
 pub use download::*;

crates/core/tedge_api/src/messages.rs

@@ -5,6 +5,7 @@ use mqtt_channel::Topic;
 use nanoid::nanoid;
 use serde::Deserialize;
 use serde::Serialize;
+use time::OffsetDateTime;
 
 const SOFTWARE_LIST_REQUEST_TOPIC: &str = "tedge/commands/req/software/list";
 const SOFTWARE_LIST_RESPONSE_TOPIC: &str = "tedge/commands/res/software/list";
@@ -534,6 +535,60 @@ impl RestartOperationResponse {
     }
 }
 
+#[derive(Debug, Deserialize, Serialize, PartialEq, Copy, Eq, Clone)]
+#[serde(rename_all = "camelCase")]
+pub enum CommandStatus {
+    Init,
+    Executing,
+    Successful,
+    Failed,
+}
+
+#[derive(Debug, Deserialize, Serialize, Eq, PartialEq)]
+#[serde(rename_all = "camelCase")]
+pub struct LogMetadata {
+    pub types: Vec<String>,
+}
+
+impl<'a> Jsonify<'a> for LogMetadata {}
+
+#[derive(Debug, Deserialize, Serialize, Eq, PartialEq, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct LogUploadCmdPayload {
+    pub status: CommandStatus, //Define a different enum if this op needs more states,
+    pub tedge_url: String,
+    #[serde(rename = "type")]
+    pub log_type: String,
+    #[serde(with = "time::serde::rfc3339")]
+    pub date_from: OffsetDateTime,
+    #[serde(with = "time::serde::rfc3339")]
+    pub date_to: OffsetDateTime,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub search_text: Option<String>,
+    pub lines: usize,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reason: Option<String>,
+}
+
+impl<'a> Jsonify<'a> for LogUploadCmdPayload {}
+
+impl LogUploadCmdPayload {
+    pub fn executing(&mut self) {
+        self.status = CommandStatus::Executing;
+        self.reason = None;
+    }
+
+    pub fn successful(&mut self) {
+        self.status = CommandStatus::Successful;
+        self.reason = None;
+    }
+
+    pub fn failed(&mut self, reason: impl Into<String>) {
+        self.status = CommandStatus::Failed;
+        self.reason = Some(reason.into());
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;