diff --git a/README.md b/README.md index ead7b0c..6e95c23 100644 --- a/README.md +++ b/README.md @@ -97,8 +97,12 @@ with a few deliberate differences: survive the protobuf Struct round-trip, so integer fields deserialize cleanly. - **Health is always on.** function-sdk-go's health service is opt-in - (`WithHealthServer`); this SDK always serves the gRPC health API and - reports the function as serving. + (`WithHealthServer`) and the function sets its status; this SDK always + serves the gRPC health API and reports the function as serving. A + function that must load something at startup passes a future to + `Server::ready`: until it completes, the server reports NOT_SERVING and + answers RunFunction with UNAVAILABLE, and if it fails, the server stops + with `Error::NotReady`. - **Graceful shutdown is built in.** The server drains in-flight requests on SIGTERM and SIGINT. - **Metrics are Go-compatible, served as OpenMetrics first.** Like @@ -117,7 +121,11 @@ with a few deliberate differences: [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. + v1beta1 series the Go SDK pre-created do not exist here. Like the Go + SDK's `WithMetricsRegistry`, `Server::metrics_registry` serves a + function's own series beside the gRPC ones: the function registers them + in a prometheus-client `Registry` built with a version semver-compatible + with the SDK's (the same minor version while prometheus-client is 0.x). [crossplane]: https://www.crossplane.io [functions]: https://docs.crossplane.io/latest/composition/compositions/ diff --git a/sdk/src/lib.rs b/sdk/src/lib.rs index d4758f1..ce10b6e 100644 --- a/sdk/src/lib.rs +++ b/sdk/src/lib.rs @@ -24,7 +24,7 @@ pub mod resource; pub mod response; pub mod server; -pub use server::{Args, serve}; +pub use server::{Args, Server, serve}; /// Errors returned by the SDK. #[derive(Debug, thiserror::Error)] @@ -67,4 +67,8 @@ pub enum Error { /// The gRPC reflection service could not be built. #[error("cannot build gRPC reflection service: {0}")] Reflection(#[from] tonic_reflection::server::Error), + + /// The readiness future set with [`Server::ready`] failed. + #[error("the function did not become ready: {0}")] + NotReady(Box), } diff --git a/sdk/src/metrics.rs b/sdk/src/metrics.rs index eb8d66f..3fd612d 100644 --- a/sdk/src/metrics.rs +++ b/sdk/src/metrics.rs @@ -16,7 +16,7 @@ //! OpenMetrics gets that instead, derived from the same rendering. use std::pin::Pin; -use std::sync::{LazyLock, RwLock}; +use std::sync::{Arc, LazyLock}; use std::task::{Context, Poll}; use prometheus_client::metrics::counter::Counter; @@ -27,10 +27,6 @@ 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 @@ -38,19 +34,25 @@ type Labels = Vec<(String, String)>; /// trailing period - the OpenMetrics encoder appends both - so the served /// text is byte-identical to the Go SDK's. struct LabeledCounter { + name: &'static str, + help: &'static str, 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 new(name: &'static str, help: &'static str, keys: &'static [&'static str]) -> Self { + LabeledCounter { + name, + help, + keys, + family: Family::default(), + } + } + + fn register(&self, registry: &mut Registry) { + let help = self.help.strip_suffix('.').unwrap_or(self.help); + registry.register(self.name, help, self.family.clone()); } fn with_label_values(&self, values: &[&str]) -> impl std::ops::Deref { @@ -136,6 +138,18 @@ fn unary_method(path: &str) -> Option<(&'static str, &'static str)> { } } +/// The registry the /metrics endpoint serves: the function's own, when it +/// set one with [`crate::Server::metrics_registry`], else an empty one, with +/// the gRPC server series added at its root. The series stay process-wide +/// statics so the counting layer needs no registry. +pub(crate) fn served_registry(registry: Option) -> Registry { + let mut registry = registry.unwrap_or_default(); + for counter in [&*STARTED, &*HANDLED, &*MSG_RECEIVED, &*MSG_SENT] { + counter.register(&mut registry); + } + registry +} + /// 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, @@ -344,10 +358,9 @@ const OPENMETRICS_CONTENT_TYPE: &str = "application/openmetrics-text; version=1. /// The registry rendered as OpenMetrics 1.0 text, straight from /// prometheus-client's encoder. -fn render_openmetrics() -> String { +fn render_openmetrics(registry: &Registry) -> String { let mut out = String::new(); - let _ = - prometheus_client::encoding::text::encode(&mut out, ®ISTRY.read().expect("poisoned")); + let _ = prometheus_client::encoding::text::encode(&mut out, registry); out } @@ -401,9 +414,10 @@ fn wants_classic_text(request_head: &str) -> bool { !accept.contains("application/openmetrics-text") && accept.contains("text/plain") } -/// Serves the metrics at /metrics on address until the process ends - the +/// Serves the registry 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> { +pub(crate) async fn serve(address: &str, registry: Registry) -> Result<(), std::io::Error> { + let registry = Arc::new(registry); // The Go SDK's ":8080" shorthand means every interface. let address = if address.starts_with(':') { format!("0.0.0.0{address}") @@ -416,13 +430,14 @@ pub(crate) async fn serve(address: &str) -> Result<(), std::io::Error> { let Ok((mut conn, _)) = listener.accept().await else { continue; }; + let registry = registry.clone(); 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 om = render_openmetrics(®istry); let (body, content_type) = if wants_classic_text(&head) { (classic_from_openmetrics(&om), "text/plain; version=0.0.4") } else { @@ -490,7 +505,7 @@ mod tests { #[test] fn renderings_differ_only_where_the_formats_do() { initialize(); - let om = render_openmetrics(); + let om = render_openmetrics(&served_registry(None)); assert!(om.ends_with("# EOF\n"), "OpenMetrics ends with EOF:\n{om}"); assert!( om.contains("# TYPE grpc_server_started counter"), @@ -511,8 +526,9 @@ mod tests { #[test] fn a_dropped_reporter_is_a_canceled_call() { initialize(); + let registry = served_registry(None); let canceled = || { - render_openmetrics() + render_openmetrics(®istry) .lines() .find(|l| { l.starts_with("grpc_server_handled_total{grpc_type=\"unary\"") diff --git a/sdk/src/server.rs b/sdk/src/server.rs index e0e72c8..244d563 100644 --- a/sdk/src/server.rs +++ b/sdk/src/server.rs @@ -1,14 +1,23 @@ //! A spec-compliant gRPC server runtime for composition functions. +use std::future::Future; use std::net::SocketAddr; use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; -use tonic::transport::{Certificate, Identity, Server, ServerTlsConfig}; +use prometheus_client::registry::Registry; +use tonic::transport::{Certificate, Identity, Server as GrpcServer, ServerTlsConfig}; +use tonic::{Request, Response, Status}; +use tonic_health::ServingStatus; +use tonic_health::server::HealthReporter; use crate::Error; use crate::proto::v1::function_runner_service_server::{ - FunctionRunnerService, FunctionRunnerServiceServer, + FunctionRunnerService, FunctionRunnerServiceServer, SERVICE_NAME, }; +use crate::proto::v1::{RunFunctionRequest, RunFunctionResponse}; /// CLI arguments required by the Crossplane composition function spec. #[derive(clap::Parser, Debug)] @@ -55,62 +64,209 @@ pub struct Args { /// 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. +/// +/// For options beyond the command-line `args`, use [`Server`]. pub async fn serve(function: F, args: &Args) -> Result<(), Error> { - let address: SocketAddr = args.address.parse()?; - - let mut builder = Server::builder(); - if !args.insecure { - let dir = args - .tls_certs_dir - .as_deref() - .ok_or(Error::MissingTlsCertsDir)?; - builder = builder.tls_config(tls_config(dir)?)?; + Server::new(function, args).serve().await +} + +type BoxError = Box; +type ReadyFuture = Pin> + Send>>; + +/// A function server with options beyond the command-line [`Args`]. Create +/// it with [`Server::new`], chain the options, then call [`Server::serve`]; +/// with no options it serves exactly like [`serve`]. New options arrive as +/// new methods, so code that builds a server keeps compiling. +/// +/// ```no_run +/// # use function_sdk_rust::proto::v1::function_runner_service_server::FunctionRunnerService; +/// # async fn run(function: impl FunctionRunnerService, args: &function_sdk_rust::Args) -> Result<(), function_sdk_rust::Error> { +/// function_sdk_rust::Server::new(function, args) +/// .ready(async { Ok::<(), std::io::Error>(()) }) +/// .serve() +/// .await +/// # } +/// ``` +pub struct Server<'a, F> { + function: F, + args: &'a Args, + ready: Option, + metrics_registry: Option, +} + +impl<'a, F: FunctionRunnerService> Server<'a, F> { + /// Creates a server for `function`, configured by the command-line `args`. + pub fn new(function: F, args: &'a Args) -> Self { + Self { + function, + args, + ready: None, + metrics_registry: None, + } } - let mut function_service = FunctionRunnerServiceServer::new(function); - if let Some(size) = args.max_recv_message_size { - function_service = function_service.max_decoding_message_size(size); + /// Holds the function back until `ready` completes, for a function that + /// must finish loading before it can run. The server listens at once, + /// but reports NOT_SERVING on the gRPC health service and answers + /// RunFunction with UNAVAILABLE. Ok flips the function to SERVING; an + /// error or a panic stops the server, and [`Server::serve`] returns it + /// as [`Error::NotReady`]. + pub fn ready(mut self, ready: R) -> Self + where + R: Future> + Send + 'static, + E: Into, + { + self.ready = Some(Box::pin(async move { ready.await.map_err(Into::into) })); + self } - let reflection_v1 = tonic_reflection::server::Builder::configure() - .register_encoded_file_descriptor_set(crate::proto::FILE_DESCRIPTOR_SET) - .build_v1()?; - let reflection_v1alpha = tonic_reflection::server::Builder::configure() - .register_encoded_file_descriptor_set(crate::proto::FILE_DESCRIPTOR_SET) - .build_v1alpha()?; - - let (health_reporter, health_service) = tonic_health::server::health_reporter(); - health_reporter - .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"); - } + /// Serves `registry` at /metrics, like function-sdk-go's + /// `WithMetricsRegistry`: register the function's own series in it + /// first, and the SDK adds its `grpc_server_*` series at the root. With + /// `--metrics-address` empty nothing is served. + /// + /// Build the registry and its metrics with a `prometheus-client` version + /// that is semver-compatible with the SDK's own dependency on it (while + /// prometheus-client is 0.x, the same minor version). Cargo then resolves + /// both to one copy of the crate; a registry from an incompatible version + /// is a different type and does not compile here. + pub fn metrics_registry(mut self, registry: Registry) -> Self { + self.metrics_registry = Some(registry); + self + } + + /// Serves RunFunctionRequests until SIGTERM or SIGINT, then shuts down + /// gracefully. See [`serve`] for what is served. + pub async fn serve(self) -> Result<(), Error> { + let Server { + function, + args, + ready, + metrics_registry, + } = self; + let address: SocketAddr = args.address.parse()?; + + let mut builder = GrpcServer::builder(); + if !args.insecure { + let dir = args + .tls_certs_dir + .as_deref() + .ok_or(Error::MissingTlsCertsDir)?; + builder = builder.tls_config(tls_config(dir)?)?; + } + + let is_ready = Arc::new(AtomicBool::new(false)); + let mut function_service = FunctionRunnerServiceServer::new(Gate { + function, + ready: is_ready.clone(), }); + if let Some(size) = args.max_recv_message_size { + function_service = function_service.max_decoding_message_size(size); + } + + let reflection_v1 = tonic_reflection::server::Builder::configure() + .register_encoded_file_descriptor_set(crate::proto::FILE_DESCRIPTOR_SET) + .build_v1()?; + let reflection_v1alpha = tonic_reflection::server::Builder::configure() + .register_encoded_file_descriptor_set(crate::proto::FILE_DESCRIPTOR_SET) + .build_v1alpha()?; + + // The health statuses are set before the server listens, so no + // probe sees SERVING ahead of the readiness future. + let (health_reporter, health_service) = tonic_health::server::health_reporter(); + let (not_ready_tx, not_ready_rx) = tokio::sync::oneshot::channel::(); + match ready { + None => { + is_ready.store(true, Ordering::Release); + set_health(&health_reporter, ServingStatus::Serving).await; + } + Some(ready) => { + set_health(&health_reporter, ServingStatus::NotServing).await; + // The future runs in a task of its own so that a panic in it + // fails readiness instead of leaving the server NOT_SERVING. + let loading = tokio::spawn(ready); + tokio::spawn(async move { + match loading.await.unwrap_or_else(|e| Err(e.into())) { + Ok(()) => { + is_ready.store(true, Ordering::Release); + set_health(&health_reporter, ServingStatus::Serving).await; + tracing::info!("function is ready"); + } + Err(e) => { + let _ = not_ready_tx.send(e); + } + } + }); + } + } + + // 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(); + let registry = crate::metrics::served_registry(metrics_registry); + tokio::spawn(async move { + if let Err(e) = crate::metrics::serve(&metrics_address, registry).await { + tracing::error!(error = %e, "cannot serve metrics"); + } + }); + } + + tracing::info!(%address, insecure = args.insecure, "serving FunctionRunnerService"); + + // A readiness failure shuts the server down like a signal does. + // Without a readiness future, or once it succeeds, the receive never + // yields Ok, and select! waits for the signal alone. + let mut not_ready = None; + builder + .layer(crate::metrics::MetricsLayer) + .add_service(function_service) + .add_service(health_service) + .add_service(reflection_v1) + .add_service(reflection_v1alpha) + .serve_with_shutdown(address, async { + tokio::select! { + () = shutdown_signal() => {} + Ok(e) = not_ready_rx => not_ready = Some(e), + } + }) + .await?; + + match not_ready { + Some(e) => Err(Error::NotReady(e)), + None => Ok(()), + } } +} - tracing::info!(%address, insecure = args.insecure, "serving FunctionRunnerService"); +/// Sets the overall ("") and the FunctionRunnerService health status. +async fn set_health(reporter: &HealthReporter, status: ServingStatus) { + for service in ["", SERVICE_NAME] { + reporter.set_service_status(service, status).await; + } +} - builder - .layer(crate::metrics::MetricsLayer) - .add_service(function_service) - .add_service(health_service) - .add_service(reflection_v1) - .add_service(reflection_v1alpha) - .serve_with_shutdown(address, shutdown_signal()) - .await?; +/// Answers RunFunction with UNAVAILABLE until the function is ready. +struct Gate { + function: F, + ready: Arc, +} - Ok(()) +#[tonic::async_trait] +impl FunctionRunnerService for Gate { + async fn run_function( + &self, + request: Request, + ) -> Result, Status> { + if !self.ready.load(Ordering::Acquire) { + return Err(Status::unavailable("the function is not ready")); + } + self.function.run_function(request).await + } } fn tls_config(dir: &Path) -> Result { diff --git a/sdk/tests/common/mod.rs b/sdk/tests/common/mod.rs new file mode 100644 index 0000000..9a25d2e --- /dev/null +++ b/sdk/tests/common/mod.rs @@ -0,0 +1,48 @@ +//! Helpers shared by the server test binaries. + +// Each test binary compiles this module and uses only some of it. +#![allow(dead_code)] + +use function_sdk_rust::proto::v1::function_runner_service_client::FunctionRunnerServiceClient; + +pub 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"); +} + +pub fn free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .expect("bind to an ephemeral port") + .local_addr() + .expect("read local addr") + .port() +} + +pub async fn connect_with_retry( + port: u16, +) -> FunctionRunnerServiceClient { + for _ in 0..50 { + if let Ok(client) = + FunctionRunnerServiceClient::connect(format!("http://127.0.0.1:{port}")).await + { + return client; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + panic!("cannot connect to the function under test"); +} diff --git a/sdk/tests/serve.rs b/sdk/tests/serve.rs index 8161b1b..0eb68cf 100644 --- a/sdk/tests/serve.rs +++ b/sdk/tests/serve.rs @@ -1,4 +1,6 @@ -use function_sdk_rust::proto::v1::function_runner_service_client::FunctionRunnerServiceClient; +mod common; + +use common::{connect_with_retry, free_port, http_get}; use function_sdk_rust::proto::v1::function_runner_service_server::FunctionRunnerService; use function_sdk_rust::proto::v1::{RequestMeta, RunFunctionRequest, RunFunctionResponse}; use function_sdk_rust::{Args, response, serve}; @@ -133,43 +135,3 @@ async fn serve_handles_a_run_function_request() { 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 { - std::net::TcpListener::bind("127.0.0.1:0") - .expect("bind to an ephemeral port") - .local_addr() - .expect("read local addr") - .port() -} - -async fn connect_with_retry(port: u16) -> FunctionRunnerServiceClient { - for _ in 0..50 { - if let Ok(client) = - FunctionRunnerServiceClient::connect(format!("http://127.0.0.1:{port}")).await - { - return client; - } - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - } - panic!("cannot connect to the function under test"); -} diff --git a/sdk/tests/server_builder.rs b/sdk/tests/server_builder.rs new file mode 100644 index 0000000..b9d60b9 --- /dev/null +++ b/sdk/tests/server_builder.rs @@ -0,0 +1,197 @@ +//! The `Server` builder's options. A separate test binary from serve.rs, +//! whose exact gRPC metric counts would see these tests' calls. + +mod common; + +use std::sync::{Arc, OnceLock}; + +use common::{connect_with_retry, free_port, http_get}; +use function_sdk_rust::proto::v1::function_runner_service_server::FunctionRunnerService; +use function_sdk_rust::proto::v1::{RunFunctionRequest, RunFunctionResponse}; +use function_sdk_rust::{Args, Error, Server, response}; +use prometheus_client::metrics::{counter::Counter, family::Family}; +use prometheus_client::registry::Registry; +use tonic::{Code, Request, Response, Status}; +use tonic_health::pb::HealthCheckRequest; +use tonic_health::pb::health_check_response::ServingStatus; +use tonic_health::pb::health_client::HealthClient; + +const SERVICES: [&str; 2] = ["", "apiextensions.fn.proto.v1.FunctionRunnerService"]; + +/// Answers with the bundle its readiness future loaded. +#[derive(Default)] +struct BundleFunction { + bundle: Arc>, +} + +#[tonic::async_trait] +impl FunctionRunnerService for BundleFunction { + async fn run_function( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let mut rsp = response::to(&req, response::DEFAULT_TTL); + let bundle = self + .bundle + .get() + .expect("Server::ready gates on the bundle"); + response::normal(&mut rsp, bundle.clone()); + Ok(Response::new(rsp)) + } +} + +fn insecure_args(port: u16) -> Args { + Args { + debug: false, + address: format!("127.0.0.1:{port}"), + tls_certs_dir: None, + insecure: true, + max_recv_message_size: None, + metrics_address: String::new(), + } +} + +async fn health_status(health: &mut HealthClient, service: &str) -> i32 { + health + .check(HealthCheckRequest { + service: service.to_string(), + }) + .await + .expect("health check must succeed") + .into_inner() + .status +} + +#[tokio::test] +async fn ready_holds_the_function_back_until_it_completes() { + let port = free_port(); + let (loaded_tx, loaded_rx) = tokio::sync::oneshot::channel::<()>(); + let function = BundleFunction::default(); + let bundle = function.bundle.clone(); + tokio::spawn(async move { + let ready = async move { + loaded_rx.await.expect("the test sends"); + bundle.set("v1".to_string()).expect("set once"); + Ok::<(), std::io::Error>(()) + }; + Server::new(function, &insecure_args(port)) + .ready(ready) + .serve() + .await + .expect("serve must start"); + }); + + let mut client = connect_with_retry(port).await; + let channel = tonic::transport::Channel::from_shared(format!("http://127.0.0.1:{port}")) + .expect("valid URI") + .connect() + .await + .expect("health channel must connect"); + let mut health = HealthClient::new(channel); + + for service in SERVICES { + assert_eq!( + health_status(&mut health, service).await, + ServingStatus::NotServing as i32, + "{service:?} before ready" + ); + } + let err = client + .run_function(RunFunctionRequest::default()) + .await + .expect_err("RunFunction must wait for ready"); + assert_eq!(err.code(), Code::Unavailable); + + loaded_tx.send(()).expect("the readiness future waits"); + for service in SERVICES { + let mut status = 0; + for _ in 0..50 { + status = health_status(&mut health, service).await; + if status == ServingStatus::Serving as i32 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + assert_eq!( + status, + ServingStatus::Serving as i32, + "{service:?} after ready" + ); + } + let rsp = client + .run_function(RunFunctionRequest::default()) + .await + .expect("RunFunction must succeed once ready") + .into_inner(); + assert_eq!(rsp.results[0].message, "v1"); +} + +#[tokio::test] +async fn a_failed_ready_stops_the_server() { + let ready = async { Err::<(), _>(std::io::Error::other("cannot pull the bundle")) }; + let err = Server::new(BundleFunction::default(), &insecure_args(free_port())) + .ready(ready) + .serve() + .await + .expect_err("serve must return the readiness failure"); + assert!(matches!(err, Error::NotReady(_)), "{err:?}"); + assert_eq!( + err.to_string(), + "the function did not become ready: cannot pull the bundle" + ); +} + +#[tokio::test] +async fn a_panicking_ready_stops_the_server() { + let ready = async { panic!("cannot parse the bundle") }; + let err = Server::new(BundleFunction::default(), &insecure_args(free_port())) + .ready::<_, std::io::Error>(ready) + .serve() + .await + .expect_err("serve must return the readiness panic"); + assert!(matches!(err, Error::NotReady(_)), "{err:?}"); + assert!( + err.to_string().contains("cannot parse the bundle"), + "the panic message: {err}" + ); +} + +#[tokio::test] +async fn metrics_registry_serves_the_function_series_beside_the_grpc_ones() { + let evaluations = Family::, Counter>::default(); + let mut registry = Registry::default(); + registry.sub_registry_with_prefix("my_function").register( + "evaluations", + "Policy evaluations by verdict", + evaluations.clone(), + ); + evaluations + .get_or_create(&vec![("verdict".to_string(), "deny".to_string())]) + .inc(); + + let metrics_port = free_port(); + let mut args = insecure_args(free_port()); + args.metrics_address = format!("127.0.0.1:{metrics_port}"); + tokio::spawn(async move { + Server::new(BundleFunction::default(), &args) + .metrics_registry(registry) + .serve() + .await + .expect("serve must start"); + }); + + let om = http_get(metrics_port, None).await; + assert!( + om.contains("my_function_evaluations_total{verdict=\"deny\"} 1"), + "the function's series: {om}" + ); + assert!( + om.contains("grpc_server_started_total{grpc_type=\"unary\",grpc_service=\"apiextensions.fn.proto.v1.FunctionRunnerService\",grpc_method=\"RunFunction\"}"), + "the SDK's series: {om}" + ); + assert!( + om.trim_end().ends_with("# EOF"), + "one OpenMetrics body: {om}" + ); +}