diff --git a/.sqlx/query-dff74f6b34839e017e159ba35d59a978f73c0c883abf61888434d4b10e5c5a9b.json b/.sqlx/query-13b90322650a962e27d279f64cae423c749b84228d534edd16e0d7e1134af183.json similarity index 59% rename from .sqlx/query-dff74f6b34839e017e159ba35d59a978f73c0c883abf61888434d4b10e5c5a9b.json rename to .sqlx/query-13b90322650a962e27d279f64cae423c749b84228d534edd16e0d7e1134af183.json index 0c26f3ddf97..f59581bb408 100644 --- a/.sqlx/query-dff74f6b34839e017e159ba35d59a978f73c0c883abf61888434d4b10e5c5a9b.json +++ b/.sqlx/query-13b90322650a962e27d279f64cae423c749b84228d534edd16e0d7e1134af183.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "with\n accounts_root_user as (\n -- Precondition: the accounts root user must exist.\n -- Use a sub-select to select either one match or an explicit null row,\n -- which will then fail a not-null constraint.\n select (select id from auth.users where email = $4 limit 1) as accounts_id\n ),\n grant_user_admin_to_tenant as (\n insert into user_grants (user_id, object_role, capability, detail) values\n ($1, $2, 'admin', $3)\n on conflict do nothing\n ),\n grant_to_tenant as (\n insert into role_grants (subject_role, object_role, capability, detail) values\n ($2, $2, 'write', $3), -- Tenant specs may write to other tenant specs.\n ($2, 'ops/dp/public/', 'read', $3) -- Tenant may access public data-planes.\n on conflict do nothing\n ),\n public_planes as (\n select json_agg(\n data_plane_name\n order by case when data_plane_name = 'ops/dp/public/aws-us-east-1-c1' then 0 else 1 end asc,\n id desc\n ) as arr\n from data_planes\n where starts_with(data_plane_name, 'ops/dp/public/')\n and data_plane_name <> 'ops/dp/public/gcp-us-central1-c1'\n and data_plane_name <> 'ops/dp/public/gcp-us-central1-c2'\n ),\n create_storage_mappings as (\n insert into storage_mappings (catalog_prefix, spec, detail) values\n ($2, json_build_object(\n 'stores', '[{\"provider\": \"GCS\", \"bucket\": \"estuary-trial\", \"prefix\": \"collection-data/\"}]'::json,\n 'data_planes', (select arr from public_planes)\n ), $3),\n ('recovery/' || $2, '{\"stores\": [{\"provider\": \"GCS\", \"bucket\": \"estuary-trial\"}]}', $3)\n on conflict do nothing\n ),\n create_alert_subscription as (\n insert into alert_subscriptions (catalog_prefix, email, include_alert_types)\n values ($2, (select email from auth.users where id = $1 limit 1), $5)\n )\n insert into tenants (tenant, detail) values ($2, $3);\n ", + "query": "with\n accounts_root_user as (\n -- Precondition: the accounts root user must exist.\n -- Use a sub-select to select either one match or an explicit null row,\n -- which will then fail a not-null constraint.\n select (select id from auth.users where email = $4 limit 1) as accounts_id\n ),\n grant_user_admin_to_tenant as (\n insert into user_grants (user_id, object_role, capability, detail) values\n ($1, $2, 'admin', $3)\n on conflict do nothing\n ),\n grant_to_tenant as (\n insert into role_grants (subject_role, object_role, capability, detail) values\n ($2, $2, 'write', $3), -- Tenant specs may write to other tenant specs.\n ($2, 'ops/dp/public/', 'read', $3) -- Tenant may access public data-planes.\n on conflict do nothing\n ),\n create_storage_mappings as (\n insert into storage_mappings (catalog_prefix, spec, detail) values\n ($2, $6::json, $3),\n ('recovery/' || $2, $7::json, $3)\n on conflict do nothing\n ),\n create_alert_subscription as (\n insert into alert_subscriptions (catalog_prefix, email, include_alert_types)\n values ($2, (select email from auth.users where id = $1 limit 1), $5)\n )\n insert into tenants (tenant, detail) values ($2, $3);\n ", "describe": { "columns": [], "parameters": { @@ -44,10 +44,12 @@ } } } - } + }, + "Json", + "Json" ] }, "nullable": [] }, - "hash": "dff74f6b34839e017e159ba35d59a978f73c0c883abf61888434d4b10e5c5a9b" + "hash": "13b90322650a962e27d279f64cae423c749b84228d534edd16e0d7e1134af183" } diff --git a/.sqlx/query-acf649e61be4084161a21f4fac3ec68c9f396739e3f878a12bd27a55c146f254.json b/.sqlx/query-acf649e61be4084161a21f4fac3ec68c9f396739e3f878a12bd27a55c146f254.json new file mode 100644 index 00000000000..54c84beb3ba --- /dev/null +++ b/.sqlx/query-acf649e61be4084161a21f4fac3ec68c9f396739e3f878a12bd27a55c146f254.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "select data_plane_name as \"data_plane_name!\"\n from data_planes\n where starts_with(data_plane_name, $1::text)\n and not closed\n order by id desc", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "data_plane_name!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "acf649e61be4084161a21f4fac3ec68c9f396739e3f878a12bd27a55c146f254" +} diff --git a/crates/agent/src/directives/beta_onboard.rs b/crates/agent/src/directives/beta_onboard.rs index 19a221c08e2..e916d968047 100644 --- a/crates/agent/src/directives/beta_onboard.rs +++ b/crates/agent/src/directives/beta_onboard.rs @@ -1,5 +1,5 @@ use super::{JobStatus, Row, extract}; -use anyhow::Context; +use control_plane_api::directives::beta_onboard::ProvisionError; use serde::{Deserialize, Serialize}; use tracing::info; use validator::Validate; @@ -13,6 +13,12 @@ pub struct Directive {} pub struct Claims { #[validate(nested)] requested_tenant: models::Token, + // Optional full catalog name of the public data-plane the user selected + // at signup, e.g. "ops/dp/public/aws-us-east-1-c1". Becomes the first + // (default) entry of the tenant's storage-mapping data_planes. + #[serde(default)] + #[validate(length(max = 256))] + requested_data_plane: Option, // Survey results for the tenant. // This is persisted in the DB but is not actually used by the agent. #[allow(dead_code)] @@ -20,6 +26,19 @@ pub struct Claims { survey: serde_json::Value, } +/// Colocated trial buckets are gated until est-dry-dock has created the +/// per-plane buckets (est-dry-dock#326) and real public planes have +/// converged. Flipping this on before then would point new tenants at +/// buckets that don't exist. Read once and cached: the process environment +/// doesn't change between reads, so there's no reason to re-parse it on +/// every signup. +fn colocate_trial_buckets_enabled() -> bool { + static ENABLED: std::sync::OnceLock = std::sync::OnceLock::new(); + *ENABLED.get_or_init(|| { + std::env::var("COLOCATED_TRIAL_BUCKETS").is_ok_and(|v| v == "1" || v == "true") + }) +} + #[tracing::instrument(skip_all, fields(directive, row.claims))] pub async fn apply( directive: Directive, @@ -31,6 +50,7 @@ pub async fn apply( Directive {}, Claims { requested_tenant, + requested_data_plane, survey: _, }, ) = match extract(directive, &row.user_claims) { @@ -60,17 +80,30 @@ pub async fn apply( ))); } - control_plane_api::directives::beta_onboard::provision_tenant( + // The submitted plane is untrusted client input. `provision_tenant` owns + // the authoritative set of selectable public planes and rejects anything + // outside it, so there's nothing to pre-check here. + match control_plane_api::directives::beta_onboard::provision_tenant( accounts_user_email, Some("applied via directive".to_string()), &requested_tenant, row.user_id, + requested_data_plane.as_deref(), + colocate_trial_buckets_enabled(), txn, ) .await - .context("provision_tenant")?; + { + Ok(()) => {} + Err(err @ ProvisionError::PlaneNotSelectable(_)) => { + return Ok(JobStatus::invalid_claims(anyhow::anyhow!("{err}"))); + } + Err(ProvisionError::Sqlx(err)) => { + return Err(anyhow::Error::from(err).context("provision_tenant")); + } + } - info!(%row.user_id, requested_tenant=%requested_tenant.as_str(), "beta onboard"); + info!(%row.user_id, requested_tenant=%requested_tenant.as_str(), requested_data_plane=?requested_data_plane, "beta onboard"); Ok(JobStatus::Success) } @@ -98,7 +131,8 @@ mod test { p3 as ( insert into auth.users (id, email) values ('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', 'new@example.com'), - ('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', 'accounts@example.com') + ('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', 'accounts@example.com'), + ('dddddddd-dddd-dddd-dddd-dddddddddddd', 'plane@example.com') on conflict do nothing ), p4 as ( @@ -109,6 +143,23 @@ mod test { ('11111111-1111-1111-1111-111111111111', 'takenTenant/', 'admin'), -- Prevents new tenant. ('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', 'takenTenant/', 'read') -- New tenant allowed. ), + p6a as ( + -- A public plane that exists but has been closed to new selection, + -- cloned from the harness's default plane. + insert into data_planes ( + data_plane_name, data_plane_fqdn, ops_logs_name, ops_stats_name, + ops_l1_inferred_name, ops_l1_stats_name, ops_l1_events_name, + ops_l2_inferred_transform, ops_l2_stats_transform, ops_l2_events_transform, + broker_address, reactor_address, hmac_keys, enable_l2, closed + ) + select + 'ops/dp/public/closed', 'closed.dp.estuary-data.com', ops_logs_name, ops_stats_name, + ops_l1_inferred_name, ops_l1_stats_name, ops_l1_events_name, + ops_l2_inferred_transform, ops_l2_stats_transform, ops_l2_events_transform, + broker_address, reactor_address, hmac_keys, enable_l2, true + from data_planes where data_plane_name = 'ops/dp/public/test' + on conflict do nothing + ), p6 as ( insert into applied_directives (directive_id, user_id, user_claims) values -- Fails: directive prefix is incorrect. @@ -123,6 +174,16 @@ mod test { ('cc00000000000000', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '{"requestedTenant":"invalid/requested/tenant"}'), -- Fails: tenant already exists. ('cc00000000000000', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '{"requestedTenant":"TakenTeNaNt"}'), + -- Fails: requestedDataPlane is not a public plane name. + ('cc00000000000000', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '{"requestedTenant":"PlaneTenantA","requestedDataPlane":"ops/dp/private/acmeCo/aws-us-east-1-c1"}'), + -- Fails: requestedDataPlane does not exist. + ('cc00000000000000', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '{"requestedTenant":"PlaneTenantB","requestedDataPlane":"ops/dp/public/aws-nope-1-c1"}'), + -- Fails: requestedDataPlane exists and is public, but is closed to + -- new selection. + ('cc00000000000000', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '{"requestedTenant":"PlaneTenantD","requestedDataPlane":"ops/dp/public/closed"}'), + -- Success: creates PlaneTenantC with a valid requestedDataPlane, using a + -- fresh user since aaaaaaaa is about to become admin of AcmeTenant below. + ('cc00000000000000', 'dddddddd-dddd-dddd-dddd-dddddddddddd', '{"requestedTenant":"PlaneTenantC","requestedDataPlane":"ops/dp/public/test"}'), -- Success: creates AcmeTenant. ('cc00000000000000', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '{"requestedTenant":"AcmeTenant","survey":"feedback"}') ) @@ -189,7 +250,7 @@ mod test { }, "did": "cc:00:00:00:00:00:00:00", "status": { - "error": "unknown field `invalid`, expected `requestedTenant` or `survey` at line 1 column 10", + "error": "unknown field `invalid`, expected one of `requestedTenant`, `requestedDataPlane`, `survey` at line 1 column 10", "type": "invalidClaims" } }, @@ -213,6 +274,49 @@ mod test { "type": "invalidClaims" } }, + { + "claims": { + "requestedDataPlane": "ops/dp/private/acmeCo/aws-us-east-1-c1", + "requestedTenant": "PlaneTenantA" + }, + "did": "cc:00:00:00:00:00:00:00", + "status": { + "error": "ops/dp/private/acmeCo/aws-us-east-1-c1 is not a selectable public data-plane", + "type": "invalidClaims" + } + }, + { + "claims": { + "requestedDataPlane": "ops/dp/public/aws-nope-1-c1", + "requestedTenant": "PlaneTenantB" + }, + "did": "cc:00:00:00:00:00:00:00", + "status": { + "error": "ops/dp/public/aws-nope-1-c1 is not a selectable public data-plane", + "type": "invalidClaims" + } + }, + { + "claims": { + "requestedDataPlane": "ops/dp/public/closed", + "requestedTenant": "PlaneTenantD" + }, + "did": "cc:00:00:00:00:00:00:00", + "status": { + "error": "ops/dp/public/closed is not a selectable public data-plane", + "type": "invalidClaims" + } + }, + { + "claims": { + "requestedDataPlane": "ops/dp/public/test", + "requestedTenant": "PlaneTenantC" + }, + "did": "cc:00:00:00:00:00:00:00", + "status": { + "type": "success" + } + }, { "claims": { "requestedTenant": "AcmeTenant", @@ -244,6 +348,10 @@ mod test { select json_build_object('prefix', m.catalog_prefix, 'storageMapping', m.spec) from storage_mappings m where m.catalog_prefix like '%AcmeTenant%' union all + -- Expect PlaneTenantC's storage mapping reflects its requestedDataPlane. + select json_build_object('prefix', m.catalog_prefix, 'storageMapping', m.spec) + from storage_mappings m where m.catalog_prefix like '%PlaneTenantC%' + union all -- Expect an alert subscription was created. select json_build_object('catalog_prefix', s.catalog_prefix, 'email', s.email) from alert_subscriptions s where s.catalog_prefix = 'AcmeTenant/' @@ -261,6 +369,10 @@ mod test { "detail": null, "tenant": "takenTenant/" }, + { + "detail": "applied via directive", + "tenant": "PlaneTenantC/" + }, { "detail": "applied via directive", "tenant": "AcmeTenant/" @@ -307,6 +419,32 @@ mod test { ] } }, + { + "prefix": "PlaneTenantC/", + "storageMapping": { + "data_planes": [ + "ops/dp/public/test" + ], + "stores": [ + { + "bucket": "estuary-trial", + "prefix": "collection-data/", + "provider": "GCS" + } + ] + } + }, + { + "prefix": "recovery/PlaneTenantC/", + "storageMapping": { + "stores": [ + { + "bucket": "estuary-trial", + "provider": "GCS" + } + ] + } + }, { "catalog_prefix": "AcmeTenant/", "email": "new@example.com" diff --git a/crates/control-plane-api/Cargo.toml b/crates/control-plane-api/Cargo.toml index fc4c6f13a71..997f11a5fa1 100644 --- a/crates/control-plane-api/Cargo.toml +++ b/crates/control-plane-api/Cargo.toml @@ -48,6 +48,7 @@ colored_json = { workspace = true } derivative = { workspace = true } enumset = { workspace = true } futures = { workspace = true } +hex = { workspace = true } humantime = { workspace = true } humantime-serde = { workspace = true } itertools = { workspace = true } @@ -60,6 +61,7 @@ rustls = { workspace = true } schemars = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +sha2 = { workspace = true } sqlx = { workspace = true } tempfile = { workspace = true } thiserror = { workspace = true } @@ -81,11 +83,9 @@ zeroize = { workspace = true } assemble = { path = "../assemble" } flow-client-next = { path = "../flow-client-next" } -hex = { workspace = true } hmac = { workspace = true } insta = { workspace = true } md5 = { workspace = true } -sha2 = { workspace = true } tokio = { workspace = true, features = ["test-util"] } tracing-subscriber = { workspace = true } diff --git a/crates/control-plane-api/src/data_plane.rs b/crates/control-plane-api/src/data_plane.rs index 820df14ea15..e56550623bf 100644 --- a/crates/control-plane-api/src/data_plane.rs +++ b/crates/control-plane-api/src/data_plane.rs @@ -66,3 +66,132 @@ pub fn build_journal_client( Ok(journal_client) } + +/// Cloud provider where the data plane is hosted. +#[derive(Debug, Clone, Copy, PartialEq, Eq, async_graphql::Enum)] +pub enum DataPlaneCloudProvider { + Aws, + Azure, + Gcp, + Local, +} + +/// Parses a data plane name into its component parts. +/// Returns None if the name format is invalid. +/// +/// Expected formats: +/// - Cloud: "ops/dp/public/aws-us-east-1-c1" or "ops/dp/private/gcp-us-central1-c2" +/// - Local: "ops/dp/local/local-foo" (any suffix after "local-") +pub(crate) fn parse_data_plane_name( + name: &str, +) -> Option<(DataPlaneCloudProvider, String, String, bool)> { + let last_segment = name.rsplit('/').next()?; + let (provider_str, after_provider) = last_segment.split_once('-')?; + + match provider_str { + "local" => Some(( + DataPlaneCloudProvider::Local, + "local".to_string(), + "c1".to_string(), + true, + )), + "aws" | "az" | "azure" | "gcp" => { + // Must have privacy indicator in path. + if !name.contains("ops/dp/private/") && !name.starts_with("ops/dp/public/") { + return None; + } + + // Parse tag (cluster) suffix (e.g., "-c1", "-c5"). + let idx = after_provider.rfind("-c")?; + let tag = &after_provider[idx + 1..]; + if tag.len() < 2 || !tag[1..].chars().all(|c| c.is_ascii_digit()) { + return None; + } + + let region = &after_provider[..idx]; + if region.is_empty() { + return None; + } + + let cloud_provider = match provider_str { + "aws" => DataPlaneCloudProvider::Aws, + "az" | "azure" => DataPlaneCloudProvider::Azure, + "gcp" => DataPlaneCloudProvider::Gcp, + _ => unreachable!(), + }; + + let is_public = name.starts_with("ops/dp/public/"); + Some(( + cloud_provider, + region.to_string(), + tag.to_string(), + is_public, + )) + } + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_aws_public() { + let (provider, region, tag, is_public) = + parse_data_plane_name("ops/dp/public/aws-us-east-1-c1").unwrap(); + assert_eq!(provider, DataPlaneCloudProvider::Aws); + assert_eq!(region, "us-east-1"); + assert_eq!(tag, "c1"); + assert!(is_public); + } + + #[test] + fn parses_gcp_private() { + let (provider, region, tag, is_public) = + parse_data_plane_name("ops/dp/private/estuary/gcp-us-central1-c5").unwrap(); + assert_eq!(provider, DataPlaneCloudProvider::Gcp); + assert_eq!(region, "us-central1"); + assert_eq!(tag, "c5"); + assert!(!is_public); + } + + #[test] + fn parses_azure_variants() { + // "az" prefix + let (provider, region, tag, _) = + parse_data_plane_name("ops/dp/private/EastPack/az-australiaeast-c1").unwrap(); + assert_eq!(provider, DataPlaneCloudProvider::Azure); + assert_eq!(region, "australiaeast"); + assert_eq!(tag, "c1"); + + // "azure" prefix + let (provider, region, tag, _) = + parse_data_plane_name("ops/dp/private/AccumTech/azure-eastus-c1").unwrap(); + assert_eq!(provider, DataPlaneCloudProvider::Azure); + assert_eq!(region, "eastus"); + assert_eq!(tag, "c1"); + } + + #[test] + fn parses_local() { + let (provider, region, tag, is_public) = + parse_data_plane_name("ops/dp/local/local-foo").unwrap(); + assert_eq!(provider, DataPlaneCloudProvider::Local); + assert_eq!(region, "local"); + assert_eq!(tag, "c1"); + assert!(is_public); + } + + #[test] + fn rejects_invalid_names() { + // Missing privacy indicator + assert!(parse_data_plane_name("ops/dp/aws-us-east-1-c1").is_none()); + // Unknown provider + assert!(parse_data_plane_name("ops/dp/public/unknown-us-east-1-c1").is_none()); + // Missing cluster suffix + assert!(parse_data_plane_name("ops/dp/public/aws-us-east-1").is_none()); + // Non-numeric cluster + assert!(parse_data_plane_name("ops/dp/public/aws-us-east-1-ca").is_none()); + } +} diff --git a/crates/control-plane-api/src/directives/beta_onboard.rs b/crates/control-plane-api/src/directives/beta_onboard.rs index 262b9dd26c2..1b002578e5f 100644 --- a/crates/control-plane-api/src/directives/beta_onboard.rs +++ b/crates/control-plane-api/src/directives/beta_onboard.rs @@ -1,5 +1,27 @@ use sqlx::types::Uuid; +/// Derives the colocated trial bucket for a public AWS data-plane, returning +/// `None` for any plane that has no colocated bucket (non-AWS, non-public, or +/// an unparseable name). +/// +/// The name is a pure function of plane identity and is deliberately not +/// stored anywhere; est-dry-dock creates the bucket from the same formula +/// (est_dry_dock/models/__init__.py::trial_bucket_name). Region is parsed from +/// the plane name rather than accepted separately, so a bucket can never be +/// derived from a region that disagrees with the plane it belongs to. +pub fn trial_bucket_name(data_plane_name: &str) -> Option<(String, String)> { + use crate::data_plane::{DataPlaneCloudProvider, parse_data_plane_name}; + use sha2::Digest; + + let (DataPlaneCloudProvider::Aws, region, _tag, true) = parse_data_plane_name(data_plane_name)? + else { + return None; + }; + + let digest = hex::encode(sha2::Sha256::digest(data_plane_name.as_bytes())); + Some((format!("estuary-trial-{region}-{}", &digest[..8]), region)) +} + pub async fn is_user_provisioned( user_id: Uuid, txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, @@ -50,13 +72,73 @@ pub async fn tenant_exists( Ok(illegal.is_some() || exists.is_some()) } +/// The plane new tenants default to when signup carries no data-plane choice. +pub const DEFAULT_PUBLIC_DATA_PLANE: &str = "ops/dp/public/aws-us-east-1-c1"; + +/// Prefix identifying a data-plane's catalog name as public. +pub const PUBLIC_DATA_PLANE_PREFIX: &str = "ops/dp/public/"; + +/// Why provisioning a tenant failed. +#[derive(Debug, thiserror::Error)] +pub enum ProvisionError { + /// The signup claim named a plane that isn't selectable: it doesn't exist, + /// isn't public, or has been closed to new selection. + #[error("{0} is not a selectable public data-plane")] + PlaneNotSelectable(String), + #[error(transparent)] + Sqlx(#[from] sqlx::Error), +} + +/// Stable-sorts `planes` (already ordered id desc) so `default_plane` is +/// first; the first entry of a storage mapping's data_planes is the default. +fn order_public_planes(mut planes: Vec, default_plane: &str) -> Vec { + planes.sort_by_key(|name| name != default_plane); + planes +} + +/// Builds the (tenant, recovery) storage_mappings specs for a new tenant. +/// +/// When `colocate` is set and the default plane is a public AWS plane, the +/// specs point at the plane's colocated S3 trial bucket (created by +/// est-dry-dock from the same derivation). GCP/Azure planes, unparseable +/// plane names, and `colocate` being unset all keep the legacy GCS bucket. +fn storage_specs(all_planes: &[String], colocate: bool) -> (serde_json::Value, serde_json::Value) { + // The first entry of the ordered list is the tenant's default plane, and + // the only one a colocated bucket could belong to. + let s3 = all_planes + .first() + .filter(|_| colocate) + .and_then(|name| trial_bucket_name(name)); + + // The recovery spec is the collection spec's store without the prefix and + // without data_planes; build one store and derive both from it, so the S3 + // and GCS shapes cannot drift apart. + let mut store = match &s3 { + Some((bucket, region)) => { + serde_json::json!({"provider": "S3", "bucket": bucket, "region": region}) + } + None => serde_json::json!({"provider": "GCS", "bucket": "estuary-trial"}), + }; + let recovery_spec = serde_json::json!({"stores": [store.clone()]}); + + store["prefix"] = serde_json::json!("collection-data/"); + let tenant_spec = serde_json::json!({ + "stores": [store], + "data_planes": all_planes, + }); + + (tenant_spec, recovery_spec) +} + pub async fn provision_tenant( accounts_user_email: &str, detail: Option, tenant: &str, tenant_user_id: Uuid, + requested_data_plane: Option<&str>, + colocate_trial_bucket: bool, txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, -) -> sqlx::Result<()> { +) -> Result<(), ProvisionError> { let prefix = format!("{tenant}/"); let default_alert_types: Vec = models::status::AlertType::all() .iter() @@ -64,8 +146,35 @@ pub async fn provision_tenant( .filter(models::status::AlertType::is_default) .collect(); - // Note that the gcp-us-central1-c1 (combustible-cronut) dataplane is excluded here - // because it's being deprecated and replaced. + // The set of planes a new tenant may be placed on. `closed` is the single + // source of truth for "retired from new selection" — the same flag + // `publicDataPlanes` honors, so the signup picker and provisioning cannot + // disagree, and retiring a plane is a data change rather than a deploy. + let public_planes: Vec = sqlx::query_scalar!( + r#"select data_plane_name as "data_plane_name!" + from data_planes + where starts_with(data_plane_name, $1::text) + and not closed + order by id desc"#, + PUBLIC_DATA_PLANE_PREFIX, + ) + .fetch_all(&mut **txn) + .await?; + + // The requested plane is untrusted client input, and this is the only + // place that knows the authoritative candidate set — so validate here + // rather than leaving it to callers. + if let Some(requested) = requested_data_plane + && !public_planes.iter().any(|plane| plane == requested) + { + return Err(ProvisionError::PlaneNotSelectable(requested.to_string())); + } + + // The first entry of the ordered list is the tenant's default data-plane. + let default_plane = requested_data_plane.unwrap_or(DEFAULT_PUBLIC_DATA_PLANE); + let public_planes = order_public_planes(public_planes, default_plane); + let (tenant_spec, recovery_spec) = storage_specs(&public_planes, colocate_trial_bucket); + sqlx::query!( r#"with accounts_root_user as ( @@ -85,24 +194,10 @@ pub async fn provision_tenant( ($2, 'ops/dp/public/', 'read', $3) -- Tenant may access public data-planes. on conflict do nothing ), - public_planes as ( - select json_agg( - data_plane_name - order by case when data_plane_name = 'ops/dp/public/aws-us-east-1-c1' then 0 else 1 end asc, - id desc - ) as arr - from data_planes - where starts_with(data_plane_name, 'ops/dp/public/') - and data_plane_name <> 'ops/dp/public/gcp-us-central1-c1' - and data_plane_name <> 'ops/dp/public/gcp-us-central1-c2' - ), create_storage_mappings as ( insert into storage_mappings (catalog_prefix, spec, detail) values - ($2, json_build_object( - 'stores', '[{"provider": "GCS", "bucket": "estuary-trial", "prefix": "collection-data/"}]'::json, - 'data_planes', (select arr from public_planes) - ), $3), - ('recovery/' || $2, '{"stores": [{"provider": "GCS", "bucket": "estuary-trial"}]}', $3) + ($2, $6::json, $3), + ('recovery/' || $2, $7::json, $3) on conflict do nothing ), create_alert_subscription as ( @@ -116,6 +211,8 @@ pub async fn provision_tenant( detail.clone() as Option, accounts_user_email as &str, &default_alert_types as &[models::status::AlertType], + tenant_spec as serde_json::Value, + recovery_spec as serde_json::Value, ) .execute(&mut **txn) .await?; @@ -150,6 +247,8 @@ pub async fn provision_test_tenant( Some("test tenant".to_string()), tenant, user_id, + None, + false, &mut txn, ) .await @@ -163,3 +262,126 @@ pub async fn provision_test_tenant( txn.commit().await.expect("commit tenant"); user_id } + +#[cfg(test)] +mod test { + // The golden vector is shared with est-dry-dock's Python implementation + // (est_dry_dock/models/__init__.py::trial_bucket_name). If this assertion + // ever fails, the two implementations have drifted and a tenant's storage + // would be misrouted — fix the drift, never the test. + #[test] + fn trial_bucket_name_golden_vector() { + assert_eq!( + super::trial_bucket_name("ops/dp/public/aws-us-east-1-c1"), + Some(( + "estuary-trial-us-east-1-ccc98e22".to_string(), + "us-east-1".to_string(), + )), + ); + } + + // Only public AWS planes have a colocated bucket. + #[test] + fn trial_bucket_name_rejects_planes_without_a_colocated_bucket() { + for name in [ + "ops/dp/public/gcp-europe-west1-c1", + "ops/dp/public/azure-eastus2-c1", + "ops/dp/public/test", // unparseable name (local dev env) + "ops/dp/private/sean-estuary/aws-eu-west-1-c1", + ] { + assert_eq!(super::trial_bucket_name(name), None, "case: {name}"); + } + } + + #[test] + fn orders_default_plane_first_preserving_id_desc_order() { + let planes = vec![ + "ops/dp/public/gcp-europe-west1-c1".to_string(), // highest id + "ops/dp/public/aws-us-east-1-c1".to_string(), + "ops/dp/public/aws-eu-west-1-c1".to_string(), + ]; + assert_eq!( + super::order_public_planes(planes.clone(), "ops/dp/public/aws-us-east-1-c1"), + vec![ + "ops/dp/public/aws-us-east-1-c1".to_string(), + "ops/dp/public/gcp-europe-west1-c1".to_string(), + "ops/dp/public/aws-eu-west-1-c1".to_string(), + ], + ); + // Default not present: order unchanged. + assert_eq!( + super::order_public_planes(planes.clone(), "ops/dp/public/aws-us-west-2-c1"), + planes, + ); + } + + #[test] + fn storage_specs_default_to_gcs_trial() { + let planes = vec!["ops/dp/public/aws-us-east-1-c1".to_string()]; + let (tenant, recovery) = super::storage_specs(&planes, false); + assert_eq!( + tenant, + serde_json::json!({ + "stores": [{"provider": "GCS", "bucket": "estuary-trial", "prefix": "collection-data/"}], + "data_planes": ["ops/dp/public/aws-us-east-1-c1"], + }), + ); + assert_eq!( + recovery, + serde_json::json!({ + "stores": [{"provider": "GCS", "bucket": "estuary-trial"}], + }), + ); + } + + #[test] + fn storage_specs_colocate_aws_default_plane() { + let planes = vec![ + "ops/dp/public/aws-us-east-1-c1".to_string(), + "ops/dp/public/gcp-europe-west1-c1".to_string(), + ]; + let (tenant, recovery) = super::storage_specs(&planes, true); + assert_eq!( + tenant, + serde_json::json!({ + "stores": [{ + "provider": "S3", + "bucket": "estuary-trial-us-east-1-ccc98e22", + "prefix": "collection-data/", + "region": "us-east-1", + }], + "data_planes": planes, + }), + ); + assert_eq!( + recovery, + serde_json::json!({ + "stores": [{ + "provider": "S3", + "bucket": "estuary-trial-us-east-1-ccc98e22", + "region": "us-east-1", + }], + }), + ); + } + + // Non-AWS default planes, unparseable names, an empty plane list, and + // colocate=false all fall back to the GCS trial bucket. + #[test] + fn storage_specs_fall_back_to_gcs() { + for (planes, colocate) in [ + (vec!["ops/dp/public/gcp-europe-west1-c1".to_string()], true), + // Unparseable name (local dev env). + (vec!["ops/dp/public/test".to_string()], true), + (vec![], true), + (vec!["ops/dp/public/aws-us-east-1-c1".to_string()], false), + ] { + let (tenant, _) = super::storage_specs(&planes, colocate); + assert_eq!( + tenant["stores"][0], + serde_json::json!({"provider": "GCS", "bucket": "estuary-trial", "prefix": "collection-data/"}), + "case: {planes:?} colocate={colocate}", + ); + } + } +} diff --git a/crates/control-plane-api/src/server/create_data_plane.rs b/crates/control-plane-api/src/server/create_data_plane.rs index 15ca92f9802..7604d3d537a 100644 --- a/crates/control-plane-api/src/server/create_data_plane.rs +++ b/crates/control-plane-api/src/server/create_data_plane.rs @@ -103,7 +103,7 @@ pub async fn create_data_plane( let data_plane_name = format!("ops/dp/{base_name}"); - if super::public::graphql::parse_data_plane_name(&data_plane_name).is_none() { + if crate::data_plane::parse_data_plane_name(&data_plane_name).is_none() { return Err(tonic::Status::invalid_argument(format!( "data plane name '{data_plane_name}' does not match the expected format (e.g., 'ops/dp/public/aws-us-east-1-c1')", )) diff --git a/crates/control-plane-api/src/server/public/graphql/data_planes.rs b/crates/control-plane-api/src/server/public/graphql/data_planes.rs index 276c344af35..705cf75002f 100644 --- a/crates/control-plane-api/src/server/public/graphql/data_planes.rs +++ b/crates/control-plane-api/src/server/public/graphql/data_planes.rs @@ -5,7 +5,7 @@ use async_graphql::{ }; use std::collections::HashMap; -const DEFAULT_PAGE_SIZE: usize = 50; +use crate::data_plane::{DataPlaneCloudProvider, parse_data_plane_name}; /// Optional filter for the `dataPlanes` query. When omitted, all accessible /// data planes are returned. @@ -15,14 +15,7 @@ pub struct DataPlanesFilter { pub closed: Option, } -/// Cloud provider where the data plane is hosted. -#[derive(Debug, Clone, Copy, PartialEq, Eq, async_graphql::Enum)] -pub enum DataPlaneCloudProvider { - Aws, - Azure, - Gcp, - Local, -} +const DEFAULT_PAGE_SIZE: usize = 50; /// Controller-observed provisioning status of a configured private link. #[derive(Debug, Clone, Copy, PartialEq, Eq, async_graphql::Enum)] @@ -323,62 +316,6 @@ struct DataPlaneDetails { gcp_psc_endpoints: Vec, } -/// Parses a data plane name into its component parts. -/// Returns None if the name format is invalid. -/// -/// Expected formats: -/// - Cloud: "ops/dp/public/aws-us-east-1-c1" or "ops/dp/private/gcp-us-central1-c2" -/// - Local: "ops/dp/local/local-foo" (any suffix after "local-") -pub(crate) fn parse_data_plane_name( - name: &str, -) -> Option<(DataPlaneCloudProvider, String, String, bool)> { - let last_segment = name.rsplit('/').next()?; - let (provider_str, after_provider) = last_segment.split_once('-')?; - - match provider_str { - "local" => Some(( - DataPlaneCloudProvider::Local, - "local".to_string(), - "c1".to_string(), - true, - )), - "aws" | "az" | "azure" | "gcp" => { - // Must have privacy indicator in path. - if !name.contains("ops/dp/private/") && !name.starts_with("ops/dp/public/") { - return None; - } - - // Parse tag (cluster) suffix (e.g., "-c1", "-c5"). - let idx = after_provider.rfind("-c")?; - let tag = &after_provider[idx + 1..]; - if tag.len() < 2 || !tag[1..].chars().all(|c| c.is_ascii_digit()) { - return None; - } - - let region = &after_provider[..idx]; - if region.is_empty() { - return None; - } - - let cloud_provider = match provider_str { - "aws" => DataPlaneCloudProvider::Aws, - "az" | "azure" => DataPlaneCloudProvider::Azure, - "gcp" => DataPlaneCloudProvider::Gcp, - _ => unreachable!(), - }; - - let is_public = name.starts_with("ops/dp/public/"); - Some(( - cloud_provider, - region.to_string(), - tag.to_string(), - is_public, - )) - } - _ => None, - } -} - /// A public data plane, as visible to unauthenticated callers. #[derive(Debug, Clone, SimpleObject)] pub struct PublicDataPlane { @@ -1689,63 +1626,4 @@ mod tests { (vec!["a", "b", "c", "d"], false, false) ); } - - #[test] - fn parses_aws_public() { - let (provider, region, tag, is_public) = - parse_data_plane_name("ops/dp/public/aws-us-east-1-c1").unwrap(); - assert_eq!(provider, DataPlaneCloudProvider::Aws); - assert_eq!(region, "us-east-1"); - assert_eq!(tag, "c1"); - assert!(is_public); - } - - #[test] - fn parses_gcp_private() { - let (provider, region, tag, is_public) = - parse_data_plane_name("ops/dp/private/estuary/gcp-us-central1-c5").unwrap(); - assert_eq!(provider, DataPlaneCloudProvider::Gcp); - assert_eq!(region, "us-central1"); - assert_eq!(tag, "c5"); - assert!(!is_public); - } - - #[test] - fn parses_azure_variants() { - // "az" prefix - let (provider, region, tag, _) = - parse_data_plane_name("ops/dp/private/EastPack/az-australiaeast-c1").unwrap(); - assert_eq!(provider, DataPlaneCloudProvider::Azure); - assert_eq!(region, "australiaeast"); - assert_eq!(tag, "c1"); - - // "azure" prefix - let (provider, region, tag, _) = - parse_data_plane_name("ops/dp/private/AccumTech/azure-eastus-c1").unwrap(); - assert_eq!(provider, DataPlaneCloudProvider::Azure); - assert_eq!(region, "eastus"); - assert_eq!(tag, "c1"); - } - - #[test] - fn parses_local() { - let (provider, region, tag, is_public) = - parse_data_plane_name("ops/dp/local/local-foo").unwrap(); - assert_eq!(provider, DataPlaneCloudProvider::Local); - assert_eq!(region, "local"); - assert_eq!(tag, "c1"); - assert!(is_public); - } - - #[test] - fn rejects_invalid_names() { - // Missing privacy indicator - assert!(parse_data_plane_name("ops/dp/aws-us-east-1-c1").is_none()); - // Unknown provider - assert!(parse_data_plane_name("ops/dp/public/unknown-us-east-1-c1").is_none()); - // Missing cluster suffix - assert!(parse_data_plane_name("ops/dp/public/aws-us-east-1").is_none()); - // Non-numeric cluster - assert!(parse_data_plane_name("ops/dp/public/aws-us-east-1-ca").is_none()); - } } diff --git a/crates/control-plane-api/src/server/public/graphql/mod.rs b/crates/control-plane-api/src/server/public/graphql/mod.rs index 9b03bb2db70..6093360d8f4 100644 --- a/crates/control-plane-api/src/server/public/graphql/mod.rs +++ b/crates/control-plane-api/src/server/public/graphql/mod.rs @@ -27,10 +27,9 @@ mod alert_types; mod alerts; mod authorized_prefixes; pub(crate) mod billing; +mod connectors; mod data_planes; mod filters; -pub(crate) use data_planes::parse_data_plane_name; -mod connectors; pub mod id; mod invite_links; mod live_spec_refs; diff --git a/supabase/migrations/20260804120000_close_deprecated_gcp_data_planes.sql b/supabase/migrations/20260804120000_close_deprecated_gcp_data_planes.sql new file mode 100644 index 00000000000..096aa3fedb0 --- /dev/null +++ b/supabase/migrations/20260804120000_close_deprecated_gcp_data_planes.sql @@ -0,0 +1,26 @@ +BEGIN; + +-- Record the deprecation of gcp-us-central1-c1 (combustible-cronut) and its +-- successor c2 as data rather than as code. +-- +-- These two planes were excluded from new-tenant storage mappings by a +-- hardcoded list in the control plane. `data_planes.closed` (added in +-- 20260716120000) is the mechanism for precisely this — "closed to new +-- selection, but still serving existing tasks" — and `publicDataPlanes` +-- already honors it, so the signup picker never offers a closed plane. +-- Marking them closed here lets tenant provisioning and claim validation read +-- the same source of truth, and lets a future plane be retired with a data +-- change instead of a code change and a deploy. +-- +-- Existing tenants already mapped to these planes are unaffected: `closed` +-- only governs new selection. +UPDATE public.data_planes +SET + closed = true +WHERE + data_plane_name IN ( + 'ops/dp/public/gcp-us-central1-c1', + 'ops/dp/public/gcp-us-central1-c2' + ); + +COMMIT;