feat(edda, forklift, pinga, rebaser, veritech): impl endpoint service

Author: sprutton1

bin/edda/src/main.rs

@@ -28,6 +28,8 @@ fn main() -> Result<()> {
 async fn async_main() -> Result<()> {
     let main_tracker = TaskTracker::new();
     let main_token = CancellationToken::new();
+    let endpoints_tracker = TaskTracker::new();
+    let endpoints_token = CancellationToken::new();
     let layer_db_tracker = TaskTracker::new();
     let layer_db_token = CancellationToken::new();
     let telemetry_tracker = TaskTracker::new();
@@ -70,6 +72,17 @@ async fn async_main() -> Result<()> {
     let config = load_config_with_provider(args, provider).await?;
     debug!(?config, "computed configuration");
 
+    let endpoints_server = if config.service_endpoints().enabled {
+        let endpoints = edda_server::DefaultServiceEndpoints::from_config("edda", &config)?;
+        Some(edda_server::EndpointsServer::new(
+            std::sync::Arc::new(endpoints),
+            config.service_endpoints().clone(),
+            endpoints_token.clone(),
+        ))
+    } else {
+        None
+    };
+
     let server = Server::from_config(
         config,
         main_token.clone(),
@@ -83,8 +96,17 @@ async fn async_main() -> Result<()> {
         server.run().await
     });
 
+    if let Some(endpoints_server) = endpoints_server {
+        endpoints_tracker.spawn(async move {
+            if let Err(err) = endpoints_server.run().await {
+                error!(error = ?err, "error running edda endpoints server");
+            }
+        });
+    }
+
     shutdown::graceful()
         .group(main_tracker, main_token)
+        .group(endpoints_tracker, endpoints_token)
         .group(layer_db_tracker, layer_db_token)
         .group(telemetry_tracker, telemetry_token)
         .telemetry_guard(telemetry_shutdown.into_future())

bin/forklift/src/main.rs

@@ -30,6 +30,8 @@ fn main() -> Result<()> {
 async fn async_main() -> Result<()> {
     let main_tracker = TaskTracker::new();
     let main_token = CancellationToken::new();
+    let endpoints_tracker = TaskTracker::new();
+    let endpoints_token = CancellationToken::new();
     let telemetry_tracker = TaskTracker::new();
     let telemetry_token = CancellationToken::new();
 
@@ -70,15 +72,35 @@ async fn async_main() -> Result<()> {
     let config = load_config_with_provider(args, provider).await?;
     debug!(?config, "computed configuration");
 
+    let endpoints_server = if config.service_endpoints().enabled {
+        let endpoints = forklift_server::DefaultServiceEndpoints::from_config("forklift", &config)?;
+        Some(forklift_server::EndpointsServer::new(
+            std::sync::Arc::new(endpoints),
+            config.service_endpoints().clone(),
+            endpoints_token.clone(),
+        ))
+    } else {
+        None
+    };
+
     let server = Server::from_config(config, main_token.clone()).await?;
 
     main_tracker.spawn(async move {
         info!("ready to receive messages");
         server.run().await
     });
 
+    if let Some(endpoints_server) = endpoints_server {
+        endpoints_tracker.spawn(async move {
+            if let Err(err) = endpoints_server.run().await {
+                error!(error = ?err, "error running forklift endpoints server");
+            }
+        });
+    }
+
     shutdown::graceful()
         .group(main_tracker, main_token)
+        .group(endpoints_tracker, endpoints_token)
         .group(telemetry_tracker, telemetry_token)
         .telemetry_guard(telemetry_shutdown.into_future())
         .timeout(GRACEFUL_SHUTDOWN_TIMEOUT)

bin/pinga/src/main.rs

@@ -30,6 +30,8 @@ fn main() -> Result<()> {
 async fn async_main() -> Result<()> {
     let main_tracker = TaskTracker::new();
     let main_token = CancellationToken::new();
+    let endpoints_tracker = TaskTracker::new();
+    let endpoints_token = CancellationToken::new();
     let layer_db_tracker = TaskTracker::new();
     let layer_db_token = CancellationToken::new();
     let telemetry_tracker = TaskTracker::new();
@@ -80,6 +82,18 @@ async fn async_main() -> Result<()> {
     let config = load_config_with_provider(args, provider).await?;
     debug!(?config, "computed configuration");
 
+    // Create optional HTTP endpoints server before Server takes ownership of config
+    let endpoints_server = if config.service_endpoints().enabled {
+        let endpoints = pinga_server::DefaultServiceEndpoints::from_config("pinga", &config)?;
+        Some(pinga_server::EndpointsServer::new(
+            std::sync::Arc::new(endpoints),
+            config.service_endpoints().clone(),
+            endpoints_token.clone(),
+        ))
+    } else {
+        None
+    };
+
     let server = Server::from_config(
         config,
         main_token.clone(),
@@ -93,8 +107,17 @@ async fn async_main() -> Result<()> {
         server.run().await
     });
 
+    if let Some(endpoints_server) = endpoints_server {
+        endpoints_tracker.spawn(async move {
+            if let Err(err) = endpoints_server.run().await {
+                error!(error = ?err, "error running pinga endpoints server");
+            }
+        });
+    }
+
     shutdown::graceful()
         .group(main_tracker, main_token)
+        .group(endpoints_tracker, endpoints_token)
         .group(layer_db_tracker, layer_db_token)
         .group(telemetry_tracker, telemetry_token)
         .telemetry_guard(telemetry_shutdown.into_future())

bin/rebaser/src/main.rs

@@ -30,6 +30,8 @@ fn main() -> Result<()> {
 async fn async_main() -> Result<()> {
     let main_tracker = TaskTracker::new();
     let main_token = CancellationToken::new();
+    let endpoints_tracker = TaskTracker::new();
+    let endpoints_token = CancellationToken::new();
     let layer_db_tracker = TaskTracker::new();
     let layer_db_token = CancellationToken::new();
     let telemetry_tracker = TaskTracker::new();
@@ -80,6 +82,17 @@ async fn async_main() -> Result<()> {
     let config = load_config_with_provider(args, provider).await?;
     debug!(?config, "computed configuration");
 
+    let endpoints_server = if config.service_endpoints().enabled {
+        let endpoints = rebaser_server::DefaultServiceEndpoints::from_config("rebaser", &config)?;
+        Some(rebaser_server::EndpointsServer::new(
+            std::sync::Arc::new(endpoints),
+            config.service_endpoints().clone(),
+            endpoints_token.clone(),
+        ))
+    } else {
+        None
+    };
+
     let server = Server::from_config(
         config,
         main_token.clone(),
@@ -93,8 +106,17 @@ async fn async_main() -> Result<()> {
         server.run().await
     });
 
+    if let Some(endpoints_server) = endpoints_server {
+        endpoints_tracker.spawn(async move {
+            if let Err(err) = endpoints_server.run().await {
+                error!(error = ?err, "error running rebaser endpoints server");
+            }
+        });
+    }
+
     shutdown::graceful()
         .group(main_tracker, main_token)
+        .group(endpoints_tracker, endpoints_token)
         .group(layer_db_tracker, layer_db_token)
         .group(telemetry_tracker, telemetry_token)
         .telemetry_guard(telemetry_shutdown.into_future())

bin/veritech/src/main.rs

@@ -29,6 +29,8 @@ fn main() -> Result<()> {
 async fn async_main(args: args::Args) -> Result<()> {
     let main_tracker = TaskTracker::new();
     let main_token = CancellationToken::new();
+    let endpoints_tracker = TaskTracker::new();
+    let endpoints_token = CancellationToken::new();
     let telemetry_tracker = TaskTracker::new();
     let telemetry_token = CancellationToken::new();
 
@@ -71,6 +73,17 @@ async fn async_main(args: args::Args) -> Result<()> {
     let config = load_config_with_provider(args, provider).await?;
     debug!(?config, "computed configuration");
 
+    let endpoints_server = if config.service_endpoints().enabled {
+        let endpoints = veritech_server::DefaultServiceEndpoints::from_config("veritech", &config)?;
+        Some(veritech_server::EndpointsServer::new(
+            std::sync::Arc::new(endpoints),
+            config.service_endpoints().clone(),
+            endpoints_token.clone(),
+        ))
+    } else {
+        None
+    };
+
     let (server, maybe_heartbeat_app) = Server::from_config(config, main_token.clone()).await?;
 
     if let Some(mut heartbeat_app) = maybe_heartbeat_app {
@@ -81,8 +94,17 @@ async fn async_main(args: args::Args) -> Result<()> {
         server.run().await
     });
 
+    if let Some(endpoints_server) = endpoints_server {
+        endpoints_tracker.spawn(async move {
+            if let Err(err) = endpoints_server.run().await {
+                error!(error = ?err, "error running veritech endpoints server");
+            }
+        });
+    }
+
     shutdown::graceful()
         .group(main_tracker, main_token)
+        .group(endpoints_tracker, endpoints_token)
         .group(telemetry_tracker, telemetry_token)
         .telemetry_guard(telemetry_shutdown.into_future())
         .timeout(graceful_shutdown_timeout)

lib/edda-server/BUCK

@@ -19,6 +19,7 @@ rust_library(
         "//lib/si-frontend-mv-types-rs:si-frontend-mv-types",
         "//lib/si-id:si-id",
         "//lib/si-layer-cache:si-layer-cache",
+        "//lib/si-service-endpoints:si-service-endpoints",
         "//lib/si-settings:si-settings",
         "//lib/si-std:si-std",
         "//lib/si-tls:si-tls",

lib/edda-server/src/config.rs

@@ -18,6 +18,7 @@ use si_crypto::{
 use si_data_nats::NatsConfig;
 use si_data_pg::PgPoolConfig;
 use si_layer_cache::db::LayerDbConfig;
+use si_service_endpoints::ServiceEndpointsConfig;
 pub(crate) use si_settings::StandardConfig;
 pub use si_settings::StandardConfigFile;
 use si_std::CanonicalFileError;
@@ -56,7 +57,7 @@ impl ConfigError {
 type Result<T> = std::result::Result<T, ConfigError>;
 
 /// The config for the forklift server.
-#[derive(Debug, Builder)]
+#[derive(Debug, Builder, Serialize)]
 pub struct Config {
     #[builder(default = "random_instance_id()")]
     instance_id: String,
@@ -87,6 +88,9 @@ pub struct Config {
 
     #[builder(default = "default_quiescent_period()")]
     quiescent_period: Duration,
+
+    #[builder(default = "default_service_endpoints_config()")]
+    service_endpoints: ServiceEndpointsConfig,
 }
 
 impl StandardConfig for Config {
@@ -152,6 +156,12 @@ impl Config {
     pub fn quiescent_period(&self) -> Duration {
         self.quiescent_period
     }
+
+    /// Gets a reference to the config's service endpoints configuration.
+    #[must_use]
+    pub fn service_endpoints(&self) -> &ServiceEndpointsConfig {
+        &self.service_endpoints
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -179,6 +189,8 @@ pub struct ConfigFile {
     layer_db_config: LayerDbConfig,
     #[serde(default = "default_quiescent_period_secs")]
     quiescent_period_secs: u64,
+    #[serde(default = "default_service_endpoints_config")]
+    service_endpoints: ServiceEndpointsConfig,
 }
 
 impl Default for ConfigFile {
@@ -194,6 +206,7 @@ impl Default for ConfigFile {
             symmetric_crypto_service: default_symmetric_crypto_config(),
             layer_db_config: default_layer_db_config(),
             quiescent_period_secs: default_quiescent_period_secs(),
+            service_endpoints: default_service_endpoints_config(),
         }
     }
 }
@@ -219,6 +232,7 @@ impl TryFrom<ConfigFile> for Config {
         config.parallel_build_limit(value.edda_parallel_build_limit);
         config.instance_id(value.instance_id);
         config.quiescent_period(Duration::from_secs(value.quiescent_period_secs));
+        config.service_endpoints(value.service_endpoints);
         config.build().map_err(Into::into)
     }
 }
@@ -259,6 +273,10 @@ fn default_quiescent_period_secs() -> u64 {
     DEFAULT_QUIESCENT_PERIOD_SECS
 }
 
+fn default_service_endpoints_config() -> ServiceEndpointsConfig {
+    ServiceEndpointsConfig::new(0)
+}
+
 #[allow(clippy::disallowed_methods)] // Used to determine if running in development
 pub fn detect_and_configure_development(config: &mut ConfigFile) -> Result<()> {
     if env::var("BUCK_RUN_BUILD_ID").is_ok() || env::var("BUCK_BUILD_ID").is_ok() {

lib/edda-server/src/lib.rs

@@ -11,6 +11,11 @@ mod deployment_processor_task;
 mod handlers;
 mod local_message;
 mod server;
+pub use si_service_endpoints::{
+    DefaultServiceEndpoints,
+    ServiceEndpointsConfig,
+    server::EndpointsServer,
+};
 pub use si_settings::{
     ConfigMap,
     ParameterProvider,

lib/forklift-server/BUCK

@@ -12,6 +12,7 @@ rust_library(
         "//lib/naxum:naxum",
         "//lib/si-data-nats:si-data-nats",
         "//lib/si-events-rs:si-events",
+        "//lib/si-service-endpoints:si-service-endpoints",
         "//lib/si-settings:si-settings",
         "//lib/si-std:si-std",
         "//lib/si-tls:si-tls",