function_sdk_rust/
server.rs

1//! A spec-compliant gRPC server runtime for composition functions.
2
3use std::net::SocketAddr;
4use std::path::{Path, PathBuf};
5
6use tonic::transport::{Certificate, Identity, Server, ServerTlsConfig};
7
8use crate::Error;
9use crate::proto::v1::function_runner_service_server::{
10    FunctionRunnerService, FunctionRunnerServiceServer,
11};
12
13/// CLI arguments required by the Crossplane composition function spec.
14#[derive(clap::Parser, Debug)]
15#[command(version, about = "A Crossplane composition function")]
16pub struct Args {
17    /// Emit debug logs.
18    #[arg(short, long, env = "DEBUG", default_value_t = false)]
19    pub debug: bool,
20
21    /// Address at which to listen for gRPC connections.
22    #[arg(long, default_value = "0.0.0.0:9443")]
23    pub address: String,
24
25    /// Directory containing tls.crt, tls.key, and ca.crt; serve using mTLS.
26    #[arg(long, env = "TLS_SERVER_CERTS_DIR")]
27    pub tls_certs_dir: Option<PathBuf>,
28
29    /// Run without mTLS credentials. If set, --tls-certs-dir is ignored.
30    #[arg(long, default_value_t = false)]
31    pub insecure: bool,
32
33    /// Maximum size in bytes of gRPC messages the function accepts.
34    /// Defaults to the gRPC default of 4MB.
35    #[arg(long)]
36    pub max_recv_message_size: Option<usize>,
37}
38
39/// Starts a gRPC server and serves RunFunctionRequests until SIGTERM or
40/// SIGINT, then shuts down gracefully.
41///
42/// Serves with mTLS from `--tls-certs-dir` (tls.crt and tls.key must be the
43/// function's PEM-encoded certificate and key; ca.crt a PEM-encoded CA used
44/// to authenticate Crossplane) unless `--insecure` is set. gRPC server
45/// reflection is enabled for both the v1 and v1alpha reflection APIs, and
46/// the gRPC health service reports the function as serving.
47pub async fn serve<F: FunctionRunnerService>(function: F, args: &Args) -> Result<(), Error> {
48    let address: SocketAddr = args.address.parse()?;
49
50    let mut builder = Server::builder();
51    if !args.insecure {
52        let dir = args
53            .tls_certs_dir
54            .as_deref()
55            .ok_or(Error::MissingTlsCertsDir)?;
56        builder = builder.tls_config(tls_config(dir)?)?;
57    }
58
59    let mut function_service = FunctionRunnerServiceServer::new(function);
60    if let Some(size) = args.max_recv_message_size {
61        function_service = function_service.max_decoding_message_size(size);
62    }
63
64    let reflection_v1 = tonic_reflection::server::Builder::configure()
65        .register_encoded_file_descriptor_set(crate::proto::FILE_DESCRIPTOR_SET)
66        .build_v1()?;
67    let reflection_v1alpha = tonic_reflection::server::Builder::configure()
68        .register_encoded_file_descriptor_set(crate::proto::FILE_DESCRIPTOR_SET)
69        .build_v1alpha()?;
70
71    let (health_reporter, health_service) = tonic_health::server::health_reporter();
72    health_reporter
73        .set_serving::<FunctionRunnerServiceServer<F>>()
74        .await;
75
76    tracing::info!(%address, insecure = args.insecure, "serving FunctionRunnerService");
77
78    builder
79        .add_service(function_service)
80        .add_service(health_service)
81        .add_service(reflection_v1)
82        .add_service(reflection_v1alpha)
83        .serve_with_shutdown(address, shutdown_signal())
84        .await?;
85
86    Ok(())
87}
88
89fn tls_config(dir: &Path) -> Result<ServerTlsConfig, Error> {
90    let read = |name: &str| {
91        let path = dir.join(name);
92        std::fs::read(&path).map_err(|source| Error::ReadCertificate { path, source })
93    };
94    let cert = read("tls.crt")?;
95    let key = read("tls.key")?;
96    let ca = read("ca.crt")?;
97
98    Ok(ServerTlsConfig::new()
99        .identity(Identity::from_pem(cert, key))
100        .client_ca_root(Certificate::from_pem(ca))
101        .client_auth_optional(false))
102}
103
104#[cfg(unix)]
105async fn shutdown_signal() {
106    let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
107        .expect("cannot install SIGTERM handler");
108    tokio::select! {
109        _ = sigterm.recv() => {}
110        _ = tokio::signal::ctrl_c() => {}
111    }
112    tracing::info!("shutting down");
113}
114
115#[cfg(not(unix))]
116async fn shutdown_signal() {
117    let _ = tokio::signal::ctrl_c().await;
118    tracing::info!("shutting down");
119}
function_sdk_rust/server.rs

function_sdk_rust/
server.rs
