use chrono::{DateTime, Utc};
use kube::CustomResource;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use crate::ephemeral::EphemeralSpec;
#[derive(CustomResource, Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[kube(
group = "tatara.pleme.io",
version = "v1alpha1",
kind = "EphemeralPool",
plural = "ephemeralpools",
shortname = "epool",
namespaced,
status = "PoolStatus",
printcolumn = r#"{"name":"Desired","type":"integer","jsonPath":".spec.desiredSize"}"#,
printcolumn = r#"{"name":"Ready","type":"integer","jsonPath":".status.readyCount"}"#,
printcolumn = r#"{"name":"Allocated","type":"integer","jsonPath":".status.allocatedCount"}"#,
printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#,
printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
)]
#[serde(rename_all = "camelCase")]
pub struct PoolSpec {
pub desired_size: u32,
#[serde(default)]
pub min_size: u32,
#[serde(default)]
pub max_size: u32,
#[serde(default)]
pub return_policy: ReturnPolicy,
#[serde(default)]
pub selector: PoolSelector,
pub template: EphemeralSpec,
#[serde(default = "default_free_ttl")]
pub free_ttl: String,
#[serde(default = "default_max_allocation_ttl")]
pub max_allocation_ttl: String,
#[serde(default)]
pub desired: u32,
#[serde(default)]
pub replacement_policy: ReplacementPolicy,
#[serde(default)]
pub stable_name_claim: bool,
}
#[derive(
Clone,
Copy,
Debug,
Default,
Serialize,
Deserialize,
JsonSchema,
PartialEq,
Eq,
Hash,
tatara_lisp::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum ReplacementPolicy {
#[default]
ReplaceImmediate,
HoldFailed,
PausePool,
}
impl ReplacementPolicy {
pub const ALL: [Self; 3] = [Self::ReplaceImmediate, Self::HoldFailed, Self::PausePool];
pub const fn as_str(self) -> &'static str {
match self {
Self::ReplaceImmediate => "ReplaceImmediate",
Self::HoldFailed => "HoldFailed",
Self::PausePool => "PausePool",
}
}
pub const fn replaces_failed(self) -> bool {
match self {
Self::ReplaceImmediate => true,
Self::HoldFailed | Self::PausePool => false,
}
}
pub const fn pauses_on_failure(self) -> bool {
match self {
Self::PausePool => true,
Self::ReplaceImmediate | Self::HoldFailed => false,
}
}
}
fn default_free_ttl() -> String {
"24h".to_string()
}
fn default_max_allocation_ttl() -> String {
"4h".to_string()
}
#[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolStatus {
#[serde(default)]
pub phase: PoolPhase,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub phase_since: Option<DateTime<Utc>>,
#[serde(default)]
pub ready_count: u32,
#[serde(default)]
pub allocated_count: u32,
#[serde(default)]
pub spawning_count: u32,
#[serde(default)]
pub returning_count: u32,
#[serde(default)]
pub members: Vec<PoolMember>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
#[serde(default)]
pub conditions: Vec<PoolCondition>,
}
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolMember {
pub process_name: String,
pub state: MemberState,
pub entered_state_at: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub allocation_ref: Option<AllocationRef>,
}
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct AllocationRef {
pub name: String,
pub namespace: String,
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
Hash,
Serialize,
Deserialize,
JsonSchema,
tatara_lisp::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum MemberState {
Spawning,
Free,
Allocated,
Returning,
Failed,
}
impl MemberState {
pub const ALL: [Self; 5] = [
Self::Spawning,
Self::Free,
Self::Allocated,
Self::Returning,
Self::Failed,
];
pub const fn as_str(self) -> &'static str {
match self {
Self::Spawning => "Spawning",
Self::Free => "Free",
Self::Allocated => "Allocated",
Self::Returning => "Returning",
Self::Failed => "Failed",
}
}
pub const fn is_failed(self) -> bool {
match self {
Self::Failed => true,
Self::Spawning | Self::Free | Self::Allocated | Self::Returning => false,
}
}
pub const fn counts_toward_supply(self) -> bool {
match self {
Self::Free | Self::Spawning => true,
Self::Allocated | Self::Returning | Self::Failed => false,
}
}
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
Hash,
Serialize,
Deserialize,
JsonSchema,
tatara_lisp::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum PoolPhase {
Initializing,
Steady,
ScalingUp,
ScalingDown,
Degraded,
Draining,
}
impl Default for PoolPhase {
fn default() -> Self {
Self::Initializing
}
}
impl PoolPhase {
pub const ALL: [Self; 6] = [
Self::Initializing,
Self::Steady,
Self::ScalingUp,
Self::ScalingDown,
Self::Degraded,
Self::Draining,
];
pub const fn as_str(self) -> &'static str {
match self {
Self::Initializing => "Initializing",
Self::Steady => "Steady",
Self::ScalingUp => "ScalingUp",
Self::ScalingDown => "ScalingDown",
Self::Degraded => "Degraded",
Self::Draining => "Draining",
}
}
pub const fn is_steady(self) -> bool {
match self {
Self::Steady => true,
Self::Initializing
| Self::ScalingUp
| Self::ScalingDown
| Self::Degraded
| Self::Draining => false,
}
}
pub const fn is_terminal(self) -> bool {
match self {
Self::Draining => true,
Self::Initializing
| Self::Steady
| Self::ScalingUp
| Self::ScalingDown
| Self::Degraded => false,
}
}
}
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolCondition {
pub type_: String,
pub status: String,
pub reason: String,
pub message: String,
pub last_transition_time: DateTime<Utc>,
}
#[derive(
Clone,
Copy,
Debug,
Hash,
PartialEq,
Eq,
Serialize,
Deserialize,
JsonSchema,
Default,
tatara_lisp::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum ReturnPolicy {
#[default]
Replace,
Reset,
Keep,
}
impl ReturnPolicy {
pub const ALL: [Self; 3] = [Self::Replace, Self::Reset, Self::Keep];
pub const fn as_str(self) -> &'static str {
match self {
Self::Replace => "Replace",
Self::Reset => "Reset",
Self::Keep => "Keep",
}
}
pub const fn keeps_process(self) -> bool {
match self {
Self::Replace => false,
Self::Reset | Self::Keep => true,
}
}
pub const fn runs_reset_job(self) -> bool {
match self {
Self::Reset => true,
Self::Replace | Self::Keep => false,
}
}
}
#[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PoolSelector {
#[serde(default)]
pub repos: Vec<String>,
#[serde(default)]
pub branches: Vec<String>,
#[serde(default)]
pub pr_labels: Vec<String>,
#[serde(default)]
pub kinds: Vec<String>,
}
impl PoolSelector {
pub fn matches(&self, key: &MatchKey<'_>) -> bool {
glob_any(&self.repos, key.repo)
&& glob_any(&self.branches, key.branch)
&& labels_subset(&self.pr_labels, key.pr_labels)
&& kind_any(&self.kinds, key.kind)
}
pub fn specificity(&self) -> u32 {
let mut score = 0;
if !self.repos.is_empty() {
score += 8;
}
if !self.branches.is_empty() {
score += 4;
}
score += (self.pr_labels.len() as u32) * 2;
if !self.kinds.is_empty() {
score += 1;
}
score
}
}
#[derive(Clone, Copy, Debug)]
pub struct MatchKey<'a> {
pub repo: &'a str,
pub branch: &'a str,
pub pr_labels: &'a [String],
pub kind: &'a str,
}
fn glob_any(patterns: &[String], value: &str) -> bool {
if patterns.is_empty() {
return true;
}
patterns.iter().any(|p| glob_match(p, value))
}
fn kind_any(kinds: &[String], value: &str) -> bool {
if kinds.is_empty() {
return true;
}
kinds.iter().any(|k| k == value)
}
fn labels_subset(required: &[String], present: &[String]) -> bool {
required.iter().all(|r| present.iter().any(|p| p == r))
}
fn glob_match(pattern: &str, value: &str) -> bool {
if pattern.is_empty() {
return true;
}
if let Some(prefix) = pattern.strip_suffix('*') {
value.starts_with(prefix)
} else {
pattern == value
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::str::FromStr;
#[test]
fn glob_trailing_star_matches_prefix() {
assert!(glob_match("pleme-io/*", "pleme-io/akeyless-deployment"));
assert!(!glob_match("pleme-io/*", "drzln/dotfiles"));
assert!(glob_match("release-*", "release-2026-05"));
assert!(!glob_match("release-*", "main"));
assert!(glob_match("main", "main"));
assert!(!glob_match("main", "develop"));
}
#[test]
fn empty_selector_matches_anything() {
let s = PoolSelector::default();
assert!(s.matches(&MatchKey {
repo: "any/repo",
branch: "any-branch",
pr_labels: &[],
kind: "any",
}));
}
#[test]
fn repo_glob_filters_match_key() {
let s = PoolSelector {
repos: vec!["pleme-io/akeyless-*".into()],
..Default::default()
};
assert!(s.matches(&MatchKey {
repo: "pleme-io/akeyless-deployment",
branch: "x",
pr_labels: &[],
kind: "y",
}));
assert!(!s.matches(&MatchKey {
repo: "pleme-io/other-repo",
branch: "x",
pr_labels: &[],
kind: "y",
}));
}
#[test]
fn pr_labels_require_all() {
let s = PoolSelector {
pr_labels: vec!["needs-akeyless".into(), "integration".into()],
..Default::default()
};
assert!(s.matches(&MatchKey {
repo: "x",
branch: "y",
pr_labels: &[
"needs-akeyless".into(),
"integration".into(),
"extra".into()
],
kind: "z",
}));
assert!(!s.matches(&MatchKey {
repo: "x",
branch: "y",
pr_labels: &["needs-akeyless".into()],
kind: "z",
}));
}
#[test]
fn specificity_ranks_more_constrained_higher() {
let general = PoolSelector::default();
let specific = PoolSelector {
repos: vec!["pleme-io/*".into()],
branches: vec!["main".into()],
pr_labels: vec!["needs-akeyless".into()],
kinds: vec!["github-pr".into()],
};
assert!(specific.specificity() > general.specificity());
}
#[test]
fn return_policy_defaults_to_replace() {
assert_eq!(ReturnPolicy::default(), ReturnPolicy::Replace);
}
#[test]
fn pool_phase_defaults_to_initializing() {
assert_eq!(PoolPhase::default(), PoolPhase::Initializing);
}
#[test]
fn replacement_policy_is_well_formed_closed_set() {
tatara_lisp::assert_closed_set_well_formed::<ReplacementPolicy>();
}
#[test]
fn replacement_policy_as_str_matches_serde() {
for policy in ReplacementPolicy::ALL {
let serialized = serde_json::to_string(&policy).expect("serialize");
let unquoted = serialized
.trim_start_matches('"')
.trim_end_matches('"')
.to_string();
assert_eq!(
unquoted,
policy.as_str(),
"as_str drift for {policy:?}: as_str={} serde={unquoted}",
policy.as_str()
);
}
}
#[test]
fn replacement_policy_display_matches_as_str() {
for policy in ReplacementPolicy::ALL {
assert_eq!(policy.to_string(), policy.as_str());
}
}
#[test]
fn unknown_replacement_policy_errors() {
for bad in [
"replaceimmediate",
"PAUSEPOOL",
"Replace-Immediate",
"hold_failed",
"Pause",
"Reset",
] {
let err = ReplacementPolicy::from_str(bad).unwrap_err();
assert_eq!(err.0, bad, "error payload should echo input verbatim");
}
}
#[test]
fn replacement_policy_predicate_truth_tables() {
assert!(ReplacementPolicy::ReplaceImmediate.replaces_failed());
assert!(!ReplacementPolicy::ReplaceImmediate.pauses_on_failure());
assert!(!ReplacementPolicy::HoldFailed.replaces_failed());
assert!(!ReplacementPolicy::HoldFailed.pauses_on_failure());
assert!(!ReplacementPolicy::PausePool.replaces_failed());
assert!(ReplacementPolicy::PausePool.pauses_on_failure());
}
#[test]
fn replacement_policy_predicates_are_disjoint() {
for policy in ReplacementPolicy::ALL {
assert!(
!(policy.replaces_failed() && policy.pauses_on_failure()),
"{policy:?} returns true from both replaces_failed and pauses_on_failure",
);
}
}
#[test]
fn replacement_policy_predicate_pair_is_injective() {
let projections: Vec<(bool, bool)> = ReplacementPolicy::ALL
.into_iter()
.map(|p| (p.replaces_failed(), p.pauses_on_failure()))
.collect();
let unique: std::collections::HashSet<_> = projections.iter().copied().collect();
assert_eq!(
projections.len(),
unique.len(),
"predicate pair projection is not injective: {projections:?}",
);
}
#[test]
fn replacement_policy_default_replaces_failed() {
let d = ReplacementPolicy::default();
assert_eq!(d, ReplacementPolicy::ReplaceImmediate);
assert!(d.replaces_failed());
assert!(!d.pauses_on_failure());
}
#[test]
fn kinds_filter_to_known_set() {
let s = PoolSelector {
kinds: vec!["github-pr".into(), "manual".into()],
..Default::default()
};
assert!(s.matches(&MatchKey {
repo: "x",
branch: "y",
pr_labels: &[],
kind: "github-pr",
}));
assert!(!s.matches(&MatchKey {
repo: "x",
branch: "y",
pr_labels: &[],
kind: "scheduled",
}));
}
#[test]
fn return_policy_is_well_formed_closed_set() {
tatara_lisp::assert_closed_set_well_formed::<ReturnPolicy>();
}
#[test]
fn return_policy_as_str_matches_serde() {
for policy in ReturnPolicy::ALL {
let serialized = serde_json::to_string(&policy).expect("serialize");
let unquoted = serialized
.trim_start_matches('"')
.trim_end_matches('"')
.to_string();
assert_eq!(
unquoted,
policy.as_str(),
"as_str drift for {policy:?}: as_str={} serde={unquoted}",
policy.as_str()
);
}
}
#[test]
fn return_policy_display_matches_as_str() {
for policy in ReturnPolicy::ALL {
assert_eq!(policy.to_string(), policy.as_str());
}
}
#[test]
fn unknown_return_policy_errors() {
for bad in [
"replace",
"RESET",
"Re-place",
"keep_for_inspection",
"DeleteAndRespawn",
"ReplaceImmediate",
] {
let err = ReturnPolicy::from_str(bad).unwrap_err();
assert_eq!(err.0, bad, "error payload should echo input verbatim");
}
}
#[test]
fn return_policy_predicate_truth_tables() {
assert!(!ReturnPolicy::Replace.keeps_process());
assert!(!ReturnPolicy::Replace.runs_reset_job());
assert!(ReturnPolicy::Reset.keeps_process());
assert!(ReturnPolicy::Reset.runs_reset_job());
assert!(ReturnPolicy::Keep.keeps_process());
assert!(!ReturnPolicy::Keep.runs_reset_job());
}
#[test]
fn return_policy_reset_implies_keeps_process() {
for policy in ReturnPolicy::ALL {
if policy.runs_reset_job() {
assert!(
policy.keeps_process(),
"{policy:?} runs a reset job but does not keep the process",
);
}
}
}
#[test]
fn return_policy_predicate_pair_is_injective() {
let projections: Vec<(bool, bool)> = ReturnPolicy::ALL
.into_iter()
.map(|p| (p.keeps_process(), p.runs_reset_job()))
.collect();
let unique: std::collections::HashSet<_> = projections.iter().copied().collect();
assert_eq!(
projections.len(),
unique.len(),
"predicate pair projection is not injective: {projections:?}",
);
}
#[test]
fn return_policy_default_is_replace_and_neither_predicate_fires() {
let d = ReturnPolicy::default();
assert_eq!(d, ReturnPolicy::Replace);
assert!(!d.keeps_process());
assert!(!d.runs_reset_job());
}
#[test]
fn member_state_is_well_formed_closed_set() {
tatara_lisp::assert_closed_set_well_formed::<MemberState>();
}
#[test]
fn member_state_as_str_matches_serde() {
for state in MemberState::ALL {
let serialized = serde_json::to_string(&state).expect("serialize");
let unquoted = serialized
.trim_start_matches('"')
.trim_end_matches('"')
.to_string();
assert_eq!(
unquoted,
state.as_str(),
"as_str drift for {state:?}: as_str={} serde={unquoted}",
state.as_str()
);
}
}
#[test]
fn member_state_display_matches_as_str() {
for state in MemberState::ALL {
assert_eq!(state.to_string(), state.as_str());
}
}
#[test]
fn unknown_member_state_errors() {
for bad in [
"free",
"SPAWNING",
"Free-State",
"allocated_now",
"ReplaceImmediate", "Reset", "Attested", ] {
let err = MemberState::from_str(bad).unwrap_err();
assert_eq!(err.0, bad, "error payload should echo input verbatim");
}
}
#[test]
fn member_state_predicate_truth_tables() {
assert!(!MemberState::Spawning.is_failed());
assert!(MemberState::Spawning.counts_toward_supply());
assert!(!MemberState::Free.is_failed());
assert!(MemberState::Free.counts_toward_supply());
assert!(!MemberState::Allocated.is_failed());
assert!(!MemberState::Allocated.counts_toward_supply());
assert!(!MemberState::Returning.is_failed());
assert!(!MemberState::Returning.counts_toward_supply());
assert!(MemberState::Failed.is_failed());
assert!(!MemberState::Failed.counts_toward_supply());
}
#[test]
fn member_state_failed_implies_no_supply() {
for state in MemberState::ALL {
assert!(
!(state.is_failed() && state.counts_toward_supply()),
"{state:?} returns true from both is_failed and counts_toward_supply — \
a failed member can never be counted as available pool capacity",
);
}
}
#[test]
fn member_state_buckets_cover_every_variant() {
let mut supply = 0u32;
let mut failed = 0u32;
let mut in_use = 0u32;
for state in MemberState::ALL {
match (state.is_failed(), state.counts_toward_supply()) {
(true, false) => failed += 1,
(false, true) => supply += 1,
(false, false) => in_use += 1,
(true, true) => panic!("disjointness already pins this empty for {state:?}"),
}
}
assert_eq!(supply, 2, "supply bucket: Free + Spawning");
assert_eq!(failed, 1, "failed bucket: Failed");
assert_eq!(in_use, 2, "in-use bucket: Allocated + Returning");
assert_eq!(supply + failed + in_use, MemberState::ALL.len() as u32);
}
#[test]
fn pool_phase_is_well_formed_closed_set() {
tatara_lisp::assert_closed_set_well_formed::<PoolPhase>();
}
#[test]
fn pool_phase_as_str_matches_serde() {
for phase in PoolPhase::ALL {
let serialized = serde_json::to_string(&phase).expect("serialize");
let unquoted = serialized
.trim_start_matches('"')
.trim_end_matches('"')
.to_string();
assert_eq!(
unquoted,
phase.as_str(),
"as_str drift for {phase:?}: as_str={} serde={unquoted}",
phase.as_str()
);
}
}
#[test]
fn pool_phase_display_matches_as_str() {
for phase in PoolPhase::ALL {
assert_eq!(phase.to_string(), phase.as_str());
}
}
#[test]
fn unknown_pool_phase_errors() {
for bad in [
"steady",
"SCALINGUP",
"Scaling-Up",
"scaling_down",
"Free", "Replace", "Attested", "HoldFailed", ] {
let err = PoolPhase::from_str(bad).unwrap_err();
assert_eq!(err.0, bad, "error payload should echo input verbatim");
}
}
#[test]
fn pool_phase_predicate_truth_tables() {
assert!(!PoolPhase::Initializing.is_steady());
assert!(!PoolPhase::Initializing.is_terminal());
assert!(PoolPhase::Steady.is_steady());
assert!(!PoolPhase::Steady.is_terminal());
assert!(!PoolPhase::ScalingUp.is_steady());
assert!(!PoolPhase::ScalingUp.is_terminal());
assert!(!PoolPhase::ScalingDown.is_steady());
assert!(!PoolPhase::ScalingDown.is_terminal());
assert!(!PoolPhase::Degraded.is_steady());
assert!(!PoolPhase::Degraded.is_terminal());
assert!(!PoolPhase::Draining.is_steady());
assert!(PoolPhase::Draining.is_terminal());
}
#[test]
fn pool_phase_steady_excludes_terminal() {
for phase in PoolPhase::ALL {
assert!(
!(phase.is_steady() && phase.is_terminal()),
"{phase:?} returns true from both is_steady and is_terminal — \
a draining pool is by definition not the converged goal state",
);
}
}
#[test]
fn pool_phase_buckets_cover_every_variant() {
let mut converged = 0u32;
let mut terminal = 0u32;
let mut transient = 0u32;
for phase in PoolPhase::ALL {
match (phase.is_steady(), phase.is_terminal()) {
(true, false) => converged += 1,
(false, true) => terminal += 1,
(false, false) => transient += 1,
(true, true) => panic!("disjointness already pins this empty for {phase:?}"),
}
}
assert_eq!(converged, 1, "converged bucket: Steady");
assert_eq!(terminal, 1, "terminal bucket: Draining");
assert_eq!(
transient, 4,
"transient bucket: Initializing + ScalingUp + ScalingDown + Degraded"
);
assert_eq!(
converged + terminal + transient,
PoolPhase::ALL.len() as u32
);
}
#[test]
fn pool_phase_default_is_initializing_in_transient_bucket() {
let d = PoolPhase::default();
assert_eq!(d, PoolPhase::Initializing);
assert!(!d.is_steady());
assert!(!d.is_terminal());
}
}