add system images to os build process (#620)

* update compat build and add to os build process

* copy to explicit file

* fix paths and loading input

* temp save docker images output

* add docker config file to suppress warning

* notes and first attempt of load image script

* pr fixes

* run as root, fix executable path

* wip fixes

* fix pool name and stop docker before umount

* start docker again

* reset docker fs

* remove mkdir

* load system images during embassy-init

* add utils to make build

* fix utils source

* create system-images dir as root

* cleanup

* make loading docker images datadir agnostic

* address PR feedback

* rework load images

* create shutdown channel on failed embassy-init run

* pr feedback

* fix import

appmgr/src/bin/embassy-init.rs

@@ -4,16 +4,19 @@ use embassy::context::rpc::RpcContextConfig;
 use embassy::context::{DiagnosticContext, SetupContext};
 use embassy::db::model::ServerStatus;
 use embassy::disk::main::DEFAULT_PASSWORD;
+use embassy::install::PKG_DOCKER_DIR;
 use embassy::middleware::cors::cors;
 use embassy::middleware::diagnostic::diagnostic;
 use embassy::middleware::encrypt::encrypt;
 #[cfg(feature = "avahi")]
 use embassy::net::mdns::MdnsController;
+use embassy::shutdown::Shutdown;
 use embassy::sound::MARIO_COIN;
 use embassy::util::logger::EmbassyLogger;
 use embassy::util::Invoke;
 use embassy::{Error, ResultExt};
 use http::StatusCode;
+use nix::sys::socket::shutdown;
 use rpc_toolkit::rpc_server;
 use tokio::process::Command;
 use tracing::instrument;
@@ -76,7 +79,7 @@ async fn init(cfg_path: Option<&str>) -> Result<(), Error> {
     }
 
     embassy::disk::main::load(
-        tokio::fs::read_to_string("/embassy-os/disk.guid")
+        tokio::fs::read_to_string("/embassy-os/disk.guid") // unique identifier for zfs pool - keeps track of the disk that goes with your embassy
             .await?
             .trim(),
         cfg.zfs_pool_name(),
@@ -124,8 +127,13 @@ async fn init(cfg_path: Option<&str>) -> Result<(), Error> {
         .invoke(embassy::ErrorKind::Docker)
         .await?;
     tracing::info!("Mounted Docker Data");
-    embassy::install::load_images(cfg.datadir()).await?;
+
+    embassy::install::load_images(cfg.datadir().join(PKG_DOCKER_DIR)).await?;
     tracing::info!("Loaded Docker Images");
+    // Loading system images
+    embassy::install::load_images("/var/lib/embassy/system-images").await?;
+    tracing::info!("Loaded System Docker Images");
+
     embassy::ssh::sync_keys_from_db(&secret_store, "/root/.ssh/authorized_keys").await?;
     tracing::info!("Synced SSH Keys");
 
@@ -175,7 +183,7 @@ async fn run_script_if_exists<P: AsRef<Path>>(path: P) {
 }
 
 #[instrument]
-async fn inner_main(cfg_path: Option<&str>) -> Result<(), Error> {
+async fn inner_main(cfg_path: Option<&str>) -> Result<Option<Shutdown>, Error> {
     embassy::sound::BEP.play().await?;
 
     run_script_if_exists("/embassy-os/preinit.sh").await;
@@ -204,6 +212,7 @@ async fn inner_main(cfg_path: Option<&str>) -> Result<(), Error> {
                 .invoke(embassy::ErrorKind::Nginx)
                 .await?;
             let ctx = DiagnosticContext::init(cfg_path, e).await?;
+            let mut shutdown_recv = ctx.shutdown.subscribe();
             rpc_server!({
                 command: embassy::diagnostic_api,
                 context: ctx.clone(),
@@ -221,11 +230,17 @@ async fn inner_main(cfg_path: Option<&str>) -> Result<(), Error> {
             })
             .await
             .with_kind(embassy::ErrorKind::Network)?;
-            Ok::<_, Error>(())
+
+            Ok::<_, Error>(
+                shutdown_recv
+                    .recv()
+                    .await
+                    .with_kind(embassy::ErrorKind::Network)?,
+            )
         })()
         .await
     } else {
-        Ok(())
+        Ok(None)
     };
 
     run_script_if_exists("/embassy-os/postinit.sh").await;
@@ -255,7 +270,8 @@ fn main() {
     };
 
     match res {
-        Ok(_) => (),
+        Ok(Some(shutdown)) => shutdown.execute(),
+        Ok(None) => (),
         Err(e) => {
             eprintln!("{}", e.source);
             tracing::debug!("{:?}", e.source);

appmgr/src/install/mod.rs

@@ -1,23 +1,11 @@
 use std::collections::{BTreeMap, BTreeSet};
+use std::ffi::OsStr;
 use std::io::SeekFrom;
 use std::path::Path;
 use std::process::Stdio;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
-use color_eyre::eyre::{self, eyre};
-use emver::VersionRange;
-use futures::TryStreamExt;
-use http::StatusCode;
-use patch_db::{DbHandle, LockType};
-use reqwest::Response;
-use rpc_toolkit::command;
-use tokio::fs::{File, OpenOptions};
-use tokio::io::{AsyncRead, AsyncSeek, AsyncSeekExt};
-use tokio::process::Command;
-use tokio_stream::wrappers::ReadDirStream;
-use tracing::instrument;
-
 use self::cleanup::cleanup_failed;
 use crate::context::RpcContext;
 use crate::db::model::{
@@ -38,6 +26,19 @@ use crate::util::io::copy_and_shutdown;
 use crate::util::{display_none, display_serializable, AsyncFileExt, Version};
 use crate::volume::asset_dir;
 use crate::{Error, ResultExt};
+use color_eyre::eyre::{self, eyre};
+use emver::VersionRange;
+use futures::future::BoxFuture;
+use futures::{FutureExt, StreamExt, TryStreamExt};
+use http::StatusCode;
+use patch_db::{DbHandle, LockType};
+use reqwest::Response;
+use rpc_toolkit::command;
+use tokio::fs::{DirEntry, File, OpenOptions};
+use tokio::io::{AsyncRead, AsyncSeek, AsyncSeekExt};
+use tokio::process::Command;
+use tokio_stream::wrappers::ReadDirStream;
+use tracing::instrument;
 
 pub mod cleanup;
 pub mod progress;
@@ -766,58 +767,62 @@ async fn handle_recovered_package(
 }
 
 #[instrument(skip(datadir))]
-pub async fn load_images<P: AsRef<Path>>(datadir: P) -> Result<(), Error> {
-    let docker_dir = datadir.as_ref().join(PKG_DOCKER_DIR);
-    if tokio::fs::metadata(&docker_dir).await.is_ok() {
-        ReadDirStream::new(tokio::fs::read_dir(&docker_dir).await?)
-            .map_err(|e| {
-                Error::new(
-                    eyre::Report::from(e).wrap_err(format!("{:?}", &docker_dir)),
-                    crate::ErrorKind::Filesystem,
-                )
-            })
-            .try_for_each_concurrent(None, |pkg_id| async move {
-                ReadDirStream::new(tokio::fs::read_dir(pkg_id.path()).await?)
-                    .map_err(|e| {
-                        Error::new(
-                            eyre::Report::from(e).wrap_err(pkg_id.path().display().to_string()),
-                            crate::ErrorKind::Filesystem,
-                        )
-                    })
-                    .try_for_each_concurrent(None, |version| async move {
-                        let mut load = Command::new("docker")
-                            .arg("load")
-                            .stdin(Stdio::piped())
-                            .stderr(Stdio::piped())
-                            .spawn()?;
-                        let load_in = load.stdin.take().ok_or_else(|| {
-                            Error::new(
-                                eyre!("Could not write to stdin of docker load"),
-                                crate::ErrorKind::Docker,
-                            )
-                        })?;
-                        let mut docker_rdr = File::open(version.path().join("image.tar")).await?;
-                        copy_and_shutdown(&mut docker_rdr, load_in).await?;
-                        let res = load.wait_with_output().await?;
-                        if !res.status.success() {
-                            Err(Error::new(
-                                eyre!(
-                                    "{}",
-                                    String::from_utf8(res.stderr).unwrap_or_else(|e| format!(
-                                        "Could not parse stderr: {}",
-                                        e
-                                    ))
-                                ),
-                                crate::ErrorKind::Docker,
-                            ))
+pub fn load_images<'a, P: AsRef<Path> + 'a + Send + Sync>(
+    datadir: P,
+) -> BoxFuture<'a, Result<(), Error>> {
+    async move {
+        let docker_dir = datadir.as_ref();
+        if tokio::fs::metadata(&docker_dir).await.is_ok() {
+            ReadDirStream::new(tokio::fs::read_dir(&docker_dir).await?)
+                .map(|r| {
+                    r.with_ctx(|_| (crate::ErrorKind::Filesystem, format!("{:?}", &docker_dir)))
+                })
+                .try_for_each(|entry| async move {
+                    let m = entry.metadata().await?;
+                    if m.is_file() {
+                        if entry.path().extension().and_then(|ext| ext.to_str()) == Some("tar") {
+                            let mut load = Command::new("docker")
+                                .arg("load")
+                                .stdin(Stdio::piped())
+                                .stderr(Stdio::piped())
+                                .spawn()?;
+                            let load_in = load.stdin.take().ok_or_else(|| {
+                                Error::new(
+                                    eyre!("Could not write to stdin of docker load"),
+                                    crate::ErrorKind::Docker,
+                                )
+                            })?;
+                            let mut docker_rdr = File::open(&entry.path()).await?;
+                            copy_and_shutdown(&mut docker_rdr, load_in).await?;
+                            let res = load.wait_with_output().await?;
+                            if !res.status.success() {
+                                Err(Error::new(
+                                    eyre!(
+                                        "{}",
+                                        String::from_utf8(res.stderr).unwrap_or_else(|e| format!(
+                                            "Could not parse stderr: {}",
+                                            e
+                                        ))
+                                    ),
+                                    crate::ErrorKind::Docker,
+                                ))
+                            } else {
+                                Ok(())
+                            }
                         } else {
                             Ok(())
                         }
-                    })
-                    .await
-            })
-            .await
-    } else {
-        Ok(())
+                    } else if m.is_dir() {
+                        load_images(entry.path()).await?;
+                        Ok(())
+                    } else {
+                        Ok(())
+                    }
+                })
+                .await
+        } else {
+            Ok(())
+        }
     }
+    .boxed()
 }