feat: implement retries (#103)

Azure Application Insights frequently times out, returns HTTP 503
errors, or only accepts partial payloads. This adds transparent retries.
Users of this crate won't have to configure anything.

Using `backon` crate, because it seems small and easy to use.

Design decisions:

- Don't expose implementation. Add default retries that should work for
  everyone:

  - min_delay = 500ms
  - max_delay = 5s
  - with jitter (to avoid thundering herds when multiple exports run in parallel)

  We wanted no max_times or total delay, because the BatchSpanProcessor
  already provides a `max_export_timeout` option. This automatically
  ensures that the exporter will stop. However, it's only available in the
  experimental async version of the BatchSpanProcessor. And it's missing
  in the SimpleSpanProcessor and metric and logs processors. Therefore
  this sets a total delay of 35s.

- Add a `with_retry_notify` option. Can be used to log retries and
  thereby debug exporter flakyness.

- Use `futures-timer` crate to provide sleep functionality during
  retries. Ideally we could use
  opentelemetry_sdk::runtime::Runtime::delay for that. But that's behind
  an experimental feature and we the SDK doesn't provide the exporters
  with a runtime. We'd have to add a configuration to the exporter where
  users would need to specify their choosen runtime, which then has to
  fit to the rest of their setup, otherwise they see runtime errors.

  E.g. tokio-sleep fails if not executed in the context of a Tokio
  reactor with:

      there is no reactor running, must be called from the context of a Tokio 1.x runtime

  This error would also only happen if a retry has to happen, which
  means when a call to AppInsights fails. This would make it very hard
  to debug.

  We're using futures-timer instead, which works with all runtimes,
  including tokio and futures-executor. The OpenTelemetry SDK uses the
  latter for most default processors right now.

---------

Co-authored-by: Jan Kuehle <[email protected]>

Author: alexbrt

Cargo.toml

@@ -47,6 +47,7 @@ reqwest-client-rustls = ["opentelemetry-http/reqwest", "reqwest/rustls-tls"]
 
 [dependencies]
 async-trait = "0.1"
+backon = { version = "1.5", default-features = false, features = ["futures-timer-sleep"] }
 bytes = "1"
 chrono = "0.4"
 flate2 = "1"

src/lib.rs

@@ -372,7 +372,13 @@ use opentelemetry_sdk::ExportError;
 use opentelemetry_sdk::Resource;
 #[cfg(feature = "live-metrics")]
 pub use quick_pulse::LiveMetricsSpanProcessor;
-use std::{convert::TryInto, error::Error as StdError, fmt::Debug, sync::Arc};
+use std::{
+    convert::TryInto,
+    error::Error as StdError,
+    fmt::Debug,
+    sync::{Arc, Mutex},
+    time::Duration,
+};
 #[cfg(feature = "live-metrics")]
 use uploader_quick_pulse::PostOrPing;
 
@@ -386,6 +392,7 @@ pub struct Exporter<C> {
     #[cfg(feature = "live-metrics")]
     live_ping_endpoint: http::Uri,
     instrumentation_key: String,
+    retry_notify: Option<Arc<Mutex<dyn FnMut(&Error, Duration) + Send + 'static>>>,
     #[cfg(feature = "trace")]
     sample_rate: f64,
     #[cfg(any(feature = "trace", feature = "logs"))]
@@ -436,6 +443,7 @@ impl<C> Exporter<C> {
                 &instrumentation_key,
             ),
             instrumentation_key,
+            retry_notify: None,
             #[cfg(feature = "trace")]
             sample_rate: 100.0,
             #[cfg(any(feature = "trace", feature = "logs"))]
@@ -475,6 +483,7 @@ impl<C> Exporter<C> {
                 &connection_string.instrumentation_key,
             ),
             instrumentation_key: connection_string.instrumentation_key,
+            retry_notify: None,
             #[cfg(feature = "trace")]
             sample_rate: 100.0,
             #[cfg(any(feature = "trace", feature = "logs"))]
@@ -484,6 +493,16 @@ impl<C> Exporter<C> {
         })
     }
 
+    /// Set a retry notification function that is called when a request to upload telemetry to
+    /// Application Insights failed and will be retried.
+    pub fn with_retry_notify<N>(mut self, retry_notify: N) -> Self
+    where
+        N: FnMut(&Error, Duration) + Send + 'static,
+    {
+        self.retry_notify = Some(Arc::new(Mutex::new(retry_notify)));
+        self
+    }
+
     /// Set endpoint used to ingest telemetry. This should consist of scheme and authrity. The
     /// exporter will call `/v2/track` on the specified endpoint.
     ///

src/logs.rs

@@ -83,9 +83,14 @@ where
             .collect();
 
         async move {
-            crate::uploader::send(client.as_ref(), endpoint.as_ref(), envelopes)
-                .await
-                .map_err(Into::into)
+            crate::uploader::send(
+                client.as_ref(),
+                endpoint.as_ref(),
+                envelopes,
+                self.retry_notify.clone(),
+            )
+            .await
+            .map_err(Into::into)
         }
     }
 

src/metrics.rs

@@ -70,9 +70,14 @@ where
         }
 
         async move {
-            crate::uploader::send(client.as_ref(), endpoint.as_ref(), envelopes)
-                .await
-                .map_err(Into::into)
+            crate::uploader::send(
+                client.as_ref(),
+                endpoint.as_ref(),
+                envelopes,
+                self.retry_notify.clone(),
+            )
+            .await
+            .map_err(Into::into)
         }
     }
 

src/trace.rs

@@ -145,9 +145,14 @@ where
             .flat_map(|span| self.create_envelopes_for_span(span, &self.resource))
             .collect();
 
-        crate::uploader::send(client.as_ref(), endpoint.as_ref(), envelopes)
-            .await
-            .map_err(Into::into)
+        crate::uploader::send(
+            client.as_ref(),
+            endpoint.as_ref(),
+            envelopes,
+            self.retry_notify.clone(),
+        )
+        .await
+        .map_err(Into::into)
     }
 
     fn set_resource(&mut self, resource: &Resource) {