Start9/start-os
2
0
Fork 0
mirror of https://github.com/Start9Labs/start-os.git synced 2026-03-26 10:21:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

wip: Refactoring the service

Browse Source 
-> Made new skeleton
-> Added service manager
-> Manager Refactored
-> Cleanup
-> Add gid struct
-> remove synchronizer
-> Added backup into manager
-> Fix the configure signal not send
-> Fixes around backup and sync

wip: Moved over the config into the service manager
This commit is contained in:
BluJ
2023-01-31 15:36:25 -07:00
committed by Aiden McClelland
parent bada88157e
commit 93c751f6eb
17 changed files with 1611 additions and 1143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
backend/src/backup/backup_bulk.rs
View File

@@ -1,5 +1,6 @@
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use chrono::Utc;
use clap::ArgMatches;
@@ -7,8 +8,7 @@ use color_eyre::eyre::eyre;
use helpers::AtomicFile;
use patch_db::{DbHandle, LockType, PatchDbHandle};
use rpc_toolkit::command;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
use tokio::{io::AsyncWriteExt, sync::Mutex};
use tracing::instrument;
use super::target::BackupTargetId;
@@ -21,9 +21,9 @@ use crate::db::model::BackupProgress;
use crate::disk::mount::backup::BackupMountGuard;
use crate::disk::mount::filesystem::ReadWrite;
use crate::disk::mount::guard::TmpMountGuard;
use crate::manager::BackupReturn;
use crate::notifications::NotificationLevel;
use crate::s9pk::manifest::PackageId;
use crate::status::MainStatus;
use crate::util::io::dir_copy;
use crate::util::serde::IoFormat;
use crate::util::{display_none, Invoke};
@@ -206,10 +206,12 @@ async fn assure_backing_up(
async fn perform_backup<Db: DbHandle>(
ctx: &RpcContext,
mut db: Db,
mut backup_guard: BackupMountGuard<TmpMountGuard>,
backup_guard: BackupMountGuard<TmpMountGuard>,
package_ids: &BTreeSet<PackageId>,
) -> Result<BTreeMap<PackageId, PackageBackupReport>, Error> {
let mut backup_report = BTreeMap::new();
let backup_guard = Arc::new(Mutex::new(backup_guard));
for package_id in crate::db::DatabaseModel::new()
.package_data()
.keys(&mut db)
@@ -231,93 +233,55 @@ async fn perform_backup<Db: DbHandle>(
};
let main_status_model = installed_model.clone().status().main();
main_status_model.lock(&mut tx, LockType::Write).await?;
let (started, health) = match main_status_model.get(&mut tx).await?.into_owned() {
MainStatus::Starting { .. } => (Some(Utc::now()), Default::default()),
MainStatus::Running { started, health } => (Some(started), health.clone()),
MainStatus::Stopped | MainStatus::Stopping | MainStatus::Restarting => {
(None, Default::default())
}
MainStatus::BackingUp { .. } => {
backup_report.insert(
package_id,
PackageBackupReport {
error: Some(
"Can't do backup because service is in a backing up state".to_owned(),
),
},
);
continue;
}
};
main_status_model
.put(
&mut tx,
&MainStatus::BackingUp {
started,
health: health.clone(),
},
)
.await?;
tx.save().await?; // drop locks
let manifest = installed_model.clone().manifest().get(&mut tx).await?;
let manifest = installed_model.clone().manifest().get(&mut db).await?;
ctx.managers
let (response, report) = match ctx
.managers
.get(&(manifest.id.clone(), manifest.version.clone()))
.await
.ok_or_else(|| {
Error::new(eyre!("Manager not found"), crate::ErrorKind::InvalidRequest)
})?
.synchronize()
.await;
let mut tx = db.begin().await?;
installed_model.lock(&mut tx, LockType::Write).await?;
let guard = backup_guard.mount_package_backup(&package_id).await?;
let res = manifest
.backup
.create(
ctx,
&mut tx,
&package_id,
&manifest.title,
&manifest.version,
&manifest.interfaces,
&manifest.volumes,
)
.await;
guard.unmount().await?;
.backup(backup_guard.clone())
.await
{
BackupReturn::Ran { report, res } => (res, report),
BackupReturn::AlreadyRunning(report) => {
backup_report.insert(package_id, report);
continue;
}
BackupReturn::Error(error) => {
tracing::warn!("Backup thread error");
tracing::debug!("{error:?}");
backup_report.insert(
package_id,
PackageBackupReport {
error: Some("Backup thread error".to_owned()),
},
);
continue;
}
};
backup_report.insert(
package_id.clone(),
PackageBackupReport {
error: res.as_ref().err().map(|e| e.to_string()),
error: response.as_ref().err().map(|e| e.to_string()),
},
);
if let Ok(pkg_meta) = res {
if let Ok(pkg_meta) = response {
installed_model
.last_backup()
.put(&mut tx, &Some(pkg_meta.timestamp))
.await?;
backup_guard
.lock()
.await
.metadata
.package_backups
.insert(package_id.clone(), pkg_meta);
}
main_status_model
.put(
&mut tx,
&match started {
Some(started) => MainStatus::Running { started, health },
None => MainStatus::Stopped,
},
)
.await?;
let mut backup_progress = crate::db::DatabaseModel::new()
.server_info()
.status_info()
@@ -344,7 +308,7 @@ async fn perform_backup<Db: DbHandle>(
.into_owned();
let mut os_backup_file = AtomicFile::new(
backup_guard.as_ref().join("os-backup.cbor"),
backup_guard.lock().await.as_ref().join("os-backup.cbor"),
None::<PathBuf>,
)
.await
@@ -374,6 +338,14 @@ async fn perform_backup<Db: DbHandle>(
}
let timestamp = Some(Utc::now());
let mut backup_guard = Arc::try_unwrap(backup_guard)
.map_err(|_err| {
Error::new(
eyre!("Backup guard could not ensure that the others where dropped"),
ErrorKind::Unknown,
)
})?
.into_inner();
backup_guard.unencrypted_metadata.version = crate::version::Current::new().semver().into();
backup_guard.unencrypted_metadata.full = true;

427
backend/src/config/mod.rs
View File

@@ -6,7 +6,10 @@ use color_eyre::eyre::eyre;
use futures::future::{BoxFuture, FutureExt};
use indexmap::IndexSet;
use itertools::Itertools;
use patch_db::{DbHandle, LockReceipt, LockTarget, LockTargetId, LockType, Verifier};
use models::ErrorKind;
use patch_db::{
DbHandle, LockReceipt, LockTarget, LockTargetId, LockType, PatchDbHandle, Transaction, Verifier,
};
use rand::SeedableRng;
use regex::Regex;
use rpc_toolkit::command;
@@ -264,18 +267,18 @@ pub struct ConfigReceipts {
pub update_dependency_receipts: UpdateDependencyReceipts,
pub try_heal_receipts: TryHealReceipts,
pub break_transitive_receipts: BreakTransitiveReceipts,
configured: LockReceipt<bool, String>,
config_actions: LockReceipt<ConfigActions, String>,
dependencies: LockReceipt<Dependencies, String>,
volumes: LockReceipt<crate::volume::Volumes, String>,
version: LockReceipt<crate::util::Version, String>,
manifest: LockReceipt<Manifest, String>,
system_pointers: LockReceipt<Vec<spec::SystemPointerSpec>, String>,
pub configured: LockReceipt<bool, String>,
pub config_actions: LockReceipt<ConfigActions, String>,
pub dependencies: LockReceipt<Dependencies, String>,
pub volumes: LockReceipt<crate::volume::Volumes, String>,
pub version: LockReceipt<crate::util::Version, String>,
pub manifest: LockReceipt<Manifest, String>,
pub system_pointers: LockReceipt<Vec<spec::SystemPointerSpec>, String>,
pub current_dependents: LockReceipt<CurrentDependents, String>,
pub current_dependencies: LockReceipt<CurrentDependencies, String>,
dependency_errors: LockReceipt<DependencyErrors, String>,
manifest_dependencies_config: LockReceipt<DependencyConfig, (String, String)>,
docker_containers: LockReceipt<DockerContainers, String>,
pub dependency_errors: LockReceipt<DependencyErrors, String>,
pub manifest_dependencies_config: LockReceipt<DependencyConfig, (String, String)>,
pub docker_containers: LockReceipt<DockerContainers, String>,
}
impl ConfigReceipts {
@@ -418,370 +421,78 @@ pub async fn set_dry(
#[context] ctx: RpcContext,
#[parent_data] (id, config, timeout): (PackageId, Option<Config>, Option<Duration>),
) -> Result<BreakageRes, Error> {
let mut db = ctx.db.handle();
let mut tx = db.begin().await?;
let mut breakages = BTreeMap::new();
let locks = ConfigReceipts::new(&mut tx).await?;
configure(
&ctx,
&mut tx,
&id,
config,
&timeout,
true,
&mut BTreeMap::new(),
&mut breakages,
&locks,
)
.await?;
let breakages = BTreeMap::new();
let overrides = Default::default();
let configure_context = ConfigureContext {
breakages,
timeout,
config,
dry_run: true,
overrides,
};
let breakages = configure(&ctx, &id, configure_context).await?;
locks.configured.set(&mut tx, true, &id).await?;
tx.abort().await?;
Ok(BreakageRes(breakages))
}
pub struct ConfigureContext {
pub breakages: BTreeMap<PackageId, TaggedDependencyError>,
pub timeout: Option<Duration>,
pub config: Option<Config>,
pub overrides: BTreeMap<PackageId, Config>,
pub dry_run: bool,
}
#[instrument(skip_all)]
pub async fn set_impl(
ctx: RpcContext,
(id, config, timeout): (PackageId, Option<Config>, Option<Duration>),
) -> Result<(), Error> {
let mut db = ctx.db.handle();
let mut tx = db.begin().await?;
let mut breakages = BTreeMap::new();
let locks = ConfigReceipts::new(&mut tx).await?;
configure(
&ctx,
&mut tx,
&id,
let breakages = BTreeMap::new();
let overrides = Default::default();
let configure_context = ConfigureContext {
breakages,
timeout,
config,
&timeout,
false,
&mut BTreeMap::new(),
&mut breakages,
&locks,
)
.await?;
tx.commit().await?;
dry_run: false,
overrides,
};
configure(&ctx, &id, configure_context).await?;
Ok(())
}
#[instrument(skip_all)]
pub async fn configure<'a, Db: DbHandle>(
pub async fn configure(
ctx: &RpcContext,
db: &'a mut Db,
id: &PackageId,
config: Option<Config>,
timeout: &Option<Duration>,
dry_run: bool,
overrides: &mut BTreeMap<PackageId, Config>,
breakages: &mut BTreeMap<PackageId, TaggedDependencyError>,
receipts: &ConfigReceipts,
) -> Result<(), Error> {
configure_rec(
ctx, db, id, config, timeout, dry_run, overrides, breakages, receipts,
)
.await?;
receipts.configured.set(db, true, &id).await?;
Ok(())
}
#[instrument(skip_all)]
pub fn configure_rec<'a, Db: DbHandle>(
ctx: &'a RpcContext,
db: &'a mut Db,
id: &'a PackageId,
config: Option<Config>,
timeout: &'a Option<Duration>,
dry_run: bool,
overrides: &'a mut BTreeMap<PackageId, Config>,
breakages: &'a mut BTreeMap<PackageId, TaggedDependencyError>,
receipts: &'a ConfigReceipts,
) -> BoxFuture<'a, Result<(), Error>> {
async move {
// fetch data from db
let action = receipts
.config_actions
.get(db, id)
.await?
.ok_or_else(|| not_found!(id))?;
let dependencies = receipts
.dependencies
.get(db, id)
.await?
.ok_or_else(|| not_found!(id))?;
let volumes = receipts
.volumes
.get(db, id)
.await?
.ok_or_else(|| not_found!(id))?;
let is_needs_config = !receipts
.configured
.get(db, id)
.await?
.ok_or_else(|| not_found!(id))?;
let version = receipts
.version
.get(db, id)
.await?
.ok_or_else(|| not_found!(id))?;
// get current config and current spec
let ConfigRes {
config: old_config,
spec,
} = action.get(ctx, id, &version, &volumes).await?;
// determine new config to use
let mut config = if let Some(config) = config.or_else(|| old_config.clone()) {
config
} else {
spec.gen(&mut rand::rngs::StdRng::from_entropy(), timeout)?
};
let manifest = receipts
.manifest
.get(db, id)
.await?
.ok_or_else(|| not_found!(id))?;
spec.validate(&manifest)?;
spec.matches(&config)?; // check that new config matches spec
spec.update(
ctx,
db,
&manifest,
&*overrides,
&mut config,
&receipts.config_receipts,
)
.await?; // dereference pointers in the new config
// create backreferences to pointers
let mut sys = receipts
.system_pointers
.get(db, &id)
.await?
.ok_or_else(|| not_found!(id))?;
sys.truncate(0);
let mut current_dependencies: CurrentDependencies = CurrentDependencies(
dependencies
.0
.iter()
.filter_map(|(id, info)| {
if info.requirement.required() {
Some((id.clone(), CurrentDependencyInfo::default()))
} else {
None
}
})
.collect(),
);
for ptr in spec.pointers(&config)? {
match ptr {
ValueSpecPointer::Package(pkg_ptr) => {
if let Some(current_dependency) =
current_dependencies.0.get_mut(pkg_ptr.package_id())
{
current_dependency.pointers.push(pkg_ptr);
} else {
current_dependencies.0.insert(
pkg_ptr.package_id().to_owned(),
CurrentDependencyInfo {
pointers: vec![pkg_ptr],
health_checks: BTreeSet::new(),
},
);
}
}
ValueSpecPointer::System(s) => sys.push(s),
}
}
receipts.system_pointers.set(db, sys, &id).await?;
let signal = if !dry_run {
// run config action
let res = action
.set(ctx, id, &version, &dependencies, &volumes, &config)
.await?;
// track dependencies with no pointers
for (package_id, health_checks) in res.depends_on.into_iter() {
if let Some(current_dependency) = current_dependencies.0.get_mut(&package_id) {
current_dependency.health_checks.extend(health_checks);
} else {
current_dependencies.0.insert(
package_id,
CurrentDependencyInfo {
pointers: Vec::new(),
health_checks,
},
);
}
}
// track dependency health checks
current_dependencies = current_dependencies.map(|x| {
x.into_iter()
.filter(|(dep_id, _)| {
if dep_id != id && !manifest.dependencies.0.contains_key(dep_id) {
tracing::warn!("Illegal dependency specified: {}", dep_id);
false
} else {
true
}
})
.collect()
});
res.signal
} else {
None
};
// update dependencies
let prev_current_dependencies = receipts
.current_dependencies
.get(db, &id)
.await?
.unwrap_or_default();
remove_from_current_dependents_lists(
db,
id,
&prev_current_dependencies,
&receipts.current_dependents,
)
.await?; // remove previous
add_dependent_to_current_dependents_lists(
db,
id,
&current_dependencies,
&receipts.current_dependents,
)
.await?; // add new
current_dependencies.0.remove(id);
receipts
.current_dependencies
.set(db, current_dependencies.clone(), &id)
.await?;
let errs = receipts
.dependency_errors
.get(db, &id)
.await?
.ok_or_else(|| not_found!(id))?;
tracing::warn!("Dependency Errors: {:?}", errs);
let errs = DependencyErrors::init(
ctx,
db,
&manifest,
&current_dependencies,
&receipts.dependency_receipt.try_heal,
)
configure_context: ConfigureContext,
) -> Result<BTreeMap<PackageId, TaggedDependencyError>, Error> {
let mut db = ctx.db.handle();
let version = crate::db::DatabaseModel::new()
.package_data()
.idx_model(id)
.expect(&mut db)
.await?
.installed()
.expect(&mut db)
.await?
.manifest()
.version()
.get(&mut ctx.db.handle())
.await?;
receipts.dependency_errors.set(db, errs, &id).await?;
// cache current config for dependents
overrides.insert(id.clone(), config.clone());
// handle dependents
let dependents = receipts
.current_dependents
.get(db, id)
.await?
.ok_or_else(|| not_found!(id))?;
let prev = if is_needs_config { None } else { old_config }
.map(Value::Object)
.unwrap_or_default();
let next = Value::Object(config.clone());
for (dependent, dep_info) in dependents.0.iter().filter(|(dep_id, _)| dep_id != &id) {
let dependent_container = receipts.docker_containers.get(db, &dependent).await?;
let dependent_container = &dependent_container;
// check if config passes dependent check
if let Some(cfg) = receipts
.manifest_dependencies_config
.get(db, (&dependent, &id))
.await?
{
let manifest = receipts
.manifest
.get(db, &dependent)
.await?
.ok_or_else(|| not_found!(id))?;
if let Err(error) = cfg
.check(
ctx,
dependent_container,
dependent,
&manifest.version,
&manifest.volumes,
id,
&config,
)
.await?
{
let dep_err = DependencyError::ConfigUnsatisfied { error };
break_transitive(
db,
dependent,
id,
dep_err,
breakages,
&receipts.break_transitive_receipts,
)
.await?;
}
// handle backreferences
for ptr in &dep_info.pointers {
if let PackagePointerSpec::Config(cfg_ptr) = ptr {
if cfg_ptr.select(&next) != cfg_ptr.select(&prev) {
if let Err(e) = configure_rec(
ctx, db, dependent, None, timeout, dry_run, overrides, breakages,
receipts,
)
.await
{
if e.kind == crate::ErrorKind::ConfigRulesViolation {
break_transitive(
db,
dependent,
id,
DependencyError::ConfigUnsatisfied {
error: format!("{}", e),
},
breakages,
&receipts.break_transitive_receipts,
)
.await?;
} else {
return Err(e);
}
}
}
}
}
heal_all_dependents_transitive(ctx, db, id, &receipts.dependency_receipt).await?;
}
}
if let Some(signal) = signal {
match ctx.managers.get(&(id.clone(), version.clone())).await {
None => {
// in theory this should never happen, which indicates this function should be moved behind the
// Manager interface
return Err(Error::new(
eyre!("Manager Not Found for package being configured"),
crate::ErrorKind::Incoherent,
));
}
Some(m) => {
m.signal(&signal).await?;
}
}
}
Ok(())
}
.boxed()
ctx.managers
.get(&(id.clone(), version.clone()))
.await
.ok_or_else(|| {
Error::new(
eyre!("There is no manager running for {id:?} and {version:?}"),
ErrorKind::Unknown,
)
})?
.configure(configure_context)
.await
}
macro_rules! not_found {

3
backend/src/control.rs
View File

@@ -80,8 +80,7 @@ pub async fn start(#[context] ctx: RpcContext, #[arg] id: PackageId) -> Result<(
.get(&(id, version))
.await
.ok_or_else(|| Error::new(eyre!("Manager not found"), crate::ErrorKind::InvalidRequest))?
.synchronize()
.await;
.start();
Ok(())
}

49
backend/src/dependencies.rs
View File

@@ -16,7 +16,7 @@ use tracing::instrument;
use crate::config::action::{ConfigActions, ConfigRes};
use crate::config::spec::PackagePointerSpec;
use crate::config::{not_found, Config, ConfigReceipts, ConfigSpec};
use crate::config::{not_found, Config, ConfigReceipts, ConfigSpec, ConfigureContext};
use crate::context::RpcContext;
use crate::db::model::{CurrentDependencies, CurrentDependents, InstalledPackageDataEntry};
use crate::procedure::docker::DockerContainers;
@@ -665,6 +665,8 @@ pub async fn configure_impl(
(pkg_id, dep_id): (PackageId, PackageId),
) -> Result<(), Error> {
let mut db = ctx.db.handle();
let breakages = BTreeMap::new();
let overrides = Default::default();
let receipts = DependencyConfigReceipts::new(&mut db, &pkg_id, &dep_id).await?;
let ConfigDryRes {
old_config: _,
@@ -672,19 +674,15 @@ pub async fn configure_impl(
spec: _,
} = configure_logic(ctx.clone(), &mut db, (pkg_id, dep_id.clone()), &receipts).await?;
let locks = &receipts.config;
Ok(crate::config::configure(
&ctx,
&mut db,
&dep_id,
Some(new_config),
&Some(Duration::from_secs(3).into()),
false,
&mut BTreeMap::new(),
&mut BTreeMap::new(),
locks,
)
.await?)
let configure_context = ConfigureContext {
breakages,
timeout: Some(Duration::from_secs(3).into()),
config: Some(new_config),
dry_run: false,
overrides,
};
crate::config::configure(&ctx, &dep_id, configure_context).await?;
Ok(())
}
#[derive(Clone, Debug, Serialize, Deserialize)]
@@ -1064,18 +1062,17 @@ pub async fn reconfigure_dependents_with_live_pointers(
PackagePointerSpec::TorKey(_) => false,
PackagePointerSpec::Config(_) => false,
}) {
crate::config::configure(
ctx,
&mut tx,
dependent_id,
None,
&None,
false,
&mut BTreeMap::new(),
&mut BTreeMap::new(),
receipts,
)
.await?;
let breakages = BTreeMap::new();
let overrides = Default::default();
let configure_context = ConfigureContext {
breakages,
timeout: None,
config: None,
dry_run: false,
overrides,
};
crate::config::configure(&ctx, dependent_id, configure_context).await?;
}
}
Ok(())

25
backend/src/install/mod.rs
View File

@@ -26,7 +26,7 @@ use tokio_stream::wrappers::ReadDirStream;
use tracing::instrument;
use self::cleanup::{cleanup_failed, remove_from_current_dependents_lists};
use crate::config::ConfigReceipts;
use crate::config::{ConfigReceipts, ConfigureContext};
use crate::context::{CliContext, RpcContext};
use crate::core::rpc_continuations::{RequestGuid, RpcContinuation};
use crate::db::model::{
@@ -1311,18 +1311,17 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin + Send + Sync>(
false
};
if configured && manifest.config.is_some() {
crate::config::configure(
ctx,
&mut tx,
pkg_id,
None,
&None,
false,
&mut BTreeMap::new(),
&mut BTreeMap::new(),
&receipts.config,
)
.await?;
let breakages = BTreeMap::new();
let overrides = Default::default();
let configure_context = ConfigureContext {
breakages,
timeout: None,
config: None,
dry_run: false,
overrides,
};
crate::config::configure(&ctx, pkg_id, configure_context).await?;
} else {
add_dependent_to_current_dependents_lists(
&mut tx,

22
backend/src/manager/health.rs
View File

@@ -96,7 +96,6 @@ pub async fn check<Db: DbHandle>(
ctx: &RpcContext,
db: &mut Db,
id: &PackageId,
should_commit: &AtomicBool,
) -> Result<(), Error> {
let mut tx = db.begin().await?;
let (manifest, started) = {
@@ -128,27 +127,6 @@ pub async fn check<Db: DbHandle>(
return Ok(());
};
if !should_commit.load(Ordering::SeqCst) {
return Ok(());
}
if !health_results
.iter()
.any(|(_, res)| matches!(res, HealthCheckResult::Failure { .. }))
{
tracing::debug!("All health checks succeeded for {}", id);
} else {
tracing::debug!(
"Some health checks failed for {}: {}",
id,
health_results
.iter()
.filter(|(_, res)| matches!(res, HealthCheckResult::Failure { .. }))
.map(|(id, _)| &*id)
.join(", ")
);
}
let current_dependents = {
let mut checkpoint = tx.begin().await?;
let receipts = HealthCheckStatusReceipt::new(&mut checkpoint, id).await?;

289
backend/src/manager/manager_container.rs Normal file
View File

@@ -0,0 +1,289 @@
use std::sync::Arc;
use std::time::Duration;
use patch_db::PatchDbHandle;
use tokio::sync::watch;
use tokio::sync::watch::Sender;
use super::start_stop::StartStop;
use super::{manager_seed, run_main, ManagerPersistentContainer, RunMainResult};
use crate::s9pk::manifest::Manifest;
use crate::status::MainStatus;
use crate::util::{GeneralBoxedGuard, NonDetachingJoinHandle};
use crate::Error;
pub type ManageContainerOverride = Arc<watch::Sender<Option<MainStatus>>>;
pub struct ManageContainer {
current_state: Arc<watch::Sender<StartStop>>,
desired_state: Arc<watch::Sender<StartStop>>,
service: NonDetachingJoinHandle<()>,
save_state: NonDetachingJoinHandle<()>,
override_main_status: ManageContainerOverride,
}
impl ManageContainer {
pub async fn new(
seed: Arc<manager_seed::ManagerSeed>,
persistent_container: ManagerPersistentContainer,
) -> Result<Self, Error> {
let mut db = seed.ctx.db.handle();
let current_state = Arc::new(watch::channel(StartStop::Stop).0);
let desired_state = Arc::new(
watch::channel::<StartStop>(get_status(&mut db, &seed.manifest).await?.into()).0,
);
let override_main_status: ManageContainerOverride = Arc::new(watch::channel(None).0);
let service = tokio::spawn(create_service_manager(
desired_state.clone(),
seed.clone(),
current_state.clone(),
persistent_container,
))
.into();
let save_state = tokio::spawn(save_state(
desired_state.clone(),
current_state.clone(),
override_main_status.clone(),
seed.clone(),
))
.into();
Ok(ManageContainer {
current_state,
desired_state,
service,
override_main_status,
save_state,
})
}
pub fn set_override(&self, override_status: Option<MainStatus>) -> GeneralBoxedGuard {
self.override_main_status.send(override_status);
let override_main_status = self.override_main_status.clone();
let guard = GeneralBoxedGuard::new(move || {
override_main_status.send(None);
});
guard
}
pub fn to_desired(&self, new_state: StartStop) {
self.desired_state.send(new_state);
}
pub fn current_state(&self) -> watch::Receiver<StartStop> {
self.current_state.subscribe()
}
pub fn desired_state(&self) -> watch::Receiver<StartStop> {
self.desired_state.subscribe()
}
}
async fn create_service_manager(
desired_state: Arc<Sender<StartStop>>,
seed: Arc<manager_seed::ManagerSeed>,
current_state: Arc<Sender<StartStop>>,
persistent_container: Arc<Option<super::persistent_container::PersistentContainer>>,
) {
let mut desired_state_receiver = desired_state.subscribe();
let mut running_service: Option<NonDetachingJoinHandle<()>> = None;
let seed = seed.clone();
loop {
let current: StartStop = current_state.borrow().clone();
let desired: StartStop = desired_state_receiver.borrow().clone();
match (current, desired) {
(StartStop::Start, StartStop::Start) => continue,
(StartStop::Start, StartStop::Stop) => {
if let Err(err) = seed.stop_container().await {
tracing::error!("Could not stop container");
tracing::debug!("{:?}", err)
};
running_service = None;
current_state.send(StartStop::Stop);
}
(StartStop::Stop, StartStop::Start) => starting_service(
current_state.clone(),
desired_state.clone(),
seed.clone(),
persistent_container.clone(),
&mut running_service,
),
(StartStop::Stop, StartStop::Stop) => continue,
}
if let Err(_) = desired_state_receiver.changed().await {
tracing::error!("Desired state error");
break;
}
}
}
async fn save_state(
desired_state: Arc<Sender<StartStop>>,
current_state: Arc<Sender<StartStop>>,
override_main_status: Arc<Sender<Option<MainStatus>>>,
seed: Arc<manager_seed::ManagerSeed>,
) {
let mut desired_state_receiver = desired_state.subscribe();
let mut current_state_receiver = current_state.subscribe();
let mut override_main_status_receiver = override_main_status.subscribe();
loop {
let current: StartStop = current_state_receiver.borrow().clone();
let desired: StartStop = desired_state_receiver.borrow().clone();
let override_status = override_main_status_receiver.borrow().clone();
let mut db = seed.ctx.db.handle();
let res = match (override_status, current, desired) {
(Some(status), _, _) => set_status(&mut db, &seed.manifest, &status).await,
(None, StartStop::Start, StartStop::Start) => {
set_status(
&mut db,
&seed.manifest,
&MainStatus::Running {
started: chrono::Utc::now(),
health: Default::default(),
},
)
.await
}
(None, StartStop::Start, StartStop::Stop) => {
set_status(&mut db, &seed.manifest, &MainStatus::Stopping).await
}
(None, StartStop::Stop, StartStop::Start) => {
set_status(
&mut db,
&seed.manifest,
&MainStatus::Starting { restarting: false },
)
.await
}
(None, StartStop::Stop, StartStop::Stop) => {
set_status(&mut db, &seed.manifest, &MainStatus::Stopped).await
}
};
if let Err(err) = res {
tracing::error!("Did not set status for {}", seed.container_name);
tracing::debug!("{:?}", err);
}
tokio::select! {
_ = desired_state_receiver.changed() =>{},
_ = current_state_receiver.changed() => {},
_ = override_main_status_receiver.changed() => {}
}
}
}
fn starting_service(
current_state: Arc<Sender<StartStop>>,
desired_state: Arc<Sender<StartStop>>,
seed: Arc<manager_seed::ManagerSeed>,
persistent_container: ManagerPersistentContainer,
running_service: &mut Option<NonDetachingJoinHandle<()>>,
) {
let set_running = {
let current_state = current_state.clone();
Arc::new(move || {
current_state.send(StartStop::Start);
})
};
let set_stopped = {
let current_state = current_state.clone();
move || current_state.send(StartStop::Stop)
};
let running_main_loop = async move {
while desired_state.borrow().is_start() {
let result = run_main(
seed.clone(),
persistent_container.clone(),
set_running.clone(),
)
.await;
run_main_log_result(result, seed.clone());
set_stopped();
}
};
*running_service = Some(tokio::spawn(running_main_loop).into());
}
async fn run_main_log_result(result: RunMainResult, seed: Arc<manager_seed::ManagerSeed>) {
match result {
Ok(Ok(NoOutput)) => (), // restart
Ok(Err(e)) => {
#[cfg(feature = "unstable")]
{
use crate::notifications::NotificationLevel;
use crate::status::MainStatus;
let mut db = seed.ctx.db.handle();
let started = crate::db::DatabaseModel::new()
.package_data()
.idx_model(&thread_shared.seed.manifest.id)
.and_then(|pde| pde.installed())
.map::<_, MainStatus>(|i| i.status().main())
.get(db, false)
.await;
match started.as_deref() {
Ok(Some(MainStatus::Running { .. })) => {
let res = thread_shared.seed.ctx.notification_manager
.notify(
db,
Some(thread_shared.seed.manifest.id.clone()),
NotificationLevel::Warning,
String::from("Service Crashed"),
format!("The service {} has crashed with the following exit code: {}\nDetails: {}", thread_shared.seed.manifest.id.clone(), e.0, e.1),
(),
Some(3600) // 1 hour
)
.await;
if let Err(e) = res {
tracing::error!("Failed to issue notification: {}", e);
tracing::debug!("{:?}", e);
}
}
_ => {
tracing::error!("service just started. not issuing crash notification")
}
}
}
tokio::time::sleep(Duration::from_secs(15)).await;
}
Err(e) => {
tracing::error!("failed to start service: {}", e);
tracing::debug!("{:?}", e);
}
}
}
pub(super) async fn get_status(db: &mut PatchDbHandle, manifest: &Manifest) -> Result<MainStatus, Error> {
Ok(crate::db::DatabaseModel::new()
.package_data()
.idx_model(&manifest.id)
.expect(db)
.await?
.installed()
.expect(db)
.await?
.status()
.main()
.get(db)
.await?
.clone())
}
async fn set_status(
db: &mut PatchDbHandle,
manifest: &Manifest,
main_status: &MainStatus,
) -> Result<(), Error> {
crate::db::DatabaseModel::new()
.package_data()
.idx_model(&manifest.id)
.expect(db)
.await?
.installed()
.expect(db)
.await?
.status()
.main()
.put(db, main_status)
.await?
.clone();
Ok(())
}

115
backend/src/manager/manager_map.rs Normal file
View File

@@ -0,0 +1,115 @@
use std::collections::BTreeMap;
use std::sync::Arc;
use color_eyre::eyre::eyre;
use patch_db::DbHandle;
use sqlx::{Executor, Postgres};
use tokio::sync::RwLock;
use torut::onion::TorSecretKeyV3;
use tracing::instrument;
use super::Manager;
use crate::context::RpcContext;
use crate::net::interface::InterfaceId;
use crate::s9pk::manifest::{Manifest, PackageId};
use crate::util::Version;
use crate::Error;
#[derive(Default)]
pub struct ManagerMap(RwLock<BTreeMap<(PackageId, Version), Arc<Manager>>>);
impl ManagerMap {
#[instrument(skip_all)]
pub async fn init<Db: DbHandle, Ex>(
&self,
ctx: &RpcContext,
db: &mut Db,
secrets: &mut Ex,
) -> Result<(), Error>
where
for<'a> &'a mut Ex: Executor<'a, Database = Postgres>,
{
let mut res = BTreeMap::new();
for package in crate::db::DatabaseModel::new()
.package_data()
.keys(db)
.await?
{
let man: Manifest = if let Some(manifest) = crate::db::DatabaseModel::new()
.package_data()
.idx_model(&package)
.and_then(|pkg| pkg.installed())
.map(|m| m.manifest())
.get(db)
.await?
.to_owned()
{
manifest
} else {
continue;
};
let tor_keys = man.interfaces.tor_keys(secrets, &package).await?;
res.insert(
(package, man.version.clone()),
Arc::new(Manager::new(ctx.clone(), man, tor_keys).await?),
);
}
*self.0.write().await = res;
Ok(())
}
#[instrument(skip_all)]
pub async fn add(
&self,
ctx: RpcContext,
manifest: Manifest,
tor_keys: BTreeMap<InterfaceId, TorSecretKeyV3>,
) -> Result<(), Error> {
let mut lock = self.0.write().await;
let id = (manifest.id.clone(), manifest.version.clone());
if let Some(man) = lock.remove(&id) {
man.exit().await;
}
lock.insert(id, Arc::new(Manager::new(ctx, manifest, tor_keys).await?));
Ok(())
}
#[instrument(skip_all)]
pub async fn remove(&self, id: &(PackageId, Version)) {
if let Some(man) = self.0.write().await.remove(id) {
man.exit().await;
}
}
#[instrument(skip_all)]
pub async fn empty(&self) -> Result<(), Error> {
let res =
futures::future::join_all(std::mem::take(&mut *self.0.write().await).into_iter().map(
|((id, version), man)| async move {
tracing::debug!("Manager for {}@{} shutting down", id, version);
man.exit().await;
tracing::debug!("Manager for {}@{} is shutdown", id, version);
if let Err(e) = Arc::try_unwrap(man) {
tracing::trace!(
"Manager for {}@{} still has {} other open references",
id,
version,
Arc::strong_count(&e) - 1
);
}
Ok::<_, Error>(())
},
))
.await;
res.into_iter().fold(Ok(()), |res, x| match (res, x) {
(Ok(()), x) => x,
(Err(e), Ok(())) => Err(e),
(Err(e1), Err(e2)) => Err(Error::new(eyre!("{}, {}", e1.source, e2.source), e1.kind)),
})
}
#[instrument(skip_all)]
pub async fn get(&self, id: &(PackageId, Version)) -> Option<Arc<Manager>> {
self.0.read().await.get(id).cloned()
}
}

50
backend/src/manager/manager_seed.rs Normal file
View File

@@ -0,0 +1,50 @@
use std::collections::BTreeMap;
use bollard::container::StopContainerOptions;
use torut::onion::TorSecretKeyV3;
use super::sigterm_timeout;
use crate::context::RpcContext;
use crate::net::interface::InterfaceId;
use crate::s9pk::manifest::Manifest;
use crate::Error;
pub struct ManagerSeed {
pub ctx: RpcContext,
pub manifest: Manifest,
pub container_name: String,
pub tor_keys: BTreeMap<InterfaceId, TorSecretKeyV3>,
}
impl ManagerSeed {
pub async fn stop_container(&self) -> Result<(), Error> {
match self
.ctx
.docker
.stop_container(
&self.container_name,
Some(StopContainerOptions {
t: sigterm_timeout(&self.manifest)
.map(|d| d.as_secs())
.unwrap_or(30) as i64,
}),
)
.await
{
Err(bollard::errors::Error::DockerResponseServerError {
status_code: 404, // NOT FOUND
..
})
| Err(bollard::errors::Error::DockerResponseServerError {
status_code: 409, // CONFLICT
..
})
| Err(bollard::errors::Error::DockerResponseServerError {
status_code: 304, // NOT MODIFIED
..
}) => (), // Already stopped
a => a?,
}
Ok(())
}
}

1461
backend/src/manager/mod.rs
View File

File diff suppressed because it is too large Load Diff

101
backend/src/manager/persistent_container.rs Normal file
View File

@@ -0,0 +1,101 @@
use std::sync::Arc;
use std::time::Duration;
use color_eyre::eyre::eyre;
use embassy_container_init::ProcessGroupId;
use helpers::UnixRpcClient;
use tokio::sync::oneshot;
use tokio::sync::watch::{self, Receiver};
use tracing::instrument;
use super::manager_seed::ManagerSeed;
use super::{
add_network_for_main, generate_certificate, get_long_running_ip, long_running_docker,
main_interfaces, remove_network_for_main, GetRunningIp,
};
use crate::procedure::docker::DockerContainer;
use crate::util::NonDetachingJoinHandle;
use crate::Error;
pub struct PersistentContainer {
_running_docker: NonDetachingJoinHandle<()>,
pub rpc_client: Receiver<Arc<UnixRpcClient>>,
}
impl PersistentContainer {
#[instrument(skip_all)]
pub async fn init(seed: &Arc<ManagerSeed>) -> Result<Option<Self>, Error> {
Ok(if let Some(containers) = &seed.manifest.containers {
let (running_docker, rpc_client) =
spawn_persistent_container(seed.clone(), containers.main.clone()).await?;
Some(Self {
_running_docker: running_docker,
rpc_client,
})
} else {
None
})
}
pub fn rpc_client(&self) -> Arc<UnixRpcClient> {
self.rpc_client.borrow().clone()
}
}
pub async fn spawn_persistent_container(
seed: Arc<ManagerSeed>,
container: DockerContainer,
) -> Result<(NonDetachingJoinHandle<()>, Receiver<Arc<UnixRpcClient>>), Error> {
let (send_inserter, inserter) = oneshot::channel();
Ok((
tokio::task::spawn(async move {
let mut inserter_send: Option<watch::Sender<Arc<UnixRpcClient>>> = None;
let mut send_inserter: Option<oneshot::Sender<Receiver<Arc<UnixRpcClient>>>> = Some(send_inserter);
loop {
if let Err(e) = async {
let interfaces = main_interfaces(&*seed)?;
let generated_certificate = generate_certificate(&*seed, &interfaces).await?;
let (mut runtime, inserter) =
long_running_docker(&seed, &container).await?;
let ip = match get_long_running_ip(&*seed, &mut runtime).await {
GetRunningIp::Ip(x) => x,
GetRunningIp::Error(e) => return Err(e),
GetRunningIp::EarlyExit(e) => {
tracing::error!("Early Exit");
tracing::debug!("{:?}", e);
return Ok(());
}
};
add_network_for_main(&*seed, ip, interfaces, generated_certificate).await?;
if let Some(inserter_send) = inserter_send.as_mut() {
let _ = inserter_send.send(Arc::new(inserter));
} else {
let (s, r) = watch::channel(Arc::new(inserter));
inserter_send = Some(s);
if let Some(send_inserter) = send_inserter.take() {
let _ = send_inserter.send(r);
}
}
let res = tokio::select! {
a = runtime.running_output => a.map_err(|_| Error::new(eyre!("Manager runtime panicked!"), crate::ErrorKind::Docker)).map(|_| ()),
};
remove_network_for_main(&*seed, ip).await?;
res
}.await {
tracing::error!("Error in persistent container: {}", e);
tracing::debug!("{:?}", e);
} else {
break;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
})
.into(),
inserter.await.map_err(|_| Error::new(eyre!("Container handle dropped before inserter sent"), crate::ErrorKind::Unknown))?,
))
}

29
backend/src/manager/start_stop.rs Normal file
View File

@@ -0,0 +1,29 @@
use crate::status::MainStatus;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum StartStop {
Start,
Stop,
}
impl StartStop {
pub(crate) fn is_start(&self) -> bool {
matches!(self, StartStop::Start)
}
pub(crate) fn is_stop(&self) -> bool {
matches!(self, StartStop::Stop)
}
}
impl From<MainStatus> for StartStop {
fn from(value: MainStatus) -> Self {
match value {
MainStatus::Stopped => StartStop::Stop,
MainStatus::Restarting => StartStop::Start,
MainStatus::Stopping => StartStop::Stop,
MainStatus::Starting { restarting } => StartStop::Start,
MainStatus::Running { started, health } => StartStop::Start,
MainStatus::BackingUp { started, health } if started.is_some() => StartStop::Start,
MainStatus::BackingUp { started, health } => StartStop::Stop,
}
}
}

30
backend/src/manager/transition_state.rs Normal file
View File

@@ -0,0 +1,30 @@
use tokio::task::JoinHandle;
pub(crate) enum TransitionState {
// Starting(JoinHandle<()>),
// Stopping(JoinHandle<()>)
BackingUp(JoinHandle<()>),
Restarting(JoinHandle<()>),
Configuring(JoinHandle<()>),
None,
}
impl TransitionState {
pub(crate) fn join_handle(&self) -> Option<&JoinHandle<()>> {
Some(match self {
TransitionState::BackingUp(a) => a,
TransitionState::Restarting(a) => a,
TransitionState::Configuring(a) => a,
TransitionState::None => return None,
})
}
pub(crate) fn abort(&self) {
self.join_handle().map(|transition| transition.abort());
}
}
impl Default for TransitionState {
fn default() -> Self {
TransitionState::None
}
}

4
backend/src/procedure/mod.rs
View File

@@ -96,9 +96,9 @@ impl PackageProcedure {
}
Some(man) => (
if matches!(name, ProcedureName::Main) {
man.new_main_gid()
man.gid.new_main_gid()
} else {
man.new_gid()
man.gid.new_gid()
},
man.rpc_client(),
),

12
backend/src/status/mod.rs
View File

@@ -73,6 +73,18 @@ impl MainStatus {
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::Restarting => {
(None, Default::default())
}
MainStatus::BackingUp { .. } => return self.clone(),
};
MainStatus::BackingUp { started, health }
}
}
impl MainStatusModel {
pub fn started(self) -> Model<Option<DateTime<Utc>>> {

23
backend/src/util/mod.rs
View File

@@ -278,6 +278,29 @@ impl<F: FnOnce() -> T, T> Drop for GeneralGuard<F, T> {
}
}
pub struct GeneralBoxedGuard(Option<Box<dyn FnOnce() -> ()>>);
impl GeneralBoxedGuard {
pub fn new(f: impl FnOnce() -> ()) -> Self {
GeneralBoxedGuard(Some(f.boxed()))
}
pub fn drop(mut self) -> () {
self.0.take().unwrap()()
}
pub fn drop_without_action(mut self) {
self.0 = None;
}
}
impl Drop for GeneralBoxedGuard {
fn drop(&mut self) {
if let Some(destroy) = self.0.take() {
destroy();
}
}
}
pub struct FileLock(OwnedMutexGuard<()>, Option<FdLock<File>>);
impl Drop for FileLock {
fn drop(&mut self) {

2
patch-db

Submodule patch-db updated: 85959db110...fcfa917710