Backends: improvements of properties management and traits documentation

Author: JEnoch

backends/influxdb/Cargo.toml

@@ -36,7 +36,7 @@ log = "0.4"
 uuid = { version = "0.8", features = ["v4"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-influxdb = "0.3"
+influxdb = { version = "0.3", features = ["h1-client"] }
 
 [dependencies.async-std]
 version = "=1.6.5"

backends/influxdb/src/lib.rs

@@ -34,6 +34,14 @@ use zenoh_backend_traits::*;
 use zenoh_util::collections::{Timed, TimedEvent, TimedHandle, Timer};
 use zenoh_util::{zerror, zerror2};
 
+// Properies used by the Backend
+pub const PROP_BACKEND_URL: &str = "url";
+
+// Properies used by the Storage
+pub const PROP_STORAGE_DB: &str = "db";
+pub const PROP_STORAGE_CREATE_DB: &str = "create_db";
+pub const PROP_STORAGE_ON_CLOSURE: &str = "on_closure";
+
 // delay after deletion to drop a measurement
 const DROP_MEASUREMENT_TIMEOUT_MS: u64 = 5000;
 
@@ -43,15 +51,15 @@ pub fn create_backend(props: &Properties) -> ZResult<Box<dyn Backend>> {
     // Try to activate it here, ignoring failures.
     let _ = env_logger::try_init();
 
-    match props.get("url") {
+    match props.get(PROP_BACKEND_URL) {
         Some(url) => {
             // Check connectivity to InfluxDB, no need for a database for this
             let client = Client::new(url, "UNUSED");
             match async_std::task::block_on(async move { client.ping().await }) {
                 Ok(_) => {
-                    let mut admin_status_props = props.clone();
-                    let _ = admin_status_props.insert("kind".into(), "time series".into());
-                    let admin_status = zenoh::utils::properties_to_json_value(&admin_status_props);
+                    let mut p = props.clone();
+                    p.insert(PROP_BACKEND_TYPE.into(), "InfluxDB".into());
+                    let admin_status = zenoh::utils::properties_to_json_value(&p);
                     Ok(Box::new(InfluxDbBackend {
                         url: url.clone(),
                         admin_status,
@@ -101,7 +109,7 @@ enum OnClosure {
 impl TryFrom<&Properties> for OnClosure {
     type Error = ZError;
     fn try_from(p: &Properties) -> ZResult<OnClosure> {
-        match p.get("on_closure") {
+        match p.get(PROP_STORAGE_ON_CLOSURE) {
             Some(s) => {
                 if s == "drop_db" {
                     Ok(OnClosure::DropDB)
@@ -129,14 +137,14 @@ struct InfluxDbStorage {
 
 impl InfluxDbStorage {
     async fn new(props: Properties, url: &str) -> ZResult<InfluxDbStorage> {
-        let path_expr = props.get("path_expr").unwrap();
-        let path_prefix = match props.get("path_prefix") {
+        let path_expr = props.get(PROP_STORAGE_PATH_EXPR).unwrap();
+        let path_prefix = match props.get(PROP_STORAGE_PATH_PREFIX) {
             Some(p) => {
                 if !path_expr.starts_with(p) {
                     return zerror!(ZErrorKind::Other {
                         descr: format!(
-                            "The specified path_prefix={} is not a prefix of pah_expr={}",
-                            p, path_expr
+                            "The specified {}={} is not a prefix of {}={}",
+                            PROP_STORAGE_PATH_PREFIX, p, PROP_STORAGE_PATH_EXPR, path_expr
                         )
                     });
                 }
@@ -145,9 +153,9 @@ impl InfluxDbStorage {
             None => None,
         };
         let on_closure = OnClosure::try_from(&props)?;
-        let createdb = props.contains_key("create_db");
+        let createdb = props.contains_key(PROP_STORAGE_CREATE_DB);
 
-        let client = match props.get("db") {
+        let client = match props.get(PROP_STORAGE_DB) {
             Some(db) => {
                 check_db_existence(url, db, createdb).await?;
                 Client::new(url, db)
@@ -160,7 +168,7 @@ impl InfluxDbStorage {
         };
 
         let mut admin_status_props = props.clone();
-        admin_status_props.insert("db".into(), client.database_name().into());
+        admin_status_props.insert(PROP_STORAGE_DB.into(), client.database_name().into());
         let admin_status = zenoh::utils::properties_to_json_value(&admin_status_props);
         Ok(InfluxDbStorage {
             admin_status,
@@ -268,9 +276,12 @@ impl Storage for InfluxDbStorage {
                     }
                 }
 
+                // encode the value as a string to be stored in InfluxDB
                 let (encoding, base64, value) = change.value.unwrap().encode_to_string();
+
                 // Note: tags are stored as strings in InfluxDB, while fileds are typed.
                 // For simpler/faster deserialization, we store encoding, timestamp and base64 as fields.
+                // while the kind is stored as a tag to be indexed by InfluxDB and have faster queries on it.
                 let query =
                     InfluxWQuery::new(InfluxTimestamp::Nanoseconds(influx_time), measurement)
                         .add_tag("kind", "PUT")
@@ -333,16 +344,10 @@ impl Storage for InfluxDbStorage {
 
     // When receiving a Query (i.e. on GET operations)
     async fn on_query(&mut self, query: Query) -> ZResult<()> {
-        #[derive(Deserialize, Debug)]
-        struct ZenohPoint {
-            kind: String,
-            timestamp: String,
-            encoding: zenoh::net::ZInt,
-            base64: bool,
-            value: String,
-        }
-
+        // get the query's Selector
         let selector = Selector::try_from(&query)?;
+
+        // if a path_prefix is used
         let regex = if let Some(prefix) = &self.path_prefix {
             // get the list of sub-path expressions that will match the same stored keys than
             // the selector, if those keys had the path_prefix.
@@ -351,28 +356,45 @@ impl Storage for InfluxDbStorage {
                 "Query on {} with path_expr={} => sub_path_exprs = {:?}",
                 selector.path_expr, prefix, path_exprs
             );
+            // convert the sub-path expressions into an Influx regex
             path_exprs_to_influx_regex(&path_exprs)
         } else {
+            // convert the Selector's path expression into an Influx regex
             path_exprs_to_influx_regex(&[selector.path_expr.as_str()])
         };
+
+        // construct the Influx query clauses from the Selector
         let clauses = clauses_from_selector(&selector);
+
+        // the Influx query
         let influx_query_str = format!("SELECT * FROM {} {}", regex, clauses);
         let influx_query = InfluxQuery::raw_read_query(&influx_query_str);
 
+        // the expected JSon type resulting from the query
+        #[derive(Deserialize, Debug)]
+        struct ZenohPoint {
+            kind: String,
+            timestamp: String,
+            encoding: zenoh::net::ZInt,
+            base64: bool,
+            value: String,
+        }
         debug!("Get {} with Influx query: {}", selector, influx_query_str);
         match self.client.json_query(influx_query).await {
             Ok(mut query_result) => {
                 while !query_result.results.is_empty() {
                     match query_result.deserialize_next::<ZenohPoint>() {
                         Ok(retn) => {
                             for serie in retn.series {
+                                // reconstruct the path from the measurement name (same as serie.name)
                                 let mut res_name = String::with_capacity(serie.name.len());
                                 if let Some(p) = &self.path_prefix {
                                     res_name.push_str(&p);
                                 }
                                 res_name.push_str(&serie.name);
                                 debug!("Replying {} values for {}", serie.values.len(), res_name);
                                 for zpoint in serie.values {
+                                    // decode the value and the timestamp
                                     match (
                                         Value::decode_from_string(
                                             zpoint.encoding,

backends/traits/src/lib.rs

@@ -15,14 +15,27 @@ use async_std::sync::{Arc, RwLock};
 use async_trait::async_trait;
 use std::convert::TryFrom;
 use zenoh::net::Sample;
-use zenoh::{Properties, Selector, Timestamp, Value, ZError, ZResult};
+use zenoh::{Properties, Selector, Value, ZError, ZResult};
 
 pub mod utils;
 
-pub const STORAGE_PATH_EXPR_PROPERTY: &str = "path_expr";
+/// The `"type"` property key to be used in admin status reported by Backends.
+pub const PROP_BACKEND_TYPE: &str = "type";
 
-/// Trait to be implemented by a Backend.  
-/// A library implementing the Backend and Storage traits must also declare a [`create_backend()`] operation
+/// The `"path_expr"` property key to be used for configuration of each storage.
+pub const PROP_STORAGE_PATH_EXPR: &str = "path_expr";
+
+/// The `"path_prefix"` property key that could be used to specify the common path prefix
+/// to be stripped from Paths before storing them as keys in the Storage.
+///
+/// Note that it shall be a prefix of the `"path_expr"`.
+/// If you use it, you should also adapt in [`Storage::on_query()`] implementation the incoming
+/// queries' path expression to the stored keys calling [`crate::utils::get_sub_path_exprs()`].
+pub const PROP_STORAGE_PATH_PREFIX: &str = "path_prefix";
+
+/// Trait to be implemented by a Backend.
+///
+/// A library implementing the Backend and Storage traits must also declare a `create_backend()` operation
 /// with the `#[no_mangle]` attribute as an entrypoint to be called for the Backend creation.
 ///
 /// # Example
@@ -33,18 +46,27 @@ pub const STORAGE_PATH_EXPR_PROPERTY: &str = "path_expr";
 /// use zenoh_backend_traits::*;
 ///
 /// #[no_mangle]
-/// pub fn create_backend(_unused: &Properties) -> ZResult<Box<dyn Backend>> {
-///     Ok(Box::new(MyBackend {}))
+/// pub fn create_backend(properties: &Properties) -> ZResult<Box<dyn Backend>> {
+///     // The properties are the ones passed via a PUT in the admin space for Backend creation
+///     // Here we re-expose them in the admin space for GET operations, adding the PROP_BACKEND_TYPE entry.
+///     let mut p = properties.clone();
+///     p.insert(PROP_BACKEND_TYPE.into(), "my_backend_type".into());
+///     let admin_status = utils::properties_to_json_value(&p);
+///     Ok(Box::new(MyBackend { admin_status }))
 /// }
 ///
 /// // Your Backend implementation
-/// pub struct MyBackend {}
+/// struct MyBackend {
+///     admin_status: Value,
+/// }
 ///
 /// #[async_trait]
 /// impl Backend for MyBackend {
 ///     async fn get_admin_status(&self) -> Value {
-///         // TODO: possibly add more properties in returned Value for more information about this storage
-///         Value::Json(r#"{"kind"="some kind"}"#.to_string())
+///         // This operation is called on GET operation on the admin space for the Backend
+///         // Here we reply with a static status (containing the configuration properties).
+///         // But we could add dynamic properties for Backend monitoring.
+///         self.admin_status.clone()
 ///     }
 ///
 ///     async fn create_storage(&mut self, properties: Properties) -> ZResult<Box<dyn Storage>> {
@@ -67,6 +89,9 @@ pub const STORAGE_PATH_EXPR_PROPERTY: &str = "path_expr";
 ///
 /// impl MyStorage {
 ///     async fn new(properties: Properties) -> ZResult<MyStorage> {
+///         // The properties are the ones passed via a PUT in the admin space for Storage creation.
+///         // They contain at least a PROP_STORAGE_PATH_EXPR entry (i.e. "path_expr").
+///         // Here we choose to re-expose them as they are in the admin space for GET operations.
 ///         let admin_status = utils::properties_to_json_value(&properties);
 ///         Ok(MyStorage { admin_status })
 ///     }
@@ -75,7 +100,9 @@ pub const STORAGE_PATH_EXPR_PROPERTY: &str = "path_expr";
 /// #[async_trait]
 /// impl Storage for MyStorage {
 ///     async fn get_admin_status(&self) -> Value {
-///         // TODO: possibly add more properties in returned Value for more information about this storage
+///         // This operation is called on GET operation on the admin space for the Storage
+///         // Here we reply with a static status (containing the configuration properties).
+///         // But we could add dynamic properties for Storage monitoring.
 ///         self.admin_status.clone()
 ///     }
 ///
@@ -87,11 +114,11 @@ pub const STORAGE_PATH_EXPR_PROPERTY: &str = "path_expr";
 ///                 info.kind.map_or(ChangeKind::PUT, ChangeKind::from),
 ///                 match &info.timestamp {
 ///                     Some(ts) => ts.clone(),
-///                     None => new_reception_timestamp(),
+///                     None => zenoh::utils::new_reception_timestamp(),
 ///                 },
 ///             )
 ///         } else {
-///             (ChangeKind::PUT, new_reception_timestamp())
+///             (ChangeKind::PUT, zenoh::utils::new_reception_timestamp())
 ///         };
 ///         // Store or delete the sample depending the ChangeKind
 ///         match kind {
@@ -106,7 +133,7 @@ pub const STORAGE_PATH_EXPR_PROPERTY: &str = "path_expr";
 ///                 let _key = sample.res_name;
 ///                 // TODO:
 ///                 //  - check if timestamp is newer than the stored one for the same key
-///                 //  - if yes: mark (key, sample) as deleted (possibly scheduling definitive removal for later)
+///                 //  - if yes: mark key as deleted (possibly scheduling definitive removal for later)
 ///                 //  - if not: drop the sample
 ///             }
 ///             ChangeKind::PATCH => {
@@ -219,14 +246,3 @@ impl TryFrom<&Query> for Selector {
         Selector::try_from(&q.q)
     }
 }
-
-/// Generates a reception timestamp with id=0x00
-pub fn new_reception_timestamp() -> Timestamp {
-    use std::time::{SystemTime, UNIX_EPOCH};
-    use zenoh::TimestampID;
-    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
-    Timestamp::new(
-        now.into(),
-        TimestampID::new(1, [0u8; TimestampID::MAX_SIZE]),
-    )
-}

plugins/zenoh-storages/src/backends_mgt.rs

@@ -20,11 +20,11 @@ use log::{debug, error, trace, warn};
 use std::collections::HashMap;
 use std::convert::TryFrom;
 use zenoh::{ChangeKind, Path, PathExpr, Selector, Value, ZError, ZErrorKind, ZResult, Zenoh};
-use zenoh_backend_traits::{IncomingDataInterceptor, OutgoingDataInterceptor};
+use zenoh_backend_traits::{
+    IncomingDataInterceptor, OutgoingDataInterceptor, PROP_STORAGE_PATH_EXPR,
+};
 use zenoh_util::{zerror, zerror2};
 
-pub(crate) const STORAGE_PATH_EXPR_PROPERTY: &str = "path_expr";
-
 pub(crate) async fn start_backend(
     backend: Box<dyn zenoh_backend_traits::Backend>,
     admin_path: Path,
@@ -135,11 +135,11 @@ async fn create_and_start_storage(
 ) -> ZResult<Sender<bool>> {
     trace!("Create storage {}", admin_path);
     if let Value::Properties(props) = value {
-        let path_expr_str = props.get(STORAGE_PATH_EXPR_PROPERTY).ok_or_else(|| {
+        let path_expr_str = props.get(PROP_STORAGE_PATH_EXPR).ok_or_else(|| {
             zerror2!(ZErrorKind::Other {
                 descr: format!(
                     "Can't create storage {}: no {} property",
-                    admin_path, STORAGE_PATH_EXPR_PROPERTY
+                    admin_path, PROP_STORAGE_PATH_EXPR
                 )
             })
         })?;

plugins/zenoh-storages/src/lib.rs

@@ -21,7 +21,7 @@ use log::{debug, error, warn};
 use std::collections::HashMap;
 use std::convert::TryFrom;
 use zenoh::{ChangeKind, Path, Properties, Selector, Value, ZError, ZErrorKind, ZResult, Zenoh};
-use zenoh_backend_traits::Backend;
+use zenoh_backend_traits::{Backend, PROP_STORAGE_PATH_EXPR};
 use zenoh_router::runtime::Runtime;
 use zenoh_util::{zerror, LibLoader};
 
@@ -108,7 +108,7 @@ async fn run(runtime: Runtime, args: &'static ArgMatches<'_>) {
                     mem_backend_path, MEMORY_STORAGE_NAME, i
                 ))
                 .unwrap();
-                let props = Properties::from([(STORAGE_PATH_EXPR_PROPERTY, path_expr)].as_ref());
+                let props = Properties::from([(PROP_STORAGE_PATH_EXPR, path_expr)].as_ref());
                 workspace
                     .put(&storage_admin_path, Value::Properties(props))
                     .await

plugins/zenoh-storages/src/memory_backend/mod.rs

@@ -24,8 +24,9 @@ use zenoh_backend_traits::*;
 use zenoh_util::collections::{Timed, TimedEvent, TimedHandle, Timer};
 
 pub fn create_backend(_unused: Properties) -> ZResult<Box<dyn Backend>> {
-    // For now admin status is static and only contains a "kind"
-    let admin_status = Value::Json(r#"{"kind"="memory"}"#.to_string());
+    // For now admin status is static and only contains a PROP_BACKEND_TYPE entry
+    let properties = Properties::from(&[(PROP_BACKEND_TYPE, "memory")][..]);
+    let admin_status = utils::properties_to_json_value(&properties);
     Ok(Box::new(MemoryBackend { admin_status }))
 }
 
@@ -157,11 +158,11 @@ impl Storage for MemoryStorage {
                 info.kind.map_or(ChangeKind::PUT, ChangeKind::from),
                 match &info.timestamp {
                     Some(ts) => ts.clone(),
-                    None => new_reception_timestamp(),
+                    None => utils::new_reception_timestamp(),
                 },
             )
         } else {
-            (ChangeKind::PUT, new_reception_timestamp())
+            (ChangeKind::PUT, utils::new_reception_timestamp())
         };
         match kind {
             ChangeKind::PUT => match self.map.write().await.entry(sample.res_name.clone()) {

zenoh/src/utils.rs

@@ -11,8 +11,27 @@
 // Contributors:
 //   ADLINK zenoh team, <[email protected]>
 //
-use crate::{Properties, Value};
 
+//! Some useful operations for the zenoh API.
+
+use crate::{Properties, Timestamp, TimestampID, Value};
+
+/// Generates a reception [`Timestamp`] with id=0x00.  
+/// This operation should be called if a timestamp is required for an incoming [`zenoh::net::Sample`](crate::net::Sample)
+/// that doesn't contain any data_info or timestamp within its data_info.
+pub fn new_reception_timestamp() -> Timestamp {
+    use std::time::{SystemTime, UNIX_EPOCH};
+
+    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
+    Timestamp::new(
+        now.into(),
+        TimestampID::new(1, [0u8; TimestampID::MAX_SIZE]),
+    )
+}
+
+/// Convert a set of [`Properties`] into a [`Value::Json`].  
+/// For instance such Properties: `[("k1", "v1"), ("k2, v2")]`  
+/// are converted into such Json: `{ "k1": "v1", "k2": "v2" }`
 pub fn properties_to_json_value(props: &Properties) -> Value {
     let json_map = props
         .iter()