Upgrade Tokio to v1 (#421)

mraerino · djc · web-flow · commit 672879997ac9 · 2021-01-18T09:20:50.000-08:00
Co-authored-by: Dirkjan Ochtman &lt;dirkjan@ochtman.nl&gt;
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -48,11 +48,11 @@ jobs:
     - uses: actions-rs/toolchain@v1
       with:
         profile: minimal
-        toolchain: 1.42.0
+        toolchain: 1.46.0
         override: true
     - name: Run tests
       run: cargo --version &&
-        cargo test --verbose --manifest-path=opentelemetry/Cargo.toml --features trace,metrics,serialize,tokio,serde,testing &&
+        cargo test --verbose --manifest-path=opentelemetry/Cargo.toml --features trace,metrics,serialize,tokio-support,serde,testing &&
         cargo test --manifest-path=opentelemetry-jaeger/Cargo.toml &&
         cargo test --manifest-path=opentelemetry-zipkin/Cargo.toml
   meta:
diff --git a/README.md b/README.md
@@ -24,7 +24,7 @@ analysis in order to understand your software's performance and behavior. You
 can export and analyze them using [Prometheus], [Jaeger], and other
 observability tools.
 
-*Compiler support: [requires `rustc` 1.42+][msrv]*
+*Compiler support: [requires `rustc` 1.46+][msrv]*
 
 [Prometheus]: https://prometheus.io
 [Jaeger]: https://www.jaegertracing.io
@@ -120,12 +120,12 @@ above, please let us know! We'd love to add your project to the list!
 ## Supported Rust Versions
 
 OpenTelemetry is built against the latest stable release. The minimum supported
-version is 1.42. The current OpenTelemetry version is not guaranteed to build
+version is 1.46. The current OpenTelemetry version is not guaranteed to build
 on Rust versions earlier than the minimum supported version.
 
 The current stable Rust compiler and the three most recent minor versions
 before it will always be supported. For example, if the current stable compiler
-version is 1.45, the minimum supported version will not be increased past 1.42,
+version is 1.49, the minimum supported version will not be increased past 1.46,
 three minor versions prior. Increasing the minimum supported compiler version
 is not considered a semver breaking change as long as doing so complies with
 this policy.
diff --git a/examples/actix-http/Cargo.toml b/examples/actix-http/Cargo.toml
@@ -4,11 +4,11 @@ version = "0.1.0"
 edition = "2018"
 
 [dependencies]
-opentelemetry = { path = "../../opentelemetry", features = ["tokio"] }
+opentelemetry = { path = "../../opentelemetry", features = ["tokio-support"] }
 opentelemetry-jaeger = { path = "../../opentelemetry-jaeger", features = ["reqwest_collector_client", "tokio"] }
 thrift = "0.13"
 futures = "0.3"
 actix-web = "3"
 actix-service = "1"
 env_logger = "0.8.2"
-tokio = { version = "0.2.21", features = ["full"] }
+tokio = { version = "1.0", features = ["full"] }
diff --git a/examples/async/Cargo.toml b/examples/async/Cargo.toml
@@ -6,6 +6,6 @@ edition = "2018"
 [dependencies]
 futures = "0.3"
 thrift = "0.13"
-tokio = { version = "0.2", features = ["full"] }
-opentelemetry = { path = "../../opentelemetry", features = ["tokio"] }
+tokio = { version = "1.0", features = ["full"] }
+opentelemetry = { path = "../../opentelemetry", features = ["tokio-support"] }
 opentelemetry-jaeger = { path = "../../opentelemetry-jaeger", features = ["tokio"] }
diff --git a/examples/basic-otlp/Cargo.toml b/examples/basic-otlp/Cargo.toml
@@ -6,7 +6,7 @@ edition = "2018"
 [dependencies]
 futures = "0.3"
 lazy_static = "1.4"
-opentelemetry = { path = "../../opentelemetry", features = ["tokio", "metrics", "serialize"] }
+opentelemetry = { path = "../../opentelemetry", features = ["tokio-support", "metrics", "serialize"] }
 opentelemetry-otlp = { path = "../../opentelemetry-otlp", features = ["tonic", "metrics"] }
 serde_json = "1.0"
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "1.0", features = ["full"] }
diff --git a/examples/basic-otlp/src/main.rs b/examples/basic-otlp/src/main.rs
@@ -19,7 +19,7 @@ fn init_tracer() -> Result<(sdktrace::Tracer, opentelemetry_otlp::Uninstall), Tr
 
 // Skip first immediate tick from tokio, not needed for async_std.
 fn delayed_interval(duration: Duration) -> impl Stream<Item = tokio::time::Instant> {
-    tokio::time::interval(duration).skip(1)
+    opentelemetry::util::tokio_interval_stream(duration).skip(1)
 }
 
 fn init_meter() -> metrics::Result<PushController> {
@@ -98,7 +98,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     });
 
     // wait for 1 seconds so that we could see metrics being pushed via OTLP every 10 seconds.
