[Rust] [Connector] Forwarder/Destination Endpoint implementation (#797)

This PR:
- adds destination support framework
- adds destination support for both Mqtt Telemetry and Broker State
Store (Storage is stubbed, but not implemented)
- adds support for setting the destination for a dataset to be either
the one specified in the dataset or the default destination specified on
the asset
- adds reporting config errors for dataset/asset definitions that aren't
valid for their destination. Note that some things are guaranteed (ex.
if the destination is Mqtt, the topic must be present), but this is for
scenarios like the topic not being a valid MQTT topic, which the
operator doesn't validate or there not being a default destination or a
dataset destination specified.
- adds forwarding logic for Mqtt Telemetry and Broker State Store (and
reporting errors back if these don't succeed)
- adds a broker state store destination on the sample asset so both
destinations are tested
- updates the sample to work with an AIO deployment instead of the local
version (because Schema Registry is required)
- adds forwarding to the temporary sample
- temporarily fixes a bug in getting certs from the connector config,
but this should be fixed better in a future PR

Not present in this PR:
- full retry strategy, this needs more thought on what should be retried
within the base connector and what should be retried within the
application. For now, errors are reported, which aligns with our goals
for this first release

Author: vaavva

rust/Cargo.toml

@@ -27,6 +27,7 @@ module_name_repetitions = "allow"
 
 [workspace.dependencies]
 bytes = "1.5.0"
+chrono = "0.4"
 derive_builder = "0.20"
 derive-getters = { version = "0.5.0", features = ["auto_copy_getters"] }
 log = "0.4.21"                                                          # For performance, I believe this should have some filters on it

rust/azure_iot_operations_connector/Cargo.toml

@@ -15,6 +15,7 @@ publish = true
 azure_iot_operations_protocol = { version = "0.9", path = "../azure_iot_operations_protocol", registry = "aio-sdks"  }
 azure_iot_operations_services = { version = "0.8", path = "../azure_iot_operations_services", registry = "aio-sdks", features = ["state_store", "schema_registry", "azure_device_registry"]  }
 azure_iot_operations_mqtt = { version = "0.9", path = "../azure_iot_operations_mqtt", registry = "aio-sdks"  }
+chrono.workspace = true
 derive-getters.workspace = true
 jmespath = "0.3.0"
 log.workspace = true

rust/azure_iot_operations_connector/README.md

@@ -1,6 +1,6 @@
 # Instructions to deploy sample Rust connector
 
-The sample Rust connector that retrieves ADR definitions is currently under the `examples` folder of this crate.
+### The sample Rust connector that retrieves ADR definitions is currently under the `examples` folder of this crate.
 
 To deploy follow these steps (from the root of the crate):
 
@@ -11,13 +11,19 @@ To deploy follow these steps (from the root of the crate):
 5. Deploy a sample device: `kubectl apply -f examples/connector_get_adr_definitions_resources/thermostat-device-definition.yaml`
 6. Deploy a sample asset to the device: `kubectl apply -f examples/connector_get_adr_definitions_resources/rest-thermostat-asset-definition.yaml` 
 
-The sample Rust connector that uses the base connector to retrieve ADR definitions is currently under the `examples` folder of this crate.
+### The sample Rust connector that uses the base connector to retrieve ADR definitions and send transformed data is currently under the `examples` folder of this crate.
 
-To deploy follow these steps (from the root of the crate):
+#### Pre-requisites:
+1. Have AIO Deployed with necessary features
+1. Have an Azure Container Registry instance
 
+To deploy follow these steps (from the root of the crate):
 1. Create a binary release of the connector code: `cargo build --release --target-dir target_base_connector_sample --example base_connector_sample`
-2. Build the docker container: `docker build -t baseconnector:latest -f examples/base_connector_sample_resources/Dockerfile .`
-3. Import into your kubernetes cluster with: `k3d image import baseconnector:latest`
-4. Apply the connector template: `kubectl apply -f examples/base_connector_sample_resources/connector_template.yaml`
-5. Deploy a sample device: `kubectl apply -f examples/base_connector_sample_resources/thermostat-device-definition.yaml`
-6. Deploy a sample asset to the device: `kubectl apply -f examples/base_connector_sample_resources/rest-thermostat-asset-definition.yaml` 
+1. Build the docker container: `docker build -t baseconnector:latest -f examples/base_connector_sample_resources/Dockerfile .`
+1. Tag your docker image `docker tag baseconnector <your ACR name>.azurecr.io/baseconnector`
+1. Make sure you're logged into azure cli `az login`
+1. Login to your ACR `az acr login --name <your ACR name>`
+1. Upload to your ACR instance `docker push <your ACR name>.azurecr.io/baseconnector`
+1. Apply the connector template: `kubectl apply -f examples/base_connector_sample_resources/connector_template.yaml`
+1. Deploy a sample device: `kubectl apply -f examples/base_connector_sample_resources/thermostat-device-definition.yaml`
+1. Deploy a sample asset to the device: `kubectl apply -f examples/base_connector_sample_resources/rest-thermostat-asset-definition.yaml` 

rust/azure_iot_operations_connector/examples/base_connector_sample.rs

@@ -10,14 +10,19 @@
 //!
 //! To deploy and test this example, see instructions in `rust/azure_iot_operations_connector/README.md`
 
-use azure_iot_operations_connector::base_connector::{
-    BaseConnector,
-    managed_azure_device_registry::{
-        AssetClientCreationObservation, DatasetClient, DeviceEndpointClientCreationObservation,
+use std::time::Duration;
+
+use azure_iot_operations_connector::{
+    AdrConfigError, Data,
+    base_connector::{
+        BaseConnector,
+        managed_azure_device_registry::{
+            AssetClientCreationObservation, DatasetClient, DeviceEndpointClientCreationObservation,
+        },
     },
+    data_processor::derived_json,
 };
 use azure_iot_operations_protocol::application::ApplicationContextBuilder;
-use azure_iot_operations_services::{azure_device_registry, schema_registry};
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -79,9 +84,9 @@ async fn run_program(mut device_creation_observation: DeviceEndpointClientCreati
                         .name,
                     unsupported_endpoint_type
                 );
-                Err(azure_device_registry::ConfigError {
+                Err(AdrConfigError {
                     message: Some("endpoint type is not supported".to_string()),
-                    ..azure_device_registry::ConfigError::default()
+                    ..AdrConfigError::default()
                 })
             }
         };
@@ -115,9 +120,9 @@ async fn run_assets(mut asset_creation_observation: AssetClientCreationObservati
                     "Asset '{}' not accepted. Manufacturer '{m}' not supported.",
                     asset_client.asset_ref().name
                 );
-                Err(azure_device_registry::ConfigError {
+                Err(AdrConfigError {
                     message: Some("asset manufacturer type is not supported".to_string()),
-                    ..azure_device_registry::ConfigError::default()
+                    ..AdrConfigError::default()
                 })
             }
         };
@@ -141,36 +146,54 @@ async fn run_dataset(dataset_client: DatasetClient) {
 
     // now we should update the status of the dataset and report the message schema
     dataset_client.report_status(Ok(())).await;
-    match dataset_client
-        .report_message_schema(
-            schema_registry::PutRequestBuilder::default()
-                .content(
-                    r#"
-{
-  "$schema": "http://json-schema.org/draft-07/schema#",
-  "type": "object",
-  "properties": {
-    "currentTemperature": {
-      "type": "number"
-    },
-    "desiredTemperature": {
-      "type": "number"
-    }
-  }
-}
-"#,
-                )
-                .format(schema_registry::Format::JsonSchemaDraft07)
-                .build()
-                .unwrap(),
-        )
-        .await
-    {
+
+    let sample_data = mock_received_data(0);
+
+    let (_, message_schema) =
+        derived_json::transform(sample_data, dataset_client.dataset_definition()).unwrap();
+    match dataset_client.report_message_schema(message_schema).await {
         Ok(message_schema_reference) => {
             log::info!("Message Schema reported, reference returned: {message_schema_reference:?}");
         }
         Err(e) => {
             log::error!("Error reporting message schema: {e}");
         }
     }
+    let mut count = 0;
+    loop {
+        tokio::time::sleep(Duration::from_secs(5)).await;
+        let sample_data = mock_received_data(count);
+        let (transformed_data, _) =
+            derived_json::transform(sample_data.clone(), dataset_client.dataset_definition())
+                .unwrap();
+        match dataset_client.forward_data(transformed_data).await {
+            Ok(()) => {
+                log::info!(
+                    "data {} for {} forwarded",
+                    String::from_utf8(sample_data.payload).unwrap(),
+                    dataset_client.dataset_ref().dataset_name
+                );
+                count += 1;
+            }
+            Err(e) => log::error!("error forwarding data: {e}"),
+        }
+    }
+}
+
+#[must_use]
+pub fn mock_received_data(count: u32) -> Data {
+    Data {
+        // temp and newTemp
+        payload: format!(
+            r#"{{
+            "temp": {count},
+            "newTemp": {}
+        }}"#,
+            count * 2
+        )
+        .into(),
+        content_type: "application/json".to_string(),
+        custom_user_data: Vec::new(),
+        timestamp: None,
+    }
 }

rust/azure_iot_operations_connector/examples/base_connector_sample_resources/connector_template.yaml

@@ -19,13 +19,15 @@ spec:
     managedConfigurationSettings:
       managedConfigurationType: imageConfiguration
       imageConfigurationSettings:
+        registrySettings:
+          registrySettingsType: containerRegistry
+          containerRegistrySettings:
+            registry: "<your acr>.azurecr.io"
         imageName: "baseconnector"
-        imagePullPolicy: Never # This sample image is built locally, so no need to pull
+        imagePullPolicy: Always
         replicas: 1
         tagDigestSettings:
           tagDigestType: tag
           tag: "latest"
   mqttConnectionConfiguration:
-    host: "aio-broker:1883"
-    tls:
-      mode: Disabled
+    host: "aio-broker:18883"

rust/azure_iot_operations_connector/examples/base_connector_sample_resources/rest-thermostat-asset-definition.yaml

@@ -10,13 +10,13 @@ spec:
   datasets:
     - name: thermostat_status
       dataPoints:
-        - dataSource: /api/thermostat/current
+        - dataSource: temp
           name: currentTemperature
           dataPointConfiguration: |-
            {
               "HttpRequestMethod": "GET",
            }
-        - dataSource: /api/thermostat/desired
+        - dataSource: newTemp
           name: desiredTemperature
           dataPointConfiguration: |-
            {
@@ -25,8 +25,27 @@ spec:
       dataSource: /thermostat
       destinations:
         - configuration:
-            topic: /mqtt/machine/status
+            topic: mqtt/machine/status
           target: Mqtt
+    - name: oven_status
+      dataPoints:
+        - dataSource: temp
+          name: currentTemperature
+          dataPointConfiguration: |-
+           {
+              "HttpRequestMethod": "GET",
+           }
+        - dataSource: newTemp
+          name: desiredTemperature
+          dataPointConfiguration: |-
+           {
+              "HttpRequestMethod": "GET",
+           }
+      dataSource: /oven
+      destinations:
+        - configuration:
+            key: oventemp
+          target: BrokerStateStore
   defaultDatasetsConfiguration: |-
    {
       "samplingInterval": 4000,