fix(operator): prevent panic on empty ingress, improve concurrency, and add unit tests

crates/operator-common/src/lib.rs

@@ -111,3 +111,49 @@ pub fn selector_labels(name: String, kind: String) -> BTreeMap<String, String> {
 pub fn external_address_name(name: &str) -> String {
     format!("{name}-external-addresses")
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_selector_labels() {
+        let labels = selector_labels("my-node".into(), "storage".into());
+        assert_eq!(
+            labels.get("app.kubernetes.io/name").unwrap(),
+            "ipfs-storage-cluster"
+        );
+        assert_eq!(labels.get("app.kubernetes.io/instance").unwrap(), "my-node");
+        assert_eq!(labels.len(), 2);
+    }
+
+    #[test]
+    fn test_labels() {
+        let labels = labels("my-node".into(), "bootstrap".into());
+        assert_eq!(
+            labels.get("app.kubernetes.io/name").unwrap(),
+            "ipfs-bootstrap-cluster"
+        );
+        assert_eq!(labels.get("app.kubernetes.io/instance").unwrap(), "my-node");
+        assert_eq!(labels.get("app.kubernetes.io/version").unwrap(), "0.2.0");
+        assert_eq!(
+            labels.get("app.kubernetes.io/managed-by").unwrap(),
+            "ipfs-operator"
+        );
+        assert_eq!(labels.len(), 4);
+    }
+
+    #[test]
+    fn test_external_address_name() {
+        assert_eq!(
+            external_address_name("cluster1"),
+            "cluster1-external-addresses"
+        );
+    }
+
+    #[test]
+    fn test_error_metric_label() {
+        let err = Error::IllegalDocument;
+        assert_eq!(err.metric_label(), "illegaldocument");
+    }
+}

crates/operator-common/src/types/load_balancer.rs

@@ -3,7 +3,7 @@ use kube::{Api, Client, Error, ResourceExt, api::ListParams, core::ErrorResponse
 use kube_runtime::wait::{Condition, await_condition};
 use std::{collections::BTreeMap, time::Duration};
 use tokio::task::JoinSet;
-use tracing::{error, instrument};
+use tracing::instrument;
 
 use crate::{
     labels, selector_labels,
@@ -138,16 +138,26 @@ pub async fn wait(
     let exists = await_condition(service_api, name.as_str(), external_ip_exists());
     let out = tokio::time::timeout(Duration::from_secs(300), exists).await?;
     match out {
-        Ok(res) => match res.unwrap().status.unwrap().load_balancer.unwrap().ingress {
-            Some(ingress) => {
-                if !ingress.is_empty() {
-                    return Ok(ingress[0].clone().ip.unwrap());
-                } else {
-                    Err(crate::Error::IngressListEmpty)
+        Ok(svc) => {
+            let status = svc
+                .and_then(|s| s.status)
+                .and_then(|st| st.load_balancer)
+                .ok_or(crate::Error::IngressListMissing)?;
+            match status.ingress {
+                Some(ingress) => {
+                    if let Some(first) = ingress.first() {
+                        if let Some(ip) = &first.ip {
+                            Ok(ip.clone())
+                        } else {
+                            Err(crate::Error::IPTimeout)
+                        }
+                    } else {
+                        Err(crate::Error::IngressListEmpty)
+                    }
                 }
+                None => Err(crate::Error::IngressListMissing),
             }
-            None => Err(crate::Error::IngressListMissing),
-        },
+        }
         Err(e) => Err(crate::Error::WaitError { source: e }),
     }
 }
@@ -159,7 +169,8 @@ fn external_ip_exists() -> impl Condition<Service> {
             && let Some(status) = &svc.status
             && let Some(lb) = &status.load_balancer
             && let Some(ingress) = &lb.ingress
-            && let Some(_ip) = &ingress[0].ip
+            && let Some(first) = ingress.first()
+            && first.ip.is_some()
         {
             return true;
         }
@@ -198,14 +209,79 @@ async fn _create(
             ports.clone(),
             (labels(name.clone(), kind.clone().to_string()), sl),
         ));
+    }
 
-        while let Some(res) = set.join_next().await {
-            match res {
-                Ok(_) => (),
-                Err(e) => error!(error = e.to_string()),
-            }
-        }
+    while let Some(res) = set.join_next().await {
+        res??;
     }
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use k8s_openapi::api::core::v1::{LoadBalancerIngress, LoadBalancerStatus, ServiceStatus};
+
+    #[test]
+    fn test_external_ip_exists_condition() {
+        let condition = external_ip_exists();
+
+        // None service
+        assert!(!condition.matches_object(None));
+
+        // Service with None status
+        let svc_no_status = Service::default();
+        assert!(!condition.matches_object(Some(&svc_no_status)));
+
+        // Service with status but None load_balancer
+        let svc_no_lb = Service {
+            status: Some(ServiceStatus::default()),
+            ..Default::default()
+        };
+        assert!(!condition.matches_object(Some(&svc_no_lb)));
+
+        // Service with empty ingress list (Must NOT panic!)
+        let svc_empty_ingress = Service {
+            status: Some(ServiceStatus {
+                load_balancer: Some(LoadBalancerStatus {
+                    ingress: Some(vec![]),
+                }),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        assert!(!condition.matches_object(Some(&svc_empty_ingress)));
+
+        // Service with ingress having None ip
+        let svc_no_ip = Service {
+            status: Some(ServiceStatus {
+                load_balancer: Some(LoadBalancerStatus {
+                    ingress: Some(vec![LoadBalancerIngress {
+                        hostname: Some("example.com".into()),
+                        ip: None,
+                        ..Default::default()
+                    }]),
+                }),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        assert!(!condition.matches_object(Some(&svc_no_ip)));
+
+        // Service with valid IP
+        let svc_with_ip = Service {
+            status: Some(ServiceStatus {
+                load_balancer: Some(LoadBalancerStatus {
+                    ingress: Some(vec![LoadBalancerIngress {
+                        ip: Some("192.168.1.100".into()),
+                        ..Default::default()
+                    }]),
+                }),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        assert!(condition.matches_object(Some(&svc_with_ip)));
+    }
+}

crates/operator/src/crd.rs

@@ -135,7 +135,6 @@ impl Default for Persistence {
 //     pub service_port: i32,
 // }
 
-// Context for our reconciler
 #[derive(Clone)]
 pub struct Context {
     /// Kubernetes client
@@ -145,3 +144,25 @@ pub struct Context {
     /// Prometheus metrics
     pub metrics: Metrics,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_node_kind_display() {
+        assert_eq!(NodeKind::BootStrap.to_string(), "bootstrap");
+        assert_eq!(NodeKind::Storage.to_string(), "storage");
+    }
+
+    #[test]
+    fn test_node_spec_default() {
+        let spec = NodeSpec::default();
+        assert_eq!(spec.replicas, 1);
+        assert_eq!(spec.rust_log, "info");
+        assert_eq!(spec.kind, NodeKind::Storage);
+        assert_eq!(spec.p2p_port, Some(4001));
+        assert_eq!(spec.bootstrap_name, None);
+        assert_eq!(spec.ipfs.image.tag, Some("latest".into()));
+    }
+}

crates/operator/src/identity.rs

@@ -145,3 +145,37 @@ impl Display for Identity {
         write!(f, "{id}")
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_identity_creation_and_serialization() {
+        let id = Identity::new().expect("Failed to create identity");
+        assert!(!id.peer_id.is_empty());
+        assert!(!id.priv_key.is_empty());
+
+        let json = serde_json::to_string(&id).expect("Failed to serialize");
+        let deserialized: Identity = serde_json::from_str(&json).expect("Failed to deserialize");
+        assert_eq!(id.peer_id, deserialized.peer_id);
+        assert_eq!(id.priv_key, deserialized.priv_key);
+    }
+
+    #[test]
+    fn test_identity_map_conversion() {
+        let id1 = Identity::new().unwrap();
+        let id2 = Identity::new().unwrap();
+        let mut map = BTreeMap::new();
+        map.insert("node-0".to_string(), id1);
+        map.insert("node-1".to_string(), id2);
+
+        let str_map = identity_to_string(&map).unwrap();
+        assert_eq!(str_map.len(), 2);
+
+        let parsed_map = string_to_identity(&str_map).unwrap();
+        assert_eq!(parsed_map.len(), 2);
+        assert_eq!(parsed_map["node-0"].peer_id, map["node-0"].peer_id);
+        assert_eq!(parsed_map["node-1"].peer_id, map["node-1"].peer_id);
+    }
+}

crates/operator/src/types/bootstrap.rs

@@ -105,12 +105,5 @@ pub async fn delete(client: Client, name: String, namespace: String) -> Result<(
     )
     .await?;
 
-    configmap::delete(
-        client.clone(),
-        format!("{name}-identities"),
-        namespace.clone(),
-    )
-    .await?;
-
     load_balancer::delete(client, name, namespace).await
 }

crates/operator/src/types/common.rs

@@ -325,3 +325,43 @@ pub fn generate_container_ports(ports: Vec<Port>) -> Vec<ContainerPort> {
         })
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_ipfs_ports() {
+        let ports = ipfs_ports();
+        assert_eq!(ports.len(), 4);
+        assert_eq!(ports[0].name, "swarm");
+        assert_eq!(ports[0].port, 4001);
+        assert_eq!(ports[1].name, "api");
+        assert_eq!(ports[1].port, 5001);
+        assert_eq!(ports[2].name, "ws");
+        assert_eq!(ports[2].port, 8081);
+        assert_eq!(ports[3].name, "http");
+        assert_eq!(ports[3].port, 8080);
+    }
+
+    #[test]
+    fn test_ipfs_cluster_ports() {
+        let ports = ipfs_cluster_ports();
+        assert_eq!(ports.len(), 3);
+        assert_eq!(ports[0].name, "api-http");
+        assert_eq!(ports[0].port, 9094);
+        assert_eq!(ports[1].name, "proxy-http");
+        assert_eq!(ports[1].port, 9095);
+        assert_eq!(ports[2].name, "cluster-swarm");
+        assert_eq!(ports[2].port, 9096);
+    }
+
+    #[test]
+    fn test_generate_container_ports() {
+        let ports = ipfs_ports();
+        let container_ports = generate_container_ports(ports);
+        assert_eq!(container_ports.len(), 4);
+        assert_eq!(container_ports[0].name.as_deref(), Some("swarm"));
+        assert_eq!(container_ports[0].container_port, 4001);
+    }
+}

crates/operator/src/types/storage.rs

@@ -31,7 +31,7 @@ pub async fn deploy(
             return Err(Error::Api(ErrorResponse {
                 status: "Failed".to_string(),
                 message: err.to_string(),
-                reason: "Failed to create load balancers for p2p port".to_string(),
+                reason: "Failed to upsert peer info secret".to_string(),
                 code: 418,
             }));
         }
@@ -43,7 +43,7 @@ pub async fn deploy(
             return Err(Error::Api(ErrorResponse {
                 status: "Failed".to_string(),
                 message: err.to_string(),
-                reason: "Failed to create load balancers for p2p port".to_string(),
+                reason: "Failed to upsert cluster secret".to_string(),
                 code: 418,
             }));
         }