rws_k8s/k8s/
controller.rs

1//! Controllers: watch a kind, reconcile each object toward what it asks
2//! for, requeue, clean up through a finalizer — kube-runtime's controller
3//! with the parts every operator writes anyway: an [`Operator`] running
4//! several controllers under leader election, a [`Context`] with status
5//! and event helpers, metrics on `/metrics`, a readiness check, and the
6//! RBAC rules the controllers need.
7//!
8//! ```rust,no_run
9//! use std::sync::Arc;
10//! use std::time::Duration;
11//! use rws_k8s::k8s::controller::{Action, Context, Controller, Operator};
12//! use rws_k8s::k8s::kube::{api::Api, CustomResource};
13//! use rws_k8s::k8s::schemars::JsonSchema;
14//! use serde::{Deserialize, Serialize};
15//!
16//! #[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
17//! #[kube(group = "example.com", version = "v1", kind = "Backup", namespaced, status = "BackupStatus")]
18//! pub struct BackupSpec { pub schedule: String }
19//!
20//! #[derive(Deserialize, Serialize, Clone, Debug, Default, JsonSchema)]
21//! pub struct BackupStatus { pub last_run: Option<String> }
22//!
23//! async fn reconcile(backup: Arc<Backup>, ctx: Arc<Context>) -> Result<Action, rws_k8s::k8s::Error> {
24//!     // … make the world match backup.spec …
25//!     ctx.patch_status(&*backup, serde_json::json!({ "last_run": "now" })).await?;
26//!     Ok(Action::requeue(Duration::from_secs(300)))
27//! }
28//!
29//! # async fn example() -> rws_k8s::k8s::Result<()> {
30//! let client = rws_k8s::k8s::client::client().await?;
31//! Operator::new(client.clone(), "backup-operator")
32//!     .leader_election("default", "backup-operator")
33//!     .controller(Controller::new(Api::<Backup>::all(client)).reconcile(reconcile).finalizer("example.com/cleanup", |_backup, _ctx| async { Ok(()) }))
34//!     .run()
35//!     .await;
36//! # Ok(()) }
37//! ```
38//!
39//! A reconcile that fails is retried after `on_error`'s delay (5 s by
40//! default, doubling up to 5 min for the same object). With a finalizer,
41//! the cleanup runs when the object is deleted and the finalizer is removed
42//! after it succeeds.
43
44use std::collections::BTreeMap;
45use std::fmt::Debug;
46use std::future::Future;
47use std::pin::Pin;
48use std::sync::atomic::{AtomicU64, Ordering};
49use std::sync::{Arc, Mutex, OnceLock};
50use std::time::Duration;
51
52use futures_util::StreamExt;
53use kube::api::{Api, Patch, PatchParams};
54use kube::runtime::events::{Event as KubeEvent, EventType, Recorder, Reporter};
55use kube::runtime::finalizer::{finalizer, Event as FinalizerEvent};
56use kube::runtime::watcher;
57use kube::{Resource, ResourceExt};
58use serde::de::DeserializeOwned;
59use serde::Serialize;
60
61pub use kube::runtime::controller::Action;
62
63use super::manifests::Rule;
64use super::Error;
65
66type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>;
67
68/// What a reconcile gets besides the object: the client, an event
69/// recorder, and the operator's name.
70pub struct Context {
71    pub client: kube::Client,
72    pub operator: String,
73    recorder: Recorder,
74}
75
76impl Context {
77    fn new(client: kube::Client, operator: &str) -> Self {
78        let reporter = Reporter { controller: operator.to_string(), instance: Some(super::client::identity()) };
79        Context { recorder: Recorder::new(client.clone(), reporter), client, operator: operator.to_string() }
80    }
81
82    /// Merge `status` into `obj`'s status subresource.
83    pub async fn patch_status<K>(&self, obj: &K, status: serde_json::Value) -> Result<K, Error>
84    where
85        K: Resource<DynamicType = ()> + Clone + DeserializeOwned + Debug,
86        K::Scope: Scope,
87    {
88        let api = api_for(&self.client, obj);
89        let patch = serde_json::json!({ "status": status });
90        Ok(api.patch_status(&obj.name_any(), &PatchParams::default(), &Patch::Merge(&patch)).await?)
91    }
92
93    /// Server-side apply `child` as owned by `owner` (an owner reference,
94    /// so it's garbage-collected with it), with this operator as the field
95    /// manager.
96    pub async fn apply_owned<O, C>(&self, owner: &O, mut child: C) -> Result<C, Error>
97    where
98        O: Resource<DynamicType = ()>,
99        C: Resource<DynamicType = ()> + Clone + DeserializeOwned + Serialize + Debug,
100        C::Scope: Scope,
101    {
102        let reference = owner.controller_owner_ref(&()).ok_or_else(|| Error::Other("the owner has no uid yet".into()))?;
103        child.meta_mut().owner_references = Some(vec![reference]);
104        if child.meta().namespace.is_none() {
105            child.meta_mut().namespace = owner.meta().namespace.clone();
106        }
107        let api = api_for(&self.client, &child);
108        let name = child.name_any();
109        let params = PatchParams::apply(&self.operator).force();
110        Ok(api.patch(&name, &params, &Patch::Apply(&child)).await?)
111    }
112
113    /// Record a Kubernetes Event about `obj` (`kubectl describe` shows it).
114    pub async fn event<K: Resource<DynamicType = ()>>(&self, obj: &K, warning: bool, reason: &str, note: &str) -> Result<(), Error> {
115        let event = KubeEvent {
116            type_: if warning { EventType::Warning } else { EventType::Normal },
117            reason: reason.to_string(),
118            note: Some(note.to_string()),
119            action: reason.to_string(),
120            secondary: None,
121        };
122        Ok(self.recorder.publish(&event, &obj.object_ref(&())).await?)
123    }
124}
125
126use super::{api_for, Scope};
127
128type Reconcile<K> = Arc<dyn Fn(Arc<K>, Arc<Context>) -> BoxFuture<Result<Action, Error>> + Send + Sync>;
129type Cleanup<K> = Arc<dyn Fn(Arc<K>, Arc<Context>) -> BoxFuture<Result<(), Error>> + Send + Sync>;
130type Owns<K> = Box<dyn FnOnce(kube::runtime::Controller<K>, &kube::Client) -> kube::runtime::Controller<K> + Send>;
131
132/// One controller: a kind, what it owns, how it reconciles. See the
133/// [module docs](self).
134pub struct Controller<K: Resource<DynamicType = ()> + Clone + Debug + 'static> {
135    api: Api<K>,
136    config: watcher::Config,
137    owns: Vec<Owns<K>>,
138    reconcile: Option<Reconcile<K>>,
139    cleanup: Option<(String, Cleanup<K>)>,
140    error_delay: Duration,
141    rules: Vec<Rule>,
142}
143
144/// The API group and plural of `K` (`("apps", "deployments")`).
145pub fn group_and_plural<K: Resource<DynamicType = ()>>() -> (String, String) {
146    (K::group(&()).to_string(), K::plural(&()).to_string())
147}
148
149impl<K> Controller<K>
150where
151    K: Resource<DynamicType = ()> + Clone + DeserializeOwned + Serialize + Debug + Send + Sync + 'static,
152    K::Scope: Scope,
153{
154    /// Watch `api`'s objects (`Api::all` or `Api::namespaced`).
155    pub fn new(api: Api<K>) -> Self {
156        let (group, plural) = group_and_plural::<K>();
157        let rules = vec![
158            Rule::new(&group, &[&plural], &["get", "list", "watch", "patch", "update"]),
159            Rule::new(&group, &[&format!("{plural}/status")], &["get", "patch", "update"]),
160            Rule::new("events.k8s.io", &["events"], &["create", "patch"]),
161        ];
162        Controller { api, config: watcher::Config::default(), owns: Vec::new(), reconcile: None, cleanup: None, error_delay: Duration::from_secs(5), rules }
163    }
164
165    /// Only the objects `config` selects (labels, fields).
166    pub fn watching(mut self, config: watcher::Config) -> Self {
167        self.config = config;
168        self
169    }
170
171    /// Also reconcile an object when a `C` it owns (an owner reference)
172    /// changes — the Deployment of a custom resource, say.
173    pub fn owns<C>(mut self, api: Api<C>) -> Self
174    where
175        C: Resource<DynamicType = ()> + Clone + DeserializeOwned + Debug + Send + Sync + 'static,
176    {
177        let (group, plural) = group_and_plural::<C>();
178        self.rules.push(Rule::new(&group, &[&plural], &["get", "list", "watch", "create", "update", "patch", "delete"]));
179        self.owns.push(Box::new(move |c, _| c.owns(api, watcher::Config::default())));
180        self
181    }
182
183    /// The reconcile function: make the world match the object, answer
184    /// when to look again.
185    pub fn reconcile<F, Fut>(mut self, f: F) -> Self
186    where
187        F: Fn(Arc<K>, Arc<Context>) -> Fut + Send + Sync + 'static,
188        Fut: Future<Output = Result<Action, Error>> + Send + 'static,
189    {
190        self.reconcile = Some(Arc::new(move |obj, ctx| Box::pin(f(obj, ctx))));
191        self
192    }
193
194    /// Add the finalizer `name` to every object and run `cleanup` when one
195    /// is deleted; the finalizer is removed (letting the deletion finish)
196    /// once `cleanup` succeeds.
197    pub fn finalizer<F, Fut>(mut self, name: &str, cleanup: F) -> Self
198    where
199        F: Fn(Arc<K>, Arc<Context>) -> Fut + Send + Sync + 'static,
200        Fut: Future<Output = Result<(), Error>> + Send + 'static,
201    {
202        self.cleanup = Some((name.to_string(), Arc::new(move |obj, ctx| Box::pin(cleanup(obj, ctx)))));
203        self
204    }
205
206    /// The first delay before retrying a failed reconcile (doubling per
207    /// failure of the same object, up to 5 min). Default 5 s.
208    pub fn on_error(mut self, first_delay: Duration) -> Self {
209        self.error_delay = first_delay.max(Duration::from_millis(10));
210        self
211    }
212
213    /// Extra RBAC rules the reconcile needs (reading Secrets, say).
214    pub fn rule(mut self, rule: Rule) -> Self {
215        self.rules.push(rule);
216        self
217    }
218}
219
220/// Per-controller counters for `/metrics`.
221#[derive(Default)]
222struct Counters {
223    reconciles: AtomicU64,
224    errors: AtomicU64,
225}
226
227fn counters() -> &'static Mutex<BTreeMap<String, Arc<Counters>>> {
228    static C: OnceLock<Mutex<BTreeMap<String, Arc<Counters>>>> = OnceLock::new();
229    C.get_or_init(|| Mutex::new(BTreeMap::new()))
230}
231
232fn counters_for(kind: &str) -> Arc<Counters> {
233    rws_server::metrics::register_section(rws_server::metrics::section::CONTROLLERS, metrics_text);
234    Arc::clone(counters().lock().unwrap_or_else(|e| e.into_inner()).entry(kind.to_string()).or_default())
235}
236
237/// `rws_controller_reconciles_total` / `rws_controller_reconcile_errors_total`
238/// by controller kind.
239pub fn metrics_text() -> String {
240    let all = counters().lock().unwrap_or_else(|e| e.into_inner());
241    if all.is_empty() {
242        return String::new();
243    }
244    let mut out = String::from("# HELP rws_controller_reconciles_total Reconciles run, by controller.\n# TYPE rws_controller_reconciles_total counter\n");
245    for (kind, c) in all.iter() {
246        out.push_str(&format!("rws_controller_reconciles_total{{controller=\"{kind}\"}} {}\n", c.reconciles.load(Ordering::Relaxed)));
247    }
248    out.push_str("# HELP rws_controller_reconcile_errors_total Reconciles that failed, by controller.\n# TYPE rws_controller_reconcile_errors_total counter\n");
249    for (kind, c) in all.iter() {
250        out.push_str(&format!("rws_controller_reconcile_errors_total{{controller=\"{kind}\"}} {}\n", c.errors.load(Ordering::Relaxed)));
251    }
252    out
253}
254
255/// The delay before retry number `failures` (1-based): `first` doubled per
256/// failure, at most 5 minutes.
257pub fn backoff(first: Duration, failures: u32) -> Duration {
258    let factor = 1u64 << failures.saturating_sub(1).min(16);
259    first.saturating_mul(factor as u32).min(Duration::from_secs(300))
260}
261
262/// A controller, its kind erased, for the [`Operator`].
263pub trait Runnable: Send {
264    fn kind(&self) -> String;
265    fn rules(&self) -> Vec<Rule>;
266    fn start(self: Box<Self>, client: kube::Client, ctx: Arc<Context>) -> BoxFuture<()>;
267}
268
269#[derive(Debug)]
270struct Failure(Error);
271
272impl std::fmt::Display for Failure {
273    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274        write!(f, "{}", self.0)
275    }
276}
277
278impl std::error::Error for Failure {}
279
280impl<K> Runnable for Controller<K>
281where
282    K: Resource<DynamicType = ()> + Clone + DeserializeOwned + Serialize + Debug + Send + Sync + 'static,
283    K::Scope: Scope,
284{
285    fn kind(&self) -> String {
286        K::kind(&()).to_string()
287    }
288
289    fn rules(&self) -> Vec<Rule> {
290        self.rules.clone()
291    }
292
293    fn start(self: Box<Self>, client: kube::Client, ctx: Arc<Context>) -> BoxFuture<()> {
294        let this = *self;
295        let kind = K::kind(&()).to_string();
296        let counters = counters_for(&kind);
297        let Some(reconcile) = this.reconcile else {
298            eprintln!("[k8s] controller {kind}: no reconcile function; not started");
299            return Box::pin(async {});
300        };
301        let cleanup = this.cleanup;
302        let error_delay = this.error_delay;
303        let failures: Arc<Mutex<BTreeMap<String, u32>>> = Arc::new(Mutex::new(BTreeMap::new()));
304        let mut controller = kube::runtime::Controller::new(this.api, this.config);
305        for owns in this.owns {
306            controller = owns(controller, &client);
307        }
308        let reconcile_counters = Arc::clone(&counters);
309        let reconcile_failures = Arc::clone(&failures);
310        let reconciler = move |obj: Arc<K>, ctx: Arc<Context>| {
311            let (reconcile, cleanup) = (Arc::clone(&reconcile), cleanup.clone());
312            let (counters, failures) = (Arc::clone(&reconcile_counters), Arc::clone(&reconcile_failures));
313            async move {
314                counters.reconciles.fetch_add(1, Ordering::Relaxed);
315                let key = format!("{}/{}", obj.namespace().unwrap_or_default(), obj.name_any());
316                let result = match cleanup {
317                    None => reconcile(obj, ctx).await,
318                    Some((name, cleanup)) => {
319                        let object_api = api_for(&ctx.client, &*obj);
320                        finalizer(&object_api, &name, obj, |event| async {
321                            match event {
322                                FinalizerEvent::Apply(o) => reconcile(o, Arc::clone(&ctx)).await.map_err(Failure),
323                                FinalizerEvent::Cleanup(o) => cleanup(o, Arc::clone(&ctx)).await.map(|_| Action::await_change()).map_err(Failure),
324                            }
325                        })
326                        .await
327                        .map_err(|e| Error::Other(e.to_string()))
328                    }
329                };
330                if result.is_ok() {
331                    failures.lock().unwrap_or_else(|e| e.into_inner()).remove(&key);
332                }
333                result.map_err(Failure)
334            }
335        };
336        let error_counters = Arc::clone(&counters);
337        let error_policy = move |obj: Arc<K>, error: &Failure, _ctx: Arc<Context>| {
338            error_counters.errors.fetch_add(1, Ordering::Relaxed);
339            let key = format!("{}/{}", obj.namespace().unwrap_or_default(), obj.name_any());
340            let n = {
341                let mut f = failures.lock().unwrap_or_else(|e| e.into_inner());
342                let n = f.entry(key.clone()).or_insert(0);
343                *n += 1;
344                *n
345            };
346            let delay = backoff(error_delay, n);
347            eprintln!("[k8s] reconcile {} {key} failed ({error}); retry in {delay:?}", K::kind(&()));
348            Action::requeue(delay)
349        };
350        Box::pin(async move {
351            controller
352                .run(reconciler, error_policy, ctx)
353                .for_each(|result| async move {
354                    if let Err(e) = result {
355                        // Watch errors (no permission, a missing CRD) and
356                        // objects gone before their reconcile.
357                        eprintln!("[k8s] controller: {e}");
358                    }
359                })
360                .await;
361        })
362    }
363}
364
365/// Several controllers, optionally under leader election, with a readiness
366/// check. See the [module docs](self).
367pub struct Operator {
368    client: kube::Client,
369    name: String,
370    controllers: Vec<Box<dyn Runnable>>,
371    lease: Option<(String, String)>,
372}
373
374impl Operator {
375    /// `name` is the field manager and event reporter.
376    pub fn new(client: kube::Client, name: &str) -> Self {
377        Operator { client, name: name.to_string(), controllers: Vec::new(), lease: None }
378    }
379
380    pub fn controller(mut self, controller: impl Runnable + 'static) -> Self {
381        self.controllers.push(Box::new(controller));
382        self
383    }
384
385    /// Run the controllers only on the replica holding the Lease `name` in
386    /// `namespace`; the others wait, ready to take over.
387    pub fn leader_election(mut self, namespace: &str, name: &str) -> Self {
388        self.lease = Some((namespace.to_string(), name.to_string()));
389        self
390    }
391
392    /// The RBAC rules the controllers (and the Lease) need.
393    pub fn rules(&self) -> Vec<Rule> {
394        let mut rules: Vec<Rule> = self.controllers.iter().flat_map(|c| c.rules()).collect();
395        if self.lease.is_some() {
396            rules.push(Rule::new("coordination.k8s.io", &["leases"], &["get", "create", "update"]));
397        }
398        super::manifests::merge_rules(rules)
399    }
400
401    /// The kinds the controllers reconcile.
402    pub fn kinds(&self) -> Vec<String> {
403        self.controllers.iter().map(|c| c.kind()).collect()
404    }
405
406    /// Run until the task is dropped. Registers a readiness check named
407    /// after the operator: ready once this replica runs its controllers or
408    /// follows a leader.
409    pub async fn run(self) {
410        let ctx = Arc::new(Context::new(self.client.clone(), &self.name));
411        rws_server::health::readiness(&format!("operator {}", self.name), || Ok(()));
412        match self.lease {
413            None => run_all(self.controllers, self.client, ctx).await,
414            Some((namespace, lease)) => {
415                let elector = super::leader::LeaderElector::new(self.client.clone(), &namespace, &lease);
416                let mut handle = elector.handle();
417                tokio::spawn(elector.run());
418                let mut controllers = Some(self.controllers);
419                loop {
420                    handle.wait_until_leader().await;
421                    // kube's controllers can't be restarted from their parts: a
422                    // replica that loses the Lease exits, and Kubernetes starts
423                    // it again as a follower.
424                    let Some(all) = controllers.take() else {
425                        eprintln!("[k8s] operator {}: leadership lost; exiting so the pod restarts as a follower", self.name);
426                        std::process::exit(1);
427                    };
428                    println!("[k8s] operator {}: leading; starting {} controller(s)", self.name, all.len());
429                    let running = run_all(all, self.client.clone(), Arc::clone(&ctx));
430                    tokio::pin!(running);
431                    loop {
432                        tokio::select! {
433                            _ = &mut running => return,
434                            leading = handle.changed() => if !leading { break },
435                        }
436                    }
437                }
438            }
439        }
440    }
441}
442
443async fn run_all(controllers: Vec<Box<dyn Runnable>>, client: kube::Client, ctx: Arc<Context>) {
444    let futures: Vec<BoxFuture<()>> = controllers.into_iter().map(|c| c.start(client.clone(), Arc::clone(&ctx))).collect();
445    futures_util::future::join_all(futures).await;
446}
447
448#[cfg(test)]
449mod tests;
rws_k8s/k8s/controller.rs

rws_k8s/k8s/
controller.rs
