api: start on adding admin command support

Signed-off-by: Hank Donnay <[email protected]>

api/src/v1alpha1.rs

@@ -151,6 +151,12 @@ pub struct ClairStatus {
     /// Config is configuration sources for the Clair instance.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub config: Option<ConfigSource>,
+    /// Previous_version is the previous verison of a deployed Clair instance, if any.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub previous_version: Option<String>,
+    /// Current_version is the current verison of a deployed Clair instance.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub current_version: Option<String>,
     /*
     /// Database is the Service for the managed database engine, if used.
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -264,6 +270,35 @@ impl std::fmt::Display for ConfigDialect {
     }
 }
 
+// ImageRef exists to have some Object to hang pre/post Jobs off of.
+// I don't think this is actually needed -- The can/could be driven off of a Condition.
+/*
+/// ImageRefSpec is the spec of an ImageRef.
+#[derive(
+    CustomResource, Clone, Debug, Default, Deserialize, PartialEq, Serialize, Validate, JsonSchema,
+)]
+#[kube(
+    group = "projectclair.io",
+    version = "v1alpha1",
+    kind = "ImageRef",
+    namespaced,
+    status = "ImageRefStatus",
+    derive = "PartialEq",
+    shortname = "imgref",
+    category = "apps"
+)]
+#[serde(rename_all = "camelCase")]
+pub struct ImageRefSpec {
+    pub repository: String, // TODO(hank) verification
+    pub tag: String,        // TODO(hank) verification
+}
+
+/// ImageRefStatus is the status of an ImageRef.
+#[derive(Clone, Debug, Deserialize, Default, PartialEq, Serialize, Validate, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct ImageRefStatus {}
+*/
+
 /// IndexerSpec describes the desired state of an Indexer instance.
 #[derive(
     CustomResource, Clone, Debug, Default, Deserialize, PartialEq, Serialize, Validate, JsonSchema,

controller/Cargo.toml

@@ -41,6 +41,7 @@ tokio-native-tls = "0.3.1"
 tokio-util = { version = "0.7.8" }
 tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter"] }
 axum = { version = "0.6.18", features = ["http1", "json", "tracing"] }
+regex = "1.8.4"
 
 [dev-dependencies]
 reqwest = { version = "0.11.18", features = ["json"] }

controller/src/clairs.rs

@@ -17,6 +17,7 @@ use tokio_stream::wrappers::SignalStream;
 
 use crate::{
     clair_condition, prelude::*, COMPONENT_LABEL, DEFAULT_CONFIG_JSON, DEFAULT_CONFIG_YAML,
+    DEFAULT_IMAGE,
 };
 use clair_config;
 
@@ -53,6 +54,10 @@ pub fn controller(cancel: CancellationToken, ctx: Arc<Context>) -> Result<Contro
             Api::<core::v1::ConfigMap>::default_namespaced(client.clone()),
             ctlcfg.clone(),
         )
+        .owns(
+            Api::<batch::v1::Job>::default_namespaced(client.clone()),
+            ctlcfg.clone(),
+        )
         .owns(
             Api::<networking::v1::Ingress>::default_namespaced(client),
             ctlcfg,
@@ -72,6 +77,7 @@ pub fn controller(cancel: CancellationToken, ctx: Arc<Context>) -> Result<Contro
                             error!(%objref, %error, "reconcile error")
                         }
                         CtrlErr::QueueError(error) => error!(%error, "queue error"),
+                        CtrlErr::RunnerError(error) => error!(%error, "runner error"),
                     },
                 };
                 futures::future::ready(())
