Skip to content

💥 Replace OTel Prometheus Exporter #942

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Sushisource merged 19 commits into from
Jul 3, 2025
Merged

💥 Replace OTel Prometheus Exporter #942

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
f4959c0
Agents updates
Sushisource May 16, 2025
e62b836
Partially complete
Sushisource May 29, 2025
7769292
Remove otel prometheus
Sushisource Jun 2, 2025
f4ac3c5
Use prometheus directly instead of otel prometheus exporter
Sushisource Jun 3, 2025
8a8cfb4
Cache labels->metrics internally
Sushisource Jun 4, 2025
f61b292
Make registry allow for dupes
Sushisource Jun 6, 2025
7f595b1
Only one top-level cache
Sushisource Jun 11, 2025
b1db6db
Change to `with_attributes`
Sushisource Jun 18, 2025
9e15229
Boxification
Sushisource Jun 20, 2025
33f8c2d
Use Prom more efficiently
Sushisource Jun 20, 2025
40b3241
Lazily bind metrics
Sushisource Jun 21, 2025
8e3b704
Test cleanup
Sushisource Jun 23, 2025
c02ccda
Return results from with attributes
Sushisource Jun 23, 2025
5e22714
More cleanup
Sushisource Jun 23, 2025
33cc3ad
Some quick benching
Sushisource Jun 24, 2025
d4553e2
Fix span-drop issue in heavy tests
Sushisource Jun 26, 2025
f7592d7
Changes to fix issues discovered when using from Python
Sushisource Jul 3, 2025
545d130
Fix C bridge
Sushisource Jul 3, 2025
424f7bf
Lint fix
Sushisource Jul 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@ cargo integ-test # integration tests (starts ephemeral server by default)
cargo test --test heavy_tests # load tests -- agents do not need to run this and should not
```

Rust compilation can take some time. Do not interrupt builds or tests unless they are taking more
than 10 minutes.

Additional checks:

```bash
Expand All @@ -63,6 +66,8 @@ Documentation can be generated with `cargo doc`.
Reviewers will look for:

- All builds, tests, and lints passing in CI
- Note that some tests cause intentional panics. That does not mean the test failed. You should
only consider tests that have failed according to the harness to be a real problem.
- New tests covering behavior changes
- Clear and concise code following existing style (see `README.md` for error handling guidance)
- Documentation updates for any public API changes
Expand Down
8 changes: 4 additions & 4 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,11 @@ license-file = "LICENSE.txt"

[workspace.dependencies]
derive_builder = "0.20"
derive_more = { version = "2.0", features = ["constructor", "display", "from", "into", "debug"] }
derive_more = { version = "2.0", features = ["constructor", "display", "from", "into", "debug", "try_into"] }
thiserror = "2"
tonic = "0.12"
tonic-build = "0.12"
opentelemetry = { version = "0.29", features = ["metrics"] }
tonic = "0.13"
tonic-build = "0.13"
opentelemetry = { version = "0.30", features = ["metrics"] }
prost = "0.13"
prost-types = "0.13"

Expand Down
5 changes: 3 additions & 2 deletions client/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,10 @@ backoff = "0.4"
base64 = "0.22"
derive_builder = { workspace = true }
derive_more = { workspace = true }
bytes = "1.10"
futures-util = { version = "0.3", default-features = false }
futures-retry = "0.6.0"
http = "1.1.0"
http = "1.1"
http-body-util = "0.1"
hyper = { version = "1.4.1" }
hyper-util = "0.1.6"
Expand All @@ -31,7 +32,7 @@ parking_lot = "0.12"
slotmap = "1.0"
thiserror = { workspace = true }
tokio = "1.1"
tonic = { workspace = true, features = ["tls", "tls-roots"] }
tonic = { workspace = true, features = ["tls-ring", "tls-native-roots"] }
Copy link
Copy Markdown
Contributor

@cretz cretz Jun 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this related to the changes for this PR? Any concerns changing this?

Copy link
Copy Markdown
Member Author

@Sushisource Sushisource Jun 25, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's part of the reason for all this, so that we could upgrade Tonic. The new feature flags should be the exact equivalent of what it was previously, they just got renamed

