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);

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.

TraceMetadata

Traits

Context

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