use super::secret::{SECRET_TYPE_LABEL, SecretExt, SecretType};
use super::statefulset::StatefulSetExt;
use crate::kanidm::controller::context::Context;
use crate::kanidm::crd::{
DomainAppearanceImageStatus, Kanidm, KanidmReplicaState, KanidmReplicaStatus, KanidmStatus,
KanidmUpgradeCheckResult, KanidmVersionStatus, VersionCompatibilityResult,
};
use crate::version;
use kaniop_k8s_util::error::{Error, Result};
use std::sync::Arc;
use std::time::Duration;
use futures::future::join_all;
use k8s_openapi::api::apps::v1::{StatefulSet, StatefulSetStatus};
use k8s_openapi::api::core::v1::Secret;
use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
use k8s_openapi::jiff::Timestamp;
use kaniop_k8s_util::resources::get_image_tag;
use kube::Resource;
use kube::ResourceExt;
use kube::api::{Api, Patch, PatchParams};
use kube::runtime::events::{Event, EventType};
use kube::runtime::reflector::ObjectRef;
use tokio::time::sleep;
use tracing::{debug, trace, warn};
const TYPE_AVAILABLE: &str = "Available";
const TYPE_PROGRESSING: &str = "Progressing";
const TYPE_INITIALIZED: &str = "Initialized";
const TYPE_REPLICA_FAILURE: &str = "ReplicaFailure";
const CONDITION_TRUE: &str = "True";
const CONDITION_FALSE: &str = "False";
#[allow(async_fn_in_trait)]
pub trait StatusExt {
async fn update_status(&self, ctx: Arc<Context>) -> Result<KanidmStatus>;
}
impl StatusExt for Kanidm {
async fn update_status(&self, ctx: Arc<Context>) -> Result<KanidmStatus> {
let name = &self.name_any();
async fn publish_incompatible_version_event(
ctx: &Arc<Context>,
kanidm: &Kanidm,
desired: &str,
) -> VersionCompatibilityResult {
let _ignore_error = ctx
.kaniop_ctx
.recorder
.publish(
&Event {
type_: EventType::Warning,
reason: "VersionIncompatible".to_string(),
note: Some(format!(
"Kanidm image version {} is not compatible with this operator. The operator uses Kanidm client SDK v{}. Either use a compatible image version or set spec.disableUpgradeChecks: true (not recommended).",
desired,
version::KANIDM_CLIENT_VERSION
)),
action: "VersionCheck".to_string(),
secondary: None,
},
&kanidm.object_ref(&()),
)
.await
.map_err(|e| {
warn!(%e, "failed to publish VersionIncompatible event");
Error::KubeError("failed to publish event".to_string(), Box::new(e))
});
VersionCompatibilityResult::Incompatible
}
let namespace = &self.get_namespace();
let statefulsets = self
.spec
.replica_groups
.iter()
.filter_map(|rg| {
let sts_name = self.statefulset_name(&rg.name);
let sts_ref = ObjectRef::<StatefulSet>::new_with(&sts_name, ()).within(namespace);
ctx.stores.stateful_set_store.get(&sts_ref)
})
.collect::<Vec<Arc<StatefulSet>>>();
let sts_status = statefulsets
.iter()
.map(|sts| sts.status.clone())
.collect::<Vec<Option<StatefulSetStatus>>>();
let secret_ref = ObjectRef::<Secret>::new_with(&self.admins_secret_name(), ())
.within(&self.get_namespace());
let admin_secret = ctx
.stores
.secret_store
.get(&secret_ref)
.map(|s| s.name_any());
let replica_infos_futures: Vec<_> = statefulsets
.iter()
.flat_map(|sts| {
let replicas = sts.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0);
let secret_store = ctx.stores.secret_store.clone();
let ctx_clone = ctx.clone();
let namespace = namespace.to_string();
(0..replicas).map(move |i| {
let sts_name = sts.name_any();
let secret_store = secret_store.clone();
let ctx = ctx_clone.clone();
let namespace = namespace.clone();
let pod_name = format!("{sts_name}-{i}");
let pod_env_prefix = self.pod_env_prefix(&pod_name);
let host_env = format!("{pod_env_prefix}_HOST");
let replication_host = sts.spec.as_ref().and_then(|s| s.template.spec.as_ref().and_then(|t_s| t_s.init_containers.as_ref().and_then(|c| c.first().and_then(|f_c| f_c.env.as_ref().and_then(|env|
env.iter().find(|e| e.name == host_env).and_then(|e| e.value.clone()))))));
let secret_name = self.replica_secret_name(&pod_name);
let replica_cert_label = serde_plain::to_string(&SecretType::ReplicaCert)
.expect("replica cert secret type must serialize");
let secret_ref =
ObjectRef::<Secret>::new_with(&secret_name, ()).within(&namespace);
async move {
let replica_secret = secret_store.get(&secret_ref);
if let Some(secret) = replica_secret.as_deref() {
let has_replica_cert_label = secret
.metadata
.labels
.as_ref()
.is_some_and(|labels| {
labels.get(SECRET_TYPE_LABEL) == Some(&replica_cert_label)
});
if !has_replica_cert_label {
warn!(namespace, name = secret_name, "replica certificate secret is missing the required replica-cert label and will be ignored");
} else if let Err(e) = ctx.insert_repl_cert_exp(secret).await {
warn!(namespace, name = secret_name, %e, "failed to parse replica certificate secret, automatic certificate renewal may be affected");
}
}
let is_certificate_expiring =
ctx.get_repl_cert_exp(&secret_ref).await.map(|exp| {
let now = Timestamp::now().as_second();
let threshold = 30 * 24 * 60 * 60;
trace!("replica cert expiration {exp}, now {now}, threshold {threshold}");
exp - now < threshold
});
let is_certificate_host_valid = ctx.get_repl_cert_host(&secret_ref).await.map(|h| {
let matches = Some(h.clone()) == replication_host;
trace!("replica cert host {h}, expected host {:?}, matches {matches}", replication_host);
matches
});
ReplicaInformation {
pod_name,
statefulset_name: sts_name.clone(),
replica_secret_exists: replica_secret.is_some(),
is_certificate_expiring,
is_certificate_host_valid,
}
}
})
})
.collect();
let replica_infos = join_all(replica_infos_futures).await;
let version = if !self.spec.disable_upgrade_checks {
let running_image_tag = statefulsets.iter().find_map(|sts| {
sts.spec.as_ref().and_then(|s| {
s.template.spec.as_ref().and_then(|t| {
t.containers
.first()
.and_then(|c| c.image.as_ref().and_then(|i| get_image_tag(i)))
})
})
});
let desired_image_tag = get_image_tag(&self.spec.image);
match running_image_tag {
Some(tag) => {
let upgrade_check = self.run_upgrade_pre_check(ctx.clone()).await;
let compatibility_result = match desired_image_tag {
Some(desired) if !version::is_version_compatible(&desired) => {
publish_incompatible_version_event(&ctx, self, &desired).await
}
_ => VersionCompatibilityResult::Compatible,
};
Some(KanidmVersionStatus {
image_tag: tag,
upgrade_check_result: upgrade_check,
compatibility_result,
})
}
None => match desired_image_tag {
Some(desired) => {
let compatibility_result = if !version::is_version_compatible(&desired) {
publish_incompatible_version_event(&ctx, self, &desired).await
} else {
VersionCompatibilityResult::Compatible
};
Some(KanidmVersionStatus {
image_tag: desired,
upgrade_check_result: KanidmUpgradeCheckResult::Passed,
compatibility_result,
})
}
None => None,
},
}
} else {
debug!("upgrade checks are disabled");
None
};
let kanidm_api = Api::<Kanidm>::namespaced(ctx.kaniop_ctx.client.clone(), namespace);
let current_kanidm = kanidm_api.get(name).await.map_err(|e| {
Error::KubeError(
format!("failed to get current Kanidm {namespace}/{name}"),
Box::new(e),
)
})?;
let domain_appearance_image = if current_kanidm
.spec
.domain_appearance
.as_ref()
.and_then(|da| da.image.as_ref())
.is_some()
{
current_kanidm
.status
.as_ref()
.and_then(|s| s.domain_appearance_image.clone())
} else {
None
};
let new_status = generate_status(
self.status
.as_ref()
.cloned()
.unwrap_or_default()
.conditions
.unwrap_or_default(),
&sts_status,
admin_secret,
replica_infos,
self.is_replication_enabled(),
current_kanidm.metadata.generation,
version,
domain_appearance_image,
);
let new_status_patch = serde_json::json!({
"status": new_status.clone()
});
debug!("updating Kanidm status");
trace!("new status {:?}", new_status_patch);
let patch = PatchParams::default();
let _o = kanidm_api
.patch_status(name, &patch, &Patch::Merge(&new_status_patch))
.await
.map_err(|e| {
Error::KubeError(
format!("failed to patch Kanidm/status {namespace}/{name}"),
Box::new(e),
)
})?;
Ok(new_status)
}
}
impl Kanidm {
async fn run_upgrade_pre_check(&self, ctx: Arc<Context>) -> KanidmUpgradeCheckResult {
let upgrade_check = vec!["kanidmd", "domain", "upgrade-check"];
debug!("running kanidmd domain upgrade-check");
const MAX_RETRIES: u32 = 5;
const RETRY_DELAY_MS: u64 = 2000;
for attempt in 0..MAX_RETRIES {
let result = self.exec_any(ctx.clone(), upgrade_check.clone()).await;
match result {
Ok(r) => {
debug!("kanidmd domain upgrade-check passed: {:?}", r);
return KanidmUpgradeCheckResult::Passed;
}
Err(e) => {
debug!(%e, "kanidmd domain upgrade-check failed (attempt {})", attempt + 1);
if attempt < MAX_RETRIES - 1 {
trace!(%e, "kanidmd domain upgrade-check failed, retrying");
sleep(Duration::from_millis(RETRY_DELAY_MS)).await;
} else {
match e {
Error::KubeExecError(e_msg) => {
warn!(%e_msg, "`kanidmd domain upgrade-check` failed");
}
_ => {
warn!(%e, "`kanidmd domain upgrade-check` failed after retries");
}
}
let _ignore_error = ctx
.kaniop_ctx
.recorder
.publish(
&Event {
type_: EventType::Warning,
reason: "UpgradeCheckFailed".to_string(),
note: Some("`kanidmd domain upgrade-check` failed. See kanidm operator logs for details.".to_string()),
action: "UpgradeCheck".to_string(),
secondary: None,
},
&self.object_ref(&()),
)
.await
.map_err(|e| {
warn!(%e, "failed to publish KanidmError event");
Error::KubeError("failed to publish event".to_string(), Box::new(e))
});
}
}
}
}
KanidmUpgradeCheckResult::Failed
}
}
struct ReplicaInformation {
pod_name: String,
statefulset_name: String,
replica_secret_exists: bool,
is_certificate_expiring: Option<bool>,
is_certificate_host_valid: Option<bool>,
}
pub fn is_kanidm_available(status: KanidmStatus) -> bool {
status
.conditions
.unwrap_or_default()
.iter()
.any(|c| c.type_ == TYPE_AVAILABLE && c.status == CONDITION_TRUE)
}
pub fn is_kanidm_initialized(status: KanidmStatus) -> bool {
status
.conditions
.unwrap_or_default()
.iter()
.any(|c| c.type_ == TYPE_INITIALIZED && c.status == CONDITION_TRUE)
}
#[allow(clippy::too_many_arguments)]
fn generate_status(
previous_conditions: Vec<Condition>,
statefulset_statuses: &[Option<StatefulSetStatus>],
secret_name: Option<String>,
replica_infos: Vec<ReplicaInformation>,
is_replication_enabled: bool,
kanidm_generation: Option<i64>,
version: Option<KanidmVersionStatus>,
domain_appearance_image: Option<DomainAppearanceImageStatus>,
) -> KanidmStatus {
let available_replicas = statefulset_statuses
.iter()
.filter_map(|sts| sts.as_ref())
.map(|sts| sts.available_replicas.unwrap_or(0))
.sum();
let replicas = statefulset_statuses
.iter()
.filter_map(|sts| sts.as_ref())
.map(|sts| sts.replicas)
.sum();
let replica_statuses = replica_infos
.iter()
.map(|ri| KanidmReplicaStatus {
pod_name: ri.pod_name.clone(),
statefulset_name: ri.statefulset_name.clone(),
state: if ri.replica_secret_exists || !is_replication_enabled {
if ri.is_certificate_expiring == Some(true) {
debug!("replica cert is expiring for pod {}", ri.pod_name);
KanidmReplicaState::CertificateExpiring
} else if ri.is_certificate_host_valid == Some(false) {
debug!("replica cert host is invalid for pod {}", ri.pod_name);
KanidmReplicaState::CertificateHostInvalid
} else {
KanidmReplicaState::Ready
}
} else {
KanidmReplicaState::Pending
},
})
.collect::<Vec<KanidmReplicaStatus>>();
let new_conditions = generate_status_conditions(
previous_conditions,
statefulset_statuses,
secret_name.is_some(),
&replica_statuses,
kanidm_generation,
);
let replica_column = format!("{available_replicas}/{replicas}");
KanidmStatus {
conditions: Some(new_conditions),
available_replicas,
replicas,
unavailable_replicas: replicas - available_replicas,
updated_replicas: statefulset_statuses
.iter()
.filter_map(|sts| sts.as_ref())
.map(|sts| sts.updated_replicas.unwrap_or(0))
.sum(),
replica_statuses,
replica_column,
secret_name,
version,
domain_appearance_image,
mail_sender: None,
}
}
fn generate_status_conditions(
previous_conditions: Vec<Condition>,
statefulset_statuses: &[Option<StatefulSetStatus>],
secret_exists: bool,
replica_statuses: &[KanidmReplicaStatus],
kanidm_generation: Option<i64>,
) -> Vec<Condition> {
let sts_statuses = statefulset_statuses.iter().filter_map(|sts| sts.as_ref());
let available_condition = match sts_statuses
.clone()
.any(|s| s.available_replicas >= Some(1))
{
true => Condition {
type_: TYPE_AVAILABLE.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "ReplicaReady".to_string(),
message: "At least one replica is ready.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
},
false => Condition {
type_: TYPE_AVAILABLE.to_string(),
status: CONDITION_FALSE.to_string(),
reason: "NoReplicaReady".to_string(),
message: "No replicas are ready.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
},
};
let initialized_condition = match secret_exists {
true => Condition {
type_: TYPE_INITIALIZED.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "AdminSecretExists".to_string(),
message: "admin and idm_admin passwords have been generated.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
},
false => Condition {
type_: TYPE_INITIALIZED.to_string(),
status: CONDITION_FALSE.to_string(),
reason: "AdminSecretNotExists".to_string(),
message: "admin and idm_admin passwords have not been generated.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
},
};
let replicate_failure_condition = match sts_statuses.clone().any(|s| {
s.conditions
.as_ref()
.map(|conditions| {
conditions
.iter()
.any(|c| c.type_ == TYPE_REPLICA_FAILURE && c.status == CONDITION_TRUE)
})
.unwrap_or(false)
}) {
true => Condition {
type_: TYPE_REPLICA_FAILURE.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "ReplicaCreationFailure".to_string(),
message: "Failed to create or delete replicas.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
},
false => Condition {
type_: TYPE_REPLICA_FAILURE.to_string(),
status: CONDITION_FALSE.to_string(),
reason: "NoReplicaFailure".to_string(),
message: "No replica creation or deletion failures.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
},
};
let previous_observed_generation = previous_conditions
.iter()
.find_map(|c| c.observed_generation);
let generation_changed =
previous_observed_generation.is_some() && previous_observed_generation != kanidm_generation;
let progressing_condition = if generation_changed {
Condition {
type_: TYPE_PROGRESSING.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "SpecChanged".to_string(),
message: "Kanidm spec has changed.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
}
} else if sts_statuses
.clone()
.any(|s| s.current_revision != s.update_revision)
{
Condition {
type_: TYPE_PROGRESSING.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "RollingUpdateInProgress".to_string(),
message: "StatefulSet rolling update is in progress.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
}
} else if sts_statuses.clone().any(|s| {
s.conditions
.as_ref()
.map(|conditions| {
conditions
.iter()
.any(|c| c.type_ == TYPE_PROGRESSING && c.status == CONDITION_TRUE)
})
.unwrap_or(false)
}) {
Condition {
type_: TYPE_PROGRESSING.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "Progressing".to_string(),
message: "StatefulSet is progressing.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
}
} else if replica_statuses
.iter()
.any(|rs| rs.state != KanidmReplicaState::Ready)
{
Condition {
type_: TYPE_PROGRESSING.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "ReplicaStatusPending".to_string(),
message: "At least one replica is not ready.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
}
} else if sts_statuses
.clone()
.any(|s| s.replicas > s.available_replicas.unwrap_or(0))
{
Condition {
type_: TYPE_PROGRESSING.to_string(),
status: CONDITION_TRUE.to_string(),
reason: "ReplicaCreation".to_string(),
message: "Replicas are being created.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
}
} else {
Condition {
type_: TYPE_PROGRESSING.to_string(),
status: CONDITION_FALSE.to_string(),
reason: "NotProgressing".to_string(),
message: "StatefulSet is not progressing.".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: kanidm_generation,
}
};
[
available_condition,
progressing_condition,
initialized_condition,
replicate_failure_condition,
]
.into_iter()
.fold(previous_conditions, |previous_conditions, c| {
update_conditions(previous_conditions, &c)
})
}
fn update_conditions(
previous_conditions: Vec<Condition>,
new_condition: &Condition,
) -> Vec<Condition> {
if previous_conditions
.iter()
.any(|c| c.type_ == *new_condition.type_)
{
previous_conditions
.iter()
.filter(|c| c.type_ != *new_condition.type_)
.cloned()
.chain(std::iter::once(new_condition.clone()))
.collect()
} else {
previous_conditions
.iter()
.cloned()
.chain(std::iter::once(new_condition.clone()))
.collect()
}
}
#[cfg(test)]
mod test {
use super::*;
fn create_condition(type_: &str, status: &str) -> Condition {
Condition {
type_: type_.to_string(),
status: status.to_string(),
reason: "".to_string(),
message: "".to_string(),
last_transition_time: Time(Timestamp::now()),
observed_generation: None,
}
}
#[test]
fn test_update_conditions_with_existing_status_type() {
let previous_conditions = vec![
create_condition(TYPE_AVAILABLE, CONDITION_TRUE),
create_condition(TYPE_PROGRESSING, CONDITION_FALSE),
];
let new_condition = create_condition(TYPE_AVAILABLE, CONDITION_FALSE);
let updated_conditions = update_conditions(previous_conditions.clone(), &new_condition);
assert_eq!(updated_conditions.len(), 2);
assert!(
updated_conditions
.iter()
.any(|c| c.type_ == TYPE_AVAILABLE && c.status == CONDITION_FALSE)
);
assert!(
updated_conditions
.iter()
.any(|c| c.type_ == TYPE_PROGRESSING && c.status == CONDITION_FALSE)
);
}
#[test]
fn test_update_conditions_without_existing_status_type() {
let previous_conditions = vec![create_condition(TYPE_PROGRESSING, CONDITION_FALSE)];
let new_condition = create_condition(TYPE_AVAILABLE, CONDITION_TRUE);
let updated_conditions = update_conditions(previous_conditions.clone(), &new_condition);
assert_eq!(updated_conditions.len(), 2);
assert!(
updated_conditions
.iter()
.any(|c| c.type_ == TYPE_AVAILABLE && c.status == CONDITION_TRUE)
);
assert!(
updated_conditions
.iter()
.any(|c| c.type_ == TYPE_PROGRESSING && c.status == CONDITION_FALSE)
);
}
#[test]
fn test_update_conditions_with_empty_previous_conditions() {
let previous_conditions = vec![];
let new_condition = create_condition(TYPE_AVAILABLE, CONDITION_TRUE);
let updated_conditions = update_conditions(previous_conditions.clone(), &new_condition);
assert_eq!(updated_conditions.len(), 1);
assert!(
updated_conditions
.iter()
.any(|c| c.type_ == TYPE_AVAILABLE && c.status == CONDITION_TRUE)
);
}
}