Update Otel/Tonic (#698)

And other deps

Author: Sushisource

Cargo.toml

@@ -10,6 +10,8 @@ license-file = "LICENSE.txt"
 derive_builder = "0.20"
 derive_more = { version = "0.99", default-features = false, features = ["constructor", "display", "from", "into"] }
 once_cell = "1.16"
-tonic = "0.9"
-tonic-build = "0.9"
-opentelemetry = "0.21"
+tonic = "0.11"
+tonic-build = "0.11"
+opentelemetry = "0.22"
+prost = "0.12"
+prost-types = "0.12"

client/Cargo.toml

@@ -25,7 +25,7 @@ http = "0.2"
 once_cell = { workspace = true }
 opentelemetry = { workspace = true, features = ["metrics"], optional = true  }
 parking_lot = "0.12"
-prost-types = "0.11"
+prost-types = { workspace = true }
 slotmap = "1.0"
 thiserror = "1.0"
 tokio = "1.1"
@@ -43,4 +43,4 @@ path = "../core-api"
 
 [dev-dependencies]
 assert_matches = "1"
-mockall = "0.11"
+mockall = "0.12"

core-api/Cargo.toml

@@ -20,7 +20,7 @@ async-trait = "0.1"
 derive_builder = { workspace = true }
 derive_more = { workspace = true }
 opentelemetry = { workspace = true, optional = true }
-prost-types = "0.11"
+prost-types = { workspace = true }
 serde = { version = "1.0", default_features = false, features = ["derive"] }
 serde_json = "1.0"
 thiserror = "1.0"

core/Cargo.toml

@@ -18,16 +18,16 @@ default = ["otel"]
 # record WF input data, we can build them a custom SDK or they can build - it adds significant extra
 # code size in the form of [de]serializers.
 save_wf_inputs = ["rmp-serde", "temporal-sdk-core-protos/serde_serialize"]
-otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-otlp", "dep:opentelemetry-prometheus"]
+otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-otlp",
+    "dep:opentelemetry-prometheus", "dep:hyper", "dep:hyper-util", "dep:http-body-util"]
 tokio-console = ["console-subscriber"]
 ephemeral-server = ["dep:flate2", "dep:nix", "dep:reqwest", "dep:tar", "dep:zip"]
 
 [dependencies]
 anyhow = "1.0"
-arc-swap = "1.3"
 async-trait = "0.1"
 base64 = "0.21"
-console-subscriber = { version = "0.1", optional = true }
+console-subscriber = { version = "0.2", optional = true }
 crossbeam-channel = "0.5"
 crossbeam-queue = "0.3"
 dashmap = "5.5"
@@ -39,22 +39,23 @@ flate2 = { version = "1.0", optional = true }
 futures = "0.3"
 futures-util = "0.3"
 governor = "0.6"
-http = "0.2"
-hyper = "0.14"
+http-body-util = { version = "0.1", optional = true }
+hyper = { version = "1.2", optional = true }
+hyper-util = { version = "0.1", features = ["server", "http1", "http2", "tokio"], optional = true }
 itertools = "0.12"
-lru = "0.11"
-mockall = "0.11"
+lru = "0.12"
+mockall = "0.12"
 nix = { version = "0.28", optional = true, features = ["process", "signal"] }
 once_cell = { workspace = true }
 opentelemetry = { workspace = true, features = ["metrics"], optional = true }
-opentelemetry_sdk = { version = "0.21", features = ["rt-tokio", "metrics"], optional = true }
-opentelemetry-otlp = { version = "0.14", features = ["tokio", "metrics"], optional = true }
-opentelemetry-prometheus = { version = "0.14", optional = true }
+opentelemetry_sdk = { version = "0.22", features = ["rt-tokio", "metrics"], optional = true }
+opentelemetry-otlp = { version = "0.15", features = ["tokio", "metrics"], optional = true }
+opentelemetry-prometheus = { version = "0.15", optional = true }
 parking_lot = { version = "0.12", features = ["send_guard"] }
 pin-project = "1.0"
 prometheus = "0.13"
