Change ExportResult to std::result::Result (#347)

The spec changed and ExportResult no longer includes the hint about
error being retriable. Since it now only distinguishes between success
and error, it makes sense to use Rust's standard Result type.

Author: frigus02

opentelemetry-contrib/src/trace/exporter/datadog/mod.rs

@@ -86,13 +86,13 @@
 //!
 //! #[async_trait]
 //! impl HttpClient for IsahcClient {
-//!   async fn send(&self, request: http::Request<Vec<u8>>) -> Result<ExportResult, Box<dyn Error + Send + Sync + 'static>> {
+//!   async fn send(&self, request: http::Request<Vec<u8>>) -> ExportResult {
 //!     let result = self.0.send_async(request).await?;
 //!
 //!     if result.status().is_success() {
-//!       Ok(ExportResult::Success)
+//!       Ok(())
 //!     } else {
-//!       Ok(ExportResult::FailedNotRetryable)
+//!       Err(result.status().as_str().into())
 //!     }
 //!   }
 //! }
@@ -279,24 +279,13 @@ impl DatadogPipelineBuilder {
 impl trace::SpanExporter for DatadogExporter {
     /// Export spans to datadog-agent
     async fn export(&self, batch: Vec<SpanData>) -> trace::ExportResult {
-        let data = match self.version.encode(&self.service_name, batch) {
-            Ok(data) => data,
-            Err(_) => return trace::ExportResult::FailedNotRetryable,
-        };
-
-        let req = match Request::builder()
+        let data = self.version.encode(&self.service_name, batch)?;
+        let req = Request::builder()
             .method(Method::POST)
             .uri(self.request_url.clone())
             .header(http::header::CONTENT_TYPE, self.version.content_type())
-            .body(data)
-        {
-            Ok(req) => req,
-            _ => return trace::ExportResult::FailedNotRetryable,
-        };
-        self.client
-            .send(req)
-            .await
-            .unwrap_or(trace::ExportResult::FailedNotRetryable)
+            .body(data)?;
+        self.client.send(req).await
     }
 }
 

opentelemetry-jaeger/src/uploader.rs

@@ -18,19 +18,16 @@ impl BatchUploader {
     /// Emit a jaeger batch for the given uploader
     pub(crate) async fn upload(&self, batch: jaeger::Batch) -> trace::ExportResult {
         match self {
-            BatchUploader::Agent(client) => match client.emit_batch(batch).await {
-                Ok(_) => trace::ExportResult::Success,
-                // TODO determine if the error is retryable
-                Err(_) => trace::ExportResult::FailedNotRetryable,
-            },
+            BatchUploader::Agent(client) => {
+                // TODO Implement retry behaviour
+                client.emit_batch(batch).await?;
+            }
             #[cfg(feature = "collector_client")]
             BatchUploader::Collector(collector) => {
-                match collector.submit_batch(batch).await {
-                    Ok(_) => trace::ExportResult::Success,
-                    // TODO determine if the error is retryable
-                    Err(_) => trace::ExportResult::FailedNotRetryable,
-                }
+                // TODO Implement retry behaviour
+                collector.submit_batch(batch).await?;
             }
-        }
+        };
+        Ok(())
     }
 }

opentelemetry-otlp/src/span.rs

@@ -7,7 +7,6 @@ use async_trait::async_trait;
 use grpcio::{
     CallOption, Channel, ChannelBuilder, ChannelCredentialsBuilder, Environment, MetadataBuilder,
 };
-use opentelemetry::exporter::trace::ExportResult::{FailedNotRetryable, Success};
 use opentelemetry::exporter::trace::{ExportResult, SpanData, SpanExporter};
 use protobuf::RepeatedField;
 use std::collections::HashMap;
@@ -154,7 +153,7 @@ impl SpanExporter for Exporter {
             cached_size: Default::default(),
         };
 
-        let mut call_options: CallOption = CallOption::default().timeout(self.timeout);
+        let mut call_options = CallOption::default().timeout(self.timeout);
 
         if let Some(headers) = self.headers.clone() {
             let mut metadata_builder: MetadataBuilder = MetadataBuilder::new();
@@ -166,12 +165,10 @@ impl SpanExporter for Exporter {
             call_options = call_options.headers(metadata_builder.build());
         }
 
-        match self.trace_exporter.export_async_opt(&request, call_options) {
-            Ok(receiver) => match receiver.await {
-                Ok(_) => Success,
-                Err(_) => FailedNotRetryable,
-            },
-            Err(_) => FailedNotRetryable,
-        }
+        let receiver = self
+            .trace_exporter
+            .export_async_opt(&request, call_options)?;
+        receiver.await?;
+        Ok(())
     }
 }

