[wip] start pulling kubernetes logic out of environmentd's kubernetes orchestrator into orchestratord

Cargo.toml

@@ -53,6 +53,7 @@ members = [
     "src/npm",
     "src/orchestrator",
     "src/orchestrator-kubernetes",
+    "src/orchestrator-external",
     "src/orchestrator-process",
     "src/orchestrator-tracing",
     "src/orchestratord",
@@ -170,6 +171,7 @@ default-members = [
     "src/npm",
     "src/orchestrator",
     "src/orchestrator-kubernetes",
+    "src/orchestrator-external",
     "src/orchestrator-process",
     "src/orchestrator-tracing",
     "src/orchestratord",

src/catalog/src/builtin.rs

@@ -9142,6 +9142,20 @@ pub const MZ_ANALYTICS_CLUSTER: BuiltinCluster = BuiltinCluster {
     replication_factor: 0,
 };
 
+pub static MZ_EXTERNAL_ORCHESTRATOR_SERVICES: LazyLock<BuiltinTable> =
+    LazyLock::new(|| BuiltinTable {
+        name: "mz_external_orchestrator_services",
+        schema: MZ_INTERNAL_SCHEMA,
+        oid: oid::TABLE_MZ_EXTERNAL_ORCHESTRATOR_SERVICES_OID,
+        desc: RelationDesc::builder()
+            .with_column("id", ScalarType::String.nullable(false))
+            .with_column("state", ScalarType::String.nullable(false))
+            .with_key(vec![0])
+            .finish(),
+        is_retained_metrics_object: false,
+        access: vec![PUBLIC_SELECT],
+    });
+
 /// List of all builtin objects sorted topologically by dependency.
 pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = LazyLock::new(|| {
     let mut builtins = vec![

src/environmentd/Cargo.toml

@@ -55,6 +55,7 @@ mz-http-util = { path = "../http-util" }
 mz-interchange = { path = "../interchange" }
 mz-metrics = { path = "../metrics" }
 mz-orchestrator = { path = "../orchestrator" }
+mz-orchestrator-external = { path = "../orchestrator-external" }
 mz-orchestrator-kubernetes = { path = "../orchestrator-kubernetes" }
 mz-orchestrator-process = { path = "../orchestrator-process" }
 mz-orchestrator-tracing = { path = "../orchestrator-tracing" }

src/environmentd/src/environmentd/main.rs

@@ -39,6 +39,7 @@ use mz_cloud_resources::{AwsExternalIdPrefix, CloudResourceController};
 use mz_controller::ControllerConfig;
 use mz_frontegg_auth::{Authenticator, FronteggCliArgs};
 use mz_orchestrator::Orchestrator;
+use mz_orchestrator_external::{ExternalOrchestrator, ExternalOrchestratorConfig};
 use mz_orchestrator_kubernetes::{
     KubernetesImagePullPolicy, KubernetesOrchestrator, KubernetesOrchestratorConfig,
 };
@@ -317,6 +318,10 @@ pub struct Args {
     /// production, only testing.
     #[structopt(long, env = "ORCHESTRATOR_KUBERNETES_COVERAGE")]
     orchestrator_kubernetes_coverage: bool,
+    /// Template to use to generate the hostname for cluster replicas. Will
+    /// replace {name} with the replica name and {id} with the replica id.
+    #[structopt(long, env = "ORCHESTRATOR_EXTERNAL_HOSTNAME_TEMPLATE")]
+    orchestrator_external_hostname_template: Option<String>,
     /// The secrets controller implementation to use.
     #[structopt(
         long,
@@ -599,6 +604,7 @@ pub struct Args {
 enum OrchestratorKind {
     Kubernetes,
     Process,
+    External,
 }
 
 // TODO [Alex Hunt] move this to a shared function that can be imported by the
@@ -851,6 +857,45 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
             };
             (orchestrator, secrets_controller, None)
         }
+        OrchestratorKind::External => {
+            if args.orchestrator_process_scratch_directory.is_some() {
+                bail!(
+                    "--orchestrator-process-scratch-directory is \
+                      not currently usable with the external orchestrator"
+                );
+            }
+            let Some(hostname_template) = args.orchestrator_external_hostname_template else {
+                bail!("--orchestrator-external-hostname-template is required when using Orchestrator::External")
+            };
+
+            let orchestrator = Arc::new(ExternalOrchestrator::new(ExternalOrchestratorConfig {
+                hostname_template,
+            }));
+            let secrets_controller: Arc<dyn SecretsController> = match args.secrets_controller {
+                SecretsControllerKind::Kubernetes => bail!(
+                    "SecretsControllerKind::Kubernetes is not yet implemented for Orchestrator::External."
+                ),
+                SecretsControllerKind::AwsSecretsManager => {
+                    Arc::new(
+                        runtime.block_on(AwsSecretsController::new(
+                            // TODO [Alex Hunt] move this to a shared function that can be imported by the
+                            // region-controller.
+                            &aws_secrets_controller_prefix(&args.environment_id),
+                            &aws_secrets_controller_key_alias(&args.environment_id),
+                            args.aws_secrets_controller_tags
+                                .into_iter()
+                                .map(|tag| (tag.key, tag.value))
+                                .collect(),
+                        )),
+                    )
+                }
+                SecretsControllerKind::LocalFile => bail!(
+                    "SecretsControllerKind::LocalFile is not compatible with Orchestrator::External."
+                ),
+            };
+            // TODO: secrets and vpc endpoints need to be implemented here
+            (orchestrator, secrets_controller, None)
+        }
     };
     drop(entered);
     let cloud_resource_reader = cloud_resource_controller.as_ref().map(|c| c.reader());

src/environmentd/src/lib.rs

@@ -36,6 +36,7 @@ use mz_catalog::durable::BootstrapArgs;
 use mz_cloud_resources::CloudResourceController;
 use mz_controller::ControllerConfig;
 use mz_frontegg_auth::Authenticator as FronteggAuthentication;
+use mz_orchestrator_external::ExternalOrchestrator;
 use mz_ore::future::OreFutureExt;
 use mz_ore::metrics::MetricsRegistry;
 use mz_ore::now::NowFn;
@@ -612,6 +613,8 @@ impl Listeners {
             connection_limiter.update_superuser_reserved(superuser_reserved);
         });
 
+        let orchestrator = Arc::clone(&config.controller.orchestrator);
+
         let webhook_concurrency_limit = WebhookConcurrencyLimiter::default();
         let (adapter_handle, adapter_client) = mz_adapter::serve(mz_adapter::Config {
             connection_context: config.controller.connection_context.clone(),
@@ -668,6 +671,13 @@ impl Listeners {
         let serve_postamble_start = Instant::now();
         info!("startup: envd serve: postamble beginning");
 
+        // Install an adapter client in the orchestrator
+        if let Some(external_orchestrator) =
+            orchestrator.as_any().downcast_ref::<ExternalOrchestrator>()
+        {
+            external_orchestrator.set_adapter_client(adapter_client.clone());
+        }
+
         // Install an adapter client in the internal HTTP server.
         internal_http_adapter_client_tx
             .send(adapter_client.clone())

src/orchestrator-external/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "mz-orchestrator-external"
+description = "Service orchestration via an external orchestrator."
+version = "0.0.0"
+edition.workspace = true
+rust-version.workspace = true
+publish = false
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow = "1.0.66"
+async-trait = "0.1.68"
+futures = "0.3.25"
+mz-adapter = { path = "../adapter" }
+mz-orchestrator = { path = "../orchestrator" }
+mz-ore = { path = "../ore" }
+mz-repr = { path = "../repr" }
+mz-sql = { path = "../sql" }
+tokio = "1.32.0"
+uuid = { version = "1.7.0", features = ["v4"] }
+workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
+
+[features]
+default = ["workspace-hack"]
+
+[package.metadata.cargo-udeps.ignore]
+normal = ["workspace-hack"]