rust(feat): Run and Asset Tags and Metadata (#319)

Author: nathan-sift

rust/crates/sift_error/src/lib.rs

@@ -108,6 +108,8 @@ pub enum ErrorKind {
     RetrieveRunError,
     /// Indicates that the program was unable to retrieve the asset being requested.
     RetrieveAssetError,
+    /// Indicates that the program was unable to update the asset being requested.
+    UpdateAssetError,
     /// Indicates a failure to update a run.
     UpdateRunError,
     /// Indicates that the program was unable to retrieve the ingestion config being requested.
@@ -193,6 +195,7 @@ impl fmt::Display for ErrorKind {
             Self::GrpcConnectError => write!(f, "GrpcConnectError"),
             Self::RetriesExhausted => write!(f, "RetriesExhausted"),
             Self::RetrieveAssetError => write!(f, "RetrieveAssetError"),
+            Self::UpdateAssetError => write!(f, "UpdateAssetError"),
             Self::RetrieveRunError => write!(f, "RetrieveRunError"),
             Self::RetrieveIngestionConfigError => write!(f, "RetrieveIngestionConfigError"),
             Self::EmptyResponseError => write!(f, "EmptyResponseError"),

rust/crates/sift_rs/src/wrappers/assets.rs

@@ -1,10 +1,13 @@
 use std::ops::{Deref, DerefMut};
 
 use async_trait::async_trait;
+use pbjson_types::FieldMask;
 use sift_connect::SiftChannel;
 use sift_error::prelude::*;
 
-use crate::assets::v1::{Asset, GetAssetRequest, asset_service_client::AssetServiceClient};
+use crate::assets::v1::{
+    Asset, GetAssetRequest, UpdateAssetRequest, asset_service_client::AssetServiceClient,
+};
 
 /// Return an implementation of [AssetServiceWrapper] which also exposes methods from the
 /// raw [AssetServiceClient].
@@ -17,6 +20,9 @@ pub fn new_asset_service(grpc_channel: SiftChannel) -> impl AssetServiceWrapper
 pub trait AssetServiceWrapper: Deref<Target = AssetServiceClient<SiftChannel>> + DerefMut {
     /// Retrieves an asset by ID
     async fn try_get_asset_by_id(&mut self, asset_id: &str) -> Result<Asset>;
+
+    /// Update an asset
+    async fn try_update_asset(&mut self, asset: Asset, update_mask: Vec<String>) -> Result<Asset>;
 }
 
 /// A convenience wrapper around [AssetServiceClient].
@@ -37,6 +43,21 @@ impl AssetServiceWrapper for AssetServiceImpl {
             Error::new_empty_response("unexpected empty response from AssetService/GetAsset")
         })
     }
+
+    async fn try_update_asset(&mut self, asset: Asset, update_mask: Vec<String>) -> Result<Asset> {
+        let req = UpdateAssetRequest {
+            asset: Some(asset),
+            update_mask: Some(FieldMask { paths: update_mask }),
+        };
+        let resp = self
+            .update_asset(req)
+            .await
+            .map_err(|e| Error::new(ErrorKind::UpdateAssetError, e))?;
+
+        resp.into_inner().asset.ok_or_else(|| {
+            Error::new_empty_response("unexpected empty response from AssetService/UpdateAsset")
+        })
+    }
 }
 
 impl Deref for AssetServiceImpl {

rust/crates/sift_rs/src/wrappers/metadata.rs

@@ -0,0 +1,75 @@
+use crate::metadata::v1::{MetadataKey, MetadataKeyType};
+
+pub use crate::metadata::v1::MetadataValue;
+pub use crate::metadata::v1::metadata_value::Value as MetadataEnumValue;
+
+impl From<f64> for MetadataEnumValue {
+    fn from(value: f64) -> MetadataEnumValue {
+        MetadataEnumValue::NumberValue(value)
+    }
+}
+
+impl From<bool> for MetadataEnumValue {
+    fn from(value: bool) -> MetadataEnumValue {
+        MetadataEnumValue::BooleanValue(value)
+    }
+}
+
+impl From<String> for MetadataEnumValue {
+    fn from(value: String) -> MetadataEnumValue {
+        MetadataEnumValue::StringValue(value)
+    }
+}
+
+impl From<&str> for MetadataEnumValue {
+    fn from(value: &str) -> MetadataEnumValue {
+        MetadataEnumValue::StringValue(value.to_string())
+    }
+}
+
+impl<T: Into<MetadataEnumValue>> From<(String, T)> for MetadataValue {
+    fn from((name, value): (String, T)) -> MetadataValue {
+        MetadataValue::from((name.as_str(), value))
+    }
+}
+
+impl<T: Into<MetadataEnumValue>> From<(&str, T)> for MetadataValue {
+    fn from((name, value): (&str, T)) -> MetadataValue {
+        let enum_value: MetadataEnumValue = value.into();
+        let key_type = match enum_value {
+            MetadataEnumValue::NumberValue(_) => MetadataKeyType::Number,
+            MetadataEnumValue::BooleanValue(_) => MetadataKeyType::Boolean,
+            MetadataEnumValue::StringValue(_) => MetadataKeyType::String,
+        };
+
+        let key = MetadataKey {
+            name: name.to_string(),
+            r#type: key_type.into(),
+            archived_date: None,
+        };
+
+        MetadataValue {
+            key: Some(key),
+            value: Some(enum_value),
+            archived_date: None,
+        }
+    }
+}
+
+/// A macro for easily creating an array of metadata to be provided to Sift.
+/// Returns a Vec<[MetadataValue]>
+///
+///  # Example
+/// ```
+/// let metadata = metadata![
+///        ("test_number", 5.0),
+///        ("is_simulation", true),
+///        ("location", "SiftHQ"),
+/// ];
+/// ```
+#[macro_export]
+macro_rules! metadata {
+    ( $( ($k:expr, $v:expr) ),* $(,)? ) => {
+        vec![ $( ($k, $v).into() ),* ]
+    }
+}

rust/crates/sift_rs/src/wrappers/mod.rs

@@ -4,6 +4,9 @@ pub mod assets;
 /// Offers a wrapper over Sift's ingestion configs API.
 pub mod ingestion_configs;
 
+/// Offers a wrapper over Sift's metadata API.
+pub mod metadata;
+
 /// Offers a wrapper over Sift's runs API.
 pub mod runs;
 

rust/crates/sift_rs/src/wrappers/runs.rs

@@ -1,7 +1,10 @@
 use super::ResourceIdentifier;
-use crate::runs::v2::{
-    CreateRunRequest, GetRunRequest, ListRunsRequest, Run, UpdateRunRequest,
-    run_service_client::RunServiceClient,
+use crate::{
+    metadata::v1::MetadataValue,
+    runs::v2::{
+        CreateRunRequest, GetRunRequest, ListRunsRequest, Run, UpdateRunRequest,
+        run_service_client::RunServiceClient,
+    },
 };
 use async_trait::async_trait;
 use pbjson_types::FieldMask;
@@ -25,16 +28,17 @@ pub trait RunServiceWrapper: Deref<Target = RunServiceClient<SiftChannel>> + Der
         client_key: &str,
         description: &str,
         tags: &[String],
+        metadata: &[MetadataValue],
     ) -> Result<Run>;
 
     /// Update a run. The `updated_run` is expected to contain the `run_id` or `client_key` used to
-    /// identify the run to update. The `field_mask` is a list of snake_cased field names used to
+    /// identify the run to update. The `update_mask` is a list of snake_cased field names used to
     /// indicate which fields should actually be updated. A list of valid field names can be found
-    /// at [`this link`]. The [Run] returned is the updated run. If `field_masks` is empty, then no
+    /// at [`this link`]. The [Run] returned is the updated run. If `update_mask` is empty, then no
     /// update is required and the `updated_run` is simply returned.
     ///
     /// [`this link`]: https://docs.siftstack.com/docs/api/grpc/protocol-buffers/runs#updaterunrequest
-    async fn try_update_run(&mut self, updated_run: Run, field_masks: &[String]) -> Result<Run>;
+    async fn try_update_run(&mut self, updated_run: Run, update_mask: &[String]) -> Result<Run>;
 
     /// Retrieve a run by ID.
     async fn try_get_run_by_id(&mut self, run_id: &str) -> Result<Run>;
@@ -94,8 +98,10 @@ impl RunServiceWrapper for RunServiceWrapperImpl {
         client_key: &str,
         description: &str,
         tags: &[String],
+        metadata: &[MetadataValue],
     ) -> Result<Run> {
         let tags = tags.to_vec();
+        let metadata = metadata.to_vec();
 
         if name.is_empty() {
             return Err(Error::new_arg_error("run name cannot be blank"));
@@ -110,6 +116,7 @@ impl RunServiceWrapper for RunServiceWrapperImpl {
                 description: description.to_string(),
                 tags,
                 client_key: Some(client_key.to_string()),
+                metadata,
                 ..Default::default()
             })
             .await
@@ -123,21 +130,21 @@ impl RunServiceWrapper for RunServiceWrapperImpl {
     }
 
     /// Update a run. The `updated_run` is expected to contain the `run_id` or `client_key` used to
-    /// identify the run to update. The `field_mask` is a list of snake_cased field names used to
+    /// identify the run to update. The `update_mask` is a list of snake_cased field names used to
     /// indicate which fields should actually be updated. A list of valid field names can be found
-    /// at [`this link`]. The [Run] returned is the updated run. If `field_masks` is empty, then no
+    /// at [`this link`]. The [Run] returned is the updated run. If `update_mask` is empty, then no
     /// update is required and the `updated_run` is simply returned.
     ///
     /// [`this link`]: https://docs.siftstack.com/docs/api/grpc/protocol-buffers/runs#updaterunrequest
-    async fn try_update_run(&mut self, updated_run: Run, field_masks: &[String]) -> Result<Run> {
-        if field_masks.is_empty() {
+    async fn try_update_run(&mut self, updated_run: Run, update_mask: &[String]) -> Result<Run> {
+        if update_mask.is_empty() {
             return Ok(updated_run);
         }
 
         let run = self
             .update_run(UpdateRunRequest {
                 update_mask: Some(FieldMask {
-                    paths: field_masks.to_vec(),
+                    paths: update_mask.to_vec(),
                 }),
                 run: Some(updated_run),
             })

rust/crates/sift_stream/examples/quick-start/main.rs

@@ -1,3 +1,4 @@
+use sift_rs::metadata;
 use sift_stream::{
     ChannelConfig, ChannelDataType, ChannelValue, Credentials, Flow, FlowConfig,
     IngestionConfigForm, RecoveryStrategy, RunForm, SiftStreamBuilder, TimeValue,
@@ -53,12 +54,20 @@ async fn run() -> Result<(), Box<dyn Error>> {
         .map(|d| d.as_millis())
         .unwrap();
 
+    // Create metadata using the metadata macro
+    let metadata = metadata![
+        ("test_number", 5.0),
+        ("is_simulation", true),
+        ("location", "SiftHQ"),
+    ];
+
     // Define an optional run to group together data for this period of telemetry ingestion.
     let run = RunForm {
         name: format!("[MarsRover0].{ts}"),
         client_key: format!("mars-rover-sim-{ts}"),
         description: Some("simulation run".into()),
         tags: Some(vec!["simulation".into()]),
+        metadata: Some(metadata),
     };
 
     // Initialize your Sift Stream