Feat: revive via gateway endpoint (#460)

oddgrd · brokad · web-flow · commit 88c877df6e94 · 2022-11-04T17:53:35.000+02:00
* feat: revive via gateway api initial commit

* feat: cleanup, document gateway testing

* refactor: use post for revive endpoint

Co-authored-by: Damien &lt;damienbroka@mailbox.org&gt;

* feat: send task to global queue

Co-authored-by: Damien &lt;damienbroka@mailbox.org&gt;
diff --git a/gateway/README.md b/gateway/README.md
@@ -0,0 +1,8 @@
+# shuttle-gateway
+
+## Tests
+To run the tests for gateway, follow the steps in [contributing](https://github.com/shuttle-hq/shuttle/blob/main/CONTRIBUTING.md) to set up your local environment. Then, from the root of the repository, run:
+
+```bash
+SHUTTLE_TESTS_RUNTIME_IMAGE=public.ecr.aws/shuttle-dev/deployer:latest SHUTTLE_TESTS_NETWORK=shuttle-dev_user-net cargo test --package shuttle-gateway --all-features -- --nocapture
+```
diff --git a/gateway/src/api/latest.rs b/gateway/src/api/latest.rs
@@ -5,10 +5,11 @@ use axum::body::{Body, BoxBody};
 use axum::extract::{Extension, Path};
 use axum::http::Request;
 use axum::response::Response;
-use axum::routing::{any, get};
+use axum::routing::{any, get, post};
 use axum::{Json as AxumJson, Router};
 use http::StatusCode;
 use serde::{Deserialize, Serialize};
+use shuttle_common::models::error::ErrorKind;
 use shuttle_common::models::{project, user};
 use tokio::sync::mpsc::Sender;
 use tower_http::trace::TraceLayer;
@@ -153,6 +154,16 @@ async fn get_status(Extension(sender): Extension<Sender<BoxedTask>>) -> Response
         .unwrap()
 }
 
+async fn revive_projects(
+    _: Admin,
+    Extension(service): Extension<Arc<GatewayService>>,
+    Extension(sender): Extension<Sender<BoxedTask>>,
+) -> Result<(), Error> {
+    crate::project::exec::revive(service, sender)
+        .await
+        .map_err(|_| Error::from_kind(ErrorKind::Internal))
+}
+
 pub fn make_api(service: Arc<GatewayService>, sender: Sender<BoxedTask>) -> Router<Body> {
     debug!("making api route");
 
@@ -167,6 +178,7 @@ pub fn make_api(service: Arc<GatewayService>, sender: Sender<BoxedTask>) -> Rout
         )
         .route("/users/:account_name", get(get_user).post(post_user))
         .route("/projects/:project/*any", any(route_project))
+        .route("/admin/revive", post(revive_projects))
         .layer(Extension(service))
         .layer(Extension(sender))
         .layer(
diff --git a/gateway/src/args.rs b/gateway/src/args.rs
@@ -19,7 +19,6 @@ pub struct Args {
 pub enum Commands {
     Start(StartArgs),
     Init(InitArgs),
-    Exec(ExecCmds),
 }
 
 #[derive(clap::Args, Debug, Clone)]
@@ -44,19 +43,6 @@ pub struct InitArgs {
     pub key: Option<Key>,
 }
 
-#[derive(clap::Args, Debug, Clone)]
-pub struct ExecCmds {
-    #[command(flatten)]
-    pub context: ContextArgs,
-    #[command(subcommand)]
-    pub command: ExecCmd,
-}
-
-#[derive(Subcommand, Debug, Clone)]
-pub enum ExecCmd {
-    Revive,
-}
-
 #[derive(clap::Args, Debug, Clone)]
 pub struct ContextArgs {
     /// Default image to deploy user runtimes into
diff --git a/gateway/src/main.rs b/gateway/src/main.rs
@@ -1,9 +1,8 @@
 use clap::Parser;
 use futures::prelude::*;
 use opentelemetry::global;
-use shuttle_gateway::args::{Args, Commands, ExecCmd, ExecCmds, InitArgs};
+use shuttle_gateway::args::{Args, Commands, InitArgs};
 use shuttle_gateway::auth::Key;
-use shuttle_gateway::project;
 use shuttle_gateway::proxy::make_proxy;
 use shuttle_gateway::service::{GatewayService, MIGRATIONS};
 use shuttle_gateway::task;
@@ -60,7 +59,6 @@ async fn main() -> io::Result<()> {
     match args.command {
         Commands::Start(start_args) => start(db, start_args).await,
         Commands::Init(init_args) => init(db, init_args).await,
-        Commands::Exec(exec_cmd) => exec(db, exec_cmd).await,
     }
 }
 
@@ -160,15 +158,3 @@ async fn init(db: SqlitePool, args: InitArgs) -> io::Result<()> {
     println!("`{}` created as super user with key: {key}", args.name);
     Ok(())
 }
-
-async fn exec(db: SqlitePool, exec_cmd: ExecCmds) -> io::Result<()> {
-    let gateway = GatewayService::init(exec_cmd.context.clone(), db).await;
-
-    match exec_cmd.command {
-        ExecCmd::Revive => project::exec::revive(gateway)
-            .await
-            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?,
-    };
-
-    Ok(())
-}
diff --git a/gateway/src/project.rs b/gateway/src/project.rs
@@ -838,21 +838,23 @@ where
 }
 
 pub mod exec {
+
     use std::sync::Arc;
 
     use bollard::service::ContainerState;
+    use tokio::sync::mpsc::Sender;
 
     use crate::{
         service::GatewayService,
-        task::{self, TaskResult},
+        task::{self, BoxedTask, TaskResult},
     };
 
     use super::*;
 
-    pub async fn revive(gateway: GatewayService) -> Result<(), ProjectError> {
-        let mut mutations = Vec::new();
-        let gateway = Arc::new(gateway);
-
+    pub async fn revive(
+        gateway: Arc<GatewayService>,
+        sender: Sender<BoxedTask>,
+    ) -> Result<(), ProjectError> {
         for (project_name, account_name) in gateway
             .iter_projects()
             .await
@@ -873,30 +875,24 @@ pub mod exec {
                             ..
                         }) = container.state
                         {
-                            mutations.push((
-                                project_name.clone(),
-                                gateway
-                                    .new_task()
-                                    .project(project_name)
-                                    .account(account_name)
-                                    .and_then(task::run(|ctx| async move {
-                                        TaskResult::Done(Project::Stopped(ProjectStopped {
-                                            container: ctx.state.container().unwrap(),
-                                        }))
+                            debug!("{} will be revived", project_name.clone());
+                            _ = gateway
+                                .new_task()
+                                .project(project_name)
+                                .account(account_name)
+                                .and_then(task::run(|ctx| async move {
+                                    TaskResult::Done(Project::Stopped(ProjectStopped {
+                                        container: ctx.state.container().unwrap(),
                                     }))
-                                    .build(),
-                            ));
+                                }))
+                                .send(&sender)
+                                .await;
                         }
                     }
                 }
             }
         }
 
-        for (project_name, mut work) in mutations {
-            debug!(?project_name, "project will be revived");
-            while let TaskResult::Pending(_) = work.poll(()).await {}
-        }
-
         Ok(())
     }
 }
