Ureq (#121)

LukeMathWalker · web-flow · commit 5211835b88a2 · 2020-05-31T20:05:32.000-07:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -43,7 +43,8 @@ serialize = ["serde", "bincode"]
 members = [
     "opentelemetry-jaeger",
     "opentelemetry-zipkin",
-    "examples/actix",
+    "examples/actix-udp",
+    "examples/actix-http",
     "examples/async",
     "examples/basic",
     "examples/grpc",
diff --git a/examples/actix-http/Cargo.toml b/examples/actix-http/Cargo.toml
@@ -0,0 +1,15 @@
+[package]
+name = "actix-http-example"
+version = "0.1.0"
+edition = "2018"
+
+[dependencies]
+opentelemetry = { path = "../../" }
+opentelemetry-jaeger = { path = "../../opentelemetry-jaeger", features = ["collector_client"] }
+thrift = "0.13.0"
+futures = "0.3"
+actix-web = "2"
+actix-service = "1"
+actix-rt = "1"
+env_logger = "0.7.1"
+tokio = { version = "0.2.21", features = ["full"] }
diff --git a/examples/actix-http/README.md b/examples/actix-http/README.md
@@ -0,0 +1,24 @@
+# Actix-web - Jaeger example with HTTP collector and batch exporter 
+
+This example shows how to export spans from an actix-web application and ship them
+ to the collector directly via HTTP.  
+ It uses the batch exporter to avoid excessive network roundtrips to Jaeger.
+
+## Usage
+
+Launch the application:
+```shell
+# Run jaeger in background
+$ docker run -d -p6831:6831/udp -p6832:6832/udp -p16686:16686 -p14268:14268 jaegertracing/all-in-one:latest
+
+# Start the actix-web server 
+$ cargo run
+
+# View spans
+$ firefox http://localhost:16686/
+```
+
+Fire a request:
+```bash
+curl http://localhost:8088
+```
diff --git a/examples/actix-http/src/main.rs b/examples/actix-http/src/main.rs
@@ -0,0 +1,66 @@
+use actix_service::Service;
+use actix_web::middleware::Logger;
+use actix_web::{web, App, HttpServer};
+use opentelemetry::api::{Key, TraceContextExt, Tracer};
+use opentelemetry::sdk::BatchSpanProcessor;
+use opentelemetry::{global, sdk};
+use opentelemetry::api::trace::futures::FutureExt;
+
+fn init_tracer() -> thrift::Result<()> {
+    let exporter = opentelemetry_jaeger::Exporter::builder()
+        .with_collector_endpoint("http://127.0.0.1:14268/api/traces")
+        .with_process(opentelemetry_jaeger::Process {
+            service_name: "trace-http-demo".to_string(),
+            tags: vec![
+                Key::new("exporter").string("jaeger"),
+                Key::new("float").f64(312.23),
+            ],
+        })
+        .init()?;
+
+    let batch_exporter = BatchSpanProcessor::builder(exporter, tokio::spawn, tokio::time::interval)
+        .build();
+
+    let provider = sdk::Provider::builder()
+        .with_batch_exporter(batch_exporter)
+        .with_config(sdk::Config {
+            default_sampler: Box::new(sdk::Sampler::Always),
+            ..Default::default()
+        })
+        .build();
+    global::set_provider(provider);
+
+    Ok(())
+}
+
+async fn index() -> &'static str {
+    let tracer = global::tracer("request");
+    tracer.in_span("index", |ctx| {
+        ctx.span().set_attribute(Key::new("parameter").i64(10));
+        "Index"
+    })
+}
+
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    std::env::set_var("RUST_LOG", "debug");
+    env_logger::init();
+    init_tracer().expect("Failed to initialise tracer.");
+
+    HttpServer::new(|| {
+        App::new()
+            .wrap(Logger::default())
+            .wrap_fn(|req, srv| {
+                let tracer = global::tracer("request");
+                tracer.in_span("middleware", move |cx| {
+                    cx.span().set_attribute(Key::new("path").string(req.path()));
+                    srv.call(req).with_context(cx)
+                })
+            })
+            .route("/", web::get().to(index))
+    })
+    .bind("127.0.0.1:8088")
+    .unwrap()
+    .run()
+    .await
+}
diff --git a/examples/actix-udp/Cargo.toml b/examples/actix-udp/Cargo.toml
@@ -1,12 +1,14 @@
 [package]
