diff --git a/Cargo.lock b/Cargo.lock index 93df0ac..93de8d4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -310,6 +310,12 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "dtoa" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c3cf4824e2d5f025c7b531afcb2325364084a16806f6d47fbc1f5fbd9960590" + [[package]] name = "dunce" version = "1.0.5" @@ -391,9 +397,13 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" name = "function-sdk-rust" version = "0.2.0" dependencies = [ + "bytes", "clap", + "http", + "http-body", "pbjson", "pbjson-types", + "prometheus-client", "prost", "serde", "serde_json", @@ -405,6 +415,8 @@ dependencies = [ "tonic-health", "tonic-prost", "tonic-reflection", + "tower-layer", + "tower-service", "tracing", "tracing-subscriber", ] @@ -678,6 +690,15 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.34" @@ -749,6 +770,29 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "pbjson" version = "0.9.0" @@ -854,6 +898,29 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prometheus-client" +version = "0.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba70bf887030e45213b4a95c9b08d5a450b157f87c1d63661ed0847a12fa2aad" +dependencies = [ + "dtoa", + "itoa", + "parking_lot", + "prometheus-client-derive-encode", +] + +[[package]] +name = "prometheus-client-derive-encode" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9adf1691c04c0a5ff46ff8f262b58beb07b0dbb61f96f9f54f6cbd82106ed87f" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "prost" version = "0.14.4" @@ -942,6 +1009,15 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.13.1" @@ -1034,6 +1110,12 @@ dependencies = [ "untrusted", ] +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "serde" version = "1.0.229" diff --git a/README.md b/README.md index 3c87daf..6f03cc6 100644 --- a/README.md +++ b/README.md @@ -97,14 +97,23 @@ with a few deliberate differences: reports the function as serving. - **Graceful shutdown is built in.** The server drains in-flight requests on SIGTERM and SIGINT. -- **No Prometheus metrics yet.** function-sdk-go serves gRPC server metrics - on `:8080` (`WithMetricsServer`). As of August 2026 there is no maintained - tonic-compatible Prometheus layer: the only purpose-built crate, - [tonic-prometheus-layer], is pinned to tonic 0.13 and lags tonic majors, - so depending on it would hold this SDK back. When metrics are needed the - plan is to hand-roll a small tower layer on [prometheus-client] (the - actively maintained official Rust client) with a `/metrics` endpoint, - rather than take that dependency. +- **Metrics are Go-compatible, served as OpenMetrics first.** Like + function-sdk-go, the server serves the gRPC server series on `:8080` at + `/metrics` (`--metrics-address`, empty disables): + `grpc_server_started_total`, `grpc_server_handled_total` (with + `grpc_code`), `grpc_server_msg_received_total` and + `grpc_server_msg_sent_total`, with the interceptor's exact names, labels + and help strings - dashboards built for Go functions work unchanged. As + in the Go SDK, only unary calls are counted (streaming reflection and + health `Watch` exist as permanently zero series), every served method is + pre-created at zero, and there is no handling-time histogram. The + exposition differs deliberately: OpenMetrics 1.0 is the main format + (readable beyond Prometheus), with the classic Prometheus text format + served to an `Accept` header that asks for it. Built on + [prometheus-client], the official OpenMetrics-native client - the only + purpose-built alternative, [tonic-prometheus-layer], is pinned to tonic + 0.13 and would hold this SDK back. Since this SDK serves no v1beta1, the + v1beta1 series the Go SDK pre-created do not exist here. [crossplane]: https://www.crossplane.io [functions]: https://docs.crossplane.io/latest/composition/compositions/ diff --git a/sdk/Cargo.toml b/sdk/Cargo.toml index 72f7eac..f3615e8 100644 --- a/sdk/Cargo.toml +++ b/sdk/Cargo.toml @@ -10,6 +10,10 @@ keywords = ["crossplane", "kubernetes", "grpc"] categories = ["api-bindings"] [dependencies] +bytes = "1" +http = "1" +http-body = "1" +prometheus-client = "0.25" tonic = { version = "0.14", features = ["tls-aws-lc"] } tonic-prost = "0.14" tonic-health = "0.14" @@ -22,6 +26,8 @@ serde_json = "1" tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal"] } clap = { version = "4", features = ["derive", "env"] } tracing = "0.1" +tower-layer = "0.3" +tower-service = "0.3" tracing-subscriber = { version = "0.3", features = ["json"] } sha2 = "0.11" thiserror = "2" diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index 82e6d7c..d4758f1 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -14,6 +14,8 @@ //! //! [`FunctionRunnerService`]: proto::v1::function_runner_service_server::FunctionRunnerService +mod metrics; + pub mod context; pub mod logging; pub mod proto; diff --git a/sdk/src/metrics.rs b/sdk/src/metrics.rs new file mode 100644 index 0000000..eb8d66f --- /dev/null +++ b/sdk/src/metrics.rs @@ -0,0 +1,534 @@ +//! The gRPC server metrics function-sdk-go serves, re-created for this +//! SDK's server: `grpc_server_started_total`, `grpc_server_handled_total`, +//! `grpc_server_msg_received_total` and `grpc_server_msg_sent_total`, with +//! the names, labels and help strings of the grpc-prometheus interceptor +//! the Go SDK installs - so dashboards and alerts work across Go and Rust +//! functions unchanged. Like that interceptor, which was unary-only, +//! streaming methods (reflection, health Watch) are not counted; like its +//! InitializeMetrics, every method the server carries starts as a zero +//! series; and like the Go SDK, there is no handling-time histogram (it +//! never enabled one). +//! +//! The /metrics endpoint (`--metrics-address`, the Go SDK's `:8080` +//! default, empty disables) serves OpenMetrics 1.0 as its main format, +//! straight from prometheus-client's encoder; a scraper whose Accept +//! header asks for the classic Prometheus text format without accepting +//! OpenMetrics gets that instead, derived from the same rendering. + +use std::pin::Pin; +use std::sync::{LazyLock, RwLock}; +use std::task::{Context, Poll}; + +use prometheus_client::metrics::counter::Counter; +use prometheus_client::metrics::family::Family; +use prometheus_client::registry::Registry; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +const FUNCTION_SERVICE: &str = "apiextensions.fn.proto.v1.FunctionRunnerService"; +const HEALTH_SERVICE: &str = "grpc.health.v1.Health"; + +/// The registry the series live in - prometheus-client has no default +/// registry, so the server's is this one. +static REGISTRY: LazyLock> = LazyLock::new(|| RwLock::new(Registry::default())); + +type Labels = Vec<(String, String)>; + +/// A counter family addressed by label values in a fixed key order. The +/// name registers without its `_total` suffix and the help without its +/// trailing period - the OpenMetrics encoder appends both - so the served +/// text is byte-identical to the Go SDK's. +struct LabeledCounter { + keys: &'static [&'static str], + family: Family, +} + +impl LabeledCounter { + fn new(name: &str, help: &str, keys: &'static [&'static str]) -> Self { + let family = Family::::default(); + let help = help.strip_suffix('.').unwrap_or(help); + REGISTRY + .write() + .expect("poisoned") + .register(name, help, family.clone()); + LabeledCounter { keys, family } + } + + fn with_label_values(&self, values: &[&str]) -> impl std::ops::Deref { + debug_assert_eq!(self.keys.len(), values.len()); + let labels: Labels = self + .keys + .iter() + .zip(values) + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + self.family.get_or_create(&labels) + } +} + +static STARTED: LazyLock = LazyLock::new(|| { + LabeledCounter::new( + "grpc_server_started", + "Total number of RPCs started on the server.", + &["grpc_type", "grpc_service", "grpc_method"], + ) +}); + +static HANDLED: LazyLock = LazyLock::new(|| { + LabeledCounter::new( + "grpc_server_handled", + "Total number of RPCs completed on the server, regardless of success or failure.", + &["grpc_type", "grpc_service", "grpc_method", "grpc_code"], + ) +}); + +static MSG_RECEIVED: LazyLock = LazyLock::new(|| { + LabeledCounter::new( + "grpc_server_msg_received", + "Total number of RPC stream messages received on the server.", + &["grpc_type", "grpc_service", "grpc_method"], + ) +}); + +static MSG_SENT: LazyLock = LazyLock::new(|| { + LabeledCounter::new( + "grpc_server_msg_sent", + "Total number of gRPC stream messages sent by the server.", + &["grpc_type", "grpc_service", "grpc_method"], + ) +}); + +/// The gRPC status code names, indexed by their wire value - the strings +/// google.golang.org/grpc/codes renders, which is what the Go SDK's +/// grpc_code label carried. +const CODES: [&str; 17] = [ + "OK", + "Canceled", + "Unknown", + "InvalidArgument", + "DeadlineExceeded", + "NotFound", + "AlreadyExists", + "PermissionDenied", + "ResourceExhausted", + "FailedPrecondition", + "Aborted", + "OutOfRange", + "Unimplemented", + "Internal", + "Unavailable", + "DataLoss", + "Unauthenticated", +]; + +fn code_str(n: u32) -> &'static str { + CODES.get(n as usize).copied().unwrap_or("Unknown") +} + +/// The unary methods the server carries - the only calls the Go SDK's +/// interceptor counted (it installed no stream interceptor). +fn unary_method(path: &str) -> Option<(&'static str, &'static str)> { + match path { + "/apiextensions.fn.proto.v1.FunctionRunnerService/RunFunction" => { + Some((FUNCTION_SERVICE, "RunFunction")) + } + "/grpc.health.v1.Health/Check" => Some((HEALTH_SERVICE, "Check")), + _ => None, + } +} + +/// Pre-creates every series the server can emit at zero, the way +/// grpc-prometheus's InitializeMetrics did for the Go SDK: a scrape sees +/// each method of each registered service from the first request on, +/// streaming methods included (present and forever zero, exactly as under +/// the Go SDK's unary-only interceptor). +pub(crate) fn initialize() { + let methods: [(&str, &str, &str); 5] = [ + ("unary", FUNCTION_SERVICE, "RunFunction"), + ("unary", HEALTH_SERVICE, "Check"), + ("server_stream", HEALTH_SERVICE, "Watch"), + ( + "bidi_stream", + "grpc.reflection.v1.ServerReflection", + "ServerReflectionInfo", + ), + ( + "bidi_stream", + "grpc.reflection.v1alpha.ServerReflection", + "ServerReflectionInfo", + ), + ]; + for (kind, service, method) in methods { + let labels = [kind, service, method]; + let _ = STARTED.with_label_values(&labels); + let _ = MSG_RECEIVED.with_label_values(&labels); + let _ = MSG_SENT.with_label_values(&labels); + for code in CODES { + let _ = HANDLED.with_label_values(&[kind, service, method, code]); + } + } +} + +/// One unary call's bookkeeping: started and msg_received on creation, +/// handled (and msg_sent on OK) exactly once when the status is known. A +/// reporter dropped before any status - the caller went away mid-call - is +/// a Canceled, which is what the Go interceptor reported for a canceled +/// context. +struct Reporter { + service: &'static str, + method: &'static str, + done: bool, +} + +impl Reporter { + fn start(service: &'static str, method: &'static str) -> Self { + STARTED.with_label_values(&["unary", service, method]).inc(); + MSG_RECEIVED + .with_label_values(&["unary", service, method]) + .inc(); + Reporter { + service, + method, + done: false, + } + } + + fn report(&mut self, code: u32) { + if self.done { + return; + } + self.done = true; + HANDLED + .with_label_values(&["unary", self.service, self.method, code_str(code)]) + .inc(); + if code == 0 { + MSG_SENT + .with_label_values(&["unary", self.service, self.method]) + .inc(); + } + } +} + +impl Drop for Reporter { + fn drop(&mut self) { + if !self.done { + self.done = true; + HANDLED + .with_label_values(&["unary", self.service, self.method, "Canceled"]) + .inc(); + } + } +} + +/// Reads a numeric grpc-status from headers or trailers; a gRPC response +/// with no grpc-status is out of spec, so the caller falls back to Unknown. +fn status_in(map: &http::HeaderMap) -> Option { + map.get("grpc-status")?.to_str().ok()?.parse().ok() +} + +/// The tower layer serve() wraps the whole router in, so the health service +/// is counted exactly as the Go SDK's server-wide interceptor counted it. +#[derive(Clone, Copy, Default)] +pub(crate) struct MetricsLayer; + +impl tower_layer::Layer for MetricsLayer { + type Service = Metrics; + + fn layer(&self, inner: S) -> Metrics { + Metrics { inner } + } +} + +#[derive(Clone)] +pub(crate) struct Metrics { + inner: S, +} + +impl tower_service::Service> for Metrics +where + S: tower_service::Service, Response = http::Response> + + Clone + + Send + + 'static, + S::Future: Send + 'static, + ReqBody: Send + 'static, +{ + type Response = http::Response; + type Error = S::Error; + type Future = + Pin> + Send + 'static>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, req: http::Request) -> Self::Future { + let unary = unary_method(req.uri().path()); + // The readied service is the one that must take the call; the clone + // parks in self (the tower Service contract). + let clone = self.inner.clone(); + let mut inner = std::mem::replace(&mut self.inner, clone); + Box::pin(async move { + let mut reporter = unary.map(|(service, method)| Reporter::start(service, method)); + let rsp = inner.call(req).await?; + let (parts, body) = rsp.into_parts(); + // A trailers-only response (tonic's immediate refusals) carries + // grpc-status in the headers; report now and wrap nothing. + if let (Some(r), Some(code)) = (reporter.as_mut(), status_in(&parts.headers)) { + r.report(code); + reporter = None; + } + Ok(http::Response::from_parts( + parts, + MetricsBody { + inner: body, + reporter, + }, + )) + }) + } +} + +/// The response body with the call's reporter riding along: the status of a +/// normal unary response is in the trailers frame, so that is where handled +/// is counted - and a body dropped before its trailers is the client going +/// away, the reporter's Canceled. +pub(crate) struct MetricsBody { + inner: tonic::body::Body, + reporter: Option, +} + +impl http_body::Body for MetricsBody { + type Data = bytes::Bytes; + type Error = tonic::Status; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + let this = self.get_mut(); + match std::task::ready!(Pin::new(&mut this.inner).poll_frame(cx)) { + Some(Ok(frame)) => { + if let (Some(trailers), Some(r)) = (frame.trailers_ref(), this.reporter.as_mut()) { + r.report(status_in(trailers).unwrap_or(2)); + } + Poll::Ready(Some(Ok(frame))) + } + Some(Err(e)) => { + if let Some(r) = this.reporter.as_mut() { + r.report(2); + } + Poll::Ready(Some(Err(e))) + } + None => { + // A stream that ends without trailers never carried a + // status; out of spec, counted as Unknown like a broken + // response would be. + if let Some(r) = this.reporter.as_mut() { + r.report(2); + } + Poll::Ready(None) + } + } + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> http_body::SizeHint { + http_body::Body::size_hint(&self.inner) + } +} + +const OPENMETRICS_CONTENT_TYPE: &str = "application/openmetrics-text; version=1.0.0; charset=utf-8"; + +/// The registry rendered as OpenMetrics 1.0 text, straight from +/// prometheus-client's encoder. +fn render_openmetrics() -> String { + let mut out = String::new(); + let _ = + prometheus_client::encoding::text::encode(&mut out, ®ISTRY.read().expect("poisoned")); + out +} + +/// The classic Prometheus text format (`text/plain; version=0.0.4`), +/// derived from the OpenMetrics rendering: the two differ only in a counter +/// family being named with its `_total` suffix in the classic HELP and TYPE +/// lines, and in the terminating `# EOF` - the samples are identical. +fn classic_from_openmetrics(om: &str) -> String { + use std::fmt::Write as _; + let counters: std::collections::HashSet<&str> = om + .lines() + .filter_map(|l| l.strip_prefix("# TYPE ")?.strip_suffix(" counter")) + .collect(); + let mut out = String::new(); + for line in om.lines() { + if line == "# EOF" { + continue; + } + if let Some(name) = line + .strip_prefix("# TYPE ") + .and_then(|rest| rest.strip_suffix(" counter")) + { + let _ = writeln!(out, "# TYPE {name}_total counter"); + continue; + } + if let Some((name, help)) = line + .strip_prefix("# HELP ") + .and_then(|rest| rest.split_once(' ')) + && counters.contains(name) + { + let _ = writeln!(out, "# HELP {name}_total {help}"); + continue; + } + let _ = writeln!(out, "{line}"); + } + out +} + +/// Whether the request's Accept header asks for the classic Prometheus text +/// format without accepting OpenMetrics. OpenMetrics is the main format: it +/// is served when the header names it (whatever the q-values), when there +/// is no Accept header, and for catch-alls like */*; only an explicit +/// text/plain-and-not-OpenMetrics preference gets the classic rendering. +fn wants_classic_text(request_head: &str) -> bool { + let Some(accept) = request_head.lines().find_map(|l| { + let (name, value) = l.split_once(':')?; + name.trim().eq_ignore_ascii_case("accept").then_some(value) + }) else { + return false; + }; + !accept.contains("application/openmetrics-text") && accept.contains("text/plain") +} + +/// Serves the metrics at /metrics on address until the process ends - the +/// plain-HTTP listener the Go SDK starts beside its gRPC server. +pub(crate) async fn serve(address: &str) -> Result<(), std::io::Error> { + // The Go SDK's ":8080" shorthand means every interface. + let address = if address.starts_with(':') { + format!("0.0.0.0{address}") + } else { + address.to_string() + }; + let listener = tokio::net::TcpListener::bind(&address).await?; + tracing::info!(address = %address, "serving metrics"); + loop { + let Ok((mut conn, _)) = listener.accept().await else { + continue; + }; + tokio::spawn(async move { + let mut buf = [0u8; 1024]; + let n = conn.read(&mut buf).await.unwrap_or(0); + let head = String::from_utf8_lossy(&buf[..n]).into_owned(); + let path = head.split_whitespace().nth(1).unwrap_or_default(); + let response = if path == "/metrics" { + let om = render_openmetrics(); + let (body, content_type) = if wants_classic_text(&head) { + (classic_from_openmetrics(&om), "text/plain; version=0.0.4") + } else { + (om, OPENMETRICS_CONTENT_TYPE) + }; + format!( + "HTTP/1.1 200 OK\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + } else { + "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + .to_string() + }; + let _ = conn.write_all(response.as_bytes()).await; + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn code_names_match_grpc_go() { + assert_eq!(code_str(0), "OK"); + assert_eq!(code_str(1), "Canceled"); + assert_eq!(code_str(12), "Unimplemented"); + assert_eq!(code_str(16), "Unauthenticated"); + assert_eq!(code_str(17), "Unknown"); + } + + #[test] + fn only_unary_methods_are_counted() { + assert_eq!( + unary_method("/apiextensions.fn.proto.v1.FunctionRunnerService/RunFunction"), + Some((FUNCTION_SERVICE, "RunFunction")) + ); + assert_eq!( + unary_method("/grpc.health.v1.Health/Check"), + Some((HEALTH_SERVICE, "Check")) + ); + // Streaming methods stay uncounted, as under the Go SDK's + // unary-only interceptor. + assert_eq!(unary_method("/grpc.health.v1.Health/Watch"), None); + assert_eq!( + unary_method("/grpc.reflection.v1.ServerReflection/ServerReflectionInfo"), + None + ); + } + + #[test] + fn openmetrics_is_the_main_format() { + assert!(!wants_classic_text("GET /metrics HTTP/1.1\r\nHost: x\r\n")); + assert!(!wants_classic_text( + "GET /metrics HTTP/1.1\r\nAccept: */*\r\n" + )); + assert!(!wants_classic_text( + "GET /metrics HTTP/1.1\r\nAccept: application/openmetrics-text;version=1.0.0;q=0.9,text/plain;version=0.0.4;q=0.5\r\n" + )); + assert!(wants_classic_text( + "GET /metrics HTTP/1.1\r\nAccept: text/plain; version=0.0.4\r\n" + )); + } + + #[test] + fn renderings_differ_only_where_the_formats_do() { + initialize(); + let om = render_openmetrics(); + assert!(om.ends_with("# EOF\n"), "OpenMetrics ends with EOF:\n{om}"); + assert!( + om.contains("# TYPE grpc_server_started counter"), + "counter family named without _total:\n{om}" + ); + assert!( + om.contains("# HELP grpc_server_started Total number of RPCs started on the server."), + "the Go SDK's help string, single trailing period:\n{om}" + ); + assert!(!om.contains("server.."), "no doubled period:\n{om}"); + + let classic = classic_from_openmetrics(&om); + assert!(classic.contains("# TYPE grpc_server_started_total counter")); + assert!(classic.contains("# HELP grpc_server_started_total ")); + assert!(!classic.contains("# EOF")); + } + + #[test] + fn a_dropped_reporter_is_a_canceled_call() { + initialize(); + let canceled = || { + render_openmetrics() + .lines() + .find(|l| { + l.starts_with("grpc_server_handled_total{grpc_type=\"unary\"") + && l.contains("grpc_method=\"RunFunction\"") + && l.contains("grpc_code=\"Canceled\"") + }) + .and_then(|l| l.rsplit(' ').next()?.parse::().ok()) + .unwrap_or(0.0) + }; + let before = canceled(); + drop(Reporter::start(FUNCTION_SERVICE, "RunFunction")); + assert_eq!(canceled(), before + 1.0); + // A reported call does not double-count on drop. + let mut r = Reporter::start(FUNCTION_SERVICE, "RunFunction"); + r.report(0); + drop(r); + assert_eq!(canceled(), before + 1.0); + } +} diff --git a/sdk/src/server.rs b/sdk/src/server.rs index 52f83ee..e0e72c8 100644 --- a/sdk/src/server.rs +++ b/sdk/src/server.rs @@ -34,6 +34,12 @@ pub struct Args { /// Defaults to the gRPC default of 4MB. #[arg(long)] pub max_recv_message_size: Option, + + /// Address at which to serve metrics at /metrics (OpenMetrics 1.0, or + /// the classic Prometheus text format for an Accept header that asks + /// for it) - the Go SDK's default; empty disables them. + #[arg(long, env = "METRICS_ADDRESS", default_value = ":8080")] + pub metrics_address: String, } /// Starts a gRPC server and serves RunFunctionRequests until SIGTERM or @@ -44,6 +50,11 @@ pub struct Args { /// to authenticate Crossplane) unless `--insecure` is set. gRPC server /// reflection is enabled for both the v1 and v1alpha reflection APIs, and /// the gRPC health service reports the function as serving. +/// +/// Unless `--metrics-address` is empty, the gRPC server metrics the Go SDK +/// serves (`grpc_server_started_total` and friends, same names and labels) +/// are served there at /metrics - OpenMetrics 1.0 as the main format, the +/// classic Prometheus text format for scrapers that ask for it. pub async fn serve(function: F, args: &Args) -> Result<(), Error> { let address: SocketAddr = args.address.parse()?; @@ -73,9 +84,25 @@ pub async fn serve(function: F, args: &Args) -> Result .set_serving::>() .await; + // The metrics endpoint runs beside the gRPC server, like the Go SDK's; + // a failure to serve it is logged, never the function's failure. The + // counting layer below is always installed - with the endpoint disabled + // the counts are simply never served, which is indistinguishable from + // the Go SDK installing no interceptor. + if !args.metrics_address.is_empty() { + crate::metrics::initialize(); + let metrics_address = args.metrics_address.clone(); + tokio::spawn(async move { + if let Err(e) = crate::metrics::serve(&metrics_address).await { + tracing::error!(error = %e, "cannot serve metrics"); + } + }); + } + tracing::info!(%address, insecure = args.insecure, "serving FunctionRunnerService"); builder + .layer(crate::metrics::MetricsLayer) .add_service(function_service) .add_service(health_service) .add_service(reflection_v1) diff --git a/sdk/tests/serve.rs b/sdk/tests/serve.rs index 45df9e1..8161b1b 100644 --- a/sdk/tests/serve.rs +++ b/sdk/tests/serve.rs @@ -22,6 +22,7 @@ impl FunctionRunnerService for EchoFunction { #[tokio::test] async fn serve_handles_a_run_function_request() { let port = free_port(); + let metrics_port = free_port(); tokio::spawn(async move { let args = Args { debug: false, @@ -29,6 +30,7 @@ async fn serve_handles_a_run_function_request() { tls_certs_dir: None, insecure: true, max_recv_message_size: None, + metrics_address: format!("127.0.0.1:{metrics_port}"), }; serve(EchoFunction, &args).await.expect("serve must start"); }); @@ -67,6 +69,89 @@ async fn serve_handles_a_run_function_request() { status.status, tonic_health::pb::health_check_response::ServingStatus::Serving as i32 ); + + // The Go SDK's gRPC server series, served as OpenMetrics: one + // RunFunction started, received, handled OK and answered. The handled + // count lands when the response's trailers go out, which can trail the + // client seeing the response, so it is polled briefly. + const RUN_FUNCTION: &str = "grpc_type=\"unary\",grpc_service=\"apiextensions.fn.proto.v1.FunctionRunnerService\",grpc_method=\"RunFunction\""; + let mut om = String::new(); + for _ in 0..50 { + om = http_get(metrics_port, None).await; + if om.contains(&format!( + "grpc_server_handled_total{{{RUN_FUNCTION},grpc_code=\"OK\"}} 1" + )) { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + assert!( + om.contains("application/openmetrics-text; version=1.0.0; charset=utf-8"), + "OpenMetrics is the main format: {om}" + ); + assert!(om.trim_end().ends_with("# EOF"), "OpenMetrics body: {om}"); + assert!( + om.contains(&format!("grpc_server_started_total{{{RUN_FUNCTION}}} 1")), + "started: {om}" + ); + assert!( + om.contains(&format!( + "grpc_server_msg_received_total{{{RUN_FUNCTION}}} 1" + )), + "msg_received: {om}" + ); + assert!( + om.contains(&format!( + "grpc_server_handled_total{{{RUN_FUNCTION},grpc_code=\"OK\"}} 1" + )), + "handled OK: {om}" + ); + assert!( + om.contains(&format!("grpc_server_msg_sent_total{{{RUN_FUNCTION}}} 1")), + "msg_sent: {om}" + ); + // The health Check above was counted too - the layer covers the whole + // router, like the Go SDK's server-wide interceptor. + assert!( + om.contains("grpc_server_started_total{grpc_type=\"unary\",grpc_service=\"grpc.health.v1.Health\",grpc_method=\"Check\"} 1"), + "health Check counted: {om}" + ); + // InitializeMetrics parity: streaming methods exist as zero series and + // are never incremented (the Go interceptor was unary-only too). + assert!( + om.contains("grpc_server_started_total{grpc_type=\"server_stream\",grpc_service=\"grpc.health.v1.Health\",grpc_method=\"Watch\"} 0"), + "streaming methods stay zero: {om}" + ); + + // A scraper that asks for the classic Prometheus text format without + // accepting OpenMetrics gets it. + let classic = http_get(metrics_port, Some("text/plain; version=0.0.4")).await; + assert!( + classic.contains("text/plain; version=0.0.4"), + "classic on request: {classic}" + ); + assert!(classic.contains("# TYPE grpc_server_started_total counter")); + assert!(!classic.contains("# EOF")); +} + +async fn http_get(port: u16, accept: Option<&str>) -> String { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + for _ in 0..50 { + let Ok(mut conn) = tokio::net::TcpStream::connect(format!("127.0.0.1:{port}")).await else { + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + continue; + }; + let header = accept + .map(|a| format!("Accept: {a}\r\n")) + .unwrap_or_default(); + conn.write_all(format!("GET /metrics HTTP/1.1\r\nHost: x\r\n{header}\r\n").as_bytes()) + .await + .expect("write request"); + let mut out = Vec::new(); + let _ = conn.read_to_end(&mut out).await; + return String::from_utf8_lossy(&out).into_owned(); + } + panic!("cannot connect to the metrics endpoint"); } fn free_port() -> u16 {