use chrono::{DateTime, Utc};
use kube::CustomResource;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use crate::pool::AllocationRef;
#[derive(CustomResource, Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[kube(
group = "tatara.pleme.io",
version = "v1alpha1",
kind = "EphemeralAllocation",
plural = "ephemeralallocations",
shortname = "ealloc",
namespaced,
status = "AllocationStatus",
printcolumn = r#"{"name":"Pool","type":"string","jsonPath":".spec.poolRef.name"}"#,
printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#,
printcolumn = r#"{"name":"Process","type":"string","jsonPath":".status.assignedProcess.name"}"#,
printcolumn = r#"{"name":"Requestor","type":"string","jsonPath":".spec.requestor.kind"}"#,
printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
)]
#[serde(rename_all = "camelCase")]
pub struct AllocationSpec {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pool_ref: Option<AllocationRef>,
pub requestor: Requestor,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ttl: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub note: Option<String>,
}
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct Requestor {
pub kind: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub repo: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub branch: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pr_number: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub sha: Option<String>,
#[serde(default)]
pub pr_labels: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub actor: Option<String>,
}
impl Requestor {
#[must_use]
pub fn known_kind(&self) -> Option<RequestorKind> {
self.kind.parse().ok()
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, tatara_closed_set::DeriveClosedSet)]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum RequestorKind {
GithubPr,
Manual,
CiRun,
Scheduled,
}
impl RequestorKind {
pub const ALL: [Self; 4] = [Self::GithubPr, Self::Manual, Self::CiRun, Self::Scheduled];
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::GithubPr => "github-pr",
Self::Manual => "manual",
Self::CiRun => "ci-run",
Self::Scheduled => "scheduled",
}
}
}
impl From<RequestorKind> for String {
fn from(k: RequestorKind) -> Self {
k.as_str().to_owned()
}
}
impl From<RequestorKind> for &'static str {
fn from(k: RequestorKind) -> Self {
k.as_str()
}
}
#[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct AllocationStatus {
#[serde(default)]
pub phase: AllocationPhase,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub phase_since: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bound_pool: Option<AllocationRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub assigned_process: Option<AllocationRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub allocated_at: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub expires_at: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message: Option<String>,
#[serde(default)]
pub conditions: Vec<AllocationCondition>,
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
Eq,
Hash,
Serialize,
Deserialize,
JsonSchema,
tatara_closed_set::DeriveClosedSet,
)]
#[serde(rename_all = "PascalCase")]
#[closed_set(via = "as_str", generate_unknown, display)]
pub enum AllocationPhase {
Pending,
Queued,
Bound,
Releasing,
Released,
NoMatchingPool,
Failed,
}
impl Default for AllocationPhase {
fn default() -> Self {
Self::Pending
}
}
impl AllocationPhase {
pub const ALL: [Self; 7] = [
Self::Pending,
Self::Queued,
Self::Bound,
Self::Releasing,
Self::Released,
Self::NoMatchingPool,
Self::Failed,
];
pub const fn as_str(self) -> &'static str {
match self {
Self::Pending => "Pending",
Self::Queued => "Queued",
Self::Bound => "Bound",
Self::Releasing => "Releasing",
Self::Released => "Released",
Self::NoMatchingPool => "NoMatchingPool",
Self::Failed => "Failed",
}
}
pub const fn is_terminal(self) -> bool {
match self {
Self::Released | Self::Failed => true,
Self::Pending | Self::Queued | Self::Bound | Self::Releasing | Self::NoMatchingPool => {
false
}
}
}
pub const fn needs_pool_routing(self) -> bool {
match self {
Self::Pending | Self::Queued | Self::NoMatchingPool => true,
Self::Bound | Self::Releasing | Self::Released | Self::Failed => false,
}
}
}
#[derive(Clone, Debug, Deserialize, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct AllocationCondition {
pub type_: String,
pub status: String,
pub reason: String,
pub message: String,
pub last_transition_time: DateTime<Utc>,
}
impl EphemeralAllocation {
#[must_use]
pub fn observed_phase(&self) -> Option<AllocationPhase> {
self.status.as_ref().map(|s| s.phase)
}
#[must_use]
pub fn observed_phase_or_pending(&self) -> AllocationPhase {
self.observed_phase().unwrap_or(AllocationPhase::Pending)
}
}
#[cfg(test)]
mod tests {
use std::str::FromStr;
use super::*;
#[test]
fn requestor_minimum_shape_round_trips() {
let r = Requestor {
kind: "github-pr".into(),
repo: Some("pleme-io/demo-app".into()),
branch: Some("fix-something".into()),
pr_number: Some(123),
sha: Some("abc123def".into()),
pr_labels: vec!["needs-ephemeral".into()],
actor: Some("drzln".into()),
};
let yaml = serde_yaml::to_string(&r).unwrap();
assert!(yaml.contains("kind: github-pr"));
assert!(yaml.contains("prNumber: 123"));
let back: Requestor = serde_yaml::from_str(&yaml).unwrap();
assert_eq!(back.kind, "github-pr");
assert_eq!(back.pr_number, Some(123));
}
#[test]
fn allocation_status_defaults_pending() {
let s = AllocationStatus::default();
assert_eq!(s.phase, AllocationPhase::Pending);
assert!(s.bound_pool.is_none());
assert!(s.assigned_process.is_none());
}
#[test]
fn allocation_phase_round_trips_via_serde() {
for p in [
AllocationPhase::Pending,
AllocationPhase::Queued,
AllocationPhase::Bound,
AllocationPhase::Releasing,
AllocationPhase::Released,
AllocationPhase::NoMatchingPool,
AllocationPhase::Failed,
] {
let s = serde_yaml::to_string(&p).unwrap();
let back: AllocationPhase = serde_yaml::from_str(&s).unwrap();
assert_eq!(back, p);
}
}
#[test]
fn allocation_phase_is_well_formed_closed_set() {
tatara_closed_set::assert_closed_set_well_formed::<AllocationPhase>();
}
#[test]
fn allocation_phase_as_str_matches_serde() {
crate::tagged_union::assert_label_matches_serde_serialization::<AllocationPhase>();
}
#[test]
fn allocation_phase_display_matches_as_str() {
crate::tagged_union::assert_display_matches_label::<AllocationPhase>();
}
#[test]
fn unknown_allocation_phase_errors() {
for bad in [
"pending",
"BOUND",
"no-matching-pool",
"release",
"failed_state",
"Reaped",
] {
let err = AllocationPhase::from_str(bad).unwrap_err();
assert_eq!(err.0, bad, "error payload should echo input verbatim");
}
}
#[test]
fn allocation_phase_predicate_truth_tables() {
assert!(!AllocationPhase::Pending.is_terminal());
assert!(AllocationPhase::Pending.needs_pool_routing());
assert!(!AllocationPhase::Queued.is_terminal());
assert!(AllocationPhase::Queued.needs_pool_routing());
assert!(!AllocationPhase::Bound.is_terminal());
assert!(!AllocationPhase::Bound.needs_pool_routing());
assert!(!AllocationPhase::Releasing.is_terminal());
assert!(!AllocationPhase::Releasing.needs_pool_routing());
assert!(AllocationPhase::Released.is_terminal());
assert!(!AllocationPhase::Released.needs_pool_routing());
assert!(!AllocationPhase::NoMatchingPool.is_terminal());
assert!(AllocationPhase::NoMatchingPool.needs_pool_routing());
assert!(AllocationPhase::Failed.is_terminal());
assert!(!AllocationPhase::Failed.needs_pool_routing());
}
#[test]
fn allocation_phase_terminal_excludes_routing() {
for phase in AllocationPhase::ALL {
assert!(
!(phase.is_terminal() && phase.needs_pool_routing()),
"{phase:?} is both terminal and routing-eligible",
);
}
}
#[test]
fn allocation_phase_default_is_pending_and_routes() {
let d = AllocationPhase::default();
assert_eq!(d, AllocationPhase::Pending);
assert!(!d.is_terminal());
assert!(d.needs_pool_routing());
}
#[test]
fn requestor_kind_is_well_formed_closed_set() {
tatara_closed_set::assert_closed_set_well_formed::<RequestorKind>();
}
#[test]
fn requestor_kind_canonical_names_pinned() {
assert_eq!(RequestorKind::GithubPr.as_str(), "github-pr");
assert_eq!(RequestorKind::Manual.as_str(), "manual");
assert_eq!(RequestorKind::CiRun.as_str(), "ci-run");
assert_eq!(RequestorKind::Scheduled.as_str(), "scheduled");
}
#[test]
fn requestor_kind_from_str_rejects_open_kinds() {
for bad in [
"github_pr",
"GithubPr",
"operator-custom-kind",
"ci_run",
"Scheduled",
] {
let err = bad.parse::<RequestorKind>().unwrap_err();
assert_eq!(err, UnknownRequestorKind(bad.to_string()));
}
}
#[test]
fn requestor_kind_display_delegates_to_as_str() {
for k in RequestorKind::ALL {
assert_eq!(format!("{k}"), k.as_str());
}
}
#[test]
fn requestor_kind_into_string_matches_as_str() {
for k in RequestorKind::ALL {
let s: String = k.into();
assert_eq!(s, k.as_str());
}
}
#[test]
fn known_kind_decodes_built_requestors() {
for k in RequestorKind::ALL {
let r = Requestor {
kind: k.into(),
repo: None,
branch: None,
pr_number: None,
sha: None,
pr_labels: vec![],
actor: None,
};
assert_eq!(r.known_kind(), Some(k), "round-trip failed for {k:?}");
}
}
#[test]
fn known_kind_returns_none_for_open_kinds() {
let r = Requestor {
kind: "operator-custom-kind".into(),
repo: None,
branch: None,
pr_number: None,
sha: None,
pr_labels: vec![],
actor: None,
};
assert_eq!(r.known_kind(), None);
}
#[test]
fn requestor_kind_matches_existing_fixture_literals() {
assert_eq!(RequestorKind::GithubPr.as_str(), "github-pr");
assert_eq!(RequestorKind::Manual.as_str(), "manual");
}
fn alloc_with_phase(phase: AllocationPhase) -> EphemeralAllocation {
let spec = AllocationSpec {
pool_ref: None,
requestor: Requestor {
kind: "manual".into(),
repo: None,
branch: None,
pr_number: None,
sha: None,
pr_labels: vec![],
actor: None,
},
ttl: None,
note: None,
};
let mut a = EphemeralAllocation::new("obs-alloc", spec);
a.status = Some(AllocationStatus {
phase,
..AllocationStatus::default()
});
a
}
fn alloc_without_status() -> EphemeralAllocation {
let spec = AllocationSpec {
pool_ref: None,
requestor: Requestor {
kind: "manual".into(),
repo: None,
branch: None,
pr_number: None,
sha: None,
pr_labels: vec![],
actor: None,
},
ttl: None,
note: None,
};
let mut a = EphemeralAllocation::new("no-status-alloc", spec);
a.status = None;
a
}
#[test]
fn observed_phase_returns_none_when_status_is_none() {
let a = alloc_without_status();
assert!(a.observed_phase().is_none());
}
#[test]
fn observed_phase_returns_populated_variant_verbatim() {
for p in AllocationPhase::ALL {
let a = alloc_with_phase(p);
assert_eq!(
a.observed_phase(),
Some(p),
"observed_phase must project the persisted variant verbatim for {p:?}"
);
}
}
#[test]
fn observed_phase_matches_pre_lift_chain_bytewise() {
let none_alloc = alloc_without_status();
assert_eq!(
none_alloc.observed_phase(),
none_alloc.status.as_ref().map(|s| s.phase),
);
for p in AllocationPhase::ALL {
let a = alloc_with_phase(p);
assert_eq!(
a.observed_phase(),
a.status.as_ref().map(|s| s.phase),
"primitive must be byte-identical to the pre-lift chain for {p:?}",
);
}
}
#[test]
fn observed_phase_or_pending_defaults_to_pending_when_status_absent() {
let a = alloc_without_status();
assert_eq!(a.observed_phase_or_pending(), AllocationPhase::Pending);
}
#[test]
fn observed_phase_or_pending_returns_populated_phase_verbatim() {
for p in AllocationPhase::ALL {
let a = alloc_with_phase(p);
assert_eq!(
a.observed_phase_or_pending(),
p,
"populated status must pass through verbatim for {p:?}"
);
}
}
#[test]
fn observed_phase_or_pending_defaults_agree_with_allocation_phase_default() {
let a = alloc_without_status();
assert_eq!(a.observed_phase_or_pending(), AllocationPhase::default());
}
#[test]
fn observed_phase_or_pending_matches_pre_lift_chain_bytewise() {
let none_alloc = alloc_without_status();
assert_eq!(
none_alloc.observed_phase_or_pending(),
none_alloc
.status
.as_ref()
.map(|s| s.phase)
.unwrap_or(AllocationPhase::Pending),
);
for p in AllocationPhase::ALL {
let a = alloc_with_phase(p);
assert_eq!(
a.observed_phase_or_pending(),
a.status
.as_ref()
.map(|s| s.phase)
.unwrap_or(AllocationPhase::Pending),
"primitive must be byte-identical to the pre-lift 5-line chain for {p:?}",
);
}
}
#[test]
fn observed_phase_or_pending_composes_from_observed_phase() {
let none_alloc = alloc_without_status();
assert_eq!(
none_alloc.observed_phase_or_pending(),
none_alloc
.observed_phase()
.unwrap_or(AllocationPhase::Pending),
);
for p in AllocationPhase::ALL {
let a = alloc_with_phase(p);
assert_eq!(
a.observed_phase_or_pending(),
a.observed_phase().unwrap_or(AllocationPhase::Pending),
"composer must ride on top of the borrow-form projection for {p:?}",
);
}
}
#[test]
fn observed_phase_is_a_pure_projection() {
let a = alloc_with_phase(AllocationPhase::Bound);
let one = a.observed_phase();
let two = a.observed_phase();
assert_eq!(one, two);
assert!(a.status.is_some(), "projection must not consume the status");
}
#[test]
fn observed_phase_pending_missing_status_and_populated_pending_collapse_to_same_composer_output(
) {
let none_alloc = alloc_without_status();
let pending_alloc = alloc_with_phase(AllocationPhase::Pending);
assert_eq!(
none_alloc.observed_phase_or_pending(),
pending_alloc.observed_phase_or_pending(),
);
assert_ne!(
none_alloc.observed_phase(),
pending_alloc.observed_phase(),
"borrow-form accessor MUST distinguish missing-status from populated-Pending",
);
}
#[test]
fn observed_phase_or_pending_missing_status_sink_agrees_with_process_peer_shape() {
let no_status_alloc = alloc_without_status();
assert_eq!(
no_status_alloc.observed_phase_or_pending(),
AllocationPhase::default(),
);
assert_eq!(AllocationPhase::default(), AllocationPhase::Pending,);
assert_eq!(
crate::phase::ProcessPhase::default(),
crate::phase::ProcessPhase::Pending,
);
}
#[test]
fn allocation_spec_omits_optional_fields() {
let s = AllocationSpec {
pool_ref: None,
requestor: Requestor {
kind: "manual".into(),
repo: None,
branch: None,
pr_number: None,
sha: None,
pr_labels: vec![],
actor: None,
},
ttl: None,
note: None,
};
let yaml = serde_yaml::to_string(&s).unwrap();
assert!(!yaml.contains("poolRef"));
assert!(!yaml.contains("ttl"));
assert!(!yaml.contains("note"));
}
}