harmony/src/modules/storage/ceph/ceph_osd_replacement_score.rs

@@ -49,6 +49,7 @@ impl<T: Topology + K8sclient> Interpret<T> for CephRemoveOsdInterpret {
         topology: &T,
     ) -> Result<Outcome, InterpretError> {
         let client = topology.k8s_client().await.unwrap();
+        self.verify_ceph_toolbox_exists(client.clone()).await?;
         self.scale_deployment(client.clone()).await?;
         self.verify_deployment_scaled(client.clone()).await?;
         self.delete_deployment(client.clone()).await?;
@@ -103,11 +104,45 @@ impl CephRemoveOsdInterpret {
         Ok(osd_id_full)
     }
 
-    fn build_timer(&self) -> (Duration, Duration, Instant) {
-        let timeout = Duration::from_secs(120);
-        let interval = Duration::from_secs(5);
-        let start = Instant::now();
-        (timeout, interval, start)
+    pub async fn verify_ceph_toolbox_exists(
+        &self,
+        client: Arc<K8sClient>,
+    ) -> Result<Outcome, InterpretError> {
+        let toolbox_dep = "rook-ceph-tools".to_string();
+
+        match client
+            .get_deployment(&toolbox_dep, Some(&self.score.rook_ceph_namespace))
+            .await
+        {
+            Ok(Some(deployment)) => {
+                if let Some(status) = deployment.status {
+                    let ready_count = status.ready_replicas.unwrap_or(0);
+                    if ready_count >= 1 {
+                        return Ok(Outcome::success(format!(
+                            "'{}' is ready with {} replica(s).",
+                            &toolbox_dep, ready_count
+                        )));
+                    } else {
+                        return Err(InterpretError::new(
+                            "ceph-tool-box not ready in cluster".to_string(),
+                        ));
+                    }
+                } else {
+                    Err(InterpretError::new(format!(
+                        "failed to get deployment status {}",
+                        &toolbox_dep
+                    )))
+                }
+            }
+            Ok(None) => Err(InterpretError::new(format!(
+                "Deployment '{}' not found in namespace '{}'.",
+                &toolbox_dep, self.score.rook_ceph_namespace
+            ))),
+            Err(e) => Err(InterpretError::new(format!(
+                "Failed to query for deployment '{}': {}",
+                &toolbox_dep, e
+            ))),
+        }
     }
 
     pub async fn scale_deployment(
@@ -167,6 +202,12 @@ impl CephRemoveOsdInterpret {
         }
     }
 
+    fn build_timer(&self) -> (Duration, Duration, Instant) {
+        let timeout = Duration::from_secs(120);
+        let interval = Duration::from_secs(5);
+        let start = Instant::now();
+        (timeout, interval, start)
+    }
     pub async fn delete_deployment(
         &self,
         client: Arc<K8sClient>,
@@ -227,9 +268,11 @@ impl CephRemoveOsdInterpret {
         let nodes = json.get("nodes").ok_or_else(|| {
             InterpretError::new("Missing 'nodes' field in ceph osd tree JSON".to_string())
         })?;
-        let tree: CephOsdTree = CephOsdTree{ nodes: serde_json::from_value(nodes.clone()).map_err(|e| {
-            InterpretError::new(format!("Failed to parse ceph osd tree JSON: {}", e))
-        })?,};
+        let tree: CephOsdTree = CephOsdTree {
+            nodes: serde_json::from_value(nodes.clone()).map_err(|e| {
+                InterpretError::new(format!("Failed to parse ceph osd tree JSON: {}", e))
+            })?,
+        };
         Ok(tree)
     }