Basic version of RS (#241)

* Basic version of RS

Signed-off-by: ichAB <[email protected]>

* fix rkl test

Signed-off-by: ichAB <[email protected]>

---------

Signed-off-by: ichAB <[email protected]>

Author: ichAB

project/common/src/lib.rs

@@ -14,9 +14,11 @@ pub struct TypeMeta {
     pub kind: String,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct ObjectMeta {
     pub name: String,
+    #[serde(default)]
+    pub uid: Option<String>,
     #[serde(default = "default_namespace")]
     pub namespace: String,
     #[serde(default)]
@@ -29,7 +31,7 @@ fn default_namespace() -> String {
     "default".to_string()
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct PodSpec {
     //if pod is distributed to a node ,then this field should be filled with node-id
     #[serde(default)]
@@ -42,18 +44,18 @@ pub struct PodSpec {
     pub tolerations: Vec<Toleration>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct ContainerRes {
     pub limits: Option<Resource>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct Resource {
     pub cpu: Option<String>,
     pub memory: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct ContainerSpec {
     pub name: String,
 
@@ -68,7 +70,7 @@ pub struct ContainerSpec {
     pub resources: Option<ContainerRes>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct Port {
     #[serde(rename = "containerPort")]
     pub container_port: i32,
@@ -96,13 +98,13 @@ pub struct PodTask {
     pub status: PodStatus,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, Default)]
+#[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq, Eq)]
 pub struct PodStatus {
     #[serde(rename = "podIP")]
     pub pod_ip: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, Default)]
+#[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq, Eq)]
 pub struct Toleration {
     /// Empty means match all taint keys.
     pub key: Option<TaintKey>,
@@ -131,7 +133,7 @@ impl Toleration {
         }
     }
 }
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub enum TolerationOperator {
     Exists,
     Equal,
@@ -322,3 +324,71 @@ pub struct ServiceTask {
     pub metadata: ObjectMeta,
     pub spec: ServiceSpec,
 }
+
+#[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq, Eq)]
+pub struct LabelSelector {
+    #[serde(rename = "matchLabels", default)]
+    pub match_labels: HashMap<String, String>,
+
+    #[serde(rename = "matchExpressions", default)]
+    pub match_expressions: Vec<LabelSelectorRequirement>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub struct LabelSelectorRequirement {
+    pub key: String,
+    pub operator: LabelSelectorOperator,
+    #[serde(default)]
+    pub values: Vec<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+#[serde(rename_all = "PascalCase")]
+pub enum LabelSelectorOperator {
+    In,
+    NotIn,
+    Exists,
+    DoesNotExist,
+}
+
+fn default_replicas() -> i32 {
+    1
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub struct PodTemplateSpec {
+    pub metadata: ObjectMeta,
+    pub spec: PodSpec,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub struct ReplicaSetSpec {
+    #[serde(default = "default_replicas")]
+    pub replicas: i32,
+    pub selector: LabelSelector,
+    pub template: PodTemplateSpec,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone, Default)]
+pub struct ReplicaSetStatus {
+    #[serde(default)]
+    pub replicas: i32,
+    #[serde(rename = "fullyLabeledReplicas", default)]
+    pub fully_labeled_replicas: i32,
+    #[serde(rename = "readyReplicas", default)]
+    pub ready_replicas: i32,
+    #[serde(rename = "availableReplicas", default)]
+    pub available_replicas: i32,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ReplicaSet {
+    #[serde(rename = "apiVersion")]
+    pub api_version: String,
+    #[serde(rename = "kind")]
+    pub kind: String,
+    pub metadata: ObjectMeta,
+    pub spec: ReplicaSetSpec,
+    #[serde(default)]
+    pub status: ReplicaSetStatus,
+}

project/libscheduler/tests/xline_test.rs

@@ -76,6 +76,7 @@ fn create_test_node(name: &str, cpu: &str, memory: &str) -> Node {
         kind: "Node".to_string(),
         metadata: ObjectMeta {
             name: name.to_string(),
+            uid: None,
             namespace: "".to_string(),
             labels: HashMap::new(),
             annotations: HashMap::new(),
@@ -123,6 +124,7 @@ fn create_test_pod(name: &str, cpu_limit: Option<&str>, memory_limit: Option<&st
         kind: "Pod".to_string(),
         metadata: ObjectMeta {
             name: name.to_string(),
+            uid: None,
             namespace: "default".to_string(),
             labels: HashMap::new(),
             annotations: HashMap::new(),

project/rkl/src/daemon/client.rs

@@ -526,6 +526,7 @@ pub async fn generate_node(ext_iface: &ExternalInterface) -> Result<Node> {
         kind: "Node".to_string(),
         metadata: ObjectMeta {
             name: hostname,
+            uid: None,
             namespace: "default".to_string(),
             labels: HashMap::new(),
             annotations: HashMap::new(),

project/rkl/tests/test_common.rs

@@ -28,6 +28,7 @@ where
         kind: "Pod".to_string(),
         metadata: ObjectMeta {
             name,
+            uid: None,
             labels: HashMap::from([
                 ("app".to_string(), "my-app".to_string()),
                 ("bundle".to_string(), bundles_path("pause")),

project/rks/src/api/xlinestore.rs

@@ -325,4 +325,84 @@ impl XlineStore {
         let (watcher, stream) = client.watch(key_prefix, Some(opts)).await?;
         Ok((watcher, stream))
     }
+
+    /// Insert a replicaset YAML definition into xline.
+    pub async fn insert_replicaset_yaml(&self, rs_name: &str, rs_yaml: &str) -> Result<()> {
+        let key = format!("/registry/replicasets/{rs_name}");
+        let mut client = self.client.write().await;
+        client.put(key, rs_yaml, Some(PutOptions::new())).await?;
+        Ok(())
+    }
+
+    /// Get a replicaset YAML definition from xline.
+    pub async fn get_replicaset_yaml(&self, rs_name: &str) -> Result<Option<String>> {
+        let key = format!("/registry/replicasets/{rs_name}");
+        let mut client = self.client.write().await;
+        let resp = client.get(key, None).await?;
+        Ok(resp
+            .kvs()
+            .first()
+            .map(|kv| String::from_utf8_lossy(kv.value()).to_string()))
+    }
+
+    /// Delete a replicaset from xline.
+    pub async fn delete_replicaset(&self, rs_name: &str) -> Result<()> {
+        let key = format!("/registry/replicasets/{rs_name}");
+        let mut client = self.client.write().await;
+        client.delete(key, None).await?;
+        Ok(())
+    }
+
+    /// List all replicaset YAMLs (deserialize values).
+    pub async fn list_replicasets(&self) -> Result<Vec<ReplicaSet>> {
+        let key = "/registry/replicasets/".to_string();
+        let mut client = self.client.write().await;
+        let resp = client
+            .get(key.clone(), Some(GetOptions::new().with_prefix()))
+            .await?;
+
+        let rss: Vec<ReplicaSet> = resp
+            .kvs()
+            .iter()
+            .filter_map(|kv| {
+                let yaml_str = String::from_utf8_lossy(kv.value());
+                serde_yaml::from_str::<ReplicaSet>(&yaml_str).ok()
+            })
+            .collect();
+
+        Ok(rss)
+    }
+
+    /// Take a snapshot of all replicasets and return them with the current revision.
+    pub async fn replicasets_snapshot_with_rev(&self) -> Result<(Vec<(String, String)>, i64)> {
+        let key_prefix = "/registry/replicasets/".to_string();
+        let mut client = self.client.write().await;
+        let resp = client
+            .get(key_prefix.clone(), Some(GetOptions::new().with_prefix()))
+            .await?;
+        let rev = resp.header().map(|h| h.revision()).unwrap_or(0);
+        let items: Vec<(String, String)> = resp
+            .kvs()
+            .iter()
+            .map(|kv| {
+                (
+                    String::from_utf8_lossy(kv.key()).replace("/registry/replicasets/", ""),
+                    String::from_utf8_lossy(kv.value()).to_string(),
+                )
+            })
+            .collect();
+        Ok((items, rev))
+    }
+
+    /// Create a watch on all replicasets with prefix `/registry/replicasets/`, starting from a given revision.
+    pub async fn watch_replicasets(&self, start_rev: i64) -> Result<(Watcher, WatchStream)> {
+        let key_prefix = "/registry/replicasets/".to_string();
+        let opts = WatchOptions::new()
+            .with_prefix()
+            .with_prev_key()
+            .with_start_revision(start_rev);
+        let mut client = self.client.write().await;
+        let (watcher, stream) = client.watch(key_prefix, Some(opts)).await?;
+        Ok((watcher, stream))
+    }
 }