Implement timeout for force_flush and shutdown.

opentelemetry/Cargo.toml

@@ -21,17 +21,17 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [dependencies]
-async-std = { version = "1.6", features = ["unstable"], default-features = false, optional = true  }
+async-std = { version = "1.6", features = ["unstable"], default-features = false, optional = true }
 async-trait = { version = "0.1", optional = true }
 bincode = { version = "1.2", optional = true }
 dashmap = { version = "4.0.0-rc6", optional = true }
 fnv = { version = "1.0", optional = true }
-futures = "0.3"
+futures = { version = "0.3", features = ["std"] }
 lazy_static = "1.4"
 percent-encoding = { version = "2.0", optional = true }
 pin-project = { version = "0.4", optional = true }
 rand = { version = "0.7", default-features = false, features = ["std"], optional = true }
-regex = { version = "1.3", default-features = false, features = ["std", "perf"], optional = true}
+regex = { version = "1.3", default-features = false, features = ["std", "perf"], optional = true }
 serde = { version = "1.0", features = ["derive", "rc"], optional = true }
 http = { version = "0.2", optional = true }
 thiserror = { version = "1.0", optional = true }
@@ -44,10 +44,11 @@ surf = { version = "2.0", default-features = false, optional = true }
 criterion = "0.3.1"
 rand_distr = "0.3.0"
 tokio = { version = "0.2", features = ["full"] }
+async-std = { version = "1.6", features = ["unstable"] }
 
 [features]
 default = ["trace"]
-trace = ["rand", "pin-project", "async-trait", "regex", "percent-encoding"]
+trace = ["rand", "pin-project", "async-trait", "regex", "percent-encoding", "thiserror"]
 metrics = ["thiserror", "dashmap", "fnv"]
 serialize = ["serde", "bincode"]
 

opentelemetry/src/api/trace/mod.rs

@@ -136,3 +136,14 @@ pub use self::{
     },
     tracer::{SpanBuilder, Tracer},
 };
+
+use thiserror::Error;
+
+/// Errors returned by the trace API.
+#[derive(Error, Debug, PartialEq)]
+#[non_exhaustive]
+pub enum TraceError {
+    /// Other errors not covered by specific cases.
+    #[error("Trace error: {0}")]
+    Other(String),
+}

opentelemetry/src/exporter/trace/mod.rs

@@ -10,7 +10,8 @@ use http::Request;
 use serde::{Deserialize, Serialize};
 #[cfg(all(feature = "http", feature = "reqwest"))]
 use std::convert::TryInto;
-use std::fmt::Debug;
+use std::error::Error;
+use std::fmt::{Debug, Display};
 use std::sync::Arc;
 use std::time::SystemTime;
 
@@ -19,6 +20,18 @@ pub mod stdout;
 /// Describes the result of an export.
 pub type ExportResult = Result<(), Box<dyn std::error::Error + Send + Sync + 'static>>;
 
+/// Timed out when exporting spans to remote
+#[derive(Debug, Default)]
+pub struct ExportTimedOutError {}
+
+impl Display for ExportTimedOutError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("export timed out")
+    }
+}
+
+impl Error for ExportTimedOutError {}
+
 /// `SpanExporter` defines the interface that protocol-specific exporters must
 /// implement so that they can be plugged into OpenTelemetry SDK and support
 /// sending of telemetry data.

opentelemetry/src/sdk/trace/provider.rs

@@ -25,7 +25,7 @@ pub(crate) struct TracerProviderInner {
 impl Drop for TracerProviderInner {
     fn drop(&mut self) {
         for processor in &mut self.processors {
-            processor.shutdown();
+            let _result = processor.shutdown();
         }
     }
 }
@@ -116,7 +116,12 @@ impl Builder {
         // drop. We cannot assume we are in a multi-threaded tokio runtime here, so use
         // `spawn_blocking` to avoid blocking the main thread.
         let spawn = |fut| tokio::task::spawn_blocking(|| futures::executor::block_on(fut));
-        let batch = sdk::trace::BatchSpanProcessor::builder(exporter, spawn, tokio::time::interval);
+        let batch = sdk::trace::BatchSpanProcessor::builder(
+            exporter,
+            spawn,
+            tokio::time::delay_for,
+            tokio::time::interval,
+        );
         self.with_batch_exporter(batch.build())
     }
 
@@ -127,6 +132,7 @@ impl Builder {
         let batch = sdk::trace::BatchSpanProcessor::builder(
             exporter,
             async_std::task::spawn,
+            async_std::task::sleep,
             async_std::stream::interval,
         );
         self.with_batch_exporter(batch.build())