opentelemetry-zipkin/README.md

@@ -109,13 +109,13 @@ struct IsahcClient(isahc::HttpClient);
 
 #[async_trait]
 impl HttpClient for IsahcClient {
-  async fn send(&self, request: http::Request<Vec<u8>>) -> Result<ExportResult, Box<dyn Error + Send + Sync + 'static>> {
+  async fn send(&self, request: http::Request<Vec<u8>>) -> ExportResult {
     let result = self.0.send_async(request).await?;
 
     if result.status().is_success() {
-      Ok(ExportResult::Success)
+      Ok(())
     } else {
-      Ok(ExportResult::FailedNotRetryable)
+      Err(result.status().as_str().into())
     }
   }
 }

opentelemetry-zipkin/src/lib.rs

@@ -89,13 +89,13 @@
 //!
 //! #[async_trait]
 //! impl HttpClient for IsahcClient {
-//!   async fn send(&self, request: http::Request<Vec<u8>>) -> Result<ExportResult, Box<dyn Error + Send + Sync + 'static>> {
+//!   async fn send(&self, request: http::Request<Vec<u8>>) -> ExportResult {
 //!     let result = self.0.send_async(request).await?;
 //!
 //!     if result.status().is_success() {
-//!       Ok(ExportResult::Success)
+//!       Ok(())
 //!     } else {
-//!       Ok(ExportResult::FailedNotRetryable)
+//!       Err(result.status().as_str().into())
 //!     }
 //!   }
 //! }

opentelemetry-zipkin/src/uploader.rs

@@ -21,10 +21,7 @@ impl Uploader {
     /// Upload spans to Zipkin
     pub(crate) async fn upload(&self, spans: Vec<Span>) -> ExportResult {
         match self {
-            Uploader::Http(client) => client
-                .upload(spans)
-                .await
-                .unwrap_or(ExportResult::FailedNotRetryable),
+            Uploader::Http(client) => client.upload(spans).await,
         }
     }
 }
