Implement version CUR

Cargo.toml

@@ -5,5 +5,6 @@ members = [
     "kiss/assets",
     "kiss/controller",
     "kiss/gateway",
+    "kiss/manager",
     "kiss/monitor",
 ]

kiss/manager/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 
 [dependencies]
 ipis = { git = "https://github.com/ulagbulag-village/ipis" }
+kiss-api = { path = "../api" }
 
 octocrab = "0.17"
 semver = "1.0"

kiss/manager/src/current.rs

@@ -0,0 +1,83 @@
+use std::collections::BTreeMap;
+
+use ipis::core::anyhow::{anyhow, Result};
+use kiss_api::{
+    k8s_openapi::{api::core::v1::ConfigMap, Resource},
+    kube::{
+        api::{Patch, PatchParams, PostParams},
+        core::ObjectMeta,
+        Api, Client,
+    },
+    serde_json::json,
+};
+use semver::Version;
+
+pub struct Handler {
+    api: Api<ConfigMap>,
+}
+
+impl Handler {
+    pub async fn try_default() -> Result<Self> {
+        // create a kubernetes client
+        let ns = "kiss";
+        let client = Client::try_default().await?;
+
+        Ok(Self {
+            api: Api::<ConfigMap>::namespaced(client, ns),
+        })
+    }
+}
+
+impl Handler {
+    pub async fn create(&self, version: &Version) -> Result<()> {
+        let config = ConfigMap {
+            metadata: ObjectMeta {
+                name: Some("manager".into()),
+                ..Default::default()
+            },
+            immutable: Some(false),
+            data: Some({
+                let mut map = BTreeMap::default();
+                map.insert("version".into(), version.to_string());
+                map
+            }),
+            ..Default::default()
+        };
+        let pp = PostParams {
+            field_manager: Some("kiss-manager".into()),
+            ..Default::default()
+        };
+        self.api.create(&pp, &config).await?;
+        Ok(())
+    }
+
+    pub async fn get(&self, latest: &Version) -> Result<Version> {
+        let config = match self.api.get("manager").await {
+            Ok(config) => config,
+            Err(_) => {
+                self.create(latest).await?;
+                return Ok(latest.clone());
+            }
+        };
+
+        let version = config
+            .data
+            .as_ref()
+            .and_then(|map| map.get("version"))
+            .ok_or_else(|| anyhow!("failed to find version field in configmap"))?;
+        version.parse().map_err(Into::into)
+    }
+
+    pub async fn patch(&self, version: Version) -> Result<()> {
+        let patch = Patch::Apply(json!({
+            "apiVersion": ConfigMap::API_VERSION,
+            "kind": ConfigMap::KIND,
+            "spec": {
+                "version": version.to_string(),
+            },
+        }));
+        let pp = PatchParams::apply("kiss-manager").force();
+        self.api.patch("manager", &pp, &patch).await?;
+        Ok(())
+    }
+}

kiss/manager/src/latest.rs

@@ -0,0 +1,45 @@
+use std::sync::Arc;
+
+use ipis::{
+    core::anyhow::{bail, Result},
+    env::infer,
+};
+use octocrab::Octocrab;
+use semver::Version;
+
+pub struct Handler {
+    instance: Arc<Octocrab>,
+    repo_name: String,
+    repo_owner: String,
+}
+
+impl Default for Handler {
+    fn default() -> Self {
+        Self {
+            instance: Default::default(),
+            repo_name: infer("REPO_NAME").unwrap_or_else(|_| Self::REPOSITORY_NAME.into()),
+            repo_owner: infer("REPO_OWNER").unwrap_or_else(|_| Self::REPOSITORY_OWNER.into()),
+        }
+    }
+}
+
+impl Handler {
+    const REPOSITORY_NAME: &str = "netai-cloud";
+    const REPOSITORY_OWNER: &str = "ulagbulag-village";
+
+    pub async fn get(&self) -> Result<Version> {
+        // request the latest release info
+        let release = self
+            .instance
+            .repos(&self.repo_owner, &self.repo_name)
+            .releases()
+            .get_latest()
+            .await?;
+
+        // compare with the current release tag
+        if !release.tag_name.starts_with("v") {
+            bail!("Received unexpected version tag: {:?}", &release.tag_name);
+        }
+        Version::parse(&release.tag_name[1..]).map_err(Into::into)
+    }
+}

kiss/manager/src/main.rs

@@ -6,34 +6,28 @@
     clippy::restriction
 )]
 
+mod current;
+mod latest;
+
 use std::time::Duration;
 
 use ipis::{
     core::anyhow::Result,
-    env::infer,
     log::{info, warn},
     tokio,
 };
-use octocrab::repos::RepoHandler;
 use semver::Version;
 
-const REPOSITORY_OWNER: &str = "ulagbulag-village";
-const REPOSITORY_NAME: &str = "netai-cloud";
-
-async fn sync_cluster(repo: &RepoHandler<'_>) -> Result<()> {
-    // request the latest release info
-    let release = repo.releases().get_latest().await?;
-
-    // compare with the current release tag
-    if !release.tag_name.starts_with("v") {
-        warn!("Received unexpected version tag: {:?}", &release.tag_name);
-        return Ok(());
-    }
-    let latest = Version::parse(&release.tag_name[1..]).unwrap();
-    let current = Version::parse("0.0.1").unwrap();
+async fn sync_cluster(
+    current_handler: &self::current::Handler,
+    latest_handler: &self::latest::Handler,
+) -> Result<()> {
+    // request the release info
+    let latest = latest_handler.get().await?;
+    let current = current_handler.get(&latest).await?;
 
     // if possible, update the cluster
-    if latest > current {
+    if &latest > &current {
         info!("Found the newer version: {current} -> {latest}");
         upgrade_cluster(latest).await
     } else if latest < current {
@@ -51,17 +45,13 @@ async fn upgrade_cluster(version: Version) -> Result<()> {
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    // get environment variables
-    let repo_owner = infer("REPO_OWNER").unwrap_or_else(|_| REPOSITORY_OWNER.to_string());
-    let repo_name = infer("REPO_NAME").unwrap_or_else(|_| REPOSITORY_NAME.to_string());
-
-    // create a repository handler
-    let instance = octocrab::instance();
-    let handler = instance.repos(&repo_owner, &repo_name);
+    // create the handlers
+    let current = self::current::Handler::try_default().await?;
+    let latest = self::latest::Handler::default();
 
     // sync the cluster periodically
     loop {
-        sync_cluster(&handler).await?;
+        sync_cluster(&current, &latest).await?;
         tokio::time::sleep(Duration::from_secs(5 * 60)).await;
     }
 }