Struct kube_runtime::controller::Controller

pub struct Controller<K> where
    K: Clone + Resource + Debug + 'static,
    K::DynamicType: Eq + Hash,  { /* fields omitted */ }

Controller

A controller is made up of:

• 1 reflector (for the core object)
• N watcher objects for each object child object
• user defined reconcile + error_policy callbacks
• a generated input stream considering all sources

And all reconcile requests through an internal scheduler

Pieces:

use kube::{Client, api::{Api, ListParams}};
use kube_derive::CustomResource;
use serde::{Deserialize, Serialize};
use tokio::time::Duration;
use futures::StreamExt;
use kube_runtime::controller::{Context, Controller, ReconcilerAction};
use k8s_openapi::api::core::v1::ConfigMap;
use schemars::JsonSchema;

use snafu::{Backtrace, OptionExt, ResultExt, Snafu};
#[derive(Debug, Snafu)]
enum Error {}
/// A custom resource
#[derive(CustomResource, Debug, Clone, Deserialize, Serialize, JsonSchema)]
#[kube(group = "nullable.se", version = "v1", kind = "ConfigMapGenerator", namespaced)]
struct ConfigMapGeneratorSpec {
    content: String,
}

/// The reconciler that will be called when either object change
async fn reconcile(g: ConfigMapGenerator, _ctx: Context<()>) -> Result<ReconcilerAction, Error> {
    // .. use api here to reconcile a child ConfigMap with ownerreferences
    // see configmapgen_controller example for full info
    Ok(ReconcilerAction {
        requeue_after: Some(Duration::from_secs(300)),
    })
}
/// an error handler that will be called when the reconciler fails
fn error_policy(_error: &Error, _ctx: Context<()>) -> ReconcilerAction {
    ReconcilerAction {
        requeue_after: Some(Duration::from_secs(60)),
    }
}

/// something to drive the controller
#[tokio::main]
async fn main() -> Result<(), kube::Error> {
    let client = Client::try_default().await?;
    let context = Context::new(()); // bad empty context - put client in here
    let cmgs = Api::<ConfigMapGenerator>::all(client.clone());
    let cms = Api::<ConfigMap>::all(client.clone());
    Controller::new(cmgs, ListParams::default())
        .owns(cms, ListParams::default())
        .run(reconcile, error_policy, context)
        .for_each(|res| async move {
            match res {
                Ok(o) => println!("reconciled {:?}", o),
                Err(e) => println!("reconcile failed: {:?}", e),
            }
        })
        .await; // controller does nothing unless polled
    Ok(())
}

Implementations

impl<K> Controller<K> where
    K: Clone + Resource + DeserializeOwned + Debug + Send + Sync + 'static,
    K::DynamicType: Eq + Hash + Clone + Default, 

#[must_use]pub fn new(owned_api: Api<K>, lp: ListParams) -> Self

Create a Controller on a type K

Configure ListParams and Api so you only get reconcile events for the correct Api scope (cluster/all/namespaced), or ListParams subset

impl<K> Controller<K> where
    K: Clone + Resource + DeserializeOwned + Debug + Send + Sync + 'static,
    K::DynamicType: Eq + Hash + Clone, 

pub fn new_with(
    owned_api: Api<K>,
    lp: ListParams,
    dyntype: K::DynamicType
) -> Self

Create a Controller on a type K

Configure ListParams and Api so you only get reconcile events for the correct Api scope (cluster/all/namespaced), or ListParams subset

Unlike new, this function accepts K::DynamicType so it can be used with dynamic resources.

pub fn store(&self) -> Store<K>

Retrieve a copy of the reader before starting the controller

pub fn owns<Child: Clone + Resource + DeserializeOwned + Debug + Send + 'static>(
    self,
    api: Api<Child>,
    lp: ListParams
) -> Self where
    Child::DynamicType: Debug + Eq + Hash, 

Indicate child objets K owns and be notified when they change

This type Child must have OwnerReference set to point back to K. You can customize the parameters used by the underlying watcher if only a subset of Child entries are required. The api must have the correct scope (cluster/all namespaces, or namespaced)

pub fn watches<Other: Clone + Resource + DeserializeOwned + Debug + Send + 'static, I: 'static + IntoIterator<Item = ObjectRef<K>>>(
    self,
    api: Api<Other>,
    lp: ListParams,
    mapper: impl Fn(Other) -> I + Send + 'static
) -> Self where
    I::IntoIter: Send, 

Indicate an object to watch with a custom mapper

This mapper should return something like Option<ObjectRef<K>>

pub fn run<ReconcilerFut, T>(
    self,
    reconciler: impl FnMut(K, Context<T>) -> ReconcilerFut,
    error_policy: impl FnMut(&ReconcilerFut::Error, Context<T>) -> ReconcilerAction,
    context: Context<T>
) -> impl Stream<Item = Result<(ObjectRef<K>, ReconcilerAction), Error<ReconcilerFut::Error, Error>>> where
    K::DynamicType: Debug + Unpin,
    ReconcilerFut: TryFuture<Ok = ReconcilerAction> + Send + 'static,
    ReconcilerFut::Error: Error + Send + 'static, 

Consume all the parameters of the Controller and start the applier stream

This creates a stream from all builder calls and starts an applier with a specified reconciler and error_policy callbacks. Each of these will be called with a configurable Context.