@@ -180,6 +186,7 @@ $(
     check_all!(
         check_config,
         check_dropins,
+        check_admin_job,
         check_subs,
         check_ingress,
         check_indexer,
@@ -440,6 +447,65 @@ async fn check_config(
     Ok(true)
 }
 
+//#[instrument(skip_all)]
+async fn check_admin_job(
+    obj: &v1alpha1::Clair,
+    ctx: &Context,
+    _req: &Request,
+    _next: &mut v1alpha1::ClairStatus,
+) -> Result<bool> {
+    trace!("checking admin job");
+    let ns = obj.namespace().unwrap_or("default".to_string());
+    let api: Api<batch::v1::Job> = Api::namespaced(ctx.client.clone(), &ns);
+    // Need to:
+    // - Determine if the image version is going to get updated.
+    //   - If not, check that the "post" job has run OK.
+    //     - If not, warn somehow.
+    //   - If so, check that the "pre" job has run OK.
+    //     - If not, block the changes.
+    if obj.status.is_none() {
+        return Ok(true);
+    }
+    let status = obj.status.as_ref().unwrap();
+    if let Some(ref v) = status.current_version {
+        if v != &obj.spec.image_default(&DEFAULT_IMAGE) {
+            // Version mismatch, find out why.
+        } else {
+            // Version is current, check for post job.
+            let cnd_post = clair_condition("AdminPostComplete");
+            match status.conditions.iter().find(|c| c.type_ == cnd_post) {
+                Some(c) => {
+                    // Condition exists
+                    match c.status.as_str() {
+                        "True" => {
+                            // Great, done.
+                            return Ok(true);
+                        }
+                        "False" | "Unknown" => {
+                            // The Job is either running or needs to be created.
+                            if let Some(j) = api.get_opt("name").await? {
+                                let s = j.status.unwrap_or_default().succeeded.unwrap_or_default();
+                                debug!("succeeded: {}", s);
+                            } else {
+                                unimplemented!("create?")
+                            }
+                        }
+                        _ => unreachable!(),
+                    }
+                }
+                None => {
+                    // No post job, seemingly. Check for Job creation.
+                    unimplemented!("no post job")
+                }
+            };
+        }
+    } else {
+        // In setup, no current version.
+        unimplemented!("in setup, no current version")
+    }
+    Ok(true)
+}
+
 #[instrument(skip_all)]
 async fn check_subs(
     obj: &v1alpha1::Clair,

controller/src/indexers.rs

@@ -50,6 +50,7 @@ pub fn controller(cancel: CancellationToken, ctx: Arc<Context>) -> Result<Contro
                             error!(%objref, %error, "reconcile error")
                         }
                         CtrlErr::QueueError(error) => error!(%error, "queue error"),
+                        CtrlErr::RunnerError(error) => error!(%error, "runner error"),
                     },
                 };
                 futures::future::ready(())

controller/src/lib.rs

@@ -12,6 +12,7 @@ use futures::Future;
 use k8s_openapi::{api::core, apimachinery::pkg::apis::meta};
 use kube::runtime::events;
 use lazy_static::lazy_static;
+use regex::Regex;
 use tracing::{instrument, trace};
 
 use api::v1alpha1;
@@ -217,6 +218,19 @@ pub fn k8s_label<S: AsRef<str>>(s: S) -> String {
     keyify("app.kubernetes.io/", s)
 }
 
+/// Image_version returns the version for an image, if present.
+///
+/// Semver versions are the only accepted version strings.
+pub fn image_version<'s>(img: &'s str) -> Option<&'s str> {
+    // The semver regexp:
+    lazy_static! {
+        static ref RE: Regex = Regex::new(r#"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$"#).unwrap();
+    }
+    img.split_once(':')
+        .map(|(_, t)| t)
+        .filter(|t| RE.is_match(t))
+}
+
 /// New_templated returns a `K` with patches for `S` applied and the owner set to `obj`.
 #[instrument(skip_all)]
 pub async fn new_templated<S, K>(obj: &S, _ctx: &Context) -> Result<K>

controller/src/matchers.rs

@@ -50,6 +50,7 @@ pub fn controller(cancel: CancellationToken, ctx: Arc<Context>) -> Result<Contro
                             error!(%objref, %error, "reconcile error")
                         }
                         CtrlErr::QueueError(error) => error!(%error, "queue error"),
+                        CtrlErr::RunnerError(error) => error!(%error, "runner error"),
                     },
                 };
                 futures::future::ready(())