NationTech/harmony
5
0
Fork 2
Code Issues 37 Pull Requests 16 Actions Packages Projects Releases 1 Wiki Activity

refactor(host_network): extract NetworkManager as a reusable component #183

Merged
letian merged 11 commits from refactor-network-manager into master 2025-11-06 00:02:53 +00:00
Conversation 2 Commits 11 Files Changed 10 +508 -306
10 changed files with 508 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
examples/nanodc/src/main.rs
View File

@@ -1,6 +1,6 @@
use std::{
net::{IpAddr, Ipv4Addr},
sync::Arc,
sync::{Arc, OnceLock},
};
use brocade::BrocadeOptions;
@@ -107,6 +107,7 @@ async fn main() {
},
],
switch_client: switch_client.clone(),
network_manager: OnceLock::new(),
};
let inventory = Inventory {

6
examples/okd_installation/src/topology.rs
View File

@@ -9,7 +9,10 @@ use harmony::{
use harmony_macros::{ip, ipv4};
use harmony_secret::{Secret, SecretManager};
use serde::{Deserialize, Serialize};
use std::{net::IpAddr, sync::Arc};
use std::{
net::IpAddr,
sync::{Arc, OnceLock},
};
#[derive(Secret, Serialize, Deserialize, Debug, PartialEq)]
struct OPNSenseFirewallConfig {
@@ -81,6 +84,7 @@ pub async fn get_topology() -> HAClusterTopology {
},
workers: vec![],
switch_client: switch_client.clone(),
network_manager: OnceLock::new(),
}
}

6
examples/okd_pxe/src/topology.rs
View File

@@ -10,7 +10,10 @@ use harmony::{
use harmony_macros::{ip, ipv4};
use harmony_secret::{Secret, SecretManager};
use serde::{Deserialize, Serialize};
use std::{net::IpAddr, sync::Arc};
use std::{
net::IpAddr,
sync::{Arc, OnceLock},
};
pub async fn get_topology() -> HAClusterTopology {
let firewall = harmony::topology::LogicalHost {
@@ -76,6 +79,7 @@ pub async fn get_topology() -> HAClusterTopology {
},
workers: vec![],
switch_client: switch_client.clone(),
network_manager: OnceLock::new(),
}
}

3
examples/opnsense/src/main.rs
View File

@@ -1,6 +1,6 @@
use std::{
net::{IpAddr, Ipv4Addr},
sync::Arc,
sync::{Arc, OnceLock},
};
use brocade::BrocadeOptions;
@@ -79,6 +79,7 @@ async fn main() {
},
workers: vec![],
switch_client: switch_client.clone(),
network_manager: OnceLock::new(),
};
let inventory = Inventory {

309
harmony/src/domain/topology/ha_cluster.rs
View File

@@ -5,30 +5,21 @@ use harmony_types::{
net::{MacAddress, Url},
switch::PortLocation,
};
use k8s_openapi::api::core::v1::Node;
use kube::{
ResourceExt,
api::{ObjectList, ObjectMeta},
};
use log::debug;
use log::info;
use crate::modules::okd::crd::nmstate::{self, NodeNetworkConfigurationPolicy};
use crate::infra::network_manager::OpenShiftNmStateNetworkManager;
use crate::topology::PxeOptions;
use crate::{data::FileContent, modules::okd::crd::nmstate::NMState};
use crate::{
executors::ExecutorError, modules::okd::crd::nmstate::NodeNetworkConfigurationPolicySpec,
};
use crate::{data::FileContent, executors::ExecutorError};
use super::{
DHCPStaticEntry, DhcpServer, DnsRecord, DnsRecordType, DnsServer, Firewall, HostNetworkConfig,
HttpServer, IpAddress, K8sclient, LoadBalancer, LoadBalancerService, LogicalHost,
PreparationError, PreparationOutcome, Router, Switch, SwitchClient, SwitchError, TftpServer,
Topology, k8s::K8sClient,
HttpServer, IpAddress, K8sclient, LoadBalancer, LoadBalancerService, LogicalHost, NetworkError,
NetworkManager, PreparationError, PreparationOutcome, Router, Switch, SwitchClient,
SwitchError, TftpServer, Topology, k8s::K8sClient,
};
use std::collections::{BTreeMap, HashSet};
use std::sync::Arc;
use std::sync::{Arc, OnceLock};
#[derive(Debug, Clone)]
pub struct HAClusterTopology {
@@ -45,6 +36,7 @@ pub struct HAClusterTopology {
pub control_plane: Vec<LogicalHost>,
pub workers: Vec<LogicalHost>,
pub kubeconfig: Option<String>,
pub network_manager: OnceLock<Arc<dyn NetworkManager>>,
}
#[async_trait]
@@ -98,263 +90,12 @@ impl HAClusterTopology {
.to_string()
}
async fn ensure_nmstate_operator_installed(&self) -> Result<(), String> {
let k8s_client = self.k8s_client().await?;
pub async fn network_manager(&self) -> &dyn NetworkManager {
let k8s_client = self.k8s_client().await.unwrap();
debug!("Installing NMState controller...");
k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/nmstate.io_nmstates.yaml
").unwrap(), Some("nmstate"))
.await
.map_err(|e| e.to_string())?;
debug!("Creating NMState namespace...");
k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/namespace.yaml
").unwrap(), Some("nmstate"))
.await
.map_err(|e| e.to_string())?;
debug!("Creating NMState service account...");
k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/service_account.yaml
").unwrap(), Some("nmstate"))
.await
.map_err(|e| e.to_string())?;
debug!("Creating NMState role...");
k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/role.yaml
").unwrap(), Some("nmstate"))
.await
.map_err(|e| e.to_string())?;
debug!("Creating NMState role binding...");
k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/role_binding.yaml
").unwrap(), Some("nmstate"))
.await
.map_err(|e| e.to_string())?;
debug!("Creating NMState operator...");
k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/operator.yaml
").unwrap(), Some("nmstate"))
.await
.map_err(|e| e.to_string())?;
k8s_client
.wait_until_deployment_ready("nmstate-operator", Some("nmstate"), None)
.await?;
let nmstate = NMState {
metadata: ObjectMeta {
name: Some("nmstate".to_string()),
..Default::default()
},
..Default::default()
};
debug!(
"Creating NMState:\n{}",
serde_yaml::to_string(&nmstate).unwrap()
);
k8s_client
.apply(&nmstate, None)
.await
.map_err(|e| e.to_string())?;
Ok(())
}
async fn configure_bond(&self, config: &HostNetworkConfig) -> Result<(), SwitchError> {
self.ensure_nmstate_operator_installed()
.await
.map_err(|e| {
SwitchError::new(format!(
"Can't configure bond, NMState operator not available: {e}"
))
})?;
let hostname = self.get_hostname(&config.host_id).await.map_err(|e| {
SwitchError::new(format!(
"Can't configure bond, can't get hostname for host '{}': {e}",
config.host_id
))
})?;
let bond_id = self.get_next_bond_id(&hostname).await.map_err(|e| {
SwitchError::new(format!(
"Can't configure bond, can't get an available bond id for host '{}': {e}",
config.host_id
))
})?;
let bond_config = self.create_bond_configuration(&hostname, &bond_id, config);
debug!(
"Applying NMState bond config for host {}:\n{}",
serde_yaml::to_string(&bond_config).unwrap(),
config.host_id
);
self.k8s_client()
.await
.unwrap()
.apply(&bond_config, None)
.await
.map_err(|e| SwitchError::new(format!("Failed to configure bond: {e}")))?;
Ok(())
}
fn create_bond_configuration(
&self,
host: &str,
bond_name: &str,
config: &HostNetworkConfig,
) -> NodeNetworkConfigurationPolicy {
info!("Configuring bond '{bond_name}' for host '{host}'...");
let mut bond_mtu: Option<u32> = None;
let mut copy_mac_from: Option<String> = None;
let mut bond_ports = Vec::new();
let mut interfaces: Vec<nmstate::Interface> = Vec::new();
for switch_port in &config.switch_ports {
let interface_name = switch_port.interface.name.clone();
interfaces.push(nmstate::Interface {
name: interface_name.clone(),
description: Some(format!("Member of bond {bond_name}")),
r#type: nmstate::InterfaceType::Ethernet,
state: "up".to_string(),
mtu: Some(switch_port.interface.mtu),
mac_address: Some(switch_port.interface.mac_address.to_string()),
ipv4: Some(nmstate::IpStackSpec {
enabled: Some(false),
..Default::default()
}),
ipv6: Some(nmstate::IpStackSpec {
enabled: Some(false),
..Default::default()
}),
link_aggregation: None,
..Default::default()
});
bond_ports.push(interface_name.clone());
// Use the first port's details for the bond mtu and mac address
if bond_mtu.is_none() {
bond_mtu = Some(switch_port.interface.mtu);
}
if copy_mac_from.is_none() {
copy_mac_from = Some(interface_name);
}
}
interfaces.push(nmstate::Interface {
name: bond_name.to_string(),
description: Some(format!("Network bond for host {host}")),
r#type: nmstate::InterfaceType::Bond,
state: "up".to_string(),
copy_mac_from,
ipv4: Some(nmstate::IpStackSpec {
dhcp: Some(true),
enabled: Some(true),
..Default::default()
}),
ipv6: Some(nmstate::IpStackSpec {
dhcp: Some(true),
autoconf: Some(true),
enabled: Some(true),
..Default::default()
}),
link_aggregation: Some(nmstate::BondSpec {
mode: "802.3ad".to_string(),
ports: bond_ports,
..Default::default()
}),
..Default::default()
});
NodeNetworkConfigurationPolicy {
metadata: ObjectMeta {
name: Some(format!("{host}-bond-config")),
..Default::default()
},
spec: NodeNetworkConfigurationPolicySpec {
node_selector: Some(BTreeMap::from([(
"kubernetes.io/hostname".to_string(),
host.to_string(),
)])),
desired_state: nmstate::NetworkState {
interfaces,
..Default::default()
},
},
}
}
async fn get_hostname(&self, host_id: &Id) -> Result<String, String> {
let nodes: ObjectList<Node> = self
.k8s_client()
.await
.unwrap()
.list_resources(None, None)
.await
.map_err(|e| format!("Failed to list nodes: {e}"))?;
let Some(node) = nodes.iter().find(|n| {
n.status
.as_ref()
.and_then(|s| s.node_info.as_ref())
.map(|i| i.system_uuid == host_id.to_string())
.unwrap_or(false)
}) else {
return Err(format!("No node found for host '{host_id}'"));
};
node.labels()
.get("kubernetes.io/hostname")
.ok_or(format!(
"Node '{host_id}' has no kubernetes.io/hostname label"
))
.cloned()
}
async fn get_next_bond_id(&self, hostname: &str) -> Result<String, String> {
let network_state: Option<nmstate::NodeNetworkState> = self
.k8s_client()
.await
.unwrap()
.get_resource(hostname, None)
.await
.map_err(|e| format!("Failed to list nodes: {e}"))?;
let interfaces = vec![];
let existing_bonds: Vec<&nmstate::Interface> = network_state
self.network_manager
.get_or_init(|| Arc::new(OpenShiftNmStateNetworkManager::new(k8s_client.clone())))
.as_ref()
.and_then(|network_state| network_state.status.current_state.as_ref())
.map_or(&interfaces, |current_state| &current_state.interfaces)
.iter()
.filter(|i| i.r#type == nmstate::InterfaceType::Bond)
.collect();
let used_ids: HashSet<u32> = existing_bonds
.iter()
.filter_map(|i| {
i.name
.strip_prefix("bond")
.and_then(|id| id.parse::<u32>().ok())
})
.collect();
let next_id = (0..).find(|id| !used_ids.contains(id)).unwrap();
Ok(format!("bond{next_id}"))
}
async fn configure_port_channel(&self, config: &HostNetworkConfig) -> Result<(), SwitchError> {
debug!("Configuring port channel: {config:#?}");
let switch_ports = config.switch_ports.iter().map(|s| s.port.clone()).collect();
self.switch_client
.configure_port_channel(&format!("Harmony_{}", config.host_id), switch_ports)
.await
.map_err(|e| SwitchError::new(format!("Failed to configure switch: {e}")))?;
Ok(())
}
pub fn autoload() -> Self {
@@ -378,6 +119,7 @@ impl HAClusterTopology {
bootstrap_host: dummy_host,
control_plane: vec![],
workers: vec![],
johnride commented 2025-11-05 21:09:18 +00:00
Outdated
Review
Copy Link

Why OnceLock instead of Oncecell or something else? I'm not familiar with it. I guess it takes a read lock when accessed after initialization?

Why OnceLock instead of Oncecell or something else? I'm not familiar with it. I guess it takes a read lock when accessed after initialization?
letian commented 2025-11-05 23:07:38 +00:00
Outdated
Review
Copy Link

In pure Rust, the main difference is that OnceLock is the multi-threaded version of OnceCell. Considering our initialization process is synchronous in that case, I preferred to use this option rather than tokio::OnceCell (which is particularly useful for async initialization process).

But I can switch to OnceCell from tokio instead if you prefer. No strong opinions here for me.

In pure Rust, the main difference is that `OnceLock` is the multi-threaded version of `OnceCell`. Considering our initialization process is _synchronous_ in that case, I preferred to use this option rather than `tokio::OnceCell` (which is particularly useful for _async_ initialization process). But I can switch to `OnceCell` from `tokio` instead if you prefer. No strong opinions here for me.
network_manager: OnceLock::new(),
}
}
}
@@ -545,9 +287,30 @@ impl Switch for HAClusterTopology {
self.switch_client.find_port(mac_address).await
}
letian marked this conversation as resolved Outdated
johnride commented 2025-11-05 21:10:47 +00:00
Outdated
Review
Copy Link

Why is this function and logic in ha_cluster instead of network manager?

Why is this function and logic in ha_cluster instead of network manager?
letian commented 2025-11-05 23:08:26 +00:00
Outdated
Review
Copy Link

it's actually the impl Switch for HAClusterTopology

it's actually the `impl Switch for HAClusterTopology`
async fn configure_host_network(&self, config: &HostNetworkConfig) -> Result<(), SwitchError> {
self.configure_bond(config).await?;
self.configure_port_channel(config).await
async fn configure_port_channel(&self, config: &HostNetworkConfig) -> Result<(), SwitchError> {
debug!("Configuring port channel: {config:#?}");
let switch_ports = config.switch_ports.iter().map(|s| s.port.clone()).collect();
self.switch_client
.configure_port_channel(&format!("Harmony_{}", config.host_id), switch_ports)
.await
.map_err(|e| SwitchError::new(format!("Failed to configure port-channel: {e}")))?;
Ok(())
}
}
#[async_trait]
impl NetworkManager for HAClusterTopology {
async fn ensure_network_manager_installed(&self) -> Result<(), NetworkError> {
self.network_manager()
.await
.ensure_network_manager_installed()
.await
}
async fn configure_bond(&self, config: &HostNetworkConfig) -> Result<(), NetworkError> {
self.network_manager().await.configure_bond(config).await
}
}

2
harmony/src/domain/topology/k8s.rs
View File

@@ -25,7 +25,7 @@ use kube::{
api::{ApiResource, GroupVersionKind},
runtime::wait::await_condition,
};
use log::{debug, error, info, trace, warn};
use log::{debug, error, trace, warn};
use serde::{Serialize, de::DeserializeOwned};
use serde_json::json;
use similar::TextDiff;

35
harmony/src/domain/topology/network.rs
View File

@@ -15,7 +15,7 @@ use harmony_types::{
};
use serde::Serialize;
use crate::{executors::ExecutorError, hardware::PhysicalHost};
use crate::executors::ExecutorError;
use super::{LogicalHost, k8s::K8sClient};
@@ -183,6 +183,37 @@ impl FromStr for DnsRecordType {
}
}
#[async_trait]
pub trait NetworkManager: Debug + Send + Sync {
async fn ensure_network_manager_installed(&self) -> Result<(), NetworkError>;
async fn configure_bond(&self, config: &HostNetworkConfig) -> Result<(), NetworkError>;
}
#[derive(Debug, Clone, new)]
pub struct NetworkError {
johnride commented 2025-11-05 21:15:33 +00:00
Review
Copy Link

I really don't like the boilerplate added by having multiple error types for each bit of business logic.

I know passing strings around has its limitations but there's a middle ground for sure.

We should be evaluating these options sooner rather than later :

  • String
  • Specific error types like this
  • thiserror crate
  • anyhow
  • std::Error

See this comment, which I largely agree with https://www.reddit.com/r/rust/comments/1bb7dco/comment/ku7io8o/?utm_source=share&utm_medium=web3x&utm_name=web3xcss&utm_term=1&utm_content=share_button

I really don't like the boilerplate added by having multiple error types for each bit of business logic. I know passing strings around has its limitations but there's a middle ground for sure. We should be evaluating these options sooner rather than later : - String - Specific error types like this - thiserror crate - anyhow - std::Error See this comment, which I largely agree with https://www.reddit.com/r/rust/comments/1bb7dco/comment/ku7io8o/?utm_source=share&utm_medium=web3x&utm_name=web3xcss&utm_term=1&utm_content=share_button
letian commented 2025-11-05 23:21:56 +00:00
Review
Copy Link

Agree with using stuff like thiserror in general, especially for the core.

The tricky is thing is that here, it is not part of the core of Harmony strictly speaking. The Switch and the NetworkManager are extra capabilities that we offer to enrich topologies. Also, Switch and NetworkManager are 2 different capabilities. So they will surely evolve separately and change for different reasons. Which means it's tricky to have a single Error that represent both. It's not designed as such right now but a future refactoring to make it an Enum with different options would make sense to allow user to inspect the error and react accordingly.

Agree with using stuff like `thiserror` in general, especially for the core. The tricky is thing is that here, it is not part of the core of Harmony strictly speaking. The `Switch` and the `NetworkManager` are extra capabilities that we offer to enrich topologies. Also, `Switch` and `NetworkManager` are 2 different capabilities. So they will surely evolve separately and change for different reasons. Which means it's tricky to have a single `Error` that represent both. It's not designed as such right now but a future refactoring to make it an Enum with different options would make sense to allow user to inspect the error and react accordingly.
msg: String,
}
impl fmt::Display for NetworkError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.msg)
}
}
impl Error for NetworkError {}
impl From<kube::Error> for NetworkError {
fn from(value: kube::Error) -> Self {
NetworkError::new(value.to_string())
}
}
impl From<String> for NetworkError {
fn from(value: String) -> Self {
NetworkError::new(value)
}
}
#[async_trait]
pub trait Switch: Send + Sync {
async fn setup_switch(&self) -> Result<(), SwitchError>;
@@ -192,7 +223,7 @@ pub trait Switch: Send + Sync {
mac_address: &MacAddress,
) -> Result<Option<PortLocation>, SwitchError>;
async fn configure_host_network(&self, config: &HostNetworkConfig) -> Result<(), SwitchError>;
async fn configure_port_channel(&self, config: &HostNetworkConfig) -> Result<(), SwitchError>;
}
#[derive(Clone, Debug, PartialEq)]