@@ -36,10 +33,7 @@ pub(crate) struct JsonV2Client {
 }
 
 impl JsonV2Client {
-    async fn upload(
-        &self,
-        spans: Vec<Span>,
-    ) -> Result<ExportResult, Box<dyn std::error::Error + Send + Sync + 'static>> {
+    async fn upload(&self, spans: Vec<Span>) -> ExportResult {
         let req = Request::builder()
             .method(Method::POST)
             .uri(self.collector_endpoint.clone())

opentelemetry/src/api/trace/noop.rs

@@ -182,7 +182,7 @@ impl NoopSpanExporter {
 #[async_trait]
 impl SpanExporter for NoopSpanExporter {
     async fn export(&self, _batch: Vec<SpanData>) -> ExportResult {
-        ExportResult::Success
+        Ok(())
     }
 }
 

opentelemetry/src/exporter/trace/mod.rs

@@ -4,28 +4,20 @@ use crate::{
     trace::{Event, Link, SpanContext, SpanId, SpanKind, StatusCode},
 };
 use async_trait::async_trait;
+#[cfg(feature = "http")]
+use http::Request;
 #[cfg(feature = "serialize")]
 use serde::{Deserialize, Serialize};
 #[cfg(all(feature = "http", feature = "reqwest"))]
 use std::convert::TryInto;
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::SystemTime;
-#[cfg(feature = "http")]
-use {http::Request, std::error::Error};
 
 pub mod stdout;
 
 /// Describes the result of an export.
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub enum ExportResult {
-    /// Batch is successfully exported.
-    Success,
-    /// Batch export failed. Caller must not retry.
-    FailedNotRetryable,
-    /// Batch export failed transiently. Caller should record error and may retry.
-    FailedRetryable,
-}
+pub type ExportResult = Result<(), Box<dyn std::error::Error + Send + Sync + 'static>>;
 
 /// `SpanExporter` defines the interface that protocol-specific exporters must
 /// implement so that they can be plugged into OpenTelemetry SDK and support
@@ -75,10 +67,7 @@ pub trait SpanExporter: Send + Sync + std::fmt::Debug {
 #[async_trait]
 pub trait HttpClient: Debug + Send + Sync {
     /// Send a batch of spans to collectors
-    async fn send(
-        &self,
-        request: Request<Vec<u8>>,
-    ) -> Result<ExportResult, Box<dyn Error + Send + Sync + 'static>>;
+    async fn send(&self, request: Request<Vec<u8>>) -> ExportResult;
 }
 
 /// `SpanData` contains all the information collected by a `Span` and can be used
@@ -118,44 +107,28 @@ pub struct SpanData {
 #[cfg(all(feature = "reqwest", feature = "http"))]
 #[async_trait]
 impl HttpClient for reqwest::Client {
-    async fn send(
-        &self,
-        request: Request<Vec<u8>>,
-    ) -> Result<ExportResult, Box<dyn Error + Send + Sync + 'static>> {
-        let result = self.execute(request.try_into()?).await?;
-
-        if result.status().is_success() {
-            Ok(ExportResult::Success)
-        } else {
-            Ok(ExportResult::FailedNotRetryable)
-        }
+    async fn send(&self, request: Request<Vec<u8>>) -> ExportResult {
+        let _result = self
+            .execute(request.try_into()?)
+            .await?
+            .error_for_status()?;
+        Ok(())
     }
 }
 
 #[cfg(all(feature = "reqwest", feature = "http"))]
 #[async_trait]
 impl HttpClient for reqwest::blocking::Client {
-    async fn send(
-        &self,
-        request: Request<Vec<u8>>,
-    ) -> Result<ExportResult, Box<dyn Error + Send + Sync + 'static>> {
-        let result = self.execute(request.try_into()?)?;
-
-        if result.status().is_success() {
-            Ok(ExportResult::Success)
-        } else {
-            Ok(ExportResult::FailedNotRetryable)
-        }
+    async fn send(&self, request: Request<Vec<u8>>) -> ExportResult {
+        let _result = self.execute(request.try_into()?)?.error_for_status()?;
+        Ok(())
     }
 }
 
 #[cfg(all(feature = "surf", feature = "http"))]
 #[async_trait]
 impl HttpClient for surf::Client {
-    async fn send(
-        &self,
-        request: Request<Vec<u8>>,
-    ) -> Result<ExportResult, Box<dyn Error + Send + Sync + 'static>> {
+    async fn send(&self, request: Request<Vec<u8>>) -> ExportResult {
         let (parts, body) = request.into_parts();
         let uri = parts.uri.to_string().parse()?;
 
@@ -165,9 +138,9 @@ impl HttpClient for surf::Client {
         let result = self.send(req).await?;
 
         if result.status().is_success() {
-            Ok(ExportResult::Success)
+            Ok(())
         } else {
-            Ok(ExportResult::FailedNotRetryable)
+            Err(result.status().canonical_reason().into())
         }
     }
 }

opentelemetry/src/exporter/trace/stdout.rs

@@ -129,28 +129,19 @@ where
 {
     /// Export spans to stdout
     async fn export(&self, batch: Vec<SpanData>) -> ExportResult {
-        let writer = self
+        let mut writer = self
             .writer
             .lock()
-            .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()));
-        let result = writer.and_then(|mut w| {
-            for span in batch {
-                if self.pretty_print {
-                    w.write_all(format!("{:#?}\n", span).as_bytes())?;
-                } else {
-                    w.write_all(format!("{:?}\n", span).as_bytes())?;
-                }
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?;
+        for span in batch {
+            if self.pretty_print {
+                writer.write_all(format!("{:#?}\n", span).as_bytes())?;
+            } else {
+                writer.write_all(format!("{:?}\n", span).as_bytes())?;
             }
-
-            Ok(())
-        });
-
-        if result.is_ok() {
-            ExportResult::Success
-        } else {
-            // FIXME: determine retryable io::Error types
-            ExportResult::FailedNotRetryable
         }
+
+        Ok(())
     }
 }
 

opentelemetry/src/sdk/trace/span_processor.rs

@@ -110,7 +110,8 @@ impl SpanProcessor for SimpleSpanProcessor {
     }
 
     fn on_end(&self, span: SpanData) {
-        executor::block_on(self.exporter.export(vec![span]));
+        // TODO: Surface error through global error handler
+        let _result = executor::block_on(self.exporter.export(vec![span]));
     }
 
     fn shutdown(&mut self) {
@@ -239,7 +240,9 @@ impl BatchSpanProcessor {
                             let batch = spans.split_off(
                                 spans.len().saturating_sub(config.max_export_batch_size),
                             );
-                            exporter.export(batch).await;
+
+                            // TODO: Surface error through global error handler
+                            let _result = exporter.export(batch).await;
                         }
                     }
                     // Stream has terminated or processor is shutdown, return to finish execution.
@@ -248,7 +251,9 @@ impl BatchSpanProcessor {
                             let batch = spans.split_off(
                                 spans.len().saturating_sub(config.max_export_batch_size),
                             );
-                            exporter.export(batch).await;
+
+                            // TODO: Surface error through global error handler
+                            let _result = exporter.export(batch).await;
                         }
                         exporter.shutdown();
                         break;