diff --git a/src/adapter-types/src/dyncfgs.rs b/src/adapter-types/src/dyncfgs.rs index 59798c1d15000..18a3b2bdcc12b 100644 --- a/src/adapter-types/src/dyncfgs.rs +++ b/src/adapter-types/src/dyncfgs.rs @@ -11,12 +11,13 @@ use std::time::Duration; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; pub const ALLOW_USER_SESSIONS: Config = Config::new( "allow_user_sessions", true, "Whether to allow user roles to create new sessions. When false, only system roles will be permitted to create new sessions.", + ParameterScope::Environment, ); // Slightly awkward with the WITH prefix, but we can't start with a 0.. @@ -26,18 +27,21 @@ pub const WITH_0DT_DEPLOYMENT_MAX_WAIT: Config = Config::new( // hydrated. To prevent cutting over unilaterally when there is an issue. Duration::from_hours(365 * 24), "How long to wait at most for clusters to be hydrated, when doing a zero-downtime deployment.", + ParameterScope::Environment, ); pub const WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL: Config = Config::new( "with_0dt_deployment_ddl_check_interval", Duration::from_secs(5 * 60), "How often to check for DDL changes during zero-downtime deployment.", + ParameterScope::Environment, ); pub const ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT: Config = Config::new( "enable_0dt_deployment_panic_after_timeout", false, "Whether to panic if the maximum wait time is reached but preflight checks have not succeeded.", + ParameterScope::Environment, ); pub const WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL: Config = Config::new( @@ -45,24 +49,28 @@ pub const WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL: Config = Confi "0dt_deployment_hydration_check_interval", Duration::from_secs(10), "Interval at which to check whether clusters are caught up, when doing zero-downtime deployment.", + ParameterScope::Environment, ); pub const WITH_0DT_CAUGHT_UP_CHECK_ALLOWED_LAG: Config = Config::new( "with_0dt_caught_up_check_allowed_lag", Duration::from_secs(60), "Maximum allowed lag when determining whether collections are caught up for 0dt deployments.", + ParameterScope::Environment, ); pub const WITH_0DT_CAUGHT_UP_CHECK_CUTOFF: Config = Config::new( "with_0dt_caught_up_check_cutoff", Duration::from_secs(2 * 60 * 60), // 2 hours "Collections whose write frontier is behind 'now' by more than the cutoff are ignored when doing caught-up checks for 0dt deployments.", + ParameterScope::Environment, ); pub const ENABLE_0DT_CAUGHT_UP_REPLICA_STATUS_CHECK: Config = Config::new( "enable_0dt_caught_up_replica_status_check", true, "Enable checking for crash/OOM-looping replicas during 0dt caught-up checks. Emergency break-glass flag to disable this feature if needed.", + ParameterScope::Environment, ); // TODO(aljoscha): Remove this break-glass flag after a couple of releases, once @@ -72,12 +80,14 @@ pub const ENABLE_0DT_CAUGHT_UP_STABILITY_CHECK: Config = Config::new( "enable_0dt_caught_up_stability_check", true, "Require clusters to stay caught-up and healthy for a stability period before being considered ready during 0dt deployments. Emergency break-glass flag: disabling reverts to treating a caught-up cluster as ready with no replica-health requirement, which differs from setting the stability period to zero (a zero period still requires all replicas to be healthy).", + ParameterScope::Environment, ); pub const WITH_0DT_CAUGHT_UP_CHECK_STABILITY_PERIOD: Config = Config::new( "with_0dt_caught_up_check_stability_period", Duration::from_secs(10 * 60), // 10 minutes "How long a cluster must continuously be caught-up and have all replicas healthy before it is considered ready to cut over during a 0dt deployment.", + ParameterScope::Environment, ); /// Enable logging of statement lifecycle events in mz_internal.mz_statement_lifecycle_history. @@ -85,6 +95,7 @@ pub const ENABLE_STATEMENT_LIFECYCLE_LOGGING: Config = Config::new( "enable_statement_lifecycle_logging", true, "Enable logging of statement lifecycle events in mz_internal.mz_statement_lifecycle_history.", + ParameterScope::Environment, ); /// Enable installation of introspection subscribes. @@ -92,6 +103,7 @@ pub const ENABLE_INTROSPECTION_SUBSCRIBES: Config = Config::new( "enable_introspection_subscribes", true, "Enable installation of introspection subscribes.", + ParameterScope::Environment, ); /// Enable sending subscribes down the new frontend-peek path. @@ -99,6 +111,7 @@ pub const ENABLE_FRONTEND_SUBSCRIBES: Config = Config::new( "enable_frontend_subscribes", true, "Enable sending subscribes down the new frontend-peek path.", + ParameterScope::Environment, ); /// The plan insights notice will not investigate fast path clusters if plan optimization took longer than this. @@ -111,6 +124,7 @@ pub const PLAN_INSIGHTS_NOTICE_FAST_PATH_CLUSTERS_OPTIMIZE_DURATION: Config = Config::new( true, "Use a cache to store optimized expressions to help speed up start times. \ Read at startup, so changing it takes effect on the next restart.", + ParameterScope::Environment, ); /// Whether to enable password authentication. @@ -126,6 +141,7 @@ pub const ENABLE_PASSWORD_AUTH: Config = Config::new( "enable_password_auth", false, "Enable password authentication.", + ParameterScope::Environment, ); /// Upper bound on the number of transitive dependencies validated for a @@ -138,11 +154,16 @@ pub const READ_THEN_WRITE_MAX_DEPENDENCIES: Config = Config::new( 100_000, "Maximum number of transitive dependencies validated for a read-then-write \ statement before it is rejected.", + ParameterScope::Environment, ); /// OIDC issuer URL. -pub const OIDC_ISSUER: Config> = - Config::new("oidc_issuer", None, "OIDC issuer URL."); +pub const OIDC_ISSUER: Config> = Config::new( + "oidc_issuer", + None, + "OIDC issuer URL.", + ParameterScope::Environment, +); /// OIDC audience (client IDs). When empty, audience validation is skipped. /// Validates that the JWT's `aud` claim contains at least one of these values. @@ -153,6 +174,7 @@ pub const OIDC_AUDIENCE: Config serde_json::Value> = Config::new( "oidc_audience", || serde_json::json!([]), "OIDC audience (client IDs). A JSON array of strings. When empty, audience validation is skipped.", + ParameterScope::Environment, ); /// OIDC authentication claim to use as username @@ -160,6 +182,7 @@ pub const OIDC_AUTHENTICATION_CLAIM: Config<&'static str> = Config::new( "oidc_authentication_claim", "sub", "OIDC authentication claim to use as username.", + ParameterScope::Environment, ); /// Whether OIDC group-to-role sync is enabled. @@ -168,6 +191,7 @@ pub const OIDC_GROUP_ROLE_SYNC_ENABLED: Config = Config::new( "oidc_group_role_sync_enabled", false, "Enable OIDC JWT group-to-role membership sync on login.", + ParameterScope::Environment, ); /// The JWT claim path that contains group memberships. May be a bare claim @@ -177,6 +201,7 @@ pub const OIDC_GROUP_CLAIM: Config<&'static str> = Config::new( "oidc_group_claim", "groups", "JWT claim path containing group memberships for role sync. Supports dot-separated paths into nested objects (e.g. customClaims.groups).", + ParameterScope::Environment, ); /// Whether to reject login when group sync fails (strict/fail-closed mode). @@ -185,12 +210,14 @@ pub const OIDC_GROUP_ROLE_SYNC_STRICT: Config = Config::new( "oidc_group_role_sync_strict", false, "When true, reject login if OIDC group-to-role sync fails (fail-closed).", + ParameterScope::Environment, ); pub const PERSIST_FAST_PATH_ORDER: Config = Config::new( "persist_fast_path_order", false, "If set, send queries with a compatible literal constraint or ordering clause down the Persist fast path.", + ParameterScope::Environment, ); /// Whether to enforce that S3 Tables connections are in the same region as the Materialize @@ -199,6 +226,7 @@ pub const ENABLE_S3_TABLES_REGION_CHECK: Config = Config::new( "enable_s3_tables_region_check", false, "Whether to enforce that S3 Tables connections are in the same region as the environment.", + ParameterScope::Environment, ); /// Whether the MCP agent endpoint is enabled. @@ -206,6 +234,7 @@ pub const ENABLE_MCP_AGENT: Config = Config::new( "enable_mcp_agent", true, "Whether the MCP agent HTTP endpoint is enabled. When false, requests to /api/mcp/agent return 503 Service Unavailable.", + ParameterScope::Environment, ); /// Whether the MCP agent query tool is enabled. @@ -215,6 +244,7 @@ pub const ENABLE_MCP_AGENT_QUERY_TOOL: Config = Config::new( "enable_mcp_agent_query_tool", true, "Whether the MCP agent query tool is enabled. When false, the query tool is not advertised and calls to it are rejected. Agents can still discover and inspect data products.", + ParameterScope::Environment, ); /// Whether the MCP agent read_data_product tool is enabled. @@ -224,6 +254,7 @@ pub const ENABLE_MCP_AGENT_READ_DATA_PRODUCT_TOOL: Config = Config::new( "enable_mcp_agent_read_data_product_tool", true, "Whether the MCP agent read_data_product tool is enabled. When false, the read_data_product tool is not advertised and calls to it are rejected. Agents can use the query tool to read data products.", + ParameterScope::Environment, ); /// Whether the MCP developer endpoint is enabled. @@ -231,6 +262,7 @@ pub const ENABLE_MCP_DEVELOPER: Config = Config::new( "enable_mcp_developer", true, "Whether the MCP developer HTTP endpoint is enabled. When false, requests to /api/mcp/developer return 503 Service Unavailable.", + ParameterScope::Environment, ); /// Whether the MCP developer query tool is enabled. @@ -240,6 +272,7 @@ pub const ENABLE_MCP_DEVELOPER_QUERY_TOOL: Config = Config::new( "enable_mcp_developer_query_tool", true, "Whether the MCP developer query tool is enabled. When false, the query tool is not advertised and calls to it are rejected. Developers can still use query_system_catalog.", + ParameterScope::Environment, ); /// Whether the external metrics endpoint on environmentd is enabled. @@ -247,6 +280,7 @@ pub const ENABLE_PUBLIC_METRICS_ENDPOINT: Config = Config::new( "enable_public_metrics_endpoint", true, "Whether the external metrics endpoint on environmentd is enabled. When false, requests return 503.", + ParameterScope::Environment, ); /// Maximum size (in bytes) of MCP tool response content after JSON serialization. @@ -256,6 +290,7 @@ pub const MCP_MAX_RESPONSE_SIZE: Config = Config::new( "mcp_max_response_size", 1_000_000, "Maximum size in bytes of MCP tool response content. Responses exceeding this limit are rejected with an error telling the agent to narrow its query.", + ParameterScope::Environment, ); /// Maximum time an MCP request may run before it is aborted and a timeout @@ -264,6 +299,7 @@ pub const MCP_REQUEST_TIMEOUT: Config = Config::new( "mcp_request_timeout", Duration::from_secs(60), "Maximum time an MCP request may run before it is aborted with a timeout error.", + ParameterScope::Environment, ); /// Maximum size (in bytes) of a webhook request body, measured after @@ -275,6 +311,7 @@ pub const WEBHOOK_MAX_REQUEST_SIZE_BYTES: Config = Config::new( // Matches `MAX_REQUEST_SIZE`, the static limit the other environmentd HTTP routes use. 5 * 1024 * 1024, "The maximum size in bytes of a webhook request body, measured after decompression.", + ParameterScope::Environment, ); /// Maximum temporary storage a webhook `CHECK` expression may allocate while @@ -295,6 +332,7 @@ pub const WEBHOOK_VALIDATION_MEMORY_BUDGET_BYTES: Config = Config::new( "webhook_validation_memory_budget_bytes", 20 * 1024 * 1024, "The maximum bytes of temporary storage a webhook CHECK expression may allocate while validating one request.", + ParameterScope::Environment, ); /// Budget for the backlog a `SUBSCRIBE` (or `COPY (SUBSCRIBE ...) TO STDOUT`) @@ -314,6 +352,7 @@ pub const SUBSCRIBE_MAX_BUFFERED_BYTES: Config = Config::new( "subscribe_max_buffered_bytes", 128 * 1024 * 1024, "Maximum bytes a SUBSCRIBE may buffer in environmentd for a slow client before it is retired with an error.", + ParameterScope::Environment, ); /// Number of user IDs to pre-allocate in a batch. Pre-allocating IDs avoids @@ -322,6 +361,7 @@ pub const USER_ID_POOL_BATCH_SIZE: Config = Config::new( "user_id_pool_batch_size", 512, "Number of user IDs to pre-allocate in a batch for DDL operations.", + ParameterScope::Environment, ); /// Maximum number of txns-shard write attempts before rebuilding `environmentd`. @@ -331,6 +371,7 @@ pub const GROUP_COMMIT_MAX_ATTEMPTS: Config = Config::new( "group_commit_max_attempts", 100, "Maximum number of txns-shard write attempts before rebuilding environmentd. Values below 1 are treated as 1.", + ParameterScope::Environment, ); /// OIDC client ID for the web console. @@ -338,6 +379,7 @@ pub const CONSOLE_OIDC_CLIENT_ID: Config<&'static str> = Config::new( "console_oidc_client_id", "", "OIDC client ID for the web console.", + ParameterScope::Environment, ); /// Space-separated OIDC scopes requested by the web console. @@ -345,6 +387,7 @@ pub const CONSOLE_OIDC_SCOPES: Config<&'static str> = Config::new( "console_oidc_scopes", "", "Space-separated OIDC scopes requested by the web console.", + ParameterScope::Environment, ); /// Interval at which to collect per-object arrangement size snapshots for the history table. @@ -354,6 +397,7 @@ pub const ARRANGEMENT_SIZE_HISTORY_COLLECTION_INTERVAL: Config = Confi Duration::ZERO, "Interval at which to collect and snapshot per-object arrangement sizes \ into mz_internal.mz_object_arrangement_size_history.", + ParameterScope::Environment, ); /// How long to retain per-object arrangement size history. @@ -361,6 +405,7 @@ pub const ARRANGEMENT_SIZE_HISTORY_RETENTION_PERIOD: Config = Config:: "arrangement_size_history_retention_period", Duration::from_hours(7 * 24), "How long to retain rows in mz_internal.mz_object_arrangement_size_history.", + ParameterScope::Environment, ); /// How frequently the catalog `*_info` metrics (`mz_object_info`, @@ -370,6 +415,7 @@ pub const CATALOG_INFO_METRICS_RECONCILE_INTERVAL: Config = Config::ne "catalog_info_metrics_reconcile_interval", Duration::from_secs(30), "How frequently to reconcile the catalog `*_info` metrics with the catalog. A zero duration disables reconciliation.", + ParameterScope::Environment, ); /// Server-side `statement_timeout` to set on Postgres/CRDB connections used by @@ -380,6 +426,7 @@ pub const PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT: Config = Config::new( crate::timestamp_oracle::DEFAULT_PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT, "The server-side statement timeout to set on Postgres/CRDB connections used by the \ Postgres/CRDB timestamp oracle. A value of zero leaves the statement timeout unset.", + ParameterScope::Environment, ); /// Cadence of the cluster controller's reconcile tick. @@ -387,6 +434,7 @@ pub const CLUSTER_CONTROLLER_TICK_INTERVAL: Config = Config::new( "cluster_controller_tick_interval", Duration::from_secs(5), "How often the cluster controller runs a reconcile tick.", + ParameterScope::Environment, ); /// Whether a config-shape `ALTER CLUSTER` returns immediately, with the @@ -399,6 +447,7 @@ pub const ENABLE_BACKGROUND_ALTER_CLUSTER: Config = Config::new( "enable_background_alter_cluster", true, "Whether a config-shape ALTER CLUSTER returns immediately (true) or the session blocks on a wait-shim over the durable reconfiguration record (false).", + ParameterScope::Environment, ); /// The reconfiguration deadline written when a config-shape `ALTER CLUSTER` @@ -408,6 +457,7 @@ pub const DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT: Config = Config::ne "default_cluster_reconfiguration_timeout", Duration::from_secs(60 * 60 * 24), "The reconfiguration deadline written when a config-shape ALTER CLUSTER omits WITH (WAIT ...).", + ParameterScope::Environment, ); /// Break-glass for the hydration-burst strategy: when off the controller never @@ -421,6 +471,7 @@ pub const ENABLE_HYDRATION_BURST: Config = Config::new( "enable_hydration_burst", true, "Whether the cluster controller's hydration-burst strategy may run a burst replica (break-glass; leaves graceful reconfiguration and ON REFRESH untouched).", + ParameterScope::Environment, ); /// The burst-replica linger duration written into a new `burst` record when the @@ -430,6 +481,7 @@ pub const DEFAULT_HYDRATION_BURST_LINGER: Config = Config::new( "default_hydration_burst_linger", Duration::from_secs(0), "The burst-replica linger duration written when an AUTO SCALING STRATEGY omits LINGER DURATION.", + ParameterScope::Environment, ); pub const FRONTEND_READ_THEN_WRITE: Config = Config::new( @@ -438,6 +490,7 @@ pub const FRONTEND_READ_THEN_WRITE: Config = Config::new( "Use frontend sequencing (with optimistic concurrency control) for \ DELETE, UPDATE, and INSERT operations. Read at startup, so changing it \ takes effect on the next restart.", + ParameterScope::Environment, ); /// Adds the full set of all adapter `Config`s. diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index a4b48fd5e2513..f621713c9272a 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -2349,20 +2349,15 @@ impl Coordinator { }) } - /// Resolves the replica-local scoped overrides from the catalog working copy - /// into the compute controller's per-replica dyncfg layer, then re-pushes - /// the environment-wide compute configuration so replicas observe the new - /// values. Driven by the catalog implication for replica-scoped - /// configuration changes, and called once on bootstrap. - pub(crate) fn push_replica_dyncfg_overrides(&mut self) { - // Clone the (sparse) replica overrides so we don't hold a catalog borrow - // across the mutable controller calls below. - let replica_overrides = self - .catalog() - .state() - .scoped_system_parameters() - .replica - .clone(); + /// Renders the replica-local scoped overrides in the catalog working copy as + /// per-replica [`ConfigUpdates`], grouped by cluster. + /// + /// Sparse: only replicas with an override are present. Parameters that are + /// not dyncfgs are skipped, as are values that fail to parse. + pub(crate) fn replica_dyncfg_overrides( + &self, + ) -> BTreeMap> { + let replica_overrides = &self.catalog().state().scoped_system_parameters().replica; let dyncfgs = self.catalog().system_config().dyncfgs(); let mut instance_overrides: BTreeMap< @@ -2397,22 +2392,32 @@ impl Coordinator { } } + instance_overrides + } + + /// Resolves the replica-local scoped overrides from the catalog working copy + /// into the controllers' per-replica dyncfg layers, then re-pushes the + /// environment-wide configuration so replicas observe the new values. + /// Driven by the catalog implication for replica-scoped configuration + /// changes, and called once on bootstrap. + pub(crate) fn push_replica_dyncfg_overrides(&mut self) { + let instance_overrides = self.replica_dyncfg_overrides(); + // Both controllers carry a per-replica dyncfg layer, because the two // protocols realize configs in different worker `ConfigSet`s on // `clusterd`. The compute worker's `handle_update_configuration` // applies the pushed dyncfg updates to compute's own worker - // `ConfigSet` and to the shared persist client `ConfigSet` + // `ConfigSet`, to the shared persist client `ConfigSet` // (`persist_clients.cfg()`) that the co-located storage server reads - // from the same `Arc`, which covers persist-backed and process-global - // configs such as persist client tuning and `lgalloc`. Configs - // realized from the storage worker's own `ConfigSet` (read in its - // `UpdateConfiguration` handler) are reached only by the storage - // controller's layer. - self.controller - .compute - .update_replica_dyncfg_overrides(instance_overrides.clone()); + // from the same `Arc`, and to `mz_metrics`, which covers + // persist-backed and process-global configs such as persist client + // tuning and `lgalloc`. Configs realized from the storage worker's own + // `ConfigSet` (read in its `UpdateConfiguration` handler) are reached + // only by the storage controller's layer. A third class is not pushed + // to a running replica at all but baked into its process configuration + // when the controller provisions it, which is why the overrides also go + // to the outer controller. self.controller - .storage .update_replica_dyncfg_overrides(instance_overrides); // Re-push the env-wide configs so existing replicas pick up their // (possibly changed) overrides. This also reverts a removed override: @@ -2501,6 +2506,16 @@ impl Coordinator { .update_orchestrator_scheduling_config(scheduling_config); self.controller.update_configuration(dyncfg_updates); + // Install the replica-local scoped overrides before creating any + // replica below. Parts of a replica's configuration (its `TimelyConfig`, + // its expiration offset) are resolved once, when the controller + // provisions the replica, and must see its overrides at that point. The + // push after the creation loop cannot serve this purpose, because those + // values are frozen by then. + let replica_dyncfg_overrides = self.replica_dyncfg_overrides(); + self.controller + .update_replica_dyncfg_overrides(replica_dyncfg_overrides); + // Skip the credit consumption check at bootstrap under DisableClusterCreation behavior: // this codepath validates existing replicas at startup, not cluster creation, so it // must not block startup. New cluster creation is still gated by the DDL-time check. @@ -2556,7 +2571,7 @@ impl Coordinator { } // Now that the compute instances and their replicas exist, push the - // replica-local scoped overrides into the compute controller so existing + // replica-local scoped overrides into the controllers so existing // replicas observe them at startup. The scoped (per-cluster and // per-replica) working copy was restored from the durable cache into // `CatalogState` while opening the catalog, so the last-known values are diff --git a/src/adapter/src/coord/catalog_implications.rs b/src/adapter/src/coord/catalog_implications.rs index 961f8cafaed48..432865aa6c757 100644 --- a/src/adapter/src/coord/catalog_implications.rs +++ b/src/adapter/src/coord/catalog_implications.rs @@ -697,9 +697,11 @@ impl Coordinator { // Apply replica-scoped overrides after clusters are created (so their // compute instances exist) but before replicas are created below. The // override layer must be set before `create_replica`, so the new - // replica's first configuration replays with its override. The push - // reads the catalog working copy, which already reflects this - // transaction's scoped-config changes. + // replica's first configuration replays with its override, and so the + // configuration the controller freezes into the replica's process at + // provisioning time resolves against it. The push reads the catalog + // working copy, which already reflects this transaction's scoped-config + // changes. if replica_scoped_config_changed { self.push_replica_dyncfg_overrides(); } diff --git a/src/balancerd/src/dyncfgs.rs b/src/balancerd/src/dyncfgs.rs index 6803c3b7497ef..bc1760f98063e 100644 --- a/src/balancerd/src/dyncfgs.rs +++ b/src/balancerd/src/dyncfgs.rs @@ -13,7 +13,7 @@ use std::str::FromStr; use std::time::Duration; use anyhow::anyhow; -use mz_dyncfg::{Config, ConfigSet, ConfigUpdates}; +use mz_dyncfg::{Config, ConfigSet, ConfigUpdates, ParameterScope}; use mz_tracing::params::TracingParameters; use mz_tracing::{CloneableEnvFilter, SerializableDirective}; use tracing_subscriber::filter::Directive; @@ -27,6 +27,7 @@ pub const SIGTERM_CONNECTION_WAIT: Config = Config::new( "balancerd_sigterm_connection_wait", Duration::from_secs(60 * 9), "Duration to wait after listeners closed via SIGTERM for outstanding connections to complete.", + ParameterScope::Environment, ); /// Duration to wait after SIGTERM to begin shutdown of servers. @@ -34,6 +35,7 @@ pub const SIGTERM_LISTEN_WAIT: Config = Config::new( "balancerd_sigterm_listen_wait", Duration::from_secs(60), "Duration to wait after SIGTERM to begin shutdown of servers.", + ParameterScope::Environment, ); /// Whether to inject tcp proxy protocol headers to downstream http servers. @@ -41,6 +43,7 @@ pub const INJECT_PROXY_PROTOCOL_HEADER_HTTP: Config = Config::new( "balancerd_inject_proxy_protocol_header_http", false, "Whether to inject tcp proxy protocol headers to downstream http servers.", + ParameterScope::Environment, ); /// Sets the filter to apply to stderr logging. @@ -48,6 +51,7 @@ pub const LOGGING_FILTER: Config<&str> = Config::new( "balancerd_log_filter", "info", "Sets the filter to apply to stderr logging.", + ParameterScope::Environment, ); /// Sets the filter to apply to OpenTelemetry-backed distributed tracing. @@ -55,6 +59,7 @@ pub const OPENTELEMETRY_FILTER: Config<&str> = Config::new( "balancerd_opentelemetry_filter", "info", "Sets the filter to apply to OpenTelemetry-backed distributed tracing.", + ParameterScope::Environment, ); /// Sets additional default directives to apply to stderr logging. @@ -66,6 +71,7 @@ pub const LOGGING_FILTER_DEFAULTS: Config String> = Config::new( "Sets additional default directives to apply to stderr logging. \ These apply to all variations of `log_filter`. Directives other than \ `module=off` are likely incorrect. Comma separated list.", + ParameterScope::Environment, ); /// Sets additional default directives to apply to OpenTelemetry-backed @@ -79,6 +85,7 @@ pub const OPENTELEMETRY_FILTER_DEFAULTS: Config String> = Config::new( distributed tracing. \ These apply to all variations of `opentelemetry_filter`. Directives other than \ `module=off` are likely incorrect. Comma separated list.", + ParameterScope::Environment, ); /// Sets additional default directives to apply to sentry logging. \ @@ -90,6 +97,7 @@ pub const SENTRY_FILTERS: Config String> = Config::new( "Sets additional default directives to apply to sentry logging. \ These apply on top of a default `info` directive. Directives other than \ `module=off` are likely incorrect. Comma separated list.", + ParameterScope::Environment, ); /// Adds the full set of all balancer `Config`s. diff --git a/src/compute-client/src/controller.rs b/src/compute-client/src/controller.rs index 3d9e145b0076a..f3ca5b1e9dd99 100644 --- a/src/compute-client/src/controller.rs +++ b/src/compute-client/src/controller.rs @@ -212,6 +212,12 @@ pub struct ComputeController { /// Updated through `ComputeController::update_configuration` calls and shared with all /// subcomponents of the compute controller. dyncfg: Arc, + /// The replica-local scoped overrides of [`Self::dyncfg`], by replica. + /// + /// Sparse, and kept here in addition to on the `Instance`s because replica + /// configuration that the controller resolves once, at replica creation, + /// must be read through the new replica's overrides. + replica_dyncfg_overrides: BTreeMap, /// Receiver for responses produced by `Instance`s. response_rx: mpsc::UnboundedReceiver, @@ -307,6 +313,7 @@ impl ComputeController { now, wallclock_lag, dyncfg: Arc::new(mz_dyncfgs::all_dyncfgs()), + replica_dyncfg_overrides: BTreeMap::new(), response_rx, response_tx, introspection_rx: Some(introspection_rx), @@ -471,6 +478,7 @@ impl ComputeController { now: _, wallclock_lag: _, dyncfg: _, + replica_dyncfg_overrides: _, response_rx: _, response_tx: _, introspection_rx: _, @@ -641,16 +649,22 @@ impl ComputeController { /// Replaces the per-replica dyncfg overrides for the given instances. /// - /// This only stores the overrides; callers should follow with a - /// configuration push (e.g. [`Self::update_configuration`]) so existing - /// replicas observe the new values. Instances absent from `overrides` have - /// their overrides cleared, so a replica that no longer has an override - /// reverts to the environment-wide configuration. Used by the scoped - /// feature flags (replica-local) layer. + /// This only stores the overrides, here and on the instances; callers + /// should follow with a configuration push (e.g. + /// [`Self::update_configuration`]) so existing replicas observe the new + /// values. Instances absent from `overrides` have their overrides cleared, + /// so a replica that no longer has an override reverts to the + /// environment-wide configuration. Used by the scoped feature flags + /// (replica-local) layer. pub fn update_replica_dyncfg_overrides( &mut self, mut overrides: BTreeMap>, ) { + self.replica_dyncfg_overrides = overrides + .values() + .flat_map(|replicas| replicas.iter()) + .map(|(replica_id, updates)| (*replica_id, updates.clone())) + .collect(); for (id, instance) in self.instances.iter_mut() { let instance_overrides = overrides.remove(id).unwrap_or_default(); instance.call(move |i| i.update_replica_dyncfg_overrides(instance_overrides)); @@ -719,7 +733,17 @@ impl ComputeController { None => (false, Duration::from_secs(1)), }; - let expiration_offset = COMPUTE_REPLICA_EXPIRATION_OFFSET.get(&self.dyncfg); + // Both configs below are `ParameterScope::Replica` and are resolved + // here, once, for the replica being created. Reading them through the + // new replica's scoped overrides is what makes those declarations + // effective: the values are frozen into `ReplicaConfig` and never + // re-read from the environment-wide set. The overrides for a replica + // created by DDL are committed in the same transaction that creates it, + // so they are already installed by the time we get here. + let overrides = self.replica_dyncfg_overrides.get(&replica_id); + + let expiration_offset = + COMPUTE_REPLICA_EXPIRATION_OFFSET.get_with_overrides(&self.dyncfg, overrides); // Capture dictionary compression once, at replica creation, and hold it fixed for the // replica's lifetime (see `InstanceConfig::arrangement_dictionary_compression`). This is @@ -728,7 +752,7 @@ impl ComputeController { // while the flag is enabled, so turning the flag off disables compression on new or // restarted replicas regardless of their configuration. let arrangement_dictionary_compression = ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA - .get(&self.dyncfg) + .get_with_overrides(&self.dyncfg, overrides) && config.arrangement_compression; let replica_config = ReplicaConfig { @@ -772,6 +796,12 @@ impl ComputeController { instance.replicas.remove(&replica_id); + // The coordinator only re-pushes the override map when the scoped + // configuration itself changes, so a dropped replica's entry would + // otherwise be retained until the next such change. + self.replica_dyncfg_overrides.remove(&replica_id); + + let instance = self.instance_mut(instance_id).expect("validated"); instance.call(move |i| i.remove_replica(replica_id).expect("validated")); Ok(()) diff --git a/src/compute-client/src/controller/instance.rs b/src/compute-client/src/controller/instance.rs index 44941bda7b3c0..f5490bd179e35 100644 --- a/src/compute-client/src/controller/instance.rs +++ b/src/compute-client/src/controller/instance.rs @@ -1294,6 +1294,11 @@ impl Instance { pub fn remove_replica(&mut self, id: ReplicaId) -> Result<(), ReplicaMissing> { let replica = self.replicas.remove(&id).ok_or(ReplicaMissing(id))?; + // The coordinator only re-pushes the override map when the scoped configuration itself + // changes, so a dropped replica's entry would otherwise be retained until the next such + // change. + self.replica_dyncfg_overrides.remove(&id); + // Before dropping the replica state (and the contained input read holds), log read holds // that are the last line of defense against compaction of a dataflow's storage inputs. If // the corresponding global read hold has already been released, dropping the per-replica diff --git a/src/compute-client/src/controller/replica.rs b/src/compute-client/src/controller/replica.rs index 023718f94bede..aaede56101a16 100644 --- a/src/compute-client/src/controller/replica.rs +++ b/src/compute-client/src/controller/replica.rs @@ -17,7 +17,7 @@ use anyhow::bail; use mz_build_info::BuildInfo; use mz_cluster_client::client::ClusterReplicaLocation; use mz_compute_types::dyncfgs::ENABLE_COMPUTE_REPLICA_EXPIRATION; -use mz_dyncfg::ConfigSet; +use mz_dyncfg::{ConfigSet, ConfigUpdates}; use mz_ore::channel::InstrumentedUnboundedSender; use mz_ore::retry::{Retry, RetryState}; use mz_ore::task::AbortOnDropHandle; @@ -95,6 +95,7 @@ impl ReplicaClient { epoch, metrics: metrics.clone(), connected: Arc::clone(&connected), + replica_dyncfg: seed_replica_dyncfg(&dyncfg), dyncfg, } .run(), @@ -131,6 +132,27 @@ impl ReplicaClient { type ComputeCtpClient = transport::Client; +/// Creates a replica's effective configuration, seeded from the environment-wide one. +/// +/// The seed covers the window before the first configuration command arrives, and is replaced +/// wholesale by the snapshot that `CreateInstance` carries. +fn seed_replica_dyncfg(dyncfg: &ConfigSet) -> ConfigSet { + let replica_dyncfg = mz_dyncfgs::all_dyncfgs(); + ConfigUpdates::from(dyncfg).apply(&replica_dyncfg); + replica_dyncfg +} + +/// Applies the configuration a command carries, if any, to a replica's effective configuration. +/// +/// `CreateInstance` carries a full snapshot, `UpdateConfiguration` the subsequent deltas. +fn apply_config_command(command: &ComputeCommand, dyncfg: &ConfigSet) { + match command { + ComputeCommand::CreateInstance(config) => config.initial_config.apply(dyncfg), + ComputeCommand::UpdateConfiguration(params) => params.dyncfg_updates.apply(dyncfg), + _ => (), + } +} + /// Configuration for `replica_task`. struct ReplicaTask { /// The ID of the replica. @@ -150,8 +172,21 @@ struct ReplicaTask { metrics: ReplicaMetrics, /// Flag to report successful replica connection. connected: Arc, - /// Dynamic system configuration. + /// The controller's environment-wide dynamic system configuration. dyncfg: Arc, + /// This replica's effective dynamic system configuration. + /// + /// Holds what the replica itself reads, including its scoped overrides, as opposed to + /// [`Self::dyncfg`], which holds the environment-wide values. Seeded from the environment-wide + /// configuration and then kept current from the configuration commands passing through this + /// task, which `Instance::specialize_command_for_replica` has already specialized for this + /// replica. Read it for any `ParameterScope::Replica` config the controller realizes on this + /// replica's behalf, else the scope declaration is a silent no-op. + /// + /// A set of its own, rather than a clone of the controller's: a cloned `ConfigSet` shares its + /// values with the original, so applying this replica's overrides to a clone would overwrite + /// the environment-wide configuration for everyone. + replica_dyncfg: ConfigSet, } impl ReplicaTask { @@ -229,8 +264,7 @@ impl ReplicaTask { // The sequential hydration interceptor holds back `Schedule` commands and releases them as // hydration capacity frees up. It is recreated per incarnation, matching the lifetime of // the connection: any in-flight hydration state is reset when we reconnect. - let mut hydration = - SequentialHydration::new(Arc::clone(&self.dyncfg), self.metrics.clone()); + let mut hydration = SequentialHydration::new(self.metrics.clone()); loop { select! { @@ -243,7 +277,8 @@ impl ReplicaTask { self.specialize_command(&mut command); self.observe_command(&command); - for command in hydration.absorb_command(command) { + apply_config_command(&command, &self.replica_dyncfg); + for command in hydration.absorb_command(command, &self.replica_dyncfg) { client.send(command).await?; } }, @@ -255,7 +290,7 @@ impl ReplicaTask { self.observe_response(&response); - for command in hydration.observe_response(&response) { + for command in hydration.observe_response(&response, &self.replica_dyncfg) { client.send(command).await?; } @@ -321,3 +356,52 @@ impl ReplicaTask { ); } } + +#[cfg(test)] +mod tests { + use mz_compute_types::dyncfgs::HYDRATION_CONCURRENCY; + + use crate::protocol::command::{ComputeParameters, InstanceConfig}; + + use super::*; + + /// A replica's effective configuration tracks the configuration commands passing through its + /// task, which carry the replica's scoped overrides, and leaves the environment-wide + /// configuration alone. + #[mz_ore::test] + fn replica_dyncfg_tracks_config_commands() { + let env_wide = mz_dyncfgs::all_dyncfgs(); + let mut updates = ConfigUpdates::default(); + updates.add(&HYDRATION_CONCURRENCY, 1); + updates.apply(&env_wide); + + let replica_dyncfg = seed_replica_dyncfg(&env_wide); + assert_eq!(HYDRATION_CONCURRENCY.get(&replica_dyncfg), 1); + + // A replica-scoped override arrives merged into the create-time snapshot. + let mut initial_config = ConfigUpdates::default(); + initial_config.add(&HYDRATION_CONCURRENCY, 2); + let create = ComputeCommand::CreateInstance(Box::new(InstanceConfig { + logging: Default::default(), + expiration_offset: None, + peek_stash_persist_location: mz_persist_client::PersistLocation::new_in_mem(), + arrangement_dictionary_compression: false, + initial_config, + })); + apply_config_command(&create, &replica_dyncfg); + assert_eq!(HYDRATION_CONCURRENCY.get(&replica_dyncfg), 2); + + // And into subsequent configuration updates. + let mut dyncfg_updates = ConfigUpdates::default(); + dyncfg_updates.add(&HYDRATION_CONCURRENCY, 3); + let update = ComputeCommand::UpdateConfiguration(Box::new(ComputeParameters { + dyncfg_updates, + ..Default::default() + })); + apply_config_command(&update, &replica_dyncfg); + assert_eq!(HYDRATION_CONCURRENCY.get(&replica_dyncfg), 3); + + // The environment-wide configuration is untouched by the replica's overrides. + assert_eq!(HYDRATION_CONCURRENCY.get(&env_wide), 1); + } +} diff --git a/src/compute-client/src/controller/sequential_hydration.rs b/src/compute-client/src/controller/sequential_hydration.rs index bbf4b014365a6..3f07e963efeff 100644 --- a/src/compute-client/src/controller/sequential_hydration.rs +++ b/src/compute-client/src/controller/sequential_hydration.rs @@ -70,10 +70,13 @@ type Token = Arc<()>; /// [`SequentialHydration::observe_response`]). Both methods return the commands the task should /// send to the replica, with `Schedule` commands held back or released according to the configured /// hydration concurrency. +/// +/// Both methods take the replica's effective configuration, which the task owns and keeps current. +/// Reading [`HYDRATION_CONCURRENCY`] from there rather than from the controller's environment-wide +/// set is what makes its `Replica` scope effective, given that the config is enforced here and +/// never read on the replica itself. #[derive(Debug)] pub(super) struct SequentialHydration { - /// Dynamic system configuration. - dyncfg: Arc, /// Tracked metrics. metrics: ReplicaMetrics, /// Tracked collections. @@ -93,9 +96,8 @@ pub(super) struct SequentialHydration { impl SequentialHydration { /// Create a new `SequentialHydration` interceptor. - pub(super) fn new(dyncfg: Arc, metrics: ReplicaMetrics) -> Self { + pub(super) fn new(metrics: ReplicaMetrics) -> Self { Self { - dyncfg, metrics, collections: Default::default(), hydration_queue: Default::default(), @@ -109,7 +111,13 @@ impl SequentialHydration { } /// Absorb a command the task intends to send, returning the commands it should actually send. - pub(super) fn absorb_command(&mut self, cmd: ComputeCommand) -> Vec { + /// + /// `dyncfg` is the replica's effective configuration, as maintained by the task. + pub(super) fn absorb_command( + &mut self, + cmd: ComputeCommand, + dyncfg: &ConfigSet, + ) -> Vec { // Whether to forward this command to the replica. let mut forward = true; @@ -149,12 +157,18 @@ impl SequentialHydration { } // Schedule collections that are ready now. - commands.extend(self.hydrate_collections()); + commands.extend(self.hydrate_collections(dyncfg)); commands } /// Observe a response the task received, returning the commands it should send in reaction. - pub(super) fn observe_response(&mut self, resp: &ComputeResponse) -> Vec { + /// + /// `dyncfg` is the replica's effective configuration, as maintained by the task. + pub(super) fn observe_response( + &mut self, + resp: &ComputeResponse, + dyncfg: &ConfigSet, + ) -> Vec { let mut commands = Vec::new(); if let ComputeResponse::Frontiers( @@ -191,7 +205,7 @@ impl SequentialHydration { // We freed some hydration capacity and may be able to start hydrating // new collections. drop(token); - commands.extend(self.hydrate_collections()); + commands.extend(self.hydrate_collections(dyncfg)); } } } else { @@ -204,10 +218,10 @@ impl SequentialHydration { } /// Allow hydration based on the available capacity, returning the `Schedule` commands to send. - fn hydrate_collections(&mut self) -> Vec { + fn hydrate_collections(&mut self, dyncfg: &ConfigSet) -> Vec { let mut commands = Vec::new(); - let capacity = HYDRATION_CONCURRENCY.get(&self.dyncfg); + let capacity = HYDRATION_CONCURRENCY.get(dyncfg); while self.hydration_count() < capacity { let Some(id) = self.hydration_queue.pop_front() else { // Hydration queue is empty. @@ -273,3 +287,79 @@ enum State { /// Collection is hydrating and waiting for hydration to complete. Hydrating(Token), } + +#[cfg(test)] +mod tests { + use mz_cluster_client::metrics::ControllerMetrics; + use mz_compute_types::ComputeInstanceId; + use mz_compute_types::dataflows::{DataflowDescription, IndexDesc}; + use mz_dyncfg::ConfigUpdates; + use mz_ore::metrics::MetricsRegistry; + use mz_repr::ReprRelationType; + + use crate::metrics::ComputeControllerMetrics; + use crate::protocol::command::ComputeParameters; + + use super::*; + + fn metrics() -> ReplicaMetrics { + let registry = MetricsRegistry::new(); + let shared = ControllerMetrics::new(®istry); + ComputeControllerMetrics::new(®istry, shared) + .for_instance(ComputeInstanceId::User(1)) + .for_replica(mz_cluster_client::ReplicaId::User(1)) + } + + /// A `CreateDataflow` command for a non-transient dataflow exporting `id`. + fn create_dataflow(id: GlobalId) -> ComputeCommand { + let mut desc = DataflowDescription::new("test".into()); + desc.as_of = Some(Antichain::from_elem(Timestamp::MIN)); + desc.index_exports.insert( + id, + ( + IndexDesc { + on_id: id, + key: Vec::new(), + }, + ReprRelationType::empty(), + ), + ); + ComputeCommand::CreateDataflow(Box::new(desc)) + } + + /// The interceptor enforces the hydration concurrency of the configuration it is handed, which + /// is the replica's own, specialized by the replica task. This is the regression guard for it + /// reading the environment-wide value instead, which would make the config's `Replica` scope + /// inert, given that it is enforced here and never read on the replica. + #[mz_ore::test] + fn hydration_concurrency_follows_supplied_config() { + let dyncfg = mz_dyncfgs::all_dyncfgs(); + let mut updates = ConfigUpdates::default(); + updates.add(&HYDRATION_CONCURRENCY, 1); + updates.apply(&dyncfg); + + let mut hydration = SequentialHydration::new(metrics()); + + let id1 = GlobalId::User(1); + let id2 = GlobalId::User(2); + for id in [id1, id2] { + let commands = hydration.absorb_command(create_dataflow(id), &dyncfg); + assert_eq!(commands, vec![create_dataflow(id)]); + } + + // At a concurrency of one, only the first `Schedule` is released. + let commands = hydration.absorb_command(ComputeCommand::Schedule(id1), &dyncfg); + assert_eq!(commands, vec![ComputeCommand::Schedule(id1)]); + let commands = hydration.absorb_command(ComputeCommand::Schedule(id2), &dyncfg); + assert_eq!(commands, vec![]); + + // Raising the concurrency in the supplied configuration releases the held-back command. + let mut updates = ConfigUpdates::default(); + updates.add(&HYDRATION_CONCURRENCY, 2); + updates.apply(&dyncfg); + + let update = ComputeCommand::UpdateConfiguration(Box::new(ComputeParameters::default())); + let commands = hydration.absorb_command(update.clone(), &dyncfg); + assert_eq!(commands, vec![update, ComputeCommand::Schedule(id2)]); + } +} diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index bfc0802b28541..ff03b29b64aed 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -21,6 +21,7 @@ pub const ENABLE_HALF_JOIN2: Config = Config::new( "enable_compute_half_join2", true, "Whether compute should use `half_join2` rather than DD's `half_join` to render delta joins.", + ParameterScope::Environment, ); /// Whether rendering should collapse error multiplicities to one where it arranges errors. @@ -40,6 +41,7 @@ pub const ENABLE_ERROR_DISTINCT: Config = Config::new( true, "Whether compute rendering should collapse error multiplicities to one where it arranges \ errors.", + ParameterScope::Environment, ); /// Use the column-paged merge batcher code path at arrange sites. When @@ -59,8 +61,8 @@ pub const ENABLE_COLUMN_PAGED_BATCHER: Config = Config::new( false, "Use the columnar-native paged merge batcher at arrange sites. When `false` (default), \ arranges fall back to the legacy columnation `Col2ValBatcher` / `RowRowBuilder` path.", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Allow the column-paged batcher's pager to evict chunks under memory /// pressure. Only meaningful when [`ENABLE_COLUMN_PAGED_BATCHER`] is `true`. @@ -81,8 +83,8 @@ pub const ENABLE_COLUMN_PAGED_BATCHER_SPILL: Config = Config::new( false, "Allow the column-paged batcher's pager to evict chunks under memory pressure. Only \ meaningful when `enable_column_paged_batcher = true`.", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Resident-bytes budget fraction for chunk spilling. Two consumers read /// it: the column pager's tiered policy multiplies it against the @@ -104,8 +106,8 @@ pub const COLUMN_PAGED_BATCHER_BUDGET_FRACTION: Config = Config::new( "Budget fraction for chunk spilling: the buffer pool multiplies it against physical \ RAM and the column pager's tiered policy against the announced memory limit. \ Total pool budget = max(ram * fraction, 128 MiB).", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Number of buffer-pool spill threads performing eviction I/O (lz4 /// compression plus the synchronous-reclaim `MADV_PAGEOUT`) off the threads @@ -118,8 +120,8 @@ pub const COLUMN_PAGED_BATCHER_SPILL_WORKER_COUNT: Config = Config::new( "column_paged_batcher_spill_worker_count", 2, "Buffer-pool spill threads for off-worker eviction I/O; 0 evicts inline on the caller.", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Compress chunks the column-paged batcher spills, using lz4. Only /// meaningful when [`ENABLE_COLUMN_PAGED_BATCHER_SPILL`] is `true`; the codec @@ -133,8 +135,8 @@ pub const COLUMN_PAGED_BATCHER_LZ4: Config = Config::new( false, "Compress column-paged batcher chunks with lz4 on the spill path. Only meaningful when \ `enable_column_paged_batcher_spill = true`.", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Proactively evict the column-paged batcher's lz4-compressed spill chunks /// from RSS via `MADV_PAGEOUT` when spilling to the swap backend. Only @@ -156,8 +158,8 @@ pub const COLUMN_PAGED_BATCHER_SWAP_PAGEOUT: Config = Config::new( "Eagerly evict the column-paged batcher's lz4-compressed swap-backend spill chunks from RSS \ via `MADV_PAGEOUT` (they otherwise receive no madvise and are reclaimed only lazily). Only \ meaningful when `column_paged_batcher_lz4 = true` and the swap backend is active.", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Eagerly compress unbacked buffer-pool chunks to `BackedResident` on idle /// spill threads (write-behind). The chunk stays readable in its slot while @@ -170,8 +172,8 @@ pub const COLUMN_PAGED_BATCHER_EAGER_BACKING: Config = Config::new( false, "Eagerly compress buffer-pool chunks to compressed-but-resident on idle spill threads, so \ budget-driven eviction is a pure page release. Only meaningful with spill workers.", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Ceiling on the buffer pool's total RSS, as a fraction of *physical RAM* /// (never the announced limit, which includes swap on swap-provisioned @@ -192,8 +194,8 @@ pub const COLUMN_PAGED_BATCHER_POOL_RSS_TARGET_FRACTION: Config = Config::n 0.25, "Ceiling on the buffer pool's total RSS as a fraction of physical RAM; the headroom above \ the slot budget holds compressed-but-resident extents. Zero pages extents out immediately.", -) -.scoped(ParameterScope::Replica); + ParameterScope::Replica, +); /// Whether rendering should use `mz_join_core` rather than DD's `JoinCore::join_core`. pub const ENABLE_MZ_JOIN_CORE: Config = Config::new( @@ -201,6 +203,7 @@ pub const ENABLE_MZ_JOIN_CORE: Config = Config::new( true, "Whether compute should use `mz_join_core` rather than DD's `JoinCore::join_core` to render \ linear joins.", + ParameterScope::Environment, ); /// Use sync Timely operators with Tokio tasks for the MV sink. @@ -208,6 +211,7 @@ pub const ENABLE_SYNC_MV_SINK: Config = Config::new( "enable_compute_sync_mv_sink", false, "Use sync Timely operators with Tokio tasks for the MV sink.", + ParameterScope::Environment, ); /// Whether rendering should use the new MV sink correction buffer implementation. @@ -215,6 +219,7 @@ pub const ENABLE_CORRECTION_V2: Config = Config::new( "enable_compute_correction_v2", true, "Whether compute should use the new MV sink correction buffer implementation.", + ParameterScope::Environment, ); /// The size factor of subsequent chains in the correction V2 buffer. @@ -222,6 +227,7 @@ pub const CORRECTION_V2_CHAIN_PROPORTIONALITY: Config = Config::new( "compute_correction_v2_chain_proportionality", 3.0, "The size factor of subsequent chains in the correction V2 buffer.", + ParameterScope::Replica, ); /// The byte size of chunks in the correction V2 buffer. @@ -229,6 +235,7 @@ pub const CORRECTION_V2_CHUNK_SIZE: Config = Config::new( "compute_correction_v2_chunk_size", 8 * 1024, "The byte size of chunks in the correction V2 buffer.", + ParameterScope::Replica, ); /// Whether to enable temporal bucketing in compute. @@ -236,6 +243,7 @@ pub const ENABLE_COMPUTE_TEMPORAL_BUCKETING: Config = Config::new( "enable_compute_temporal_bucketing", false, "Whether to enable temporal bucketing in compute.", + ParameterScope::Environment, ); /// The summary to apply to the frontier in temporal bucketing in compute. @@ -243,6 +251,7 @@ pub const TEMPORAL_BUCKETING_SUMMARY: Config = Config::new( "compute_temporal_bucketing_summary", Duration::from_secs(2), "The summary to apply to frontiers in temporal bucketing in compute.", + ParameterScope::Environment, ); /// The yielding behavior with which linear joins should be rendered. @@ -253,17 +262,23 @@ pub const LINEAR_JOIN_YIELDING: Config<&str> = Config::new( 'work:' or 'time:' or 'work:,time:'. Note \ that omitting one of 'work' or 'time' will entirely disable join yielding by time or \ work, respectively, rather than falling back to some default.", + ParameterScope::Replica, ); /// Enable lgalloc. -pub const ENABLE_LGALLOC: Config = - Config::new("enable_lgalloc", true, "Enable lgalloc.").scoped(ParameterScope::Replica); +pub const ENABLE_LGALLOC: Config = Config::new( + "enable_lgalloc", + true, + "Enable lgalloc.", + ParameterScope::Replica, +); /// Enable lgalloc's eager memory return/reclamation feature. pub const ENABLE_LGALLOC_EAGER_RECLAMATION: Config = Config::new( "enable_lgalloc_eager_reclamation", true, "Enable lgalloc's eager return behavior.", + ParameterScope::Replica, ); /// The interval at which the background thread wakes. @@ -271,6 +286,7 @@ pub const LGALLOC_BACKGROUND_INTERVAL: Config = Config::new( "lgalloc_background_interval", Duration::from_secs(1), "Scheduling interval for lgalloc's background worker.", + ParameterScope::Replica, ); /// Enable lgalloc's eager memory return/reclamation feature. @@ -278,6 +294,7 @@ pub const LGALLOC_FILE_GROWTH_DAMPENER: Config = Config::new( "lgalloc_file_growth_dampener", 2, "Lgalloc's file growth dampener parameter.", + ParameterScope::Replica, ); /// Enable lgalloc's eager memory return/reclamation feature. @@ -285,6 +302,7 @@ pub const LGALLOC_LOCAL_BUFFER_BYTES: Config = Config::new( "lgalloc_local_buffer_bytes", 64 << 20, "Lgalloc's local buffer bytes parameter.", + ParameterScope::Replica, ); /// The bytes to reclaim (slow path) per size class, for each background thread activation. @@ -292,6 +310,7 @@ pub const LGALLOC_SLOW_CLEAR_BYTES: Config = Config::new( "lgalloc_slow_clear_bytes", 128 << 20, "Clear byte size per size class for every invocation", + ParameterScope::Replica, ); /// Interval to run the memory limiter. A zero duration disables the limiter. @@ -299,6 +318,7 @@ pub const MEMORY_LIMITER_INTERVAL: Config = Config::new( "memory_limiter_interval", Duration::from_secs(10), "Interval to run the memory limiter. A zero duration disables the limiter.", + ParameterScope::Replica, ); /// Bias to the memory limiter usage factor. @@ -306,6 +326,7 @@ pub const MEMORY_LIMITER_USAGE_BIAS: Config = Config::new( "memory_limiter_usage_bias", 1., "Multiplicative bias to the memory limiter's limit.", + ParameterScope::Replica, ); /// Burst factor to memory limit. @@ -313,6 +334,7 @@ pub const MEMORY_LIMITER_BURST_FACTOR: Config = Config::new( "memory_limiter_burst_factor", 0., "Multiplicative burst factor to the memory limiter's limit.", + ParameterScope::Replica, ); /// Enable lgalloc for columnation. @@ -320,6 +342,7 @@ pub const ENABLE_COLUMNATION_LGALLOC: Config = Config::new( "enable_columnation_lgalloc", true, "Enable allocating regions from lgalloc.", + ParameterScope::Replica, ); /// The interval at which the compute server performs maintenance tasks. @@ -327,6 +350,7 @@ pub const COMPUTE_SERVER_MAINTENANCE_INTERVAL: Config = Config::new( "compute_server_maintenance_interval", Duration::from_millis(10), "The interval at which the compute server performs maintenance tasks. Zero enables maintenance on every iteration.", + ParameterScope::Replica, ); /// Maximum number of in-flight bytes emitted by persist_sources feeding dataflows. @@ -335,6 +359,7 @@ pub const DATAFLOW_MAX_INFLIGHT_BYTES: Config> = Config::new( None, "The maximum number of in-flight bytes emitted by persist_sources feeding \ compute dataflows in non-cc clusters.", + ParameterScope::Replica, ); /// The "physical backpressure" of `compute_dataflow_max_inflight_bytes_cc` has @@ -346,6 +371,7 @@ pub const DATAFLOW_MAX_INFLIGHT_BYTES_CC: Config> = Config::new( None, "The maximum number of in-flight bytes emitted by persist_sources feeding \ compute dataflows in cc clusters.", + ParameterScope::Replica, ); /// The term `n` in the growth rate `1 + 1/(n + 1)` for `ConsolidatingVec`. @@ -354,13 +380,21 @@ pub const CONSOLIDATING_VEC_GROWTH_DAMPENER: Config = Config::new( "consolidating_vec_growth_dampener", 1, "Dampener in growth rate for consolidating vector size", + ParameterScope::Replica, ); /// The number of dataflows that may hydrate concurrently. +/// +/// Enforced in `environmentd`, by the controller's per-replica hydration +/// interceptor withholding `Schedule` commands, rather than by the replica. The +/// interceptor resolves it from the configuration commands it observes, which +/// are already specialized for its replica, so the limit still follows the +/// replica's scoped override. pub const HYDRATION_CONCURRENCY: Config = Config::new( "compute_hydration_concurrency", 4, "Controls how many compute dataflows may hydrate concurrently.", + ParameterScope::Replica, ); /// See `src/storage-operators/src/s3_oneshot_sink/parquet.rs` for more details. @@ -369,6 +403,7 @@ pub const COPY_TO_S3_PARQUET_ROW_GROUP_FILE_RATIO: Config = Config::new( 20, "The ratio (defined as a percentage) of row-group size to max-file-size. \ Must be <= 100.", + ParameterScope::Environment, ); /// See `src/storage-operators/src/s3_oneshot_sink/parquet.rs` for more details. @@ -377,6 +412,7 @@ pub const COPY_TO_S3_ARROW_BUILDER_BUFFER_RATIO: Config = Config::new( 150, "The ratio (defined as a percentage) of arrow-builder size to row-group size. \ Must be >= 100.", + ParameterScope::Environment, ); /// The size of each part in the multi-part upload to use when uploading files to S3. @@ -384,15 +420,21 @@ pub const COPY_TO_S3_MULTIPART_PART_SIZE_BYTES: Config = Config::new( "copy_to_s3_multipart_part_size_bytes", 1024 * 1024 * 8, "The size of each part in a multipart upload to S3.", + ParameterScope::Environment, ); /// Main switch to enable or disable replica expiration. /// /// Changes affect existing replicas only after restart. +/// +/// The env-wide kill switch for the feature, read in `environmentd` when +/// specializing `CreateInstance` for a replica. [`COMPUTE_REPLICA_EXPIRATION_OFFSET`] +/// is the replica-scoped half of the pair. pub const ENABLE_COMPUTE_REPLICA_EXPIRATION: Config = Config::new( "enable_compute_replica_expiration", true, "Main switch to disable replica expiration.", + ParameterScope::Environment, ); /// The maximum lifetime of a replica configured as an offset to the replica start time. @@ -404,6 +446,7 @@ pub const COMPUTE_REPLICA_EXPIRATION_OFFSET: Config = Config::new( "compute_replica_expiration_offset", Duration::ZERO, "The expiration time offset for replicas. Zero disables expiration.", + ParameterScope::Replica, ); /// When enabled, applies the column demands from a MapFilterProject onto the RelationDesc used to @@ -413,6 +456,7 @@ pub const COMPUTE_APPLY_COLUMN_DEMANDS: Config = Config::new( "compute_apply_column_demands", true, "When enabled, passes applys column demands to the RelationDesc used to read out of Persist.", + ParameterScope::Environment, ); /// The amount of output the flat-map operator produces before yielding. Set to a high value to @@ -421,6 +465,7 @@ pub const COMPUTE_FLAT_MAP_FUEL: Config = Config::new( "compute_flat_map_fuel", 1_000_000, "The amount of output the flat-map operator produces before yielding.", + ParameterScope::Replica, ); /// Whether to render `as_specific_collection` using a fueled flat-map operator. @@ -428,6 +473,7 @@ pub const ENABLE_COMPUTE_RENDER_FUELED_AS_SPECIFIC_COLLECTION: Config = Co "enable_compute_render_fueled_as_specific_collection", true, "When enabled, renders `as_specific_collection` using a fueled flat-map operator.", + ParameterScope::Environment, ); /// Whether to apply logical backpressure in compute dataflows. @@ -435,6 +481,7 @@ pub const ENABLE_COMPUTE_LOGICAL_BACKPRESSURE: Config = Config::new( "enable_compute_logical_backpressure", false, "When enabled, compute dataflows will apply logical backpressure.", + ParameterScope::Replica, ); /// Maximal number of capabilities retained by the logical backpressure operator. @@ -450,6 +497,7 @@ pub const COMPUTE_LOGICAL_BACKPRESSURE_MAX_RETAINED_CAPABILITIES: Config = Config "compute_logical_backpressure_inflight_slack", Duration::from_secs(1), "Round observed timestamps to slack.", + ParameterScope::Replica, ); /// Enable per-column dictionary compression for row containers in arrangements. @@ -474,6 +523,7 @@ pub const ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA: Config = Config "enable_arrangement_dictionary_compression_alpha", true, "Enable arrangement dictionary compression (alpha; not yet production-ready).", + ParameterScope::Replica, ); /// Whether to enable the peek response stash, for sending back large peek @@ -483,6 +533,7 @@ pub const ENABLE_PEEK_RESPONSE_STASH: Config = Config::new( "enable_compute_peek_response_stash", true, "Whether to enable the peek response stash, for sending back large peek responses. Will only be used for results that exceed compute_peek_response_stash_threshold_bytes.", + ParameterScope::Environment, ); /// The threshold for peek response size above which we should use the peek @@ -492,6 +543,7 @@ pub const PEEK_RESPONSE_STASH_THRESHOLD_BYTES: Config = Config::new( "compute_peek_response_stash_threshold_bytes", 1024 * 10, /* 10KB */ "The threshold above which to use the peek response stash, for sending back large peek responses.", + ParameterScope::Environment, ); /// The target number of maximum runs in the batches written to the stash. @@ -506,6 +558,7 @@ pub const PEEK_RESPONSE_STASH_BATCH_MAX_RUNS: Config = Config::new( // `clusterd` side. 2, "The target number of maximum runs in the batches written to the stash.", + ParameterScope::Environment, ); /// The target size for batches of rows we read out of the peek stash. @@ -513,6 +566,7 @@ pub const PEEK_RESPONSE_STASH_READ_BATCH_SIZE_BYTES: Config = Config::new "compute_peek_response_stash_read_batch_size_bytes", 1024 * 1024 * 100, /* 100mb */ "The target size for batches of rows we read out of the peek stash.", + ParameterScope::Environment, ); /// The memory budget for consolidating stashed peek responses in @@ -521,6 +575,7 @@ pub const PEEK_RESPONSE_STASH_READ_MEMORY_BUDGET_BYTES: Config = Config:: "compute_peek_response_stash_read_memory_budget_bytes", 1024 * 1024 * 64, /* 64mb */ "The memory budget for consolidating stashed peek responses in environmentd.", + ParameterScope::Environment, ); /// The number of batches to pump from the peek result iterator when stashing peek responses. @@ -528,6 +583,7 @@ pub const PEEK_STASH_NUM_BATCHES: Config = Config::new( "compute_peek_stash_num_batches", 100, "The number of batches to pump from the peek result iterator (in one iteration through the worker loop) when stashing peek responses.", + ParameterScope::Environment, ); /// The size of each batch, as number of rows, pumped from the peek result @@ -536,6 +592,7 @@ pub const PEEK_STASH_BATCH_SIZE: Config = Config::new( "compute_peek_stash_batch_size", 100000, "The size, as number of rows, of each batch pumped from the peek result iterator (in one iteration through the worker loop) when stashing peek responses.", + ParameterScope::Environment, ); /// The collection interval for the Prometheus metrics introspection source. @@ -545,13 +602,25 @@ pub const COMPUTE_PROMETHEUS_INTROSPECTION_SCRAPE_INTERVAL: Config = C "compute_prometheus_introspection_scrape_interval", Duration::from_secs(10), "The collection interval for the Prometheus metrics introspection source. Set to zero to disable.", + ParameterScope::Replica, ); /// If set, skip fetching or processing the snapshot data for subscribes when possible. +/// +/// Read twice. At plan time in `environmentd` it gates whether snapshot elision runs at all, and +/// at render time on the replica it gates whether an elided snapshot is honored. The replica-side +/// read only ever puts a snapshot back, never takes one away, so the two reads disagreeing costs +/// work rather than correctness. +/// +/// Environment-scoped because the plan-time read has no replica in scope. Making it +/// cluster-coherent instead would need plan-time resolution of cluster overrides for +/// `OptimizerConfig` fields that are not `OptimizerFeatures`, which is the only place cluster +/// overrides are resolved today. pub const SUBSCRIBE_SNAPSHOT_OPTIMIZATION: Config = Config::new( "compute_subscribe_snapshot_optimization", true, "If set, skip fetching or processing the snapshot data for subscribes when possible.", + ParameterScope::Environment, ); /// Temporary flag to de-risk the rollout of a release-blocker fix. @@ -561,6 +630,7 @@ pub const MV_SINK_ADVANCE_PERSIST_FRONTIERS: Config = Config::new( "compute_mv_sink_advance_persist_frontiers", true, "Whether the MV sink's write operator advances its internal persist frontiers to the as_of.", + ParameterScope::Environment, ); /// Adds the full set of all compute `Config`s. diff --git a/src/controller-types/src/dyncfgs.rs b/src/controller-types/src/dyncfgs.rs index c3ff6909a7edd..7a37b862df65f 100644 --- a/src/controller-types/src/dyncfgs.rs +++ b/src/controller-types/src/dyncfgs.rs @@ -11,61 +11,76 @@ use std::time::Duration; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; /// The interval at which to retry cleaning replicas from past generatinos. pub const CONTROLLER_PAST_GENERATION_REPLICA_CLEANUP_RETRY_INTERVAL: Config = Config::new( "controller_past_generation_replica_cleanup_retry_interval", Duration::from_secs(300), "The interval at which to attempt to retry cleaning up replicas from past generations.", + ParameterScope::Environment, ); pub const ENABLE_0DT_DEPLOYMENT_SOURCES: Config = Config::new( "enable_0dt_deployment_sources", true, "Whether to enable zero-downtime deployments for sources that support it (experimental).", + ParameterScope::Environment, ); pub const WALLCLOCK_LAG_RECORDING_INTERVAL: Config = Config::new( "wallclock_lag_recording_interval", Duration::from_secs(60), "The interval at which to record `WallclockLagHistory` introspection.", + ParameterScope::Environment, ); pub const WALLCLOCK_LAG_HISTOGRAM_PERIOD_INTERVAL: Config = Config::new( "wallclock_lag_histogram_period_interval", Duration::from_secs(24 * 60 * 60), "The period interval of histograms in `WallclockLagHistogram` introspection.", + ParameterScope::Environment, ); +// The four configs below make up the `TimelyConfig` of a replica's `clusterd` +// processes. They are replica-scoped: `Controller::provision_replica` resolves +// them against the replica's scoped overrides and bakes the result into the +// process configuration, so a change reaches a replica only when it is next +// provisioned. + pub const ENABLE_TIMELY_ZERO_COPY: Config = Config::new( "enable_timely_zero_copy", false, "Enable the zero copy allocator (timely dataflow).", + ParameterScope::Replica, ); pub const ENABLE_TIMELY_ZERO_COPY_LGALLOC: Config = Config::new( "enable_timely_zero_copy_lgalloc", false, "Enable backing the zero copy allocator with lgalloc (timely dataflow).", + ParameterScope::Replica, ); pub const TIMELY_ZERO_COPY_LIMIT: Config> = Config::new( "timely_zero_copy_limit", None, "Optional limit of the zero copy allocator in allocations (timely dataflow).", + ParameterScope::Replica, ); pub const ARRANGEMENT_EXERT_PROPORTIONALITY: Config = Config::new( "arrangement_exert_proportionality", 16, "Value that controls how much merge effort to exert on arrangements.", + ParameterScope::Replica, ); pub const ENABLE_PAUSED_CLUSTER_READHOLD_DOWNGRADE: Config = Config::new( "enable_paused_cluster_readhold_downgrade", true, "Aggressively downgrade input read holds for indexes on zero-replica clusters.", + ParameterScope::Environment, ); /// Adds the full set of all controller `Config`s. diff --git a/src/controller/src/clusters.rs b/src/controller/src/clusters.rs index 01ad1c3c78a1f..3bdc4b9c2ce28 100644 --- a/src/controller/src/clusters.rs +++ b/src/controller/src/clusters.rs @@ -534,6 +534,11 @@ impl Controller { self.replica_http_locator .remove_replica(cluster_id, replica_id); + // The coordinator only re-pushes the override map when the scoped + // configuration itself changes, so a dropped replica's entry would + // otherwise be retained until the next such change. + self.replica_dyncfg_overrides.remove(&replica_id); + self.compute.drop_replica(cluster_id, replica_id)?; self.storage.drop_replica(cluster_id, replica_id); Ok(()) @@ -703,11 +708,20 @@ impl Controller { arrangement_exert_proportionality: 1337, ..Default::default() }; + // These configure the replica's process rather than environmentd's, so + // they are `ParameterScope::Replica` and must be read through this + // replica's scoped overrides. They are baked into the process + // configuration at provisioning time, so a later change to either the + // environment-wide value or the override reaches the replica only when + // it is next provisioned. + let overrides = self.replica_dyncfg_overrides.get(&replica_id); let compute_proto_timely_config = TimelyConfig { - arrangement_exert_proportionality: ARRANGEMENT_EXERT_PROPORTIONALITY.get(&self.dyncfg), - enable_zero_copy: ENABLE_TIMELY_ZERO_COPY.get(&self.dyncfg), - enable_zero_copy_lgalloc: ENABLE_TIMELY_ZERO_COPY_LGALLOC.get(&self.dyncfg), - zero_copy_limit: TIMELY_ZERO_COPY_LIMIT.get(&self.dyncfg), + arrangement_exert_proportionality: ARRANGEMENT_EXERT_PROPORTIONALITY + .get_with_overrides(&self.dyncfg, overrides), + enable_zero_copy: ENABLE_TIMELY_ZERO_COPY.get_with_overrides(&self.dyncfg, overrides), + enable_zero_copy_lgalloc: ENABLE_TIMELY_ZERO_COPY_LGALLOC + .get_with_overrides(&self.dyncfg, overrides), + zero_copy_limit: TIMELY_ZERO_COPY_LIMIT.get_with_overrides(&self.dyncfg, overrides), ..Default::default() }; diff --git a/src/controller/src/lib.rs b/src/controller/src/lib.rs index dbb28dfc3d016..6e675eb3001d1 100644 --- a/src/controller/src/lib.rs +++ b/src/controller/src/lib.rs @@ -37,7 +37,7 @@ use mz_compute_client::controller::{ ComputeController, ComputeControllerResponse, PeekNotification, }; use mz_compute_client::protocol::response::SubscribeBatch; -use mz_controller_types::WatchSetId; +use mz_controller_types::{ClusterId, WatchSetId}; use mz_dyncfg::{ConfigSet, ConfigUpdates}; use mz_orchestrator::{NamespacedOrchestrator, Orchestrator, ServiceProcessMetrics}; use mz_ore::cast::CastFrom; @@ -203,6 +203,12 @@ pub struct Controller { /// Dynamic system configuration. dyncfg: ConfigSet, + /// The replica-local scoped overrides of [`Self::dyncfg`], by replica. + /// + /// Sparse: only replicas LaunchDarkly targets to a replica-specific value + /// have an entry. See [`Self::update_replica_dyncfg_overrides`]. + replica_dyncfg_overrides: BTreeMap, + /// Locator for HTTP addresses of cluster replicas. replica_http_locator: Arc, } @@ -213,6 +219,36 @@ impl Controller { updates.apply(&self.dyncfg); } + /// Replaces the per-replica dyncfg overrides of the replica-local scoped + /// system parameters, in this controller and in the compute and storage + /// controllers beneath it. + /// + /// Replicas absent from `overrides` have their overrides cleared, so a + /// replica that no longer has one reverts to the environment-wide + /// configuration. Callers should follow with a configuration push so + /// running replicas observe the new values. + /// + /// The three layers realize a [`ParameterScope::Replica`] config in + /// different places, which is why all three are fed from one call. The + /// compute and storage controllers specialize the configuration they push + /// to a running replica. This controller resolves the overrides that are + /// baked into a replica's process configuration when it is provisioned. + /// + /// [`ParameterScope::Replica`]: mz_dyncfg::ParameterScope::Replica + pub fn update_replica_dyncfg_overrides( + &mut self, + overrides: BTreeMap>, + ) { + self.replica_dyncfg_overrides = overrides + .values() + .flat_map(|replicas| replicas.iter()) + .map(|(replica_id, updates)| (*replica_id, updates.clone())) + .collect(); + self.compute + .update_replica_dyncfg_overrides(overrides.clone()); + self.storage.update_replica_dyncfg_overrides(overrides); + } + /// Start sinking the compute controller's introspection data into storage. /// /// This method should be called once the introspection collections have been registered with @@ -266,6 +302,7 @@ impl Controller { watch_set_id_gen: _, immediate_watch_sets, dyncfg: _, + replica_dyncfg_overrides: _, replica_http_locator: _, } = self; @@ -717,6 +754,7 @@ impl Controller { watch_set_id_gen: Gen::default(), immediate_watch_sets: Vec::new(), dyncfg: mz_dyncfgs::all_dyncfgs(), + replica_dyncfg_overrides: BTreeMap::new(), replica_http_locator: config.replica_http_locator, }; diff --git a/src/dyncfg-file/src/lib.rs b/src/dyncfg-file/src/lib.rs index c05c11b25879f..acb3115302a96 100644 --- a/src/dyncfg-file/src/lib.rs +++ b/src/dyncfg-file/src/lib.rs @@ -191,7 +191,7 @@ fn json_to_config_val(json: &JsonValue, template: &ConfigVal) -> Result = Config::new("test_bool", true, "A test boolean config"); - const STRING_CONFIG: Config<&str> = - Config::new("test_string", "default", "A test string config"); + const BOOL_CONFIG: Config = Config::new( + "test_bool", + true, + "A test boolean config", + ParameterScope::Environment, + ); + const STRING_CONFIG: Config<&str> = Config::new( + "test_string", + "default", + "A test string config", + ParameterScope::Environment, + ); let set = ConfigSet::default().add(&BOOL_CONFIG).add(&STRING_CONFIG); // Start sync with empty file (should create it) @@ -247,8 +256,12 @@ mod tests { #[mz_ore::test(tokio::test)] async fn test_file_sync_opt_string() { - const OPT_STRING_CONFIG: Config> = - Config::new("test_opt_string", None, "A test optional string config"); + const OPT_STRING_CONFIG: Config> = Config::new( + "test_opt_string", + None, + "A test optional string config", + ParameterScope::Environment, + ); let set = ConfigSet::default().add(&OPT_STRING_CONFIG); let mut config_file = tempfile::NamedTempFile::new().unwrap(); diff --git a/src/dyncfg/src/lib.rs b/src/dyncfg/src/lib.rs index 0ae19678a82bd..afedcbdd4cba4 100644 --- a/src/dyncfg/src/lib.rs +++ b/src/dyncfg/src/lib.rs @@ -19,8 +19,9 @@ //! set the value of `Config`. //! //! ``` -//! # use mz_dyncfg::{Config, ConfigSet}; -//! const FOO: Config = Config::new("foo", false, "description of foo"); +//! # use mz_dyncfg::{Config, ConfigSet, ParameterScope}; +//! const FOO: Config = +//! Config::new("foo", false, "description of foo", ParameterScope::Environment); //! fn bar(cfg: &ConfigSet) { //! assert_eq!(FOO.get(&cfg), false); //! } @@ -71,10 +72,73 @@ use tracing::error; /// The declaration is the single source of truth for which contexts the /// LaunchDarkly sync loop evaluates and where the resolved value may be /// overridden. See `doc/developer/design/20260609_scoped_feature_flags.md`. +/// +/// [`Environment`] is the safe declaration, and the right one when in doubt. It +/// preserves the unscoped behavior of a single value everywhere. A finer scope +/// *enables* divergence, so declaring one asserts that divergence is both safe +/// and useful for this config. Getting that wrong introduces a way to break an +/// environment that did not previously exist, while erring toward +/// [`Environment`] only forgoes a capability. Prefer forgoing the capability. +/// +/// Where a config is *realized* is therefore a necessary condition for a finer +/// scope, not a sufficient one. A config qualifies for [`Replica`] only if it is +/// realized per replica, and should be declared [`Replica`] only if it also +/// tunes that replica process's own resource usage (memory, CPU, I/O, +/// concurrency, timing) and cannot change what a dataflow produces, what reaches +/// durable state, or anything externally visible. `lgalloc`, the memory limiter, +/// the spill and pager knobs, and the timely zero-copy settings are the shape to +/// match. +/// +/// In particular, declare [`Environment`] for a flag selecting a different +/// implementation or code path whose output-equivalence is an assumption rather +/// than a guarantee. Where the assumption holds, [`Environment`] costs nothing. +/// Where it does not, a per-replica rollout turns one bug into query results +/// that differ by which replica served them. +/// +/// Which contexts a config can be realized in: +/// +/// - A config realized inside a `clusterd` process is a [`Replica`] candidate. +/// That covers the compute and storage worker config sets and `mz_metrics`, +/// which the per-replica dyncfg push reaches. `environmentd` may read such a +/// config too, for its own process. That read legitimately sees the +/// environment-wide value. +/// - A config that `environmentd` resolves for *one specific replica*, whether +/// it ships the value there or acts on it itself, is also a candidate. The +/// read site has to resolve that replica's override, either with +/// [`Config::get_with_overrides`] or from a per-replica config set. Without +/// that, the override never takes effect and the declaration is a silent +/// no-op. +/// - A config realized in `environmentd` with no single replica in scope is +/// [`Environment`]. So is every `balancerd` config. `balancerd` syncs +/// LaunchDarkly itself, against a `balancer` context keyed by cloud provider, +/// region and build version, and has no environment, cluster or replica +/// beneath it to target. +/// - A config consumed at plan time, once per cluster, is [`Cluster`]. +/// +/// NOTE: a config whose value must agree across the replicas of one cluster +/// (because they render the same dataflow and their outputs are compared) is +/// [`Environment`] even when it is read on `clusterd`. Per-replica divergence in +/// *how* a dataflow is rendered is fine, in *what* it produces is not. +/// +/// NOTE: persist configs are [`Environment`] as a class, even though the persist +/// client runs on `clusterd` and the per-replica push reaches its config set. +/// The same client code also runs in `environmentd`, and every copy of it acts +/// on shared durable state. A replica-scoped persist config could never reach +/// the `environmentd` client, so a rollout targeting replicas would leave a +/// shard's other writer on the old value indefinitely. [`Environment`] is the +/// only scope covering every persist client in an environment. +/// +/// [`Cluster`]: ParameterScope::Cluster +/// [`Environment`]: ParameterScope::Environment +/// [`Replica`]: ParameterScope::Replica #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum ParameterScope { /// Environment-wide only; no cluster/replica overrides. The default, so all /// existing synced parameters are unchanged. + /// + /// NOTE: this names the coarsest targeting granularity, not the + /// `environmentd` process. A config a process other than `environmentd` + /// resolves for itself, with nothing finer beneath it, is `Environment`. Environment, /// Cluster-coherent: env-wide base plus per-cluster overrides. Evaluated /// with the `cluster` context (replica-free) and resolved at plan time via @@ -132,6 +196,11 @@ impl Config { /// It is best practice, but not strictly required, for the name to be /// globally unique within a process. /// + /// `scope` declares where the config's value may be overridden. It is a + /// required parameter rather than a builder step so that every new config + /// makes the choice deliberately. [`ParameterScope`] documents how to pick + /// one from the config's read sites. + /// /// TODO(cfg): Add some sort of categorization of config purpose here: e.g. /// limited-lifetime rollout flag, CYA, magic number that we never expect to /// tune, magic number that we DO expect to tune, etc. This could be used to @@ -141,26 +210,20 @@ impl Config { /// TODO(cfg): See if we can make this more Rust-y and take these params as /// a struct (the obvious thing hits some issues with const combined with /// Drop). - pub const fn new(name: &'static str, default: D, desc: &'static str) -> Self { + pub const fn new( + name: &'static str, + default: D, + desc: &'static str, + scope: ParameterScope, + ) -> Self { Config { name, default, desc, - scope: ParameterScope::DEFAULT, + scope, } } - /// Declares the [`ParameterScope`] of this config, overriding the - /// [default](ParameterScope::DEFAULT). - /// - /// Use this to mark a config as cluster-coherent or replica-local so the - /// LaunchDarkly sync loop evaluates the appropriate scoped contexts and - /// resolution applies the override at the right boundary. - pub const fn scoped(mut self, scope: ParameterScope) -> Self { - self.scope = scope; - self - } - /// The name of this config. pub fn name(&self) -> &str { self.name @@ -194,6 +257,38 @@ impl Config { D::ConfigType::from_val(self.shared(set).load()) } + /// Returns the value of this config within the given set, with `overrides` + /// layered on top. + /// + /// This is how `environmentd` must read a [`ParameterScope::Replica`] config + /// whose value it ships to one specific replica: the set holds the + /// environment-wide value and `overrides` holds that replica's scoped + /// overrides, which win. Reading such a config with [`Self::get`] instead + /// makes its scope declaration a silent no-op. + /// + /// Panics if this config was not previously registered to the set. An + /// override whose type does not match the config's is logged and ignored, + /// rather than panicking a read site that is often on a critical path. + pub fn get_with_overrides( + &self, + set: &ConfigSet, + overrides: Option<&ConfigUpdates>, + ) -> D::ConfigType { + let val = self.shared(set).load(); + let val = match overrides.and_then(|o| o.updates.get(self.name)) { + None => val, + Some(o) if std::mem::discriminant(o) == std::mem::discriminant(&val) => o.clone(), + Some(o) => { + error!( + "override {:?} for config {} does not match its type {:?}", + o, self.name, val + ); + val + } + }; + D::ConfigType::from_val(val) + } + /// Returns a handle to the value of this config in the given set. /// /// This allows users to amortize the cost of the name lookup. @@ -830,16 +925,27 @@ mod tests { use mz_ore::assert_err; - const BOOL: Config = Config::new("bool", true, ""); - const U32: Config = Config::new("u32", 4, ""); - const USIZE: Config = Config::new("usize", 1, ""); - const OPT_USIZE: Config> = Config::new("opt_usize", Some(2), ""); - const F64: Config = Config::new("f64", 5.0, ""); - const STRING: Config<&str> = Config::new("string", "a", ""); - const OPT_STRING: Config> = Config::new("opt_string", Some("a"), ""); - const DURATION: Config = Config::new("duration", Duration::from_nanos(3), ""); - const JSON: Config serde_json::Value> = - Config::new("json", || serde_json::json!({}), ""); + const BOOL: Config = Config::new("bool", true, "", ParameterScope::Environment); + const U32: Config = Config::new("u32", 4, "", ParameterScope::Environment); + const USIZE: Config = Config::new("usize", 1, "", ParameterScope::Environment); + const OPT_USIZE: Config> = + Config::new("opt_usize", Some(2), "", ParameterScope::Environment); + const F64: Config = Config::new("f64", 5.0, "", ParameterScope::Environment); + const STRING: Config<&str> = Config::new("string", "a", "", ParameterScope::Environment); + const OPT_STRING: Config> = + Config::new("opt_string", Some("a"), "", ParameterScope::Environment); + const DURATION: Config = Config::new( + "duration", + Duration::from_nanos(3), + "", + ParameterScope::Environment, + ); + const JSON: Config serde_json::Value> = Config::new( + "json", + || serde_json::json!({}), + "", + ParameterScope::Environment, + ); #[mz_ore::test] fn all_types() { @@ -888,12 +994,17 @@ mod tests { #[mz_ore::test] fn fn_default() { - const BOOL_FN_DEFAULT: Config bool> = Config::new("bool", || !true, ""); + const BOOL_FN_DEFAULT: Config bool> = + Config::new("bool", || !true, "", ParameterScope::Environment); const STRING_FN_DEFAULT: Config String> = - Config::new("string", || "x".repeat(3), ""); + Config::new("string", || "x".repeat(3), "", ParameterScope::Environment); - const OPT_STRING_FN_DEFAULT: Config Option> = - Config::new("opt_string", || Some("x".repeat(3)), ""); + const OPT_STRING_FN_DEFAULT: Config Option> = Config::new( + "opt_string", + || Some("x".repeat(3)), + "", + ParameterScope::Environment, + ); let configs = ConfigSet::default() .add(&BOOL_FN_DEFAULT) @@ -933,6 +1044,28 @@ mod tests { assert_eq!(USIZE.get(&c1), 2); } + #[mz_ore::test] + fn get_with_overrides() { + let configs = ConfigSet::default().add(&USIZE).add(&STRING); + + // No overrides at all, and an override map that doesn't mention the + // config, both resolve to the set's value. + assert_eq!(USIZE.get_with_overrides(&configs, None), 1); + let mut overrides = ConfigUpdates::default(); + overrides.add(&STRING, "b"); + assert_eq!(USIZE.get_with_overrides(&configs, Some(&overrides)), 1); + + // An override wins over the set's value, without disturbing it. + overrides.add(&USIZE, 2); + assert_eq!(USIZE.get_with_overrides(&configs, Some(&overrides)), 2); + assert_eq!(USIZE.get(&configs), 1); + + // An override of the wrong type is ignored rather than panicking. + let mut mistyped = ConfigUpdates::default(); + mistyped.add_dynamic(USIZE.name(), ConfigVal::Bool(true)); + assert_eq!(USIZE.get_with_overrides(&configs, Some(&mistyped)), 1); + } + #[mz_ore::test] fn config_updates_extend() { // Regression test for database-issues#7793. diff --git a/src/metrics/src/dyncfgs.rs b/src/metrics/src/dyncfgs.rs index 7d44206b35dd2..f75c44a314ee6 100644 --- a/src/metrics/src/dyncfgs.rs +++ b/src/metrics/src/dyncfgs.rs @@ -11,13 +11,14 @@ use std::time::Duration; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; /// How frequently to refresh lgalloc map stats. pub(crate) const MZ_METRICS_LGALLOC_MAP_REFRESH_INTERVAL: Config = Config::new( "mz_metrics_lgalloc_map_refresh_interval", Duration::from_secs(0), "How frequently to refresh lgalloc stats. A zero duration disables refreshing.", + ParameterScope::Replica, ); /// How frequently to refresh lgalloc stats. @@ -25,6 +26,7 @@ pub(crate) const MZ_METRICS_LGALLOC_REFRESH_INTERVAL: Config = Config: "mz_metrics_lgalloc_refresh_interval", Duration::from_secs(30), "How frequently to refresh lgalloc stats. A zero duration disables refreshing.", + ParameterScope::Replica, ); /// How frequently to refresh lgalloc stats. @@ -32,6 +34,7 @@ pub(crate) const MZ_METRICS_RUSAGE_REFRESH_INTERVAL: Config = Config:: "mz_metrics_rusage_refresh_interval", Duration::from_secs(30), "How frequently to refresh rusage stats. A zero duration disables refreshing.", + ParameterScope::Replica, ); /// Adds the full set of all storage `Config`s. diff --git a/src/persist-client/src/batch.rs b/src/persist-client/src/batch.rs index 7c15d3df7ca63..3c80d5e5084a9 100644 --- a/src/persist-client/src/batch.rs +++ b/src/persist-client/src/batch.rs @@ -24,7 +24,7 @@ use differential_dataflow::lattice::Lattice; use differential_dataflow::trace::Description; use futures_util::stream::StreamExt; use futures_util::{FutureExt, stream}; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::cast::CastFrom; use mz_ore::instrument; use mz_persist::indexed::encoding::{BatchColumnarFormat, BlobTraceBatchPart, BlobTraceUpdates}; @@ -376,18 +376,21 @@ pub(crate) const BATCH_DELETE_ENABLED: Config = Config::new( "persist_batch_delete_enabled", true, "Whether to actually delete blobs when batch delete is called (Materialize).", + ParameterScope::Environment, ); pub(crate) const ENCODING_ENABLE_DICTIONARY: Config = Config::new( "persist_encoding_enable_dictionary", true, "A feature flag to enable dictionary encoding for Parquet data (Materialize).", + ParameterScope::Environment, ); pub(crate) const ENCODING_COMPRESSION_FORMAT: Config<&'static str> = Config::new( "persist_encoding_compression_format", "none", "A feature flag to enable compression of Parquet data (Materialize).", + ParameterScope::Environment, ); pub(crate) const STRUCTURED_KEY_LOWER_LEN: Config = Config::new( @@ -395,6 +398,7 @@ pub(crate) const STRUCTURED_KEY_LOWER_LEN: Config = Config::new( 256, "The maximum size in proto bytes of any structured key-lower metadata to preserve. \ (If we're unable to fit the lower in budget, or the budget is zero, no metadata is kept.)", + ParameterScope::Environment, ); pub(crate) const MAX_RUN_LEN: Config = Config::new( @@ -402,6 +406,7 @@ pub(crate) const MAX_RUN_LEN: Config = Config::new( usize::MAX, "The maximum length a run can have before it will be spilled as a hollow run \ into the blob store.", + ParameterScope::Environment, ); pub(crate) const MAX_RUNS: Config = Config::new( @@ -410,6 +415,7 @@ pub(crate) const MAX_RUNS: Config = Config::new( "The maximum number of runs a batch builder should generate for user batches. \ (Compaction outputs always generate a single run.) \ The minimum value is 2; below this, compaction is disabled.", + ParameterScope::Environment, ); /// A target maximum size of blob payloads in bytes. If a logical "batch" is @@ -422,12 +428,14 @@ pub(crate) const BLOB_TARGET_SIZE: Config = Config::new( "persist_blob_target_size", 128 * MiB, "A target maximum size of persist blob payloads in bytes (Materialize).", + ParameterScope::Environment, ); pub(crate) const INLINE_WRITES_SINGLE_MAX_BYTES: Config = Config::new( "persist_inline_writes_single_max_bytes", 4096, "The (exclusive) maximum size of a write that persist will inline in metadata.", + ParameterScope::Environment, ); pub(crate) const INLINE_WRITES_TOTAL_MAX_BYTES: Config = Config::new( @@ -436,6 +444,7 @@ pub(crate) const INLINE_WRITES_TOTAL_MAX_BYTES: Config = Config::new( "\ The (exclusive) maximum total size of inline writes in metadata before \ persist will backpressure them by flushing out to s3.", + ParameterScope::Environment, ); impl BatchBuilderConfig { diff --git a/src/persist-client/src/cache.rs b/src/persist-client/src/cache.rs index 2f7707cf59091..3e52d7bdf2048 100644 --- a/src/persist-client/src/cache.rs +++ b/src/persist-client/src/cache.rs @@ -19,7 +19,7 @@ use std::time::{Duration, Instant}; use differential_dataflow::difference::Monoid; use differential_dataflow::lattice::Lattice; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::instrument; use mz_ore::metrics::MetricsRegistry; use mz_ore::task::{AbortOnDropHandle, JoinHandle}; @@ -673,6 +673,7 @@ pub(crate) const STATE_UPDATE_LEASE_TIMEOUT: Config = Config::new( "The amount of time for a command to wait for a previous command to finish before executing. \ (If zero, commands will not wait for others to complete.) Higher values reduce database contention \ at the cost of higher worst-case latencies for individual requests.", + ParameterScope::Environment, ); impl LockingTypedState { diff --git a/src/persist-client/src/cfg.rs b/src/persist-client/src/cfg.rs index 88a8333844b53..a228f7a1aef35 100644 --- a/src/persist-client/src/cfg.rs +++ b/src/persist-client/src/cfg.rs @@ -10,13 +10,20 @@ #![allow(missing_docs)] //! The tunable knobs for persist. +//! +//! Persist configs are `ParameterScope::Environment` as a class, including the +//! ones only ever read on `clusterd`. The same client code runs in +//! `environmentd`, and every copy of it acts on shared durable state, so a +//! replica-scoped persist config could not reach the `environmentd` client and +//! a rollout targeting replicas would leave a shard's other writer on the old +//! value indefinitely. Declare new persist configs `Environment` too. use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use mz_build_info::BuildInfo; -use mz_dyncfg::{Config, ConfigDefault, ConfigSet, ConfigUpdates}; +use mz_dyncfg::{Config, ConfigDefault, ConfigSet, ConfigUpdates, ParameterScope}; use mz_ore::instrument; use mz_ore::now::NowFn; use mz_persist::cfg::BlobKnobs; @@ -281,6 +288,15 @@ pub(crate) const MiB: usize = 1024 * 1024; /// Adds the full set of all persist [Config]s. /// +/// Persist configs are [`ParameterScope::Environment`] by default, because a +/// persist client runs in every `clusterd` process and the per-replica dyncfg +/// push reaches its `ConfigSet` (the compute worker applies the pushed updates +/// to `persist_clients.cfg()`). `environmentd`'s own persist clients read the +/// same configs and see the environment-wide value, which is what "replica- +/// local" means for a config that both processes run. The exceptions are the +/// configs read only from `environmentd`'s catalog and expression cache, which +/// are [`ParameterScope::Environment`]. +/// /// TODO(cfg): Consider replacing this with a static global registry powered by /// something like the `ctor` or `inventory` crate. This would involve managing /// the footgun of a Config being linked into one binary but not the other. @@ -392,6 +408,7 @@ pub const CONSENSUS_CONNECTION_POOL_MAX_SIZE: Config = Config::new( "persist_consensus_connection_pool_max_size", 50, "The maximum size the connection pool to Postgres/CRDB will grow to.", + ParameterScope::Environment, ); /// Sets the maximum amount of time we'll wait to acquire a connection from @@ -402,6 +419,7 @@ const CONSENSUS_CONNECTION_POOL_MAX_WAIT: Config = Config::new( "persist_consensus_connection_pool_max_wait", Duration::from_secs(60), "The amount of time we'll wait for a connection to become available.", + ParameterScope::Environment, ); /// The minimum TTL of a connection to Postgres/CRDB before it is proactively @@ -413,6 +431,7 @@ const CONSENSUS_CONNECTION_POOL_TTL: Config = Config::new( "\ The minimum TTL of a Consensus connection to Postgres/CRDB before it is \ proactively terminated", + ParameterScope::Environment, ); /// The minimum time between TTLing connections to Postgres/CRDB. This delay is @@ -426,6 +445,7 @@ const CONSENSUS_CONNECTION_POOL_TTL_STAGGER: Config = Config::new( "persist_consensus_connection_pool_ttl_stagger", Duration::from_secs(6), "The minimum time between TTLing Consensus connections to Postgres/CRDB.", + ParameterScope::Environment, ); /// The duration to wait for a Consensus Postgres/CRDB connection to be made @@ -434,6 +454,7 @@ pub const CRDB_CONNECT_TIMEOUT: Config = Config::new( "crdb_connect_timeout", Duration::from_secs(5), "The time to connect to CockroachDB before timing out and retrying.", + ParameterScope::Environment, ); /// The TCP user timeout for a Consensus Postgres/CRDB connection. Specifies the @@ -446,6 +467,7 @@ pub const CRDB_TCP_USER_TIMEOUT: Config = Config::new( The TCP timeout for connections to CockroachDB. Specifies the amount of \ time that transmitted data may remain unacknowledged before the TCP \ connection is forcibly closed.", + ParameterScope::Environment, ); pub const CRDB_KEEPALIVES_IDLE: Config = Config::new( @@ -454,12 +476,14 @@ pub const CRDB_KEEPALIVES_IDLE: Config = Config::new( "\ The amount of idle time before a TCP keepalive packet is sent on CRDB \ connections.", + ParameterScope::Environment, ); pub const CRDB_KEEPALIVES_INTERVAL: Config = Config::new( "crdb_keepalives_interval", Duration::from_secs(5), "The time interval between TCP keepalive probes on CRDB connections.", + ParameterScope::Environment, ); pub const CRDB_KEEPALIVES_RETRIES: Config = Config::new( @@ -468,6 +492,7 @@ pub const CRDB_KEEPALIVES_RETRIES: Config = Config::new( "\ The maximum number of TCP keepalive probes that will be sent before \ dropping a CRDB connection.", + ParameterScope::Environment, ); /// Migrate the txns code to use the critical since when opening a new read handle. @@ -475,6 +500,7 @@ pub const USE_CRITICAL_SINCE_TXN: Config = Config::new( "persist_use_critical_since_txn", true, "Use the critical since (instead of the overall since) when initializing a subscribe.", + ParameterScope::Environment, ); /// Migrate the catalog to use the critical since when opening a new read handle. @@ -482,6 +508,7 @@ pub const USE_CRITICAL_SINCE_CATALOG: Config = Config::new( "persist_use_critical_since_catalog", false, "Use the critical since (instead of the overall since) for the Persist-backed catalog.", + ParameterScope::Environment, ); /// Migrate the persist source to use the critical since when opening a new read handle. @@ -489,6 +516,7 @@ pub const USE_CRITICAL_SINCE_SOURCE: Config = Config::new( "persist_use_critical_since_source", false, "Use the critical since (instead of the overall since) in the Persist source.", + ParameterScope::Environment, ); /// While the source is catching up to the shard upper observed at hydration, @@ -500,6 +528,7 @@ pub const SOURCE_HYDRATION_FRONTIER_COALESCE_BYTES: Config = Config::new( 0, "While catching up to the hydration-time upper, the persist source coalesces \ frontier downgrades until this many encoded bytes have been emitted (0 disables).", + ParameterScope::Environment, ); /// Maximum number of part fetches the persist source issues concurrently, per @@ -511,6 +540,7 @@ pub const SOURCE_FETCH_CONCURRENCY: Config = Config::new( 1, "Maximum number of part fetches the persist source issues concurrently per worker \ (1 = serial).", + ParameterScope::Environment, ); /// Migrate snapshots to use the critical since when opening a new read handle. @@ -518,6 +548,7 @@ pub const USE_CRITICAL_SINCE_SNAPSHOT: Config = Config::new( "persist_use_critical_since_snapshot", false, "Use the critical since (instead of the overall since) when taking snapshots in the controller or in fast-path peeks.", + ParameterScope::Environment, ); /// The maximum number of parts (s3 blobs) that [crate::batch::BatchBuilder] @@ -527,6 +558,7 @@ pub const BATCH_BUILDER_MAX_OUTSTANDING_PARTS: Config = Config::new( "persist_batch_builder_max_outstanding_parts", 2, "The number of writes a batch builder can have outstanding before we slow down the writer.", + ParameterScope::Environment, ); /// In Compactor::compact_and_apply, we do the compaction (don't skip it) @@ -536,6 +568,7 @@ pub const COMPACTION_HEURISTIC_MIN_INPUTS: Config = Config::new( "persist_compaction_heuristic_min_inputs", 8, "Don't skip compaction if we have more than this many hollow batches as input.", + ParameterScope::Environment, ); /// In Compactor::compact_and_apply, we do the compaction (don't skip it) @@ -545,6 +578,7 @@ pub const COMPACTION_HEURISTIC_MIN_PARTS: Config = Config::new( "persist_compaction_heuristic_min_parts", 8, "Don't skip compaction if we have more than this many parts as input.", + ParameterScope::Environment, ); /// In Compactor::compact_and_apply, we do the compaction (don't skip it) @@ -554,6 +588,7 @@ pub const COMPACTION_HEURISTIC_MIN_UPDATES: Config = Config::new( "persist_compaction_heuristic_min_updates", 1024, "Don't skip compaction if we have more than this many updates as input.", + ParameterScope::Environment, ); /// The upper bound on compaction's memory consumption. The value must be at @@ -564,6 +599,7 @@ pub const COMPACTION_MEMORY_BOUND_BYTES: Config = Config::new( "persist_compaction_memory_bound_bytes", 1024 * MiB, "Attempt to limit compaction to this amount of memory.", + ParameterScope::Environment, ); /// The maximum number of concurrent blob deletes during garbage collection. @@ -571,6 +607,7 @@ pub const GC_BLOB_DELETE_CONCURRENCY_LIMIT: Config = Config::new( "persist_gc_blob_delete_concurrency_limit", 32, "Limit the number of concurrent deletes GC can perform to this threshold.", + ParameterScope::Environment, ); /// The # of diffs to initially scan when fetching the latest consensus state, to @@ -586,6 +623,7 @@ pub const STATE_VERSIONS_RECENT_LIVE_DIFFS_LIMIT: Config = Config::new( "persist_state_versions_recent_live_diffs_limit", 30 * 128, "Fetch this many diffs when fetching recent diffs.", + ParameterScope::Environment, ); /// The maximum number of concurrent state fetches during usage computation. @@ -593,6 +631,7 @@ pub const USAGE_STATE_FETCH_CONCURRENCY_LIMIT: Config = Config::new( "persist_usage_state_fetch_concurrency_limit", 8, "Limit the concurrency in of fetching in the perioding Persist-storage-usage calculation.", + ParameterScope::Environment, ); impl PostgresClientKnobs for PersistConfig { @@ -678,24 +717,28 @@ pub(crate) const BLOB_OPERATION_TIMEOUT: Config = Config::new( "persist_blob_operation_timeout", Duration::from_secs(180), "Maximum time allowed for a network call, including retry attempts.", + ParameterScope::Environment, ); pub(crate) const BLOB_OPERATION_ATTEMPT_TIMEOUT: Config = Config::new( "persist_blob_operation_attempt_timeout", Duration::from_secs(90), "Maximum time allowed for a single network call.", + ParameterScope::Environment, ); pub(crate) const BLOB_CONNECT_TIMEOUT: Config = Config::new( "persist_blob_connect_timeout", Duration::from_secs(7), "Maximum time to wait for a socket connection to be made.", + ParameterScope::Environment, ); pub(crate) const BLOB_READ_TIMEOUT: Config = Config::new( "persist_blob_read_timeout", Duration::from_secs(10), "Maximum time to wait to read the first byte of a response, including connection time.", + ParameterScope::Environment, ); impl BlobKnobs for PersistConfig { diff --git a/src/persist-client/src/cli/admin.rs b/src/persist-client/src/cli/admin.rs index 5f7b09b4e3eae..5050177eb4936 100644 --- a/src/persist-client/src/cli/admin.rs +++ b/src/persist-client/src/cli/admin.rs @@ -19,7 +19,7 @@ use anyhow::{anyhow, bail}; use differential_dataflow::difference::Monoid; use differential_dataflow::lattice::Lattice; use futures_util::{StreamExt, TryStreamExt, stream}; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; use mz_ore::metrics::MetricsRegistry; use mz_ore::now::SYSTEM_TIME; use mz_ore::url::SensitiveUrl; @@ -666,6 +666,7 @@ pub const CATALOG_FORCE_COMPACTION_FUEL: Config = Config::new( "persist_catalog_force_compaction_fuel", 1024, "fuel to use in catalog dangerous_force_compaction task", + ParameterScope::Environment, ); /// Exposed for `mz-catalog`. @@ -673,6 +674,7 @@ pub const CATALOG_FORCE_COMPACTION_WAIT: Config = Config::new( "persist_catalog_force_compaction_wait", Duration::from_secs(60), "wait to use in catalog dangerous_force_compaction task", + ParameterScope::Environment, ); /// Exposed for `mz-catalog`. @@ -680,6 +682,7 @@ pub const EXPRESSION_CACHE_FORCE_COMPACTION_FUEL: Config = Config::new( "persist_expression_cache_force_compaction_fuel", 131_072, "fuel to use in expression cache dangerous_force_compaction", + ParameterScope::Environment, ); /// Exposed for `mz-catalog`. @@ -687,6 +690,7 @@ pub const EXPRESSION_CACHE_FORCE_COMPACTION_WAIT: Config = Config::new "persist_expression_cache_force_compaction_wait", Duration::from_secs(0), "wait to use in expression cache dangerous_force_compaction", + ParameterScope::Environment, ); /// Attempts to compact all batches in a shard into a minimal number. diff --git a/src/persist-client/src/fetch.rs b/src/persist-client/src/fetch.rs index 1d5c956162654..e84323ef79b3c 100644 --- a/src/persist-client/src/fetch.rs +++ b/src/persist-client/src/fetch.rs @@ -21,7 +21,7 @@ use differential_dataflow::difference::Monoid; use differential_dataflow::lattice::Lattice; use differential_dataflow::trace::Description; use itertools::EitherOrBoth; -use mz_dyncfg::{Config, ConfigSet, ConfigValHandle}; +use mz_dyncfg::{Config, ConfigSet, ConfigValHandle, ParameterScope}; use mz_ore::bytes::SegmentedBytes; use mz_ore::cast::CastFrom; use mz_ore::{soft_assert_or_log, soft_panic_no_log, soft_panic_or_log}; @@ -66,6 +66,7 @@ pub(crate) const FETCH_SEMAPHORE_COST_ADJUSTMENT: Config = Config::new( "\ An adjustment multiplied by encoded_size_bytes to approximate an upper \ bound on the size in lgalloc, which includes the decoded version.", + ParameterScope::Environment, ); pub(crate) const FETCH_SEMAPHORE_PERMIT_ADJUSTMENT: Config = Config::new( @@ -76,6 +77,7 @@ pub(crate) const FETCH_SEMAPHORE_PERMIT_ADJUSTMENT: Config = Config::new( parsed, expressed as a multiplier of the process's memory limit. This data \ all spills to lgalloc, so values > 1.0 are safe. Only applied to cc \ replicas.", + ParameterScope::Environment, ); pub(crate) const PART_DECODE_FORMAT: Config<&'static str> = Config::new( @@ -84,12 +86,14 @@ pub(crate) const PART_DECODE_FORMAT: Config<&'static str> = Config::new( "\ Format we'll use to decode a Persist Part, either 'row', \ 'row_with_validate', or 'arrow' (Materialize).", + ParameterScope::Environment, ); pub(crate) const OPTIMIZE_IGNORED_DATA_FETCH: Config = Config::new( "persist_optimize_ignored_data_fetch", true, "CYA to allow opt-out of a performance optimization to skip fetching ignored data", + ParameterScope::Environment, ); pub(crate) const VALIDATE_PART_BOUNDS_ON_READ: Config = Config::new( @@ -97,6 +101,7 @@ pub(crate) const VALIDATE_PART_BOUNDS_ON_READ: Config = Config::new( false, "Validate the part lower <= the batch lower and the part upper <= batch upper,\ for the batch containing that part", + ParameterScope::Environment, ); #[derive(Debug, Clone)] diff --git a/src/persist-client/src/internal/cache.rs b/src/persist-client/src/internal/cache.rs index 1af8ea077bd5b..82d1e0e5e255d 100644 --- a/src/persist-client/src/internal/cache.rs +++ b/src/persist-client/src/internal/cache.rs @@ -13,7 +13,7 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use bytes::Bytes; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; use mz_ore::bytes::SegmentedBytes; use mz_ore::cast::CastFrom; use mz_persist::location::{Blob, BlobMetadata, ExternalError}; @@ -38,12 +38,14 @@ pub(crate) const BLOB_CACHE_MEM_LIMIT_BYTES: Config = Config::new( // 128MiB 128 * 1024 * 1024, "Capacity of in-mem blob cache in bytes (Materialize).", + ParameterScope::Environment, ); pub(crate) const BLOB_CACHE_SCALE_WITH_THREADS: Config = Config::new( "persist_blob_cache_scale_with_threads", false, "Whether or not the size of the in-mem blob cache scales with the number of threads in the current process (Materialize).", + ParameterScope::Environment, ); pub(crate) const BLOB_CACHE_SCALE_FACTOR_BYTES: Config = Config::new( @@ -51,6 +53,7 @@ pub(crate) const BLOB_CACHE_SCALE_FACTOR_BYTES: Config = Config::new( // 32MiB 32 * 1024 * 1024, "Scale factor for the in-mem blob cache, in bytes, if scaling with threads (Materialize).", + ParameterScope::Environment, ); impl BlobMemCache { diff --git a/src/persist-client/src/internal/compact.rs b/src/persist-client/src/internal/compact.rs index 28d141b6ff178..68a97d0bd121b 100644 --- a/src/persist-client/src/internal/compact.rs +++ b/src/persist-client/src/internal/compact.rs @@ -21,7 +21,7 @@ use differential_dataflow::trace::Description; use futures::{Stream, pin_mut}; use futures_util::StreamExt; use itertools::Either; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::cast::CastFrom; use mz_ore::error::ErrorExt; use mz_ore::now::NowFn; @@ -145,6 +145,7 @@ pub(crate) const COMPACTION_MINIMUM_TIMEOUT: Config = Config::new( "\ The minimum amount of time to allow a persist compaction request to run \ before timing it out (Materialize).", + ParameterScope::Environment, ); pub(crate) const COMPACTION_CHECK_PROCESS_FLAG: Config = Config::new( @@ -152,6 +153,7 @@ pub(crate) const COMPACTION_CHECK_PROCESS_FLAG: Config = Config::new( true, "Whether Compactor will obey the process_requests flag in PersistConfig, \ which allows dynamically disabling compaction. If false, all compaction requests will be processed.", + ParameterScope::Environment, ); /// Create a `[CompactionInput::IdRange]` from a set of `SpineId`s. diff --git a/src/persist-client/src/internal/machine.rs b/src/persist-client/src/internal/machine.rs index afde9b12e2812..939c95b49496e 100644 --- a/src/persist-client/src/internal/machine.rs +++ b/src/persist-client/src/internal/machine.rs @@ -18,7 +18,7 @@ use differential_dataflow::difference::Monoid; use differential_dataflow::lattice::Lattice; use futures::FutureExt; use futures::future::{self, BoxFuture}; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; use mz_ore::cast::CastFrom; use mz_ore::error::ErrorExt; #[allow(unused_imports)] // False positive. @@ -78,6 +78,7 @@ pub(crate) const CLAIM_UNCLAIMED_COMPACTIONS: Config = Config::new( false, "If an append doesn't result in a compaction request, but there is some uncompacted batch \ in state, compact that instead.", + ParameterScope::Environment, ); pub(crate) const CLAIM_COMPACTION_PERCENT: Config = Config::new( @@ -86,12 +87,14 @@ pub(crate) const CLAIM_COMPACTION_PERCENT: Config = Config::new( "Claim a compaction with the given percent chance, if claiming compactions is enabled. \ (If over 100, we'll always claim at least one; for example, if set to 365, we'll claim at least \ three and have a 65% chance of claiming a fourth.)", + ParameterScope::Environment, ); pub(crate) const CLAIM_COMPACTION_MIN_VERSION: Config = Config::new( "persist_claim_compaction_min_version", String::new(), "If set to a valid version string, compact away any earlier versions if possible.", + ParameterScope::Environment, ); impl Machine @@ -1144,24 +1147,28 @@ pub(crate) const NEXT_LISTEN_BATCH_RETRYER_FIXED_SLEEP: Config = Confi Duration::from_millis(1200), // pubsub is on by default! "\ The fixed sleep when polling for new batches from a Listen or Subscribe. Skipped if zero.", + ParameterScope::Environment, ); pub(crate) const NEXT_LISTEN_BATCH_RETRYER_INITIAL_BACKOFF: Config = Config::new( "persist_next_listen_batch_retryer_initial_backoff", Duration::from_millis(100), // pubsub is on by default! "The initial backoff when polling for new batches from a Listen or Subscribe.", + ParameterScope::Environment, ); pub(crate) const NEXT_LISTEN_BATCH_RETRYER_MULTIPLIER: Config = Config::new( "persist_next_listen_batch_retryer_multiplier", 2, "The backoff multiplier when polling for new batches from a Listen or Subscribe.", + ParameterScope::Environment, ); pub(crate) const NEXT_LISTEN_BATCH_RETRYER_CLAMP: Config = Config::new( "persist_next_listen_batch_retryer_clamp", Duration::from_secs(16), // pubsub is on by default! "The backoff clamp duration when polling for new batches from a Listen or Subscribe.", + ParameterScope::Environment, ); pub(crate) fn next_listen_batch_retry_params(cfg: &ConfigSet) -> RetryParameters { diff --git a/src/persist-client/src/internal/state.rs b/src/persist-client/src/internal/state.rs index b63012c688e34..980e8eed55717 100644 --- a/src/persist-client/src/internal/state.rs +++ b/src/persist-client/src/internal/state.rs @@ -31,7 +31,7 @@ use differential_dataflow::trace::implementations::BatchContainer; use futures::Stream; use futures_util::StreamExt; use itertools::Itertools; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::cast::CastFrom; use mz_ore::now::EpochMillis; use mz_ore::soft_panic_or_log; @@ -92,6 +92,7 @@ pub(crate) const ROLLUP_THRESHOLD: Config = Config::new( "persist_rollup_threshold", 128, "The number of seqnos between rollups.", + ParameterScope::Environment, ); /// Determines how long to wait before an active rollup is considered @@ -100,6 +101,7 @@ pub(crate) const ROLLUP_FALLBACK_THRESHOLD_MS: Config = Config::new( "persist_rollup_fallback_threshold_ms", 5000, "The number of milliseconds before a worker claims an already claimed rollup.", + ParameterScope::Environment, ); /// Feature flag the new active rollup tracking mechanism. @@ -108,6 +110,7 @@ pub(crate) const ROLLUP_USE_ACTIVE_ROLLUP: Config = Config::new( "persist_rollup_use_active_rollup", true, "Whether to use the new active rollup tracking mechanism.", + ParameterScope::Environment, ); /// Determines how long to wait before an active GC is considered @@ -116,6 +119,7 @@ pub(crate) const GC_FALLBACK_THRESHOLD_MS: Config = Config::new( "persist_gc_fallback_threshold_ms", 900000, "The number of milliseconds before a worker claims an already claimed GC.", + ParameterScope::Environment, ); /// See the config description string. @@ -123,6 +127,7 @@ pub(crate) const GC_MIN_VERSIONS: Config = Config::new( "persist_gc_min_versions", 32, "The number of un-GCd versions that may exist in state before we'll trigger a GC.", + ParameterScope::Environment, ); /// See the config description string. @@ -130,6 +135,7 @@ pub(crate) const GC_MAX_VERSIONS: Config = Config::new( "persist_gc_max_versions", 128_000, "The maximum number of versions to GC in a single GC run.", + ParameterScope::Environment, ); /// Feature flag the new active GC tracking mechanism. @@ -138,12 +144,14 @@ pub(crate) const GC_USE_ACTIVE_GC: Config = Config::new( "persist_gc_use_active_gc", false, "Whether to use the new active GC tracking mechanism.", + ParameterScope::Environment, ); pub(crate) const ENABLE_INCREMENTAL_COMPACTION: Config = Config::new( "persist_enable_incremental_compaction", false, "Whether to enable incremental compaction.", + ParameterScope::Environment, ); /// A token to disambiguate state commands that could not otherwise be diff --git a/src/persist-client/src/lib.rs b/src/persist-client/src/lib.rs index e381230376c6a..1c22327b27d2e 100644 --- a/src/persist-client/src/lib.rs +++ b/src/persist-client/src/lib.rs @@ -84,7 +84,7 @@ pub mod metrics { pub mod operators { //! [timely] operators for reading and writing persist Shards. - use mz_dyncfg::Config; + use mz_dyncfg::{Config, ParameterScope}; pub mod shard_source; @@ -95,6 +95,7 @@ pub mod operators { "\ The maximum amount of work to do in the persist_source mfp_and_decode \ operator before yielding.", + ParameterScope::Environment, ); } pub mod read; diff --git a/src/persist-client/src/read.rs b/src/persist-client/src/read.rs index 428e5cce074d3..058bce194e7e4 100644 --- a/src/persist-client/src/read.rs +++ b/src/persist-client/src/read.rs @@ -22,7 +22,7 @@ use differential_dataflow::difference::Monoid; use differential_dataflow::lattice::Lattice; use futures::Stream; use futures_util::{StreamExt, stream}; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::cast::CastLossy; use mz_ore::halt; use mz_ore::instrument; @@ -617,6 +617,7 @@ pub(crate) const READER_LEASE_DURATION: Config = Config::new( "persist_reader_lease_duration", Duration::from_secs(60 * 15), "The time after which we'll clean up stale read leases", + ParameterScope::Environment, ); impl ReadHandle diff --git a/src/persist-client/src/rpc.rs b/src/persist-client/src/rpc.rs index 799597350fe75..87bd55aff68f5 100644 --- a/src/persist-client/src/rpc.rs +++ b/src/persist-client/src/rpc.rs @@ -24,7 +24,7 @@ use async_trait::async_trait; use bytes::Bytes; use futures::Stream; use futures_util::StreamExt; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::cast::CastFrom; use mz_ore::collections::{HashMap, HashSet}; use mz_ore::metrics::MetricsRegistry; @@ -59,6 +59,7 @@ pub(crate) const PUBSUB_CLIENT_ENABLED: Config = Config::new( "persist_pubsub_client_enabled", true, "Whether to connect to the Persist PubSub service.", + ParameterScope::Environment, ); /// For connected clients, determines whether to push state diffs to the PubSub @@ -68,6 +69,7 @@ pub(crate) const PUBSUB_PUSH_DIFF_ENABLED: Config = Config::new( "persist_pubsub_push_diff_enabled", true, "Whether to push state diffs to Persist PubSub.", + ParameterScope::Environment, ); /// For connected clients, determines whether to push state diffs to the PubSub @@ -77,6 +79,7 @@ pub(crate) const PUBSUB_SAME_PROCESS_DELEGATE_ENABLED: Config = Config::ne "persist_pubsub_same_process_delegate_enabled", true, "Whether to push state diffs to Persist PubSub on the same process.", + ParameterScope::Environment, ); /// Timeout per connection attempt to Persist PubSub service. @@ -84,6 +87,7 @@ pub(crate) const PUBSUB_CONNECT_ATTEMPT_TIMEOUT: Config = Config::new( "persist_pubsub_connect_attempt_timeout", Duration::from_secs(5), "Timeout per connection attempt to Persist PubSub service.", + ParameterScope::Environment, ); /// Timeout per request attempt to Persist PubSub service. @@ -91,6 +95,7 @@ pub(crate) const PUBSUB_REQUEST_TIMEOUT: Config = Config::new( "persist_pubsub_request_timeout", Duration::from_secs(5), "Timeout per request attempt to Persist PubSub service.", + ParameterScope::Environment, ); /// Maximum backoff when retrying connection establishment to Persist PubSub service. @@ -98,6 +103,7 @@ pub(crate) const PUBSUB_CONNECT_MAX_BACKOFF: Config = Config::new( "persist_pubsub_connect_max_backoff", Duration::from_secs(60), "Maximum backoff when retrying connection establishment to Persist PubSub service.", + ParameterScope::Environment, ); /// Size of channel used to buffer send messages to PubSub service. @@ -105,6 +111,7 @@ pub(crate) const PUBSUB_CLIENT_SENDER_CHANNEL_SIZE: Config = Config::new( "persist_pubsub_client_sender_channel_size", 25, "Size of channel used to buffer send messages to PubSub service.", + ParameterScope::Environment, ); /// Size of channel used to buffer received messages from PubSub service. @@ -112,6 +119,7 @@ pub(crate) const PUBSUB_CLIENT_RECEIVER_CHANNEL_SIZE: Config = Config::ne "persist_pubsub_client_receiver_channel_size", 25, "Size of channel used to buffer received messages from PubSub service.", + ParameterScope::Environment, ); /// Size of channel used per connection to buffer broadcasted messages from PubSub server. @@ -119,6 +127,7 @@ pub(crate) const PUBSUB_SERVER_CONNECTION_CHANNEL_SIZE: Config = Config:: "persist_pubsub_server_connection_channel_size", 25, "Size of channel used per connection to buffer broadcasted messages from PubSub server.", + ParameterScope::Environment, ); /// Size of channel used by the state cache to broadcast shard state references. @@ -126,6 +135,7 @@ pub(crate) const PUBSUB_STATE_CACHE_SHARD_REF_CHANNEL_SIZE: Config = Conf "persist_pubsub_state_cache_shard_ref_channel_size", 25, "Size of channel used by the state cache to broadcast shard state references.", + ParameterScope::Environment, ); /// Backoff after an established connection to Persist PubSub service fails. @@ -133,6 +143,7 @@ pub(crate) const PUBSUB_RECONNECT_BACKOFF: Config = Config::new( "persist_pubsub_reconnect_backoff", Duration::from_secs(5), "Backoff after an established connection to Persist PubSub service fails.", + ParameterScope::Environment, ); /// Max message size, used to configure gRPC servers and clients. diff --git a/src/persist-client/src/stats.rs b/src/persist-client/src/stats.rs index b509c425a7bce..25a8ab57bd6d2 100644 --- a/src/persist-client/src/stats.rs +++ b/src/persist-client/src/stats.rs @@ -12,7 +12,7 @@ use std::borrow::Cow; use std::sync::Arc; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; use crate::batch::UntrimmableColumns; use crate::metrics::Metrics; @@ -25,6 +25,7 @@ pub(crate) const STATS_AUDIT_PERCENT: Config = Config::new( "persist_stats_audit_percent", 1, "Percent of filtered data to opt in to correctness auditing (Materialize).", + ParameterScope::Environment, ); /// See description for usage. @@ -34,6 +35,7 @@ pub const STATS_AUDIT_PANIC: Config = Config::new( "If set (as it is by default), panic on any auditing failure. If not, report an error but \ pass along the data as normal. This should almost certainly be paired with an audit rate of 100%, \ so all parts are audited, for consistency.", + ParameterScope::Environment, ); /// Computes and stores statistics about each batch part. @@ -47,6 +49,7 @@ pub(crate) const STATS_COLLECTION_ENABLED: Config = Config::new( Whether to calculate and record statistics about the data stored in \ persist to be used at read time, see persist_stats_filter_enabled \ (Materialize).", + ParameterScope::Environment, ); /// Uses previously computed statistics about batch parts to entirely skip @@ -59,6 +62,7 @@ pub const STATS_FILTER_ENABLED: Config = Config::new( "\ Whether to use recorded statistics about the data stored in persist to \ filter at read time, see persist_stats_collection_enabled (Materialize).", + ParameterScope::Environment, ); /// The budget (in bytes) of how many stats to write down per batch part. When @@ -68,6 +72,7 @@ pub(crate) const STATS_BUDGET_BYTES: Config = Config::new( "persist_stats_budget_bytes", 1024, "The budget (in bytes) of how many stats to maintain per batch part.", + ParameterScope::Environment, ); pub(crate) const STATS_UNTRIMMABLE_COLUMNS_EQUALS: Config String> = Config::new( @@ -91,6 +96,7 @@ pub(crate) const STATS_UNTRIMMABLE_COLUMNS_EQUALS: Config String> = Conf Which columns to always retain during persist stats trimming. Any column \ with a name exactly equal (case-insensitive) to one of these will be kept. \ Comma separated list.", + ParameterScope::Environment, ); pub(crate) const STATS_UNTRIMMABLE_COLUMNS_PREFIX: Config String> = Config::new( @@ -100,6 +106,7 @@ pub(crate) const STATS_UNTRIMMABLE_COLUMNS_PREFIX: Config String> = Conf Which columns to always retain during persist stats trimming. Any column \ with a name starting with (case-insensitive) one of these will be kept. \ Comma separated list.", + ParameterScope::Environment, ); pub(crate) const STATS_UNTRIMMABLE_COLUMNS_SUFFIX: Config String> = Config::new( @@ -109,6 +116,7 @@ pub(crate) const STATS_UNTRIMMABLE_COLUMNS_SUFFIX: Config String> = Conf Which columns to always retain during persist stats trimming. Any column \ with a name ending with (case-insensitive) one of these will be kept. \ Comma separated list.", + ParameterScope::Environment, ); pub(crate) fn untrimmable_columns(cfg: &ConfigSet) -> UntrimmableColumns { diff --git a/src/persist-client/src/write.rs b/src/persist-client/src/write.rs index 66023a50185bc..287595aa9769a 100644 --- a/src/persist-client/src/write.rs +++ b/src/persist-client/src/write.rs @@ -18,7 +18,7 @@ use differential_dataflow::lattice::Lattice; use differential_dataflow::trace::Description; use futures::StreamExt; use futures::stream::FuturesUnordered; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::task::RuntimeExt; use mz_ore::{instrument, soft_panic_or_log}; use mz_persist::location::Blob; @@ -58,6 +58,7 @@ pub(crate) const COMBINE_INLINE_WRITES: Config = Config::new( "persist_write_combine_inline_writes", true, "If set, re-encode inline writes if they don't fit into the batch metadata limits.", + ParameterScope::Environment, ); pub(crate) const VALIDATE_PART_BOUNDS_ON_WRITE: Config = Config::new( @@ -65,6 +66,7 @@ pub(crate) const VALIDATE_PART_BOUNDS_ON_WRITE: Config = Config::new( false, "Validate the part lower <= the batch lower and the part upper <= batch upper,\ for the batch being appended.", + ParameterScope::Environment, ); /// An opaque identifier for a writer of a persist durable TVC (aka shard). diff --git a/src/persist/src/postgres.rs b/src/persist/src/postgres.rs index 66801f5d9e3bb..7771ace833b02 100644 --- a/src/persist/src/postgres.rs +++ b/src/persist/src/postgres.rs @@ -58,6 +58,7 @@ pub const PG_CONSENSUS_READ_COMMITTED: mz_dyncfg::Config = mz_dyncfg::Conf false, "Run consensus connections under READ COMMITTED isolation instead of SERIALIZABLE when targetting PostgreSQL backends. This flag must be off when targetting CockroachDB.", + mz_dyncfg::ParameterScope::Environment, ); const SCHEMA: &str = " diff --git a/src/storage-controller/src/instance.rs b/src/storage-controller/src/instance.rs index 57b506f362943..aba62e0bd8cc7 100644 --- a/src/storage-controller/src/instance.rs +++ b/src/storage-controller/src/instance.rs @@ -222,6 +222,11 @@ impl Instance { pub fn drop_replica(&mut self, id: ReplicaId) { let replica = self.replicas.remove(&id); + // The coordinator only re-pushes the override map when the scoped configuration itself + // changes, so a dropped replica's entry would otherwise be retained until the next such + // change. + self.replica_dyncfg_overrides.remove(&id); + let mut needs_rescheduling = false; for (ingestion_id, ingestion) in self.active_ingestions.iter_mut() { let was_running = ingestion.active_replicas.remove(&id); diff --git a/src/storage-types/src/dyncfgs.rs b/src/storage-types/src/dyncfgs.rs index 30002f480f94f..d7bbdc26c77f4 100644 --- a/src/storage-types/src/dyncfgs.rs +++ b/src/storage-types/src/dyncfgs.rs @@ -10,7 +10,7 @@ //! Dyncfgs used by the storage layer. Despite their name, these can be used //! "statically" during rendering, or dynamically within timely operators. -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; use std::time::Duration; /// When dataflows observe an invariant violation it is either due to a bug or due to the cluster @@ -22,6 +22,7 @@ pub const CLUSTER_SHUTDOWN_GRACE_PERIOD: Config = Config::new( "When dataflows observe an invariant violation it is either due to a bug or due to \ the cluster being shut down. This configuration defines the amount of time to \ wait before panicking the process, which will register the invariant violation.", + ParameterScope::Replica, ); // Flow control @@ -34,6 +35,7 @@ pub const DELAY_SOURCES_PAST_REHYDRATION: Config = Config::new( true, "Whether or not to delay sources producing values in some scenarios \ (namely, upsert) till after rehydration is finished", + ParameterScope::Environment, ); /// Whether storage dataflows should suspend execution while downstream operators are still @@ -43,6 +45,7 @@ pub const SUSPENDABLE_SOURCES: Config = Config::new( true, "Whether storage dataflows should suspend execution while downstream operators are still \ processing data.", + ParameterScope::Environment, ); // Controller @@ -55,6 +58,7 @@ pub const STORAGE_DOWNGRADE_SINCE_DURING_FINALIZATION: Config = Config::ne true, "When enabled, force-downgrade the controller's since handle on the shard\ during shard finalization", + ParameterScope::Environment, ); /// The interval of time to keep when truncating the replica metrics history. @@ -62,6 +66,7 @@ pub const REPLICA_METRICS_HISTORY_RETENTION_INTERVAL: Config = Config: "replica_metrics_history_retention_interval", Duration::from_secs(60 * 60 * 24 * 30), // 30 days "The interval of time to keep when truncating the replica metrics history.", + ParameterScope::Environment, ); /// The interval of time to keep when truncating the wallclock lag history. @@ -69,6 +74,7 @@ pub const WALLCLOCK_LAG_HISTORY_RETENTION_INTERVAL: Config = Config::n "wallclock_lag_history_retention_interval", Duration::from_secs(60 * 60 * 24 * 30), // 30 days "The interval of time to keep when truncating the wallclock lag history.", + ParameterScope::Environment, ); /// The interval of time to keep when truncating the wallclock lag histogram. @@ -76,6 +82,7 @@ pub const WALLCLOCK_GLOBAL_LAG_HISTOGRAM_RETENTION_INTERVAL: Config = "wallclock_global_lag_histogram_retention_interval", Duration::from_secs(60 * 60 * 24 * 30), // 30 days "The interval of time to keep when truncating the wallclock lag histogram.", + ParameterScope::Environment, ); // Kafka @@ -94,6 +101,7 @@ pub const KAFKA_CLIENT_ID_ENRICHMENT_RULES: Config serde_json::Value> = "kafka_client_id_enrichment_rules", || serde_json::json!([]), "Rules for enriching the `client.id` property of Kafka clients with additional data.", + ParameterScope::Environment, ); /// The maximum time we will wait before re-polling rdkafka to see if new partitions/data are @@ -103,15 +111,20 @@ pub const KAFKA_POLL_MAX_WAIT: Config = Config::new( Duration::from_secs(1), "The maximum time we will wait before re-polling rdkafka to see if new partitions/data are \ available.", + ParameterScope::Replica, ); /// Whether to check the low watermark for Kafka sources and error if the start offset/resume /// upper has been compacted away. +/// Environment-scoped because it decides whether a definite error is emitted. +/// Replicas of one cluster disagreeing would write different collection +/// contents, so the value has to be coherent across them. pub const KAFKA_LOW_WATERMARK_CHECK: Config = Config::new( "kafka_low_watermark_check", true, "Whether to check the low watermark for Kafka sources and error if the start \ offset/resume upper has been compacted away.", + ParameterScope::Environment, ); pub const KAFKA_DEFAULT_AWS_PRIVATELINK_ENDPOINT_IDENTIFICATION_ALGORITHM: Config<&'static str> = @@ -121,6 +134,7 @@ pub const KAFKA_DEFAULT_AWS_PRIVATELINK_ENDPOINT_IDENTIFICATION_ALGORITHM: Confi "none", "The value we set for the 'ssl.endpoint.identification.algorithm' option in the Kafka \ Connection config. default: 'none'", + ParameterScope::Environment, ); pub const KAFKA_BUFFERED_EVENT_RESIZE_THRESHOLD_ELEMENTS: Config = Config::new( @@ -129,6 +143,7 @@ pub const KAFKA_BUFFERED_EVENT_RESIZE_THRESHOLD_ELEMENTS: Config = Config "In the Kafka sink operator we might need to buffer messages before emitting them. As a \ performance optimization we reuse the buffer allocations, but shrink it to retain at \ most this number of elements.", + ParameterScope::Replica, ); /// Sets retry.backoff.ms in librdkafka for sources and sinks. @@ -137,6 +152,7 @@ pub const KAFKA_RETRY_BACKOFF: Config = Config::new( "kafka_retry_backoff", Duration::from_millis(100), "Sets retry.backoff.ms in librdkafka for sources and sinks.", + ParameterScope::Replica, ); /// Sets retry.backoff.max.ms in librdkafka for sources and sinks. @@ -145,6 +161,7 @@ pub const KAFKA_RETRY_BACKOFF_MAX: Config = Config::new( "kafka_retry_backoff_max", Duration::from_secs(1), "Sets retry.backoff.max.ms in librdkafka for sources and sinks.", + ParameterScope::Replica, ); /// Sets reconnect.backoff.ms in librdkafka for sources and sinks. @@ -153,6 +170,7 @@ pub const KAFKA_RECONNECT_BACKOFF: Config = Config::new( "kafka_reconnect_backoff", Duration::from_millis(100), "Sets reconnect.backoff.ms in librdkafka for sources and sinks.", + ParameterScope::Replica, ); /// Sets reconnect.backoff.max.ms in librdkafka for sources and sinks. @@ -163,6 +181,7 @@ pub const KAFKA_RECONNECT_BACKOFF_MAX: Config = Config::new( "kafka_reconnect_backoff_max", Duration::from_secs(30), "Sets reconnect.backoff.max.ms in librdkafka for sources and sinks.", + ParameterScope::Replica, ); /// Sets message.max.bytes in librdkafka for Kafka sink producers. @@ -174,6 +193,7 @@ pub const KAFKA_SINK_MESSAGE_MAX_BYTES: Config = Config::new( "kafka_sink_message_max_bytes", 1_000_000, "Sets message.max.bytes in librdkafka for Kafka sink producers.", + ParameterScope::Environment, ); /// Sets batch.size in librdkafka for Kafka sink producers. @@ -185,6 +205,7 @@ pub const KAFKA_SINK_BATCH_SIZE: Config = Config::new( "kafka_sink_batch_size", 1_000_000, "Sets batch.size in librdkafka for Kafka sink producers.", + ParameterScope::Environment, ); /// Sets batch.num.messages in librdkafka for Kafka sink producers. @@ -195,6 +216,7 @@ pub const KAFKA_SINK_BATCH_NUM_MESSAGES: Config = Config::new( "kafka_sink_batch_num_messages", 10_000, "Sets batch.num.messages in librdkafka for Kafka sink producers.", + ParameterScope::Environment, ); // MySQL @@ -204,6 +226,7 @@ pub const MYSQL_REPLICATION_HEARTBEAT_INTERVAL: Config = Config::new( "mysql_replication_heartbeat_interval", Duration::from_secs(30), "Replication heartbeat interval requested from the MySQL server.", + ParameterScope::Replica, ); /// Whether to split snapshot reads of tables with a supported single-column @@ -213,6 +236,7 @@ pub static MYSQL_SOURCE_SNAPSHOT_PARALLELISM: Config = Config::new( "mysql_source_snapshot_parallelism", false, "Whether to split MySQL snapshot reads across workers by primary-key ranges.", + ParameterScope::Replica, ); /// Smallest estimated row count the MySQL snapshot partitioner attempts to subdivide. @@ -220,6 +244,7 @@ pub static MYSQL_SOURCE_SNAPSHOT_PARTITION_MIN_ROWS: Config = Config::new "mysql_source_snapshot_partition_min_rows", 50_000, "Minimum estimated rows the MySQL snapshot partitioner attempts to split.", + ParameterScope::Replica, ); /// Cap on string primary key prefixes visited when attempting to partition a table for @@ -231,6 +256,7 @@ pub static MYSQL_SOURCE_SNAPSHOT_PARTITION_PROBED_PREFIXES_PER_BILLION_ROWS: Con 1_000, "Cap on MySQL snapshot PK-prefix partitioning probed prefixes per table, per billion \ estimated rows; when exhausted, splitting stops early with coarser partition boundaries.", + ParameterScope::Replica, ); /// If the optimizer estimates the table has fewer rows than this, compute the exact row count @@ -240,6 +266,7 @@ pub static MYSQL_SOURCE_SNAPSHOT_EXACT_COUNT_MAX_ROWS: Config = Config::n 1_000_000, "Maximum estimated table size for which MySQL snapshots compute an exact COUNT(*) \ for the size gauge; larger tables report the information_schema estimate.", + ParameterScope::Replica, ); // Postgres @@ -249,6 +276,7 @@ pub const PG_FETCH_SLOT_RESUME_LSN_INTERVAL: Config = Config::new( "postgres_fetch_slot_resume_lsn_interval", Duration::from_millis(500), "Interval to poll `confirmed_flush_lsn` to get a resumption lsn.", + ParameterScope::Replica, ); /// Interval to re-validate the schemas of ingested tables. @@ -256,6 +284,7 @@ pub const PG_SCHEMA_VALIDATION_INTERVAL: Config = Config::new( "pg_schema_validation_interval", Duration::from_secs(15), "Interval to re-validate the schemas of ingested tables.", + ParameterScope::Environment, ); /// Controls behavior of PG Source when the upstream DB timeline changes. The default behavior @@ -263,10 +292,14 @@ pub const PG_SCHEMA_VALIDATION_INTERVAL: Config = Config::new( /// provide guarantees of failover without loss of data (e.g. CloudSQL maintenance). Changing this /// flag puts the onus on the customer to recreate the source if the upstream DB changes timeline /// in a way that introduces data loss (e.g. manual failover, restore, etc.). +/// Environment-scoped because it decides whether a definite error is emitted. +/// Replicas of one cluster disagreeing would write different collection +/// contents, so the value has to be coherent across them. pub static PG_SOURCE_VALIDATE_TIMELINE: Config = Config::new( "pg_source_validate_timeline", true, "Whether to treat a timeline switch as a definite error", + ParameterScope::Environment, ); /// Controls behavior of the SQL Server source when the upstream DB restore history changes. The @@ -274,10 +307,14 @@ pub static PG_SOURCE_VALIDATE_TIMELINE: Config = Config::new( /// On Availability Group (AOAG), the upstream DB may guarantee continuity without loss of data. /// Changing this flag puts the onus on the customer to recreate the source if the upstream DB /// changes in a way that introduces data loss. +/// Environment-scoped because it decides whether a definite error is emitted. +/// Replicas of one cluster disagreeing would write different collection +/// contents, so the value has to be coherent across them. pub static SQL_SERVER_SOURCE_VALIDATE_RESTORE_HISTORY: Config = Config::new( "sql_server_source_validate_restore_history", true, "Whether to treat a restore history change as a definite error", + ParameterScope::Environment, ); // AWS @@ -294,17 +331,23 @@ pub const AWS_PREFETCH_STS_CONNECT_TIMEOUT: Config = Config::new( "aws_prefetch_sts_connect_timeout", Duration::from_millis(3100), "Connect timeout for the AWS AssumeRole credentials prefetcher's STS calls.", + ParameterScope::Replica, ); // Networking /// Whether or not to enforce that external connection addresses are global /// (not private or local) when resolving them. +/// +/// Read on both `environmentd` (purification, `COPY` planning) and the replica. +/// Deliberately environment-scoped even so: this is a security control, and a +/// per-replica override would weaken it for part of the environment only. pub const ENFORCE_EXTERNAL_ADDRESSES: Config = Config::new( "storage_enforce_external_addresses", false, "Whether or not to enforce that external connection addresses are global \ (not private or local) when resolving them", + ParameterScope::Environment, ); // Upsert @@ -327,6 +370,7 @@ pub const STORAGE_UPSERT_PREVENT_SNAPSHOT_BUFFERING: Config = Config::new( "storage_upsert_prevent_snapshot_buffering", true, "Prevent snapshot buffering in upsert.", + ParameterScope::Replica, ); /// Whether to enable the merge operator in upsert for the RocksDB backend. @@ -334,6 +378,7 @@ pub const STORAGE_ROCKSDB_USE_MERGE_OPERATOR: Config = Config::new( "storage_rocksdb_use_merge_operator", true, "Use the native rocksdb merge operator where possible.", + ParameterScope::Environment, ); /// If `storage_upsert_prevent_snapshot_buffering` is true, this prevents the upsert @@ -344,6 +389,7 @@ pub const STORAGE_UPSERT_MAX_SNAPSHOT_BATCH_BUFFERING: Config> = C "storage_upsert_max_snapshot_batch_buffering", None, "Limit snapshot buffering in upsert.", + ParameterScope::Replica, ); /// Allow the upsert-v2 source stash's chunk batcher to spill cold chains out @@ -361,6 +407,7 @@ pub const ENABLE_UPSERT_PAGED_SPILL: Config = Config::new( false, "Allow the upsert-v2 source stash to spill chunks to the shared buffer pool, gated \ independently of the compute `enable_column_paged_batcher_spill`.", + ParameterScope::Replica, ); // RocksDB @@ -370,6 +417,7 @@ pub const STORAGE_ROCKSDB_CLEANUP_TRIES: Config = Config::new( "storage_rocksdb_cleanup_tries", 5, "How many times to try to cleanup old RocksDB DB's on disk before giving up.", + ParameterScope::Replica, ); /// Delay interval when reconnecting to a source / sink after halt. @@ -377,6 +425,7 @@ pub const STORAGE_SUSPEND_AND_RESTART_DELAY: Config = Config::new( "storage_suspend_and_restart_delay", Duration::from_secs(5), "Delay interval when reconnecting to a source / sink after halt.", + ParameterScope::Replica, ); /// Whether to use the new continual feedback upsert operator. @@ -384,6 +433,7 @@ pub const STORAGE_USE_CONTINUAL_FEEDBACK_UPSERT: Config = Config::new( "storage_use_continual_feedback_upsert", true, "Whether to use the new continual feedback upsert operator.", + ParameterScope::Environment, ); /// Whether to use the v2 upsert operator. @@ -391,6 +441,7 @@ pub const ENABLE_UPSERT_V2: Config = Config::new( "enable_upsert_v2", false, "Whether to use the v2 upsert operator.", + ParameterScope::Environment, ); /// The interval at which the storage server performs maintenance tasks. @@ -398,6 +449,7 @@ pub const STORAGE_SERVER_MAINTENANCE_INTERVAL: Config = Config::new( "storage_server_maintenance_interval", Duration::from_millis(10), "The interval at which the storage server performs maintenance tasks. Zero enables maintenance on every iteration.", + ParameterScope::Replica, ); /// If set, iteratively search the progress topic for a progress record with increasing lookback. @@ -405,6 +457,7 @@ pub const SINK_PROGRESS_SEARCH: Config = Config::new( "storage_sink_progress_search", true, "If set, iteratively search the progress topic for a progress record with increasing lookback.", + ParameterScope::Environment, ); /// Configure how to behave when trying to create an existing topic with specified configs. @@ -414,6 +467,7 @@ pub const SINK_ENSURE_TOPIC_CONFIG: Config<&'static str> = Config::new( "If `skip`, don't check the config of existing topics; if `check`, fetch the config and \ warn if it does not match the expected configs; if `alter`, attempt to change the upstream to \ match the expected configs.", + ParameterScope::Environment, ); /// Configure mz-ore overflowing type behavior. @@ -421,6 +475,7 @@ pub const ORE_OVERFLOWING_BEHAVIOR: Config<&'static str> = Config::new( "ore_overflowing_behavior", "soft_panic", "Overflow behavior for Overflowing types. One of 'ignore', 'panic', 'soft_panic'.", + ParameterScope::Environment, ); /// The time after which we delete per-replica statistics (for sources and @@ -432,6 +487,7 @@ pub const STATISTICS_RETENTION_DURATION: Config = Config::new( "storage_statistics_retention_duration", Duration::from_secs(86_400), /* one day */ "The time after which we delete per replica statistics (for sources and sinks) after there have been no updates.", + ParameterScope::Environment, ); /// Adds the full set of all storage `Config`s. diff --git a/src/storage-types/src/sources/sql_server.rs b/src/storage-types/src/sources/sql_server.rs index db3615f85439b..64bc59e564a33 100644 --- a/src/storage-types/src/sources/sql_server.rs +++ b/src/storage-types/src/sources/sql_server.rs @@ -12,7 +12,7 @@ use std::sync::{Arc, LazyLock}; use std::time::Duration; -use mz_dyncfg::Config; +use mz_dyncfg::{Config, ParameterScope}; use mz_ore::future::InTask; use mz_proto::RustType; use mz_repr::{CatalogItemId, Datum, GlobalId, RelationDesc, Row, SqlScalarType}; @@ -33,17 +33,22 @@ include!(concat!( "/mz_storage_types.sources.sql_server.rs" )); +/// Environment-scoped because source purification reads it in `environmentd` +/// (`mz_sql::pure`) as well as the source dataflow on the replica, and the +/// purification read has no replica in scope. pub const MAX_LSN_WAIT: Config = Config::new( "sql_server_max_lsn_wait", Duration::from_secs(30), "Maximum amount of time we'll wait for SQL Server to report an LSN (in other words for \ CDC to be fully enabled)", + ParameterScope::Environment, ); pub const SNAPSHOT_PROGRESS_REPORT_INTERVAL: Config = Config::new( "sql_server_snapshot_progress_report_interval", Duration::from_secs(2), "Interval at which we'll report progress for currently running snapshots.", + ParameterScope::Replica, ); pub const CDC_CLEANUP_CHANGE_TABLE: Config = Config::new( @@ -51,6 +56,7 @@ pub const CDC_CLEANUP_CHANGE_TABLE: Config = Config::new( false, "When enabled we'll notify SQL Server that it can cleanup the change tables \ as the source makes progress and commits data.", + ParameterScope::Environment, ); /// Maximum number of deletes that we'll make from a single SQL Server change table. @@ -64,6 +70,7 @@ pub const CDC_CLEANUP_CHANGE_TABLE_MAX_DELETES: Config = Config::new( // TODO(sql_server2): Call the cleanup function iteratively. 1_000_000, "Maximum number of entries that can be deleted by using a single statement.", + ParameterScope::Environment, ); pub static SQL_SERVER_PROGRESS_DESC: LazyLock = LazyLock::new(|| { diff --git a/src/txn-wal/src/operator.rs b/src/txn-wal/src/operator.rs index 0466aedece13f..06166c0d502a5 100644 --- a/src/txn-wal/src/operator.rs +++ b/src/txn-wal/src/operator.rs @@ -20,7 +20,7 @@ use std::time::Duration; use differential_dataflow::Hashable; use differential_dataflow::difference::Monoid; use differential_dataflow::lattice::Lattice; -use mz_dyncfg::{Config, ConfigSet}; +use mz_dyncfg::{Config, ConfigSet, ParameterScope}; use mz_ore::cast::CastFrom; use mz_persist_client::cfg::RetryParameters; use mz_persist_client::operators::shard_source::{ @@ -591,18 +591,21 @@ pub(crate) const DATA_SHARD_RETRYER_INITIAL_BACKOFF: Config = Config:: "persist_txns_data_shard_retryer_initial_backoff", Duration::from_millis(1024), "The initial backoff when polling for new batches from a txns data shard persist_source.", + ParameterScope::Environment, ); pub(crate) const DATA_SHARD_RETRYER_MULTIPLIER: Config = Config::new( "persist_txns_data_shard_retryer_multiplier", 2, "The backoff multiplier when polling for new batches from a txns data shard persist_source.", + ParameterScope::Environment, ); pub(crate) const DATA_SHARD_RETRYER_CLAMP: Config = Config::new( "persist_txns_data_shard_retryer_clamp", Duration::from_secs(16), "The backoff clamp duration when polling for new batches from a txns data shard persist_source.", + ParameterScope::Environment, ); /// Retry configuration for txn-wal data shard override of diff --git a/src/txn-wal/src/txns.rs b/src/txn-wal/src/txns.rs index 59e8c68414494..1b1b502251aa2 100644 --- a/src/txn-wal/src/txns.rs +++ b/src/txn-wal/src/txns.rs @@ -18,7 +18,7 @@ use differential_dataflow::difference::Monoid; use differential_dataflow::lattice::Lattice; use futures::StreamExt; use futures::stream::FuturesUnordered; -use mz_dyncfg::{Config, ConfigSet, ConfigValHandle}; +use mz_dyncfg::{Config, ConfigSet, ConfigValHandle, ParameterScope}; use mz_ore::collections::HashSet; use mz_ore::instrument; use mz_persist_client::batch::Batch; @@ -888,6 +888,7 @@ pub(crate) const APPLY_ENSURE_SCHEMA_MATCH: Config = Config::new( "txn_wal_apply_ensure_schema_match", true, "CYA to skip updating write handle to batch schema in apply", + ParameterScope::Environment, ); fn at_most_one_schema(