1
harmony/src/infra/mod.rs
View File

@@ -4,5 +4,6 @@ pub mod hp_ilo;
pub mod intel_amt;
pub mod inventory;
pub mod kube;
pub mod network_manager;
pub mod opnsense;
mod sqlx;

259
harmony/src/infra/network_manager.rs Normal file
View File

@@ -0,0 +1,259 @@
use std::{
collections::{BTreeMap, HashSet},
sync::Arc,
};
use async_trait::async_trait;
use harmony_types::id::Id;
use k8s_openapi::api::core::v1::Node;
use kube::{
ResourceExt,
api::{ObjectList, ObjectMeta},
};
use log::{debug, info};
use crate::{
modules::okd::crd::nmstate,
topology::{HostNetworkConfig, NetworkError, NetworkManager, k8s::K8sClient},
};
pub struct OpenShiftNmStateNetworkManager {
k8s_client: Arc<K8sClient>,
}
impl std::fmt::Debug for OpenShiftNmStateNetworkManager {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OpenShiftNmStateNetworkManager").finish()
}
}
#[async_trait]
impl NetworkManager for OpenShiftNmStateNetworkManager {
async fn ensure_network_manager_installed(&self) -> Result<(), NetworkError> {
letian marked this conversation as resolved
johnride commented 2025-11-05 21:17:15 +00:00
Review
Copy Link

I guess this is required because the operator from the catalog does not work?

I guess this is required because the operator from the catalog does not work?
👍 1
debug!("Installing NMState controller...");
self.k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/nmstate.io_nmstates.yaml
").unwrap(), Some("nmstate"))
.await?;
debug!("Creating NMState namespace...");
self.k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/namespace.yaml
johnride commented 2025-11-05 21:17:45 +00:00
Review
Copy Link

url::parse can fail at runtime, would hurl! work here?

url::parse can fail at runtime, would hurl! work here?
letian commented 2025-11-05 23:26:10 +00:00
Review
Copy Link

I wanted to use hurl! at first but it creates an enum that match either an URL or a local folder. The url would have been nice, but I'm not sure what to do if it's a local folder. So it made the code more complicated and less elegant in the end.

We can definitely revisit the apply_url function if we really want to support both URLs and folders.

I wanted to use `hurl!` at first but it creates an enum that match either an URL or a _local folder_. The url would have been nice, but I'm not sure what to do if it's a _local folder_. So it made the code more complicated and less elegant in the end. We can definitely revisit the `apply_url` function if we really want to support both URLs and _folders_.
").unwrap(), Some("nmstate"))
.await?;
debug!("Creating NMState service account...");
self.k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/service_account.yaml
").unwrap(), Some("nmstate"))
.await?;
debug!("Creating NMState role...");
self.k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/role.yaml
").unwrap(), Some("nmstate"))
.await?;
debug!("Creating NMState role binding...");
self.k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/role_binding.yaml
").unwrap(), Some("nmstate"))
.await?;
debug!("Creating NMState operator...");
self.k8s_client.apply_url(url::Url::parse("https://github.com/nmstate/kubernetes-nmstate/releases/download/v0.84.0/operator.yaml
").unwrap(), Some("nmstate"))
.await?;
self.k8s_client
.wait_until_deployment_ready("nmstate-operator", Some("nmstate"), None)
.await?;
let nmstate = nmstate::NMState {
metadata: ObjectMeta {
name: Some("nmstate".to_string()),
..Default::default()
},
..Default::default()
};
debug!(
"Creating NMState:\n{}",
serde_yaml::to_string(&nmstate).unwrap()
);
self.k8s_client.apply(&nmstate, None).await?;
Ok(())
}
async fn configure_bond(&self, config: &HostNetworkConfig) -> Result<(), NetworkError> {
let hostname = self.get_hostname(&config.host_id).await.map_err(|e| {
NetworkError::new(format!(
"Can't configure bond, can't get hostname for host '{}': {e}",
config.host_id
))
})?;
let bond_id = self.get_next_bond_id(&hostname).await.map_err(|e| {
NetworkError::new(format!(
"Can't configure bond, can't get an available bond id for host '{}': {e}",
config.host_id
))
})?;
let bond_config = self.create_bond_configuration(&hostname, &bond_id, config);
debug!(
"Applying NMState bond config for host {}:\n{}",
config.host_id,
serde_yaml::to_string(&bond_config).unwrap(),
);
self.k8s_client
.apply(&bond_config, None)
.await
.map_err(|e| NetworkError::new(format!("Failed to configure bond: {e}")))?;
Ok(())
}
}
impl OpenShiftNmStateNetworkManager {
pub fn new(k8s_client: Arc<K8sClient>) -> Self {
Self { k8s_client }
}
fn create_bond_configuration(
&self,
host: &str,
bond_name: &str,
config: &HostNetworkConfig,
) -> nmstate::NodeNetworkConfigurationPolicy {
info!("Configuring bond '{bond_name}' for host '{host}'...");
let mut bond_mtu: Option<u32> = None;
let mut copy_mac_from: Option<String> = None;
let mut bond_ports = Vec::new();
let mut interfaces: Vec<nmstate::Interface> = Vec::new();
for switch_port in &config.switch_ports {
let interface_name = switch_port.interface.name.clone();
interfaces.push(nmstate::Interface {
name: interface_name.clone(),
description: Some(format!("Member of bond {bond_name}")),
r#type: nmstate::InterfaceType::Ethernet,
state: "up".to_string(),
mtu: Some(switch_port.interface.mtu),
mac_address: Some(switch_port.interface.mac_address.to_string()),
ipv4: Some(nmstate::IpStackSpec {
enabled: Some(false),
..Default::default()
}),
ipv6: Some(nmstate::IpStackSpec {
enabled: Some(false),
..Default::default()
}),
link_aggregation: None,
..Default::default()
});
bond_ports.push(interface_name.clone());
// Use the first port's details for the bond mtu and mac address
if bond_mtu.is_none() {
bond_mtu = Some(switch_port.interface.mtu);
}
if copy_mac_from.is_none() {
copy_mac_from = Some(interface_name);
}
}
interfaces.push(nmstate::Interface {
name: bond_name.to_string(),
description: Some(format!("Network bond for host {host}")),
r#type: nmstate::InterfaceType::Bond,
state: "up".to_string(),
copy_mac_from,
ipv4: Some(nmstate::IpStackSpec {
dhcp: Some(true),
enabled: Some(true),
..Default::default()
}),
ipv6: Some(nmstate::IpStackSpec {
dhcp: Some(true),
autoconf: Some(true),
enabled: Some(true),
..Default::default()
}),
link_aggregation: Some(nmstate::BondSpec {
mode: "802.3ad".to_string(),
ports: bond_ports,
..Default::default()
}),
..Default::default()
});
nmstate::NodeNetworkConfigurationPolicy {
metadata: ObjectMeta {
name: Some(format!("{host}-bond-config")),
..Default::default()
},
spec: nmstate::NodeNetworkConfigurationPolicySpec {
node_selector: Some(BTreeMap::from([(
"kubernetes.io/hostname".to_string(),
host.to_string(),
)])),
desired_state: nmstate::NetworkState {
interfaces,
..Default::default()
},
},
}
}
async fn get_hostname(&self, host_id: &Id) -> Result<String, String> {
let nodes: ObjectList<Node> = self
.k8s_client
.list_resources(None, None)
.await
.map_err(|e| format!("Failed to list nodes: {e}"))?;
let Some(node) = nodes.iter().find(|n| {
n.status
.as_ref()
.and_then(|s| s.node_info.as_ref())
.map(|i| i.system_uuid == host_id.to_string())
.unwrap_or(false)
}) else {
return Err(format!("No node found for host '{host_id}'"));
};
node.labels()
.get("kubernetes.io/hostname")
.ok_or(format!(
"Node '{host_id}' has no kubernetes.io/hostname label"
))
.cloned()
}
async fn get_next_bond_id(&self, hostname: &str) -> Result<String, String> {
let network_state: Option<nmstate::NodeNetworkState> = self
.k8s_client
.get_resource(hostname, None)
.await
.map_err(|e| format!("Failed to list nodes: {e}"))?;
let interfaces = vec![];
let existing_bonds: Vec<&nmstate::Interface> = network_state
.as_ref()
.and_then(|network_state| network_state.status.current_state.as_ref())
.map_or(&interfaces, |current_state| &current_state.interfaces)
.iter()
.filter(|i| i.r#type == nmstate::InterfaceType::Bond)
.collect();
let used_ids: HashSet<u32> = existing_bonds
.iter()
.filter_map(|i| {
i.name
.strip_prefix("bond")
.and_then(|id| id.parse::<u32>().ok())
})
.collect();
let next_id = (0..).find(|id| !used_ids.contains(id)).unwrap();
Ok(format!("bond{next_id}"))
}
}

190
harmony/src/modules/okd/host_network.rs
View File

@@ -1,6 +1,6 @@
use async_trait::async_trait;
use harmony_types::id::Id;
use log::{debug, info};
use log::info;
use serde::Serialize;
use crate::{
@@ -9,7 +9,7 @@ use crate::{
interpret::{Interpret, InterpretError, InterpretName, InterpretStatus, Outcome},
inventory::Inventory,
score::Score,
topology::{HostNetworkConfig, NetworkInterface, Switch, SwitchPort, Topology},
topology::{HostNetworkConfig, NetworkInterface, NetworkManager, Switch, SwitchPort, Topology},
};
#[derive(Debug, Clone, Serialize)]
@@ -17,7 +17,7 @@ pub struct HostNetworkConfigurationScore {
pub hosts: Vec<PhysicalHost>,
}
impl<T: Topology + Switch> Score<T> for HostNetworkConfigurationScore {
impl<T: Topology + NetworkManager + Switch> Score<T> for HostNetworkConfigurationScore {
fn name(&self) -> String {
"HostNetworkConfigurationScore".into()
}
@@ -35,7 +35,7 @@ pub struct HostNetworkConfigurationInterpret {
}
impl HostNetworkConfigurationInterpret {
async fn configure_network_for_host<T: Topology + Switch>(
async fn configure_network_for_host<T: Topology + NetworkManager + Switch>(
&self,
topology: &T,
host: &PhysicalHost,
@@ -67,10 +67,15 @@ impl HostNetworkConfigurationInterpret {
);
info!("[Host {current_host}/{total_hosts}] Configuring host network...");
topology.configure_bond(&config).await.map_err(|e| {
InterpretError::new(format!("Failed to configure host network: {e}"))
})?;
topology
.configure_host_network(&config)
.configure_port_channel(&config)
.await
.map_err(|e| InterpretError::new(format!("Failed to configure host: {e}")))?;
.map_err(|e| {
InterpretError::new(format!("Failed to configure host network: {e}"))
})?;
} else {
info!(
"[Host {current_host}/{total_hosts}] No ports found for {} interfaces, skipping",
@@ -113,7 +118,7 @@ impl HostNetworkConfigurationInterpret {
port,
});
}
Ok(None) => debug!("No port found for '{mac_address}', skipping"),
Ok(None) => {}
Err(e) => {
return Err(InterpretError::new(format!(
"Failed to get port for host '{}': {}",
@@ -169,7 +174,7 @@ impl HostNetworkConfigurationInterpret {
}
#[async_trait]
impl<T: Topology + Switch> Interpret<T> for HostNetworkConfigurationInterpret {
impl<T: Topology + NetworkManager + Switch> Interpret<T> for HostNetworkConfigurationInterpret {
fn get_name(&self) -> InterpretName {
InterpretName::Custom("HostNetworkConfigurationInterpret")
}
@@ -198,6 +203,12 @@ impl<T: Topology + Switch> Interpret<T> for HostNetworkConfigurationInterpret {
let host_count = self.score.hosts.len();
info!("Started network configuration for {host_count} host(s)...",);
info!("Setting up NetworkManager...",);
topology
.ensure_network_manager_installed()
.await
.map_err(|e| InterpretError::new(format!("NetworkManager setup failed: {e}")))?;
info!("Setting up switch with sane defaults...");
topology
.setup_switch()
@@ -242,7 +253,8 @@ mod tests {
use crate::{
hardware::HostCategory,
topology::{
HostNetworkConfig, PreparationError, PreparationOutcome, SwitchError, SwitchPort,
HostNetworkConfig, NetworkError, PreparationError, PreparationOutcome, SwitchError,
SwitchPort,
},
};
use std::{
@@ -290,15 +302,27 @@ mod tests {
}
#[tokio::test]
async fn host_with_one_mac_address_should_create_bond_with_one_interface() {
async fn should_setup_network_manager() {
let host = given_host(&HOST_ID, vec![EXISTING_INTERFACE.clone()]);
let score = given_score(vec![host]);
let topology = TopologyWithSwitch::new();
let _ = score.interpret(&Inventory::empty(), &topology).await;
let configured_host_networks = topology.configured_host_networks.lock().unwrap();
assert_that!(*configured_host_networks).contains_exactly(vec![(
let network_manager_setup = topology.network_manager_setup.lock().unwrap();
assert_that!(*network_manager_setup).is_true();
}
#[tokio::test]
async fn host_with_one_mac_address_should_configure_bond_with_one_interface() {
let host = given_host(&HOST_ID, vec![EXISTING_INTERFACE.clone()]);
let score = given_score(vec![host]);
let topology = TopologyWithSwitch::new();
let _ = score.interpret(&Inventory::empty(), &topology).await;
let config = topology.configured_bonds.lock().unwrap();
assert_that!(*config).contains_exactly(vec![(
HOST_ID.clone(),
HostNetworkConfig {
host_id: HOST_ID.clone(),
@@ -311,7 +335,28 @@ mod tests {
}
#[tokio::test]
async fn host_with_multiple_mac_addresses_should_create_one_bond_with_all_interfaces() {
async fn host_with_one_mac_address_should_configure_port_channel_with_one_interface() {
let host = given_host(&HOST_ID, vec![EXISTING_INTERFACE.clone()]);
let score = given_score(vec![host]);
let topology = TopologyWithSwitch::new();
let _ = score.interpret(&Inventory::empty(), &topology).await;
let config = topology.configured_port_channels.lock().unwrap();
assert_that!(*config).contains_exactly(vec![(
HOST_ID.clone(),
HostNetworkConfig {
host_id: HOST_ID.clone(),
switch_ports: vec![SwitchPort {
interface: EXISTING_INTERFACE.clone(),
port: PORT.clone(),
}],
},
)]);
}
#[tokio::test]
async fn host_with_multiple_mac_addresses_should_configure_one_bond_with_all_interfaces() {
let score = given_score(vec![given_host(
&HOST_ID,
vec![
@@ -323,8 +368,8 @@ mod tests {
let _ = score.interpret(&Inventory::empty(), &topology).await;
let configured_host_networks = topology.configured_host_networks.lock().unwrap();
assert_that!(*configured_host_networks).contains_exactly(vec![(
let config = topology.configured_bonds.lock().unwrap();
assert_that!(*config).contains_exactly(vec![(
HOST_ID.clone(),
HostNetworkConfig {
host_id: HOST_ID.clone(),
@@ -343,7 +388,40 @@ mod tests {
}
#[tokio::test]
async fn multiple_hosts_should_create_one_bond_per_host() {
async fn host_with_multiple_mac_addresses_should_configure_one_port_channel_with_all_interfaces()
{
let score = given_score(vec![given_host(
&HOST_ID,
vec![
EXISTING_INTERFACE.clone(),
ANOTHER_EXISTING_INTERFACE.clone(),
],
)]);
let topology = TopologyWithSwitch::new();
let _ = score.interpret(&Inventory::empty(), &topology).await;
let config = topology.configured_port_channels.lock().unwrap();
assert_that!(*config).contains_exactly(vec![(
HOST_ID.clone(),
HostNetworkConfig {
host_id: HOST_ID.clone(),
switch_ports: vec![
SwitchPort {
interface: EXISTING_INTERFACE.clone(),
port: PORT.clone(),
},
SwitchPort {
interface: ANOTHER_EXISTING_INTERFACE.clone(),
port: ANOTHER_PORT.clone(),
},
],
},
)]);
}
#[tokio::test]
async fn multiple_hosts_should_configure_one_bond_per_host() {
let score = given_score(vec![
given_host(&HOST_ID, vec![EXISTING_INTERFACE.clone()]),
given_host(&ANOTHER_HOST_ID, vec![ANOTHER_EXISTING_INTERFACE.clone()]),
@@ -352,8 +430,43 @@ mod tests {
let _ = score.interpret(&Inventory::empty(), &topology).await;
let configured_host_networks = topology.configured_host_networks.lock().unwrap();
assert_that!(*configured_host_networks).contains_exactly(vec![
let config = topology.configured_bonds.lock().unwrap();
assert_that!(*config).contains_exactly(vec![
(
HOST_ID.clone(),
HostNetworkConfig {
host_id: HOST_ID.clone(),
switch_ports: vec![SwitchPort {
interface: EXISTING_INTERFACE.clone(),
port: PORT.clone(),
}],
},
),
(
ANOTHER_HOST_ID.clone(),
HostNetworkConfig {
host_id: ANOTHER_HOST_ID.clone(),
switch_ports: vec![SwitchPort {
interface: ANOTHER_EXISTING_INTERFACE.clone(),
port: ANOTHER_PORT.clone(),
}],
},
),
]);
}
#[tokio::test]
async fn multiple_hosts_should_configure_one_port_channel_per_host() {
let score = given_score(vec![
given_host(&HOST_ID, vec![EXISTING_INTERFACE.clone()]),
given_host(&ANOTHER_HOST_ID, vec![ANOTHER_EXISTING_INTERFACE.clone()]),
]);
let topology = TopologyWithSwitch::new();
let _ = score.interpret(&Inventory::empty(), &topology).await;
let config = topology.configured_port_channels.lock().unwrap();
assert_that!(*config).contains_exactly(vec![
(
HOST_ID.clone(),
HostNetworkConfig {
@@ -384,8 +497,10 @@ mod tests {
let _ = score.interpret(&Inventory::empty(), &topology).await;
let configured_host_networks = topology.configured_host_networks.lock().unwrap();
assert_that!(*configured_host_networks).is_empty();
let config = topology.configured_port_channels.lock().unwrap();
assert_that!(*config).is_empty();
let config = topology.configured_bonds.lock().unwrap();
assert_that!(*config).is_empty();
}
fn given_score(hosts: Vec<PhysicalHost>) -> HostNetworkConfigurationScore {
@@ -422,26 +537,33 @@ mod tests {
}
}
#[derive(Debug)]
struct TopologyWithSwitch {
available_ports: Arc<Mutex<Vec<PortLocation>>>,
configured_host_networks: Arc<Mutex<Vec<(Id, HostNetworkConfig)>>>,
configured_port_channels: Arc<Mutex<Vec<(Id, HostNetworkConfig)>>>,
switch_setup: Arc<Mutex<bool>>,
network_manager_setup: Arc<Mutex<bool>>,
configured_bonds: Arc<Mutex<Vec<(Id, HostNetworkConfig)>>>,
}
impl TopologyWithSwitch {
fn new() -> Self {
Self {
available_ports: Arc::new(Mutex::new(vec![PORT.clone(), ANOTHER_PORT.clone()])),
configured_host_networks: Arc::new(Mutex::new(vec![])),
configured_port_channels: Arc::new(Mutex::new(vec![])),
switch_setup: Arc::new(Mutex::new(false)),
network_manager_setup: Arc::new(Mutex::new(false)),
configured_bonds: Arc::new(Mutex::new(vec![])),
}
}
fn new_port_not_found() -> Self {
Self {
available_ports: Arc::new(Mutex::new(vec![])),
configured_host_networks: Arc::new(Mutex::new(vec![])),
configured_port_channels: Arc::new(Mutex::new(vec![])),
switch_setup: Arc::new(Mutex::new(false)),
network_manager_setup: Arc::new(Mutex::new(false)),
configured_bonds: Arc::new(Mutex::new(vec![])),
}
}
}
@@ -457,6 +579,22 @@ mod tests {
}
}
#[async_trait]
impl NetworkManager for TopologyWithSwitch {
async fn ensure_network_manager_installed(&self) -> Result<(), NetworkError> {
let mut network_manager_installed = self.network_manager_setup.lock().unwrap();
*network_manager_installed = true;
Ok(())
}
async fn configure_bond(&self, config: &HostNetworkConfig) -> Result<(), NetworkError> {
let mut configured_bonds = self.configured_bonds.lock().unwrap();
configured_bonds.push((config.host_id.clone(), config.clone()));
Ok(())
}
}
#[async_trait]
impl Switch for TopologyWithSwitch {
async fn setup_switch(&self) -> Result<(), SwitchError> {
@@ -476,12 +614,12 @@ mod tests {
Ok(Some(ports.remove(0)))
}
async fn configure_host_network(
async fn configure_port_channel(
&self,
config: &HostNetworkConfig,
) -> Result<(), SwitchError> {
let mut configured_host_networks = self.configured_host_networks.lock().unwrap();
configured_host_networks.push((config.host_id.clone(), config.clone()));
let mut configured_port_channels = self.configured_port_channels.lock().unwrap();
configured_port_channels.push((config.host_id.clone(), config.clone()));
Ok(())
}