Operators

cargo generate --git https://github.com/bohdaq/rust-web-server starters/operator

use rust_web_server::k8s::{kube::CustomResource, schemars::JsonSchema};
use serde::{Deserialize, Serialize};

#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
#[kube(group = "example.com", version = "v1", kind = "Backup", namespaced, status = "BackupStatus")]
pub struct BackupSpec {
    pub schedule: String,
    pub database: String,
}

#[derive(Deserialize, Serialize, Clone, Debug, Default, JsonSchema)]
pub struct BackupStatus {
    pub last_run: Option<String>,
}

use std::sync::Arc;
use std::time::Duration;
use rust_web_server::k8s::controller::{Action, Context, Controller, Operator};
use rust_web_server::k8s::kube::api::Api;
use rust_web_server::k8s::Error;
use k8s_openapi::api::batch::v1::CronJob;

async fn reconcile(backup: Arc<Backup>, ctx: Arc<Context>) -> Result<Action, Error> {
    ctx.apply_owned(&*backup, cron_job_for(&backup)).await?;                 // owned: deleted with it
    ctx.patch_status(&*backup, serde_json::json!({ "last_run": null })).await?;
    ctx.event(&*backup, false, "Scheduled", "the CronJob is up to date").await?;
    Ok(Action::requeue(Duration::from_secs(300)))
}

let kube = rust_web_server::k8s::client::client().await?;
Operator::new(kube.clone(), "backup-operator")
    .leader_election(&rust_web_server::k8s::client::current_namespace(), "backup-operator")
    .controller(
        Controller::new(Api::<Backup>::all(kube.clone()))
            .owns(Api::<CronJob>::all(kube.clone()))          // a change to its CronJob reconciles the Backup
            .reconcile(reconcile)
            .finalizer("example.com/cleanup", |backup, ctx| async move { delete_backups_of(&backup).await }),
    )
    .run()
    .await;

let operator = tokio::spawn(operator.run());
let (listener, pool) = Server::setup()?;
Server::run_tls(listener, pool, App::new()).await;     // /livez, /readyz, /metrics

use rust_web_server::k8s::webhook::{Webhooks, WebhookCertificate};

let webhooks = Webhooks::new()
    .validate::<Backup, _>("/validate-backups", |request| match &request.object {
        Some(b) if b.spec.schedule.is_empty() => Err("spec.schedule is required".into()),
        _ => Ok(()),
    })
    .mutate::<Backup, _>("/mutate-backups", |mut backup| {
        backup.labels_mut().insert("example.com/managed".into(), "true".into());
        Ok(backup)
    })
    .convert("/convert-backups", |object, desired_version| v1_to_v2(object, desired_version))
    .wrap(App::new());                                       // probes and metrics on the same server

// The API server only calls webhooks over TLS it trusts.
let cert = WebhookCertificate::shared(&kube, &namespace, "backup-operator-webhook-tls", "backup-operator").await?;
cert.install("/tmp/tls")?;                                    // RWS_CONFIG_TLS_CERT_FILE / KEY_FILE
cert.patch_ca_bundle(&kube, Some("backup-operator"), None, None).await?;
Server::run_tls(listener, pool, webhooks).await;

use rust_web_server::k8s::manifests::{self, Rbac, WebhookRule, WebhookTarget};

manifests::print_on_request(|| {                // `backup-operator --manifests | kubectl apply -f -`
    let rbac = Rbac::new("backup-operator", "backups").rules(operator.rules());
    let target = WebhookTarget { service: "backup-operator".into(), namespace: "backups".into(), port: 443, path: "/validate-backups".into(), ca_pem: ca.clone() };
    let mut docs = vec![manifests::crd::<Backup>()];
    docs.extend(rbac.documents());
    docs.push(manifests::webhook_configuration("backups", false, &target, &[WebhookRule::new("example.com", "v1", "backups", &["CREATE", "UPDATE"])], true));
    docs
});

use rust_web_server::k8s::testing::{eventually, TestCluster};

#[tokio::test]
async fn backups_get_a_cron_job() {
    let Some(cluster) = TestCluster::from_env().await else { return };   // skipped without a cluster
    cluster.install_crd::<Backup>().await.unwrap();                       // applied, then waited for
    let ns = cluster.namespace("backup").await.unwrap();                  // backup-<random>, labelled rws-test=true
    tokio::spawn(operator(cluster.client.clone()).run());
    ns.api::<Backup>().create(&Default::default(), &Backup::new("nightly", spec())).await.unwrap();
    let job = eventually(Duration::from_secs(30), || async { ns.api::<CronJob>().get_opt("nightly").await.ok().flatten() }).await;
    assert!(job.is_some());
    ns.delete().await;
}

kind create cluster --name test --kubeconfig /tmp/kubeconfig
RWS_TEST_KUBECONFIG=/tmp/kubeconfig cargo test