-    tokio::time::delay_for(Duration::from_secs(60)).await;
+    tokio::time::sleep(Duration::from_secs(60)).await;
 
     Ok(())
 }
diff --git a/examples/basic/Cargo.toml b/examples/basic/Cargo.toml
@@ -6,8 +6,8 @@ edition = "2018"
 [dependencies]
 futures = "0.3"
 lazy_static = "1.4"
-opentelemetry = { path = "../../opentelemetry", features = ["serialize", "tokio", "metrics"] }
+opentelemetry = { path = "../../opentelemetry", features = ["serialize", "tokio-support", "metrics"] }
 opentelemetry-jaeger = { path = "../../opentelemetry-jaeger", features = ["tokio"] }
 serde_json = "1.0"
 thrift = "0.13"
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "1.0", features = ["full"] }
diff --git a/examples/basic/src/main.rs b/examples/basic/src/main.rs
@@ -23,7 +23,7 @@ fn init_tracer() -> Result<(sdktrace::Tracer, opentelemetry_jaeger::Uninstall),
 
 // Skip first immediate tick from tokio, not needed for async_std.
 fn delayed_interval(duration: Duration) -> impl Stream<Item = tokio::time::Instant> {
-    tokio::time::interval(duration).skip(1)
+    opentelemetry::util::tokio_interval_stream(duration).skip(1)
 }
 
 fn init_meter() -> metrics::Result<PushController> {
diff --git a/examples/grpc/Cargo.toml b/examples/grpc/Cargo.toml
@@ -15,7 +15,7 @@ path = "src/client.rs"
 http = "0.2"
 tonic = "0.3"
 prost = "0.6"
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "1.0", features = ["full"] }
 opentelemetry = { path = "../../opentelemetry" }
 opentelemetry-jaeger = { path = "../../opentelemetry-jaeger" }
 
diff --git a/examples/tracing-grpc/Cargo.toml b/examples/tracing-grpc/Cargo.toml
@@ -15,9 +15,9 @@ path = "src/client.rs"
 http = "0.2"
 tonic = "0.3"
 prost = "0.6"
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "1.0", features = ["full"] }
 opentelemetry = { version = "0.11", features = ["tonic"] }
-opentelemetry-jaeger ="0.10" 
+opentelemetry-jaeger ="0.10"
 tracing = "0.1"
 tracing-subscriber = "0.2"
 tracing-opentelemetry = "0.10"