-name = "actix-example"
+name = "actix-udp-example"
 version = "0.1.0"
 edition = "2018"
 
 [dependencies]
 opentelemetry = { path = "../../" }
 opentelemetry-jaeger = { path = "../../opentelemetry-jaeger" }
 thrift = "0.13.0"
-futures = "0.1.25"
-actix-web = "1.0.9"
-actix-service = "0.4.1"
+futures = "0.3"
+actix-web = "2"
+actix-service = "1"
+actix-rt = "1"
+env_logger = "0.7.1"
diff --git a/examples/actix-udp/README.md b/examples/actix-udp/README.md
@@ -0,0 +1,23 @@
+# Actix-web - Jaeger example with UDP agent 
+
+This example shows how to export spans from an actix-web application and ship them
+ to the Jaeger agent over UDP.  
+
+## Usage
+
+Launch the application:
+```shell
+# Run jaeger in background
+$ docker run -d -p6831:6831/udp -p6832:6832/udp -p16686:16686 -p14268:14268 jaegertracing/all-in-one:latest
+
+# Start the actix-web server 
+$ cargo run
+
+# View spans
+$ firefox http://localhost:16686/
+```
+
+Fire a request:
+```bash
+curl http://localhost:8088
+```
diff --git a/examples/actix-udp/src/main.rs b/examples/actix-udp/src/main.rs
@@ -1,6 +1,7 @@
 use actix_service::Service;
+use actix_web::middleware::Logger;
 use actix_web::{web, App, HttpServer};
-use futures::future::Future;
+use opentelemetry::api::trace::futures::FutureExt;
 use opentelemetry::api::{Key, TraceContextExt, Tracer};
 use opentelemetry::{global, sdk};
 
@@ -27,31 +28,34 @@ fn init_tracer() -> thrift::Result<()> {
     Ok(())
 }
 
-fn index() -> &'static str {
+async fn index() -> &'static str {
     let tracer = global::tracer("request");
     tracer.in_span("index", |ctx| {
         ctx.span().set_attribute(Key::new("parameter").i64(10));
         "Index"
     })
 }
 
-fn main() -> thrift::Result<()> {
-    init_tracer()?;
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    std::env::set_var("RUST_LOG", "debug");
+    env_logger::init();
+    init_tracer().expect("Failed to initialise tracer.");
 
     HttpServer::new(|| {
         App::new()
+            .wrap(Logger::default())
             .wrap_fn(|req, srv| {
                 let tracer = global::tracer("request");
                 tracer.in_span("middleware", move |cx| {
                     cx.span().set_attribute(Key::new("path").string(req.path()));
-                    srv.call(req).map(|res| res)
+                    srv.call(req).with_context(cx)
                 })
             })
             .route("/", web::get().to(index))
     })
     .bind("127.0.0.1:8088")
     .unwrap()
-    .run()?;
-
-    Ok(())
+    .run()
+    .await
 }
