Crate k8s_controller 

This crate implements a lightweight framework around kube_runtime::Controller which provides a simpler interface for common controller patterns. To use it, you define the data that your controller is going to operate over, and implement the Context trait on that struct:

#[derive(Default, Clone)]
struct PodCounter {
    pods: Arc<Mutex<BTreeSet<String>>>,
}

impl PodCounter {
    fn pod_count(&self) -> usize {
        let mut pods = self.pods.lock().unwrap();
        pods.len()
    }
}

#[async_trait::async_trait]
impl k8s_controller::Context for PodCounter {
    type Resource = Pod;
    type Error = kube::Error;

    const FINALIZER_NAME: Option<&'static str> = Some("example.com/pod-counter");

    async fn apply(
        &self,
        client: Client,
        pod: &Self::Resource,
        _metadata: &mut k8s_controller::TraceMetadata,
    ) -> Result<Option<Action>, Self::Error> {
        let mut pods = self.pods.lock().unwrap();
        pods.insert(pod.meta().uid.as_ref().unwrap().clone());
        Ok(None)
    }

    async fn cleanup(
        &self,
        client: Client,
        pod: &Self::Resource,
        _metadata: &mut k8s_controller::TraceMetadata,
    ) -> Result<Option<Action>, Self::Error> {
        let mut pods = self.pods.lock().unwrap();
        pods.remove(pod.meta().uid.as_ref().unwrap());
        Ok(None)
    }
}

Then you can run it against your Kubernetes cluster by creating a Controller:

let kube_config = Config::infer().await.unwrap();
let kube_client = Client::try_from(kube_config).unwrap();
let context = PodCounter::default();
let controller = k8s_controller::Controller::namespaced_all(
    kube_client,
    context.clone(),
    watcher::Config::default(),
);
task::spawn(controller.run());

loop {
    println!("{} pods running", context.pod_count());
    sleep(Duration::from_secs(1));
}

If you run multiple replicas of your controller (for instance, to avoid downtime of webhooks served by the same process during rollouts), you can use leader election to ensure that only one replica reconciles at a time:

let leader_election = k8s_controller::LeaderElection::new(
    kube_client.clone(),
    "my-namespace",
    "pod-counter",
    // must be unique per replica; the pod name is a good choice
    &std::env::var("HOSTNAME").unwrap(),
);
loop {
    let controller = k8s_controller::Controller::namespaced_all(
        kube_client.clone(),
        context.clone(),
        watcher::Config::default(),
    );
    leader_election.with_lease(controller.run()).await;
    // leadership was lost; the controller has been stopped, and we loop
    // to rejoin the election. Exiting the process (and letting
    // Kubernetes restart it) works too, and is preferable if your
    // reconcilers spawn tasks or do blocking work that stopping the
    // controller can't cancel.
}

A process that runs several controllers should usually guard them all with a single lease, rather than electing a separate leader per controller (which could scatter the controllers across replicas). Use LeaderElection::with_lease with a future that runs all of them:

let controller_a = k8s_controller::Controller::namespaced(
    kube_client.clone(),
    PodCounter::default(),
    "namespace-a",
    watcher::Config::default(),
);
let controller_b = k8s_controller::Controller::namespaced(
    kube_client.clone(),
    PodCounter::default(),
    "namespace-b",
    watcher::Config::default(),
);
leader_election
    .with_lease(futures::future::join(controller_a.run(), controller_b.run()))
    .await;
// leadership was lost; both controllers have been stopped
std::process::exit(1);

Observability

Every reconciliation pass is logged, as a reconcile tracing span. Beyond that, a Controller can be configured to:

• report each pass, and each step a reconciler divides its work into, to a ReconcileObserver, for metrics. With the prometheus feature, PrometheusMetrics exports these as Prometheus metrics. See the observe module.

• publish a Kubernetes event on the resource whenever reconciling it fails, through an EventRecorder, so that kubectl describe explains why it is not converging. Reconcilers can publish events of their own through the same recorder. See the events module.

• report the state of a resource through the standard status.conditions, maintained with the conditions module, which keeps lastTransitionTime and observedGeneration consistent with Kubernetes conventions. Writing the status remains up to the reconciler.

A single observer is typically shared by every controller in a process, while each controller gets its own event recorder, whose reporter names that controller:

use k8s_controller::events::{EventRecorder, Reporter};

let observer: Arc<dyn k8s_controller::ReconcileObserver> = Arc::new(MyObserver);
let events = Arc::new(EventRecorder::new(
    kube_client.clone(),
    Reporter {
        controller: "example.com/pod-counter".to_owned(),
        instance: std::env::var("HOSTNAME").ok(),
    },
));
let controller = k8s_controller::Controller::namespaced_all(
    kube_client,
    PodCounter::default(),
    watcher::Config::default(),
)
.with_name("pod-counter")
.with_observer(Arc::clone(&observer))
.with_event_recorder(Arc::clone(&events));
controller.run().await;

Within a reconciler, steps are started from the TraceMetadata passed to Context::apply and Context::cleanup:

async fn apply(
    &self,
    client: Client,
    widget: &Self::Resource,
    metadata: &mut k8s_controller::TraceMetadata,
) -> Result<Option<Action>, Self::Error> {
    let step = metadata.step("deployment");
    sync_deployment().await?;
    if !deployment_ready().await? {
        step.finish(k8s_controller::Outcome::Waiting);
        return Ok(Some(Action::requeue(std::time::Duration::from_secs(5))));
    }
    step.finish(k8s_controller::Outcome::Completed);
    Ok(None)
}

Re-exports

pub use observe::Outcome;

pub use observe::Phase;

pub use observe::ReconcileObserver;

pub use observe::ReconcileRecord;

pub use observe::Step;

pub use observe::StepRecord;

pub use observe::TraceMetadata;

Modules

conditions

Maintaining the standard status.conditions of a resource.

events

Publishing Kubernetes events about the resources a controller reconciles.

observe

Observing what reconciliation does, for metrics.

Structs

Controller

The Controller watches a set of resources, calling methods on the provided Context when events occur.

LeaderElection

Lease-based leader election, allowing multiple replicas of a controller to run while ensuring that only one of them is reconciling at a time.

PrometheusMetrics

A ReconcileObserver that exports Prometheus metrics. Requires the prometheus feature.

Enums

Error

An error from a reconciliation pass, as passed to Context::error_action and Context::failure_event.

Traits

Context

The Context trait should be implemented in order to provide callbacks for events that happen to resources watched by a Controller.