use std::fmt;
use std::path::PathBuf;
use std::time::Duration;
use engenho_config::{
ClusterConfig, ConsistencyConfig, ConsistencyTierKind, ControllerEnable, ControllersConfig,
DatapathMode, EngenhoConfig, Fabric, HealthReasons, KubeconfigVisibility, KubeletBackendKind,
LegacyTeiaSection, NativeCgroups, NetworkingConfig, RevoadaConfig, RuntimeConfig,
SchedulerConfig, SharedCluster, TlsConfig, TopologyConfig, TopologyStrategyKind,
};
use crate::health::Windows;
#[derive(Clone, Debug)]
pub(crate) struct BootConfig {
pub(crate) cluster_name: String,
pub(crate) membership: Membership,
pub(crate) fabric: MeshFabric,
pub(crate) listen_addr: String,
pub(crate) data_dir: PathBuf,
pub(crate) durable: bool,
pub(crate) node_name: String,
pub(crate) kubeconfig_publish_path: String,
pub(crate) kubeconfig_publish_visibility: KubeconfigVisibility,
pub(crate) pod_kubeconfig_publish_path: String,
pub(crate) advertise_address: String,
pub(crate) remote_kubeconfig_publish_path: String,
pub(crate) kubelet_listen_addr: String,
pub(crate) etcd_listen_addr: String,
pub(crate) kubelet_backend: KubeletBackendKind,
pub(crate) native_cgroups: NativeCgroups,
pub(crate) health_reasons: HealthReasons,
pub(crate) podman_binary: Option<String>,
pub(crate) host_path_allowlist: Vec<String>,
pub(crate) node_manifests_dir: PathBuf,
pub(crate) leadership_timeout_seconds: u32,
pub(crate) tls: ApiserverTls,
pub(crate) scheduler: SchedulerConfig,
pub(crate) enable: ControllerEnable,
pub(crate) controllers_namespace: Option<String>,
fallback: Duration,
debounce: Duration,
pub(crate) service_cidr: String,
pub(crate) datapath_mode: DatapathMode,
pub(crate) not_run: Vec<NotRun>,
}
impl BootConfig {
pub(crate) fn read(config: &EngenhoConfig) -> Result<Self, Unhonoured> {
let EngenhoConfig {
cluster,
revoada,
fabric,
legacy_teia,
scheduler,
controllers,
consistency,
networking,
runtime,
control: _,
} = config;
let ClusterConfig {
name,
region,
shared,
} = cluster;
let membership = Membership::read(shared.as_ref());
let (mesh_fabric, fabric_not_run) = read_fabric(fabric, legacy_teia.as_ref())?;
if mesh_fabric.carries_peers() && membership.shared().is_none() {
return Err(Unhonoured::MeshWithoutIdentity);
}
if membership.joins() && !mesh_fabric.carries_peers() {
return Err(Unhonoured::SeedsWithoutMesh);
}
let cluster_name = shared.as_ref().map_or(name, |shared| &shared.name).clone();
let not_run = [
shared.is_some().then_some(NotRun::HostClusterName),
(!region.is_empty()).then_some(NotRun::Region),
Some(read_topology(revoada, membership.joins())?),
fabric_not_run,
]
.into_iter()
.flatten()
.collect();
read_consistency(consistency)?;
let ControllersConfig {
enable,
namespace,
fallback_interval_seconds,
debounce_milliseconds,
} = controllers;
let NetworkingConfig {
service_cidr,
datapath_mode,
} = networking;
let RuntimeConfig {
listen_addr,
data_dir,
durable,
node_name,
kubeconfig_publish_path,
kubeconfig_publish_visibility,
pod_kubeconfig_publish_path,
advertise_address,
remote_kubeconfig_publish_path,
kubelet_listen_addr,
etcd_listen_addr,
kubelet_backend,
native_cgroups,
health_reasons,
podman_binary,
host_path_allowlist,
node_manifests_dir,
leadership_timeout_seconds,
tls,
} = runtime;
Ok(Self {
cluster_name,
membership,
fabric: mesh_fabric,
listen_addr: listen_addr.clone(),
data_dir: data_dir.clone(),
durable: *durable,
node_name: node_name.clone(),
kubeconfig_publish_path: kubeconfig_publish_path.clone(),
kubeconfig_publish_visibility: *kubeconfig_publish_visibility,
pod_kubeconfig_publish_path: pod_kubeconfig_publish_path.clone(),
advertise_address: advertise_address.clone(),
remote_kubeconfig_publish_path: remote_kubeconfig_publish_path.clone(),
kubelet_listen_addr: kubelet_listen_addr.clone(),
etcd_listen_addr: etcd_listen_addr.clone(),
kubelet_backend: *kubelet_backend,
native_cgroups: *native_cgroups,
health_reasons: *health_reasons,
podman_binary: podman_binary.clone(),
host_path_allowlist: host_path_allowlist.clone(),
node_manifests_dir: node_manifests_dir.clone(),
leadership_timeout_seconds: *leadership_timeout_seconds,
tls: ApiserverTls::read(tls)?,
scheduler: scheduler.clone(),
enable: enable.clone(),
controllers_namespace: Some(namespace).filter(|n| !n.is_empty()).cloned(),
fallback: Duration::from_secs(u64::from(*fallback_interval_seconds)),
debounce: Duration::from_millis(u64::from(*debounce_milliseconds)),
service_cidr: service_cidr.clone(),
datapath_mode: *datapath_mode,
not_run,
})
}
pub(crate) fn windows(&self, scheduler_fallback: Duration) -> Windows {
Windows::new(self.fallback, self.debounce, scheduler_fallback)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum Membership {
Own,
Shared(SharedMembership),
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct SharedMembership {
pub(crate) ca_cert: PathBuf,
pub(crate) ca_key: PathBuf,
pub(crate) identity_key: PathBuf,
pub(crate) seeds: Vec<std::net::SocketAddr>,
}
impl Membership {
fn read(shared: Option<&SharedCluster>) -> Self {
let Some(SharedCluster {
name: _,
ca_cert_path,
ca_key_path,
identity_key_path,
seeds,
}) = shared
else {
return Self::Own;
};
Self::Shared(SharedMembership {
ca_cert: ca_cert_path.clone(),
ca_key: ca_key_path.clone(),
identity_key: identity_key_path.clone(),
seeds: seeds.clone(),
})
}
pub(crate) fn joins(&self) -> bool {
self.shared().is_some_and(|shared| !shared.seeds.is_empty())
}
pub(crate) const fn shared(&self) -> Option<&SharedMembership> {
match self {
Self::Own => None,
Self::Shared(shared) => Some(shared),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum ApiserverTls {
Plaintext,
SelfIssued {
extra_sans: Vec<String>,
},
}
impl ApiserverTls {
fn read(tls: &TlsConfig) -> Result<Self, Unhonoured> {
let TlsConfig {
enabled,
auto_generate,
ca_cert_path,
ca_key_path,
cert_path,
key_path,
extra_sans,
} = tls;
if !*enabled {
return Ok(Self::Plaintext);
}
if !*auto_generate {
return Err(Unhonoured::OperatorPki {
field: PkiField::AutoGenerateOff,
});
}
for (field, path) in [
(PkiField::CaCertPath, ca_cert_path),
(PkiField::CaKeyPath, ca_key_path),
(PkiField::CertPath, cert_path),
(PkiField::KeyPath, key_path),
] {
if path.is_some() {
return Err(Unhonoured::OperatorPki { field });
}
}
Ok(Self::SelfIssued {
extra_sans: extra_sans.clone(),
})
}
pub(crate) const fn is_enabled(&self) -> bool {
matches!(self, Self::SelfIssued { .. })
}
}
fn read_topology(revoada: &RevoadaConfig, joins: bool) -> Result<NotRun, Unhonoured> {
let RevoadaConfig { topology } = revoada;
let TopologyConfig {
strategy,
min_nodes: _,
grace_period_seconds: _,
} = topology;
match masters_named(*strategy) {
Some(_) if joins => Ok(NotRun::Topology {
strategy: *strategy,
}),
Some(masters) => Err(Unhonoured::MultiNodeFormation {
strategy: *strategy,
masters,
}),
None => Ok(NotRun::Topology {
strategy: *strategy,
}),
}
}
#[cfg_attr(feature = "mesh-quic", allow(clippy::unnecessary_wraps))]
fn read_fabric(
fabric: &Fabric,
legacy_teia: Option<&LegacyTeiaSection>,
) -> Result<(MeshFabric, Option<NotRun>), Unhonoured> {
#[allow(unreachable_patterns)]
let mesh = match fabric {
Fabric::InBinary => MeshFabric::InProcess,
#[cfg(feature = "mesh-quic")]
Fabric::Quic { bind } => MeshFabric::Quic { bind: *bind },
#[cfg(not(feature = "mesh-quic"))]
_ => return Err(Unhonoured::FabricNotBuilt),
};
Ok((mesh, legacy_teia.map(|_| NotRun::Teia)))
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum MeshFabric {
InProcess,
#[cfg(feature = "mesh-quic")]
Quic {
bind: std::net::SocketAddr,
},
}
impl MeshFabric {
pub(crate) const fn carries_peers(self) -> bool {
match self {
Self::InProcess => false,
#[cfg(feature = "mesh-quic")]
Self::Quic { .. } => true,
}
}
}
fn read_consistency(consistency: &ConsistencyConfig) -> Result<(), Unhonoured> {
let ConsistencyConfig { default_tier } = consistency;
match default_tier {
ConsistencyTierKind::Strong => Ok(()),
requested @ (ConsistencyTierKind::EventualGossip
| ConsistencyTierKind::DurableStream
| ConsistencyTierKind::Content) => Err(Unhonoured::ConsistencyTier {
requested: *requested,
}),
}
}
const fn masters_named(strategy: TopologyStrategyKind) -> Option<u32> {
match strategy {
TopologyStrategyKind::Solo
| TopologyStrategyKind::MeshAllPeers
| TopologyStrategyKind::Phalanx => None,
TopologyStrategyKind::Pair => Some(2),
TopologyStrategyKind::Quorum3M | TopologyStrategyKind::Cluster3MNW => Some(3),
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PkiField {
AutoGenerateOff,
CaCertPath,
CaKeyPath,
CertPath,
KeyPath,
}
impl fmt::Display for PkiField {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::AutoGenerateOff => "runtime.tls.auto_generate: false",
Self::CaCertPath => "runtime.tls.ca_cert_path",
Self::CaKeyPath => "runtime.tls.ca_key_path",
Self::CertPath => "runtime.tls.cert_path",
Self::KeyPath => "runtime.tls.key_path",
})
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub enum Unhonoured {
#[error(
"{field} is set, and this runtime serves only the CA it loads or mints under \
data_dir/pki; unset it (with runtime.tls.auto_generate: true) or disable TLS"
)]
OperatorPki {
field: PkiField,
},
#[error(
"revoada.topology.strategy {strategy:?} names {masters} masters, and this runtime is \
one in-process raft voter; use solo, phalanx or mesh_all_peers"
)]
MultiNodeFormation {
strategy: TopologyStrategyKind,
masters: u32,
},
#[error("fabric names a transport this build does not carry; build with mesh-quic")]
FabricNotBuilt,
#[error(
"fabric quic carries raft between shared-cluster members, and cluster.shared is not set"
)]
MeshWithoutIdentity,
#[error(
"cluster.shared.seeds names members to join, and fabric is in_binary, which reaches none"
)]
SeedsWithoutMesh,
#[error(
"consistency.default_tier {requested:?} is not served: the store runs strong \
(linearizable raft) only"
)]
ConsistencyTier {
requested: ConsistencyTierKind,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum NotRun {
HostClusterName,
Region,
Topology {
strategy: TopologyStrategyKind,
},
Teia,
}
impl NotRun {
pub(crate) const fn field(self) -> &'static str {
match self {
Self::HostClusterName => "cluster.name",
Self::Region => "cluster.region",
Self::Topology { .. } => "revoada.topology",
Self::Teia => "teia",
}
}
}
impl fmt::Display for NotRun {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::HostClusterName => {
f.write_str("this node is a member of cluster.shared.name, which it answers to")
}
Self::Region => f.write_str("nothing in a single-node runtime places by region"),
Self::Topology { strategy } => write!(
f,
"{strategy:?} is a formation one node already is; no topology strategy runs"
),
Self::Teia => f.write_str("no NATS fabric runs; the store's raft is in-process"),
}
}
}
#[cfg(test)]
impl BootConfig {
pub(crate) fn prescribed() -> Self {
use shikumi::TieredConfig as _;
Self::read(&EngenhoConfig::prescribed_default()).expect("the prescribed default is run")
}
pub(crate) fn prescribed_windows() -> Windows {
let boot = Self::prescribed();
let scheduler_fallback =
Duration::from_secs(u64::from(boot.scheduler.tick_interval_seconds));
boot.windows(scheduler_fallback)
}
}
#[cfg(test)]
mod tests {
use engenho_config::{ConsistencyTierKind, EngenhoConfig, TopologyStrategyKind};
use shikumi::TieredConfig as _;
use super::{
ApiserverTls, BootConfig, Membership, NotRun, PkiField, SharedMembership, Unhonoured,
};
fn read(edit: impl FnOnce(&mut EngenhoConfig)) -> Result<BootConfig, Unhonoured> {
let mut config = EngenhoConfig::prescribed_default();
edit(&mut config);
BootConfig::read(&config)
}
#[test]
fn the_prescribed_default_is_run_and_names_what_it_does_not_run() {
let boot = BootConfig::prescribed();
assert_eq!(
boot.not_run,
vec![
NotRun::Region,
NotRun::Topology {
strategy: TopologyStrategyKind::Phalanx
},
]
);
assert!(matches!(boot.tls, ApiserverTls::SelfIssued { .. }));
}
#[test]
fn a_retired_teia_section_is_accepted_and_recorded() {
let boot = read(|c| {
c.legacy_teia = serde_json::from_value(serde_json::json!({
"servers": ["nats://127.0.0.1:4222"],
"cluster": "engenho",
}))
.expect("the retired section still deserializes");
})
.expect("a config with the retired section still boots");
assert!(boot.not_run.contains(&NotRun::Teia), "{:?}", boot.not_run);
}
#[test]
fn a_shared_block_names_the_cluster_and_is_the_membership() {
let boot = read(|c| {
c.cluster.shared = Some(engenho_config::SharedCluster {
name: "aldeia".into(),
ca_cert_path: "/pki/ca.crt".into(),
ca_key_path: "/pki/ca.key".into(),
identity_key_path: "/pki/node.key".into(),
seeds: Vec::new(),
});
})
.expect("run");
assert_eq!(boot.cluster_name, "aldeia");
assert_eq!(
boot.membership,
Membership::Shared(SharedMembership {
ca_cert: "/pki/ca.crt".into(),
ca_key: "/pki/ca.key".into(),
identity_key: "/pki/node.key".into(),
seeds: Vec::new(),
})
);
assert_eq!(boot.not_run.first(), Some(&NotRun::HostClusterName));
assert_eq!(boot.fabric, super::MeshFabric::InProcess);
let own = BootConfig::prescribed();
assert_eq!(own.membership, Membership::Own);
assert_eq!(own.fabric, super::MeshFabric::InProcess);
assert!(
own.cluster_name.starts_with("engenho-"),
"{}",
own.cluster_name
);
}
#[test]
fn an_empty_region_is_not_recorded() {
let boot = read(|c| c.cluster.region.clear()).expect("run");
assert!(
!boot.not_run.contains(&NotRun::Region),
"{:?}",
boot.not_run
);
}
#[test]
fn an_operator_pki_is_refused_not_replaced_by_the_minted_ca() {
let refused = |edit: fn(&mut EngenhoConfig)| read(edit).map(|_| ()).unwrap_err();
assert_eq!(
refused(|c| {
c.runtime.tls.auto_generate = false;
c.runtime.tls.ca_cert_path = Some("/pki/ca.crt".into());
c.runtime.tls.ca_key_path = Some("/pki/ca.key".into());
c.runtime.tls.cert_path = Some("/pki/srv.crt".into());
c.runtime.tls.key_path = Some("/pki/srv.key".into());
}),
Unhonoured::OperatorPki {
field: PkiField::AutoGenerateOff
}
);
assert_eq!(
refused(|c| c.runtime.tls.ca_cert_path = Some("/pki/ca.crt".into())),
Unhonoured::OperatorPki {
field: PkiField::CaCertPath
}
);
assert_eq!(
refused(|c| c.runtime.tls.key_path = Some("/pki/srv.key".into())),
Unhonoured::OperatorPki {
field: PkiField::KeyPath
}
);
}
#[test]
fn plaintext_reads_no_pki_field() {
let boot = read(|c| {
c.runtime.tls.enabled = false;
c.runtime.tls.auto_generate = false;
c.runtime.tls.ca_cert_path = Some("/pki/ca.crt".into());
})
.expect("run");
assert_eq!(boot.tls, ApiserverTls::Plaintext);
}
#[test]
fn a_multi_master_formation_is_refused() {
for (strategy, masters) in [
(TopologyStrategyKind::Pair, 2),
(TopologyStrategyKind::Quorum3M, 3),
(TopologyStrategyKind::Cluster3MNW, 3),
] {
assert_eq!(
read(|c| {
c.revoada.topology.strategy = strategy;
c.revoada.topology.min_nodes = 3;
})
.map(|_| ())
.unwrap_err(),
Unhonoured::MultiNodeFormation { strategy, masters },
);
}
}
#[test]
fn a_multi_master_formation_is_read_not_refused_only_for_a_shared_node_that_joins() {
let shared = |seeds: Vec<std::net::SocketAddr>| engenho_config::SharedCluster {
name: "aldeia".into(),
ca_cert_path: "/pki/ca.crt".into(),
ca_key_path: "/pki/ca.key".into(),
identity_key_path: "/pki/node.key".into(),
seeds,
};
let formation = |c: &mut EngenhoConfig| {
c.revoada.topology.strategy = TopologyStrategyKind::Quorum3M;
c.revoada.topology.min_nodes = 3;
};
assert_eq!(
read(|c| {
formation(c);
c.cluster.shared = Some(shared(Vec::new()));
})
.map(|_| ())
.unwrap_err(),
Unhonoured::MultiNodeFormation {
strategy: TopologyStrategyKind::Quorum3M,
masters: 3
},
"a node that forms is still one voter"
);
let joins = read(|c| {
formation(c);
c.cluster.shared = Some(shared(vec!["127.0.0.1:7946".parse().unwrap()]));
#[cfg(feature = "mesh-quic")]
{
c.fabric = engenho_config::Fabric::Quic {
bind: "127.0.0.1:0".parse().unwrap(),
};
}
});
#[cfg(feature = "mesh-quic")]
assert!(joins.unwrap().not_run.contains(&NotRun::Topology {
strategy: TopologyStrategyKind::Quorum3M
}));
#[cfg(not(feature = "mesh-quic"))]
assert_eq!(joins.map(|_| ()).unwrap_err(), Unhonoured::SeedsWithoutMesh);
}
#[test]
fn a_formation_one_node_already_is_is_run() {
for strategy in [
TopologyStrategyKind::Solo,
TopologyStrategyKind::MeshAllPeers,
TopologyStrategyKind::Phalanx,
] {
let boot = read(|c| c.revoada.topology.strategy = strategy).expect("run");
assert!(
boot.not_run.contains(&NotRun::Topology { strategy }),
"{:?}",
boot.not_run
);
}
}
#[test]
fn only_the_strong_tier_is_served() {
for requested in [
ConsistencyTierKind::EventualGossip,
ConsistencyTierKind::DurableStream,
ConsistencyTierKind::Content,
] {
assert_eq!(
read(|c| c.consistency.default_tier = requested)
.map(|_| ())
.unwrap_err(),
Unhonoured::ConsistencyTier { requested },
);
}
}
#[test]
fn an_empty_controllers_namespace_is_every_namespace() {
assert_eq!(BootConfig::prescribed().controllers_namespace, None);
let boot = read(|c| c.controllers.namespace = "team-a".into()).expect("run");
assert_eq!(boot.controllers_namespace.as_deref(), Some("team-a"));
}
}