use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::sync::Arc;
use anyhow::{Context, bail};
use names::{Generator, Name};
use tokio::sync::RwLock;
use tracing::{debug, trace, warn};
use crate::engine::Engine;
use crate::engine::workload::ResolvedWorkload;
use crate::plugin::HostPlugin;
use crate::types::*;
use crate::wit::WitWorld;
mod sysinfo;
use sysinfo::SystemMonitor;
pub trait HostApi {
fn heartbeat(&self) -> impl Future<Output = anyhow::Result<HostHeartbeat>>;
fn workload_start(
&self,
request: WorkloadStartRequest,
) -> impl Future<Output = anyhow::Result<WorkloadStartResponse>>;
fn workload_status(
&self,
request: WorkloadStatusRequest,
) -> impl Future<Output = anyhow::Result<WorkloadStatusResponse>>;
fn workload_stop(
&self,
request: WorkloadStopRequest,
) -> impl Future<Output = anyhow::Result<WorkloadStopResponse>>;
}
impl<T: HostApi> HostApi for Arc<T> {
async fn heartbeat(&self) -> anyhow::Result<HostHeartbeat> {
self.as_ref().heartbeat().await
}
async fn workload_start(
&self,
request: WorkloadStartRequest,
) -> anyhow::Result<WorkloadStartResponse> {
self.as_ref().workload_start(request).await
}
async fn workload_stop(
&self,
request: WorkloadStopRequest,
) -> anyhow::Result<WorkloadStopResponse> {
self.as_ref().workload_stop(request).await
}
async fn workload_status(
&self,
request: WorkloadStatusRequest,
) -> anyhow::Result<WorkloadStatusResponse> {
self.as_ref().workload_status(request).await
}
}
#[derive(Debug, Clone)]
pub enum HostWorkload {
Starting,
Running(Box<ResolvedWorkload>),
Stopping,
Error,
}
impl From<&HostWorkload> for WorkloadState {
fn from(hw: &HostWorkload) -> Self {
match hw {
HostWorkload::Starting => WorkloadState::Starting,
HostWorkload::Running(_) => WorkloadState::Running,
HostWorkload::Stopping => WorkloadState::Stopping,
HostWorkload::Error => WorkloadState::Error,
}
}
}
pub struct Host {
engine: Engine,
workloads: Arc<RwLock<HashMap<String, HostWorkload>>>,
plugins: HashMap<&'static str, Arc<dyn HostPlugin>>,
id: String,
hostname: String,
friendly_name: String,
version: String,
labels: HashMap<String, String>,
started_at: chrono::DateTime<chrono::Utc>,
system_monitor: Arc<RwLock<SystemMonitor>>,
}
impl Host {
pub fn builder() -> HostBuilder {
HostBuilder::default()
}
pub async fn start(self) -> anyhow::Result<Arc<Self>> {
for (id, plugin) in &self.plugins {
if let Err(e) = plugin.start().await {
tracing::error!(id = id, err = ?e, "failed to start plugin");
bail!(e)
}
}
Ok(Arc::new(self))
}
pub async fn stop(self: Arc<Self>) -> anyhow::Result<()> {
for (id, plugin) in &self.plugins {
let stop_fut = plugin.stop();
match tokio::time::timeout(std::time::Duration::from_secs(3), stop_fut).await {
Ok(Err(e)) => {
tracing::error!(id = id, err = ?e, "failed to stop plugin");
}
Err(_) => {
tracing::error!(id = id, "plugin stop timed out after 3 seconds");
}
_ => {}
}
}
Ok(())
}
pub fn label(&self, label: impl AsRef<str>) -> Option<&String> {
self.labels.get(label.as_ref())
}
pub fn id(&self) -> &str {
&self.id
}
pub fn friendly_name(&self) -> &str {
&self.friendly_name
}
fn generate_workload_id(&self) -> String {
uuid::Uuid::new_v4().to_string()
}
pub fn wit_world(&self) -> WitWorld {
let mut imports = HashSet::new();
let mut exports = HashSet::from([
"wasi:io/poll,error,streams@0.2.0".into(),
"wasi:clocks/monotonic-clock,wall-time@0.2.0".into(),
"wasi:random/random@0.2.0".into(),
"wasi:cli/environment,exit,stderr,stdin,stdout,terminal-input,terminal-output,terminal-stderr,terminal-stdin,terminal-stdout@0.2.0".into(),
"wasi:clocks/monotonic-clock,wall-clock@0.2.0".into(),
"wasi:filesystem/preopens,types@0.2.0".into(),
"wasi:random/insecure-seed,insecure,random@0.2.0".into(),
"wasi:sockets/instance-network,ip-name-lookup,network,tcp-create-socket,tcp,udp-create-socket,udp@0.2.0".into(),
]);
imports.extend(
self.plugins
.values()
.flat_map(|p| p.world().imports.into_iter().collect::<Vec<_>>()),
);
exports.extend(
self.plugins
.values()
.flat_map(|p| p.world().exports.into_iter().collect::<Vec<_>>()),
);
WitWorld { imports, exports }
}
async fn get_system_info(&self) -> (String, String, String) {
let os_name = std::env::consts::OS.to_string();
let os_arch = std::env::consts::ARCH.to_string();
let os_kernel = std::env::consts::FAMILY.to_string();
(os_arch, os_name, os_kernel)
}
async fn get_memory_info(&self) -> anyhow::Result<(u64, u64)> {
let monitor = self.system_monitor.read().await;
let mem = monitor.memory_usage();
Ok((mem.total_memory, mem.free_memory))
}
async fn get_cpu_usage(&self) -> anyhow::Result<f32> {
let monitor = self.system_monitor.read().await;
Ok(monitor.cpu_usage().global_usage)
}
}
impl HostApi for Host {
async fn heartbeat(&self) -> anyhow::Result<HostHeartbeat> {
{
let mut monitor = self.system_monitor.write().await;
monitor.refresh();
monitor.report_usage();
}
let (os_arch, os_name, os_kernel) = self.get_system_info().await;
let (system_memory_total, system_memory_free) = self
.get_memory_info()
.await
.context("failed to get memory info")?;
let system_cpu_usage = self
.get_cpu_usage()
.await
.context("failed to get CPU usage")?;
let (workload_count, component_count) = {
let workloads = self.workloads.read().await;
let workload_count: u64 = workloads.len() as u64;
let mut component_count: u64 = 0;
for workload in workloads.values() {
if let HostWorkload::Running(workload) = workload {
component_count += workload.component_count().await as u64;
}
}
(workload_count, component_count)
};
let mut imports = Vec::new();
let mut exports = Vec::new();
for plugin in self.plugins.values() {
let world = plugin.world();
imports.extend(world.imports.into_iter());
exports.extend(world.exports.into_iter());
}
Ok(HostHeartbeat {
id: self.id.clone(),
hostname: self.hostname.clone(),
friendly_name: self.friendly_name.clone(),
version: self.version.clone(),
labels: self.labels.clone(),
started_at: self.started_at,
os_arch,
os_name,
os_kernel,
system_cpu_usage,
system_memory_total,
system_memory_free,
component_count,
workload_count,
imports,
exports,
})
}
async fn workload_start(
&self,
request: WorkloadStartRequest,
) -> anyhow::Result<WorkloadStartResponse> {
let workload_id = self.generate_workload_id();
self.workloads
.write()
.await
.insert(workload_id.clone(), HostWorkload::Starting);
let service_present = request.workload.service.is_some();
let unresolved_workload = self
.engine
.initialize_workload(&workload_id, request.workload)?;
let mut resolved_workload = unresolved_workload.resolve(Some(&self.plugins)).await?;
if resolved_workload.execute_service().await? != service_present {
warn!(
workload_id = workload_id,
"service did not properly execute"
);
}
self.workloads
.write()
.await
.entry(workload_id.clone())
.and_modify(|workload| {
*workload = HostWorkload::Running(Box::new(resolved_workload));
});
Ok(WorkloadStartResponse {
workload_status: WorkloadStatus {
workload_id,
workload_state: WorkloadState::Running,
message: "Workload started successfully".to_string(),
},
})
}
async fn workload_status(
&self,
request: WorkloadStatusRequest,
) -> anyhow::Result<WorkloadStatusResponse> {
if let Some(workload) = self.workloads.read().await.get(&request.workload_id) {
let workload_state = workload.into();
Ok(WorkloadStatusResponse {
workload_status: WorkloadStatus {
workload_id: request.workload_id,
message: format!("Workload is {workload_state:?}"),
workload_state,
},
})
} else {
anyhow::bail!("Workload not found: {}", request.workload_id)
}
}
async fn workload_stop(
&self,
request: WorkloadStopRequest,
) -> anyhow::Result<WorkloadStopResponse> {
let has_workload = self
.workloads
.read()
.await
.contains_key(&request.workload_id);
let (workload_state, message) = if has_workload {
let resolved_workload = {
let mut workloads = self.workloads.write().await;
trace!(
workload_id = request.workload_id,
"updating workload state to stopping"
);
workloads
.insert(request.workload_id.clone(), HostWorkload::Stopping)
.and_then(|hw| match hw {
HostWorkload::Running(rw) => Some(*rw),
_ => None,
})
};
if let Some(resolved_workload) = resolved_workload {
debug!(
workload_id = request.workload_id,
workload_name = resolved_workload.name(),
"stopping workload"
);
resolved_workload.stop_service();
if let Err(e) = resolved_workload.unbind_all_plugins().await {
warn!(
workload_id = request.workload_id,
error = ?e,
"error unbinding plugins during workload stop, continuing"
);
}
}
self.workloads.write().await.remove(&request.workload_id);
debug!(
workload_id = request.workload_id,
"workload stopped successfully"
);
(
WorkloadState::Stopping,
"Workload stopped successfully".to_string(),
)
} else {
(WorkloadState::Unspecified, "Workload not found".to_string())
};
Ok(WorkloadStopResponse {
workload_status: WorkloadStatus {
workload_id: request.workload_id,
workload_state,
message,
},
})
}
}
impl std::fmt::Debug for Host {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Host")
.field("id", &self.id)
.field("hostname", &self.hostname)
.field("friendly_name", &self.friendly_name)
.field("version", &self.version)
.field("labels", &self.labels)
.field("started_at", &self.started_at)
.field("workloads", &self.workloads)
.finish()
}
}
#[derive(Default)]
pub struct HostBuilder {
engine: Option<Engine>,
plugins: HashMap<&'static str, Arc<dyn HostPlugin>>,
hostname: Option<String>,
friendly_name: Option<String>,
labels: HashMap<String, String>,
}
impl HostBuilder {
pub fn new() -> Self {
Self::default()
}
pub fn with_engine(mut self, engine: Engine) -> Self {
self.engine = Some(engine);
self
}
pub fn with_plugin<T: HostPlugin>(mut self, plugin: Arc<T>) -> anyhow::Result<Self> {
let plugin_id = plugin.id();
if self.plugins.contains_key(plugin_id) {
bail!("Duplicate plugin ID '{plugin_id}' - plugin IDs must be unique");
}
self.plugins.insert(plugin_id, plugin);
Ok(self)
}
pub fn with_hostname(mut self, hostname: impl AsRef<str>) -> Self {
self.hostname = Some(hostname.as_ref().to_string());
self
}
pub fn with_friendly_name(mut self, name: impl AsRef<str>) -> Self {
self.friendly_name = Some(name.as_ref().to_string());
self
}
pub fn with_label(mut self, key: impl AsRef<str>, value: impl AsRef<str>) -> Self {
self.labels
.insert(key.as_ref().to_string(), value.as_ref().to_string());
self
}
pub fn build(self) -> anyhow::Result<Host> {
let engine = if let Some(engine) = self.engine {
engine
} else {
Engine::builder().build()?
};
let hostname = self.hostname.unwrap_or_else(|| {
hostname::get()
.map(|h| h.to_string_lossy().to_string())
.unwrap_or_else(|_| "unknown".to_string())
});
let friendly_name = self.friendly_name.unwrap_or_else(|| {
let mut generator = Generator::with_naming(Name::Numbered);
generator
.next()
.unwrap_or_else(|| format!("host-{}", uuid::Uuid::new_v4()))
});
Ok(Host {
engine,
workloads: Arc::default(),
plugins: self.plugins,
id: uuid::Uuid::new_v4().to_string(),
hostname,
friendly_name,
version: env!("CARGO_PKG_VERSION").to_string(),
labels: self.labels,
started_at: chrono::Utc::now(),
system_monitor: Arc::new(RwLock::new(SystemMonitor::new())),
})
}
}