Feat/long running (#1676)

* feat: Start the long running container * feat: Long running docker, running, stoping, and uninstalling * feat: Just make the folders that we would like to mount. * fix: Uninstall not working * chore: remove some logging * feat: Smarter cleanup * feat: Wait for start * wip: Need to kill * chore: Remove the bad tracing * feat: Stopping the long running processes without killing the long running * Mino Feat: Change the Manifest To have a new type (#1736) * Add build-essential to README.md (#1716) Update README.md * write image to sparse-aware archive format (#1709) * fix: Add modification to the max_user_watches (#1695) * fix: Add modification to the max_user_watches * chore: Move to initialization * [Feat] follow logs (#1714) * tail logs * add cli * add FE * abstract http to shared * batch new logs * file download for logs * fix modal error when no config Co-authored-by: Chris Guida <chrisguida@users.noreply.github.com> Co-authored-by: Aiden McClelland <me@drbonez.dev> Co-authored-by: Matt Hill <matthewonthemoon@gmail.com> Co-authored-by: BluJ <mogulslayer@gmail.com> * Update README.md (#1728) * fix build for patch-db client for consistency (#1722) * fix cli install (#1720) * highlight instructions if not viewed (#1731) * wip: * [ ] Fix the build (dependencies:634 map for option) * fix: Cargo build * fix: Long running wasn't starting * fix: uninstall works Co-authored-by: Chris Guida <chrisguida@users.noreply.github.com> Co-authored-by: Aiden McClelland <3732071+dr-bonez@users.noreply.github.com> Co-authored-by: Aiden McClelland <me@drbonez.dev> Co-authored-by: Matt Hill <matthewonthemoon@gmail.com> Co-authored-by: Lucy C <12953208+elvece@users.noreply.github.com> Co-authored-by: Matt Hill <MattDHill@users.noreply.github.com> * chore: Fix a dbg! * chore: Make the commands of the docker-inject do inject instead of exec * chore: Fix compile mistake * chore: Change to use simpler Co-authored-by: Chris Guida <chrisguida@users.noreply.github.com> Co-authored-by: Aiden McClelland <3732071+dr-bonez@users.noreply.github.com> Co-authored-by: Aiden McClelland <me@drbonez.dev> Co-authored-by: Matt Hill <matthewonthemoon@gmail.com> Co-authored-by: Lucy C <12953208+elvece@users.noreply.github.com> Co-authored-by: Matt Hill <MattDHill@users.noreply.github.com>
2026-03-30 12:11:56 +00:00 · 2022-08-19 11:05:23 -06:00
parent 8093faee19
commit 35b220d7a5
23 changed files with 1236 additions and 322 deletions
--- a/backend/src/procedure/docker.rs
+++ b/backend/src/procedure/docker.rs
@@ -41,6 +41,16 @@ lazy_static::lazy_static! {
    };
 }

+#[derive(Clone, Debug, Deserialize, Serialize, patch_db::HasModel)]
+#[serde(rename_all = "kebab-case")]
+pub struct DockerContainer {
+    pub image: ImageId,
+    #[serde(default)]
+    pub mounts: BTreeMap<VolumeId, PathBuf>,
+    #[serde(default)]
+    pub shm_size_mb: Option<usize>, // TODO: use postfix sizing? like 1k vs 1m vs 1g
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "kebab-case")]
 pub struct DockerProcedure {
@@ -55,12 +65,40 @@ pub struct DockerProcedure {
    #[serde(default)]
    pub io_format: Option<IoFormat>,
    #[serde(default)]
-    pub inject: bool,
+    pub sigterm_timeout: Option<SerdeDuration>,
    #[serde(default)]
    pub shm_size_mb: Option<usize>, // TODO: use postfix sizing? like 1k vs 1m vs 1g
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct DockerInject {
+    #[serde(default)]
+    pub system: bool,
+    pub entrypoint: String,
+    #[serde(default)]
+    pub args: Vec<String>,
+    #[serde(default)]
+    pub io_format: Option<IoFormat>,
    #[serde(default)]
    pub sigterm_timeout: Option<SerdeDuration>,
 }
+
+impl From<(&DockerContainer, &DockerInject)> for DockerProcedure {
+    fn from((container, injectable): (&DockerContainer, &DockerInject)) -> Self {
+        DockerProcedure {
+            image: container.image.clone(),
+            system: injectable.system.clone(),
+            entrypoint: injectable.entrypoint.clone(),
+            args: injectable.args.clone(),
+            mounts: container.mounts.clone(),
+            io_format: injectable.io_format.clone(),
+            sigterm_timeout: injectable.sigterm_timeout.clone(),
+            shm_size_mb: container.shm_size_mb.clone(),
+        }
+    }
+}
+
 impl DockerProcedure {
    pub fn validate(
        &self,
@@ -86,12 +124,6 @@ impl DockerProcedure {
        if expected_io && self.io_format.is_none() {
            color_eyre::eyre::bail!("expected io-format");
        }
-        if &**eos_version >= &emver::Version::new(0, 3, 1, 1)
-            && self.inject
-            && !self.mounts.is_empty()
-        {
-            color_eyre::eyre::bail!("mounts not allowed in inject actions");
-        }
        Ok(())
    }

@@ -104,48 +136,196 @@ impl DockerProcedure {
        name: ProcedureName,
        volumes: &Volumes,
        input: Option<I>,
-        allow_inject: bool,
        timeout: Option<Duration>,
    ) -> Result<Result<O, (i32, String)>, Error> {
        let name = name.docker_name();
        let name: Option<&str> = name.as_ref().map(|x| &**x);
        let mut cmd = tokio::process::Command::new("docker");
-        if self.inject && allow_inject {
-            cmd.arg("exec");
+        tracing::debug!("{:?} is run", name);
+        let container_name = Self::container_name(pkg_id, name);
+        cmd.arg("run")
+            .arg("--rm")
+            .arg("--network=start9")
+            .arg(format!("--add-host=embassy:{}", Ipv4Addr::from(HOST_IP)))
+            .arg("--name")
+            .arg(&container_name)
+            .arg(format!("--hostname={}", &container_name))
+            .arg("--no-healthcheck");
+        match ctx
+            .docker
+            .remove_container(
+                &container_name,
+                Some(RemoveContainerOptions {
+                    v: false,
+                    force: true,
+                    link: false,
+                }),
+            )
+            .await
+        {
+            Ok(())
+            | Err(bollard::errors::Error::DockerResponseServerError {
+                status_code: 404, // NOT FOUND
+                ..
+            }) => Ok(()),
+            Err(e) => Err(e),
+        }?;
+        cmd.args(self.docker_args(ctx, pkg_id, pkg_version, volumes).await?);
+        let input_buf = if let (Some(input), Some(format)) = (&input, &self.io_format) {
+            cmd.stdin(std::process::Stdio::piped());
+            Some(format.to_vec(input)?)
        } else {
-            let container_name = Self::container_name(pkg_id, name);
-            cmd.arg("run")
-                .arg("--rm")
-                .arg("--network=start9")
-                .arg(format!("--add-host=embassy:{}", Ipv4Addr::from(HOST_IP)))
-                .arg("--name")
-                .arg(&container_name)
-                .arg(format!("--hostname={}", &container_name))
-                .arg("--no-healthcheck");
-            match ctx
-                .docker
-                .remove_container(
-                    &container_name,
-                    Some(RemoveContainerOptions {
-                        v: false,
-                        force: true,
-                        link: false,
-                    }),
-                )
-                .await
-            {
-                Ok(())
-                | Err(bollard::errors::Error::DockerResponseServerError {
-                    status_code: 404, // NOT FOUND
-                    ..
-                }) => Ok(()),
-                Err(e) => Err(e),
-            }?;
-        }
-        cmd.args(
-            self.docker_args(ctx, pkg_id, pkg_version, volumes, allow_inject)
-                .await,
+            None
+        };
+        cmd.stdout(std::process::Stdio::piped());
+        cmd.stderr(std::process::Stdio::piped());
+        tracing::trace!(
+            "{}",
+            format!("{:?}", cmd)
+                .split(r#"" ""#)
+                .collect::<Vec<&str>>()
+                .join(" ")
        );
+        let mut handle = cmd.spawn().with_kind(crate::ErrorKind::Docker)?;
+        let id = handle.id();
+        let timeout_fut = if let Some(timeout) = timeout {
+            EitherFuture::Right(async move {
+                tokio::time::sleep(timeout).await;
+
+                Ok(())
+            })
+        } else {
+            EitherFuture::Left(futures::future::pending::<Result<_, Error>>())
+        };
+        if let (Some(input), Some(mut stdin)) = (&input_buf, handle.stdin.take()) {
+            use tokio::io::AsyncWriteExt;
+            stdin
+                .write_all(input)
+                .await
+                .with_kind(crate::ErrorKind::Docker)?;
+            stdin.flush().await?;
+            stdin.shutdown().await?;
+            drop(stdin);
+        }
+        enum Race<T> {
+            Done(T),
+            TimedOut,
+        }
+
+        let io_format = self.io_format;
+        let mut output = BufReader::new(
+            handle
+                .stdout
+                .take()
+                .ok_or_else(|| eyre!("Can't takeout stout"))
+                .with_kind(crate::ErrorKind::Docker)?,
+        );
+        let output = NonDetachingJoinHandle::from(tokio::spawn(async move {
+            match async {
+                if let Some(format) = io_format {
+                    return match max_by_lines(&mut output, None).await {
+                        MaxByLines::Done(buffer) => {
+                            Ok::<Value, Error>(
+                                match format.from_slice(buffer.as_bytes()) {
+                                    Ok(a) => a,
+                                    Err(e) => {
+                                        tracing::trace!(
+                                        "Failed to deserialize stdout from {}: {}, falling back to UTF-8 string.",
+                                        format,
+                                        e
+                                    );
+                                        Value::String(buffer)
+                                    }
+                                },
+                            )
+                        },
+                        MaxByLines::Error(e) => Err(e),
+                        MaxByLines::Overflow(buffer) => Ok(Value::String(buffer))
+                    }
+                }
+
+                let lines = buf_reader_to_lines(&mut output, 1000).await?;
+                if lines.is_empty() {
+                    return Ok(Value::Null);
+                }
+
+                let joined_output = lines.join("\n");
+                Ok(Value::String(joined_output))
+            }.await {
+                Ok(a) => Ok((a, output)),
+                Err(e) => Err((e, output))
+            }
+        }));
+        let err_output = BufReader::new(
+            handle
+                .stderr
+                .take()
+                .ok_or_else(|| eyre!("Can't takeout std err"))
+                .with_kind(crate::ErrorKind::Docker)?,
+        );
+
+        let err_output = NonDetachingJoinHandle::from(tokio::spawn(async move {
+            let lines = buf_reader_to_lines(err_output, 1000).await?;
+            let joined_output = lines.join("\n");
+            Ok::<_, Error>(joined_output)
+        }));
+
+        let res = tokio::select! {
+            res = handle.wait() => Race::Done(res.with_kind(crate::ErrorKind::Docker)?),
+            res = timeout_fut => {
+                res?;
+                Race::TimedOut
+            },
+        };
+        let exit_status = match res {
+            Race::Done(x) => x,
+            Race::TimedOut => {
+                if let Some(id) = id {
+                    signal::kill(Pid::from_raw(id as i32), signal::SIGKILL)
+                        .with_kind(crate::ErrorKind::Docker)?;
+                }
+                return Ok(Err((143, "Timed out. Retrying soon...".to_owned())));
+            }
+        };
+        Ok(
+            if exit_status.success() || exit_status.code() == Some(143) {
+                Ok(serde_json::from_value(
+                    output
+                        .await
+                        .with_kind(crate::ErrorKind::Unknown)?
+                        .map(|(v, _)| v)
+                        .map_err(|(e, _)| tracing::warn!("{}", e))
+                        .unwrap_or_default(),
+                )
+                .with_kind(crate::ErrorKind::Deserialization)?)
+            } else {
+                Err((
+                    exit_status.code().unwrap_or_default(),
+                    err_output.await.with_kind(crate::ErrorKind::Unknown)??,
+                ))
+            },
+        )
+    }
+
+    #[instrument(skip(ctx, input))]
+    pub async fn inject<I: Serialize, O: for<'de> Deserialize<'de>>(
+        &self,
+        ctx: &RpcContext,
+        pkg_id: &PackageId,
+        pkg_version: &Version,
+        name: ProcedureName,
+        volumes: &Volumes,
+        input: Option<I>,
+        timeout: Option<Duration>,
+    ) -> Result<Result<O, (i32, String)>, Error> {
+        let name = name.docker_name();
+        let name: Option<&str> = name.as_ref().map(|x| &**x);
+        let mut cmd = tokio::process::Command::new("docker");
+
+        tracing::debug!("{:?} is exec", name);
+        cmd.arg("exec");
+
+        cmd.args(self.docker_args_inject(ctx, pkg_id, pkg_version).await?);
        let input_buf = if let (Some(input), Some(format)) = (&input, &self.io_format) {
            cmd.stdin(std::process::Stdio::piped());
            Some(format.to_vec(input)?)
@@ -295,8 +475,8 @@ impl DockerProcedure {
        let mut cmd = tokio::process::Command::new("docker");
        cmd.arg("run").arg("--rm").arg("--network=none");
        cmd.args(
-            self.docker_args(ctx, pkg_id, pkg_version, &volumes.to_readonly(), false)
-                .await,
+            self.docker_args(ctx, pkg_id, pkg_version, &volumes.to_readonly())
+                .await?,
        );
        let input_buf = if let (Some(input), Some(format)) = (&input, &self.io_format) {
            cmd.stdin(std::process::Stdio::piped());
@@ -418,14 +598,8 @@ impl DockerProcedure {
        pkg_id: &PackageId,
        pkg_version: &Version,
        volumes: &Volumes,
-        allow_inject: bool,
-    ) -> Vec<Cow<'_, OsStr>> {
-        let mut res = Vec::with_capacity(
-            (2 * self.mounts.len()) // --mount <MOUNT_ARG>
-                + (2 * self.shm_size_mb.is_some() as usize) // --shm-size <SHM_SIZE>
-                + 5 // --interactive --log-driver=journald --entrypoint <ENTRYPOINT> <IMAGE>
-                + self.args.len(), // [ARG...]
-        );
+    ) -> Result<Vec<Cow<'_, OsStr>>, Error> {
+        let mut res = self.new_docker_args();
        for (volume_id, dst) in &self.mounts {
            let volume = if let Some(v) = volumes.get(volume_id) {
                v
@@ -434,8 +608,7 @@ impl DockerProcedure {
            };
            let src = volume.path_for(&ctx.datadir, pkg_id, pkg_version, volume_id);
            if let Err(e) = tokio::fs::metadata(&src).await {
-                tracing::warn!("{} not mounted to container: {}", src.display(), e);
-                continue;
+                tokio::fs::create_dir_all(&src).await?;
            }
            res.push(OsStr::new("--mount").into());
            res.push(
@@ -453,22 +626,48 @@ impl DockerProcedure {
            res.push(OsString::from(format!("{}m", shm_size_mb)).into());
        }
        res.push(OsStr::new("--interactive").into());
-        if self.inject && allow_inject {
-            res.push(OsString::from(Self::container_name(pkg_id, None)).into());
-            res.push(OsStr::new(&self.entrypoint).into());
+
+        res.push(OsStr::new("--log-driver=journald").into());
+        res.push(OsStr::new("--entrypoint").into());
+        res.push(OsStr::new(&self.entrypoint).into());
+        if self.system {
+            res.push(OsString::from(self.image.for_package(SYSTEM_PACKAGE_ID, None)).into());
        } else {
-            res.push(OsStr::new("--log-driver=journald").into());
-            res.push(OsStr::new("--entrypoint").into());
-            res.push(OsStr::new(&self.entrypoint).into());
-            if self.system {
-                res.push(OsString::from(self.image.for_package(SYSTEM_PACKAGE_ID, None)).into());
-            } else {
-                res.push(OsString::from(self.image.for_package(pkg_id, Some(pkg_version))).into());
-            }
+            res.push(OsString::from(self.image.for_package(pkg_id, Some(pkg_version))).into());
        }
+
        res.extend(self.args.iter().map(|s| OsStr::new(s).into()));

-        res
+        Ok(res)
+    }
+
+    fn new_docker_args(&self) -> Vec<Cow<OsStr>> {
+        Vec::with_capacity(
+            (2 * self.mounts.len()) // --mount <MOUNT_ARG>
+                + (2 * self.shm_size_mb.is_some() as usize) // --shm-size <SHM_SIZE>
+                + 5 // --interactive --log-driver=journald --entrypoint <ENTRYPOINT> <IMAGE>
+                + self.args.len(), // [ARG...]
+        )
+    }
+    async fn docker_args_inject(
+        &self,
+        ctx: &RpcContext,
+        pkg_id: &PackageId,
+        pkg_version: &Version,
+    ) -> Result<Vec<Cow<'_, OsStr>>, Error> {
+        let mut res = self.new_docker_args();
+        if let Some(shm_size_mb) = self.shm_size_mb {
+            res.push(OsStr::new("--shm-size").into());
+            res.push(OsString::from(format!("{}m", shm_size_mb)).into());
+        }
+        res.push(OsStr::new("--interactive").into());
+
+        res.push(OsString::from(Self::container_name(pkg_id, None)).into());
+        res.push(OsStr::new(&self.entrypoint).into());
+
+        res.extend(self.args.iter().map(|s| OsStr::new(s).into()));
+
+        Ok(res)
    }
 }

--- a/backend/src/procedure/mod.rs
+++ b/backend/src/procedure/mod.rs
@@ -1,11 +1,12 @@
 use std::collections::BTreeSet;
 use std::time::Duration;

+use color_eyre::eyre::{bail, eyre};
 use patch_db::HasModel;
 use serde::{Deserialize, Serialize};
 use tracing::instrument;

-use self::docker::DockerProcedure;
+use self::docker::{DockerContainer, DockerInject, DockerProcedure};
 use crate::context::RpcContext;
 use crate::id::ImageId;
 use crate::s9pk::manifest::PackageId;
@@ -25,10 +26,12 @@ pub use models::ProcedureName;
 #[serde(tag = "type")]
 pub enum PackageProcedure {
    Docker(DockerProcedure),
+    DockerInject(DockerInject),

    #[cfg(feature = "js_engine")]
    Script(js_scripts::JsProcedure),
 }
+
 impl PackageProcedure {
    pub fn is_script(&self) -> bool {
        match self {
@@ -40,6 +43,7 @@ impl PackageProcedure {
    #[instrument]
    pub fn validate(
        &self,
+        container: &Option<DockerContainer>,
        eos_version: &Version,
        volumes: &Volumes,
        image_ids: &BTreeSet<ImageId>,
@@ -49,40 +53,95 @@ impl PackageProcedure {
            PackageProcedure::Docker(action) => {
                action.validate(eos_version, volumes, image_ids, expected_io)
            }
+            PackageProcedure::DockerInject(injectable) => {
+                let container = match container {
+                    None => bail!("For the docker injectable procedure, a container must be exist on the config"),
+                    Some(container) => container,
+                } ;
+                let docker_procedure: DockerProcedure = (container, injectable).into();
+                docker_procedure.validate(eos_version, volumes, image_ids, expected_io)
+            }

            #[cfg(feature = "js_engine")]
            PackageProcedure::Script(action) => action.validate(volumes),
        }
    }

-    #[instrument(skip(ctx, input))]
+    #[instrument(skip(ctx, input, container))]
    pub async fn execute<I: Serialize, O: for<'de> Deserialize<'de>>(
        &self,
        ctx: &RpcContext,
+        container: &Option<DockerContainer>,
        pkg_id: &PackageId,
        pkg_version: &Version,
        name: ProcedureName,
        volumes: &Volumes,
        input: Option<I>,
-        allow_inject: bool,
        timeout: Option<Duration>,
    ) -> Result<Result<O, (i32, String)>, Error> {
        tracing::trace!("Procedure execute {} {} - {:?}", self, pkg_id, name);
        match self {
            PackageProcedure::Docker(procedure) => {
+                procedure
+                    .execute(ctx, pkg_id, pkg_version, name, volumes, input, timeout)
+                    .await
+            }
+            PackageProcedure::DockerInject(injectable) => {
+                let container = match container {
+                    None => return Err(Error::new(eyre!("For the docker injectable procedure, a container must be exist on the config"), crate::ErrorKind::Action)),
+                    Some(container) => container,
+                } ;
+                let docker_procedure: DockerProcedure = (container, injectable).into();
+                docker_procedure
+                    .inject(ctx, pkg_id, pkg_version, name, volumes, input, timeout)
+                    .await
+            }
+            #[cfg(feature = "js_engine")]
+            PackageProcedure::Script(procedure) => {
                procedure
                    .execute(
-                        ctx,
+                        &ctx.datadir,
                        pkg_id,
                        pkg_version,
                        name,
                        volumes,
                        input,
-                        allow_inject,
                        timeout,
                    )
                    .await
            }
+        }
+    }
+
+    #[instrument(skip(ctx, input, container))]
+    pub async fn inject<I: Serialize, O: for<'de> Deserialize<'de>>(
+        &self,
+        ctx: &RpcContext,
+        container: &Option<DockerContainer>,
+        pkg_id: &PackageId,
+        pkg_version: &Version,
+        name: ProcedureName,
+        volumes: &Volumes,
+        input: Option<I>,
+        timeout: Option<Duration>,
+    ) -> Result<Result<O, (i32, String)>, Error> {
+        tracing::trace!("Procedure inject {} {} - {:?}", self, pkg_id, name);
+        match self {
+            PackageProcedure::Docker(procedure) => {
+                procedure
+                    .inject(ctx, pkg_id, pkg_version, name, volumes, input, timeout)
+                    .await
+            }
+            PackageProcedure::DockerInject(injectable) => {
+                let container = match container {
+                    None => return Err(Error::new(eyre!("For the docker injectable procedure, a container must be exist on the config"), crate::ErrorKind::Action)),
+                    Some(container) => container,
+                } ;
+                let docker_procedure: DockerProcedure = (container, injectable).into();
+                docker_procedure
+                    .inject(ctx, pkg_id, pkg_version, name, volumes, input, timeout)
+                    .await
+            }
            #[cfg(feature = "js_engine")]
            PackageProcedure::Script(procedure) => {
                procedure
@@ -102,6 +161,7 @@ impl PackageProcedure {
    #[instrument(skip(ctx, input))]
    pub async fn sandboxed<I: Serialize, O: for<'de> Deserialize<'de>>(
        &self,
+        container: &Option<DockerContainer>,
        ctx: &RpcContext,
        pkg_id: &PackageId,
        pkg_version: &Version,
@@ -117,6 +177,16 @@ impl PackageProcedure {
                    .sandboxed(ctx, pkg_id, pkg_version, volumes, input, timeout)
                    .await
            }
+            PackageProcedure::DockerInject(injectable) => {
+                let container = match container {
+                    None => return Err(Error::new(eyre!("For the docker injectable procedure, a container must be exist on the config"), crate::ErrorKind::Action)),
+                    Some(container) => container,
+                } ;
+                let docker_procedure: DockerProcedure = (container, injectable).into();
+                docker_procedure
+                    .sandboxed(ctx, pkg_id, pkg_version, volumes, input, timeout)
+                    .await
+            }
            #[cfg(feature = "js_engine")]
            PackageProcedure::Script(procedure) => {
                procedure
@@ -130,6 +200,7 @@ impl PackageProcedure {
 impl std::fmt::Display for PackageProcedure {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
+            PackageProcedure::DockerInject(_) => write!(f, "Docker Injectable")?,
            PackageProcedure::Docker(_) => write!(f, "Docker")?,
            #[cfg(feature = "js_engine")]
            PackageProcedure::Script(_) => write!(f, "JS")?,