[Rust] [Connector] Cleanup (#800)

This PR:
- removes the now unused data transformer
- adds some more logging around retries
- adds escapes to the retries that aren't critical. Some of these maybe
should be put back in later, but for now this will make our first drop
more stable

Author: vaavva

rust/azure_iot_operations_connector/src/base_connector/managed_azure_device_registry.rs

@@ -29,8 +29,8 @@ use crate::{
 };
 
 /// Used as the strategy when using [`tokio_retry2::Retry`]
-const RETRY_STRATEGY: tokio_retry2::strategy::ExponentialBackoff =
-    tokio_retry2::strategy::ExponentialBackoff::from_millis(100);
+const RETRY_STRATEGY: tokio_retry2::strategy::ExponentialFactorBackoff =
+    tokio_retry2::strategy::ExponentialFactorBackoff::from_millis(500, 2.0);
 
 /// An Observation for device endpoint creation events that uses
 /// multiple underlying clients to get full device endpoint information.
@@ -70,7 +70,7 @@ impl DeviceEndpointClientCreationObservation {
                 .await?;
 
             // and then get device update observation as well and turn it into a DeviceEndpointClientUpdateObservation
-            let device_endpoint_client_update_observation =  match Retry::spawn(RETRY_STRATEGY, async || -> Result<DeviceUpdateObservation, RetryError<azure_device_registry::Error>> {
+            let device_endpoint_client_update_observation =  match Retry::spawn(RETRY_STRATEGY.take(10), async || -> Result<DeviceUpdateObservation, RetryError<azure_device_registry::Error>> {
                 self.connector_context
                     .azure_device_registry_client
                     .observe_device_update_notifications(
@@ -110,6 +110,7 @@ impl DeviceEndpointClientCreationObservation {
                             match e.kind() {
                                 // network/retriable
                                 azure_device_registry::ErrorKind::AIOProtocolError(_) => {
+                                    log::warn!("Get device definition failed. Retrying: {e}");
                                     RetryError::transient(e)
                                 }
                                 // indicates an error in the configuration, so we want to get a new notification instead of retrying this operation
@@ -135,7 +136,7 @@ impl DeviceEndpointClientCreationObservation {
                     );
                     // unobserve as cleanup
                     let _ = Retry::spawn(
-                        RETRY_STRATEGY,
+                        RETRY_STRATEGY.take(10),
                         async || -> Result<(), RetryError<azure_device_registry::Error>> {
                             self.connector_context
                                 .azure_device_registry_client
@@ -175,7 +176,7 @@ impl DeviceEndpointClientCreationObservation {
                     );
                     // unobserve
                     let _ = Retry::spawn(
-                        RETRY_STRATEGY,
+                        RETRY_STRATEGY.take(10),
                         async || -> Result<(), RetryError<azure_device_registry::Error>> {
                             self.connector_context
                                 .azure_device_registry_client
@@ -325,7 +326,7 @@ impl DeviceEndpointClient {
     ) {
         // send status update to the service
         match Retry::spawn(
-            RETRY_STRATEGY,
+            RETRY_STRATEGY.take(10),
             async || -> Result<Device, RetryError<azure_device_registry::Error>> {
                 self.connector_context
                     .azure_device_registry_client
@@ -340,6 +341,7 @@ impl DeviceEndpointClient {
                         match e.kind() {
                             // network/retriable
                             azure_device_registry::ErrorKind::AIOProtocolError(_) => {
+                                log::warn!("Update device status failed. Retrying: {e}");
                                 RetryError::transient(e)
                             }
                             // indicates an error in the configuration, might be transient in the future depending on what it can indicate
@@ -428,7 +430,7 @@ impl AssetClientCreationObservation {
                 self.asset_create_observation.recv_notification().await?;
 
             // Get asset update observation as well and turn it into a AssetClientUpdateObservation
-            let asset_client_update_observation =  match Retry::spawn(RETRY_STRATEGY, async || -> Result<AssetUpdateObservation, RetryError<azure_device_registry::Error>> {
+            let asset_client_update_observation =  match Retry::spawn(RETRY_STRATEGY.take(10), async || -> Result<AssetUpdateObservation, RetryError<azure_device_registry::Error>> {
                 self.connector_context
                     .azure_device_registry_client
                     .observe_asset_update_notifications(
@@ -470,6 +472,7 @@ impl AssetClientCreationObservation {
                             match e.kind() {
                                 // network/retriable
                                 azure_device_registry::ErrorKind::AIOProtocolError(_) => {
+                                    log::warn!("Get asset definition failed. Retrying: {e}");
                                     RetryError::transient(e)
                                 }
                                 // indicates an error in the configuration, so we want to get a new notification instead of retrying this operation
@@ -502,7 +505,7 @@ impl AssetClientCreationObservation {
                     log::error!("Dropping asset create notification: {asset_ref:?}");
                     // unobserve as cleanup
                     let _ = Retry::spawn(
-                        RETRY_STRATEGY,
+                        RETRY_STRATEGY.take(10),
                         async || -> Result<(), RetryError<azure_device_registry::Error>> {
                             self.connector_context
                                 .azure_device_registry_client
@@ -717,7 +720,7 @@ impl AssetClient {
     ) {
         // send status update to the service
         match Retry::spawn(
-            RETRY_STRATEGY,
+            RETRY_STRATEGY.take(10),
             async || -> Result<Asset, RetryError<azure_device_registry::Error>> {
                 connector_context
                     .azure_device_registry_client
@@ -733,6 +736,7 @@ impl AssetClient {
                         match e.kind() {
                             // network/retriable
                             azure_device_registry::ErrorKind::AIOProtocolError(_) => {
+                                log::warn!("Update asset status failed. Retrying: {e}");
                                 RetryError::transient(e)
                             }
                             // indicates an error in the configuration, might be transient in the future depending on what it can indicate
@@ -884,6 +888,7 @@ impl DatasetClient {
                         match e.kind() {
                             // network/retriable
                             schema_registry::ErrorKind::AIOProtocolError(_) => {
+                                log::warn!("Reporting message schema failed. Retrying: {e}");
                                 RetryError::transient(e)
                             }
                             // indicates an error in the provided message schema, return to caller so they can fix

rust/azure_iot_operations_connector/src/data_transformer.rs

@@ -411,7 +411,7 @@ impl FileMountMap {
 
                 // Add the new asset to the tracked assets with the one shot sender so when the asset is
                 // deleted the channel is closed and the receiver is notified.
-                log::warn!("New asset: {asset:?}");
+                log::info!("New asset: {asset:?}");
                 tracked_assets.insert(asset.clone(), asset_deletion_tx);
 
                 // Notify that an asset has been created

rust/azure_iot_operations_connector/src/filemount/azure_device_registry.rs

@@ -13,7 +13,6 @@ use azure_iot_operations_services::{
 
 pub mod base_connector;
 pub mod data_processor;
-pub mod data_transformer;
 pub mod destination_endpoint;
 pub mod filemount;
 pub mod source_endpoint;