NationTech/harmony
6
0
Fork 1
Code Issues 35 Pull Requests 6 Actions Packages Projects Releases 1 Wiki Activity

wip: saving harmony inventory, currently messing with async stuff, properly understanding stuff now so I should fix it soon. The recv in the inventory agent is sync and blocking the whole thread so the request cannot be sent until the recv is killed, which is wrong. Will fix this by isolating on another thread

Browse Source
This commit is contained in:
Jean-Gabriel Gill-Couture 2025-08-31 00:31:55 -04:00
parent e548bf619a
commit d9c26f43ee
12 changed files with 233 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -87,8 +87,7 @@ async fn main() {
let inventory = Inventory {
location: Location::new("I am mobile".to_string(), "earth".to_string()),
switch: SwitchGroup::from([]),
firewall: FirewallGroup::from([PhysicalHost::empty(HostCategory::Firewall)
.management(Arc::new(OPNSenseManagementInterface::new()))]),
firewall_mgmt: Box::new(OPNSenseManagementInterface::new()),
storage_host: vec![],
worker_host: vec![
PhysicalHost::empty(HostCategory::Server)

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

@ -69,8 +69,7 @@ pub fn get_inventory() -> Inventory {
"testopnsense".to_string(),
),
switch: SwitchGroup::from([]),
firewall: FirewallGroup::from([PhysicalHost::empty(HostCategory::Firewall)
.management(Arc::new(OPNSenseManagementInterface::new()))]),
firewall_mgmt: Box::new(OPNSenseManagementInterface::new()),
storage_host: vec![],
worker_host: vec![],
control_plane_host: vec![],

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

@ -63,8 +63,7 @@ async fn main() {
"wk".to_string(),
),
switch: SwitchGroup::from([]),
firewall: FirewallGroup::from([PhysicalHost::empty(HostCategory::Firewall)
.management(Arc::new(OPNSenseManagementInterface::new()))]),
firewall_mgmt: Box::new(OPNSenseManagementInterface::new()),
storage_host: vec![],
worker_host: vec![],
control_plane_host: vec![

8
harmony/src/domain/config.rs
View File

@ -12,4 +12,12 @@ lazy_static! {
std::env::var("HARMONY_REGISTRY_PROJECT").unwrap_or_else(|_| "harmony".to_string());
pub static ref DRY_RUN: bool =
std::env::var("HARMONY_DRY_RUN").is_ok_and(|value| value.parse().unwrap_or(false));
pub static ref DEFAULT_DATABASE_URL: String = "sqlite://harmony.sqlite".to_string();
pub static ref DATABASE_URL: String = std::env::var("HARMONY_DATABASE_URL")
.map(|value| if value.is_empty() {
(*DEFAULT_DATABASE_URL).clone()
} else {
value
})
.unwrap_or((*DEFAULT_DATABASE_URL).clone());
}

242
harmony/src/domain/hardware/mod.rs
View File

@ -1,25 +1,24 @@
use std::{str::FromStr, sync::Arc};
use std::sync::Arc;
use derive_new::new;
use harmony_inventory_agent::hwinfo::{CPU, MemoryModule, NetworkInterface, StorageDrive};
use harmony_types::net::MacAddress;
use serde::{Deserialize, Serialize, Serializer, ser::SerializeStruct};
use serde_value::Value;
use harmony_inventory_agent::hwinfo::NetworkInterface;
pub type HostGroup = Vec<PhysicalHost>;
pub type SwitchGroup = Vec<Switch>;
pub type FirewallGroup = Vec<PhysicalHost>;
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Serialize)]
pub struct PhysicalHost {
pub id: Id,
pub category: HostCategory,
pub network: Vec<NetworkInterface>,
pub management: Arc<dyn ManagementInterface>,
pub storage: Vec<Storage>,
pub storage: Vec<StorageDrive>,
pub labels: Vec<Label>,
pub memory_size: Option<u64>,
pub cpu_count: Option<u64>,
pub memory_modules: Vec<MemoryModule>,
pub cpus: Vec<CPU>,
}
impl PhysicalHost {
@ -30,12 +29,15 @@ impl PhysicalHost {
network: vec![],
storage: vec![],
labels: vec![],
management: Arc::new(ManualManagementInterface {}),
memory_size: None,
cpu_count: None,
memory_modules: vec![],
cpus: vec![],
}
}
pub fn summary(&self) -> String {
todo!();
}
pub fn cluster_mac(&self) -> MacAddress {
self.network
.first()
@ -43,32 +45,6 @@ impl PhysicalHost {
.mac_address
}
pub fn cpu(mut self, cpu_count: Option<u64>) -> Self {
self.cpu_count = cpu_count;
self
}
pub fn memory_size(mut self, memory_size: Option<u64>) -> Self {
self.memory_size = memory_size;
self
}
pub fn storage(
mut self,
connection: StorageConnectionType,
kind: StorageKind,
size: u64,
serial: String,
) -> Self {
self.storage.push(Storage {
connection,
kind,
size,
serial,
});
self
}
pub fn mac_address(mut self, mac_address: MacAddress) -> Self {
self.network.push(NetworkInterface {
name: String::new(),
@ -88,54 +64,49 @@ impl PhysicalHost {
self.labels.push(Label { name, value });
self
}
pub fn management(mut self, management: Arc<dyn ManagementInterface>) -> Self {
self.management = management;
self
}
}
// Custom Serialize implementation for PhysicalHost
impl Serialize for PhysicalHost {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
// Determine the number of fields
let mut num_fields = 5; // category, network, storage, labels, management
if self.memory_size.is_some() {
num_fields += 1;
}
if self.cpu_count.is_some() {
num_fields += 1;
}
// Create a serialization structure
let mut state = serializer.serialize_struct("PhysicalHost", num_fields)?;
// Serialize the standard fields
state.serialize_field("category", &self.category)?;
state.serialize_field("network", &self.network)?;
state.serialize_field("storage", &self.storage)?;
state.serialize_field("labels", &self.labels)?;
// Serialize optional fields
if let Some(memory) = self.memory_size {
state.serialize_field("memory_size", &memory)?;
}
if let Some(cpu) = self.cpu_count {
state.serialize_field("cpu_count", &cpu)?;
}
let mgmt_data = self.management.serialize_management();
// pub management: Arc<dyn ManagementInterface>,
// Handle management interface - either as a field or flattened
state.serialize_field("management", &mgmt_data)?;
state.end()
}
}
// impl Serialize for PhysicalHost {
// fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
// where
// S: Serializer,
// {
// // Determine the number of fields
// let mut num_fields = 5; // category, network, storage, labels, management
// if self.memory_modules.is_some() {
// num_fields += 1;
// }
// if self.cpus.is_some() {
// num_fields += 1;
// }
//
// // Create a serialization structure
// let mut state = serializer.serialize_struct("PhysicalHost", num_fields)?;
//
// // Serialize the standard fields
// state.serialize_field("category", &self.category)?;
// state.serialize_field("network", &self.network)?;
// state.serialize_field("storage", &self.storage)?;
// state.serialize_field("labels", &self.labels)?;
//
// // Serialize optional fields
// if let Some(memory) = self.memory_modules {
// state.serialize_field("memory_size", &memory)?;
// }
// if let Some(cpu) = self.cpus {
// state.serialize_field("cpu_count", &cpu)?;
// }
//
// let mgmt_data = self.management.serialize_management();
// // pub management: Arc<dyn ManagementInterface>,
//
// // Handle management interface - either as a field or flattened
// state.serialize_field("management", &mgmt_data)?;
//
// state.end()
// }
// }
impl<'de> Deserialize<'de> for PhysicalHost {
fn deserialize<D>(_deserializer: D) -> Result<Self, D::Error>
@ -200,39 +171,6 @@ pub enum HostCategory {
use harmony_macros::mac_address;
use harmony_types::id::Id;
#[derive(Debug, new, Clone, Serialize)]
pub enum StorageConnectionType {
Sata3g,
Sata6g,
Sas6g,
Sas12g,
PCIE,
}
#[derive(Debug, Clone, Serialize)]
pub enum StorageKind {
SSD,
NVME,
HDD,
}
#[derive(Debug, new, Clone, Serialize)]
pub struct Storage {
pub connection: StorageConnectionType,
pub kind: StorageKind,
pub size: u64,
pub serial: String,
}
#[cfg(test)]
impl Storage {
pub fn dummy() -> Self {
Self {
connection: StorageConnectionType::Sata3g,
kind: StorageKind::SSD,
size: 0,
serial: String::new(),
}
}
}
#[derive(Debug, Clone, Serialize)]
pub struct Switch {
@ -263,6 +201,40 @@ impl Location {
}
}
impl std::fmt::Display for HostCategory {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
HostCategory::Server => write!(f, "Server"),
HostCategory::Firewall => write!(f, "Firewall"),
HostCategory::Switch => write!(f, "Switch"),
}
}
}
impl std::fmt::Display for Label {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}: {}", self.name, self.value)
}
}
impl std::fmt::Display for Location {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Address: {}, Name: {}", self.address, self.name)
}
}
impl std::fmt::Display for PhysicalHost {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.summary())
}
}
impl std::fmt::Display for Switch {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Switch with {} interfaces", self._interface.len())
}
}
#[cfg(test)]
mod tests {
use super::*;
@ -309,16 +281,10 @@ mod tests {
id: Id::empty(),
category: HostCategory::Server,
network: vec![NetworkInterface::dummy()],
management: Arc::new(MockHPIlo {
ip: "192.168.1.100".to_string(),
username: "admin".to_string(),
password: "password123".to_string(),
firmware_version: "2.5.0".to_string(),
}),
storage: vec![Storage::dummy()],
storage: vec![StorageDrive::dummy()],
labels: vec![Label::new("datacenter".to_string(), "us-east".to_string())],
memory_size: Some(64_000_000),
cpu_count: Some(16),
memory_modules: vec![],
cpus: vec![],
};
// Serialize to JSON
@ -347,15 +313,10 @@ mod tests {
id: Id::empty(),
category: HostCategory::Server,
network: vec![NetworkInterface::dummy()],
management: Arc::new(MockDellIdrac {
hostname: "idrac-server01".to_string(),
port: 443,
api_token: "abcdef123456".to_string(),
}),
storage: vec![Storage::dummy()],
storage: vec![StorageDrive::dummy()],
labels: vec![Label::new("env".to_string(), "production".to_string())],
memory_size: Some(128_000_000),
cpu_count: Some(32),
memory_modules: vec![],
cpus: vec![],
};
// Serialize to JSON
@ -382,31 +343,20 @@ mod tests {
id: Id::empty(),
category: HostCategory::Server,
network: vec![],
management: Arc::new(MockHPIlo {
ip: "10.0.0.1".to_string(),
username: "root".to_string(),
password: "secret".to_string(),
firmware_version: "3.0.0".to_string(),
}),
storage: vec![],
labels: vec![],
memory_size: None,
cpu_count: None,
memory_modules: vec![],
cpus: vec![],
};
let host2 = PhysicalHost {
id: Id::empty(),
category: HostCategory::Server,
network: vec![],
management: Arc::new(MockDellIdrac {
hostname: "server02-idrac".to_string(),
port: 8443,
api_token: "token123".to_string(),
}),
storage: vec![],
labels: vec![],
memory_size: None,
cpu_count: None,
memory_modules: vec![],
cpus: vec![],
};
// Both should serialize successfully

8
harmony/src/domain/inventory/mod.rs
View File

@ -18,6 +18,8 @@ impl InventoryFilter {
use derive_new::new;
use log::info;
use crate::hardware::{ManagementInterface, ManualManagementInterface};
use super::{
filter::Filter,
hardware::{FirewallGroup, HostGroup, Location, SwitchGroup},
@ -30,7 +32,7 @@ pub struct Inventory {
// Firewall is really just a host but with somewhat specialized hardware
// I'm not entirely sure it belongs to its own category but it helps make things easier and
// clearer for now so let's try it this way.
pub firewall: FirewallGroup,
pub firewall_mgmt: Box<dyn ManagementInterface>,
pub worker_host: HostGroup,
pub storage_host: HostGroup,
pub control_plane_host: HostGroup,
@ -41,7 +43,7 @@ impl Inventory {
Self {
location: Location::new("Empty".to_string(), "location".to_string()),
switch: vec![],
firewall: vec![],
firewall_mgmt: Box::new(ManualManagementInterface {}),
worker_host: vec![],
storage_host: vec![],
control_plane_host: vec![],
@ -52,7 +54,7 @@ impl Inventory {
Self {
location: Location::test_building(),
switch: SwitchGroup::new(),
firewall: FirewallGroup::new(),
firewall_mgmt: Box::new(ManualManagementInterface {}),
worker_host: HostGroup::new(),
storage_host: HostGroup::new(),
control_plane_host: HostGroup::new(),

18
harmony/src/infra/inventory/mod.rs
View File

@ -1 +1,17 @@
mod sqlite;
use crate::{
config::DATABASE_URL,
infra::inventory::sqlite::SqliteInventoryRepository,
inventory::{InventoryRepository, RepoError},
};
pub mod sqlite;
pub struct InventoryRepositoryFactory;
impl InventoryRepositoryFactory {
pub async fn build() -> Result<Box<dyn InventoryRepository>, RepoError> {
Ok(Box::new(
SqliteInventoryRepository::new(&(*DATABASE_URL)).await?,
))
}
}

94
harmony/src/modules/inventory/mod.rs
View File

@ -1,3 +1,6 @@
use std::sync::Arc;
use tokio::time::Duration;
use async_trait::async_trait;
use harmony_inventory_agent::local_presence::DiscoveryEvent;
use log::{debug, info, trace};
@ -5,7 +8,8 @@ use serde::{Deserialize, Serialize};
use crate::{
data::Version,
hardware::{HostCategory, PhysicalHost},
hardware::{HostCategory, Label, PhysicalHost},
infra::inventory::InventoryRepositoryFactory,
interpret::{Interpret, InterpretError, InterpretName, InterpretStatus, Outcome},
inventory::Inventory,
score::Score,
@ -48,10 +52,10 @@ impl<T: Topology> Interpret<T> for DiscoverInventoryAgentInterpret {
harmony_inventory_agent::local_presence::discover_agents(
self.score.discovery_timeout,
|event: DiscoveryEvent| -> Result<(), String> {
println!("Discovery event {event:?}");
debug!("Discovery event {event:?}");
match event {
DiscoveryEvent::ServiceResolved(service) => {
debug!("Found instance {service:?}");
info!("Found instance {service:?}");
let address = match service.get_addresses().iter().next() {
Some(address) => address,
None => {
@ -61,34 +65,82 @@ impl<T: Topology> Interpret<T> for DiscoverInventoryAgentInterpret {
}
};
let address = &address.to_string();
let address = address.to_string();
let port = service.get_port();
debug!("Getting host inventory on service at {address} port {port}");
info!("Getting host inventory on service at {address} port {port}");
let host =
harmony_inventory_agent::client::get_host_inventory(address, port)?;
tokio::task::spawn(async move {
todo!("are we here");
info!("Getting inventory for host {address} {port}");
let host =
harmony_inventory_agent::client::get_host_inventory(&address, port)
.await
.unwrap();
trace!("Found host information {host:?}");
// TODO its useless to have two distinct host types but requires a bit much
// refactoring to do it now
let host = PhysicalHost {
id: Id::from(host.host_uuid),
category: HostCategory::Server,
network: todo!(),
management: todo!(),
storage: todo!(),
labels: todo!(),
memory_size: todo!(),
cpu_count: todo!(),
};
trace!("Found host information {host:?}");
// TODO its useless to have two distinct host types but requires a bit much
// refactoring to do it now
let harmony_inventory_agent::hwinfo::PhysicalHost {
storage_drives,
storage_controller,
memory_modules,
cpus,
chipset,
network_interfaces,
management_interface,
host_uuid,
} = host;
let host = PhysicalHost {
id: Id::from(host_uuid),
category: HostCategory::Server,
network: network_interfaces,
storage: storage_drives,
labels: vec![Label {
name: "discovered-by".to_string(),
value: "harmony-inventory-agent".to_string(),
}],
memory_modules,
cpus,
};
let repo = InventoryRepositoryFactory::build()
.await
.map_err(|e| format!("Could not build repository : {e}"))
.unwrap();
repo.save(&host)
.await
.map_err(|e| format!("Could not save host : {e}"))
.unwrap();
info!(
"Saved new host id {}, summary : {}",
host.id,
host.summary()
);
});
}
_ => debug!("Unhandled event {event:?}"),
};
Ok(())
},
)
.await;
info!("Launched inventory host information gathering");
info!(
"tokio current {:?}",
tokio::runtime::Handle::try_current().unwrap()
);
todo!()
tokio::spawn(async {
info!("Spawned a sleeper");
tokio::time::sleep(Duration::from_millis(100)).await;
info!("done a sleeper");
});
tokio::time::sleep(Duration::from_millis(1000)).await;
Ok(Outcome {
status: InterpretStatus::RUNNING,
message: "Launched discovery process".to_string(),
})
}
fn get_name(&self) -> InterpretName {

7
harmony_inventory_agent/src/client.rs
View File

@ -1,14 +1,15 @@
use crate::hwinfo::PhysicalHost;
pub fn get_host_inventory(host: &str, port: u16) -> Result<PhysicalHost, String> {
pub async fn get_host_inventory(host: &str, port: u16) -> Result<PhysicalHost, String> {
let url = format!("http://{host}:{port}/inventory");
let client = reqwest::blocking::Client::new();
let client = reqwest::Client::new();
let response = client
.get(url)
.send()
.await
.map_err(|e| format!("Failed to download file: {e}"))?;
let host = response.json().map_err(|e| e.to_string())?;
let host = response.json().await.map_err(|e| e.to_string())?;
Ok(host)
}

24
harmony_inventory_agent/src/hwinfo.rs
View File

@ -19,7 +19,7 @@ pub struct PhysicalHost {
pub host_uuid: String,
}
#[derive(Serialize, Deserialize, Debug)]
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct StorageDrive {
pub name: String,
pub model: String,
@ -33,13 +33,30 @@ pub struct StorageDrive {
pub smart_status: Option<String>,
}
impl StorageDrive {
pub fn dummy() -> Self {
Self {
name: String::new(),
model: String::new(),
serial: String::new(),
size_bytes: 0,
logical_block_size: 0,
physical_block_size: 0,
rotational: false,
wwn: None,
interface_type: String::new(),
smart_status: None,
}
}
}
#[derive(Serialize, Deserialize, Debug)]
pub struct StorageController {
pub name: String,
pub driver: String,
}
#[derive(Serialize, Deserialize, Debug)]
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct MemoryModule {
pub size_bytes: u64,
pub speed_mhz: Option<u32>,
@ -49,7 +66,7 @@ pub struct MemoryModule {
pub rank: Option<u8>,
}
#[derive(Serialize, Deserialize, Debug)]
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct CPU {
pub model: String,
pub vendor: String,
@ -77,7 +94,6 @@ pub struct NetworkInterface {
pub firmware_version: Option<String>,
}
#[cfg(test)]
impl NetworkInterface {
pub fn dummy() -> Self {
use harmony_macros::mac_address;

2
harmony_inventory_agent/src/lib.rs
View File

@ -1,3 +1,3 @@
pub mod client;
pub mod hwinfo;
pub mod local_presence;
pub mod client;

13
harmony_inventory_agent/src/local_presence/discover.rs
View File

@ -5,10 +5,10 @@ use crate::local_presence::SERVICE_NAME;
pub type DiscoveryEvent = ServiceEvent;
pub fn discover_agents(
timeout: Option<u64>,
on_event: impl Fn(DiscoveryEvent) -> Result<(), String> + Send + 'static,
) {
pub async fn discover_agents<F>(timeout: Option<u64>, on_event: F)
where
F: FnOnce(DiscoveryEvent) -> Result<(), String> + Send + 'static + Copy,
{
// Create a new mDNS daemon.
let mdns = ServiceDaemon::new().expect("Failed to create mDNS daemon");
@ -16,7 +16,7 @@ pub fn discover_agents(
// The receiver will be a stream of events.
let receiver = mdns.browse(SERVICE_NAME).expect("Failed to browse");
std::thread::spawn(move || {
tokio::spawn(async move {
while let Ok(event) = receiver.recv() {
if let Err(e) = on_event(event.clone()) {
error!("Event callback failed : {e}");
@ -33,8 +33,7 @@ pub fn discover_agents(
});
if let Some(timeout) = timeout {
// Gracefully shutdown the daemon.
std::thread::sleep(std::time::Duration::from_secs(timeout));
tokio::time::sleep(std::time::Duration::from_secs(timeout)).await;
mdns.shutdown().unwrap();
}
}