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

switch to managers: wip 2

Browse Source
This commit is contained in:
Aiden McClelland
2021-07-19 11:38:04 -06:00
parent 7eaf6dabcc
commit d9fa354e85
10 changed files with 315 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
appmgr/src/config/mod.rs
View File

@@ -161,7 +161,7 @@ pub async fn get(
.clone() .clone()
.manifest() .manifest()
.config() .config()
.get(&mut db) .get(&mut db, true)
.await? .await?
.to_owned() .to_owned()
.ok_or_else(|| { .ok_or_else(|| {
@@ -170,8 +170,13 @@ pub async fn get(
crate::ErrorKind::NotFound, crate::ErrorKind::NotFound,
) )
})?; })?;
let version = pkg_model.clone().manifest().version().get(&mut db).await?; let version = pkg_model
let volumes = pkg_model.manifest().volumes().get(&mut db).await?; .clone()
.manifest()
.version()
.get(&mut db, true)
.await?;
let volumes = pkg_model.manifest().volumes().get(&mut db, true).await?;
action.get(ctx.extension(), &*version, &*volumes).await action.get(ctx.extension(), &*version, &*volumes).await
} }
@@ -295,15 +300,20 @@ pub fn configure<'a, Db: DbHandle>(
.clone() .clone()
.manifest() .manifest()
.config() .config()
.get(db) .get(db, true)
.await? .await?
.to_owned() .to_owned()
.ok_or_else(|| { .ok_or_else(|| {
Error::new(anyhow!("{} has no config", id), crate::ErrorKind::NotFound) Error::new(anyhow!("{} has no config", id), crate::ErrorKind::NotFound)
})?; })?;
let version = pkg_model.clone().manifest().version().get(db).await?; let version = pkg_model.clone().manifest().version().get(db, true).await?;
let dependencies = pkg_model.clone().manifest().dependencies().get(db).await?; let dependencies = pkg_model
let volumes = pkg_model.clone().manifest().volumes().get(db).await?; .clone()
.manifest()
.dependencies()
.get(db, true)
.await?;
let volumes = pkg_model.clone().manifest().volumes().get(db, true).await?;
// get current config and current spec // get current config and current spec
let ConfigRes { let ConfigRes {
@@ -406,7 +416,7 @@ pub fn configure<'a, Db: DbHandle>(
overrides.insert(id.clone(), config.clone()); overrides.insert(id.clone(), config.clone());
// handle dependents // handle dependents
let dependents = pkg_model.clone().current_dependents().get(db).await?; let dependents = pkg_model.clone().current_dependents().get(db, true).await?;
let prev = old_config.map(Value::Object).unwrap_or_default(); let prev = old_config.map(Value::Object).unwrap_or_default();
let next = Value::Object(config.clone()); let next = Value::Object(config.clone());
for (dependent, dep_info) in &*dependents { for (dependent, dep_info) in &*dependents {
@@ -447,12 +457,12 @@ pub fn configure<'a, Db: DbHandle>(
.idx_model(dependency) .idx_model(dependency)
.expect(db) .expect(db)
.await? .await?
.get(db) .get(db, true)
.await? .await?
.critical .critical
{ {
status.main.stop(); status.main.stop();
let dependents = model.current_dependents().get(db).await?; let dependents = model.current_dependents().get(db, true).await?;
for (dependent, _) in &*dependents { for (dependent, _) in &*dependents {
let dependent_model = crate::db::DatabaseModel::new() let dependent_model = crate::db::DatabaseModel::new()
.package_data() .package_data()
@@ -496,10 +506,15 @@ pub fn configure<'a, Db: DbHandle>(
.expect(db) .expect(db)
.await? .await?
.config() .config()
.get(db) .get(db, true)
.await? .await?
{ {
let version = dependent_model.clone().manifest().version().get(db).await?; let version = dependent_model
.clone()
.manifest()
.version()
.get(db, true)
.await?;
if let Err(error) = cfg.check(dependent, &*version, &config).await? { if let Err(error) = cfg.check(dependent, &*version, &config).await? {
let dep_err = DependencyError::ConfigUnsatisfied { error }; let dep_err = DependencyError::ConfigUnsatisfied { error };
handle_broken_dependents( handle_broken_dependents(

10
appmgr/src/config/spec.rs
View File

@@ -1564,7 +1564,7 @@ impl PackagePointerSpec {
.and_then(|pde| pde.installed()) .and_then(|pde| pde.installed())
.and_then(|installed| installed.interface_addresses().idx_model(interface)) .and_then(|installed| installed.interface_addresses().idx_model(interface))
.and_then(|addresses| addresses.tor_address()) .and_then(|addresses| addresses.tor_address())
.get(db) .get(db, true)
.await .await
.map_err(|e| ConfigurationError::SystemError(Error::from(e)))?; .map_err(|e| ConfigurationError::SystemError(Error::from(e)))?;
Ok(addr.to_owned().map(Value::String).unwrap_or(Value::Null)) Ok(addr.to_owned().map(Value::String).unwrap_or(Value::Null))
@@ -1576,7 +1576,7 @@ impl PackagePointerSpec {
.and_then(|pde| pde.installed()) .and_then(|pde| pde.installed())
.and_then(|installed| installed.interface_addresses().idx_model(interface)) .and_then(|installed| installed.interface_addresses().idx_model(interface))
.and_then(|addresses| addresses.lan_address()) .and_then(|addresses| addresses.lan_address())
.get(db) .get(db, true)
.await .await
.map_err(|e| ConfigurationError::SystemError(Error::from(e)))?; .map_err(|e| ConfigurationError::SystemError(Error::from(e)))?;
Ok(addr.to_owned().map(Value::String).unwrap_or(Value::Null)) Ok(addr.to_owned().map(Value::String).unwrap_or(Value::Null))
@@ -1594,18 +1594,18 @@ impl PackagePointerSpec {
let version = manifest_model let version = manifest_model
.clone() .clone()
.map(|manifest| manifest.version()) .map(|manifest| manifest.version())
.get(db) .get(db, true)
.await .await
.map_err(|e| ConfigurationError::SystemError(Error::from(e)))?; .map_err(|e| ConfigurationError::SystemError(Error::from(e)))?;
let cfg_actions = manifest_model let cfg_actions = manifest_model
.clone() .clone()
.and_then(|manifest| manifest.config()) .and_then(|manifest| manifest.config())
.get(db) .get(db, true)
.await .await
.map_err(|e| ConfigurationError::SystemError(Error::from(e)))?; .map_err(|e| ConfigurationError::SystemError(Error::from(e)))?;
let volumes = manifest_model let volumes = manifest_model
.map(|manifest| manifest.volumes()) .map(|manifest| manifest.volumes())
.get(db) .get(db, true)
.await .await
.map_err(|e| ConfigurationError::SystemError(Error::from(e)))?; .map_err(|e| ConfigurationError::SystemError(Error::from(e)))?;
if let (Some(version), Some(cfg_actions), Some(volumes)) = if let (Some(version), Some(cfg_actions), Some(volumes)) =

2
appmgr/src/context/rpc.rs
View File

@@ -93,7 +93,7 @@ impl RpcContext {
Ok(crate::db::DatabaseModel::new() Ok(crate::db::DatabaseModel::new()
.server_info() .server_info()
.registry() .registry()
.get(&mut self.db.handle()) .get(&mut self.db.handle(), false)
.await? .await?
.to_owned()) .to_owned())
} }

4
appmgr/src/dependencies.rs
View File

@@ -144,8 +144,8 @@ impl DepInfo {
.await? .await?
{ {
( (
dep_model.clone().manifest().get(db).await?, dep_model.clone().manifest().get(db, true).await?,
dep_model.get(db).await?, dep_model.get(db, true).await?,
) )
} else { } else {
return Ok(Err(DependencyError::NotInstalled)); return Ok(Err(DependencyError::NotInstalled));

47
appmgr/src/install/mod.rs
View File

@@ -74,8 +74,6 @@ pub async fn download_install_s9pk(
let pkg_id = &temp_manifest.id; let pkg_id = &temp_manifest.id;
let version = &temp_manifest.version; let version = &temp_manifest.version;
let mut db = ctx.db.handle();
let pkg_cache_dir = Path::new(PKG_CACHE).join(pkg_id).join(version.as_str()); let pkg_cache_dir = Path::new(PKG_CACHE).join(pkg_id).join(version.as_str());
tokio::fs::create_dir_all(&pkg_cache_dir).await?; tokio::fs::create_dir_all(&pkg_cache_dir).await?;
let pkg_cache = AsRef::<Path>::as_ref(pkg_id).with_extension("s9pk"); let pkg_cache = AsRef::<Path>::as_ref(pkg_id).with_extension("s9pk");
@@ -87,7 +85,8 @@ pub async fn download_install_s9pk(
let res = (|| async { let res = (|| async {
let progress = InstallProgress::new(s9pk.content_length()); let progress = InstallProgress::new(s9pk.content_length());
let static_files = StaticFiles::remote(pkg_id, version, temp_manifest.assets.icon_type()); let static_files = StaticFiles::remote(pkg_id, version, temp_manifest.assets.icon_type());
let mut pde = pkg_data_entry.get_mut(&mut db).await?; let mut db_handle = ctx.db.handle();
let mut pde = pkg_data_entry.get_mut(&mut db_handle).await?;
match pde.take() { match pde.take() {
Some(PackageDataEntry::Installed { Some(PackageDataEntry::Installed {
installed, installed,
@@ -115,7 +114,8 @@ pub async fn download_install_s9pk(
)) ))
} }
} }
pde.save(&mut db).await?; pde.save(&mut db_handle).await?;
drop(db_handle);
let progress_model = pkg_data_entry.and_then(|pde| pde.install_progress()); let progress_model = pkg_data_entry.and_then(|pde| pde.install_progress());
async fn check_cache( async fn check_cache(
@@ -153,9 +153,7 @@ pub async fn download_install_s9pk(
pkg_id, pkg_id,
version, version,
progress progress
.track_read_during(model, &ctx.db, db, || { .track_read_during(model, &ctx.db, || S9pkReader::from_reader(progress_reader))
S9pkReader::from_reader(progress_reader)
})
.await, .await,
)?; )?;
if hash.as_bytes() == rdr.hash_str().as_bytes() { if hash.as_bytes() == rdr.hash_str().as_bytes() {
@@ -172,7 +170,7 @@ pub async fn download_install_s9pk(
&progress, &progress,
progress_model.clone(), progress_model.clone(),
&ctx, &ctx,
&mut db, &mut ctx.db.handle(),
) )
.await; .await;
@@ -188,7 +186,7 @@ pub async fn download_install_s9pk(
.await?; .await?;
progress progress
.track_download_during(progress_model.clone(), &ctx.db, &mut db, || async { .track_download_during(progress_model.clone(), &ctx.db, || async {
let mut progress_writer = let mut progress_writer =
InstallProgressTracker::new(&mut dst, progress.clone()); InstallProgressTracker::new(&mut dst, progress.clone());
tokio::io::copy( tokio::io::copy(
@@ -216,25 +214,26 @@ pub async fn download_install_s9pk(
let progress_reader = InstallProgressTracker::new(dst, progress.clone()); let progress_reader = InstallProgressTracker::new(dst, progress.clone());
let rdr = progress let rdr = progress
.track_read_during(progress_model.clone(), &ctx.db, &mut db, || { .track_read_during(progress_model.clone(), &ctx.db, || {
S9pkReader::from_reader(progress_reader) S9pkReader::from_reader(progress_reader)
}) })
.await?; .await?;
rdr rdr
}; };
install_s9pk(&ctx, &mut db, pkg_id, version, &mut s9pk_reader, progress).await?; install_s9pk(&ctx, pkg_id, version, &mut s9pk_reader, progress).await?;
Ok(()) Ok(())
})() })()
.await; .await;
if let Err(e) = res { if let Err(e) = res {
let mut handle = ctx.db.handle();
let mut broken = crate::db::DatabaseModel::new() let mut broken = crate::db::DatabaseModel::new()
.broken_packages() .broken_packages()
.get_mut(&mut db) .get_mut(&mut handle)
.await?; .await?;
broken.push(pkg_id.clone()); broken.push(pkg_id.clone());
broken.save(&mut db).await?; broken.save(&mut handle).await?;
Err(e) Err(e)
} else { } else {
Ok(()) Ok(())
@@ -243,7 +242,6 @@ pub async fn download_install_s9pk(
pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>( pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
ctx: &RpcContext, ctx: &RpcContext,
mut db: &mut PatchDbHandle,
pkg_id: &PackageId, pkg_id: &PackageId,
version: &Version, version: &Version,
rdr: &mut S9pkReader<InstallProgressTracker<R>>, rdr: &mut S9pkReader<InstallProgressTracker<R>>,
@@ -254,7 +252,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
let model = crate::db::DatabaseModel::new() let model = crate::db::DatabaseModel::new()
.package_data() .package_data()
.idx_model(pkg_id) .idx_model(pkg_id)
.check(db) .check(&mut ctx.db.handle())
.await? .await?
.ok_or_else(|| { .ok_or_else(|| {
Error::new( Error::new(
@@ -266,7 +264,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
log::info!("Install {}@{}: Unpacking Manifest", pkg_id, version); log::info!("Install {}@{}: Unpacking Manifest", pkg_id, version);
let manifest = progress let manifest = progress
.track_read_during(progress_model.clone(), &ctx.db, db, || rdr.manifest()) .track_read_during(progress_model.clone(), &ctx.db, || rdr.manifest())
.await?; .await?;
log::info!("Install {}@{}: Unpacked Manifest", pkg_id, version); log::info!("Install {}@{}: Unpacked Manifest", pkg_id, version);
@@ -277,7 +275,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
log::info!("Install {}@{}: Unpacking LICENSE.md", pkg_id, version); log::info!("Install {}@{}: Unpacking LICENSE.md", pkg_id, version);
progress progress
.track_read_during(progress_model.clone(), &ctx.db, db, || async { .track_read_during(progress_model.clone(), &ctx.db, || async {
let license_path = public_dir_path.join("LICENSE.md"); let license_path = public_dir_path.join("LICENSE.md");
let mut dst = File::create(&license_path).await?; let mut dst = File::create(&license_path).await?;
tokio::io::copy(&mut rdr.license().await?, &mut dst).await?; tokio::io::copy(&mut rdr.license().await?, &mut dst).await?;
@@ -289,7 +287,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
log::info!("Install {}@{}: Unpacking INSTRUCTIONS.md", pkg_id, version); log::info!("Install {}@{}: Unpacking INSTRUCTIONS.md", pkg_id, version);
progress progress
.track_read_during(progress_model.clone(), &ctx.db, db, || async { .track_read_during(progress_model.clone(), &ctx.db, || async {
let instructions_path = public_dir_path.join("INSTRUCTIONS.md"); let instructions_path = public_dir_path.join("INSTRUCTIONS.md");
let mut dst = File::create(&instructions_path).await?; let mut dst = File::create(&instructions_path).await?;
tokio::io::copy(&mut rdr.instructions().await?, &mut dst).await?; tokio::io::copy(&mut rdr.instructions().await?, &mut dst).await?;
@@ -307,7 +305,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
icon_path.display() icon_path.display()
); );
progress progress
.track_read_during(progress_model.clone(), &ctx.db, db, || async { .track_read_during(progress_model.clone(), &ctx.db, || async {
let icon_path = public_dir_path.join(&icon_path); let icon_path = public_dir_path.join(&icon_path);
let mut dst = File::create(&icon_path).await?; let mut dst = File::create(&icon_path).await?;
tokio::io::copy(&mut rdr.icon().await?, &mut dst).await?; tokio::io::copy(&mut rdr.icon().await?, &mut dst).await?;
@@ -324,7 +322,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
log::info!("Install {}@{}: Unpacking Docker Images", pkg_id, version); log::info!("Install {}@{}: Unpacking Docker Images", pkg_id, version);
progress progress
.track_read_during(progress_model.clone(), &ctx.db, db, || async { .track_read_during(progress_model.clone(), &ctx.db, || async {
let mut load = tokio::process::Command::new("docker") let mut load = tokio::process::Command::new("docker")
.arg("load") .arg("load")
.stdin(Stdio::piped()) .stdin(Stdio::piped())
@@ -359,9 +357,10 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
progress.read_complete.store(true, Ordering::SeqCst); progress.read_complete.store(true, Ordering::SeqCst);
progress_model.put(&mut db, &progress).await?; progress_model.put(&mut ctx.db.handle(), &progress).await?;
let mut tx = db.begin().await?; let mut handle = ctx.db.handle();
let mut tx = handle.begin().await?;
let mut sql_tx = ctx.secret_store.begin().await?; let mut sql_tx = ctx.secret_store.begin().await?;
log::info!("Install {}@{}: Creating manager", pkg_id, version); log::info!("Install {}@{}: Creating manager", pkg_id, version);
@@ -398,7 +397,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
let mut deps = IndexMap::new(); let mut deps = IndexMap::new();
for package in crate::db::DatabaseModel::new() for package in crate::db::DatabaseModel::new()
.package_data() .package_data()
.keys(&mut tx) .keys(&mut tx, true)
.await? .await?
{ {
if let Some(dep) = crate::db::DatabaseModel::new() if let Some(dep) = crate::db::DatabaseModel::new()
@@ -408,7 +407,7 @@ pub async fn install_s9pk<R: AsyncRead + AsyncSeek + Unpin>(
.await? .await?
.installed() .installed()
.and_then(|i| i.current_dependencies().idx_model(pkg_id)) .and_then(|i| i.current_dependencies().idx_model(pkg_id))
.get(&mut tx) .get(&mut tx, true)
.await? .await?
.to_owned() .to_owned()
{ {

36
appmgr/src/install/progress.rs
View File

@@ -43,18 +43,13 @@ impl InstallProgress {
self: Arc<Self>, self: Arc<Self>,
model: OptionModel<InstallProgress>, model: OptionModel<InstallProgress>,
mut db: Db, mut db: Db,
) -> (Db, Result<(), Error>) { ) -> Result<(), Error> {
while !self.download_complete.load(Ordering::SeqCst) { while !self.download_complete.load(Ordering::SeqCst) {
if let Err(e) = model.put(&mut db, &self).await { model.put(&mut db, &self).await?;
return (db, Err(e.into()));
}
tokio::time::sleep(Duration::from_secs(1)).await; tokio::time::sleep(Duration::from_secs(1)).await;
} }
if let Err(e) = model.put(&mut db, &self).await { model.put(&mut db, &self).await?;
(db, Err(e.into())) Ok(())
} else {
(db, Ok(()))
}
} }
pub async fn track_download_during< pub async fn track_download_during<
F: FnOnce() -> Fut, F: FnOnce() -> Fut,
@@ -64,16 +59,13 @@ impl InstallProgress {
self: &Arc<Self>, self: &Arc<Self>,
model: OptionModel<InstallProgress>, model: OptionModel<InstallProgress>,
db: &PatchDb, db: &PatchDb,
handle: &mut PatchDbHandle,
f: F, f: F,
) -> Result<T, Error> { ) -> Result<T, Error> {
let local_db = std::mem::replace(handle, db.handle()); let local_db = db.handle();
let tracker = tokio::spawn(self.clone().track_download(model.clone(), local_db)); let tracker = tokio::spawn(self.clone().track_download(model.clone(), local_db));
let res = f().await; let res = f().await;
self.download_complete.store(true, Ordering::SeqCst); self.download_complete.store(true, Ordering::SeqCst);
let (local_db, tracker_res) = tracker.await.unwrap(); tracker.await.unwrap()?;
let _ = std::mem::replace(handle, local_db);
tracker_res?;
res res
} }
pub async fn track_read<Db: DbHandle>( pub async fn track_read<Db: DbHandle>(
@@ -81,14 +73,13 @@ impl InstallProgress {
model: OptionModel<InstallProgress>, model: OptionModel<InstallProgress>,
mut db: Db, mut db: Db,
complete: Arc<AtomicBool>, complete: Arc<AtomicBool>,
) -> (Db, Result<(), Error>) { ) -> Result<(), Error> {
while !complete.load(Ordering::SeqCst) { while !complete.load(Ordering::SeqCst) {
if let Err(e) = model.put(&mut db, &self).await { model.put(&mut db, &self).await?;
return (db, Err(e.into()));
}
tokio::time::sleep(Duration::from_secs(1)).await; tokio::time::sleep(Duration::from_secs(1)).await;
} }
(db, Ok(())) model.put(&mut db, &self).await?;
Ok(())
} }
pub async fn track_read_during< pub async fn track_read_during<
F: FnOnce() -> Fut, F: FnOnce() -> Fut,
@@ -98,10 +89,9 @@ impl InstallProgress {
self: &Arc<Self>, self: &Arc<Self>,
model: OptionModel<InstallProgress>, model: OptionModel<InstallProgress>,
db: &PatchDb, db: &PatchDb,
handle: &mut PatchDbHandle,
f: F, f: F,
) -> Result<T, Error> { ) -> Result<T, Error> {
let local_db = std::mem::replace(handle, db.handle()); let local_db = db.handle();
let complete = Arc::new(AtomicBool::new(false)); let complete = Arc::new(AtomicBool::new(false));
let tracker = tokio::spawn(self.clone().track_read( let tracker = tokio::spawn(self.clone().track_read(
model.clone(), model.clone(),
@@ -110,9 +100,7 @@ impl InstallProgress {
)); ));
let res = f().await; let res = f().await;
complete.store(true, Ordering::SeqCst); complete.store(true, Ordering::SeqCst);
let (local_db, tracker_res) = tracker.await.unwrap(); tracker.await.unwrap()?;
let _ = std::mem::replace(handle, local_db);
tracker_res?;
res res
} }
} }

227
appmgr/src/manager/mod.rs
View File

@@ -1,12 +1,12 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::future::Future; use std::future::Future;
use std::net::Ipv4Addr; use std::net::Ipv4Addr;
use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicUsize;
use std::sync::Arc; use std::sync::Arc;
use std::task::Poll; use std::task::Poll;
use std::time::Duration;
use anyhow::anyhow; use anyhow::anyhow;
use bollard::container::StopContainerOptions;
use bollard::Docker; use bollard::Docker;
use patch_db::{DbHandle, PatchDbHandle}; use patch_db::{DbHandle, PatchDbHandle};
use sqlx::{Executor, Sqlite}; use sqlx::{Executor, Sqlite};
@@ -14,17 +14,13 @@ use tokio::sync::watch::error::RecvError;
use tokio::sync::watch::{channel, Receiver, Sender}; use tokio::sync::watch::{channel, Receiver, Sender};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
use tokio_stream::wrappers::WatchStream;
use torut::onion::TorSecretKeyV3; use torut::onion::TorSecretKeyV3;
use crate::action::docker::DockerAction; use crate::action::docker::DockerAction;
use crate::context::RpcContext;
use crate::net::interface::InterfaceId; use crate::net::interface::InterfaceId;
use crate::net::mdns::MdnsController;
use crate::net::tor::TorController;
use crate::net::NetController; use crate::net::NetController;
use crate::s9pk::manifest::{Manifest, PackageId}; use crate::s9pk::manifest::{Manifest, PackageId};
use crate::util::Version; use crate::util::{Container, Version};
use crate::{Error, ResultExt}; use crate::{Error, ResultExt};
pub struct ManagerMap(RwLock<HashMap<(PackageId, Version), Arc<Manager>>>); pub struct ManagerMap(RwLock<HashMap<(PackageId, Version), Arc<Manager>>>);
@@ -49,8 +45,10 @@ impl ManagerMap {
) -> Result<(), Error> { ) -> Result<(), Error> {
let mut lock = self.0.write().await; let mut lock = self.0.write().await;
let id = (manifest.id.clone(), manifest.version.clone()); let id = (manifest.id.clone(), manifest.version.clone());
if lock.contains_key(&id) { if let Some(man) = lock.get(&id) {
return Ok(()); if !man.thread.is_empty().await {
return Ok(());
}
} }
lock.insert( lock.insert(
id, id,
@@ -60,7 +58,11 @@ impl ManagerMap {
} }
pub async fn remove(&self, id: &(PackageId, Version)) { pub async fn remove(&self, id: &(PackageId, Version)) {
self.0.write().await.remove(id); if let Some(man) = self.0.write().await.remove(id) {
if let Err(e) = man.exit().await {
log::error!("Error shutting down manager: {}", e);
}
}
} }
pub async fn get(&self, id: &(PackageId, Version)) -> Option<Arc<Manager>> { pub async fn get(&self, id: &(PackageId, Version)) -> Option<Arc<Manager>> {
@@ -69,8 +71,24 @@ impl ManagerMap {
} }
pub struct Manager { pub struct Manager {
shared: Arc<ManagerSharedState>,
thread: Container<JoinHandle<()>>,
}
pub enum Status {
Running = 0,
Stopped = 1,
Paused = 2,
}
struct ManagerSharedState {
status: AtomicUsize,
on_stop: Sender<OnStop>, on_stop: Sender<OnStop>,
thread: JoinHandle<()>, docker: Docker,
net_ctl: Arc<NetController>,
manifest: Manifest,
container_name: String,
tor_keys: HashMap<InterfaceId, TorSecretKeyV3>,
} }
#[derive(Clone, Copy)] #[derive(Clone, Copy)]
@@ -80,21 +98,17 @@ pub enum OnStop {
Exit, Exit,
} }
async fn run_main( async fn run_main(state: &Arc<ManagerSharedState>) -> Result<Result<(), (i32, String)>, Error> {
docker: &Docker, let rt_state = state.clone();
net_ctl: &NetController,
manifest: &Manifest,
tor_keys: &HashMap<InterfaceId, TorSecretKeyV3>,
) -> Result<Result<(), (i32, String)>, Error> {
let rt_manifest = manifest.clone();
let mut runtime = tokio::spawn(async move { let mut runtime = tokio::spawn(async move {
rt_manifest rt_state
.manifest
.main .main
.execute::<(), ()>( .execute::<(), ()>(
&rt_manifest.id, &rt_state.manifest.id,
&rt_manifest.version, &rt_state.manifest.version,
None, None,
&rt_manifest.volumes, &rt_state.manifest.volumes,
None, None,
false, false,
) )
@@ -102,8 +116,9 @@ async fn run_main(
}); });
let mut ip = None::<Ipv4Addr>; let mut ip = None::<Ipv4Addr>;
loop { loop {
match docker match state
.inspect_container(&DockerAction::container_name(&manifest.id, None), None) .docker
.inspect_container(&state.container_name, None)
.await .await
{ {
Ok(res) => { Ok(res) => {
@@ -140,11 +155,13 @@ async fn run_main(
) )
})?; })?;
net_ctl state
.net_ctl
.add( .add(
&manifest.id, &state.manifest.id,
ip, ip,
manifest state
.manifest
.interfaces .interfaces
.0 .0
.iter() .iter()
@@ -152,7 +169,8 @@ async fn run_main(
Ok(( Ok((
id.clone(), id.clone(),
info, info,
tor_keys state
.tor_keys
.get(id) .get(id)
.ok_or_else(|| { .ok_or_else(|| {
Error::new( Error::new(
@@ -175,7 +193,10 @@ async fn run_main(
) )
}) })
.and_then(|a| a); .and_then(|a| a);
net_ctl.remove(&manifest.id, manifest.interfaces.0.keys().cloned()); state.net_ctl.remove(
&state.manifest.id,
state.manifest.interfaces.0.keys().cloned(),
);
res res
} }
@@ -187,6 +208,16 @@ impl Manager {
tor_keys: HashMap<InterfaceId, TorSecretKeyV3>, tor_keys: HashMap<InterfaceId, TorSecretKeyV3>,
) -> Result<Self, Error> { ) -> Result<Self, Error> {
let (on_stop, mut recv) = channel(OnStop::Sleep); let (on_stop, mut recv) = channel(OnStop::Sleep);
let shared = Arc::new(ManagerSharedState {
status: AtomicUsize::new(Status::Stopped as usize),
on_stop,
docker,
net_ctl,
container_name: DockerAction::container_name(&manifest.id, None),
manifest,
tor_keys,
});
let thread_shared = shared.clone();
let thread = tokio::spawn(async move { let thread = tokio::spawn(async move {
loop { loop {
fn handle_stop_action<'a>( fn handle_stop_action<'a>(
@@ -205,17 +236,36 @@ impl Manager {
match stop_action { match stop_action {
OnStop::Sleep => { OnStop::Sleep => {
if let Some(fut) = fut { if let Some(fut) = fut {
thread_shared.status.store(
Status::Stopped as usize,
std::sync::atomic::Ordering::SeqCst,
);
fut.await.unwrap(); fut.await.unwrap();
continue; continue;
} }
} }
OnStop::Exit => { OnStop::Exit => {
thread_shared.status.store(
Status::Stopped as usize,
std::sync::atomic::Ordering::SeqCst,
);
break; break;
} }
OnStop::Restart => (), OnStop::Restart => {
thread_shared.status.store(
Status::Running as usize,
std::sync::atomic::Ordering::SeqCst,
);
}
} }
match run_main(&docker, &*net_ctl, &manifest, &tor_keys).await { match run_main(&thread_shared).await {
Ok(Ok(())) => break, Ok(Ok(())) => {
thread_shared
.on_stop
.send(OnStop::Sleep)
.map_err(|_| ())
.unwrap(); // recv is still in scope, cannot fail
}
Ok(Err(e)) => { Ok(Err(e)) => {
todo!("application crashed") todo!("application crashed")
} }
@@ -225,12 +275,115 @@ impl Manager {
} }
} }
}); });
Ok(Manager { on_stop, thread }) Ok(Manager {
shared,
thread: Container::new(Some(thread)),
})
} }
}
impl Drop for Manager { pub fn status(&self) -> Status {
fn drop(&mut self) { match self.shared.status.load(std::sync::atomic::Ordering::SeqCst) {
let _ = self.on_stop.send(OnStop::Exit); 0 => Status::Running,
1 => Status::Stopped,
2 => Status::Paused,
_ => unreachable!(),
}
}
pub async fn stop(&self) -> Result<(), Error> {
self.shared.on_stop.send(OnStop::Sleep).map_err(|_| {
Error::new(
anyhow!("Manager has already been shutdown"),
crate::ErrorKind::Docker,
)
})?;
if matches!(self.status(), Status::Paused) {
self.resume().await?;
}
match self
.shared
.docker
.stop_container(
&self.shared.container_name,
Some(StopContainerOptions { t: 30 }),
)
.await
{
Err(bollard::errors::Error::DockerResponseNotFoundError { .. })
| Err(bollard::errors::Error::DockerResponseConflictError { .. }) => (), // Already stopped
a => a?,
};
self.shared.status.store(
Status::Stopped as usize,
std::sync::atomic::Ordering::SeqCst,
);
Ok(())
}
pub async fn start(&self) -> Result<(), Error> {
self.shared.on_stop.send(OnStop::Restart).map_err(|_| {
Error::new(
anyhow!("Manager has already been shutdown"),
crate::ErrorKind::Docker,
)
})?;
self.shared.status.store(
Status::Running as usize,
std::sync::atomic::Ordering::SeqCst,
);
Ok(())
}
pub async fn pause(&self) -> Result<(), Error> {
self.shared
.docker
.pause_container(&self.shared.container_name)
.await?;
self.shared
.status
.store(Status::Paused as usize, std::sync::atomic::Ordering::SeqCst);
Ok(())
}
pub async fn resume(&self) -> Result<(), Error> {
self.shared
.docker
.unpause_container(&self.shared.container_name)
.await?;
self.shared.status.store(
Status::Running as usize,
std::sync::atomic::Ordering::SeqCst,
);
Ok(())
}
async fn exit(&self) -> Result<(), Error> {
let _ = self.shared.on_stop.send(OnStop::Exit);
match self
.shared
.docker
.stop_container(
&self.shared.container_name,
Some(StopContainerOptions { t: 30 }),
)
.await
{
Err(bollard::errors::Error::DockerResponseNotFoundError { .. })
| Err(bollard::errors::Error::DockerResponseConflictError { .. }) => (),
a => a?,
};
self.shared.status.store(
Status::Stopped as usize,
std::sync::atomic::Ordering::SeqCst,
);
if let Some(thread) = self.thread.take().await {
thread.await.map_err(|e| {
Error::new(
anyhow!("Manager thread panicked: {}", e),
crate::ErrorKind::Docker,
)
})?;
}
Ok(())
} }
} }

4
appmgr/src/net/mdns.rs
View File

@@ -119,7 +119,7 @@ impl MdnsControllerInner {
} }
} }
fn add<'a, I: IntoIterator<Item = (InterfaceId, TorSecretKeyV3)>>( fn add<'a, I: IntoIterator<Item = (InterfaceId, TorSecretKeyV3)>>(
&self, &mut self,
pkg_id: &PackageId, pkg_id: &PackageId,
interfaces: I, interfaces: I,
) { ) {
@@ -130,7 +130,7 @@ impl MdnsControllerInner {
); );
self.sync(); self.sync();
} }
fn remove<I: IntoIterator<Item = InterfaceId>>(&self, pkg_id: &PackageId, interfaces: I) { fn remove<I: IntoIterator<Item = InterfaceId>>(&mut self, pkg_id: &PackageId, interfaces: I) {
for interface_id in interfaces { for interface_id in interfaces {
self.services.remove(&(pkg_id.clone(), interface_id)); self.services.remove(&(pkg_id.clone(), interface_id));
} }

140
appmgr/src/status/mod.rs
View File

@@ -18,6 +18,7 @@ use crate::db::model::{
CurrentDependencyInfo, InstalledPackageDataEntryModel, PackageDataEntryModel, CurrentDependencyInfo, InstalledPackageDataEntryModel, PackageDataEntryModel,
}; };
use crate::dependencies::{Dependencies, DependencyError}; use crate::dependencies::{Dependencies, DependencyError};
use crate::manager::{Manager, Status as ManagerStatus};
use crate::net::interface::InterfaceId; use crate::net::interface::InterfaceId;
use crate::s9pk::manifest::{Manifest, PackageId}; use crate::s9pk::manifest::{Manifest, PackageId};
use crate::status::health_check::HealthCheckResultVariant; use crate::status::health_check::HealthCheckResultVariant;
@@ -31,7 +32,7 @@ pub async fn synchronize_all(ctx: &RpcContext) -> Result<(), Error> {
let mut db = ctx.db.handle(); let mut db = ctx.db.handle();
let mut pkg_ids = crate::db::DatabaseModel::new() let mut pkg_ids = crate::db::DatabaseModel::new()
.package_data() .package_data()
.keys(&mut db) .keys(&mut db, true)
.await?; .await?;
let mut container_names = Vec::with_capacity(pkg_ids.len()); let mut container_names = Vec::with_capacity(pkg_ids.len());
for id in pkg_ids.clone().into_iter() { for id in pkg_ids.clone().into_iter() {
@@ -42,7 +43,7 @@ pub async fn synchronize_all(ctx: &RpcContext) -> Result<(), Error> {
.await? .await?
.installed() .installed()
.map(|i| i.manifest().version()) .map(|i| i.manifest().version())
.get(&mut db) .get(&mut db, true)
.await? .await?
{ {
container_names.push(DockerAction::container_name(id.as_ref(), None)); container_names.push(DockerAction::container_name(id.as_ref(), None));
@@ -61,10 +62,10 @@ pub async fn synchronize_all(ctx: &RpcContext) -> Result<(), Error> {
filters, filters,
})) }))
.await?; .await?;
let mut fuckening = false;
for summary in info { for summary in info {
let id = if let Some(id) = summary.names.iter().flatten().find_map(|s| { let id = if let Some(id) = summary.names.iter().flatten().find_map(|s| {
DockerAction::uncontainer_name(s.as_str()).and_then(|(id, _)| pkg_ids.take(id)) // DockerAction::uncontainer_name(s.as_str()).and_then(|(id, _)| pkg_ids.take(&id))
todo!()
}) { }) {
id id
} else { } else {
@@ -75,7 +76,7 @@ pub async fn synchronize_all(ctx: &RpcContext) -> Result<(), Error> {
id: &PackageId, id: &PackageId,
db: &mut Db, db: &mut Db,
summary: &ContainerSummaryInner, summary: &ContainerSummaryInner,
) -> Result<bool, Error> { ) -> Result<(), Error> {
let pkg_data = crate::db::DatabaseModel::new() let pkg_data = crate::db::DatabaseModel::new()
.package_data() .package_data()
.idx_model(id) .idx_model(id)
@@ -91,32 +92,23 @@ pub async fn synchronize_all(ctx: &RpcContext) -> Result<(), Error> {
if let Some(installed) = pkg_data.installed().check(db).await? { if let Some(installed) = pkg_data.installed().check(db).await? {
( (
installed.clone().status().get_mut(db).await?, installed.clone().status().get_mut(db).await?,
installed.manifest().get(db).await?, installed.manifest().get(db, true).await?,
) )
} else { } else {
return Ok(false); return Ok(());
}; };
let res = status.main.synchronize(docker, &*manifest, summary).await?; let res = status.main.synchronize(todo!()).await?;
status.save(db).await?; status.save(db).await?;
Ok(res) Ok(res)
} }
match status(&ctx.docker, &id, &mut db, &summary).await { if let Err(e) = status(&ctx.docker, &id, &mut db, &summary).await {
Ok(a) => fuckening |= a, log::error!("Error syncronizing status of {}: {}", id, e);
Err(e) => log::error!("Error syncronizing status of {}: {}", id, e),
} }
} }
if fuckening {
tokio::process::Command::new("service")
.arg("docker")
.arg("restart")
.invoke(crate::ErrorKind::Docker)
.await?;
}
for id in pkg_ids { for id in pkg_ids {
log::warn!("No container for {}", id); log::warn!("No container for {}", id);
} }
@@ -126,17 +118,9 @@ pub async fn synchronize_all(ctx: &RpcContext) -> Result<(), Error> {
pub async fn check_all(ctx: &RpcContext) -> Result<(), Error> { pub async fn check_all(ctx: &RpcContext) -> Result<(), Error> {
let mut db = ctx.db.handle(); let mut db = ctx.db.handle();
let hosts = Arc::new(
crate::db::DatabaseModel::new()
.network()
.hosts()
.get(&mut db)
.await?
.to_owned(),
);
let pkg_ids = crate::db::DatabaseModel::new() let pkg_ids = crate::db::DatabaseModel::new()
.package_data() .package_data()
.keys(&mut db) .keys(&mut db, true)
.await?; .await?;
let mut status_manifest = Vec::with_capacity(pkg_ids.len()); let mut status_manifest = Vec::with_capacity(pkg_ids.len());
let mut status_deps = Vec::with_capacity(pkg_ids.len()); let mut status_deps = Vec::with_capacity(pkg_ids.len());
@@ -155,11 +139,11 @@ pub async fn check_all(ctx: &RpcContext) -> Result<(), Error> {
if let Some(installed) = model.installed().check(&mut db).await? { if let Some(installed) = model.installed().check(&mut db).await? {
status_manifest.push(( status_manifest.push((
installed.clone().status(), installed.clone().status(),
Arc::new(installed.clone().manifest().get(&mut db).await?), Arc::new(installed.clone().manifest().get(&mut db, true).await?),
)); ));
status_deps.push(( status_deps.push((
installed.clone().status(), installed.clone().status(),
Arc::new(installed.current_dependencies().get(&mut db).await?), Arc::new(installed.current_dependencies().get(&mut db, true).await?),
)); ));
} }
} }
@@ -181,30 +165,25 @@ pub async fn check_all(ctx: &RpcContext) -> Result<(), Error> {
} }
let (status_sender, mut statuses_recv) = tokio::sync::mpsc::channel(status_manifest.len() + 1); let (status_sender, mut statuses_recv) = tokio::sync::mpsc::channel(status_manifest.len() + 1);
let mut statuses = HashMap::with_capacity(status_manifest.len()); let mut statuses = HashMap::with_capacity(status_manifest.len());
futures::stream::iter( futures::stream::iter(status_manifest.into_iter().zip(pkg_ids.clone()))
status_manifest .for_each_concurrent(None, move |((status, manifest), id)| {
.into_iter() let status_sender = status_sender.clone();
.zip(pkg_ids.clone()) async move {
.zip(std::iter::repeat(hosts)), match tokio::spawn(main_status(status, manifest, ctx.db.handle()))
) .await
.for_each_concurrent(None, move |(((status, manifest), id), hosts)| { .unwrap()
let status_sender = status_sender.clone(); {
async move { Err(e) => {
match tokio::spawn(main_status(status, manifest, ctx.db.handle())) log::error!("Error running main health check for {}: {}", id, e);
.await log::debug!("{:?}", e);
.unwrap() }
{ Ok(status) => {
Err(e) => { status_sender.send((id, status)).await.expect("unreachable");
log::error!("Error running main health check for {}: {}", id, e); }
log::debug!("{:?}", e);
}
Ok(status) => {
status_sender.send((id, status)).await.expect("unreachable");
} }
} }
} })
}) .await;
.await;
while let Some((id, status)) = statuses_recv.recv().await { while let Some((id, status)) = statuses_recv.recv().await {
statuses.insert(id, status); statuses.insert(id, status);
} }
@@ -271,75 +250,40 @@ pub enum MainStatus {
}, },
} }
impl MainStatus { impl MainStatus {
pub async fn synchronize( pub async fn synchronize(&mut self, manager: &Manager) -> Result<(), Error> {
&mut self, match manager.status() {
docker: &Docker, ManagerStatus::Stopped => match self {
manifest: &Manifest,
summary: &ContainerSummaryInner,
) -> Result<bool, Error> {
// true if Docker Fuckening
async fn check_fuckening(docker: &Docker, manifest: &Manifest) -> Result<bool, Error> {
Ok(docker
.inspect_container(
&DockerAction::container_name(&manifest.id, &manifest.version),
None,
)
.await?
.state
.as_ref()
.and_then(|s| s.status)
== Some(ContainerStateStatusEnum::RUNNING))
}
let name = DockerAction::container_name(&manifest.id, None);
let state = summary.state.as_ref().map(|s| s.as_str());
match state {
Some("created") | Some("exited") => match self {
MainStatus::Stopped => (), MainStatus::Stopped => (),
MainStatus::Stopping => { MainStatus::Stopping => {
*self = MainStatus::Stopped; *self = MainStatus::Stopped;
} }
MainStatus::Running { started, .. } => { MainStatus::Running { started, .. } => {
*started = Utc::now(); *started = Utc::now();
docker manager.start().await?;
.start_container(&name, None::<StartContainerOptions<String>>)
.await?;
} }
MainStatus::BackingUp { .. } => (), MainStatus::BackingUp { .. } => (),
MainStatus::Restoring { .. } => (), MainStatus::Restoring { .. } => (),
}, },
Some("running") | Some("restarting") => match self { ManagerStatus::Running => match self {
MainStatus::Stopped | MainStatus::Stopping | MainStatus::Restoring { .. } => { MainStatus::Stopped | MainStatus::Stopping | MainStatus::Restoring { .. } => {
docker manager.stop().await?;
.stop_container(&name, Some(StopContainerOptions { t: 30 }))
.await?;
return check_fuckening(docker, manifest).await;
} }
MainStatus::Running { .. } => (), MainStatus::Running { .. } => (),
MainStatus::BackingUp { .. } => { MainStatus::BackingUp { .. } => {
docker.pause_container(&name).await?; manager.pause().await?;
} }
}, },
Some("paused") => match self { ManagerStatus::Paused => match self {
MainStatus::Stopped | MainStatus::Stopping | MainStatus::Restoring { .. } => { MainStatus::Stopped | MainStatus::Stopping | MainStatus::Restoring { .. } => {
docker.unpause_container(&name).await?; manager.stop().await?;
docker
.stop_container(&name, Some(StopContainerOptions { t: 30 }))
.await?;
return check_fuckening(docker, manifest).await;
} }
MainStatus::Running { .. } => { MainStatus::Running { .. } => {
docker.unpause_container(&name).await?; manager.resume().await?;
} }
MainStatus::BackingUp { .. } => (), MainStatus::BackingUp { .. } => (),
}, },
unknown => {
return Err(Error::new(
anyhow!("Unexpected Docker Status: {:?}", unknown),
crate::ErrorKind::Docker,
));
}
} }
Ok(false) Ok(())
} }
pub async fn check(&mut self, manifest: &Manifest) -> Result<(), Error> { pub async fn check(&mut self, manifest: &Manifest) -> Result<(), Error> {
match self { match self {

16
appmgr/src/util.rs
View File

@@ -16,7 +16,7 @@ use serde_json::Value;
use sqlx::{Executor, Sqlite}; use sqlx::{Executor, Sqlite};
use tokio::fs::File; use tokio::fs::File;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
use tokio::sync::RwLock; use tokio::sync::{Mutex, RwLock};
use crate::{Error, ResultExt as _}; use crate::{Error, ResultExt as _};
@@ -826,11 +826,17 @@ pub fn parse_duration(arg: &str, matches: &ArgMatches<'_>) -> Result<Duration, E
pub struct Container<T>(RwLock<Option<T>>); pub struct Container<T>(RwLock<Option<T>>);
impl<T> Container<T> { impl<T> Container<T> {
pub fn new() -> Self { pub fn new(value: Option<T>) -> Self {
Container(RwLock::new(None)) Container(RwLock::new(value))
} }
pub async fn set(&self, value: T) { pub async fn set(&self, value: T) -> Option<T> {
*self.0.write().await = Some(value); std::mem::replace(&mut *self.0.write().await, Some(value))
}
pub async fn take(&self) -> Option<T> {
std::mem::replace(&mut *self.0.write().await, None)
}
pub async fn is_empty(&self) -> bool {
self.0.read().await.is_none()
} }
pub async fn drop(&self) { pub async fn drop(&self) {
*self.0.write().await = None; *self.0.write().await = None;