diff --git a/opentelemetry-contrib/Cargo.toml b/opentelemetry-contrib/Cargo.toml
@@ -35,7 +35,7 @@ opentelemetry = { version = "0.11", path = "../opentelemetry", features = ["trac
 opentelemetry-http = { version = "0.1", path = "../opentelemetry-http", optional = true }
 rmp = { version = "0.8", optional = true }
 lazy_static = "1.4"
-reqwest = { version = "0.10", optional = true }
+reqwest = { version = "0.11", optional = true }
 surf = { version = "2.0", optional = true }
 http = "0.2"
 base64 = { version = "0.13", optional = true }
diff --git a/opentelemetry-contrib/src/trace/exporter/datadog/mod.rs b/opentelemetry-contrib/src/trace/exporter/datadog/mod.rs
@@ -35,16 +35,16 @@
 //!
 //! For optimal performance, a batch exporter is recommended as the simple
 //! exporter will export each span synchronously on drop. You can enable the
-//! [`tokio`] or [`async-std`] features to have a batch exporter configured for
+//! [`tokio-support`] or [`async-std`] features to have a batch exporter configured for
 //! you automatically for either executor when you install the pipeline.
 //!
 //! ```toml
 //! [dependencies]
-//! opentelemetry = { version = "*", features = ["tokio"] }
+//! opentelemetry = { version = "*", features = ["tokio-support"] }
 //! opentelemetry-datadog = "*"
 //! ```
 //!
-//! [`tokio`]: https://tokio.rs
+//! [`tokio-support`]: https://tokio.rs
 //! [`async-std`]: https://async.rs
 //!
 
diff --git a/opentelemetry-http/Cargo.toml b/opentelemetry-http/Cargo.toml
@@ -14,6 +14,6 @@ async-trait = "0.1.42"
 http = "0.2.2"
 isahc = { version = "0.9", default-features = false, optional = true }
 opentelemetry = { version = "0.11.2", path = "../opentelemetry", features = ["trace"] }
-reqwest = { version = "0.10", default-features = false, features = ["blocking"], optional = true }
+reqwest = { version = "0.11", default-features = false, features = ["blocking"], optional = true }
 surf = { version = "2.0", default-features = false, optional = true }
 thiserror = "1"
diff --git a/opentelemetry-http/README.md b/opentelemetry-http/README.md
@@ -21,7 +21,7 @@ analysis in order to understand your software's performance and behavior. This
 crate provides a HTTP client interface for use by trace exporters, as well as
 helper types to inject and extract key value pairs into/from HTTP headers.
 
-*Compiler support: [requires `rustc` 1.42+][msrv]*
+*Compiler support: [requires `rustc` 1.46+][msrv]*
 
 [`Jaeger`]: https://www.jaegertracing.io/
 [`OpenTelemetry`]: https://crates.io/crates/opentelemetry
@@ -30,12 +30,12 @@ helper types to inject and extract key value pairs into/from HTTP headers.
 ## Supported Rust Versions
 
 OpenTelemetry is built against the latest stable release. The minimum supported
-version is 1.42. The current OpenTelemetry version is not guaranteed to build
+version is 1.46. The current OpenTelemetry version is not guaranteed to build
 on Rust versions earlier than the minimum supported version.
 
 The current stable Rust compiler and the three most recent minor versions
 before it will always be supported. For example, if the current stable compiler
-version is 1.45, the minimum supported version will not be increased past 1.42,
+version is 1.49, the minimum supported version will not be increased past 1.46,
 three minor versions prior. Increasing the minimum supported compiler version
 is not considered a semver breaking change as long as doing so complies with
 this policy.
diff --git a/opentelemetry-jaeger/Cargo.toml b/opentelemetry-jaeger/Cargo.toml
@@ -31,12 +31,12 @@ opentelemetry = { version = "0.11", default-features = false, features = ["trace
 opentelemetry-http = { version = "0.1", path = "../opentelemetry-http", optional = true }
 pin-project = { version = "1.0", optional = true }
 thrift = "0.13"
-tokio = { version = "0.2", features = ["udp", "sync"], optional = true }
+tokio = { version = "1.0", features = ["net", "sync"], optional = true }
 wasm-bindgen = { version = "0.2", optional = true }
 wasm-bindgen-futures = { version = "0.4.18", optional = true }
 thiserror = "1.0"
 lazy_static = "1.4"
-reqwest = { version = "0.10", default-features = false, optional = true }
+reqwest = { version = "0.11", default-features = false, optional = true }
 headers = { version = "0.3.2", optional = true }
 surf = { version = "2.0", optional = true }
 
diff --git a/opentelemetry-jaeger/README.md b/opentelemetry-jaeger/README.md
@@ -23,7 +23,7 @@ analysis in order to understand your software's performance and behavior. This
 crate provides a trace pipeline and exporter for sending span information to a
 Jaeger `agent` or `collector` endpoint for processing and visualization.
 
-*Compiler support: [requires `rustc` 1.42+][msrv]*
+*Compiler support: [requires `rustc` 1.46+][msrv]*
 
 [`Jaeger`]: https://www.jaegertracing.io/
 [`OpenTelemetry`]: https://crates.io/crates/opentelemetry
@@ -62,17 +62,17 @@ fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
 ## Performance
 
 For optimal performance, a batch exporter is recommended as the simple exporter
-will export each span synchronously on drop. You can enable the [`tokio`] or
+will export each span synchronously on drop. You can enable the [`tokio-support`] or
 [`async-std`] features to have a batch exporter configured for you automatically
 for either executor when you install the pipeline.
 
 ```toml
 [dependencies]
-opentelemetry = { version = "*", features = ["tokio"] }
+opentelemetry = { version = "*", features = ["tokio-support"] }
 opentelemetry-jaeger = { version = "*", features = ["tokio"] }
 ```
 
-[`tokio`]: https://tokio.rs
+[`tokio-support`]: https://tokio.rs
 [`async-std`]: https://async.rs
 
 ### Jaeger Exporter From Environment Variables
@@ -123,8 +123,8 @@ Then you can use the [`with_collector_endpoint`] method to specify the endpoint:
 // * reqwest_blocking_collector_client
 // * isahc_collector_client
 
-// You can also provide your own implementation by enable 
-// `collecor_client` and set it with 
+// You can also provide your own implementation by enable
+// `collecor_client` and set it with
 // new_pipeline().with_http_client() method.
 use opentelemetry::tracer;
 
@@ -186,12 +186,12 @@ fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
 ## Supported Rust Versions
 
 OpenTelemetry is built against the latest stable release. The minimum supported
-version is 1.42. The current OpenTelemetry version is not guaranteed to build
+version is 1.46. The current OpenTelemetry version is not guaranteed to build
 on Rust versions earlier than the minimum supported version.
 
 The current stable Rust compiler and the three most recent minor versions
 before it will always be supported. For example, if the current stable compiler
-version is 1.45, the minimum supported version will not be increased past 1.42,
+version is 1.49, the minimum supported version will not be increased past 1.46,
 three minor versions prior. Increasing the minimum supported compiler version
 is not considered a semver breaking change as long as doing so complies with
 this policy.
diff --git a/opentelemetry-jaeger/src/lib.rs b/opentelemetry-jaeger/src/lib.rs
@@ -2,7 +2,7 @@
 //! `agent` or `collector` endpoint. See the [Jaeger Docs] for details
 //! about Jaeger and deployment information.
 //!
-//! *Compiler support: [requires `rustc` 1.42+][msrv]*
+//! *Compiler support: [requires `rustc` 1.46+][msrv]*
 //!
 //! [Jaeger Docs]: https://www.jaegertracing.io/docs/
 //! [msrv]: #supported-rust-versions
@@ -44,7 +44,7 @@
 //!
 //! ```toml
 //! [dependencies]
-//! opentelemetry = { version = "*", features = ["tokio"] }
+//! opentelemetry = { version = "*", features = ["tokio-support"] }
 //! opentelemetry-jaeger = { version = "*", features = ["tokio"] }
 //! ```
 //!
@@ -171,14 +171,14 @@
 //! ## Supported Rust Versions
 //!
 //! OpenTelemetry is built against the latest stable release. The minimum
-//! supported version is 1.42. The current OpenTelemetry version is not
+//! supported version is 1.46. The current OpenTelemetry version is not
 //! guaranteed to build on Rust versions earlier than the minimum supported
 //! version.
 //!
 //! The current stable Rust compiler and the three most recent minor versions
 //! before it will always be supported. For example, if the current stable
-//! compiler version is 1.45, the minimum supported version will not be
-//! increased past 1.42, three minor versions prior. Increasing the minimum
+//! compiler version is 1.49, the minimum supported version will not be
+//! increased past 1.46, three minor versions prior. Increasing the minimum
 //! supported compiler version is not considered a semver breaking change as
 //! long as doing so complies with this policy.
 #![warn(
diff --git a/opentelemetry-otlp/Cargo.toml b/opentelemetry-otlp/Cargo.toml
@@ -34,14 +34,15 @@ async-trait = "0.1"
 futures = "0.3"
 grpcio = { version = "0.7", optional = true }
 opentelemetry = { version = "0.11", default-features = false, features = ["trace"], path = "../opentelemetry" }
-prost = { version = "0.6", optional = true }
+prost = { version = "0.7", optional = true }
 protobuf = { version = "2.18", optional = true }
 thiserror = "1.0"
-tonic = { package = "tonic", version = "0.3", optional = true }
-tokio = { version = "0.2", features = ["full"], optional = true }
+tonic = { version = "0.4", optional = true }
+tokio = { version = "1.0", features = ["full"], optional = true }
 
 [dev-dependencies]
 chrono = "0.4"
+tokio-stream = { version = "0.1", features = ["net"] }
 
 [features]
 async = ["default"]
diff --git a/opentelemetry-otlp/src/span.rs b/opentelemetry-otlp/src/span.rs
@@ -181,8 +181,7 @@ impl Default for TraceExporter {
             metadata: config.metadata,
             #[cfg(not(feature = "async"))]
             runtime: config.runtime.unwrap_or_else(|| {
-                tokio::runtime::Builder::new()
-                    .basic_scheduler()
+                tokio::runtime::Builder::new_current_thread()
                     .enable_all()
                     .build()
                     .unwrap()
@@ -270,8 +269,7 @@ impl TraceExporter {
             metadata: config.metadata,
             #[cfg(not(feature = "async"))]
             runtime: config.runtime.unwrap_or_else(|| {
-                tokio::runtime::Builder::new()
-                    .basic_scheduler()
+                tokio::runtime::Builder::new_current_thread()
                     .enable_all()
                     .build()
                     .unwrap()
diff --git a/opentelemetry-otlp/tests/smoke.rs b/opentelemetry-otlp/tests/smoke.rs
@@ -6,6 +6,7 @@ use opentelemetry_otlp::proto::collector::trace::v1::{
 };
 use std::{net::SocketAddr, sync::Mutex};
 use tokio::sync::mpsc;
+use tokio_stream::wrappers::TcpListenerStream;
 
 struct MockServer {
     tx: Mutex<mpsc::Sender<ExportTraceServiceRequest>>,
@@ -39,7 +40,7 @@ async fn setup() -> (SocketAddr, mpsc::Receiver<ExportTraceServiceRequest>) {
         .await
         .expect("failed to bind");
     let addr = listener.local_addr().unwrap();
-    let stream = listener.map(|s| {
+    let stream = TcpListenerStream::new(listener).map(|s| {
         if let Ok(ref s) = s {
             println!("Got new conn at {}", s.peer_addr().unwrap());
         }
@@ -58,7 +59,7 @@ async fn setup() -> (SocketAddr, mpsc::Receiver<ExportTraceServiceRequest>) {
     (addr, req_rx)
 }
 
-#[tokio::test(threaded_scheduler)]
+#[tokio::test(flavor = "multi_thread")]
 async fn smoke_tracer() {
     println!("Starting server setup...");
     let (addr, mut req_rx) = setup().await;
diff --git a/opentelemetry-prometheus/Cargo.toml b/opentelemetry-prometheus/Cargo.toml
@@ -25,6 +25,6 @@ prometheus = "0.10"
 protobuf = "2.14"
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["full"] }
-hyper = "0.13"
+tokio = { version = "1.0", features = ["full"] }
+hyper = "0.14"
 lazy_static = "1.4"
diff --git a/opentelemetry-zipkin/Cargo.toml b/opentelemetry-zipkin/Cargo.toml
@@ -34,7 +34,7 @@ serde = { version = "1.0", features = ["derive"] }
 typed-builder = "0.7"
 lazy_static = "1.4"
 http = "0.2"
-reqwest = { version = "0.10", optional = true }
+reqwest = { version = "0.11", optional = true }
 surf = { version = "2.0", optional = true }
 thiserror = { version = "1.0"}
 
diff --git a/opentelemetry-zipkin/README.md b/opentelemetry-zipkin/README.md
diff --git a/opentelemetry-zipkin/src/lib.rs b/opentelemetry-zipkin/src/lib.rs
diff --git a/opentelemetry/Cargo.toml b/opentelemetry/Cargo.toml
diff --git a/opentelemetry/README.md b/opentelemetry/README.md
diff --git a/opentelemetry/src/lib.rs b/opentelemetry/src/lib.rs
diff --git a/opentelemetry/src/sdk/trace/provider.rs b/opentelemetry/src/sdk/trace/provider.rs
diff --git a/opentelemetry/src/sdk/trace/span_processor.rs b/opentelemetry/src/sdk/trace/span_processor.rs
diff --git a/opentelemetry/src/util.rs b/opentelemetry/src/util.rs

Original file line number	Diff line number	Diff line change
`@@ -19,7 +19,7 @@ fn init_tracer() -> Result<(sdktrace::Tracer, opentelemetry_otlp::Uninstall), Tr`
`19`	`19`
`20`	`20`	`// Skip first immediate tick from tokio, not needed for async_std.`
`21`	`21`	`fn delayed_interval(duration: Duration) -> impl Stream<Item = tokio::time::Instant> {`
`22`		`- tokio::time::interval(duration).skip(1)`
	`22`	`+ opentelemetry::util::tokio_interval_stream(duration).skip(1)`
`23`	`23`	`}`
`24`	`24`
`25`	`25`	`fn init_meter() -> metrics::Result<PushController> {`
`@@ -98,7 +98,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {`
`98`	`98`	`});`
`99`	`99`
`100`	`100`	`// wait for 1 seconds so that we could see metrics being pushed via OTLP every 10 seconds.`
`101`		`- tokio::time::delay_for(Duration::from_secs(60)).await;`
	`101`	`+ tokio::time::sleep(Duration::from_secs(60)).await;`
`102`	`102`
`103`	`103`	`Ok(())`
`104`	`104`	`}`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ fn init_tracer() -> Result<(sdktrace::Tracer, opentelemetry_jaeger::Uninstall),`
`23`	`23`
`24`	`24`	`// Skip first immediate tick from tokio, not needed for async_std.`
`25`	`25`	`fn delayed_interval(duration: Duration) -> impl Stream<Item = tokio::time::Instant> {`
`26`		`- tokio::time::interval(duration).skip(1)`
	`26`	`+ opentelemetry::util::tokio_interval_stream(duration).skip(1)`
`27`	`27`	`}`
`28`	`28`
`29`	`29`	`fn init_meter() -> metrics::Result<PushController> {`