diff --git a/examples/basic/README.md b/examples/basic/README.md
@@ -6,7 +6,7 @@ This example shows basic span and metric usage, and exports to Jaeger.
 
 ```shell
 # Run jaeger in background
-$ docker run -d -p6831:6831/udp -p6832:6832/udp -p16686:16686 jaegertracing/all-in-one:latest
+$ docker run -d -p6831:6831/udp -p6832:6832/udp -p16686:16686 -p14268:14268 jaegertracing/all-in-one:latest
 
 # Report spans
 $ cargo run
diff --git a/opentelemetry-jaeger/Cargo.toml b/opentelemetry-jaeger/Cargo.toml
@@ -17,9 +17,9 @@ edition = "2018"
 
 [dependencies]
 opentelemetry = { version = "0.5.0", default-features = false, features = ["trace"], path = ".." }
-reqwest = { version = "0.10.1", features = ["blocking"], optional = true }
+ureq = { version = "1.0.0", optional = true }
 thrift = "0.13.0"
 
 [features]
 default = []
-collector_client = ["reqwest"]
+collector_client = ["ureq"]
diff --git a/opentelemetry-jaeger/src/lib.rs b/opentelemetry-jaeger/src/lib.rs
@@ -277,7 +277,10 @@ impl<T: net::ToSocketAddrs> Builder<T> {
                 self.collector_username,
                 self.collector_password,
             )?;
-            Ok((self.process, uploader::BatchUploader::Collector(collector)))
+            Ok((
+                self.process,
+                uploader::BatchUploader::Collector(Box::new(collector)),
+            ))
         } else {
             Err(::thrift::Error::from(
                 "Collector endpoint or agent endpoint must be set",
diff --git a/opentelemetry-jaeger/src/transport/http.rs b/opentelemetry-jaeger/src/transport/http.rs
@@ -1,16 +1,14 @@
 //! Thrift HTTP transport
-use reqwest::header;
 use std::fmt;
 use std::sync::{Arc, Mutex};
-use thrift::{TransportError, TransportErrorKind};
 
 /// Thrift channel over HTTP
 #[derive(Clone)]
 pub(crate) struct THttpChannel {
-    client: reqwest::blocking::Client,
     endpoint: String,
     username: Option<String>,
     password: Option<String>,
+    request: ureq::Request,
     read_buffer: Arc<Mutex<Vec<u8>>>,
     write_buffer: Arc<Mutex<Vec<u8>>>,
 }
@@ -19,7 +17,6 @@ impl fmt::Debug for THttpChannel {
     /// Debug info
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         fmt.debug_struct("THttpChannel")
-            .field("client", &self.client)
             .field("endpoint", &self.endpoint)
             .field("username", &self.username)
             .field("password", &self.password.as_ref().map(|_| "************"))
@@ -36,24 +33,21 @@ impl THttpChannel {
         username: Option<String>,
         password: Option<String>,
     ) -> thrift::Result<Self> {
-        let mut headers = header::HeaderMap::new();
-        headers.insert(
-            header::CONTENT_TYPE,
-            header::HeaderValue::from_static("application/vnd.apache.thrift.binary"),
-        );
-        let client = reqwest::blocking::Client::builder()
-            .default_headers(headers)
-            .build()
-            .map_err(|err| {
-                thrift::Error::Transport(TransportError::new(
-                    TransportErrorKind::Unknown,
-                    err.to_string(),
-                ))
-            })?;
+        let endpoint: String = endpoint.into();
+        let mut request = ureq::post(&endpoint);
+        request.set("Content-Type", "application/vnd.apache.thrift.binary");
+        // Some arbitrary default timeouts to avoid hanging forever
+        request.timeout_connect(10000);
+        request.timeout_read(10000);
+        request.timeout_write(10000);
+
+        if let (Some(username), Some(password)) = (username.as_ref(), password.as_ref()) {
+            request.auth(&username, &password);
+        }
 
         Ok(THttpChannel {
-            client,
-            endpoint: endpoint.into(),
+            request,
+            endpoint,
             username,
             password,
             read_buffer: Default::default(),
@@ -120,30 +114,30 @@ impl std::io::Write for THttpChannel {
             .write_buffer
             .lock()
             .map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err.to_string()))?;
-        let mut req = self.client.post(&self.endpoint).body(write_buffer.clone());
-        if let (Some(username), Some(password)) = (self.username.as_ref(), self.password.as_ref()) {
-            req = req.basic_auth(username, Some(password));
-        }
-        let mut resp = req
-            .send()
-            .map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err))?;
 
-        if !resp.status().is_success() {
+        let mut req = self.request.build();
+        let resp = req.send_bytes(&write_buffer.clone());
+        if !resp.ok() {
             return Err(std::io::Error::new(
                 std::io::ErrorKind::Other,
-                format!("Expected success response, got {:?}", 1),
+                format!("Expected success response, got {:?}", resp.status()),
             ));
         }
 
         write_buffer.clear();
 
-        if resp.content_length().filter(|len| *len > 0).is_some() {
+        let has_content = resp
+            .header("Content-Length")
+            .and_then(|s| s.parse::<usize>().ok())
+            .map(|len| len > 0)
+            .unwrap_or(false);
+        if has_content {
+            let mut response_reader = resp.into_reader();
             let mut read_buffer = self
                 .read_buffer
                 .lock()
                 .map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err.to_string()))?;
-
-            resp.copy_to(&mut *read_buffer)
+            std::io::copy(&mut response_reader, &mut *read_buffer)
                 .map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err))?;
         }
 
diff --git a/opentelemetry-jaeger/src/uploader.rs b/opentelemetry-jaeger/src/uploader.rs
@@ -11,7 +11,7 @@ pub(crate) enum BatchUploader {
     Agent(agent::AgentSyncClientUDP),
     /// Collector sync client
     #[cfg(feature = "collector_client")]
-    Collector(collector::CollectorSyncClientHttp),
+    Collector(Box<collector::CollectorSyncClientHttp>),
 }
 
 impl BatchUploader {
