use std::{future::Future, pin::Pin, sync::Arc, time::Duration};
use k8s_openapi::{NamespaceResourceScope, serde};
use kube::{
Resource,
api::{Api, ResourceExt},
runtime::{controller::Action, finalizer},
};
use crate::controller::{Reconciler, RunError, RunOutcome};
pub fn requeue_on_error<K, E: std::fmt::Debug, C>(duration: Duration) -> impl Fn(Arc<K>, &E, Arc<C>) -> Action {
#[allow(unused_variables)]
move |_, e, _| {
#[cfg(feature = "controller_trace")]
tracing::warn!(?e, "reconcile failed");
Action::requeue(duration)
}
}
#[expect(clippy::type_complexity)]
pub fn reconcile_with_finalizer<R, D, RR, F>(
default_reconcile_fn: F,
) -> impl Fn(Arc<RR>, Arc<R>) -> Pin<Box<dyn Future<Output = Result<Action, finalizer::Error<RunError<R::Error>>>> + Send>>
where
R: Reconciler<Resource = RR, EvaluationResource = finalizer::Event<RR>, InitialData = D> + Send + Sync + 'static,
<R as Reconciler>::Error: std::error::Error + Send + 'static,
<R as Reconciler>::EvaluationOutcome: Send,
RR: Resource<Scope = NamespaceResourceScope>
+ serde::Serialize
+ for<'de> serde::Deserialize<'de>
+ std::fmt::Debug
+ Clone
+ kube::CustomResourceExt
+ kube::core::object::HasStatus
+ Send
+ Sync
+ 'static,
<RR as kube::Resource>::DynamicType: std::default::Default,
<RR as kube::core::object::HasStatus>::Status: Send,
D: Send + 'static,
F: FnOnce(RunOutcome, R::EvaluationOutcome) -> Action + Send + Sync + Clone + 'static,
{
move |obj: Arc<RR>, reconciler: Arc<R>| {
let default_reconcile_fn = default_reconcile_fn.clone();
Box::pin(async move {
let ns = obj.namespace().unwrap();
let client = reconciler
.client()
.await
.map_err(|source| finalizer::Error::ApplyFailed(RunError::Reconciler { source }))?;
let objects: Api<R::Resource> = Api::namespaced(client, &ns);
let finalizer_name = &R::Resource::crd_name();
finalizer::finalizer(&objects, finalizer_name, obj, |event| async {
let (outcome, eval_outcome) = crate::controller::run(reconciler.as_ref(), event).await?;
Ok(default_reconcile_fn(outcome, eval_outcome))
})
.await
})
}
}
#[cfg(test)]
mod tests {
use std::{
collections::VecDeque,
convert::Infallible,
error::Error,
fmt,
sync::{Arc, Mutex},
time::Duration,
};
use http::{HeaderMap, Method, StatusCode, Uri};
use k8s_openapi::{
apimachinery::pkg::apis::meta::v1::{ObjectMeta, Time},
jiff::Timestamp,
};
use kube::{CustomResource, CustomResourceExt, client::Body};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use crate::controller::{BoxPlan, Plan, RunContext, StatusUpdater};
use super::*;
#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
#[kube(
group = "tests.staircase.dev",
version = "v1",
kind = "FinalizerTest",
status = "FinalizerTestStatus",
namespaced
)]
struct FinalizerTestSpec {}
#[derive(Deserialize, Serialize, Clone, Debug, Default, JsonSchema)]
struct FinalizerTestStatus {}
#[derive(Debug, PartialEq, Eq)]
enum TestError {}
impl fmt::Display for TestError {
fn fmt(&self, _: &mut fmt::Formatter<'_>) -> fmt::Result { match *self {} }
}
impl Error for TestError {}
#[derive(Clone, Debug)]
struct CapturedRequest {
method: Method,
uri: Uri,
headers: HeaderMap,
body: Vec<u8>,
}
#[derive(Clone, Debug)]
struct QueuedResponse {
status: StatusCode,
body: Value,
}
struct FakeReconciler {
client: kube::Client,
log: Arc<Mutex<Vec<&'static str>>>,
}
impl Reconciler for FakeReconciler {
type Error = TestError;
type EvaluationOutcome = &'static str;
type EvaluationResource = finalizer::Event<FinalizerTest>;
type InitialData = ();
type Plan = BoxPlan<Self::Resource, Self::Error, Self::InitialData, ()>;
type Resource = FinalizerTest;
async fn evaluate_plan(
&self,
event: Self::EvaluationResource,
) -> ((), Self::EvaluationOutcome, Arc<Self::Resource>, Self::Plan) {
let (event_name, resource) = match event {
finalizer::Event::Apply(resource) => ("apply", resource),
finalizer::Event::Cleanup(resource) => ("cleanup", resource),
};
self.log.lock().unwrap().push(event_name);
(
(),
"evaluated",
resource,
Plan::empty::<FinalizerTest, TestError, ()>().boxed(),
)
}
fn build_config(&self) -> crate::controller::RunReconcilerConfig { Default::default() }
fn build_status_updater(&self) -> impl StatusUpdater<Self::Resource> + Send + Sync { NoopStatusUpdater }
async fn client(&self) -> Result<kube::Client, Self::Error> {
self.log.lock().unwrap().push("client");
Ok(self.client.clone())
}
}
struct NoopStatusUpdater;
impl StatusUpdater<FinalizerTest> for NoopStatusUpdater {
type Error = TestError;
async fn update(&self, _: &RunContext<FinalizerTest>, _: FinalizerTestStatus) -> Result<(), Self::Error> {
Ok(())
}
}
fn client_with_responses(responses: Vec<QueuedResponse>) -> (kube::Client, Arc<Mutex<Vec<CapturedRequest>>>) {
let captured = Arc::new(Mutex::new(Vec::new()));
let responses = Arc::new(Mutex::new(VecDeque::from(responses)));
let service = tower::service_fn({
let captured = captured.clone();
let responses = responses.clone();
move |request: http::Request<Body>| {
let captured = captured.clone();
let responses = responses.clone();
async move {
let (parts, body) = request.into_parts();
let body = body.collect_bytes().await.unwrap().to_vec();
captured.lock().unwrap().push(CapturedRequest {
method: parts.method,
uri: parts.uri,
headers: parts.headers,
body,
});
let response = responses.lock().unwrap().pop_front().expect("unexpected request");
Ok::<_, Infallible>(
http::Response::builder()
.status(response.status)
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&response.body).unwrap()))
.unwrap(),
)
}
}
});
(kube::Client::new(service, "default"), captured)
}
fn json_response(body: impl Serialize) -> QueuedResponse {
QueuedResponse {
status: StatusCode::OK,
body: serde_json::to_value(body).unwrap(),
}
}
fn server_error_response() -> QueuedResponse {
QueuedResponse {
status: StatusCode::INTERNAL_SERVER_ERROR,
body: json!({
"kind": "Status",
"apiVersion": "v1",
"metadata": {},
"status": "Failure",
"message": "mock finalizer patch failure",
"reason": "InternalError",
"code": 500
}),
}
}
fn resource(finalizers: Option<Vec<String>>, deleting: bool) -> FinalizerTest {
FinalizerTest {
metadata: ObjectMeta {
name: Some("sample".to_string()),
namespace: Some("apps".to_string()),
finalizers,
deletion_timestamp: deleting.then(|| Time(Timestamp::now())),
..Default::default()
},
spec: FinalizerTestSpec {},
status: None,
}
}
fn request_json(request: &CapturedRequest) -> Value { serde_json::from_slice(&request.body).unwrap() }
#[test]
fn requeue_on_error_returns_action() {
let policy = requeue_on_error::<FinalizerTest, _, ()>(Duration::from_secs(5));
let _action = policy(Arc::new(resource(None, false)), &"error", Arc::new(()));
}
#[test]
fn reconcile_with_finalizer_adds_finalizer_before_apply() {
let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
let mut patched = resource(Some(vec![FinalizerTest::crd_name().to_string()]), false);
patched.metadata.resource_version = Some("rv-2".to_string());
let (client, captured) = client_with_responses(vec![json_response(patched)]);
let log = Arc::new(Mutex::new(Vec::new()));
let reconciler = Arc::new(FakeReconciler {
client,
log: log.clone(),
});
let reconcile = reconcile_with_finalizer(|outcome, eval| {
panic!("unexpected reconcile call: {outcome:?} {eval}");
});
reconcile(Arc::new(resource(None, false)), reconciler).await.unwrap();
assert_eq!(*log.lock().unwrap(), vec!["client"]);
let captured = captured.lock().unwrap();
assert_eq!(captured.len(), 1);
assert_eq!(captured[0].method, Method::PATCH);
assert_eq!(
captured[0].uri.path(),
"/apis/tests.staircase.dev/v1/namespaces/apps/finalizertests/sample"
);
assert_eq!(
captured[0].headers.get("content-type").unwrap(),
"application/json-patch+json"
);
assert_eq!(
request_json(&captured[0]),
json!([
{"op": "test", "path": "/metadata/finalizers", "value": null},
{"op": "add", "path": "/metadata/finalizers", "value": [FinalizerTest::crd_name()]}
])
);
});
}
#[test]
fn reconcile_with_finalizer_appends_finalizer_when_other_finalizers_exist() {
let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
let mut patched = resource(
Some(vec![
"other.example.com".to_string(),
FinalizerTest::crd_name().to_string(),
]),
false,
);
patched.metadata.resource_version = Some("rv-2".to_string());
let (client, captured) = client_with_responses(vec![json_response(patched)]);
let log = Arc::new(Mutex::new(Vec::new()));
let reconciler = Arc::new(FakeReconciler {
client,
log: log.clone(),
});
let reconcile = reconcile_with_finalizer(|outcome, eval| {
panic!("unexpected reconcile call: {outcome:?} {eval}");
});
reconcile(
Arc::new(resource(Some(vec!["other.example.com".to_string()]), false)),
reconciler,
)
.await
.unwrap();
assert_eq!(*log.lock().unwrap(), vec!["client"]);
let captured = captured.lock().unwrap();
assert_eq!(captured.len(), 1);
assert_eq!(captured[0].method, Method::PATCH);
assert_eq!(
request_json(&captured[0]),
json!([
{"op": "test", "path": "/metadata/finalizers", "value": ["other.example.com"]},
{"op": "add", "path": "/metadata/finalizers/-", "value": FinalizerTest::crd_name()}
])
);
});
}
#[test]
fn reconcile_with_finalizer_runs_apply_when_finalizer_exists() {
let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
let (client, captured) = client_with_responses(Vec::new());
let log = Arc::new(Mutex::new(Vec::new()));
let reconciler = Arc::new(FakeReconciler {
client,
log: log.clone(),
});
let reconcile = reconcile_with_finalizer(|outcome, eval| {
assert!(matches!(outcome, RunOutcome::NoOp));
assert_eq!(eval, "evaluated");
Action::await_change()
});
reconcile(
Arc::new(resource(Some(vec![FinalizerTest::crd_name().to_string()]), false)),
reconciler,
)
.await
.unwrap();
assert_eq!(*log.lock().unwrap(), vec!["client", "client", "apply"]);
assert!(captured.lock().unwrap().is_empty());
});
}
#[test]
fn reconcile_with_finalizer_deleting_without_our_finalizer_is_noop() {
let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
let (client, captured) = client_with_responses(Vec::new());
let log = Arc::new(Mutex::new(Vec::new()));
let reconciler = Arc::new(FakeReconciler {
client,
log: log.clone(),
});
let reconcile = reconcile_with_finalizer(|outcome, eval| {
panic!("unexpected reconcile call: {outcome:?} {eval}");
});
reconcile(
Arc::new(resource(Some(vec!["other.example.com".to_string()]), true)),
reconciler,
)
.await
.unwrap();
assert_eq!(*log.lock().unwrap(), vec!["client"]);
assert!(captured.lock().unwrap().is_empty());
});
}
#[test]
fn reconcile_with_finalizer_runs_cleanup_and_removes_finalizer() {
let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
let patched = resource(None, true);
let (client, captured) = client_with_responses(vec![json_response(patched)]);
let log = Arc::new(Mutex::new(Vec::new()));
let reconciler = Arc::new(FakeReconciler {
client,
log: log.clone(),
});
let reconcile = reconcile_with_finalizer(|outcome, eval| {
assert!(matches!(outcome, RunOutcome::NoOp));
assert_eq!(eval, "evaluated");
Action::await_change()
});
reconcile(
Arc::new(resource(Some(vec![FinalizerTest::crd_name().to_string()]), true)),
reconciler,
)
.await
.unwrap();
assert_eq!(*log.lock().unwrap(), vec!["client", "client", "cleanup"]);
let captured = captured.lock().unwrap();
assert_eq!(captured.len(), 1);
assert_eq!(captured[0].method, Method::PATCH);
assert_eq!(
captured[0].headers.get("content-type").unwrap(),
"application/json-patch+json"
);
assert_eq!(
request_json(&captured[0]),
json!([
{"op": "test", "path": "/metadata/finalizers/0", "value": FinalizerTest::crd_name()},
{"op": "remove", "path": "/metadata/finalizers/0"}
])
);
});
}
#[test]
fn reconcile_with_finalizer_maps_add_and_remove_patch_failures() {
let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
let (client, captured) = client_with_responses(vec![server_error_response()]);
let log = Arc::new(Mutex::new(Vec::new()));
let reconciler = Arc::new(FakeReconciler {
client,
log: log.clone(),
});
let reconcile = reconcile_with_finalizer(|outcome, eval| {
panic!("unexpected reconcile call: {outcome:?} {eval}");
});
let error = reconcile(Arc::new(resource(None, false)), reconciler)
.await
.unwrap_err();
assert!(matches!(error, finalizer::Error::AddFinalizer(_)));
assert_eq!(*log.lock().unwrap(), vec!["client"]);
assert_eq!(captured.lock().unwrap().len(), 1);
});
rt.block_on(async {
let (client, captured) = client_with_responses(vec![server_error_response()]);
let log = Arc::new(Mutex::new(Vec::new()));
let reconciler = Arc::new(FakeReconciler {
client,
log: log.clone(),
});
let reconcile = reconcile_with_finalizer(|outcome, eval| {
assert!(matches!(outcome, RunOutcome::NoOp));
assert_eq!(eval, "evaluated");
Action::await_change()
});
let error = reconcile(
Arc::new(resource(Some(vec![FinalizerTest::crd_name().to_string()]), true)),
reconciler,
)
.await
.unwrap_err();
assert!(matches!(error, finalizer::Error::RemoveFinalizer(_)));
assert_eq!(*log.lock().unwrap(), vec!["client", "client", "cleanup"]);
assert_eq!(captured.lock().unwrap().len(), 1);
});
}
}