k8s_controller/
controller.rs

1use std::collections::BTreeMap;
2use std::error::Error as _;
3use std::fmt::Display;
4use std::sync::{Arc, Mutex};
5use std::time::{Duration, Instant};
6
7use futures::future::FutureExt;
8use futures::stream::StreamExt;
9use kube::api::Api;
10use kube::core::{ClusterResourceScope, NamespaceResourceScope};
11use kube::{Client, Resource, ResourceExt};
12use kube_runtime::controller::Action;
13use kube_runtime::finalizer::{Event, finalizer};
14use kube_runtime::watcher;
15use rand::{Rng, rng};
16use tracing::field::Empty;
17use tracing::{Instrument, Span, error, info, info_span, trace, warn};
18
19#[derive(Debug, thiserror::Error)]
20pub enum Error<E: std::error::Error + 'static> {
21    #[error("{0}")]
22    ControllerError(#[source] E),
23    #[error("{0}")]
24    FinalizerError(#[source] kube_runtime::finalizer::Error<E>),
25}
26
27#[derive(Debug, Default)]
28pub struct TraceMetadata(BTreeMap<String, String>);
29
30impl TraceMetadata {
31    pub fn annotate<K: Display, V: Display>(&mut self, key: K, val: V) {
32        self.0.insert(key.to_string(), val.to_string());
33    }
34}
35
36/// The [`Controller`] watches a set of resources, calling methods on the
37/// provided [`Context`] when events occur.
38pub struct Controller<Ctx: Context>
39where
40    Ctx: Send + Sync + 'static,
41    Ctx::Error: Send + Sync + 'static,
42    Ctx::Resource: Send + Sync + 'static,
43    Ctx::Resource: Clone + std::fmt::Debug + serde::Serialize,
44    for<'de> Ctx::Resource: serde::Deserialize<'de>,
45    <Ctx::Resource as Resource>::DynamicType:
46        Eq + Clone + std::hash::Hash + std::default::Default + std::fmt::Debug + std::marker::Unpin,
47{
48    client: kube::Client,
49    make_api: Box<dyn Fn(&Ctx::Resource) -> Api<Ctx::Resource> + Sync + Send + 'static>,
50    controller: kube_runtime::controller::Controller<Ctx::Resource>,
51    context: Ctx,
52}
53
54impl<Ctx: Context> Controller<Ctx>
55where
56    Ctx: Send + Sync + 'static,
57    Ctx::Error: Send + Sync + 'static,
58    Ctx::Resource: Clone + std::fmt::Debug + serde::Serialize,
59    for<'de> Ctx::Resource: serde::Deserialize<'de>,
60    <Ctx::Resource as Resource>::DynamicType:
61        Eq + Clone + std::hash::Hash + std::default::Default + std::fmt::Debug + std::marker::Unpin,
62{
63    /// Creates a new controller for a namespaced resource using the given
64    /// `client`. The `context` given determines the type of resource
65    /// to watch (via the [`Context::Resource`] type provided as part of
66    /// the trait implementation). The resources to be watched will be
67    /// limited to resources in the given `namespace`. A [`watcher::Config`]
68    /// can be given to limit the resources watched (for instance,
69    /// `watcher::Config::default().labels("app=myapp")`).
70    pub fn namespaced(client: Client, context: Ctx, namespace: &str, wc: watcher::Config) -> Self
71    where
72        Ctx::Resource: Resource<Scope = NamespaceResourceScope>,
73    {
74        let make_api = {
75            let client = client.clone();
76            Box::new(move |resource: &Ctx::Resource| {
77                Api::<Ctx::Resource>::namespaced(client.clone(), &resource.namespace().unwrap())
78            })
79        };
80        let controller = kube_runtime::controller::Controller::new(
81            Api::<Ctx::Resource>::namespaced(client.clone(), namespace),
82            wc,
83        );
84        Self {
85            client,
86            make_api,
87            controller,
88            context,
89        }
90    }
91
92    /// Creates a new controller for a namespaced resource using the given
93    /// `client`. The `context` given determines the type of resource to
94    /// watch (via the [`Context::Resource`] type provided as part of the
95    /// trait implementation). The resources to be watched will not be
96    /// limited by namespace. A [`watcher::Config`] can be given to limit the
97    /// resources watched (for instance,
98    /// `watcher::Config::default().labels("app=myapp")`).
99    pub fn namespaced_all(client: Client, context: Ctx, wc: watcher::Config) -> Self
100    where
101        Ctx::Resource: Resource<Scope = NamespaceResourceScope>,
102    {
103        let make_api = {
104            let client = client.clone();
105            Box::new(move |resource: &Ctx::Resource| {
106                Api::<Ctx::Resource>::namespaced(client.clone(), &resource.namespace().unwrap())
107            })
108        };
109        let controller = kube_runtime::controller::Controller::new(
110            Api::<Ctx::Resource>::all(client.clone()),
111            wc,
112        );
113        Self {
114            client,
115            make_api,
116            controller,
117            context,
118        }
119    }
120
121    /// Creates a new controller for a cluster-scoped resource using the
122    /// given `client`. The `context` given determines the type of resource
123    /// to watch (via the [`Context::Resource`] type provided as part of the
124    /// trait implementation). A [`watcher::Config`] can be given to limit the
125    /// resources watched (for instance,
126    /// `watcher::Config::default().labels("app=myapp")`).
127    pub fn cluster(client: Client, context: Ctx, wc: watcher::Config) -> Self
128    where
129        Ctx::Resource: Resource<Scope = ClusterResourceScope>,
130    {
131        let make_api = {
132            let client = client.clone();
133            Box::new(move |_: &Ctx::Resource| Api::<Ctx::Resource>::all(client.clone()))
134        };
135        let controller = kube_runtime::controller::Controller::new(
136            Api::<Ctx::Resource>::all(client.clone()),
137            wc,
138        );
139        Self {
140            client,
141            make_api,
142            controller,
143            context,
144        }
145    }
146
147    /// Run the controller. This method will not return. The [`Context`]
148    /// given to the constructor will have its [`apply`](Context::apply)
149    /// method called when a resource is created or updated, and its
150    /// [`cleanup`](Context::cleanup) method called when a resource is about
151    /// to be deleted.
152    ///
153    /// To run multiple replicas of a controller with only one reconciling
154    /// at a time, pass this method's future to
155    /// [`LeaderElection::with_lease`](crate::LeaderElection::with_lease).
156    pub async fn run(self) {
157        let Self {
158            client,
159            make_api,
160            controller,
161            context,
162        } = self;
163        let backoffs = Arc::new(Mutex::new(BTreeMap::new()));
164        let backoffs = &backoffs;
165        controller
166            .run(
167                |resource, context| {
168                    let uid = resource.uid().unwrap();
169                    let backoffs = Arc::clone(backoffs);
170                    context
171                        ._reconcile(client.clone(), make_api(&resource), resource)
172                        .inspect(move |result| {
173                            if result.is_ok() {
174                                backoffs.lock().unwrap().remove(&uid);
175                            }
176                        })
177                },
178                |resource, err, context| {
179                    let consecutive_errors = {
180                        let uid = resource.uid().unwrap();
181                        let mut backoffs = backoffs.lock().unwrap();
182                        let consecutive_errors: u32 =
183                            backoffs.get(&uid).copied().unwrap_or_default();
184                        backoffs.insert(uid, consecutive_errors.saturating_add(1));
185                        consecutive_errors
186                    };
187                    context.error_action(resource, err, consecutive_errors)
188                },
189                Arc::new(context),
190            )
191            .for_each(|res| async {
192                // ReconcilerFailed errors will already have been reported by
193                // the _reconcile function
194                if let Err(e) = res
195                    && !matches!(e, kube_runtime::controller::Error::ReconcilerFailed(..))
196                {
197                    // warn instead of error because these kinds of errors
198                    // are almost always recoverable
199                    warn!(
200                        error = %e,
201                        source = e.source(),
202                        "internal kube controller error",
203                    );
204                }
205            })
206            .await
207    }
208
209    /// Allow configuring the underlying [`kube_runtime::Controller`]. For
210    /// example, you can use
211    /// `controller.with_controller(|controller| controller.with_config(Config::default().concurrency(10)))`
212    /// to limit the created controller to reconciling 10 resources at once.
213    pub fn with_controller<F>(mut self, f: F) -> Self
214    where
215        F: FnOnce(
216            kube_runtime::Controller<Ctx::Resource>,
217        ) -> kube_runtime::Controller<Ctx::Resource>,
218    {
219        self.controller = f(self.controller);
220        self
221    }
222}
223
224/// The [`Context`] trait should be implemented in order to provide callbacks
225/// for events that happen to resources watched by a [`Controller`].
226#[cfg_attr(not(docsrs), async_trait::async_trait)]
227pub trait Context {
228    /// The type of Kubernetes [resource](Resource) that will be watched by
229    /// the [`Controller`] this context is passed to
230    type Resource: Resource + Send + Sync + 'static;
231    /// The error type which will be returned by the [`apply`](Self::apply)
232    /// and [`cleanup`](Self::cleanup) methods
233    type Error: std::error::Error;
234
235    /// The name to use for the finalizer. This must be unique across
236    /// controllers - if multiple controllers with the same finalizer name
237    /// run against the same resource, unexpected behavior can occur.
238    ///
239    /// If this is None (the default), a finalizer will not be used, and
240    /// cleanup events will not be reported.
241    const FINALIZER_NAME: Option<&'static str> = None;
242
243    /// This method is called when a watched resource is created or updated.
244    /// The [`Client`] used by the controller is passed in to allow making
245    /// additional API requests, as is the resource which triggered this
246    /// event. If this method returns `Some(action)`, the given action will
247    /// be performed, otherwise if `None` is returned,
248    /// [`success_action`](Self::success_action) will be called to find the
249    /// action to perform.
250    async fn apply(
251        &self,
252        client: Client,
253        resource: &Self::Resource,
254        metadata: &mut TraceMetadata,
255    ) -> Result<Option<Action>, Self::Error>;
256
257    /// This method is called when a watched resource is marked for deletion.
258    /// The [`Client`] used by the controller is passed in to allow making
259    /// additional API requests, as is the resource which triggered this
260    /// event. If this method returns `Some(action)`, the given action will
261    /// be performed, otherwise if `None` is returned,
262    /// [`success_action`](Self::success_action) will be called to find the
263    /// action to perform.
264    ///
265    /// Note that this method will only be called if a finalizer is used.
266    async fn cleanup(
267        &self,
268        client: Client,
269        resource: &Self::Resource,
270        metadata: &mut TraceMetadata,
271    ) -> Result<Option<Action>, Self::Error> {
272        // use a better name for the parameter name in the docs
273        let _client = client;
274        let _resource = resource;
275        let _metadata = metadata;
276
277        Ok(Some(Action::await_change()))
278    }
279
280    /// This method is called when a call to [`apply`](Self::apply) or
281    /// [`cleanup`](Self::cleanup) returns `Ok(None)`. It should return the
282    /// default [`Action`] to perform. The default implementation will
283    /// requeue the event at a random time between 40 and 60 minutes in the
284    /// future.
285    fn success_action(&self, resource: &Self::Resource) -> Action {
286        // use a better name for the parameter name in the docs
287        let _resource = resource;
288
289        Action::requeue(Duration::from_secs(rng().random_range(2400..3600)))
290    }
291
292    /// This method is called when a call to [`apply`](Self::apply) or
293    /// [`cleanup`](Self::cleanup) returns `Err`. It should return the
294    /// default [`Action`] to perform. The error returned will be passed in
295    /// here, as well as a count of how many consecutive errors have happened
296    /// for this resource, to allow for an exponential backoff strategy. The
297    /// default implementation uses exponential backoff with a max of 256
298    /// seconds and some added randomization to avoid thundering herds.
299    fn error_action(
300        self: Arc<Self>,
301        resource: Arc<Self::Resource>,
302        err: &Error<Self::Error>,
303        consecutive_errors: u32,
304    ) -> Action {
305        // use a better name for the parameter name in the docs
306        let _resource = resource;
307        let _err = err;
308
309        let seconds = 2u64.pow(consecutive_errors.min(7) + 1);
310        Action::requeue(Duration::from_millis(
311            rng().random_range((seconds * 500)..(seconds * 1000)),
312        ))
313    }
314
315    #[doc(hidden)]
316    async fn _reconcile(
317        self: Arc<Self>,
318        client: Client,
319        api: Api<Self::Resource>,
320        resource: Arc<Self::Resource>,
321    ) -> Result<Action, Error<Self::Error>>
322    where
323        Self: Send + Sync + 'static,
324        Self::Error: Send + Sync + 'static,
325        Self::Resource: Send + Sync + 'static,
326        Self::Resource: Clone + std::fmt::Debug + serde::Serialize,
327        for<'de> Self::Resource: serde::Deserialize<'de>,
328        <Self::Resource as Resource>::DynamicType: Eq
329            + Clone
330            + std::hash::Hash
331            + std::default::Default
332            + std::fmt::Debug
333            + std::marker::Unpin,
334    {
335        let span = info_span!(
336            "reconcile",
337            resource_type = Self::Resource::kind(&Default::default()).as_ref(),
338            resource_name = resource.name_unchecked().as_str(),
339            controller = Self::FINALIZER_NAME,
340            event_type = Empty,
341            success = Empty,
342            duration_seconds = Empty,
343            metadata = Empty,
344        );
345        async {
346            trace!("beginning reconciliation");
347
348            let mut metadata = TraceMetadata::default();
349            let mut ran = false;
350            let start = Instant::now();
351
352            let res = if let Some(finalizer_name) = Self::FINALIZER_NAME {
353                finalizer(&api, finalizer_name, Arc::clone(&resource), |event| async {
354                    ran = true;
355                    Span::current().record(
356                        "event_type",
357                        match event {
358                            Event::Apply(_) => "apply",
359                            Event::Cleanup(_) => "cleanup",
360                        },
361                    );
362                    match event {
363                        Event::Apply(resource) => self
364                            .apply(client, &resource, &mut metadata)
365                            .await
366                            .map(|action| action.unwrap_or_else(|| self.success_action(&resource))),
367                        Event::Cleanup(resource) => self
368                            .cleanup(client, &resource, &mut metadata)
369                            .await
370                            .map(|action| action.unwrap_or_else(Action::await_change)),
371                    }
372                })
373                .await
374                .map_err(Error::FinalizerError)
375            } else if resource.meta().deletion_timestamp.is_none() {
376                ran = true;
377                Span::current().record("event_type", "apply");
378                self.apply(client, &resource, &mut metadata)
379                    .await
380                    .map(|action| action.unwrap_or_else(|| self.success_action(&resource)))
381                    .map_err(Error::ControllerError)
382            } else {
383                Ok(Action::await_change())
384            };
385
386            // eventually we should set up metrics here, but this should
387            // help for now
388            Span::current().record("duration_seconds", start.elapsed().as_secs_f64());
389
390            if !ran {
391                Span::current().record(
392                    "event_type",
393                    if resource.meta().deletion_timestamp.is_some() {
394                        "delete"
395                    } else {
396                        "init"
397                    },
398                );
399            }
400
401            if !metadata.0.is_empty()
402                && let Ok(s) = serde_json::to_string(&metadata.0)
403            {
404                Span::current().record("metadata", s);
405            }
406
407            if let Err(e) = &res {
408                Span::current().record("success", false);
409                error!(error = %e, source = e.source(), "reconcile");
410            } else {
411                Span::current().record("success", true);
412                info!("reconcile");
413            }
414
415            res
416        }
417        .instrument(span)
418        .await
419    }
420}
k8s_controller/controller.rs

k8s_controller/
controller.rs
