mz_environmentd/
lib.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10#![recursion_limit = "256"]
11
12//! A SQL stream processor built on top of [timely dataflow] and
13//! [differential dataflow].
14//!
15//! [differential dataflow]: ../differential_dataflow/index.html
16//! [timely dataflow]: ../timely/index.html
17
18use ::http::HeaderValue;
19use std::collections::{BTreeMap, BTreeSet};
20use std::panic::AssertUnwindSafe;
21use std::path::PathBuf;
22use std::pin::Pin;
23use std::sync::{Arc, LazyLock, Mutex};
24use std::time::{Duration, Instant};
25use std::{env, io};
26
27use anyhow::{Context, anyhow};
28use derivative::Derivative;
29use futures::FutureExt;
30use ipnet::IpNet;
31use mz_adapter::config::{
32    SystemParameterSyncClientConfig, SystemParameterSyncConfig, system_parameter_sync,
33};
34use mz_adapter::webhook::WebhookConcurrencyLimiter;
35use mz_adapter::{
36    AdapterError, Client as AdapterClient, RemoteSystemParameters, load_remote_system_parameters,
37};
38use mz_adapter_types::bootstrap_builtin_cluster_config::BootstrapBuiltinClusterConfig;
39use mz_adapter_types::dyncfgs::{
40    ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT, WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL,
41    WITH_0DT_DEPLOYMENT_MAX_WAIT,
42};
43use mz_auth::password::Password;
44use mz_authenticator::GenericOidcAuthenticator;
45use mz_build_info::{BuildInfo, build_info};
46use mz_catalog::config::ClusterReplicaSizeMap;
47use mz_catalog::durable::BootstrapArgs;
48use mz_cloud_resources::CloudResourceController;
49use mz_controller::ControllerConfig;
50use mz_dyncfg::ConfigSet;
51use mz_frontegg_auth::Authenticator as FronteggAuthenticator;
52use mz_license_keys::ValidatedLicenseKey;
53use mz_ore::future::OreFutureExt;
54use mz_ore::metrics::MetricsRegistry;
55use mz_ore::now::NowFn;
56use mz_ore::tracing::TracingHandle;
57use mz_ore::url::SensitiveUrl;
58use mz_ore::{instrument, task};
59use mz_persist_client::cache::PersistClientCache;
60use mz_persist_client::usage::StorageUsageClient;
61use mz_pgwire::MetricsConfig;
62use mz_pgwire_common::ConnectionCounter;
63use mz_repr::strconv;
64use mz_secrets::SecretsController;
65use mz_server_core::listeners::v26_32_0::ListenersConfig;
66use mz_server_core::listeners::{HttpListenerConfig, ListenerConfig, SqlListenerConfig};
67use mz_server_core::{
68    ConnectionStream, ListenerHandle, ReloadTrigger, ReloadingSslContext, ServeConfig,
69    TlsCertConfig, TlsMode,
70};
71use mz_sql::catalog::EnvironmentId;
72use mz_sql::session::vars::{Value, VarInput};
73use tokio::sync::oneshot;
74use tower_http::cors::AllowOrigin;
75use tracing::{Instrument, info, info_span};
76
77use crate::deployment::preflight::{self, CatchupConfig};
78use crate::deployment::state::DeploymentState;
79use crate::http::{HttpConfig, HttpServer, InternalRouteConfig};
80
81pub use crate::http::{SqlResponse, WebSocketAuth, WebSocketResponse};
82
83mod deployment;
84pub mod environmentd;
85pub mod http;
86mod telemetry;
87#[cfg(feature = "test")]
88pub mod test_util;
89
90pub const BUILD_INFO: BuildInfo = build_info!();
91
92/// Configuration for an `environmentd` server.
93#[derive(Derivative)]
94#[derivative(Debug)]
95pub struct Config {
96    // === Special modes. ===
97    /// Whether to permit usage of unsafe features. This is never meant to run
98    /// in production.
99    pub unsafe_mode: bool,
100    /// Whether the environmentd is running on a local dev machine. This is
101    /// never meant to run in production or CI.
102    pub all_features: bool,
103    /// Whether the container entrypoint restarted this process after
104    /// `mz_ore::restart_in_place!`.
105    pub restarted_in_place: bool,
106
107    // === Connection options. ===
108    /// TLS encryption and authentication configuration.
109    pub tls: Option<TlsCertConfig>,
110    /// Trigger to attempt to reload TLS certififcates.
111    #[derivative(Debug = "ignore")]
112    pub tls_reload_certs: ReloadTrigger,
113    /// Password of the mz_system user.
114    pub external_login_password_mz_system: Option<Password>,
115    /// Frontegg JWT authenticator.
116    pub frontegg: Option<FronteggAuthenticator>,
117    /// Frontegg workspace URL advertised in MCP OAuth discovery.
118    pub frontegg_oauth_issuer_url: Option<String>,
119    /// Origins for which cross-origin resource sharing (CORS) for HTTP requests
120    /// is permitted.
121    pub cors_allowed_origin: AllowOrigin,
122    /// Raw list of allowed CORS origins. Retained alongside `cors_allowed_origin`
123    /// (which is the computed predicate) so that endpoints like MCP can perform
124    /// server-side Origin validation to defend against DNS rebinding attacks
125    /// (where same-origin requests bypass CORS enforcement).
126    pub cors_allowed_origin_list: Vec<HeaderValue>,
127    /// Public IP addresses which the cloud environment has configured for
128    /// egress.
129    pub egress_addresses: Vec<IpNet>,
130    /// The external host name to connect to the HTTP server of this
131    /// environment.
132    ///
133    /// Presently used to render webhook URLs for end users in notices and the
134    /// system catalog. Not used to establish connections directly.
135    pub http_host_name: Option<String>,
136    /// The URL of the Materialize console to proxy from the /internal-console
137    /// endpoint on the internal HTTP server.
138    pub internal_console_redirect_url: Option<String>,
139
140    // === Controller options. ===
141    /// Storage and compute controller configuration.
142    pub controller: ControllerConfig,
143    /// Secrets controller configuration.
144    pub secrets_controller: Arc<dyn SecretsController>,
145    /// VpcEndpoint controller configuration.
146    pub cloud_resource_controller: Option<Arc<dyn CloudResourceController>>,
147    /// The process-wide live system dyncfg set.
148    pub system_dyncfgs: Arc<ConfigSet>,
149
150    // === Storage options. ===
151    /// The interval at which to collect storage usage information.
152    pub storage_usage_collection_interval: Duration,
153    /// How long to retain storage usage records for.
154    pub storage_usage_retention_period: Option<Duration>,
155
156    // === Adapter options. ===
157    /// Catalog configuration.
158    pub catalog_config: CatalogConfig,
159    /// Availability zones in which storage and compute resources may be
160    /// deployed.
161    pub availability_zones: Vec<String>,
162    /// A map from size name to resource allocations for cluster replicas.
163    pub cluster_replica_sizes: ClusterReplicaSizeMap,
164    /// The PostgreSQL URL for the Postgres-backed timestamp oracle.
165    pub timestamp_oracle_url: Option<SensitiveUrl>,
166    /// An API key for Segment. Enables export of audit events to Segment.
167    pub segment_api_key: Option<String>,
168    /// Whether the Segment client is being used on the client side
169    /// (rather than the server side).
170    pub segment_client_side: bool,
171    /// Only create a dummy segment client, only to get more testing coverage.
172    pub test_only_dummy_segment_client: bool,
173    /// An SDK key for LaunchDarkly. Enables system parameter synchronization
174    /// with LaunchDarkly.
175    pub launchdarkly_sdk_key: Option<String>,
176    /// Overrides the LaunchDarkly service endpoints with a single base URL, as
177    /// for a relay proxy or a mock server in tests.
178    pub launchdarkly_base_uri: Option<String>,
179    /// An invertible map from system parameter names to LaunchDarkly feature
180    /// keys to use when propagating values from the latter to the former.
181    pub launchdarkly_key_map: BTreeMap<String, String>,
182    /// The duration at which the system parameter synchronization times out during startup.
183    pub config_sync_timeout: Duration,
184    /// The interval in seconds at which to synchronize system parameter values.
185    pub config_sync_loop_interval: Option<Duration>,
186    /// The path for file based config sync
187    pub config_sync_file_path: Option<PathBuf>,
188
189    // === Bootstrap options. ===
190    /// The cloud ID of this environment.
191    pub environment_id: EnvironmentId,
192    /// What role, if any, should be initially created with elevated privileges.
193    pub bootstrap_role: Option<String>,
194    /// The size of the default cluster replica if bootstrapping.
195    pub bootstrap_default_cluster_replica_size: String,
196    /// The default number of replicas if bootstrapping.
197    pub bootstrap_default_cluster_replication_factor: u32,
198    /// The config of the builtin system cluster replicas if bootstrapping.
199    pub bootstrap_builtin_system_cluster_config: BootstrapBuiltinClusterConfig,
200    /// The config of the builtin catalog server cluster replicas if bootstrapping.
201    pub bootstrap_builtin_catalog_server_cluster_config: BootstrapBuiltinClusterConfig,
202    /// The config of the builtin probe cluster replicas if bootstrapping.
203    pub bootstrap_builtin_probe_cluster_config: BootstrapBuiltinClusterConfig,
204    /// The config of the builtin support cluster replicas if bootstrapping.
205    pub bootstrap_builtin_support_cluster_config: BootstrapBuiltinClusterConfig,
206    /// The config of the builtin analytics cluster replicas if bootstrapping.
207    pub bootstrap_builtin_analytics_cluster_config: BootstrapBuiltinClusterConfig,
208    /// Values to set for system parameters, if those system parameters have not
209    /// already been set by the system user.
210    pub system_parameter_defaults: BTreeMap<String, String>,
211    /// Helm chart version
212    pub helm_chart_version: Option<String>,
213    /// Configuration managed by license keys
214    pub license_key: ValidatedLicenseKey,
215
216    // === AWS options. ===
217    /// The AWS account ID, which will be used to generate ARNs for
218    /// Materialize-controlled AWS resources.
219    pub aws_account_id: Option<String>,
220    /// Supported AWS PrivateLink availability zone ids.
221    pub aws_privatelink_availability_zones: Option<Vec<String>>,
222
223    // === Observability options. ===
224    /// The metrics registry to use.
225    pub metrics_registry: MetricsRegistry,
226    /// Handle to tracing.
227    pub tracing_handle: TracingHandle,
228
229    // === Testing options. ===
230    /// A now generation function for mocking time.
231    pub now: NowFn,
232    /// If `Some`, force running builtin schema migration using the specified
233    /// migration mechanism ("evolution" or "replacement").
234    pub force_builtin_schema_migration: Option<String>,
235}
236
237/// Configuration for the Catalog.
238#[derive(Debug, Clone)]
239pub struct CatalogConfig {
240    /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient.
241    pub persist_clients: Arc<PersistClientCache>,
242    /// Persist catalog metrics.
243    pub metrics: Arc<mz_catalog::durable::Metrics>,
244}
245
246pub struct Listener<C> {
247    pub handle: ListenerHandle,
248    connection_stream: Pin<Box<dyn ConnectionStream>>,
249    config: C,
250}
251impl<C> Listener<C>
252where
253    C: ListenerConfig,
254{
255    /// Initializes network listeners for a later call to `serve` at the
256    /// specified addresses.
257    ///
258    /// Splitting this function out from `serve` has two benefits:
259    ///
260    ///   * It initializes network listeners as early as possible, so that the OS
261    ///     will queue incoming connections while the server is booting.
262    ///
263    ///   * It allows the caller to communicate with the server via the internal
264    ///     HTTP port while it is booting.
265    ///
266    fn bind(config: C) -> Result<Self, io::Error> {
267        let (handle, connection_stream) = mz_server_core::listen(&config.addr())?;
268        Ok(Self {
269            handle,
270            connection_stream,
271            config,
272        })
273    }
274}
275
276impl Listener<SqlListenerConfig> {
277    #[instrument(name = "environmentd::serve_sql")]
278    pub async fn serve_sql(
279        self,
280        name: String,
281        active_connection_counter: ConnectionCounter,
282        tls_reloading_context: Option<ReloadingSslContext>,
283        frontegg: Option<FronteggAuthenticator>,
284        adapter_client: AdapterClient,
285        oidc: GenericOidcAuthenticator,
286        metrics: MetricsConfig,
287        helm_chart_version: Option<String>,
288    ) -> ListenerHandle {
289        let label = leak_listener_name(&name);
290        let tls = tls_reloading_context.map(|context| mz_server_core::ReloadingTlsConfig {
291            context,
292            mode: if self.config.enable_tls {
293                TlsMode::Require
294            } else {
295                TlsMode::Allow
296            },
297        });
298
299        task::spawn(|| format!("{}_sql_server", label), {
300            let sql_server = mz_pgwire::Server::new(mz_pgwire::Config {
301                label,
302                tls,
303                adapter_client,
304                authenticator_kind: self.config.authenticator_kind,
305                frontegg,
306                oidc,
307                metrics,
308                active_connection_counter,
309                helm_chart_version,
310                allowed_roles: self.config.allowed_roles,
311            });
312            mz_server_core::serve(ServeConfig {
313                conns: self.connection_stream,
314                server: sql_server,
315                // `environmentd` does not currently need to dynamically
316                // configure graceful termination behavior.
317                dyncfg: None,
318            })
319        });
320        self.handle
321    }
322}
323
324impl Listener<HttpListenerConfig> {
325    #[instrument(name = "environmentd::serve_http")]
326    pub async fn serve_http(self, config: HttpConfig) -> ListenerHandle {
327        let task_name = format!("{}_http_server", config.source);
328        task::spawn(|| task_name, {
329            let http_server = HttpServer::new(config);
330            mz_server_core::serve(ServeConfig {
331                conns: self.connection_stream,
332                server: http_server,
333                // `environmentd` does not currently need to dynamically
334                // configure graceful termination behavior.
335                dyncfg: None,
336            })
337        });
338        self.handle
339    }
340}
341
342/// The `&'static str` listener names handed to the metrics layers, kept so that
343/// a process which serves repeatedly reuses one allocation per name.
344static LISTENER_NAMES: LazyLock<Mutex<BTreeSet<&'static str>>> =
345    LazyLock::new(|| Mutex::new(BTreeSet::new()));
346
347/// Returns a `&'static str` for `name`, leaking it on first use.
348///
349/// The metrics layer a listener is wired into wants a `'static` label. Leaking
350/// is how it gets one, but an unconditional leak both grows with every restart
351/// in a process that serves more than once, such as `sqllogictest`, and reads
352/// to LeakSanitizer as a genuine leak that fails the process at exit. Holding
353/// the leaked strs in a static solves both: one allocation per distinct name,
354/// still reachable at exit.
355fn leak_listener_name(name: &str) -> &'static str {
356    let mut names = LISTENER_NAMES.lock().expect("lock poisoned");
357    if let Some(name) = names.get(name) {
358        return name;
359    }
360    let name: &'static str = Box::leak(name.to_owned().into_boxed_str());
361    names.insert(name);
362    name
363}
364
365pub struct Listeners {
366    pub http: BTreeMap<String, Listener<HttpListenerConfig>>,
367    pub sql: BTreeMap<String, Listener<SqlListenerConfig>>,
368}
369
370impl Listeners {
371    pub fn bind(config: ListenersConfig) -> Result<Self, io::Error> {
372        let mut sql = BTreeMap::new();
373        for (name, config) in config.sql {
374            sql.insert(name, Listener::bind(config)?);
375        }
376
377        let mut http = BTreeMap::new();
378        for (name, config) in config.http {
379            http.insert(name, Listener::bind(config)?);
380        }
381
382        Ok(Listeners { http, sql })
383    }
384
385    /// Starts an `environmentd` server.
386    ///
387    /// Returns a handle to the server once it is fully booted.
388    #[instrument(name = "environmentd::serve")]
389    pub async fn serve(self, config: Config) -> Result<Server, AdapterError> {
390        let serve_start = Instant::now();
391        info!("startup: envd serve: beginning");
392        info!("startup: envd serve: preamble beginning");
393
394        // Validate TLS configuration, if present.
395        let tls_reloading_context = match config.tls {
396            Some(tls_config) => Some(tls_config.reloading_context(config.tls_reload_certs)?),
397            None => None,
398        };
399
400        let active_connection_counter = ConnectionCounter::default();
401        let (deployment_state, deployment_state_handle) = DeploymentState::new();
402
403        // Launch HTTP servers.
404        //
405        // We start these servers before we've completed initialization so that
406        // metrics are accessible during initialization. Some HTTP
407        // endpoints require the adapter to be initialized; requests to those
408        // endpoints block until the adapter client is installed.
409        // One of these endpoints is /api/readyz,
410        // which assumes we're ready when the adapter client exists.
411        let webhook_concurrency_limit = WebhookConcurrencyLimiter::default();
412        let internal_route_config = Arc::new(InternalRouteConfig {
413            deployment_state_handle,
414            internal_console_redirect_url: config.internal_console_redirect_url,
415        });
416
417        let (authenticator_oidc_tx, authenticator_oidc_rx) = oneshot::channel();
418        let authenticator_oidc_rx = authenticator_oidc_rx.shared();
419        let (adapter_client_tx, adapter_client_rx) = oneshot::channel();
420        let adapter_client_rx = adapter_client_rx.shared();
421
422        let metrics_registry = config.metrics_registry.clone();
423        let metrics = http::Metrics::register_into(&metrics_registry, "mz_http");
424        let mcp_metrics = http::mcp_metrics::McpMetrics::register_into(&metrics_registry);
425        let oauth_metadata_metrics =
426            http::oauth_metadata::OauthMetadataMetrics::register_into(&metrics_registry);
427        let mut http_listener_handles = BTreeMap::new();
428        for (name, listener) in self.http {
429            let authenticator_kind = listener.config.authenticator_kind();
430            let source = leak_listener_name(&name);
431            let tls = if listener.config.enable_tls() {
432                tls_reloading_context.clone()
433            } else {
434                None
435            };
436            let http_config = HttpConfig {
437                adapter_client_rx: adapter_client_rx.clone(),
438                active_connection_counter: active_connection_counter.clone(),
439                helm_chart_version: config.helm_chart_version.clone(),
440                http_host_name: config.http_host_name.clone(),
441                frontegg_oauth_issuer_url: config.frontegg_oauth_issuer_url.clone(),
442                source,
443                tls,
444                authenticator_kind,
445                frontegg: config.frontegg.clone(),
446                oidc_rx: authenticator_oidc_rx.clone(),
447                allowed_origin: config.cors_allowed_origin.clone(),
448                allowed_origin_list: config.cors_allowed_origin_list.clone(),
449                concurrent_webhook_req: webhook_concurrency_limit.semaphore(),
450                dyncfgs: Arc::clone(&config.system_dyncfgs),
451                metrics: metrics.clone(),
452                metrics_registry: metrics_registry.clone(),
453                mcp_metrics: mcp_metrics.clone(),
454                oauth_metadata_metrics: oauth_metadata_metrics.clone(),
455                internal_route_config: Arc::clone(&internal_route_config),
456                routes_enabled: listener.config.routes,
457                replica_http_locator: Arc::clone(&config.controller.replica_http_locator),
458            };
459            http_listener_handles.insert(name.clone(), listener.serve_http(http_config).await);
460        }
461
462        info!(
463            "startup: envd serve: preamble complete in {:?}",
464            serve_start.elapsed()
465        );
466
467        let catalog_init_start = Instant::now();
468        info!("startup: envd serve: catalog init beginning");
469
470        // Get the current timestamp so we can record when we booted.
471        let boot_ts = (config.now)().into();
472
473        let persist_client = config
474            .catalog_config
475            .persist_clients
476            .open(config.controller.persist_location.clone())
477            .await
478            .context("opening persist client")?;
479        let mut openable_adapter_storage = mz_catalog::durable::persist_backed_catalog_state(
480            persist_client.clone(),
481            config.environment_id.organization_id(),
482            BUILD_INFO.semver_version(),
483            Some(config.controller.deploy_generation),
484            Arc::clone(&config.catalog_config.metrics),
485        )
486        .await?;
487
488        info!(
489            "startup: envd serve: catalog init complete in {:?}",
490            catalog_init_start.elapsed()
491        );
492
493        let system_param_sync_start = Instant::now();
494        info!("startup: envd serve: system parameter sync beginning");
495        // Initialize the system parameter frontend
496        let system_parameter_sync_config =
497            match (config.launchdarkly_sdk_key, config.config_sync_file_path) {
498                (None, None) => None,
499                (None, Some(f)) => {
500                    info!("Using config file path {:?}", f);
501                    Some(SystemParameterSyncConfig::new(
502                        config.environment_id.clone(),
503                        &BUILD_INFO,
504                        &config.metrics_registry,
505                        config.launchdarkly_key_map,
506                        SystemParameterSyncClientConfig::File { path: f },
507                    ))
508                }
509                (Some(key), None) => Some(SystemParameterSyncConfig::new(
510                    config.environment_id.clone(),
511                    &BUILD_INFO,
512                    &config.metrics_registry,
513                    config.launchdarkly_key_map,
514                    SystemParameterSyncClientConfig::LaunchDarkly {
515                        sdk_key: key,
516                        base_uri: config.launchdarkly_base_uri,
517                        now_fn: config.now.clone(),
518                    },
519                )),
520
521                (Some(_), Some(_)) => {
522                    panic!("Cannot configure both file and Launchdarkly based config syncing")
523                }
524            };
525
526        let RemoteSystemParameters {
527            params: remote_system_parameters,
528            frontend: boot_frontend,
529        } = load_remote_system_parameters(
530            &mut openable_adapter_storage,
531            system_parameter_sync_config.clone(),
532            config.config_sync_timeout,
533        )
534        .await?;
535        info!(
536            "startup: envd serve: system parameter sync complete in {:?}",
537            system_param_sync_start.elapsed()
538        );
539
540        let preflight_checks_start = Instant::now();
541        info!("startup: envd serve: preflight checks beginning");
542
543        // Determine the maximum wait time when doing a 0dt deployment.
544        let with_0dt_deployment_max_wait = {
545            let cli_default = config
546                .system_parameter_defaults
547                .get(WITH_0DT_DEPLOYMENT_MAX_WAIT.name())
548                .map(|x| {
549                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
550                        anyhow!(
551                            "failed to parse default for {}: {:?}",
552                            WITH_0DT_DEPLOYMENT_MAX_WAIT.name(),
553                            err
554                        )
555                    })
556                })
557                .transpose()?;
558            let compiled_default = WITH_0DT_DEPLOYMENT_MAX_WAIT.default().clone();
559            let ld = get_ld_value(
560                WITH_0DT_DEPLOYMENT_MAX_WAIT.name(),
561                &remote_system_parameters,
562                |x| {
563                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
564                        format!(
565                            "failed to parse LD value {} for {}: {:?}",
566                            x,
567                            WITH_0DT_DEPLOYMENT_MAX_WAIT.name(),
568                            err
569                        )
570                    })
571                },
572            )?;
573            let catalog = openable_adapter_storage
574                .get_0dt_deployment_max_wait()
575                .await?;
576            let computed = ld.or(catalog).or(cli_default).unwrap_or(compiled_default);
577            info!(
578                ?computed,
579                ?ld,
580                ?catalog,
581                ?cli_default,
582                ?compiled_default,
583                "determined value for {} system parameter",
584                WITH_0DT_DEPLOYMENT_MAX_WAIT.name()
585            );
586            computed
587        };
588        // Determine the DDL check interval when doing a 0dt deployment.
589        let with_0dt_deployment_ddl_check_interval = {
590            let cli_default = config
591                .system_parameter_defaults
592                .get(WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name())
593                .map(|x| {
594                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
595                        anyhow!(
596                            "failed to parse default for {}: {:?}",
597                            WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name(),
598                            err
599                        )
600                    })
601                })
602                .transpose()?;
603            let compiled_default = WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.default().clone();
604            let ld = get_ld_value(
605                WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name(),
606                &remote_system_parameters,
607                |x| {
608                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
609                        format!(
610                            "failed to parse LD value {} for {}: {:?}",
611                            x,
612                            WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name(),
613                            err
614                        )
615                    })
616                },
617            )?;
618            let catalog = openable_adapter_storage
619                .get_0dt_deployment_ddl_check_interval()
620                .await?;
621            let computed = ld.or(catalog).or(cli_default).unwrap_or(compiled_default);
622            info!(
623                ?computed,
624                ?ld,
625                ?catalog,
626                ?cli_default,
627                ?compiled_default,
628                "determined value for {} system parameter",
629                WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name()
630            );
631            computed
632        };
633
634        // Determine whether we should panic if we reach the maximum wait time
635        // without the preflight checks succeeding.
636        let enable_0dt_deployment_panic_after_timeout = {
637            let cli_default = config
638                .system_parameter_defaults
639                .get(ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT.name())
640                .map(|x| {
641                    strconv::parse_bool(x).map_err(|err| {
642                        anyhow!(
643                            "failed to parse default for {}: {}",
644                            ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT.name(),
645                            err
646                        )
647                    })
648                })
649                .transpose()?;
650            let compiled_default = ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT.default().clone();
651            let ld = get_ld_value(
652                "enable_0dt_deployment_panic_after_timeout",
653                &remote_system_parameters,
654                |x| strconv::parse_bool(x).map_err(|x| x.to_string()),
655            )?;
656            let catalog = openable_adapter_storage
657                .get_enable_0dt_deployment_panic_after_timeout()
658                .await?;
659            let computed = ld.or(catalog).or(cli_default).unwrap_or(compiled_default);
660            info!(
661                %computed,
662                ?ld,
663                ?catalog,
664                ?cli_default,
665                ?compiled_default,
666                "determined value for enable_0dt_deployment_panic_after_timeout system parameter",
667            );
668            computed
669        };
670
671        // Perform preflight checks.
672        //
673        // Preflight checks determine whether to boot in read-only mode or not.
674        let read_only = preflight::preflight_0dt(
675            openable_adapter_storage.as_mut(),
676            config.controller.deploy_generation,
677            config.restarted_in_place,
678        )
679        .await?;
680
681        let bootstrap_args = BootstrapArgs {
682            default_cluster_replica_size: config.bootstrap_default_cluster_replica_size.clone(),
683            default_cluster_replication_factor: config.bootstrap_default_cluster_replication_factor,
684            bootstrap_role: config.bootstrap_role,
685            cluster_replica_size_map: config.cluster_replica_sizes.clone(),
686        };
687
688        let (caught_up_trigger, bootstrapped) = if read_only {
689            let (caught_up_trigger, caught_up_receiver) = mz_ore::channel::trigger::channel();
690            let (bootstrapped, bootstrapped_receiver) = tokio::sync::oneshot::channel();
691            let catchup_config = CatchupConfig {
692                boot_ts,
693                environment_id: config.environment_id.clone(),
694                persist_client,
695                deploy_generation: config.controller.deploy_generation,
696                deployment_state: deployment_state.clone(),
697                catalog_metrics: Arc::clone(&config.catalog_config.metrics),
698                caught_up_max_wait: with_0dt_deployment_max_wait,
699                panic_after_timeout: enable_0dt_deployment_panic_after_timeout,
700                bootstrap_args: bootstrap_args.clone(),
701                ddl_check_interval: with_0dt_deployment_ddl_check_interval,
702            };
703            preflight::spawn_catchup(catchup_config, caught_up_receiver, bootstrapped_receiver);
704            (Some(caught_up_trigger), Some(bootstrapped))
705        } else {
706            (None, None)
707        };
708
709        info!(
710            "startup: envd serve: preflight checks complete in {:?}",
711            preflight_checks_start.elapsed()
712        );
713
714        let catalog_open_start = Instant::now();
715        info!("startup: envd serve: durable catalog open beginning");
716
717        // Load the adapter durable storage.
718        let mut adapter_storage = if read_only {
719            // TODO: behavior of migrations when booting in savepoint mode is
720            // not well defined.
721            let adapter_storage = openable_adapter_storage
722                .open_savepoint(boot_ts, &bootstrap_args)
723                .await?;
724            // In read-only mode, we intentionally do not call `set_is_leader`,
725            // because we are by definition not the leader if we are in
726            // read-only mode.
727
728            adapter_storage
729        } else {
730            let adapter_storage = openable_adapter_storage
731                .open(boot_ts, &bootstrap_args)
732                .await?;
733
734            // Once we have successfully opened the adapter storage in
735            // read/write mode, we can announce we are the leader, as we've
736            // fenced out all other environments using the adapter storage.
737            deployment_state.set_is_leader();
738
739            adapter_storage
740        };
741
742        let bootstrapped = match bootstrapped {
743            Some(tx) => Some((tx, preflight::get_user_ids(adapter_storage.as_mut()).await?)),
744            None => None,
745        };
746
747        // Enable Persist compaction if we're not in read only.
748        if !read_only {
749            config.controller.persist_clients.cfg().enable_compaction();
750        }
751
752        info!(
753            "startup: envd serve: durable catalog open complete in {:?}",
754            catalog_open_start.elapsed()
755        );
756
757        let coord_init_start = Instant::now();
758        info!("startup: envd serve: coordinator init beginning");
759
760        if !config
761            .cluster_replica_sizes
762            .0
763            .contains_key(&config.bootstrap_default_cluster_replica_size)
764        {
765            return Err(anyhow!("bootstrap default cluster replica size is unknown").into());
766        }
767        let envd_epoch = adapter_storage.epoch();
768
769        // Initialize storage usage client.
770        let storage_usage_client = StorageUsageClient::open(
771            config
772                .controller
773                .persist_clients
774                .open(config.controller.persist_location.clone())
775                .await
776                .context("opening storage usage client")?,
777        );
778
779        // Initialize adapter.
780        let license_key = config.license_key.clone();
781        let segment_client = config.segment_api_key.map(|api_key| {
782            mz_segment::Client::new(mz_segment::Config {
783                api_key,
784                client_side: config.segment_client_side,
785            })
786        });
787        let connection_limiter = active_connection_counter.clone();
788        let connection_limit_callback = Box::new(move |limit, superuser_reserved| {
789            connection_limiter.update_limit(limit);
790            connection_limiter.update_superuser_reserved(superuser_reserved);
791        });
792
793        let (adapter_handle, adapter_client) = mz_adapter::serve(mz_adapter::Config {
794            connection_context: config.controller.connection_context.clone(),
795            connection_limit_callback,
796            controller_config: config.controller,
797            controller_envd_epoch: envd_epoch,
798            storage: adapter_storage,
799            timestamp_oracle_url: config.timestamp_oracle_url,
800            unsafe_mode: config.unsafe_mode,
801            all_features: config.all_features,
802            build_info: &BUILD_INFO,
803            environment_id: config.environment_id.clone(),
804            metrics_registry: config.metrics_registry.clone(),
805            now: config.now,
806            secrets_controller: config.secrets_controller,
807            cloud_resource_controller: config.cloud_resource_controller,
808            cluster_replica_sizes: config.cluster_replica_sizes,
809            builtin_system_cluster_config: config.bootstrap_builtin_system_cluster_config,
810            builtin_catalog_server_cluster_config: config
811                .bootstrap_builtin_catalog_server_cluster_config,
812            builtin_probe_cluster_config: config.bootstrap_builtin_probe_cluster_config,
813            builtin_support_cluster_config: config.bootstrap_builtin_support_cluster_config,
814            builtin_analytics_cluster_config: config.bootstrap_builtin_analytics_cluster_config,
815            availability_zones: config.availability_zones,
816            system_parameter_defaults: config.system_parameter_defaults,
817            storage_usage_client,
818            storage_usage_collection_interval: config.storage_usage_collection_interval,
819            storage_usage_retention_period: config.storage_usage_retention_period,
820            segment_client: segment_client.clone(),
821            egress_addresses: config.egress_addresses,
822            remote_system_parameters,
823            aws_account_id: config.aws_account_id,
824            aws_privatelink_availability_zones: config.aws_privatelink_availability_zones,
825            webhook_concurrency_limit: webhook_concurrency_limit.clone(),
826            http_host_name: config.http_host_name,
827            tracing_handle: config.tracing_handle,
828            read_only_controllers: read_only,
829            caught_up_trigger,
830            helm_chart_version: config.helm_chart_version.clone(),
831            license_key: config.license_key,
832            external_login_password_mz_system: config.external_login_password_mz_system,
833            force_builtin_schema_migration: config.force_builtin_schema_migration,
834        })
835        .instrument(info_span!("adapter::serve"))
836        .await?;
837
838        if let Some((tx, initial_ids)) = bootstrapped {
839            // The receiver is gone if an administrator already skipped catch-up.
840            let _ = tx.send(initial_ids);
841        }
842
843        // Initialize the OIDC authenticator, shared between the HTTP and SQL servers.
844        let oidc = GenericOidcAuthenticator::new(adapter_client.clone());
845
846        info!(
847            "startup: envd serve: coordinator init complete in {:?}",
848            coord_init_start.elapsed()
849        );
850
851        let serve_postamble_start = Instant::now();
852        info!("startup: envd serve: postamble beginning");
853
854        // Send adapter client and OIDC authenticator to the HTTP servers.
855        authenticator_oidc_tx
856            .send(oidc.clone())
857            .expect("rx known to be live");
858        adapter_client_tx
859            .send(adapter_client.clone())
860            .expect("internal HTTP server should not drop first");
861
862        let metrics = mz_pgwire::MetricsConfig::register_into(&config.metrics_registry);
863
864        // Launch SQL server.
865        let mut sql_listener_handles = BTreeMap::new();
866        for (name, listener) in self.sql {
867            sql_listener_handles.insert(
868                name.clone(),
869                listener
870                    .serve_sql(
871                        name,
872                        active_connection_counter.clone(),
873                        tls_reloading_context.clone(),
874                        config.frontegg.clone(),
875                        adapter_client.clone(),
876                        oidc.clone(),
877                        metrics.clone(),
878                        config.helm_chart_version.clone(),
879                    )
880                    .await,
881            );
882        }
883
884        // Start telemetry reporting loop.
885        if let Some(segment_client) = segment_client {
886            telemetry::start_reporting(telemetry::Config {
887                segment_client,
888                adapter_client: adapter_client.clone(),
889                environment_id: config.environment_id,
890                license_key: license_key.clone(),
891                helm_chart_version: config.helm_chart_version.clone(),
892                report_interval: Duration::from_secs(3600),
893            });
894        } else if config.test_only_dummy_segment_client {
895            // We only have access to a segment client in production but we
896            // still want to exercise the telemetry reporting code to a degree.
897            // So we create a dummy client and report telemetry into the void.
898            // This way we at least run the telemetry queries the way a
899            // production environment would.
900            tracing::debug!("starting telemetry reporting with a dummy segment client");
901            let segment_client = mz_segment::Client::new_dummy_client();
902            telemetry::start_reporting(telemetry::Config {
903                segment_client,
904                adapter_client: adapter_client.clone(),
905                environment_id: config.environment_id,
906                license_key: license_key.clone(),
907                helm_chart_version: config.helm_chart_version.clone(),
908                report_interval: Duration::from_secs(180),
909            });
910        }
911
912        // If system_parameter_sync_config and config_sync_loop_interval are present,
913        // start the system_parameter_sync loop.
914        if let Some(system_parameter_sync_config) = system_parameter_sync_config {
915            task::spawn(
916                || "system_parameter_sync",
917                AssertUnwindSafe(system_parameter_sync(
918                    system_parameter_sync_config,
919                    boot_frontend,
920                    adapter_client.clone(),
921                    config.config_sync_loop_interval,
922                ))
923                .ore_catch_unwind(),
924            );
925        }
926
927        info!(
928            "startup: envd serve: postamble complete in {:?}",
929            serve_postamble_start.elapsed()
930        );
931        info!(
932            "startup: envd serve: complete in {:?}",
933            serve_start.elapsed()
934        );
935
936        Ok(Server {
937            sql_listener_handles,
938            http_listener_handles,
939            #[cfg(feature = "test")]
940            adapter_client,
941            _adapter_handle: adapter_handle,
942        })
943    }
944}
945
946fn get_ld_value<V>(
947    name: &str,
948    remote_system_parameters: &Option<BTreeMap<String, String>>,
949    parse: impl Fn(&str) -> Result<V, String>,
950) -> Result<Option<V>, anyhow::Error> {
951    remote_system_parameters
952        .as_ref()
953        .and_then(|params| params.get(name))
954        .map(|x| {
955            parse(x).map_err(|err| anyhow!("failed to parse remote value for {}: {}", name, err))
956        })
957        .transpose()
958}
959
960/// A running `environmentd` server.
961pub struct Server {
962    // Drop order matters for these fields.
963    pub sql_listener_handles: BTreeMap<String, ListenerHandle>,
964    pub http_listener_handles: BTreeMap<String, ListenerHandle>,
965    #[cfg(feature = "test")]
966    adapter_client: AdapterClient,
967    _adapter_handle: mz_adapter::Handle,
968}
969
970impl Server {
971    /// A client for the adapter, letting tests drive the adapter API directly.
972    #[cfg(feature = "test")]
973    pub fn adapter_client(&self) -> &AdapterClient {
974        &self.adapter_client
975    }
976}
mz_environmentd/lib.rs

mz_environmentd/
lib.rs
