From 90f3c99586cd4eb62fca897f031158e68a550979 Mon Sep 17 00:00:00 2001 From: Aphelios01-sdk <175563482+Aphelios01-sdk@users.noreply.github.com> Date: Sat, 3 Oct 2026 19:06:39 +0700 Subject: [PATCH] fix(operator): prevent panic on empty ingress, improve concurrency, and add unit tests --- crates/operator-common/src/lib.rs | 46 ++++++++ .../src/types/load_balancer.rs | 108 +++++++++++++++--- crates/operator/src/crd.rs | 23 +++- crates/operator/src/identity.rs | 34 ++++++ crates/operator/src/types/bootstrap.rs | 7 -- crates/operator/src/types/common.rs | 40 +++++++ crates/operator/src/types/storage.rs | 4 +- 7 files changed, 236 insertions(+), 26 deletions(-) diff --git a/crates/operator-common/src/lib.rs b/crates/operator-common/src/lib.rs index 47df503..03afad2 100644 --- a/crates/operator-common/src/lib.rs +++ b/crates/operator-common/src/lib.rs @@ -111,3 +111,49 @@ pub fn selector_labels(name: String, kind: String) -> BTreeMap { 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"); + } +} diff --git a/crates/operator-common/src/types/load_balancer.rs b/crates/operator-common/src/types/load_balancer.rs index ecf3b6c..731ed34 100644 --- a/crates/operator-common/src/types/load_balancer.rs +++ b/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 { && 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))); + } +} diff --git a/crates/operator/src/crd.rs b/crates/operator/src/crd.rs index 4adcc6e..a96814b 100644 --- a/crates/operator/src/crd.rs +++ b/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())); + } +} diff --git a/crates/operator/src/identity.rs b/crates/operator/src/identity.rs index 9971cab..2e7b0fe 100644 --- a/crates/operator/src/identity.rs +++ b/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); + } +} diff --git a/crates/operator/src/types/bootstrap.rs b/crates/operator/src/types/bootstrap.rs index 2b071c2..875a250 100644 --- a/crates/operator/src/types/bootstrap.rs +++ b/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 } diff --git a/crates/operator/src/types/common.rs b/crates/operator/src/types/common.rs index 56a7bf1..6b777ce 100644 --- a/crates/operator/src/types/common.rs +++ b/crates/operator/src/types/common.rs @@ -325,3 +325,43 @@ pub fn generate_container_ports(ports: Vec) -> Vec { }) .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); + } +} diff --git a/crates/operator/src/types/storage.rs b/crates/operator/src/types/storage.rs index 1ebafef..7f6e1f9 100644 --- a/crates/operator/src/types/storage.rs +++ b/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, })); }