tower = { version = "0.5", features = ["util"] }
tracing = "0.1"
url = "2.2"
Expand Down
23 changes: 16 additions & 7 deletions client/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ use temporal_sdk_core_protos::{
};
use tonic::{
Code,
body::BoxBody,
body::Body,
client::GrpcService,
codegen::InterceptedService,
metadata::{MetadataKey, MetadataMap, MetadataValue},
Expand Down Expand Up @@ -595,7 +595,7 @@ fn get_decode_max_size() -> usize {
impl<T> TemporalServiceClient<T>
where
T: Clone,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
<T::ResponseBody as tonic::codegen::Body>::Error: Into<tonic::codegen::StdError> + Send,
Expand Down Expand Up @@ -1175,13 +1175,13 @@ impl From<common::v1::Priority> for Priority {
impl<T> WorkflowClientTrait for T
where
T: RawClientLike + NamespacedClient + Clone + Send + Sync + 'static,
<Self as RawClientLike>::SvcType: GrpcService<BoxBody> + Send + Clone + 'static,
<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::ResponseBody:
<Self as RawClientLike>::SvcType: GrpcService<Body> + Send + Clone + 'static,
<<Self as RawClientLike>::SvcType as GrpcService<Body>>::ResponseBody:
tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::Error:
<<Self as RawClientLike>::SvcType as GrpcService<Body>>::Error:
Into<tonic::codegen::StdError>,
<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::Future: Send,
<<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::ResponseBody
<<Self as RawClientLike>::SvcType as GrpcService<Body>>::Future: Send,
<<<Self as RawClientLike>::SvcType as GrpcService<Body>>::ResponseBody
as tonic::codegen::Body>::Error: Into<tonic::codegen::StdError> + Send,
{
async fn start_workflow(
Expand Down Expand Up @@ -1673,6 +1673,15 @@ impl<T> RequestExt for tonic::Request<T> {
}
}

macro_rules! dbg_panic {
($($arg:tt)*) => {
use tracing::error;
error!($($arg)*);
debug_assert!(false, $($arg)*);
};
}
pub(crate) use dbg_panic;

#[cfg(test)]
mod tests {
use super::*;
Expand Down
80 changes: 57 additions & 23 deletions client/src/metrics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
use crate::{AttachMetricLabels, CallType};
use crate::{AttachMetricLabels, CallType, dbg_panic};
use futures_util::{FutureExt, future::BoxFuture};
use std::{
sync::Arc,
task::{Context, Poll},
time::{Duration, Instant},
};
use temporal_sdk_core_api::telemetry::metrics::{
CoreMeter, Counter, HistogramDuration, MetricAttributes, MetricKeyValue, MetricParameters,
TemporalMeter,
CoreMeter, Counter, CounterBase, HistogramDuration, HistogramDurationBase, MetricAttributable,
MetricAttributes, MetricKeyValue, MetricParameters, TemporalMeter,
};
use tonic::{Code, body::BoxBody, transport::Channel};
use tonic::{Code, body::Body, transport::Channel};
use tower::Service;

/// The string name (which may be prefixed) for this metric
Expand All @@ -26,22 +26,24 @@ pub(crate) struct MetricsContext {
meter: Arc<dyn CoreMeter>,
kvs: MetricAttributes,
poll_is_long: bool,
instruments: Instruments,
}
#[derive(Clone)]
struct Instruments {
svc_request: Counter,
svc_request_failed: Counter,
long_svc_request: Counter,
long_svc_request_failed: Counter,

svc_request: Arc<dyn Counter>,
svc_request_failed: Arc<dyn Counter>,
long_svc_request: Arc<dyn Counter>,
long_svc_request_failed: Arc<dyn Counter>,

svc_request_latency: Arc<dyn HistogramDuration>,
long_svc_request_latency: Arc<dyn HistogramDuration>,
svc_request_latency: HistogramDuration,
long_svc_request_latency: HistogramDuration,
}

impl MetricsContext {
pub(crate) fn new(tm: TemporalMeter) -> Self {
let meter = tm.inner;
Self {
kvs: meter.new_attributes(tm.default_attribs),
poll_is_long: false,
let kvs = meter.new_attributes(tm.default_attribs);
let instruments = Instruments {
svc_request: meter.counter(MetricParameters {
name: "request".into(),
description: "Count of client request successes by rpc name".into(),
Expand Down Expand Up @@ -72,6 +74,11 @@ impl MetricsContext {
unit: "duration".into(),
description: "Histogram of client long-poll request latencies".into(),
}),
};
Self {
kvs,
poll_is_long: false,
instruments,
meter,
}
}
Expand All @@ -81,6 +88,33 @@ impl MetricsContext {
self.kvs = self
.meter
.extend_attributes(self.kvs.clone(), new_kvs.into());

let _ = self
.instruments
.svc_request
.with_attributes(&self.kvs)
.and_then(|v| {
self.instruments.svc_request = v;
self.instruments.long_svc_request.with_attributes(&self.kvs)
})
.and_then(|v| {
self.instruments.long_svc_request = v;
self.instruments
.svc_request_latency
.with_attributes(&self.kvs)
})
.and_then(|v| {
self.instruments.svc_request_latency = v;
self.instruments
.long_svc_request_latency
.with_attributes(&self.kvs)
})
.map(|v| {
self.instruments.long_svc_request_latency = v;
})
.inspect_err(|e| {
dbg_panic!("Failed to extend client metrics attributes: {:?}", e);
});
}

pub(crate) fn set_is_long_poll(&mut self) {
Expand All @@ -90,9 +124,9 @@ impl MetricsContext {
/// A request to the temporal service was made
pub(crate) fn svc_request(&self) {
if self.poll_is_long {
self.long_svc_request.add(1, &self.kvs);
self.instruments.long_svc_request.adds(1);
} else {
self.svc_request.add(1, &self.kvs);
self.instruments.svc_request.adds(1);
}
}

Expand All @@ -108,18 +142,18 @@ impl MetricsContext {
&self.kvs
};
if self.poll_is_long {
self.long_svc_request_failed.add(1, kvs);
self.instruments.long_svc_request_failed.add(1, kvs);
} else {
self.svc_request_failed.add(1, kvs);
self.instruments.svc_request_failed.add(1, kvs);
}
}

/// Record service request latency
pub(crate) fn record_svc_req_latency(&self, dur: Duration) {
if self.poll_is_long {
self.long_svc_request_latency.record(dur, &self.kvs);
self.instruments.long_svc_request_latency.records(dur);
} else {
self.svc_request_latency.record(dur, &self.kvs);
self.instruments.svc_request_latency.records(dur);
}
}
}
Expand Down Expand Up @@ -177,16 +211,16 @@ pub struct GrpcMetricSvc {
pub(crate) disable_errcode_label: bool,
}

impl Service<http::Request<BoxBody>> for GrpcMetricSvc {
type Response = http::Response<BoxBody>;
impl Service<http::Request<Body>> for GrpcMetricSvc {
type Response = http::Response<Body>;
type Error = tonic::transport::Error;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx).map_err(Into::into)
}

fn call(&mut self, mut req: http::Request<BoxBody>) -> Self::Future {
fn call(&mut self, mut req: http::Request<Body>) -> Self::Future {
let metrics = self
.metrics
.clone()
Expand Down
26 changes: 13 additions & 13 deletions client/src/raw.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ use temporal_sdk_core_protos::{
};
use tonic::{
Request, Response, Status,
body::BoxBody,
body::Body,
client::GrpcService,
metadata::{AsciiMetadataValue, KeyAndValueRef},
};
Expand Down Expand Up @@ -166,7 +166,7 @@ where
impl<T> RawClientLike for TemporalServiceClient<T>
where
T: Send + Sync + Clone + 'static,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
<T::ResponseBody as tonic::codegen::Body>::Error: Into<tonic::codegen::StdError> + Send,
Expand Down Expand Up @@ -221,7 +221,7 @@ where
impl<T> RawClientLike for ConfiguredClient<TemporalServiceClient<T>>
where
T: Send + Sync + Clone + 'static,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
<T::ResponseBody as tonic::codegen::Body>::Error: Into<tonic::codegen::StdError> + Send,
Expand Down Expand Up @@ -373,7 +373,7 @@ pub(super) struct IsUserLongPoll;
impl<RC, T> WorkflowService for RC
where
RC: RawClientLike<SvcType = T>,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
T::Future: Send,
Expand All @@ -383,7 +383,7 @@ where
impl<RC, T> OperatorService for RC
where
RC: RawClientLike<SvcType = T>,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
T::Future: Send,
Expand All @@ -393,7 +393,7 @@ where
impl<RC, T> CloudService for RC
where
RC: RawClientLike<SvcType = T>,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
T::Future: Send,
Expand All @@ -403,7 +403,7 @@ where
impl<RC, T> TestService for RC
where
RC: RawClientLike<SvcType = T>,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
T::Future: Send,
Expand All @@ -413,7 +413,7 @@ where
impl<RC, T> HealthService for RC
where
RC: RawClientLike<SvcType = T>,
T: GrpcService<BoxBody> + Send + Clone + 'static,
T: GrpcService<Body> + Send + Clone + 'static,
T::ResponseBody: tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
T::Error: Into<tonic::codegen::StdError>,
T::Future: Send,
Expand Down Expand Up @@ -483,13 +483,13 @@ macro_rules! proxier {
pub trait $trait_name: RawClientLike
where
// Yo this is wild
<Self as RawClientLike>::SvcType: GrpcService<BoxBody> + Send + Clone + 'static,
<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::ResponseBody:
<Self as RawClientLike>::SvcType: GrpcService<Body> + Send + Clone + 'static,
<<Self as RawClientLike>::SvcType as GrpcService<Body>>::ResponseBody:
tonic::codegen::Body<Data = tonic::codegen::Bytes> + Send + 'static,
<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::Error:
<<Self as RawClientLike>::SvcType as GrpcService<Body>>::Error:
Into<tonic::codegen::StdError>,
<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::Future: Send,
<<<Self as RawClientLike>::SvcType as GrpcService<BoxBody>>::ResponseBody
<<Self as RawClientLike>::SvcType as GrpcService<Body>>::Future: Send,
<<<Self as RawClientLike>::SvcType as GrpcService<Body>>::ResponseBody
as tonic::codegen::Body>::Error: Into<tonic::codegen::StdError> + Send,
{
$(
Expand Down
4 changes: 2 additions & 2 deletions core-api/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,10 +14,9 @@ categories = ["development-tools"]

[features]
otel_impls = ["dep:opentelemetry"]
envconfig = ["dep:toml", "dep:serde", "dep:dirs", "dep:anyhow"]
envconfig = ["dep:toml", "dep:serde", "dep:dirs"]

[dependencies]
anyhow = { version = "1.0", optional = true }
async-trait = "0.1"
dirs = { version = "5.0", optional = true }
derive_builder = { workspace = true }
Expand All @@ -29,6 +28,7 @@ serde_json = "1.0"
thiserror = { workspace = true }
toml = { version = "0.8", optional = true }
tonic = { workspace = true }
tracing = "0.1"
tracing-core = "0.1"
url = "2.3"

Expand Down
2 changes: 1 addition & 1 deletion core-api/src/envconfig.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ pub enum ConfigError {
InvalidConfig(String),

#[error("Configuration loading error: {0}")]
LoadError(anyhow::Error),
LoadError(Box<dyn std::error::Error>),
Copy link
Copy Markdown
Contributor

@cretz cretz Jun 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This also seems unrelated?

Copy link
Copy Markdown
Member Author

@Sushisource Sushisource Jun 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's just something i didn't catch in the review for envconfig - I don't want anyhow in the "public" core api.

}

impl From<std::str::Utf8Error> for ConfigError {
Expand Down
9 changes: 9 additions & 0 deletions core-api/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,3 +138,12 @@ pub trait Worker: Send + Sync {
/// functions have returned `ShutDown` errors.
async fn finalize_shutdown(self);
}

macro_rules! dbg_panic {
($($arg:tt)*) => {
use tracing::error;
error!($($arg)*);
debug_assert!(false, $($arg)*);
};
}
pub(crate) use dbg_panic;
Loading
Loading