feat: create a control plane interface (part 1) (#436)

* feat: add proto for runtimes

* refactor: legacy move main to lib

* feat: impl Runtime server for legacy

* feat: impl Runtime server for next

Author: chesedo

runtimes/Cargo.toml

@@ -2,5 +2,6 @@
 members = [
   "legacy",
   "next",
+  "proto",
   "wasm"
 ]

runtimes/legacy/Cargo.toml

@@ -6,6 +6,7 @@ publish = false
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+anyhow = "1.0.62"
 async-trait = "0.1.58"
 clap ={ version = "4.0.18", features = ["derive"] }
 thiserror = "1.0.37"
@@ -18,6 +19,10 @@ tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 version = "0.7.0"
 path = "../../common"
 
+[dependencies.shuttle-runtime-proto]
+version = "0.1.0"
+path = "../proto"
+
 [dependencies.shuttle-service]
 version = "0.7.0"
 default-features = false

runtimes/legacy/README.md

@@ -2,8 +2,15 @@
 
 Load and run an .so library that implements `shuttle_service::Service`. 
 
-To load and run, pass the path to the .so file to load as an argument to the shuttle-next binary:
+To test, first start this binary using:
 
 ```bash
-cargo run -- -f "src/libhello_world.so"
+cargo run --
+```
+
+Then in another shell, load a `.so` file and start it up:
+
+``` bash
+grpcurl -plaintext -import-path ../proto -proto runtime.proto -d '{"service_name": "Tonic", "path": "../../examples/rocket/hello-world/target/debug/libhello_world.so"}' localhost:8000 runtime.Runtime/load
+grpcurl -plaintext -import-path ../proto -proto runtime.proto -d '{"service_name": "Tonic"}' localhost:8000 runtime.Runtime/start
 ```

runtimes/legacy/src/args.rs

@@ -3,10 +3,6 @@ use tonic::transport::Endpoint;
 
 #[derive(Parser, Debug)]
 pub struct Args {
-    /// Uri to the `.so` file to load
-    #[arg(long, short)]
-    pub file_path: String,
-
     /// Address to reach provisioner at
     #[clap(long, default_value = "localhost:5000")]
     pub provisioner_address: Endpoint,

runtimes/legacy/src/error.rs

@@ -7,6 +7,8 @@ pub enum Error {
     Load(#[from] LoaderError),
     #[error("Run error: {0}")]
     Run(#[from] shuttle_service::Error),
+    #[error("Start error: {0}")]
+    Start(#[from] shuttle_service::error::CustomError),
 }
 
 pub type Result<T> = std::result::Result<T, Error>;

runtimes/legacy/src/lib.rs

@@ -1,2 +1,151 @@
+use std::{
+    collections::BTreeMap,
+    net::{Ipv4Addr, SocketAddr},
+    path::PathBuf,
+    str::FromStr,
+    sync::Mutex,
+};
+
+use anyhow::anyhow;
+use async_trait::async_trait;
+use shuttle_common::{database, LogItem};
+use shuttle_runtime_proto::runtime::{
+    runtime_server::Runtime, LoadRequest, LoadResponse, StartRequest, StartResponse,
+};
+use shuttle_service::{
+    loader::{LoadedService, Loader},
+    Factory, Logger, ServiceName,
+};
+use tokio::sync::mpsc::{self, UnboundedReceiver};
+use tonic::{Request, Response, Status};
+use tracing::{info, instrument, trace};
+
 pub mod args;
 pub mod error;
+
+pub struct Legacy {
+    // Mutexes are for interior mutability
+    so_path: Mutex<Option<PathBuf>>,
+    port: Mutex<Option<u16>>,
+}
+
+impl Legacy {
+    pub fn new() -> Self {
+        Self {
+            so_path: Mutex::new(None),
+            port: Mutex::new(None),
+        }
+    }
+}
+
+#[async_trait]
+impl Runtime for Legacy {
+    async fn load(&self, request: Request<LoadRequest>) -> Result<Response<LoadResponse>, Status> {
+        let so_path = request.into_inner().path;
+        trace!(so_path, "loading");
+
+        let so_path = PathBuf::from(so_path);
+        *self.so_path.lock().unwrap() = Some(so_path);
+
+        let message = LoadResponse { success: true };
+        Ok(Response::new(message))
+    }
+
+    async fn start(
+        &self,
+        _request: Request<StartRequest>,
+    ) -> Result<Response<StartResponse>, Status> {
+        let port = 8001;
+        let address = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), port);
+        let mut factory = DummyFactory::new();
+        let (logger, _rx) = get_logger();
+        let so_path = self
+            .so_path
+            .lock()
+            .unwrap()
+            .as_ref()
+            .ok_or_else(|| -> error::Error {
+                error::Error::Start(anyhow!("trying to start a service that was not loaded"))
+            })
+            .map_err(|err| Status::from_error(Box::new(err)))?
+            .clone();
+
+        trace!(%address, "starting");
+        let service = load_service(address, so_path, &mut factory, logger)
+            .await
+            .unwrap();
+
+        _ = tokio::spawn(run(service, address));
+
+        *self.port.lock().unwrap() = Some(port);
+
+        let message = StartResponse {
+            success: true,
+            port: Some(port as u32),
+        };
+
+        Ok(Response::new(message))
+    }
+}
+
+#[instrument(skip(service))]
+async fn run(service: LoadedService, addr: SocketAddr) {
+    let (handle, library) = service;
+
+    info!("starting deployment on {}", addr);
+    handle.await.unwrap().unwrap();
+
+    tokio::spawn(async move {
+        trace!("closing .so file");
+        library.close().unwrap();
+    });
+}
+
+#[instrument(skip(addr, so_path, factory, logger))]
+async fn load_service(
+    addr: SocketAddr,
+    so_path: PathBuf,
+    factory: &mut dyn Factory,
+    logger: Logger,
+) -> error::Result<LoadedService> {
+    let loader = Loader::from_so_file(so_path)?;
+
+    Ok(loader.load(factory, addr, logger).await?)
+}
+
+struct DummyFactory {
+    service_name: ServiceName,
+}
+
+impl DummyFactory {
+    fn new() -> Self {
+        Self {
+            service_name: ServiceName::from_str("legacy").unwrap(),
+        }
+    }
+}
+
+#[async_trait]
+impl Factory for DummyFactory {
+    fn get_service_name(&self) -> ServiceName {
+        self.service_name.clone()
+    }
+
+    async fn get_db_connection_string(
+        &mut self,
+        _: database::Type,
+    ) -> Result<String, shuttle_service::Error> {
+        todo!()
+    }
+
+    async fn get_secrets(&mut self) -> Result<BTreeMap<String, String>, shuttle_service::Error> {
+        todo!()
+    }
+}
+
+fn get_logger() -> (Logger, UnboundedReceiver<LogItem>) {
+    let (tx, rx) = mpsc::unbounded_channel();
+    let logger = Logger::new(tx, Default::default());
+
+    (logger, rx)
+}

runtimes/legacy/src/main.rs

@@ -1,15 +1,10 @@
-use std::{collections::BTreeMap, net::SocketAddr, path::PathBuf, str::FromStr};
+use std::net::{Ipv4Addr, SocketAddr};
 
-use async_trait::async_trait;
 use clap::Parser;
-use shuttle_common::{database, LogItem};
-use shuttle_legacy::args::Args;
-use shuttle_service::{
-    loader::{LoadedService, Loader},
-    Factory, Logger, ServiceName,
-};
-use tokio::sync::mpsc::{self, UnboundedReceiver};
-use tracing::{info, instrument, trace};
+use shuttle_legacy::{args::Args, Legacy};
+use shuttle_runtime_proto::runtime::runtime_server::RuntimeServer;
+use tonic::transport::Server;
+use tracing::trace;
 use tracing_subscriber::{fmt, prelude::*, EnvFilter};
 
 #[tokio::main(flavor = "multi_thread")]
@@ -28,76 +23,11 @@ async fn main() {
 
     trace!(args = ?args, "parsed args");
 
-    let address: SocketAddr = "127.0.0.1:8000".parse().unwrap();
-    let mut factory = DummyFactory::new();
-    let (logger, _rx) = get_logger();
-    let so_path = PathBuf::from(args.file_path.as_str());
-
-    let service = load_service(address, so_path, &mut factory, logger)
+    let addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 8000);
+    let legacy = Legacy::new();
+    Server::builder()
+        .add_service(RuntimeServer::new(legacy))
+        .serve(addr)
         .await
         .unwrap();
-
-    _ = tokio::spawn(run(service, address)).await;
-}
-
-#[instrument(skip(service))]
-async fn run(service: LoadedService, addr: SocketAddr) {
-    let (handle, library) = service;
-
-    info!("starting deployment on {}", addr);
-    handle.await.unwrap().unwrap();
-
-    tokio::spawn(async move {
-        trace!("closing .so file");
-        library.close().unwrap();
-    });
-}
-
-#[instrument(skip(addr, so_path, factory, logger))]
-async fn load_service(
-    addr: SocketAddr,
-    so_path: PathBuf,
-    factory: &mut dyn Factory,
-    logger: Logger,
-) -> shuttle_legacy::error::Result<LoadedService> {
-    let loader = Loader::from_so_file(so_path)?;
-
-    Ok(loader.load(factory, addr, logger).await?)
-}
-
-struct DummyFactory {
-    service_name: ServiceName,
-}
-
-impl DummyFactory {
-    fn new() -> Self {
-        Self {
-            service_name: ServiceName::from_str("next").unwrap(),
-        }
-    }
-}
-
-#[async_trait]
-impl Factory for DummyFactory {
-    fn get_service_name(&self) -> ServiceName {
-        self.service_name.clone()
-    }
-
-    async fn get_db_connection_string(
-        &mut self,
-        _: database::Type,
-    ) -> Result<String, shuttle_service::Error> {
-        todo!()
-    }
-
-    async fn get_secrets(&mut self) -> Result<BTreeMap<String, String>, shuttle_service::Error> {
-        todo!()
-    }
-}
-
-fn get_logger() -> (Logger, UnboundedReceiver<LogItem>) {
-    let (tx, rx) = mpsc::unbounded_channel();
-    let logger = Logger::new(tx, Default::default());
-
-    (logger, rx)
 }

runtimes/next/Cargo.toml

@@ -13,10 +13,16 @@ async-trait = "0.1.58"
 clap ={ version = "4.0.18", features = ["derive"] }
 tokio = { version = "1.20.1", features = [ "full" ] }
 tonic = "0.8.0"
+tracing = "0.1.37"
+tracing-subscriber = { version = "0.3.16", features = ["env-filter"] }
 
 cap-std = "*"
 wasmtime = "*"
 wasmtime-wasi = "*"
 wasi-common = "*"
 
 serenity = { version = "0.11.5", default-features = false, features = ["client", "gateway", "rustls_backend", "model"] }
+
+[dependencies.shuttle-runtime-proto]
+version = "0.1.0"
+path = "../proto"

runtimes/next/README.md

@@ -4,7 +4,14 @@
 
 ```bash
 $ cd ..; make wasm
-$ DISCORD_TOKEN=xxx BOT_SRC=bot.wasm cargo run
+$ DISCORD_TOKEN=xxx cargo run
+```
+
+In another terminal:
+
+``` bash
+grpcurl -plaintext -import-path ../proto -proto runtime.proto -d '{"service_name": "Tonic", "path": "bot.wasm"}' localhost:8000 runtime.Runtime/load
+grpcurl -plaintext -import-path ../proto -proto runtime.proto -d '{"service_name": "Tonic"}' localhost:8000 runtime.Runtime/start
 ```
 
 ## Running the tests