opentelemetry-zipkin: Add Environment Variables (#718)

As described in https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/sdk-environment-variables.md
* Add support for:
** OTEL_EXPORTER_ZIPKIN_ENDPOINT
** OTEL_EXPORTER_ZIPKIN_TIMEOUT
* Update Default to use
* Add remove_var to flaky test

Closes #534

Signed-off-by: Harold Dost <h.dost@criteo.com>

opentelemetry-zipkin/CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## unreleased
+
+## Added
+
+- Add support for OTEL_EXPORTER_ZIPKIN_* variables. #718
+
+## Changed
+
+- Add defaults for timeouts to HTTP clients #718
+
 ## v0.15.0
 
 ### Changed
@@ -116,5 +126,5 @@
 
 ### Added
 
-- Exporter to Zipkin collector through HTTP API 
+- Exporter to Zipkin collector through HTTP API
 

opentelemetry-zipkin/src/exporter/env.rs

@@ -0,0 +1,63 @@
+use std::env;
+use std::time::Duration;
+
+/// Default Zipkin collector endpoint
+const DEFAULT_COLLECTOR_ENDPOINT: &str = "http://127.0.0.1:9411/api/v2/spans";
+
+/// HTTP endpoint for Zipkin collector.
+/// e.g. "http://localhost:9411/api/v2/spans"
+const ENV_ENDPOINT: &str = "OTEL_EXPORTER_ZIPKIN_ENDPOINT";
+
+/// Maximum time the Zipkin exporter will wait for each batch export
+const ENV_TIMEOUT: &str = "OTEL_EXPORTER_ZIPKIN_TIMEOUT";
+
+/// Default Zipkin timeout in milliseconds
+const DEFAULT_COLLECTOR_TIMEOUT: Duration = Duration::from_millis(10_000);
+
+pub(crate) fn get_timeout() -> Duration {
+    match env::var(ENV_TIMEOUT).ok().filter(|var| !var.is_empty()) {
+        Some(timeout) => match timeout.parse() {
+            Ok(timeout) => Duration::from_millis(timeout),
+            Err(e) => {
+                eprintln!("{} malformed defaulting to 10000: {}", ENV_TIMEOUT, e);
+                DEFAULT_COLLECTOR_TIMEOUT
+            }
+        },
+        None => DEFAULT_COLLECTOR_TIMEOUT,
+    }
+}
+
+pub(crate) fn get_endpoint() -> String {
+    match env::var(ENV_ENDPOINT).ok().filter(|var| !var.is_empty()) {
+        Some(endpoint) => endpoint,
+        None => DEFAULT_COLLECTOR_ENDPOINT.to_string(),
+    }
+}
+
+#[test]
+fn test_collector_defaults() {
+    // Ensure the variables are undefined.
+    env::remove_var(ENV_TIMEOUT);
+    env::remove_var(ENV_ENDPOINT);
+    assert_eq!(DEFAULT_COLLECTOR_TIMEOUT, get_timeout());
+    assert_eq!(DEFAULT_COLLECTOR_ENDPOINT, get_endpoint());
+}
+
+#[test]
+fn test_collector_bad_timeout() {
+    env::set_var(ENV_TIMEOUT, "a");
+    assert_eq!(DEFAULT_COLLECTOR_TIMEOUT, get_timeout());
+}
+
+#[test]
+fn test_collector_good_timeout() {
+    env::set_var(ENV_TIMEOUT, "777");
+    assert_eq!(Duration::from_millis(777), get_timeout());
+}
+
+#[test]
+fn test_collector_custom_endpoint() {
+    let custom_endpoint = "https://example.com/api/v2/spans";
+    env::set_var(ENV_ENDPOINT, custom_endpoint);
+    assert_eq!(custom_endpoint, get_endpoint());
+}

opentelemetry-zipkin/src/exporter/mod.rs

@@ -1,3 +1,4 @@
+mod env;
 mod model;
 mod uploader;
 
@@ -17,13 +18,16 @@ use opentelemetry::{
 };
 use opentelemetry_http::HttpClient;
 use opentelemetry_semantic_conventions as semcov;
+#[cfg(all(
+    not(feature = "reqwest-client"),
+    not(feature = "reqwest-blocking-client"),
+    feature = "surf-client"
+))]
+use std::convert::TryFrom;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
-/// Default Zipkin collector endpoint
-const DEFAULT_COLLECTOR_ENDPOINT: &str = "http://127.0.0.1:9411/api/v2/spans";
-
 /// Zipkin span exporter
 #[derive(Debug)]
 pub struct Exporter {
@@ -57,21 +61,35 @@ pub struct ZipkinPipelineBuilder {
 
 impl Default for ZipkinPipelineBuilder {
     fn default() -> Self {
+        let timeout = env::get_timeout();
         ZipkinPipelineBuilder {
             #[cfg(feature = "reqwest-blocking-client")]
-            client: Some(Box::new(reqwest::blocking::Client::new())),
+            client: Some(Box::new(
+                reqwest::blocking::Client::builder()
+                    .timeout(timeout)
+                    .build()
+                    .unwrap_or_else(|_| reqwest::blocking::Client::new()),
+            )),
             #[cfg(all(
                 not(feature = "reqwest-blocking-client"),
                 not(feature = "surf-client"),
                 feature = "reqwest-client"
             ))]
-            client: Some(Box::new(reqwest::Client::new())),
+            client: Some(Box::new(
+                reqwest::Client::builder()
+                    .timeout(timeout)
+                    .build()
+                    .unwrap_or_else(|_| reqwest::Client::new()),
+            )),
             #[cfg(all(
                 not(feature = "reqwest-client"),
                 not(feature = "reqwest-blocking-client"),
                 feature = "surf-client"
             ))]
-            client: Some(Box::new(surf::Client::new())),
+            client: Some(Box::new(
+                surf::Client::try_from(surf::Config::new().set_timeout(Some(timeout)))
+                    .unwrap_or_else(|_| surf::Client::new()),
+            )),
             #[cfg(all(
                 not(feature = "reqwest-client"),
                 not(feature = "surf-client"),
@@ -81,7 +99,7 @@ impl Default for ZipkinPipelineBuilder {
 
             service_name: None,
             service_addr: None,
-            collector_endpoint: DEFAULT_COLLECTOR_ENDPOINT.to_string(),
+            collector_endpoint: env::get_endpoint(),
             trace_config: None,
         }
     }

opentelemetry/src/sdk/resource/env.rs

@@ -132,6 +132,8 @@ mod tests {
     #[test]
     fn test_sdk_provided_resource_detector() {
         const SERVICE_NAME: &str = "service.name";
+        // Ensure no env var set
+        env::remove_var(OTEL_RESOURCE_ATTRIBUTES);
         let no_env = SdkProvidedResourceDetector.detect(Duration::from_secs(1));
         assert_eq!(
             no_env.get(Key::from_static_str(SERVICE_NAME)),