#![allow(clippy::missing_errors_doc)] #![allow(clippy::must_use_candidate)] #![allow(clippy::return_self_not_must_use)] #![allow(clippy::type_complexity)]
#[cfg(feature = "derive")]
#[doc(inline)]
pub use kube_condition_derive::StatusCondition;
use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition;
use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
use k8s_openapi::chrono::Utc;
use kube::api::{Api, Patch, PatchParams};
use kube::runtime::controller::Action;
use kube::{Resource, ResourceExt};
use serde::de::DeserializeOwned;
use serde::Serialize;
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, error, info, warn};
pub const CONDITION_TYPE_READY: &str = "Ready";
pub const CONDITION_TYPE_PROGRESSING: &str = "Progressing";
pub const CONDITION_TYPE_DEGRADED: &str = "Degraded";
pub const CONDITION_TYPE_AVAILABLE: &str = "Available";
pub const CONDITION_STATUS_TRUE: &str = "True";
pub const CONDITION_STATUS_FALSE: &str = "False";
pub const CONDITION_STATUS_UNKNOWN: &str = "Unknown";
pub const CONDITION_REASON_RECONCILE_SUCCEEDED: &str = "ReconcileSucceeded";
pub const CONDITION_REASON_PROGRESSING: &str = "Progressing";
pub const DEFAULT_REQUEUE_SECONDS: u64 = 30;
pub const DEFAULT_NON_RETRYABLE_REQUEUE_SECONDS: u64 = 300;
pub const FIELD_MANAGER_NAME: &str = "kube-condition";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Severity {
Info,
Warning,
Error,
}
impl Severity {
pub fn as_str(&self) -> &'static str {
match self {
Severity::Info => "Info",
Severity::Warning => "Warning",
Severity::Error => "Error",
}
}
}
#[derive(Debug, Clone)]
pub struct ConditionInfo {
pub type_: String,
pub status: String,
pub reason: String,
pub message: String,
}
impl ConditionInfo {
pub fn into_condition(self, observed_generation: Option<i64>) -> Condition {
Condition {
type_: self.type_,
status: self.status,
reason: self.reason,
message: self.message,
last_transition_time: Time(Utc::now()),
observed_generation,
}
}
}
pub trait StatusCondition: std::error::Error {
fn to_condition_info(&self) -> ConditionInfo;
fn severity(&self) -> Severity;
fn is_retryable(&self) -> bool;
fn requeue_duration(&self) -> Duration;
fn to_condition(&self, observed_generation: Option<i64>) -> Condition {
self.to_condition_info().into_condition(observed_generation)
}
fn to_action(&self) -> Action {
if self.is_retryable() {
Action::requeue(self.requeue_duration())
} else {
Action::requeue(Duration::from_secs(DEFAULT_NON_RETRYABLE_REQUEUE_SECONDS))
}
}
}
pub fn ready_condition(ready: bool, message: impl Into<String>) -> ConditionInfo {
ConditionInfo {
type_: CONDITION_TYPE_READY.to_string(),
status: if ready { CONDITION_STATUS_TRUE } else { CONDITION_STATUS_FALSE }.to_string(),
reason: if ready { CONDITION_REASON_RECONCILE_SUCCEEDED } else { "NotReady" }.to_string(),
message: message.into(),
}
}
pub fn error_condition(reason: impl Into<String>, message: impl Into<String>) -> ConditionInfo {
ConditionInfo {
type_: CONDITION_TYPE_READY.to_string(),
status: CONDITION_STATUS_FALSE.to_string(),
reason: reason.into(),
message: message.into(),
}
}
#[derive(Debug, Clone)]
pub struct ReadyCondition;
impl ReadyCondition {
pub fn success(message: impl Into<String>) -> ConditionInfo {
ConditionInfo {
type_: CONDITION_TYPE_READY.to_string(),
status: CONDITION_STATUS_TRUE.to_string(),
reason: CONDITION_REASON_RECONCILE_SUCCEEDED.to_string(),
message: message.into(),
}
}
pub fn progressing(message: impl Into<String>) -> ConditionInfo {
ConditionInfo {
type_: CONDITION_TYPE_READY.to_string(),
status: CONDITION_STATUS_FALSE.to_string(),
reason: CONDITION_REASON_PROGRESSING.to_string(),
message: message.into(),
}
}
}
#[derive(Debug, Clone)]
pub struct ConditionBuilder {
type_: String,
status: String,
reason: String,
message: String,
}
impl ConditionBuilder {
pub fn new(type_: impl Into<String>) -> Self {
Self {
type_: type_.into(),
status: CONDITION_STATUS_UNKNOWN.to_string(),
reason: "Unknown".to_string(),
message: String::new(),
}
}
pub fn status(mut self, status: impl Into<String>) -> Self {
self.status = status.into();
self
}
pub fn success(self) -> Self {
self.status(CONDITION_STATUS_TRUE)
}
pub fn failure(self) -> Self {
self.status(CONDITION_STATUS_FALSE)
}
pub fn reason(mut self, reason: impl Into<String>) -> Self {
self.reason = reason.into();
self
}
pub fn message(mut self, message: impl Into<String>) -> Self {
self.message = message.into();
self
}
pub fn build(self, observed_generation: Option<i64>) -> Condition {
Condition {
type_: self.type_,
status: self.status,
reason: self.reason,
message: self.message,
last_transition_time: Time(Utc::now()),
observed_generation,
}
}
}
#[async_trait::async_trait]
pub trait ConditionExt: Resource + Clone + Debug + Send + Sync + 'static
where
Self::DynamicType: Default,
{
async fn update_condition(
&self,
client: &kube::Client,
condition: Condition,
) -> Result<(), kube::Error>;
async fn update_conditions(
&self,
client: &kube::Client,
conditions: Vec<Condition>,
) -> Result<(), kube::Error>;
async fn set_ready(
&self,
client: &kube::Client,
message: impl Into<String> + Send,
) -> Result<(), kube::Error> {
let generation = self.meta().generation;
let condition = ReadyCondition::success(message).into_condition(generation);
self.update_condition(client, condition).await
}
async fn set_error<E: StatusCondition + Send + Sync>(
&self,
client: &kube::Client,
error: &E,
) -> Result<(), kube::Error> {
let generation = self.meta().generation;
let condition = error.to_condition(generation);
self.update_condition(client, condition).await
}
}
#[async_trait::async_trait]
impl<T> ConditionExt for T
where
T: Resource<Scope = kube::core::NamespaceResourceScope> + Clone + Debug + Send + Sync + 'static,
T::DynamicType: Default,
T: Serialize + DeserializeOwned,
{
async fn update_condition(
&self,
client: &kube::Client,
condition: Condition,
) -> Result<(), kube::Error> {
self.update_conditions(client, vec![condition]).await
}
async fn update_conditions(
&self,
client: &kube::Client,
conditions: Vec<Condition>,
) -> Result<(), kube::Error> {
let name = self.name_any();
let namespace = self.namespace().ok_or_else(|| {
kube::Error::Api(kube::error::ErrorResponse {
status: "Failure".to_string(),
message: "Resource must be namespaced".to_string(),
reason: "MissingNamespace".to_string(),
code: 400,
})
})?;
let api: Api<T> = Api::namespaced(client.clone(), &namespace);
let patch = serde_json::json!({
"status": {
"conditions": conditions
}
});
let pp = PatchParams::apply(FIELD_MANAGER_NAME).force();
debug!(
resource = %name,
namespace = %namespace,
conditions = ?conditions.iter().map(|c| &c.type_).collect::<Vec<_>>(),
"Updating status conditions"
);
api.patch_status(&name, &pp, &Patch::Merge(&patch)).await?;
Ok(())
}
}
pub struct ReconcileContext<T> {
pub client: kube::Client,
_phantom: std::marker::PhantomData<T>,
}
impl<T> ReconcileContext<T> {
pub fn new(client: kube::Client) -> Self {
Self { client, _phantom: std::marker::PhantomData }
}
}
pub trait ReconcileExt<T, E>
where
T: Resource + Clone + Debug + Send + Sync + 'static + Serialize + DeserializeOwned,
T::DynamicType: Default,
E: StatusCondition + Send + Sync,
{
fn with_status_update<F, Fut>(
reconcile_fn: F,
) -> impl Fn(
Arc<T>,
Arc<ReconcileContext<T>>,
)
-> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Action, E>> + Send>>
where
F: Fn(Arc<T>, Arc<ReconcileContext<T>>) -> Fut + Send + Sync + Clone + 'static,
Fut: std::future::Future<Output = Result<Action, E>> + Send + 'static;
}
pub async fn reconcile_with_status<T, E, F, Fut>(
obj: Arc<T>,
ctx: Arc<ReconcileContext<T>>,
reconcile_fn: F,
) -> Result<Action, E>
where
T: Resource<Scope = kube::core::NamespaceResourceScope>
+ Clone
+ Debug
+ Send
+ Sync
+ 'static
+ Serialize
+ DeserializeOwned,
T::DynamicType: Default,
E: StatusCondition + Send + Sync,
F: Fn(Arc<T>, Arc<ReconcileContext<T>>) -> Fut,
Fut: std::future::Future<Output = Result<Action, E>>,
{
let name = obj.name_any();
let generation = obj.meta().generation;
match reconcile_fn(obj.clone(), ctx.clone()).await {
Ok(action) => {
let condition =
ReadyCondition::success("Reconciliation succeeded").into_condition(generation);
if let Err(e) = obj.update_condition(&ctx.client, condition).await {
error!(resource = %name, error = %e, "Failed to update success status");
} else {
info!(resource = %name, "Reconciliation succeeded");
}
Ok(action)
}
Err(e) => {
let condition = e.to_condition(generation);
let severity = e.severity();
let action = e.to_action();
match severity {
Severity::Info => info!(resource = %name, error = %e, "Reconciliation info"),
Severity::Warning => warn!(resource = %name, error = %e, "Reconciliation warning"),
Severity::Error => error!(resource = %name, error = %e, "Reconciliation failed"),
}
if let Err(status_err) = obj.update_condition(&ctx.client, condition).await {
error!(
resource = %name,
original_error = %e,
status_error = %status_err,
"Failed to update error status"
);
}
Ok(action)
}
}
}
#[macro_export]
macro_rules! reconcile_with_status {
($reconcile_fn:expr) => {
|obj, ctx| Box::pin($crate::reconcile_with_status(obj, ctx, $reconcile_fn))
};
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_condition_builder_success() {
let condition = ConditionBuilder::new("Synchronized")
.success()
.reason("ZoneUpdated")
.message("Zone file synchronized successfully")
.build(Some(5));
assert_eq!(condition.type_, "Synchronized");
assert_eq!(condition.status, CONDITION_STATUS_TRUE);
assert_eq!(condition.reason, "ZoneUpdated");
assert_eq!(condition.message, "Zone file synchronized successfully");
assert_eq!(condition.observed_generation, Some(5));
}
#[test]
fn test_condition_builder_failure() {
let condition = ConditionBuilder::new("Ready")
.failure()
.reason("ValidationFailed")
.message("Resource validation failed")
.build(None);
assert_eq!(condition.type_, CONDITION_TYPE_READY);
assert_eq!(condition.status, CONDITION_STATUS_FALSE);
assert_eq!(condition.reason, "ValidationFailed");
assert_eq!(condition.message, "Resource validation failed");
assert_eq!(condition.observed_generation, None);
}
#[test]
fn test_condition_builder_custom_status() {
let condition = ConditionBuilder::new("Custom")
.status("Unknown")
.reason("Investigating")
.message("Status unknown")
.build(Some(1));
assert_eq!(condition.type_, "Custom");
assert_eq!(condition.status, CONDITION_STATUS_UNKNOWN);
assert_eq!(condition.reason, "Investigating");
}
#[test]
fn test_ready_condition_success() {
let info = ReadyCondition::success("All good");
assert_eq!(info.type_, CONDITION_TYPE_READY);
assert_eq!(info.status, CONDITION_STATUS_TRUE);
assert_eq!(info.reason, CONDITION_REASON_RECONCILE_SUCCEEDED);
assert_eq!(info.message, "All good");
}
#[test]
fn test_ready_condition_progressing() {
let info = ReadyCondition::progressing("Initializing resources");
assert_eq!(info.type_, CONDITION_TYPE_READY);
assert_eq!(info.status, CONDITION_STATUS_FALSE);
assert_eq!(info.reason, CONDITION_REASON_PROGRESSING);
assert_eq!(info.message, "Initializing resources");
}
#[test]
fn test_ready_condition_helper_true() {
let info = ready_condition(true, "System operational");
assert_eq!(info.type_, CONDITION_TYPE_READY);
assert_eq!(info.status, CONDITION_STATUS_TRUE);
assert_eq!(info.reason, CONDITION_REASON_RECONCILE_SUCCEEDED);
assert_eq!(info.message, "System operational");
}
#[test]
fn test_ready_condition_helper_false() {
let info = ready_condition(false, "System not ready");
assert_eq!(info.type_, CONDITION_TYPE_READY);
assert_eq!(info.status, CONDITION_STATUS_FALSE);
assert_eq!(info.reason, "NotReady");
assert_eq!(info.message, "System not ready");
}
#[test]
fn test_error_condition_helper() {
let info = error_condition("DatabaseConnectionFailed", "Unable to connect to database");
assert_eq!(info.type_, CONDITION_TYPE_READY);
assert_eq!(info.status, CONDITION_STATUS_FALSE);
assert_eq!(info.reason, "DatabaseConnectionFailed");
assert_eq!(info.message, "Unable to connect to database");
}
#[test]
fn test_condition_info_into_condition_with_generation() {
let info = ConditionInfo {
type_: CONDITION_TYPE_READY.to_string(),
status: CONDITION_STATUS_TRUE.to_string(),
reason: CONDITION_REASON_RECONCILE_SUCCEEDED.to_string(),
message: "Test message".to_string(),
};
let condition = info.into_condition(Some(42));
assert_eq!(condition.type_, CONDITION_TYPE_READY);
assert_eq!(condition.status, CONDITION_STATUS_TRUE);
assert_eq!(condition.reason, CONDITION_REASON_RECONCILE_SUCCEEDED);
assert_eq!(condition.message, "Test message");
assert_eq!(condition.observed_generation, Some(42));
}
#[test]
fn test_condition_info_into_condition_without_generation() {
let info = ConditionInfo {
type_: CONDITION_TYPE_AVAILABLE.to_string(),
status: CONDITION_STATUS_FALSE.to_string(),
reason: "Unavailable".to_string(),
message: "Service unavailable".to_string(),
};
let condition = info.into_condition(None);
assert_eq!(condition.type_, CONDITION_TYPE_AVAILABLE);
assert_eq!(condition.observed_generation, None);
}
#[test]
fn test_severity_as_str() {
assert_eq!(Severity::Info.as_str(), "Info");
assert_eq!(Severity::Warning.as_str(), "Warning");
assert_eq!(Severity::Error.as_str(), "Error");
}
#[test]
fn test_severity_equality() {
assert_eq!(Severity::Info, Severity::Info);
assert_eq!(Severity::Warning, Severity::Warning);
assert_eq!(Severity::Error, Severity::Error);
assert_ne!(Severity::Info, Severity::Warning);
assert_ne!(Severity::Warning, Severity::Error);
}
#[test]
fn test_condition_type_constants() {
assert_eq!(CONDITION_TYPE_READY, "Ready");
assert_eq!(CONDITION_TYPE_PROGRESSING, "Progressing");
assert_eq!(CONDITION_TYPE_DEGRADED, "Degraded");
assert_eq!(CONDITION_TYPE_AVAILABLE, "Available");
}
#[test]
fn test_condition_status_constants() {
assert_eq!(CONDITION_STATUS_TRUE, "True");
assert_eq!(CONDITION_STATUS_FALSE, "False");
assert_eq!(CONDITION_STATUS_UNKNOWN, "Unknown");
}
#[test]
fn test_condition_reason_constants() {
assert_eq!(CONDITION_REASON_RECONCILE_SUCCEEDED, "ReconcileSucceeded");
assert_eq!(CONDITION_REASON_PROGRESSING, "Progressing");
}
#[test]
fn test_requeue_duration_constants() {
assert_eq!(DEFAULT_REQUEUE_SECONDS, 30);
assert_eq!(DEFAULT_NON_RETRYABLE_REQUEUE_SECONDS, 300);
}
#[test]
fn test_field_manager_constant() {
assert_eq!(FIELD_MANAGER_NAME, "kube-condition");
}
#[derive(Debug)]
struct MockError {
retryable: bool,
requeue_secs: u64,
}
impl std::fmt::Display for MockError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Mock error")
}
}
impl std::error::Error for MockError {}
impl StatusCondition for MockError {
fn to_condition_info(&self) -> ConditionInfo {
ConditionInfo {
type_: CONDITION_TYPE_READY.to_string(),
status: CONDITION_STATUS_FALSE.to_string(),
reason: "MockError".to_string(),
message: "Mock error occurred".to_string(),
}
}
fn severity(&self) -> Severity {
Severity::Error
}
fn is_retryable(&self) -> bool {
self.retryable
}
fn requeue_duration(&self) -> Duration {
Duration::from_secs(self.requeue_secs)
}
}
#[test]
fn test_status_condition_to_condition() {
let error = MockError { retryable: true, requeue_secs: 60 };
let condition = error.to_condition(Some(10));
assert_eq!(condition.type_, CONDITION_TYPE_READY);
assert_eq!(condition.status, CONDITION_STATUS_FALSE);
assert_eq!(condition.reason, "MockError");
assert_eq!(condition.message, "Mock error occurred");
assert_eq!(condition.observed_generation, Some(10));
}
#[test]
fn test_status_condition_to_action_retryable() {
let error = MockError { retryable: true, requeue_secs: 60 };
let action = error.to_action();
assert!(error.is_retryable());
assert_eq!(error.requeue_duration(), Duration::from_secs(60));
let _ = action;
}
#[test]
fn test_status_condition_to_action_non_retryable() {
let error = MockError {
retryable: false,
requeue_secs: 60, };
let action = error.to_action();
assert!(!error.is_retryable());
let _ = action;
}
}