Start9/start-os
2
0
Fork 0
mirror of https://github.com/Start9Labs/start-os.git synced 2026-03-30 12:11:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'next/minor' of github.com:Start9Labs/start-os into next/major

Browse Source
This commit is contained in:
Matt Hill
2024-08-28 15:49:28 -06:00
parent 097d77f7b3 ed1bc6c215
commit b6c48d0f98
130 changed files with 4839 additions and 2114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
core/startos/src/context/rpc.rs
View File

@@ -1,4 +1,5 @@
use std::collections::BTreeMap;
use std::future::Future;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::ops::Deref;
use std::path::PathBuf;
@@ -6,6 +7,8 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use chrono::{TimeDelta, Utc};
use helpers::NonDetachingJoinHandle;
use imbl_value::InternedString;
use josekit::jwk::Jwk;
use reqwest::{Client, Proxy};
@@ -29,7 +32,7 @@ use crate::net::utils::{find_eth_iface, find_wifi_iface};
use crate::net::wifi::WpaCli;
use crate::prelude::*;
use crate::progress::{FullProgressTracker, PhaseProgressTrackerHandle};
use crate::rpc_continuations::{OpenAuthedContinuations, RpcContinuations};
use crate::rpc_continuations::{Guid, OpenAuthedContinuations, RpcContinuations};
use crate::service::effects::callbacks::ServiceCallbacks;
use crate::service::ServiceMap;
use crate::shutdown::Shutdown;
@@ -63,6 +66,7 @@ pub struct RpcContextSeed {
pub client: Client,
pub hardware: Hardware,
pub start_time: Instant,
pub crons: SyncMutex<BTreeMap<Guid, NonDetachingJoinHandle<()>>>,
#[cfg(feature = "dev")]
pub dev: Dev,
}
@@ -94,12 +98,14 @@ impl InitRpcContextPhases {
}
pub struct CleanupInitPhases {
cleanup_sessions: PhaseProgressTrackerHandle,
init_services: PhaseProgressTrackerHandle,
check_dependencies: PhaseProgressTrackerHandle,
}
impl CleanupInitPhases {
pub fn new(handle: &FullProgressTracker) -> Self {
Self {
cleanup_sessions: handle.add_phase("Cleaning up sessions".into(), Some(1)),
init_services: handle.add_phase("Initializing services".into(), Some(10)),
check_dependencies: handle.add_phase("Checking dependencies".into(), Some(1)),
}
@@ -174,6 +180,8 @@ impl RpcContext {
let ram = get_mem_info().await?.total.0 as u64 * 1024 * 1024;
read_device_info.complete();
let crons = SyncMutex::new(BTreeMap::new());
if !db
.peek()
.await
@@ -183,18 +191,24 @@ impl RpcContext {
.de()?
{
let db = db.clone();
tokio::spawn(async move {
while !check_time_is_synchronized().await.unwrap() {
tokio::time::sleep(Duration::from_secs(30)).await;
}
db.mutate(|v| {
v.as_public_mut()
.as_server_info_mut()
.as_ntp_synced_mut()
.ser(&true)
})
.await
.unwrap()
crons.mutate(|c| {
c.insert(
Guid::new(),
tokio::spawn(async move {
while !check_time_is_synchronized().await.unwrap() {
tokio::time::sleep(Duration::from_secs(30)).await;
}
db.mutate(|v| {
v.as_public_mut()
.as_server_info_mut()
.as_ntp_synced_mut()
.ser(&true)
})
.await
.unwrap()
})
.into(),
)
});
}
@@ -259,6 +273,7 @@ impl RpcContext {
.with_kind(crate::ErrorKind::ParseUrl)?,
hardware: Hardware { devices, ram },
start_time: Instant::now(),
crons,
#[cfg(feature = "dev")]
dev: Dev {
lxc: Mutex::new(BTreeMap::new()),
@@ -273,6 +288,7 @@ impl RpcContext {
#[instrument(skip_all)]
pub async fn shutdown(self) -> Result<(), Error> {
self.crons.mutate(|c| std::mem::take(c));
self.services.shutdown_all().await?;
self.is_closed.store(true, Ordering::SeqCst);
tracing::info!("RPC Context is shutdown");
@@ -280,14 +296,75 @@ impl RpcContext {
Ok(())
}
pub fn add_cron<F: Future<Output = ()> + Send + 'static>(&self, fut: F) -> Guid {
let guid = Guid::new();
self.crons
.mutate(|c| c.insert(guid.clone(), tokio::spawn(fut).into()));
guid
}
#[instrument(skip_all)]
pub async fn cleanup_and_initialize(
&self,
CleanupInitPhases {
mut cleanup_sessions,
init_services,
mut check_dependencies,
}: CleanupInitPhases,
) -> Result<(), Error> {
cleanup_sessions.start();
self.db
.mutate(|db| {
if db.as_public().as_server_info().as_ntp_synced().de()? {
for id in db.as_private().as_sessions().keys()? {
if Utc::now()
- db.as_private()
.as_sessions()
.as_idx(&id)
.unwrap()
.de()?
.last_active
> TimeDelta::days(30)
{
db.as_private_mut().as_sessions_mut().remove(&id)?;
}
}
}
Ok(())
})
.await?;
let db = self.db.clone();
self.add_cron(async move {
loop {
tokio::time::sleep(Duration::from_secs(86400)).await;
if let Err(e) = db
.mutate(|db| {
if db.as_public().as_server_info().as_ntp_synced().de()? {
for id in db.as_private().as_sessions().keys()? {
if Utc::now()
- db.as_private()
.as_sessions()
.as_idx(&id)
.unwrap()
.de()?
.last_active
> TimeDelta::days(30)
{
db.as_private_mut().as_sessions_mut().remove(&id)?;
}
}
}
Ok(())
})
.await
{
tracing::error!("Error in session cleanup cron: {e}");
tracing::debug!("{e:?}");
}
}
});
cleanup_sessions.complete();
self.services.init(&self, init_services).await?;
tracing::info!("Initialized Package Managers");

5
core/startos/src/db/model/package.rs
View File

@@ -3,7 +3,9 @@ use std::collections::{BTreeMap, BTreeSet};
use chrono::{DateTime, Utc};
use exver::VersionRange;
use imbl_value::InternedString;
use models::{ActionId, DataUrl, HealthCheckId, HostId, PackageId, ServiceInterfaceId};
use models::{
ActionId, DataUrl, HealthCheckId, HostId, PackageId, ServiceInterfaceId, VersionString,
};
use patch_db::json_ptr::JsonPointer;
use patch_db::HasModel;
use reqwest::Url;
@@ -335,6 +337,7 @@ pub struct ActionMetadata {
#[ts(export)]
pub struct PackageDataEntry {
pub state_info: PackageState,
pub data_version: Option<VersionString>,
pub status: Status,
#[ts(type = "string | null")]
pub registry: Option<Url>,

7
core/startos/src/lxc/mod.rs
View File

@@ -268,9 +268,10 @@ impl LxcContainer {
.invoke(ErrorKind::Docker)
.await?,
)?;
let out_str = output.trim();
if !out_str.is_empty() {
return Ok(out_str.parse()?);
for line in output.lines() {
if let Ok(ip) = line.trim().parse() {
return Ok(ip);
}
}
if start.elapsed() > CONTAINER_DHCP_TIMEOUT {
return Err(Error::new(

1
core/startos/src/net/service_interface.rs
View File

@@ -67,7 +67,6 @@ pub struct ServiceInterface {
pub name: String,
pub description: String,
pub has_primary: bool,
pub disabled: bool,
pub masked: bool,
pub address_info: AddressInfo,
#[serde(rename = "type")]

2
core/startos/src/s9pk/v1/manifest.rs
View File

@@ -21,7 +21,7 @@ pub struct Manifest {
#[serde(default)]
pub git_hash: Option<GitHash>,
pub title: String,
pub version: exver::emver::Version,
pub version: String,
pub description: Description,
#[serde(default)]
pub assets: Assets,

26
core/startos/src/s9pk/v2/compat.rs
View File

@@ -1,8 +1,9 @@
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::str::FromStr;
use std::sync::Arc;
use exver::ExtendedVersion;
use exver::{ExtendedVersion, VersionRange};
use models::ImageId;
use tokio::io::{AsyncRead, AsyncSeek, AsyncWriteExt};
use tokio::process::Command;
@@ -44,9 +45,9 @@ impl S9pk<TmpSource<PackSource>> {
// manifest.json
let manifest_raw = reader.manifest().await?;
let manifest = from_value::<ManifestV1>(manifest_raw.clone())?;
let mut new_manifest = Manifest::from(manifest.clone());
let mut new_manifest = Manifest::try_from(manifest.clone())?;
let images: BTreeMap<ImageId, bool> = manifest
let images: BTreeSet<(ImageId, bool)> = manifest
.package_procedures()
.filter_map(|p| {
if let PackageProcedure::Docker(p) = p {
@@ -89,8 +90,6 @@ impl S9pk<TmpSource<PackSource>> {
// images
for arch in reader.docker_arches().await? {
let images_dir = tmp_dir.join("images").join(&arch);
tokio::fs::create_dir_all(&images_dir).await?;
Command::new(CONTAINER_TOOL)
.arg("load")
.input(Some(&mut reader.docker_images(&arch).await?))
@@ -194,15 +193,22 @@ impl S9pk<TmpSource<PackSource>> {
}
}
impl From<ManifestV1> for Manifest {
fn from(value: ManifestV1) -> Self {
impl TryFrom<ManifestV1> for Manifest {
type Error = Error;
fn try_from(value: ManifestV1) -> Result<Self, Self::Error> {
let default_url = value.upstream_repo.clone();
Self {
Ok(Self {
id: value.id,
title: value.title.into(),
version: ExtendedVersion::from(value.version).into(),
version: ExtendedVersion::from(
exver::emver::Version::from_str(&value.version)
.with_kind(ErrorKind::Deserialization)?,
)
.into(),
satisfies: BTreeSet::new(),
release_notes: value.release_notes,
can_migrate_from: VersionRange::any(),
can_migrate_to: VersionRange::none(),
license: value.license.into(),
wrapper_repo: value.wrapper_repo,
upstream_repo: value.upstream_repo,
@@ -244,6 +250,6 @@ impl From<ManifestV1> for Manifest {
git_hash: value.git_hash,
os_version: value.eos_version,
has_config: value.config.is_some(),
}
})
}
}

10
core/startos/src/s9pk/v2/manifest.rs
View File

@@ -2,7 +2,7 @@ use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use color_eyre::eyre::eyre;
use exver::Version;
use exver::{Version, VersionRange};
use helpers::const_true;
use imbl_value::InternedString;
pub use models::PackageId;
@@ -37,6 +37,10 @@ pub struct Manifest {
pub satisfies: BTreeSet<VersionString>,
pub release_notes: String,
#[ts(type = "string")]
pub can_migrate_to: VersionRange,
#[ts(type = "string")]
pub can_migrate_from: VersionRange,
#[ts(type = "string")]
pub license: InternedString, // type of license
#[ts(type = "string")]
pub wrapper_repo: Url,
@@ -159,8 +163,8 @@ impl Manifest {
#[ts(export)]
pub struct HardwareRequirements {
#[serde(default)]
#[ts(type = "{ device?: string, processor?: string }")]
pub device: BTreeMap<String, Regex>,
#[ts(type = "{ display?: string, processor?: string }")]
pub device: BTreeMap<String, Regex>, // TODO: array
#[ts(type = "number | null")]
pub ram: Option<u64>,
#[ts(type = "string[] | null")]

20
core/startos/src/s9pk/v2/pack.rs
View File

@@ -60,14 +60,20 @@ impl SqfsDir {
.get_or_try_init(|| async move {
let guid = Guid::new();
let path = self.tmpdir.join(guid.as_ref()).with_extension("squashfs");
let mut cmd = Command::new("mksquashfs");
if self.path.extension().and_then(|s| s.to_str()) == Some("tar") {
cmd.arg("-tar");
Command::new("tar2sqfs")
.arg(&path)
.input(Some(&mut open_file(&self.path).await?))
.invoke(ErrorKind::Filesystem)
.await?;
} else {
Command::new("mksquashfs")
.arg(&self.path)
.arg(&path)
.invoke(ErrorKind::Filesystem)
.await?;
}
cmd.arg(&self.path)
.arg(&path)
.invoke(ErrorKind::Filesystem)
.await?;
Ok(MultiCursorFile::from(
open_file(&path)
.await
@@ -507,7 +513,7 @@ impl ImageSource {
Command::new(CONTAINER_TOOL)
.arg("export")
.arg(container.trim())
.pipe(Command::new("mksquashfs").arg("-").arg(&dest).arg("-tar"))
.pipe(Command::new("tar2sqfs").arg(&dest))
.capture(false)
.invoke(ErrorKind::Docker)
.await?;

59
core/startos/src/service/effects/dependency.rs
View File

@@ -4,8 +4,10 @@ use std::str::FromStr;
use clap::builder::ValueParserFactory;
use exver::VersionRange;
use imbl::OrdMap;
use imbl_value::InternedString;
use itertools::Itertools;
use models::{HealthCheckId, PackageId, VolumeId};
use models::{HealthCheckId, PackageId, VersionString, VolumeId};
use patch_db::json_ptr::JsonPointer;
use tokio::process::Command;
@@ -17,7 +19,7 @@ use crate::disk::mount::filesystem::idmapped::IdMapped;
use crate::disk::mount::filesystem::{FileSystem, MountType};
use crate::rpc_continuations::Guid;
use crate::service::effects::prelude::*;
use crate::status::health_check::HealthCheckResult;
use crate::status::health_check::NamedHealthCheckResult;
use crate::util::clap::FromStrParser;
use crate::util::Invoke;
use crate::volume::data_dir;
@@ -316,12 +318,16 @@ pub struct CheckDependenciesParam {
#[ts(export)]
pub struct CheckDependenciesResult {
package_id: PackageId,
is_installed: bool,
#[ts(type = "string | null")]
title: Option<InternedString>,
#[ts(type = "string | null")]
installed_version: Option<exver::ExtendedVersion>,
#[ts(type = "string[]")]
satisfies: BTreeSet<VersionString>,
is_running: bool,
config_satisfied: bool,
health_checks: BTreeMap<HealthCheckId, HealthCheckResult>,
#[ts(type = "string | null")]
version: Option<exver::ExtendedVersion>,
#[ts(as = "BTreeMap::<HealthCheckId, NamedHealthCheckResult>")]
health_checks: OrdMap<HealthCheckId, NamedHealthCheckResult>,
}
pub async fn check_dependencies(
context: EffectContext,
@@ -347,36 +353,23 @@ pub async fn check_dependencies(
let mut results = Vec::with_capacity(package_ids.len());
for (package_id, dependency_info) in package_ids {
let title = dependency_info.title.clone();
let Some(package) = db.as_public().as_package_data().as_idx(&package_id) else {
results.push(CheckDependenciesResult {
package_id,
is_installed: false,
title,
installed_version: None,
satisfies: BTreeSet::new(),
is_running: false,
config_satisfied: false,
health_checks: Default::default(),
version: None,
});
continue;
};
let manifest = package.as_state_info().as_manifest(ManifestPreference::New);
let installed_version = manifest.as_version().de()?.into_version();
let satisfies = manifest.as_satisfies().de()?;
let version = Some(installed_version.clone());
if ![installed_version]
.into_iter()
.chain(satisfies.into_iter().map(|v| v.into_version()))
.any(|v| v.satisfies(&dependency_info.version_range))
{
results.push(CheckDependenciesResult {
package_id,
is_installed: false,
is_running: false,
config_satisfied: false,
health_checks: Default::default(),
version,
});
continue;
}
let installed_version = Some(installed_version.clone());
let is_installed = true;
let status = package.as_status().as_main().de()?;
let is_running = if is_installed {
@@ -384,25 +377,15 @@ pub async fn check_dependencies(
} else {
false
};
let health_checks =
if let CurrentDependencyKind::Running { health_checks } = &dependency_info.kind {
status
.health()
.cloned()
.unwrap_or_default()
.into_iter()
.filter(|(id, _)| health_checks.contains(id))
.collect()
} else {
Default::default()
};
let health_checks = status.health().cloned().unwrap_or_default();
results.push(CheckDependenciesResult {
package_id,
is_installed,
title,
installed_version,
satisfies,
is_running,
config_satisfied: dependency_info.config_satisfied,
health_checks,
version,
});
}
Ok(results)

8
core/startos/src/service/effects/health.rs
View File

@@ -1,7 +1,7 @@
use models::HealthCheckId;
use crate::service::effects::prelude::*;
use crate::status::health_check::HealthCheckResult;
use crate::status::health_check::NamedHealthCheckResult;
use crate::status::MainStatus;
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
@@ -10,7 +10,7 @@ use crate::status::MainStatus;
pub struct SetHealth {
id: HealthCheckId,
#[serde(flatten)]
result: HealthCheckResult,
result: NamedHealthCheckResult,
}
pub async fn set_health(
context: EffectContext,
@@ -32,8 +32,8 @@ pub async fn set_health(
.as_main_mut()
.mutate(|main| {
match main {
&mut MainStatus::Running { ref mut health, .. }
| &mut MainStatus::BackingUp { ref mut health, .. } => {
MainStatus::Running { ref mut health, .. }
| MainStatus::Starting { ref mut health } => {
health.insert(id, result);
}
_ => (),

124
core/startos/src/service/effects/mod.rs
View File

@@ -1,4 +1,4 @@
use rpc_toolkit::{from_fn, from_fn_async, Context, HandlerExt, ParentHandler};
use rpc_toolkit::{from_fn, from_fn_async, from_fn_blocking, Context, HandlerExt, ParentHandler};
use crate::echo;
use crate::prelude::*;
@@ -12,44 +12,44 @@ pub mod context;
mod control;
mod dependency;
mod health;
mod image;
mod net;
mod prelude;
mod store;
mod subcontainer;
mod system;
pub fn handler<C: Context>() -> ParentHandler<C> {
ParentHandler::new()
.subcommand("gitInfo", from_fn(|_: C| crate::version::git_info()))
.subcommand("git-info", from_fn(|_: C| crate::version::git_info()))
.subcommand(
"echo",
from_fn(echo::<EffectContext>).with_call_remote::<ContainerCliContext>(),
)
// action
.subcommand(
"executeAction",
"execute-action",
from_fn_async(action::execute_action).no_cli(),
)
.subcommand(
"exportAction",
"export-action",
from_fn_async(action::export_action).no_cli(),
)
.subcommand(
"clearActions",
"clear-actions",
from_fn_async(action::clear_actions).no_cli(),
)
// callbacks
.subcommand(
"clearCallbacks",
"clear-callbacks",
from_fn(callbacks::clear_callbacks).no_cli(),
)
// config
.subcommand(
"getConfigured",
"get-configured",
from_fn_async(config::get_configured).no_cli(),
)
.subcommand(
"setConfigured",
"set-configured",
from_fn_async(config::set_configured)
.no_display()
.with_call_remote::<ContainerCliContext>(),
@@ -68,105 +68,143 @@ pub fn handler<C: Context>() -> ParentHandler<C> {
.with_call_remote::<ContainerCliContext>(),
)
.subcommand(
"setMainStatus",
"set-main-status",
from_fn_async(control::set_main_status)
.no_display()
.with_call_remote::<ContainerCliContext>(),
)
// dependency
.subcommand(
"setDependencies",
"set-dependencies",
from_fn_async(dependency::set_dependencies)
.no_display()
.with_call_remote::<ContainerCliContext>(),
)
.subcommand(
"getDependencies",
"get-dependencies",
from_fn_async(dependency::get_dependencies)
.no_display()
.with_call_remote::<ContainerCliContext>(),
)
.subcommand(
"checkDependencies",
"check-dependencies",
from_fn_async(dependency::check_dependencies)
.no_display()
.with_call_remote::<ContainerCliContext>(),
)
.subcommand("mount", from_fn_async(dependency::mount).no_cli())
.subcommand(
"getInstalledPackages",
"get-installed-packages",
from_fn_async(dependency::get_installed_packages).no_cli(),
)
.subcommand(
"exposeForDependents",
"expose-for-dependents",
from_fn_async(dependency::expose_for_dependents).no_cli(),
)
// health
.subcommand("setHealth", from_fn_async(health::set_health).no_cli())
// image
.subcommand("set-health", from_fn_async(health::set_health).no_cli())
// subcontainer
.subcommand(
"chroot",
from_fn(image::chroot::<ContainerCliContext>).no_display(),
)
.subcommand(
"createOverlayedImage",
from_fn_async(image::create_overlayed_image)
.with_custom_display_fn(|_, (path, _)| Ok(println!("{}", path.display())))
.with_call_remote::<ContainerCliContext>(),
)
.subcommand(
"destroyOverlayedImage",
from_fn_async(image::destroy_overlayed_image).no_cli(),
"subcontainer",
ParentHandler::<C>::new()
.subcommand(
"launch",
from_fn_blocking(subcontainer::launch).no_display(),
)
.subcommand(
"launch-init",
from_fn_blocking(subcontainer::launch_init).no_display(),
)
.subcommand("exec", from_fn_blocking(subcontainer::exec).no_display())
.subcommand(
"exec-command",
from_fn_blocking(subcontainer::exec_command).no_display(),
)
.subcommand(
"create-fs",
from_fn_async(subcontainer::create_subcontainer_fs)
.with_custom_display_fn(|_, (path, _)| Ok(println!("{}", path.display())))
.with_call_remote::<ContainerCliContext>(),
)
.subcommand(
"destroy-fs",
from_fn_async(subcontainer::destroy_subcontainer_fs)
.no_display()
.with_call_remote::<ContainerCliContext>(),
),
)
// net
.subcommand("bind", from_fn_async(net::bind::bind).no_cli())
.subcommand(
"getServicePortForward",
"get-service-port-forward",
from_fn_async(net::bind::get_service_port_forward).no_cli(),
)
.subcommand(
"clearBindings",
"clear-bindings",
from_fn_async(net::bind::clear_bindings).no_cli(),
)
.subcommand(
"getHostInfo",
"get-host-info",
from_fn_async(net::host::get_host_info).no_cli(),
)
.subcommand(
"getPrimaryUrl",
"get-primary-url",
from_fn_async(net::host::get_primary_url).no_cli(),
)
.subcommand(
"getContainerIp",
"get-container-ip",
from_fn_async(net::info::get_container_ip).no_cli(),
)
.subcommand(
"exportServiceInterface",
"export-service-interface",
from_fn_async(net::interface::export_service_interface).no_cli(),
)
.subcommand(
"getServiceInterface",
"get-service-interface",
from_fn_async(net::interface::get_service_interface).no_cli(),
)
.subcommand(
"listServiceInterfaces",
"list-service-interfaces",
from_fn_async(net::interface::list_service_interfaces).no_cli(),
)
.subcommand(
"clearServiceInterfaces",
"clear-service-interfaces",
from_fn_async(net::interface::clear_service_interfaces).no_cli(),
)
.subcommand(
"getSslCertificate",
"get-ssl-certificate",
from_fn_async(net::ssl::get_ssl_certificate).no_cli(),
)
.subcommand("getSslKey", from_fn_async(net::ssl::get_ssl_key).no_cli())
.subcommand("get-ssl-key", from_fn_async(net::ssl::get_ssl_key).no_cli())
// store
.subcommand("getStore", from_fn_async(store::get_store).no_cli())
.subcommand("setStore", from_fn_async(store::set_store).no_cli())
.subcommand(
"store",
ParentHandler::<C>::new()
.subcommand("get", from_fn_async(store::get_store).no_cli())
.subcommand("set", from_fn_async(store::set_store).no_cli()),
)
.subcommand(
"set-data-version",
from_fn_async(store::set_data_version)
.no_display()
.with_call_remote::<ContainerCliContext>(),
)
.subcommand(
"get-data-version",
from_fn_async(store::get_data_version)
.with_custom_display_fn(|_, v| {
if let Some(v) = v {
println!("{v}")
} else {
println!("N/A")
}
Ok(())
})
.with_call_remote::<ContainerCliContext>(),
)
// system
.subcommand(
"getSystemSmtp",
"get-system-smtp",
from_fn_async(system::get_system_smtp).no_cli(),
)

3
core/startos/src/service/effects/net/interface.rs
View File

@@ -16,7 +16,6 @@ pub struct ExportServiceInterfaceParams {
name: String,
description: String,
has_primary: bool,
disabled: bool,
masked: bool,
address_info: AddressInfo,
r#type: ServiceInterfaceType,
@@ -28,7 +27,6 @@ pub async fn export_service_interface(
name,
description,
has_primary,
disabled,
masked,
address_info,
r#type,
@@ -42,7 +40,6 @@ pub async fn export_service_interface(
name,
description,
has_primary,
disabled,
masked,
address_info,
interface_type: r#type,

49
core/startos/src/service/effects/store.rs
View File

@@ -1,6 +1,6 @@
use imbl::vector;
use imbl_value::json;
use models::PackageId;
use models::{PackageId, VersionString};
use patch_db::json_ptr::JsonPointer;
use crate::service::effects::callbacks::CallbackHandler;
@@ -91,3 +91,50 @@ pub async fn set_store(
Ok(())
}
#[derive(Debug, Clone, Serialize, Deserialize, TS, Parser)]
#[serde(rename_all = "camelCase")]
#[ts(export)]
pub struct SetDataVersionParams {
#[ts(type = "string")]
version: VersionString,
}
pub async fn set_data_version(
context: EffectContext,
SetDataVersionParams { version }: SetDataVersionParams,
) -> Result<(), Error> {
let context = context.deref()?;
let package_id = &context.seed.id;
context
.seed
.ctx
.db
.mutate(|db| {
db.as_public_mut()
.as_package_data_mut()
.as_idx_mut(package_id)
.or_not_found(package_id)?
.as_data_version_mut()
.ser(&Some(version))
})
.await?;
Ok(())
}
pub async fn get_data_version(context: EffectContext) -> Result<Option<VersionString>, Error> {
let context = context.deref()?;
let package_id = &context.seed.id;
context
.seed
.ctx
.db
.peek()
.await
.as_public()
.as_package_data()
.as_idx(package_id)
.or_not_found(package_id)?
.as_data_version()
.de()
}

91
core/startos/src/service/effects/image.rs → core/startos/src/service/effects/subcontainer/mod.rs
View File

@@ -1,9 +1,6 @@
use std::ffi::OsString;
use std::os::unix::process::CommandExt;
use std::path::{Path, PathBuf};
use models::ImageId;
use rpc_toolkit::Context;
use tokio::process::Command;
use crate::disk::mount::filesystem::overlayfs::OverlayGuard;
@@ -11,89 +8,39 @@ use crate::rpc_continuations::Guid;
use crate::service::effects::prelude::*;
use crate::util::Invoke;
#[derive(Debug, Clone, Serialize, Deserialize, Parser)]
pub struct ChrootParams {
#[arg(short = 'e', long = "env")]
env: Option<PathBuf>,
#[arg(short = 'w', long = "workdir")]
workdir: Option<PathBuf>,
#[arg(short = 'u', long = "user")]
user: Option<String>,
path: PathBuf,
command: OsString,
args: Vec<OsString>,
}
pub fn chroot<C: Context>(
_: C,
ChrootParams {
env,
workdir,
user,
path,
command,
args,
}: ChrootParams,
) -> Result<(), Error> {
let mut cmd = std::process::Command::new(command);
if let Some(env) = env {
for (k, v) in std::fs::read_to_string(env)?
.lines()
.map(|l| l.trim())
.filter_map(|l| l.split_once("="))
{
cmd.env(k, v);
}
}
nix::unistd::setsid().ok(); // https://stackoverflow.com/questions/25701333/os-setsid-operation-not-permitted
std::os::unix::fs::chroot(path)?;
if let Some(uid) = user.as_deref().and_then(|u| u.parse::<u32>().ok()) {
cmd.uid(uid);
} else if let Some(user) = user {
let (uid, gid) = std::fs::read_to_string("/etc/passwd")?
.lines()
.find_map(|l| {
let mut split = l.trim().split(":");
if user != split.next()? {
return None;
}
split.next(); // throw away x
Some((split.next()?.parse().ok()?, split.next()?.parse().ok()?))
// uid gid
})
.or_not_found(lazy_format!("{user} in /etc/passwd"))?;
cmd.uid(uid);
cmd.gid(gid);
};
if let Some(workdir) = workdir {
cmd.current_dir(workdir);
}
cmd.args(args);
Err(cmd.exec().into())
}
#[cfg(feature = "container-runtime")]
mod sync;
#[cfg(not(feature = "container-runtime"))]
mod sync_dummy;
pub use sync::*;
#[cfg(not(feature = "container-runtime"))]
use sync_dummy as sync;
#[derive(Debug, Deserialize, Serialize, Parser, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export)]
pub struct DestroyOverlayedImageParams {
pub struct DestroySubcontainerFsParams {
guid: Guid,
}
#[instrument(skip_all)]
pub async fn destroy_overlayed_image(
pub async fn destroy_subcontainer_fs(
context: EffectContext,
DestroyOverlayedImageParams { guid }: DestroyOverlayedImageParams,
DestroySubcontainerFsParams { guid }: DestroySubcontainerFsParams,
) -> Result<(), Error> {
let context = context.deref()?;
if let Some(overlay) = context
.seed
.persistent_container
.overlays
.subcontainers
.lock()
.await
.remove(&guid)
{
overlay.unmount(true).await?;
} else {
tracing::warn!("Could not find a guard to remove on the destroy overlayed image; assumming that it already is removed and will be skipping");
tracing::warn!("Could not find a subcontainer fs to destroy; assumming that it already is destroyed and will be skipping");
}
Ok(())
}
@@ -101,13 +48,13 @@ pub async fn destroy_overlayed_image(
#[derive(Debug, Deserialize, Serialize, Parser, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export)]
pub struct CreateOverlayedImageParams {
pub struct CreateSubcontainerFsParams {
image_id: ImageId,
}
#[instrument(skip_all)]
pub async fn create_overlayed_image(
pub async fn create_subcontainer_fs(
context: EffectContext,
CreateOverlayedImageParams { image_id }: CreateOverlayedImageParams,
CreateSubcontainerFsParams { image_id }: CreateSubcontainerFsParams,
) -> Result<(PathBuf, Guid), Error> {
let context = context.deref()?;
if let Some(image) = context
@@ -131,7 +78,7 @@ pub async fn create_overlayed_image(
})?
.rootfs_dir();
let mountpoint = rootfs_dir
.join("media/startos/overlays")
.join("media/startos/subcontainers")
.join(guid.as_ref());
tokio::fs::create_dir_all(&mountpoint).await?;
let container_mountpoint = Path::new("/").join(
@@ -150,7 +97,7 @@ pub async fn create_overlayed_image(
context
.seed
.persistent_container
.overlays
.subcontainers
.lock()
.await
.insert(guid.clone(), guard);

392
core/startos/src/service/effects/subcontainer/sync.rs Normal file
View File

@@ -0,0 +1,392 @@
use std::borrow::Cow;
use std::collections::BTreeMap;
use std::ffi::{c_int, OsStr, OsString};
use std::fs::File;
use std::os::unix::process::CommandExt;
use std::path::{Path, PathBuf};
use std::process::{Command as StdCommand, Stdio};
use nix::sched::CloneFlags;
use nix::unistd::Pid;
use rpc_toolkit::Context;
use signal_hook::consts::signal::*;
use tokio::sync::oneshot;
use unshare::Command as NSCommand;
use crate::service::effects::prelude::*;
use crate::service::effects::ContainerCliContext;
const FWD_SIGNALS: &[c_int] = &[
SIGABRT, SIGALRM, SIGCONT, SIGHUP, SIGINT, SIGIO, SIGPIPE, SIGPROF, SIGQUIT, SIGTERM, SIGTRAP,
SIGTSTP, SIGTTIN, SIGTTOU, SIGURG, SIGUSR1, SIGUSR2, SIGVTALRM,
];
struct NSPid(Vec<i32>);
impl procfs::FromBufRead for NSPid {
fn from_buf_read<R: std::io::BufRead>(r: R) -> procfs::ProcResult<Self> {
for line in r.lines() {
let line = line?;
if let Some(row) = line.trim().strip_prefix("NSpid") {
return Ok(Self(
row.split_ascii_whitespace()
.map(|pid| pid.parse::<i32>())
.collect::<Result<Vec<_>, _>>()?,
));
}
}
Err(procfs::ProcError::Incomplete(None))
}
}
fn open_file_read(path: impl AsRef<Path>) -> Result<File, Error> {
File::open(&path).with_ctx(|_| {
(
ErrorKind::Filesystem,
lazy_format!("open r {}", path.as_ref().display()),
)
})
}
#[derive(Debug, Clone, Serialize, Deserialize, Parser)]
pub struct ExecParams {
#[arg(short = 'e', long = "env")]
env: Option<PathBuf>,
#[arg(short = 'w', long = "workdir")]
workdir: Option<PathBuf>,
#[arg(short = 'u', long = "user")]
user: Option<String>,
chroot: PathBuf,
#[arg(trailing_var_arg = true)]
command: Vec<OsString>,
}
impl ExecParams {
fn exec(&self) -> Result<(), Error> {
let ExecParams {
env,
workdir,
user,
chroot,
command,
} = self;
let Some(([command], args)) = command.split_at_checked(1) else {
return Err(Error::new(
eyre!("command cannot be empty"),
ErrorKind::InvalidRequest,
));
};
let env_string = if let Some(env) = &env {
std::fs::read_to_string(env)
.with_ctx(|_| (ErrorKind::Filesystem, lazy_format!("read {env:?}")))?
} else {
Default::default()
};
let env = env_string
.lines()
.map(|l| l.trim())
.filter_map(|l| l.split_once("="))
.collect::<BTreeMap<_, _>>();
std::os::unix::fs::chroot(chroot)
.with_ctx(|_| (ErrorKind::Filesystem, lazy_format!("chroot {chroot:?}")))?;
let command = which::which_in(
command,
env.get("PATH")
.copied()
.map(Cow::Borrowed)
.or_else(|| std::env::var("PATH").ok().map(Cow::Owned))
.as_deref(),
workdir.as_deref().unwrap_or(Path::new("/")),
)
.with_kind(ErrorKind::Filesystem)?;
let mut cmd = StdCommand::new(command);
cmd.args(args);
for (k, v) in env {
cmd.env(k, v);
}
if let Some(uid) = user.as_deref().and_then(|u| u.parse::<u32>().ok()) {
cmd.uid(uid);
} else if let Some(user) = user {
let (uid, gid) = std::fs::read_to_string("/etc/passwd")
.with_ctx(|_| (ErrorKind::Filesystem, "read /etc/passwd"))?
.lines()
.find_map(|l| {
let mut split = l.trim().split(":");
if user != split.next()? {
return None;
}
split.next(); // throw away x
Some((split.next()?.parse().ok()?, split.next()?.parse().ok()?))
// uid gid
})
.or_not_found(lazy_format!("{user} in /etc/passwd"))?;
cmd.uid(uid);
cmd.gid(gid);
};
if let Some(workdir) = workdir {
cmd.current_dir(workdir);
} else {
cmd.current_dir("/");
}
Err(cmd.exec().into())
}
}
pub fn launch(
_: ContainerCliContext,
ExecParams {
env,
workdir,
user,
chroot,
command,
}: ExecParams,
) -> Result<(), Error> {
use unshare::{Namespace, Stdio};
use crate::service::cli::ContainerCliContext;
let mut sig = signal_hook::iterator::Signals::new(FWD_SIGNALS)?;
let mut cmd = NSCommand::new("/usr/bin/start-cli");
cmd.arg("subcontainer").arg("launch-init");
if let Some(env) = env {
cmd.arg("--env").arg(env);
}
if let Some(workdir) = workdir {
cmd.arg("--workdir").arg(workdir);
}
if let Some(user) = user {
cmd.arg("--user").arg(user);
}
cmd.arg(&chroot);
cmd.args(&command);
cmd.unshare(&[Namespace::Pid, Namespace::Cgroup, Namespace::Ipc]);
cmd.stdin(Stdio::piped());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
let (stdin_send, stdin_recv) = oneshot::channel();
std::thread::spawn(move || {
if let Ok(mut stdin) = stdin_recv.blocking_recv() {
std::io::copy(&mut std::io::stdin(), &mut stdin).unwrap();
}
});
let (stdout_send, stdout_recv) = oneshot::channel();
std::thread::spawn(move || {
if let Ok(mut stdout) = stdout_recv.blocking_recv() {
std::io::copy(&mut stdout, &mut std::io::stdout()).unwrap();
}
});
let (stderr_send, stderr_recv) = oneshot::channel();
std::thread::spawn(move || {
if let Ok(mut stderr) = stderr_recv.blocking_recv() {
std::io::copy(&mut stderr, &mut std::io::stderr()).unwrap();
}
});
if chroot.join("proc/1").exists() {
let ns_id = procfs::process::Process::new_with_root(chroot.join("proc"))
.with_ctx(|_| (ErrorKind::Filesystem, "open subcontainer procfs"))?
.namespaces()
.with_ctx(|_| (ErrorKind::Filesystem, "read subcontainer pid 1 ns"))?
.0
.get(OsStr::new("pid"))
.or_not_found("pid namespace")?
.identifier;
for proc in
procfs::process::all_processes().with_ctx(|_| (ErrorKind::Filesystem, "open procfs"))?
{
let proc = proc.with_ctx(|_| (ErrorKind::Filesystem, "read single process details"))?;
let pid = proc.pid();
if proc
.namespaces()
.with_ctx(|_| (ErrorKind::Filesystem, lazy_format!("read pid {} ns", pid)))?
.0
.get(OsStr::new("pid"))
.map_or(false, |ns| ns.identifier == ns_id)
{
let pids = proc.read::<NSPid>("status").with_ctx(|_| {
(
ErrorKind::Filesystem,
lazy_format!("read pid {} NSpid", pid),
)
})?;
if pids.0.len() == 2 && pids.0[1] == 1 {
nix::sys::signal::kill(Pid::from_raw(pid), nix::sys::signal::SIGKILL)
.with_ctx(|_| {
(
ErrorKind::Filesystem,
lazy_format!(
"kill pid {} (determined to be pid 1 in subcontainer)",
pid
),
)
})?;
}
}
}
nix::mount::umount(&chroot.join("proc"))
.with_ctx(|_| (ErrorKind::Filesystem, "unmounting subcontainer procfs"))?;
}
let mut child = cmd
.spawn()
.map_err(color_eyre::eyre::Report::msg)
.with_ctx(|_| (ErrorKind::Filesystem, "spawning child process"))?;
let pid = child.pid();
std::thread::spawn(move || {
for sig in sig.forever() {
nix::sys::signal::kill(
Pid::from_raw(pid),
Some(nix::sys::signal::Signal::try_from(sig).unwrap()),
)
.unwrap();
}
});
stdin_send
.send(child.stdin.take().unwrap())
.unwrap_or_default();
stdout_send
.send(child.stdout.take().unwrap())
.unwrap_or_default();
stderr_send
.send(child.stderr.take().unwrap())
.unwrap_or_default();
// TODO: subreaping, signal handling
let exit = child
.wait()
.with_ctx(|_| (ErrorKind::Filesystem, "waiting on child process"))?;
if let Some(code) = exit.code() {
std::process::exit(code);
} else {
if exit.success() {
Ok(())
} else {
Err(Error::new(
color_eyre::eyre::Report::msg(exit),
ErrorKind::Unknown,
))
}
}
}
pub fn launch_init(_: ContainerCliContext, params: ExecParams) -> Result<(), Error> {
nix::mount::mount(
Some("proc"),
&params.chroot.join("proc"),
Some("proc"),
nix::mount::MsFlags::empty(),
None::<&str>,
)
.with_ctx(|_| (ErrorKind::Filesystem, "mount procfs"))?;
if params.command.is_empty() {
signal_hook::iterator::Signals::new(signal_hook::consts::TERM_SIGNALS)?
.forever()
.next();
std::process::exit(0)
} else {
params.exec()
}
}
pub fn exec(
_: ContainerCliContext,
ExecParams {
env,
workdir,
user,
chroot,
command,
}: ExecParams,
) -> Result<(), Error> {
let mut sig = signal_hook::iterator::Signals::new(FWD_SIGNALS)?;
let (send_pid, recv_pid) = oneshot::channel();
std::thread::spawn(move || {
if let Ok(pid) = recv_pid.blocking_recv() {
for sig in sig.forever() {
nix::sys::signal::kill(
Pid::from_raw(pid),
Some(nix::sys::signal::Signal::try_from(sig).unwrap()),
)
.unwrap();
}
}
});
let mut cmd = StdCommand::new("/usr/bin/start-cli");
cmd.arg("subcontainer").arg("exec-command");
if let Some(env) = env {
cmd.arg("--env").arg(env);
}
if let Some(workdir) = workdir {
cmd.arg("--workdir").arg(workdir);
}
if let Some(user) = user {
cmd.arg("--user").arg(user);
}
cmd.arg(&chroot);
cmd.args(&command);
cmd.stdin(Stdio::piped());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
let (stdin_send, stdin_recv) = oneshot::channel();
std::thread::spawn(move || {
if let Ok(mut stdin) = stdin_recv.blocking_recv() {
std::io::copy(&mut std::io::stdin(), &mut stdin).unwrap();
}
});
let (stdout_send, stdout_recv) = oneshot::channel();
std::thread::spawn(move || {
if let Ok(mut stdout) = stdout_recv.blocking_recv() {
std::io::copy(&mut stdout, &mut std::io::stdout()).unwrap();
}
});
let (stderr_send, stderr_recv) = oneshot::channel();
std::thread::spawn(move || {
if let Ok(mut stderr) = stderr_recv.blocking_recv() {
std::io::copy(&mut stderr, &mut std::io::stderr()).unwrap();
}
});
nix::sched::setns(
open_file_read(chroot.join("proc/1/ns/pid"))?,
CloneFlags::CLONE_NEWPID,
)
.with_ctx(|_| (ErrorKind::Filesystem, "set pid ns"))?;
nix::sched::setns(
open_file_read(chroot.join("proc/1/ns/cgroup"))?,
CloneFlags::CLONE_NEWCGROUP,
)
.with_ctx(|_| (ErrorKind::Filesystem, "set cgroup ns"))?;
nix::sched::setns(
open_file_read(chroot.join("proc/1/ns/ipc"))?,
CloneFlags::CLONE_NEWIPC,
)
.with_ctx(|_| (ErrorKind::Filesystem, "set ipc ns"))?;
let mut child = cmd
.spawn()
.map_err(color_eyre::eyre::Report::msg)
.with_ctx(|_| (ErrorKind::Filesystem, "spawning child process"))?;
send_pid.send(child.id() as i32).unwrap_or_default();
stdin_send
.send(child.stdin.take().unwrap())
.unwrap_or_default();
stdout_send
.send(child.stdout.take().unwrap())
.unwrap_or_default();
stderr_send
.send(child.stderr.take().unwrap())
.unwrap_or_default();
let exit = child
.wait()
.with_ctx(|_| (ErrorKind::Filesystem, "waiting on child process"))?;
if let Some(code) = exit.code() {
std::process::exit(code);
} else {
if exit.success() {
Ok(())
} else {
Err(Error::new(
color_eyre::eyre::Report::msg(exit),
ErrorKind::Unknown,
))
}
}
}
pub fn exec_command(_: ContainerCliContext, params: ExecParams) -> Result<(), Error> {
params.exec()
}

30
core/startos/src/service/effects/subcontainer/sync_dummy.rs Normal file
View File

@@ -0,0 +1,30 @@
use crate::service::effects::prelude::*;
use crate::service::effects::ContainerCliContext;
pub fn launch(_: ContainerCliContext) -> Result<(), Error> {
Err(Error::new(
eyre!("requires feature container-runtime"),
ErrorKind::InvalidRequest,
))
}
pub fn launch_init(_: ContainerCliContext) -> Result<(), Error> {
Err(Error::new(
eyre!("requires feature container-runtime"),
ErrorKind::InvalidRequest,
))
}
pub fn exec(_: ContainerCliContext) -> Result<(), Error> {
Err(Error::new(
eyre!("requires feature container-runtime"),
ErrorKind::InvalidRequest,
))
}
pub fn exec_command(_: ContainerCliContext) -> Result<(), Error> {
Err(Error::new(
eyre!("requires feature container-runtime"),
ErrorKind::InvalidRequest,
))
}

6
core/startos/src/service/mod.rs
View File

@@ -27,7 +27,7 @@ use crate::progress::{NamedProgress, Progress};
use crate::rpc_continuations::Guid;
use crate::s9pk::S9pk;
use crate::service::service_map::InstallProgressHandles;
use crate::status::health_check::HealthCheckResult;
use crate::status::health_check::NamedHealthCheckResult;
use crate::util::actor::concurrent::ConcurrentActor;
use crate::util::io::create_file;
use crate::util::serde::{NoOutput, Pem};
@@ -45,7 +45,7 @@ mod properties;
mod rpc;
mod service_actor;
pub mod service_map;
mod start_stop;
pub mod start_stop;
mod transition;
mod util;
@@ -493,7 +493,6 @@ impl Service {
#[derive(Debug, Clone)]
pub struct RunningStatus {
health: OrdMap<HealthCheckId, HealthCheckResult>,
started: DateTime<Utc>,
}
@@ -516,7 +515,6 @@ impl ServiceActorSeed {
.running_status
.take()
.unwrap_or_else(|| RunningStatus {
health: Default::default(),
started: Utc::now(),
}),
);

8
core/startos/src/service/persistent_container.rs
View File

@@ -98,7 +98,7 @@ pub struct PersistentContainer {
volumes: BTreeMap<VolumeId, MountGuard>,
assets: BTreeMap<VolumeId, MountGuard>,
pub(super) images: BTreeMap<ImageId, Arc<MountGuard>>,
pub(super) overlays: Arc<Mutex<BTreeMap<Guid, OverlayGuard<Arc<MountGuard>>>>>,
pub(super) subcontainers: Arc<Mutex<BTreeMap<Guid, OverlayGuard<Arc<MountGuard>>>>>,
pub(super) state: Arc<watch::Sender<ServiceState>>,
pub(super) net_service: Mutex<NetService>,
destroyed: bool,
@@ -273,7 +273,7 @@ impl PersistentContainer {
volumes,
assets,
images,
overlays: Arc::new(Mutex::new(BTreeMap::new())),
subcontainers: Arc::new(Mutex::new(BTreeMap::new())),
state: Arc::new(watch::channel(ServiceState::new(start)).0),
net_service: Mutex::new(net_service),
destroyed: false,
@@ -388,7 +388,7 @@ impl PersistentContainer {
let volumes = std::mem::take(&mut self.volumes);
let assets = std::mem::take(&mut self.assets);
let images = std::mem::take(&mut self.images);
let overlays = self.overlays.clone();
let subcontainers = self.subcontainers.clone();
let lxc_container = self.lxc_container.take();
self.destroyed = true;
Some(async move {
@@ -404,7 +404,7 @@ impl PersistentContainer {
for (_, assets) in assets {
errs.handle(assets.unmount(true).await);
}
for (_, overlay) in std::mem::take(&mut *overlays.lock().await) {
for (_, overlay) in std::mem::take(&mut *subcontainers.lock().await) {
errs.handle(overlay.unmount(true).await);
}
for (_, images) in images {

142
core/startos/src/service/service_actor.rs
View File

@@ -1,11 +1,10 @@
use std::sync::Arc;
use std::time::Duration;
use imbl::OrdMap;
use super::start_stop::StartStop;
use super::ServiceActorSeed;
use crate::prelude::*;
use crate::service::persistent_container::ServiceStateKinds;
use crate::service::transition::TransitionKind;
use crate::service::SYNC_RETRY_COOLDOWN_SECONDS;
use crate::status::MainStatus;
@@ -46,96 +45,77 @@ async fn service_actor_loop(
let id = &seed.id;
let kinds = current.borrow().kinds();
if let Err(e) = async {
let main_status = match (
kinds.transition_state,
kinds.desired_state,
kinds.running_status,
) {
(Some(TransitionKind::Restarting), StartStop::Stop, Some(_)) => {
seed.persistent_container.stop().await?;
MainStatus::Restarting
}
(Some(TransitionKind::Restarting), StartStop::Start, _) => {
seed.persistent_container.start().await?;
MainStatus::Restarting
}
(Some(TransitionKind::Restarting), _, _) => MainStatus::Restarting,
(Some(TransitionKind::Restoring), _, _) => MainStatus::Restoring,
(Some(TransitionKind::BackingUp), StartStop::Stop, Some(status)) => {
seed.persistent_container.stop().await?;
MainStatus::BackingUp {
started: Some(status.started),
health: status.health.clone(),
}
}
(Some(TransitionKind::BackingUp), StartStop::Start, _) => {
seed.persistent_container.start().await?;
MainStatus::BackingUp {
started: None,
health: OrdMap::new(),
}
}
(Some(TransitionKind::BackingUp), _, _) => MainStatus::BackingUp {
started: None,
health: OrdMap::new(),
},
(None, StartStop::Stop, None) => MainStatus::Stopped,
(None, StartStop::Stop, Some(_)) => {
let task_seed = seed.clone();
seed.ctx
.db
.mutate(|d| {
if let Some(i) = d.as_public_mut().as_package_data_mut().as_idx_mut(&id) {
i.as_status_mut().as_main_mut().ser(&MainStatus::Stopping)?;
}
Ok(())
})
.await?;
task_seed.persistent_container.stop().await?;
MainStatus::Stopped
}
(None, StartStop::Start, Some(status)) => MainStatus::Running {
started: status.started,
health: status.health.clone(),
},
(None, StartStop::Start, None) => {
seed.persistent_container.start().await?;
MainStatus::Starting
}
};
seed.ctx
.db
.mutate(|d| {
if let Some(i) = d.as_public_mut().as_package_data_mut().as_idx_mut(&id) {
let previous = i.as_status().as_main().de()?;
let previous_health = previous.health();
let previous_started = previous.started();
let mut main_status = main_status;
match &mut main_status {
&mut MainStatus::Running { ref mut health, .. }
| &mut MainStatus::BackingUp { ref mut health, .. } => {
*health = previous_health.unwrap_or(health).clone();
}
_ => (),
};
match &mut main_status {
MainStatus::Running {
ref mut started, ..
} => {
*started = previous_started.unwrap_or(*started);
}
MainStatus::BackingUp {
ref mut started, ..
} => {
*started = previous_started.map(Some).unwrap_or(*started);
}
_ => (),
let main_status = match &kinds {
ServiceStateKinds {
transition_state: Some(TransitionKind::Restarting),
..
} => MainStatus::Restarting,
ServiceStateKinds {
transition_state: Some(TransitionKind::Restoring),
..
} => MainStatus::Restoring,
ServiceStateKinds {
transition_state: Some(TransitionKind::BackingUp),
..
} => previous.backing_up(),
ServiceStateKinds {
running_status: Some(status),
desired_state: StartStop::Start,
..
} => MainStatus::Running {
started: status.started,
health: previous.health().cloned().unwrap_or_default(),
},
ServiceStateKinds {
running_status: None,
desired_state: StartStop::Start,
..
} => MainStatus::Starting {
health: previous.health().cloned().unwrap_or_default(),
},
ServiceStateKinds {
running_status: Some(_),
desired_state: StartStop::Stop,
..
} => MainStatus::Stopping,
ServiceStateKinds {
running_status: None,
desired_state: StartStop::Stop,
..
} => MainStatus::Stopped,
};
i.as_status_mut().as_main_mut().ser(&main_status)?;
}
Ok(())
})
.await?;
seed.synchronized.notify_waiters();
match kinds {
ServiceStateKinds {
running_status: None,
desired_state: StartStop::Start,
..
} => {
seed.persistent_container.start().await?;
}
ServiceStateKinds {
running_status: Some(_),
desired_state: StartStop::Stop,
..
} => {
seed.persistent_container.stop().await?;
seed.persistent_container
.state
.send_if_modified(|s| s.running_status.take().is_some());
}
_ => (),
};
Ok::<_, Error>(())
}

1
core/startos/src/service/service_map.rs
View File

@@ -173,6 +173,7 @@ impl ServiceMap {
} else {
PackageState::Installing(installing)
},
data_version: None,
status: Status {
configured: false,
main: MainStatus::Stopped,

42
core/startos/src/service/start_stop.rs
View File

@@ -1,6 +1,10 @@
use serde::{Deserialize, Serialize};
use ts_rs::TS;
use crate::status::MainStatus;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize, TS)]
#[serde(rename_all = "camelCase")]
pub enum StartStop {
Start,
Stop,
@@ -11,23 +15,19 @@ impl StartStop {
matches!(self, StartStop::Start)
}
}
impl From<MainStatus> for StartStop {
fn from(value: MainStatus) -> Self {
match value {
MainStatus::Stopped => StartStop::Stop,
MainStatus::Restoring => StartStop::Stop,
MainStatus::Restarting => StartStop::Start,
MainStatus::Stopping { .. } => StartStop::Stop,
MainStatus::Starting => StartStop::Start,
MainStatus::Running {
started: _,
health: _,
} => StartStop::Start,
MainStatus::BackingUp { started, health: _ } if started.is_some() => StartStop::Start,
MainStatus::BackingUp {
started: _,
health: _,
} => StartStop::Stop,
}
}
}
// impl From<MainStatus> for StartStop {
// fn from(value: MainStatus) -> Self {
// match value {
// MainStatus::Stopped => StartStop::Stop,
// MainStatus::Restoring => StartStop::Stop,
// MainStatus::Restarting => StartStop::Start,
// MainStatus::Stopping { .. } => StartStop::Stop,
// MainStatus::Starting => StartStop::Start,
// MainStatus::Running {
// started: _,
// health: _,
// } => StartStop::Start,
// MainStatus::BackingUp { on_complete } => on_complete,
// }
// }
// }

36
core/startos/src/status/health_check.rs
View File

@@ -9,25 +9,25 @@ use crate::util::clap::FromStrParser;
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq, TS)]
#[serde(rename_all = "camelCase")]
pub struct HealthCheckResult {
pub struct NamedHealthCheckResult {
pub name: String,
#[serde(flatten)]
pub kind: HealthCheckResultKind,
pub kind: NamedHealthCheckResultKind,
}
// healthCheckName:kind:message OR healthCheckName:kind
impl FromStr for HealthCheckResult {
impl FromStr for NamedHealthCheckResult {
type Err = color_eyre::eyre::Report;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let from_parts = |name: &str, kind: &str, message: Option<&str>| {
let message = message.map(|x| x.to_string());
let kind = match kind {
"success" => HealthCheckResultKind::Success { message },
"disabled" => HealthCheckResultKind::Disabled { message },
"starting" => HealthCheckResultKind::Starting { message },
"loading" => HealthCheckResultKind::Loading {
"success" => NamedHealthCheckResultKind::Success { message },
"disabled" => NamedHealthCheckResultKind::Disabled { message },
"starting" => NamedHealthCheckResultKind::Starting { message },
"loading" => NamedHealthCheckResultKind::Loading {
message: message.unwrap_or_default(),
},
"failure" => HealthCheckResultKind::Failure {
"failure" => NamedHealthCheckResultKind::Failure {
message: message.unwrap_or_default(),
},
_ => return Err(color_eyre::eyre::eyre!("Invalid health check kind")),
@@ -47,7 +47,7 @@ impl FromStr for HealthCheckResult {
}
}
}
impl ValueParserFactory for HealthCheckResult {
impl ValueParserFactory for NamedHealthCheckResult {
type Parser = FromStrParser<Self>;
fn value_parser() -> Self::Parser {
FromStrParser::new()
@@ -57,40 +57,44 @@ impl ValueParserFactory for HealthCheckResult {
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq, TS)]
#[serde(rename_all = "camelCase")]
#[serde(tag = "result")]
pub enum HealthCheckResultKind {
pub enum NamedHealthCheckResultKind {
Success { message: Option<String> },
Disabled { message: Option<String> },
Starting { message: Option<String> },
Loading { message: String },
Failure { message: String },
}
impl std::fmt::Display for HealthCheckResult {
impl std::fmt::Display for NamedHealthCheckResult {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let name = &self.name;
match &self.kind {
HealthCheckResultKind::Success { message } => {
NamedHealthCheckResultKind::Success { message } => {
if let Some(message) = message {
write!(f, "{name}: Succeeded ({message})")
} else {
write!(f, "{name}: Succeeded")
}
}
HealthCheckResultKind::Disabled { message } => {
NamedHealthCheckResultKind::Disabled { message } => {
if let Some(message) = message {
write!(f, "{name}: Disabled ({message})")
} else {
write!(f, "{name}: Disabled")
}
}
HealthCheckResultKind::Starting { message } => {
NamedHealthCheckResultKind::Starting { message } => {
if let Some(message) = message {
write!(f, "{name}: Starting ({message})")
} else {
write!(f, "{name}: Starting")
}
}
HealthCheckResultKind::Loading { message } => write!(f, "{name}: Loading ({message})"),
HealthCheckResultKind::Failure { message } => write!(f, "{name}: Failed ({message})"),
NamedHealthCheckResultKind::Loading { message } => {
write!(f, "{name}: Loading ({message})")
}
NamedHealthCheckResultKind::Failure { message } => {
write!(f, "{name}: Failed ({message})")
}
}
}
}

85
core/startos/src/status/mod.rs
View File

@@ -1,4 +1,4 @@
use std::{collections::BTreeMap, sync::Arc};
use std::collections::BTreeMap;
use chrono::{DateTime, Utc};
use imbl::OrdMap;
@@ -6,8 +6,9 @@ use serde::{Deserialize, Serialize};
use ts_rs::TS;
use self::health_check::HealthCheckId;
use crate::status::health_check::HealthCheckResult;
use crate::{prelude::*, util::GeneralGuard};
use crate::prelude::*;
use crate::service::start_stop::StartStop;
use crate::status::health_check::NamedHealthCheckResult;
pub mod health_check;
#[derive(Clone, Debug, Deserialize, Serialize, HasModel, TS)]
@@ -22,25 +23,24 @@ pub struct Status {
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, TS)]
#[serde(tag = "status")]
#[serde(rename_all = "camelCase")]
#[serde(rename_all_fields = "camelCase")]
pub enum MainStatus {
Stopped,
Restarting,
Restoring,
Stopping,
Starting,
#[serde(rename_all = "camelCase")]
Starting {
#[ts(as = "BTreeMap<HealthCheckId, NamedHealthCheckResult>")]
health: OrdMap<HealthCheckId, NamedHealthCheckResult>,
},
Running {
#[ts(type = "string")]
started: DateTime<Utc>,
#[ts(as = "BTreeMap<HealthCheckId, HealthCheckResult>")]
health: OrdMap<HealthCheckId, HealthCheckResult>,
#[ts(as = "BTreeMap<HealthCheckId, NamedHealthCheckResult>")]
health: OrdMap<HealthCheckId, NamedHealthCheckResult>,
},
#[serde(rename_all = "camelCase")]
BackingUp {
#[ts(type = "string | null")]
started: Option<DateTime<Utc>>,
#[ts(as = "BTreeMap<HealthCheckId, HealthCheckResult>")]
health: OrdMap<HealthCheckId, HealthCheckResult>,
on_complete: StartStop,
},
}
impl MainStatus {
@@ -48,60 +48,37 @@ impl MainStatus {
match self {
MainStatus::Starting { .. }
| MainStatus::Running { .. }
| MainStatus::Restarting
| MainStatus::BackingUp {
started: Some(_), ..
on_complete: StartStop::Start,
} => true,
MainStatus::Stopped
| MainStatus::Restoring
| MainStatus::Stopping { .. }
| MainStatus::Restarting
| MainStatus::BackingUp { started: None, .. } => false,
| MainStatus::BackingUp {
on_complete: StartStop::Stop,
} => false,
}
}
// pub fn stop(&mut self) {
// match self {
// MainStatus::Starting { .. } | MainStatus::Running { .. } => {
// *self = MainStatus::Stopping;
// }
// MainStatus::BackingUp { started, .. } => {
// *started = None;
// }
// MainStatus::Stopped | MainStatus::Stopping | MainStatus::Restarting => (),
// }
// }
pub fn started(&self) -> Option<DateTime<Utc>> {
match self {
MainStatus::Running { started, .. } => Some(*started),
MainStatus::BackingUp { started, .. } => *started,
MainStatus::Stopped => None,
MainStatus::Restoring => None,
MainStatus::Restarting => None,
MainStatus::Stopping { .. } => None,
MainStatus::Starting { .. } => None,
}
}
pub fn backing_up(&self) -> Self {
let (started, health) = match self {
MainStatus::Starting { .. } => (Some(Utc::now()), Default::default()),
MainStatus::Running { started, health } => (Some(started.clone()), health.clone()),
MainStatus::Stopped
| MainStatus::Stopping { .. }
| MainStatus::Restoring
| MainStatus::Restarting => (None, Default::default()),
MainStatus::BackingUp { .. } => return self.clone(),
};
MainStatus::BackingUp { started, health }
}
pub fn health(&self) -> Option<&OrdMap<HealthCheckId, HealthCheckResult>> {
pub fn backing_up(self) -> Self {
MainStatus::BackingUp {
on_complete: if self.running() {
StartStop::Start
} else {
StartStop::Stop
},
}
}
pub fn health(&self) -> Option<&OrdMap<HealthCheckId, NamedHealthCheckResult>> {
match self {
MainStatus::Running { health, .. } => Some(health),
MainStatus::BackingUp { health, .. } => Some(health),
MainStatus::Stopped
MainStatus::Running { health, .. } | MainStatus::Starting { health } => Some(health),
MainStatus::BackingUp { .. }
| MainStatus::Stopped
| MainStatus::Restoring
| MainStatus::Stopping { .. }
| MainStatus::Restarting => None,
MainStatus::Starting { .. } => None,
}
}
}

2
core/startos/src/version/mod.rs
View File

@@ -22,7 +22,7 @@ mod v0_3_6_alpha_5;
mod v0_3_6_alpha_6;
mod v0_3_6_alpha_7;
pub type Current = v0_3_6_alpha_3::Version; // VERSION_BUMP
pub type Current = v0_3_6_alpha_5::Version; // VERSION_BUMP
#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
#[serde(untagged)]