src/talos_apid.rs
//! Talos apid gRPC push (feature `talos`) — the async half of the bare-metal
//! rebirth target (B5). The sync machineconfig **generator** lives in
//! [`crate::unfold::talos`]; this module owns the **apply**: a real gRPC client
//! for Talos's `machine.MachineService`, driving
//! `ApplyConfiguration` (push the rendered machineconfig) and `Bootstrap`
//! (init etcd on the first control-plane node) over the apid endpoint.
//!
//! 100% Rust, no `talosctl`: the Talos machine API is protobuf/gRPC, so we speak
//! it directly with [`tonic`] + hand-written [`prost`] messages (no `protoc` /
//! `tonic-build` in the build — the messages are small and stable, so the lean
//! default build never pulls a proto compiler). Only compiled with the `talos`
//! feature on.
//!
//! Transport: apid is **mutual-TLS** on port `50000`. [`ApidClient::connect`]
//! configures the client identity (the admin cert from a `talosconfig`) + the
//! cluster CA via [`TalosTls`]; without TLS material it falls back to a plaintext
//! `http://` channel (used by the proto-level mock-server test, which asserts the
//! exact RPC path + machineconfig bytes on the wire without a live node).
use anyhow::Result;
// ── proto messages (machine.proto, hand-written prost) ───────────────────────
// Only the fields we send/read; prost skips unknown fields on decode, so partial
// response models are safe.
/// `machine.ApplyConfigurationRequest.Mode` — how Talos applies the config.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum ApplyMode {
/// Apply + reboot into the new config.
Reboot = 0,
/// Apply without reboot when possible, else reboot (Talos default).
Auto = 1,
/// Apply, never reboot (fails if a reboot would be required).
NoReboot = 2,
/// Stage for the next boot.
Staged = 3,
/// Apply, then roll back unless confirmed within a timeout.
Try = 4,
}
/// `machine.ApplyConfigurationRequest` — push a machineconfig to apid.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ApplyConfigurationRequest {
/// The raw machineconfig YAML bytes.
#[prost(bytes = "vec", tag = "1")]
pub data: Vec<u8>,
/// One of [`ApplyMode`].
#[prost(enumeration = "ApplyMode", tag = "5")]
pub mode: i32,
#[prost(bool, tag = "6")]
pub dry_run: bool,
}
/// `common.Metadata` — the per-node envelope on every response message; a
/// non-empty `error` means that node rejected the request.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Metadata {
#[prost(string, tag = "1")]
pub hostname: String,
#[prost(string, tag = "2")]
pub error: String,
}
/// One node's `ApplyConfiguration` response entry.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ApplyConfiguration {
#[prost(message, optional, tag = "1")]
pub metadata: Option<Metadata>,
#[prost(string, tag = "3")]
pub mode_details: String,
}
/// `machine.ApplyConfigurationResponse`.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct ApplyConfigurationResponse {
#[prost(message, repeated, tag = "1")]
pub messages: Vec<ApplyConfiguration>,
}
/// `machine.BootstrapRequest` — init etcd on the first control-plane node.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct BootstrapRequest {
#[prost(bool, tag = "1")]
pub recover_etcd: bool,
#[prost(bool, tag = "2")]
pub recover_skip_hash_check: bool,
}
/// One node's `Bootstrap` response entry.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Bootstrap {
#[prost(message, optional, tag = "1")]
pub metadata: Option<Metadata>,
}
/// `machine.BootstrapResponse`.
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct BootstrapResponse {
#[prost(message, repeated, tag = "1")]
pub messages: Vec<Bootstrap>,
}
/// The Talos machine-API service name + method paths (gRPC routes).
const APPLY_PATH: &str = "/machine.MachineService/ApplyConfiguration";
const BOOTSTRAP_PATH: &str = "/machine.MachineService/Bootstrap";
/// mTLS material for the apid channel — the admin client identity (cert + key)
/// from a `talosconfig`, plus the cluster CA to pin the server. `domain`
/// overrides the SNI/verification name when the endpoint is an IP.
#[derive(Debug, Clone)]
pub struct TalosTls {
/// Cluster CA cert (PEM) to verify the apid server against.
pub ca_pem: Option<Vec<u8>>,
/// Admin client cert (PEM).
pub cert_pem: Vec<u8>,
/// Admin client key (PEM).
pub key_pem: Vec<u8>,
/// Override the verified server name (apid certs are SAN'd for the node).
pub domain: Option<String>,
}
/// A connected gRPC client for Talos's `machine.MachineService` (apid).
pub struct ApidClient {
inner: tonic::client::Grpc<tonic::transport::Channel>,
}
impl ApidClient {
/// Connect to apid at `endpoint` (`host:port`, or a full `http(s)://` URL).
/// With [`TalosTls`] present the channel is mutual-TLS (the real apid path);
/// without it, a plaintext `http://` channel (the mock-server test path).
pub async fn connect(endpoint: &str, tls: Option<TalosTls>) -> Result<Self> {
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
let uri = if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
endpoint.to_string()
} else if tls.is_some() {
format!("https://{endpoint}")
} else {
format!("http://{endpoint}")
};
let mut ep = Channel::from_shared(uri.clone())
.map_err(|e| anyhow::anyhow!("invalid apid endpoint `{uri}`: {e}"))?;
if let Some(tls) = tls {
let mut cfg =
ClientTlsConfig::new().identity(Identity::from_pem(&tls.cert_pem, &tls.key_pem));
if let Some(ca) = &tls.ca_pem {
cfg = cfg.ca_certificate(Certificate::from_pem(ca));
}
if let Some(domain) = &tls.domain {
cfg = cfg.domain_name(domain.clone());
}
ep = ep
.tls_config(cfg)
.map_err(|e| anyhow::anyhow!("apid TLS config: {e}"))?;
}
let channel = ep
.connect()
.await
.map_err(|e| anyhow::anyhow!("connect apid `{uri}`: {e}"))?;
Ok(ApidClient {
inner: tonic::client::Grpc::new(channel),
})
}
/// Push the rendered machineconfig via `ApplyConfiguration`.
pub async fn apply_configuration(
&mut self,
machineconfig: Vec<u8>,
mode: ApplyMode,
) -> Result<()> {
self.inner
.ready()
.await
.map_err(|e| anyhow::anyhow!("apid not ready: {e}"))?;
let req = tonic::Request::new(ApplyConfigurationRequest {
data: machineconfig,
mode: mode as i32,
dry_run: false,
});
let path = tonic::codegen::http::uri::PathAndQuery::from_static(APPLY_PATH);
let codec = tonic_prost::ProstCodec::default();
let resp: tonic::Response<ApplyConfigurationResponse> = self
.inner
.unary(req, path, codec)
.await
.map_err(|s| anyhow::anyhow!("apid ApplyConfiguration failed: {s}"))?;
for m in &resp.into_inner().messages {
if let Some(meta) = &m.metadata {
if !meta.error.is_empty() {
anyhow::bail!("apid ApplyConfiguration rejected by node: {}", meta.error);
}
}
}
Ok(())
}
/// Bootstrap etcd on the first control-plane node via `Bootstrap`.
pub async fn bootstrap(&mut self) -> Result<()> {
self.inner
.ready()
.await
.map_err(|e| anyhow::anyhow!("apid not ready: {e}"))?;
let req = tonic::Request::new(BootstrapRequest::default());
let path = tonic::codegen::http::uri::PathAndQuery::from_static(BOOTSTRAP_PATH);
let codec = tonic_prost::ProstCodec::default();
let resp: tonic::Response<BootstrapResponse> = self
.inner
.unary(req, path, codec)
.await
.map_err(|s| anyhow::anyhow!("apid Bootstrap failed: {s}"))?;
for m in &resp.into_inner().messages {
if let Some(meta) = &m.metadata {
if !meta.error.is_empty() {
anyhow::bail!("apid Bootstrap rejected by node: {}", meta.error);
}
}
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Mutex};
/// What the mock apid captured — the bytes + mode on the wire.
#[derive(Default)]
struct Captured {
applied: Vec<(Vec<u8>, i32)>,
bootstrapped: usize,
}
/// A hand-written `tower::Service` mock of `machine.MachineService` — routes
/// the two RPC paths through `tonic::server::Grpc` (real gRPC framing) and
/// records the decoded requests, so the test asserts the exact machineconfig
/// bytes + `ApplyMode` reached apid, no live Talos node needed.
#[derive(Clone)]
struct MockApid {
cap: Arc<Mutex<Captured>>,
}
impl tonic::server::NamedService for MockApid {
const NAME: &'static str = "machine.MachineService";
}
impl tonic::codegen::Service<tonic::codegen::http::Request<tonic::body::Body>> for MockApid {
type Response = tonic::codegen::http::Response<tonic::body::Body>;
type Error = std::convert::Infallible;
type Future = tonic::codegen::BoxFuture<Self::Response, Self::Error>;
fn poll_ready(
&mut self,
_cx: &mut tonic::codegen::Context<'_>,
) -> tonic::codegen::Poll<Result<(), Self::Error>> {
tonic::codegen::Poll::Ready(Ok(()))
}
fn call(&mut self, req: tonic::codegen::http::Request<tonic::body::Body>) -> Self::Future {
let cap = self.cap.clone();
match req.uri().path() {
APPLY_PATH => Box::pin(async move {
struct Svc(Arc<Mutex<Captured>>);
impl tonic::server::UnaryService<ApplyConfigurationRequest> for Svc {
type Response = ApplyConfigurationResponse;
type Future = tonic::codegen::BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<ApplyConfigurationRequest>,
) -> Self::Future {
let cap = self.0.clone();
Box::pin(async move {
let msg = request.into_inner();
cap.lock().unwrap().applied.push((msg.data, msg.mode));
Ok(tonic::Response::new(ApplyConfigurationResponse {
messages: vec![ApplyConfiguration {
metadata: Some(Metadata {
hostname: "talos-cp1".into(),
error: String::new(),
}),
mode_details: "applied".into(),
}],
}))
})
}
}
let mut grpc = tonic::server::Grpc::new(tonic_prost::ProstCodec::default());
Ok(grpc.unary(Svc(cap), req).await)
}),
BOOTSTRAP_PATH => Box::pin(async move {
struct Svc(Arc<Mutex<Captured>>);
impl tonic::server::UnaryService<BootstrapRequest> for Svc {
type Response = BootstrapResponse;
type Future = tonic::codegen::BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
_request: tonic::Request<BootstrapRequest>,
) -> Self::Future {
let cap = self.0.clone();
Box::pin(async move {
cap.lock().unwrap().bootstrapped += 1;
Ok(tonic::Response::new(BootstrapResponse {
messages: vec![Bootstrap {
metadata: Some(Metadata {
hostname: "talos-cp1".into(),
error: String::new(),
}),
}],
}))
})
}
}
let mut grpc = tonic::server::Grpc::new(tonic_prost::ProstCodec::default());
Ok(grpc.unary(Svc(cap), req).await)
}),
_ => Box::pin(async move {
Ok(tonic::codegen::http::Response::builder()
.status(200)
.header("grpc-status", "12") // UNIMPLEMENTED
.body(tonic::body::Body::empty())
.unwrap())
}),
}
}
}
/// LAW 1 (inject-assert): a control-plane apply drives BOTH RPCs over real
/// gRPC framing — `ApplyConfiguration` with the exact machineconfig bytes +
/// `ApplyMode::Auto`, then `Bootstrap` — asserted against a mock apid.
#[tokio::test]
async fn apply_and_bootstrap_against_mock_apid() {
let cap = Arc::new(Mutex::new(Captured::default()));
let svc = MockApid { cap: cap.clone() };
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener);
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
let server = tokio::spawn(async move {
tonic::transport::Server::builder()
.add_service(svc)
.serve_with_incoming_shutdown(incoming, async {
let _ = rx.await;
})
.await
.unwrap();
});
let machineconfig = b"version: v1alpha1\nmachine:\n type: controlplane\n".to_vec();
let mut client = ApidClient::connect(&format!("http://{addr}"), None)
.await
.expect("connect mock apid");
client
.apply_configuration(machineconfig.clone(), ApplyMode::Auto)
.await
.expect("apply");
client.bootstrap().await.expect("bootstrap");
let _ = tx.send(());
let _ = server.await;
let cap = cap.lock().unwrap();
assert_eq!(cap.applied.len(), 1, "one ApplyConfiguration call");
assert_eq!(
cap.applied[0].0, machineconfig,
"exact machineconfig bytes on the wire"
);
assert_eq!(cap.applied[0].1, ApplyMode::Auto as i32, "ApplyMode::Auto");
assert_eq!(cap.bootstrapped, 1, "one Bootstrap call");
}
/// A node-level rejection (non-empty `metadata.error`) surfaces as an error.
#[tokio::test]
async fn apply_surfaces_node_rejection() {
#[derive(Clone)]
struct RejectApid;
impl tonic::server::NamedService for RejectApid {
const NAME: &'static str = "machine.MachineService";
}
impl tonic::codegen::Service<tonic::codegen::http::Request<tonic::body::Body>> for RejectApid {
type Response = tonic::codegen::http::Response<tonic::body::Body>;
type Error = std::convert::Infallible;
type Future = tonic::codegen::BoxFuture<Self::Response, Self::Error>;
fn poll_ready(
&mut self,
_cx: &mut tonic::codegen::Context<'_>,
) -> tonic::codegen::Poll<Result<(), Self::Error>> {
tonic::codegen::Poll::Ready(Ok(()))
}
fn call(
&mut self,
req: tonic::codegen::http::Request<tonic::body::Body>,
) -> Self::Future {
Box::pin(async move {
struct Svc;
impl tonic::server::UnaryService<ApplyConfigurationRequest> for Svc {
type Response = ApplyConfigurationResponse;
type Future = tonic::codegen::BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
_r: tonic::Request<ApplyConfigurationRequest>,
) -> Self::Future {
Box::pin(async move {
Ok(tonic::Response::new(ApplyConfigurationResponse {
messages: vec![ApplyConfiguration {
metadata: Some(Metadata {
hostname: "n".into(),
error: "invalid machineconfig".into(),
}),
mode_details: String::new(),
}],
}))
})
}
}
let mut grpc = tonic::server::Grpc::new(tonic_prost::ProstCodec::default());
Ok(grpc.unary(Svc, req).await)
})
}
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener);
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
let server = tokio::spawn(async move {
tonic::transport::Server::builder()
.add_service(RejectApid)
.serve_with_incoming_shutdown(incoming, async {
let _ = rx.await;
})
.await
.unwrap();
});
let mut client = ApidClient::connect(&format!("http://{addr}"), None)
.await
.unwrap();
let err = client
.apply_configuration(b"x".to_vec(), ApplyMode::Auto)
.await
.unwrap_err()
.to_string();
assert!(
err.contains("invalid machineconfig"),
"surfaces node error: {err}"
);
let _ = tx.send(());
let _ = server.await;
}
}