feat: recording of provisioned resources

This returns provisioned resources from runtime so that they can be
recorded by deployer. Closes #ENG-252.

common/Cargo.toml

@@ -31,7 +31,7 @@ reqwest = { version = "0.11.13", optional = true }
 rmp-serde = { version = "1.1.1", optional = true }
 rustrict = { version = "0.5.5", optional = true }
 serde = { workspace = true, features = ["derive", "std"] }
-serde_json = { workspace = true, optional = true }
+serde_json = { workspace = true }
 strum = { workspace = true, features = ["derive"], optional = true }
 thiserror = { workspace = true, optional = true }
 tonic = { version = "0.8.3", optional = true }
@@ -47,10 +47,10 @@ uuid = { workspace = true, features = ["v4", "serde"], optional = true }
 backend = ["async-trait", "axum/matched-path", "claims", "hyper/client", "opentelemetry-otlp", "thiserror", "tower-http", "tracing-subscriber/env-filter", "tracing-subscriber/fmt", "ttl_cache"]
 claims = ["bytes", "chrono/clock", "headers", "http", "http-body", "jsonwebtoken", "opentelemetry", "opentelemetry-http", "pin-project", "tower", "tracing", "tracing-opentelemetry"]
 display = ["chrono/clock", "comfy-table", "crossterm"]
-error = ["prost-types", "serde_json", "thiserror", "uuid"]
-models = ["anyhow", "async-trait", "display", "http", "reqwest", "serde_json", "service"]
+error = ["prost-types", "thiserror", "uuid"]
+models = ["anyhow", "async-trait", "display", "http", "reqwest", "service"]
 service = ["chrono/serde", "once_cell", "rustrict", "serde/derive", "strum", "uuid"]
-tracing = ["serde_json"]
+tracing = []
 wasm = ["chrono/clock", "http-serde", "http", "rmp-serde", "tracing", "tracing-subscriber"]
 
 [dev-dependencies]
@@ -59,7 +59,6 @@ base64 = "0.13.1"
 cap-std = "1.0.2"
 hyper = { workspace = true }
 ring = { workspace = true }
-serde_json = { workspace = true }
 tokio = { version = "1.22.0", features = ["macros", "rt-multi-thread"] }
 tower = { workspace = true, features = ["util"] }
 tracing-fluent-assertions = "0.3.0"

common/src/database.rs

@@ -3,14 +3,14 @@ use std::fmt::Display;
 use serde::{Deserialize, Serialize};
 use strum::Display;
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 pub enum Type {
     AwsRds(AwsRdsEngine),
     Shared(SharedEngine),
 }
 