-prost = "0.11"
-prost-types = { version = "0.4", package = "prost-wkt-types" }
+prost = { workspace = true }
+prost-types = { version = "0.5", package = "prost-wkt-types" }
 rand = "0.8.3"
 reqwest = { version = "0.11", features = ["json", "stream", "rustls-tls", "tokio-rustls"], default-features = false, optional = true }
 ringbuf = "0.3"
@@ -70,7 +71,6 @@ tokio-util = { version = "0.7", features = ["io", "io-util"] }
 tokio-stream = "0.1"
 tonic = { workspace = true, features = ["tls", "tls-roots"] }
 tracing = "0.1"
-tracing-futures = "0.2"
 tracing-subscriber = { version = "0.3", features = ["parking_lot", "env-filter", "registry"] }
 url = "2.2"
 uuid = { version = "1.1", features = ["v4"] }

core/src/protosext/mod.rs

@@ -406,7 +406,7 @@ impl ValidScheduleLA {
             .clone()
             .try_into_or_none()
             .unwrap_or_else(|| Duration::from_secs(60));
-        let cancellation_type = ActivityCancellationType::from_i32(v.cancellation_type)
+        let cancellation_type = ActivityCancellationType::try_from(v.cancellation_type)
             .unwrap_or(ActivityCancellationType::WaitCancellationCompleted);
         Ok(ValidScheduleLA {
             seq: v.seq,

core/src/telemetry/mod.rs

@@ -10,7 +10,6 @@ mod prometheus_server;
 
 #[cfg(feature = "otel")]
 pub use metrics::{default_buckets_for, MetricsCallBuffer};
-
 #[cfg(feature = "otel")]
 pub use otel::{build_otlp_metric_exporter, start_prometheus_metric_exporter};
 

core/src/telemetry/otel.rs

@@ -21,8 +21,7 @@ use opentelemetry_sdk::{
         data::Temporality,
         new_view,
         reader::{AggregationSelector, DefaultAggregationSelector, TemporalitySelector},
-        Aggregation, Instrument, InstrumentKind, MeterProvider, MeterProviderBuilder,
-        PeriodicReader, View,
+        Aggregation, Instrument, InstrumentKind, MeterProviderBuilder, PeriodicReader, View,
     },
     runtime, AttributeSet, Resource,
 };
@@ -152,7 +151,7 @@ pub fn build_otlp_metric_exporter(
         .with_interval(opts.metric_periodicity)
         .build();
     let mp = augment_meter_provider_with_defaults(
-        MeterProvider::builder().with_reader(reader),
+        MeterProviderBuilder::default().with_reader(reader),
         &opts.global_tags,
     )?
     .build();
@@ -167,16 +166,18 @@ pub struct StartedPromServer {
 
 /// Builds and runs a prometheus endpoint which can be scraped by prom instances for metrics export.
 /// Returns the meter that can be used as a [CoreMeter].
+///
+/// Requires a Tokio runtime to exist, and will block briefly while binding the server endpoint.
 pub fn start_prometheus_metric_exporter(
     opts: PrometheusExporterOptions,
 ) -> Result<StartedPromServer, anyhow::Error> {
     let (srv, exporter) = PromServer::new(&opts, SDKAggSelector::default())?;
     let meter_provider = augment_meter_provider_with_defaults(
-        MeterProvider::builder().with_reader(exporter),
+        MeterProviderBuilder::default().with_reader(exporter),
         &opts.global_tags,
     )?
     .build();
-    let bound_addr = srv.bound_addr();
+    let bound_addr = srv.bound_addr()?;
     let handle = tokio::spawn(async move { srv.run().await });
     Ok(StartedPromServer {
         meter: Arc::new(CoreOtelMeter(meter_provider.meter(TELEM_SERVICE_NAME))),

core/src/telemetry/prometheus_server.rs

@@ -1,18 +1,19 @@
-use hyper::{
-    header::CONTENT_TYPE,
-    server::conn::AddrIncoming,
-    service::{make_service_fn, service_fn},
-    Body, Method, Request, Response, Server,
+use http_body_util::Full;
+use hyper::{body::Bytes, header::CONTENT_TYPE, service::service_fn, Method, Request, Response};
+use hyper_util::{
+    rt::{TokioExecutor, TokioIo},
+    server::conn::auto,
 };
 use opentelemetry_prometheus::PrometheusExporter;
 use opentelemetry_sdk::metrics::reader::AggregationSelector;
 use prometheus::{Encoder, Registry, TextEncoder};
-use std::{convert::Infallible, net::SocketAddr};
+use std::net::{SocketAddr, TcpListener};
 use temporal_sdk_core_api::telemetry::PrometheusExporterOptions;
+use tokio::io;
 
 /// Exposes prometheus metrics for scraping
 pub(super) struct PromServer {
-    bound_addr: AddrIncoming,
+    listener: TcpListener,
     registry: Registry,
 }
 
@@ -36,37 +37,49 @@ impl PromServer {
         } else {
             exporter
         };
-        let bound_addr = AddrIncoming::bind(&opts.socket_addr)?;
         Ok((
             Self {
-                bound_addr,
+                listener: TcpListener::bind(opts.socket_addr)?,
                 registry,
             },
             exporter.build()?,
         ))
     }
 
-    pub async fn run(self) -> hyper::Result<()> {
+    pub async fn run(self) -> Result<(), anyhow::Error> {
         // Spin up hyper server to serve metrics for scraping. We use hyper since we already depend
         // on it via Tonic.
-        let svc = make_service_fn(move |_conn| {
+        self.listener.set_nonblocking(true)?;
+        let listener = tokio::net::TcpListener::from_std(self.listener)?;
+        loop {
+            let (stream, _) = listener.accept().await?;
+            let io = TokioIo::new(stream);
             let regclone = self.registry.clone();
-            async move { Ok::<_, Infallible>(service_fn(move |req| metrics_req(req, regclone.clone()))) }
-        });
-        let server = Server::builder(self.bound_addr).serve(svc);
-        server.await
+            tokio::task::spawn(async move {
+                let server = auto::Builder::new(TokioExecutor::new());
+                if let Err(e) = server
+                    .serve_connection(
+                        io,
+                        service_fn(move |req| metrics_req(req, regclone.clone())),
+                    )
+                    .await
+                {
+                    warn!("Error serving metrics connection: {:?}", e);
+                }
+            });
+        }
     }
 
-    pub fn bound_addr(&self) -> SocketAddr {
-        self.bound_addr.local_addr()
+    pub fn bound_addr(&self) -> io::Result<SocketAddr> {
+        self.listener.local_addr()
     }
 }
 
 /// Serves prometheus metrics in the expected format for scraping
 async fn metrics_req(
-    req: Request<Body>,
+    req: Request<hyper::body::Incoming>,
     registry: Registry,
-) -> Result<Response<Body>, hyper::Error> {
+) -> Result<Response<Full<Bytes>>, hyper::Error> {
     let response = match (req.method(), req.uri().path()) {
         (&Method::GET, "/metrics") => {
             let mut buffer = vec![];
@@ -77,12 +90,12 @@ async fn metrics_req(
             Response::builder()
                 .status(200)
                 .header(CONTENT_TYPE, encoder.format_type())
-                .body(Body::from(buffer))
+                .body(buffer.into())
                 .unwrap()
         }
         _ => Response::builder()
             .status(404)
-            .body(Body::empty())
+            .body(vec![].into())
             .expect("Can't fail to construct empty resp"),
     };
     Ok(response)

core/src/worker/activities/local_activities.rs

@@ -97,7 +97,7 @@ impl LocalActivityExecutionResult {
                         ..
                     }),
                 ..
-            }) => TimeoutType::from_i32(*timeout_type),
+            }) => TimeoutType::try_from(*timeout_type).ok(),
             _ => None,
         }
     }

core/src/worker/workflow/machines/activity_state_machine.rs

@@ -117,8 +117,8 @@ impl ActivityMachine {
         let mut s = Self::from_parts(
             Created {}.into(),
             SharedState {
-                cancellation_type: ActivityCancellationType::from_i32(attrs.cancellation_type)
-                    .unwrap(),
+                cancellation_type: ActivityCancellationType::try_from(attrs.cancellation_type)
+                    .unwrap_or(ActivityCancellationType::TryCancel),
                 attrs,
                 internal_flags,
                 scheduled_event_id: 0,