-#[derive(Clone, Debug, Deserialize, Display, Serialize)]
+#[derive(Clone, Debug, Deserialize, Display, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 #[strum(serialize_all = "lowercase")]
 pub enum AwsRdsEngine {
@@ -19,7 +19,7 @@ pub enum AwsRdsEngine {
     MariaDB,
 }
 
-#[derive(Clone, Debug, Deserialize, Display, Serialize)]
+#[derive(Clone, Debug, Deserialize, Display, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 #[strum(serialize_all = "lowercase")]
 pub enum SharedEngine {

common/src/lib.rs

@@ -12,6 +12,7 @@ pub mod log;
 pub mod models;
 #[cfg(feature = "service")]
 pub mod project;
+pub mod resource;
 #[cfg(feature = "service")]
 pub mod storage_manager;
 #[cfg(feature = "tracing")]

common/src/models/mod.rs

@@ -1,7 +1,6 @@
 pub mod deployment;
 pub mod error;
 pub mod project;
-pub mod resource;
 pub mod secret;
 pub mod service;
 pub mod stats;

common/src/models/service.rs

@@ -1,6 +1,6 @@
 use crate::{
-    models::{deployment, resource, resource::ResourceInfo, secret},
-    DatabaseReadyInfo,
+    models::{deployment, secret},
+    resource::{self, ResourceInfo},
 };
 
 use comfy_table::{
@@ -34,12 +34,6 @@ pub struct Summary {
     pub uri: String,
 }
 
-impl ResourceInfo for DatabaseReadyInfo {
-    fn connection_string_public(&self) -> String {
-        self.connection_string_public()
-    }
-}
-
 impl Display for Detailed {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let deploys = get_deployments_table(&self.deployments, &self.name);

common/src/models/resource.rs → common/src/resource.rs

@@ -2,13 +2,11 @@ use std::fmt::Display;
 
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use uuid::Uuid;
 
 use crate::{database, DatabaseReadyInfo};
 
-#[derive(Deserialize, Serialize)]
+#[derive(Clone, Deserialize, Serialize)]
 pub struct Response {
-    pub service_id: Uuid,
     pub r#type: Type,
     pub data: Value,
 }
@@ -17,9 +15,22 @@ pub struct Response {
 pub trait ResourceInfo {
     /// String to connect to this resource from a public location
     fn connection_string_public(&self) -> String;
+
+    /// String to connect to this resource from within shuttle
+    fn connection_string_private(&self) -> String;
+}
+
+impl ResourceInfo for DatabaseReadyInfo {
+    fn connection_string_public(&self) -> String {
+        self.connection_string_public()
+    }
+
+    fn connection_string_private(&self) -> String {
+        self.connection_string_private()
+    }
 }
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]
 pub enum Type {
     Database(database::Type),
@@ -33,6 +44,10 @@ impl Response {
             }
         }
     }
+
+    pub fn into_bytes(self) -> Vec<u8> {
+        serde_json::to_vec(&self).expect("to turn resource into a vec")
+    }
 }
 
 impl Display for Type {

deployer/src/deployment/deploy_layer.rs

@@ -310,7 +310,10 @@ mod tests {
         time::Duration,
     };
 
-    use crate::{persistence::DeploymentUpdater, RuntimeManager};
+    use crate::{
+        persistence::{DeploymentUpdater, Resource, ResourceManager},
+        RuntimeManager,
+    };
     use async_trait::async_trait;
     use axum::body::Bytes;
     use ctor::ctor;
@@ -553,6 +556,21 @@ mod tests {
         }
     }
 
+    #[derive(Clone)]
+    struct StubResourceManager;
+
+    #[async_trait::async_trait]
+    impl ResourceManager for StubResourceManager {
+        type Err = std::io::Error;
+
+        async fn insert_resource(&self, _resource: &Resource) -> Result<(), Self::Err> {
+            Ok(())
+        }
+        async fn get_resources(&self, _service_id: &Uuid) -> Result<Vec<Resource>, Self::Err> {
+            Ok(Vec::new())
+        }
+    }
+
     async fn test_states(id: &Uuid, expected_states: Vec<StateLog>) {
         loop {
             let states = RECORDER.lock().unwrap().get_deployment_states(id);
@@ -861,6 +879,7 @@ mod tests {
             .active_deployment_getter(StubActiveDeploymentGetter)
             .artifacts_path(PathBuf::from("/tmp"))
             .secret_getter(StubSecretGetter)
+            .resource_manager(StubResourceManager)
             .runtime(get_runtime_manager())
             .deployment_updater(StubDeploymentUpdater)
             .queue_client(StubBuildQueueClient)

deployer/src/deployment/mod.rs

@@ -12,7 +12,7 @@ use tracing::{instrument, Span};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 
 use crate::{
-    persistence::{DeploymentUpdater, SecretGetter, SecretRecorder, State},
+    persistence::{DeploymentUpdater, ResourceManager, SecretGetter, SecretRecorder, State},
     RuntimeManager,
 };
 use tokio::sync::{mpsc, Mutex};
@@ -23,24 +23,26 @@ use self::{deploy_layer::LogRecorder, gateway_client::BuildQueueClient};
 const QUEUE_BUFFER_SIZE: usize = 100;
 const RUN_BUFFER_SIZE: usize = 100;
 
-pub struct DeploymentManagerBuilder<LR, SR, ADG, DU, SG, QC> {
+pub struct DeploymentManagerBuilder<LR, SR, ADG, DU, SG, RM, QC> {
     build_log_recorder: Option<LR>,
     secret_recorder: Option<SR>,
     active_deployment_getter: Option<ADG>,
     artifacts_path: Option<PathBuf>,
     runtime_manager: Option<Arc<Mutex<RuntimeManager>>>,
     deployment_updater: Option<DU>,
     secret_getter: Option<SG>,
+    resource_manager: Option<RM>,
     queue_client: Option<QC>,
 }
 
-impl<LR, SR, ADG, DU, SG, QC> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, QC>
+impl<LR, SR, ADG, DU, SG, RM, QC> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, RM, QC>
 where
     LR: LogRecorder,
     SR: SecretRecorder,
     ADG: ActiveDeploymentsGetter,
     DU: DeploymentUpdater,
     SG: SecretGetter,
+    RM: ResourceManager,
     QC: BuildQueueClient,
 {
     pub fn build_log_recorder(mut self, build_log_recorder: LR) -> Self {
@@ -79,6 +81,12 @@ where
         self
     }
 
+    pub fn resource_manager(mut self, resource_manager: RM) -> Self {
+        self.resource_manager = Some(resource_manager);
+
+        self
+    }
+
     pub fn runtime(mut self, runtime_manager: Arc<Mutex<RuntimeManager>>) -> Self {
         self.runtime_manager = Some(runtime_manager);
 
@@ -110,6 +118,7 @@ where
             .deployment_updater
             .expect("a deployment updater to be set");
         let secret_getter = self.secret_getter.expect("a secret getter to be set");
+        let resource_manager = self.resource_manager.expect("a resource manager to be set");
 
         let (queue_send, queue_recv) = mpsc::channel(QUEUE_BUFFER_SIZE);
         let (run_send, run_recv) = mpsc::channel(RUN_BUFFER_SIZE);
@@ -132,6 +141,7 @@ where
             deployment_updater,
             active_deployment_getter,
             secret_getter,
+            resource_manager,
             storage_manager.clone(),
         ));
 
@@ -169,7 +179,8 @@ pub struct DeploymentManager {
 impl DeploymentManager {
     /// Create a new deployment manager. Manages one or more 'pipelines' for
     /// processing service building, loading, and deployment.
-    pub fn builder<LR, SR, ADG, DU, SG, QC>() -> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, QC> {
+    pub fn builder<LR, SR, ADG, DU, SG, RM, QC>(
+    ) -> DeploymentManagerBuilder<LR, SR, ADG, DU, SG, RM, QC> {
         DeploymentManagerBuilder {
             build_log_recorder: None,
             secret_recorder: None,
@@ -178,6 +189,7 @@ impl DeploymentManager {
             runtime_manager: None,
             deployment_updater: None,
             secret_getter: None,
+            resource_manager: None,
             queue_client: None,
         }
     }