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    /// See [`BootstrapArgs::force_catalog_migration`].
236    pub force_catalog_migration: bool,
237}
238
239/// Configuration for the Catalog.
240#[derive(Debug, Clone)]
241pub struct CatalogConfig {
242    /// A process-global cache of (blob_uri, consensus_uri) -> PersistClient.
243    pub persist_clients: Arc<PersistClientCache>,
244    /// Persist catalog metrics.
245    pub metrics: Arc<mz_catalog::durable::Metrics>,
246}
247
248pub struct Listener<C> {
249    pub handle: ListenerHandle,
250    connection_stream: Pin<Box<dyn ConnectionStream>>,
251    config: C,
252}
253impl<C> Listener<C>
254where
255    C: ListenerConfig,
256{
257    /// Initializes network listeners for a later call to `serve` at the
258    /// specified addresses.
259    ///
260    /// Splitting this function out from `serve` has two benefits:
261    ///
262    ///   * It initializes network listeners as early as possible, so that the OS
263    ///     will queue incoming connections while the server is booting.
264    ///
265    ///   * It allows the caller to communicate with the server via the internal
266    ///     HTTP port while it is booting.
267    ///
268    fn bind(config: C) -> Result<Self, io::Error> {
269        let (handle, connection_stream) = mz_server_core::listen(&config.addr())?;
270        Ok(Self {
271            handle,
272            connection_stream,
273            config,
274        })
275    }
276}
277
278impl Listener<SqlListenerConfig> {
279    #[instrument(name = "environmentd::serve_sql")]
280    pub async fn serve_sql(
281        self,
282        name: String,
283        active_connection_counter: ConnectionCounter,
284        tls_reloading_context: Option<ReloadingSslContext>,
285        frontegg: Option<FronteggAuthenticator>,
286        adapter_client: AdapterClient,
287        oidc: GenericOidcAuthenticator,
288        metrics: MetricsConfig,
289        helm_chart_version: Option<String>,
290    ) -> ListenerHandle {
291        let label = leak_listener_name(&name);
292        let tls = tls_reloading_context.map(|context| mz_server_core::ReloadingTlsConfig {
293            context,
294            mode: if self.config.enable_tls {
295                TlsMode::Require
296            } else {
297                TlsMode::Allow
298            },
299        });
300
301        task::spawn(|| format!("{}_sql_server", label), {
302            let sql_server = mz_pgwire::Server::new(mz_pgwire::Config {
303                label,
304                tls,
305                adapter_client,
306                authenticator_kind: self.config.authenticator_kind,
307                frontegg,
308                oidc,
309                metrics,
310                active_connection_counter,
311                helm_chart_version,
312                allowed_roles: self.config.allowed_roles,
313            });
314            mz_server_core::serve(ServeConfig {
315                conns: self.connection_stream,
316                server: sql_server,
317                // `environmentd` does not currently need to dynamically
318                // configure graceful termination behavior.
319                dyncfg: None,
320            })
321        });
322        self.handle
323    }
324}
325
326impl Listener<HttpListenerConfig> {
327    #[instrument(name = "environmentd::serve_http")]
328    pub async fn serve_http(self, config: HttpConfig) -> ListenerHandle {
329        let task_name = format!("{}_http_server", config.source);
330        task::spawn(|| task_name, {
331            let http_server = HttpServer::new(config);
332            mz_server_core::serve(ServeConfig {
333                conns: self.connection_stream,
334                server: http_server,
335                // `environmentd` does not currently need to dynamically
336                // configure graceful termination behavior.
337                dyncfg: None,
338            })
339        });
340        self.handle
341    }
342}
343
344/// The `&'static str` listener names handed to the metrics layers, kept so that
345/// a process which serves repeatedly reuses one allocation per name.
346static LISTENER_NAMES: LazyLock<Mutex<BTreeSet<&'static str>>> =
347    LazyLock::new(|| Mutex::new(BTreeSet::new()));
348
349/// Returns a `&'static str` for `name`, leaking it on first use.
350///
351/// The metrics layer a listener is wired into wants a `'static` label. Leaking
352/// is how it gets one, but an unconditional leak both grows with every restart
353/// in a process that serves more than once, such as `sqllogictest`, and reads
354/// to LeakSanitizer as a genuine leak that fails the process at exit. Holding
355/// the leaked strs in a static solves both: one allocation per distinct name,
356/// still reachable at exit.
357fn leak_listener_name(name: &str) -> &'static str {
358    let mut names = LISTENER_NAMES.lock().expect("lock poisoned");
359    if let Some(name) = names.get(name) {
360        return name;
361    }
362    let name: &'static str = Box::leak(name.to_owned().into_boxed_str());
363    names.insert(name);
364    name
365}
366
367pub struct Listeners {
368    pub http: BTreeMap<String, Listener<HttpListenerConfig>>,
369    pub sql: BTreeMap<String, Listener<SqlListenerConfig>>,
370}
371
372impl Listeners {
373    pub fn bind(config: ListenersConfig) -> Result<Self, io::Error> {
374        let mut sql = BTreeMap::new();
375        for (name, config) in config.sql {
376            sql.insert(name, Listener::bind(config)?);
377        }
378
379        let mut http = BTreeMap::new();
380        for (name, config) in config.http {
381            http.insert(name, Listener::bind(config)?);
382        }
383
384        Ok(Listeners { http, sql })
385    }
386
387    /// Starts an `environmentd` server.
388    ///
389    /// Returns a handle to the server once it is fully booted.
390    #[instrument(name = "environmentd::serve")]
391    pub async fn serve(self, config: Config) -> Result<Server, AdapterError> {
392        let serve_start = Instant::now();
393        info!("startup: envd serve: beginning");
394        info!("startup: envd serve: preamble beginning");
395
396        // Validate TLS configuration, if present.
397        let tls_reloading_context = match config.tls {
398            Some(tls_config) => Some(tls_config.reloading_context(config.tls_reload_certs)?),
399            None => None,
400        };
401
402        let active_connection_counter = ConnectionCounter::default();
403        let (deployment_state, deployment_state_handle) = DeploymentState::new();
404
405        // Launch HTTP servers.
406        //
407        // We start these servers before we've completed initialization so that
408        // metrics are accessible during initialization. Some HTTP
409        // endpoints require the adapter to be initialized; requests to those
410        // endpoints block until the adapter client is installed.
411        // One of these endpoints is /api/readyz,
412        // which assumes we're ready when the adapter client exists.
413        let webhook_concurrency_limit = WebhookConcurrencyLimiter::default();
414        let internal_route_config = Arc::new(InternalRouteConfig {
415            deployment_state_handle,
416            internal_console_redirect_url: config.internal_console_redirect_url,
417        });
418
419        let (authenticator_oidc_tx, authenticator_oidc_rx) = oneshot::channel();
420        let authenticator_oidc_rx = authenticator_oidc_rx.shared();
421        let (adapter_client_tx, adapter_client_rx) = oneshot::channel();
422        let adapter_client_rx = adapter_client_rx.shared();
423
424        let metrics_registry = config.metrics_registry.clone();
425        let metrics = http::Metrics::register_into(&metrics_registry, "mz_http");
426        let mcp_metrics = http::mcp_metrics::McpMetrics::register_into(&metrics_registry);
427        let oauth_metadata_metrics =
428            http::oauth_metadata::OauthMetadataMetrics::register_into(&metrics_registry);
429        let mut http_listener_handles = BTreeMap::new();
430        for (name, listener) in self.http {
431            let authenticator_kind = listener.config.authenticator_kind();
432            let source = leak_listener_name(&name);
433            let tls = if listener.config.enable_tls() {
434                tls_reloading_context.clone()
435            } else {
436                None
437            };
438            let http_config = HttpConfig {
439                adapter_client_rx: adapter_client_rx.clone(),
440                active_connection_counter: active_connection_counter.clone(),
441                helm_chart_version: config.helm_chart_version.clone(),
442                http_host_name: config.http_host_name.clone(),
443                frontegg_oauth_issuer_url: config.frontegg_oauth_issuer_url.clone(),
444                source,
445                tls,
446                authenticator_kind,
447                frontegg: config.frontegg.clone(),
448                oidc_rx: authenticator_oidc_rx.clone(),
449                allowed_origin: config.cors_allowed_origin.clone(),
450                allowed_origin_list: config.cors_allowed_origin_list.clone(),
451                concurrent_webhook_req: webhook_concurrency_limit.semaphore(),
452                dyncfgs: Arc::clone(&config.system_dyncfgs),
453                metrics: metrics.clone(),
454                metrics_registry: metrics_registry.clone(),
455                mcp_metrics: mcp_metrics.clone(),
456                oauth_metadata_metrics: oauth_metadata_metrics.clone(),
457                internal_route_config: Arc::clone(&internal_route_config),
458                routes_enabled: listener.config.routes,
459                replica_http_locator: Arc::clone(&config.controller.replica_http_locator),
460            };
461            http_listener_handles.insert(name.clone(), listener.serve_http(http_config).await);
462        }
463
464        info!(
465            "startup: envd serve: preamble complete in {:?}",
466            serve_start.elapsed()
467        );
468
469        let catalog_init_start = Instant::now();
470        info!("startup: envd serve: catalog init beginning");
471
472        // Get the current timestamp so we can record when we booted.
473        let boot_ts = (config.now)().into();
474
475        let persist_client = config
476            .catalog_config
477            .persist_clients
478            .open(config.controller.persist_location.clone())
479            .await
480            .context("opening persist client")?;
481        let mut openable_adapter_storage = mz_catalog::durable::persist_backed_catalog_state(
482            persist_client.clone(),
483            config.environment_id.organization_id(),
484            BUILD_INFO.semver_version(),
485            Some(config.controller.deploy_generation),
486            Arc::clone(&config.catalog_config.metrics),
487        )
488        .await?;
489
490        info!(
491            "startup: envd serve: catalog init complete in {:?}",
492            catalog_init_start.elapsed()
493        );
494
495        let system_param_sync_start = Instant::now();
496        info!("startup: envd serve: system parameter sync beginning");
497        // Initialize the system parameter frontend
498        let system_parameter_sync_config =
499            match (config.launchdarkly_sdk_key, config.config_sync_file_path) {
500                (None, None) => None,
501                (None, Some(f)) => {
502                    info!("Using config file path {:?}", f);
503                    Some(SystemParameterSyncConfig::new(
504                        config.environment_id.clone(),
505                        &BUILD_INFO,
506                        &config.metrics_registry,
507                        config.launchdarkly_key_map,
508                        SystemParameterSyncClientConfig::File { path: f },
509                    ))
510                }
511                (Some(key), None) => Some(SystemParameterSyncConfig::new(
512                    config.environment_id.clone(),
513                    &BUILD_INFO,
514                    &config.metrics_registry,
515                    config.launchdarkly_key_map,
516                    SystemParameterSyncClientConfig::LaunchDarkly {
517                        sdk_key: key,
518                        base_uri: config.launchdarkly_base_uri,
519                        now_fn: config.now.clone(),
520                    },
521                )),
522
523                (Some(_), Some(_)) => {
524                    panic!("Cannot configure both file and Launchdarkly based config syncing")
525                }
526            };
527
528        let RemoteSystemParameters {
529            params: remote_system_parameters,
530            frontend: boot_frontend,
531        } = load_remote_system_parameters(
532            &mut openable_adapter_storage,
533            system_parameter_sync_config.clone(),
534            config.config_sync_timeout,
535        )
536        .await?;
537        info!(
538            "startup: envd serve: system parameter sync complete in {:?}",
539            system_param_sync_start.elapsed()
540        );
541
542        let preflight_checks_start = Instant::now();
543        info!("startup: envd serve: preflight checks beginning");
544
545        // Determine the maximum wait time when doing a 0dt deployment.
546        let with_0dt_deployment_max_wait = {
547            let cli_default = config
548                .system_parameter_defaults
549                .get(WITH_0DT_DEPLOYMENT_MAX_WAIT.name())
550                .map(|x| {
551                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
552                        anyhow!(
553                            "failed to parse default for {}: {:?}",
554                            WITH_0DT_DEPLOYMENT_MAX_WAIT.name(),
555                            err
556                        )
557                    })
558                })
559                .transpose()?;
560            let compiled_default = WITH_0DT_DEPLOYMENT_MAX_WAIT.default().clone();
561            let ld = get_ld_value(
562                WITH_0DT_DEPLOYMENT_MAX_WAIT.name(),
563                &remote_system_parameters,
564                |x| {
565                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
566                        format!(
567                            "failed to parse LD value {} for {}: {:?}",
568                            x,
569                            WITH_0DT_DEPLOYMENT_MAX_WAIT.name(),
570                            err
571                        )
572                    })
573                },
574            )?;
575            let catalog = openable_adapter_storage
576                .get_0dt_deployment_max_wait()
577                .await?;
578            let computed = ld.or(catalog).or(cli_default).unwrap_or(compiled_default);
579            info!(
580                ?computed,
581                ?ld,
582                ?catalog,
583                ?cli_default,
584                ?compiled_default,
585                "determined value for {} system parameter",
586                WITH_0DT_DEPLOYMENT_MAX_WAIT.name()
587            );
588            computed
589        };
590        // Determine the DDL check interval when doing a 0dt deployment.
591        let with_0dt_deployment_ddl_check_interval = {
592            let cli_default = config
593                .system_parameter_defaults
594                .get(WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name())
595                .map(|x| {
596                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
597                        anyhow!(
598                            "failed to parse default for {}: {:?}",
599                            WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name(),
600                            err
601                        )
602                    })
603                })
604                .transpose()?;
605            let compiled_default = WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.default().clone();
606            let ld = get_ld_value(
607                WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name(),
608                &remote_system_parameters,
609                |x| {
610                    Duration::parse(VarInput::Flat(x)).map_err(|err| {
611                        format!(
612                            "failed to parse LD value {} for {}: {:?}",
613                            x,
614                            WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name(),
615                            err
616                        )
617                    })
618                },
619            )?;
620            let catalog = openable_adapter_storage
621                .get_0dt_deployment_ddl_check_interval()
622                .await?;
623            let computed = ld.or(catalog).or(cli_default).unwrap_or(compiled_default);
624            info!(
625                ?computed,
626                ?ld,
627                ?catalog,
628                ?cli_default,
629                ?compiled_default,
630                "determined value for {} system parameter",
631                WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL.name()
632            );
633            computed
634        };
635
636        // Determine whether we should panic if we reach the maximum wait time
637        // without the preflight checks succeeding.
638        let enable_0dt_deployment_panic_after_timeout = {
639            let cli_default = config
640                .system_parameter_defaults
641                .get(ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT.name())
642                .map(|x| {
643                    strconv::parse_bool(x).map_err(|err| {
644                        anyhow!(
645                            "failed to parse default for {}: {}",
646                            ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT.name(),
647                            err
648                        )
649                    })
650                })
651                .transpose()?;
652            let compiled_default = ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT.default().clone();
653            let ld = get_ld_value(
654                "enable_0dt_deployment_panic_after_timeout",
655                &remote_system_parameters,
656                |x| strconv::parse_bool(x).map_err(|x| x.to_string()),
657            )?;
658            let catalog = openable_adapter_storage
659                .get_enable_0dt_deployment_panic_after_timeout()
660                .await?;
661            let computed = ld.or(catalog).or(cli_default).unwrap_or(compiled_default);
662            info!(
663                %computed,
664                ?ld,
665                ?catalog,
666                ?cli_default,
667                ?compiled_default,
668                "determined value for enable_0dt_deployment_panic_after_timeout system parameter",
669            );
670            computed
671        };
672
673        // Perform preflight checks.
674        //
675        // Preflight checks determine whether to boot in read-only mode or not.
676        let read_only = preflight::preflight_0dt(
677            openable_adapter_storage.as_mut(),
678            config.controller.deploy_generation,
679            config.restarted_in_place,
680        )
681        .await?;
682
683        let bootstrap_args = BootstrapArgs {
684            default_cluster_replica_size: config.bootstrap_default_cluster_replica_size.clone(),
685            default_cluster_replication_factor: config.bootstrap_default_cluster_replication_factor,
686            bootstrap_role: config.bootstrap_role,
687            cluster_replica_size_map: config.cluster_replica_sizes.clone(),
688            force_catalog_migration: config.force_catalog_migration,
689        };
690
691        let (caught_up_trigger, bootstrapped) = if read_only {
692            let (caught_up_trigger, caught_up_receiver) = mz_ore::channel::trigger::channel();
693            let (bootstrapped, bootstrapped_receiver) = tokio::sync::oneshot::channel();
694            let catchup_config = CatchupConfig {
695                boot_ts,
696                environment_id: config.environment_id.clone(),
697                persist_client,
698                deploy_generation: config.controller.deploy_generation,
699                deployment_state: deployment_state.clone(),
700                catalog_metrics: Arc::clone(&config.catalog_config.metrics),
701                caught_up_max_wait: with_0dt_deployment_max_wait,
702                panic_after_timeout: enable_0dt_deployment_panic_after_timeout,
703                bootstrap_args: bootstrap_args.clone(),
704                ddl_check_interval: with_0dt_deployment_ddl_check_interval,
705            };
706            preflight::spawn_catchup(catchup_config, caught_up_receiver, bootstrapped_receiver);
707            (Some(caught_up_trigger), Some(bootstrapped))
708        } else {
709            (None, None)
710        };
711
712        info!(
713            "startup: envd serve: preflight checks complete in {:?}",
714            preflight_checks_start.elapsed()
715        );
716
717        let catalog_open_start = Instant::now();
718        info!("startup: envd serve: durable catalog open beginning");
719
720        // Load the adapter durable storage.
721        let mut adapter_storage = if read_only {
722            // TODO: behavior of migrations when booting in savepoint mode is
723            // not well defined.
724            let adapter_storage = openable_adapter_storage
725                .open_savepoint(boot_ts, &bootstrap_args)
726                .await?;
727            // In read-only mode, we intentionally do not call `set_is_leader`,
728            // because we are by definition not the leader if we are in
729            // read-only mode.
730
731            adapter_storage
732        } else {
733            let adapter_storage = openable_adapter_storage
734                .open(boot_ts, &bootstrap_args)
735                .await?;
736
737            // Once we have successfully opened the adapter storage in
738            // read/write mode, we can announce we are the leader, as we've
739            // fenced out all other environments using the adapter storage.
740            deployment_state.set_is_leader();
741
742            adapter_storage
743        };
744
745        let bootstrapped = match bootstrapped {
746            Some(tx) => Some((tx, preflight::get_user_ids(adapter_storage.as_mut()).await?)),
747            None => None,
748        };
749
750        // Enable Persist compaction if we're not in read only.
751        if !read_only {
752            config.controller.persist_clients.cfg().enable_compaction();
753        }
754
755        info!(
756            "startup: envd serve: durable catalog open complete in {:?}",
757            catalog_open_start.elapsed()
758        );
759
760        let coord_init_start = Instant::now();
761        info!("startup: envd serve: coordinator init beginning");
762
763        if !config
764            .cluster_replica_sizes
765            .0
766            .contains_key(&config.bootstrap_default_cluster_replica_size)
767        {
768            return Err(anyhow!("bootstrap default cluster replica size is unknown").into());
769        }
770        let envd_epoch = adapter_storage.epoch();
771
772        // Initialize storage usage client.
773        let storage_usage_client = StorageUsageClient::open(
774            config
775                .controller
776                .persist_clients
777                .open(config.controller.persist_location.clone())
778                .await
779                .context("opening storage usage client")?,
780        );
781
782        // Initialize adapter.
783        let license_key = config.license_key.clone();
784        let segment_client = config.segment_api_key.map(|api_key| {
785            mz_segment::Client::new(mz_segment::Config {
786                api_key,
787                client_side: config.segment_client_side,
788            })
789        });
790        let connection_limiter = active_connection_counter.clone();
791        let connection_limit_callback = Box::new(move |limit, superuser_reserved| {
792            connection_limiter.update_limit(limit);
793            connection_limiter.update_superuser_reserved(superuser_reserved);
794        });
795
796        let (adapter_handle, adapter_client) = mz_adapter::serve(mz_adapter::Config {
797            connection_context: config.controller.connection_context.clone(),
798            connection_limit_callback,
799            controller_config: config.controller,
800            controller_envd_epoch: envd_epoch,
801            storage: adapter_storage,
802            timestamp_oracle_url: config.timestamp_oracle_url,
803            unsafe_mode: config.unsafe_mode,
804            all_features: config.all_features,
805            build_info: &BUILD_INFO,
806            environment_id: config.environment_id.clone(),
807            metrics_registry: config.metrics_registry.clone(),
808            now: config.now,
809            secrets_controller: config.secrets_controller,
810            cloud_resource_controller: config.cloud_resource_controller,
811            cluster_replica_sizes: config.cluster_replica_sizes,
812            builtin_system_cluster_config: config.bootstrap_builtin_system_cluster_config,
813            builtin_catalog_server_cluster_config: config
814                .bootstrap_builtin_catalog_server_cluster_config,
815            builtin_probe_cluster_config: config.bootstrap_builtin_probe_cluster_config,
816            builtin_support_cluster_config: config.bootstrap_builtin_support_cluster_config,
817            builtin_analytics_cluster_config: config.bootstrap_builtin_analytics_cluster_config,
818            availability_zones: config.availability_zones,
819            system_parameter_defaults: config.system_parameter_defaults,
820            storage_usage_client,
821            storage_usage_collection_interval: config.storage_usage_collection_interval,
822            storage_usage_retention_period: config.storage_usage_retention_period,
823            segment_client: segment_client.clone(),
824            egress_addresses: config.egress_addresses,
825            remote_system_parameters,
826            aws_account_id: config.aws_account_id,
827            aws_privatelink_availability_zones: config.aws_privatelink_availability_zones,
828            webhook_concurrency_limit: webhook_concurrency_limit.clone(),
829            http_host_name: config.http_host_name,
830            tracing_handle: config.tracing_handle,
831            read_only_controllers: read_only,
832            caught_up_trigger,
833            helm_chart_version: config.helm_chart_version.clone(),
834            license_key: config.license_key,
835            external_login_password_mz_system: config.external_login_password_mz_system,
836            force_builtin_schema_migration: config.force_builtin_schema_migration,
837        })
838        .instrument(info_span!("adapter::serve"))
839        .await?;
840
841        if let Some((tx, initial_ids)) = bootstrapped {
842            // The receiver is gone if an administrator already skipped catch-up.
843            let _ = tx.send(initial_ids);
844        }
845
846        // Initialize the OIDC authenticator, shared between the HTTP and SQL servers.
847        let oidc = GenericOidcAuthenticator::new(adapter_client.clone());
848
849        info!(
850            "startup: envd serve: coordinator init complete in {:?}",
851            coord_init_start.elapsed()
852        );
853
854        let serve_postamble_start = Instant::now();
855        info!("startup: envd serve: postamble beginning");
856
857        // Send adapter client and OIDC authenticator to the HTTP servers.
858        authenticator_oidc_tx
859            .send(oidc.clone())
860            .expect("rx known to be live");
861        adapter_client_tx
862            .send(adapter_client.clone())
863            .expect("internal HTTP server should not drop first");
864
865        let metrics = mz_pgwire::MetricsConfig::register_into(&config.metrics_registry);
866
867        // Launch SQL server.
868        let mut sql_listener_handles = BTreeMap::new();
869        for (name, listener) in self.sql {
870            sql_listener_handles.insert(
871                name.clone(),
872                listener
873                    .serve_sql(
874                        name,
875                        active_connection_counter.clone(),
876                        tls_reloading_context.clone(),
877                        config.frontegg.clone(),
878                        adapter_client.clone(),
879                        oidc.clone(),
880                        metrics.clone(),
881                        config.helm_chart_version.clone(),
882                    )
883                    .await,
884            );
885        }
886
887        // Start telemetry reporting loop.
888        if let Some(segment_client) = segment_client {
889            telemetry::start_reporting(telemetry::Config {
890                segment_client,
891                adapter_client: adapter_client.clone(),
892                environment_id: config.environment_id,
893                license_key: license_key.clone(),
894                helm_chart_version: config.helm_chart_version.clone(),
895                report_interval: Duration::from_secs(3600),
896            });
897        } else if config.test_only_dummy_segment_client {
898            // We only have access to a segment client in production but we
899            // still want to exercise the telemetry reporting code to a degree.
900            // So we create a dummy client and report telemetry into the void.
901            // This way we at least run the telemetry queries the way a
902            // production environment would.
903            tracing::debug!("starting telemetry reporting with a dummy segment client");
904            let segment_client = mz_segment::Client::new_dummy_client();
905            telemetry::start_reporting(telemetry::Config {
906                segment_client,
907                adapter_client: adapter_client.clone(),
908                environment_id: config.environment_id,
909                license_key: license_key.clone(),
910                helm_chart_version: config.helm_chart_version.clone(),
911                report_interval: Duration::from_secs(180),
912            });
913        }
914
915        // If system_parameter_sync_config and config_sync_loop_interval are present,
916        // start the system_parameter_sync loop.
917        if let Some(system_parameter_sync_config) = system_parameter_sync_config {
918            task::spawn(
919                || "system_parameter_sync",
920                AssertUnwindSafe(system_parameter_sync(
921                    system_parameter_sync_config,
922                    boot_frontend,
923                    adapter_client.clone(),
924                    config.config_sync_loop_interval,
925                ))
926                .ore_catch_unwind(),
927            );
928        }
929
930        info!(
931            "startup: envd serve: postamble complete in {:?}",
932            serve_postamble_start.elapsed()
933        );
934        info!(
935            "startup: envd serve: complete in {:?}",
936            serve_start.elapsed()
937        );
938
939        Ok(Server {
940            sql_listener_handles,
941            http_listener_handles,
942            #[cfg(feature = "test")]
943            adapter_client,
944            _adapter_handle: adapter_handle,
945        })
946    }
947}
948
949fn get_ld_value<V>(
950    name: &str,
951    remote_system_parameters: &Option<BTreeMap<String, String>>,
952    parse: impl Fn(&str) -> Result<V, String>,
953) -> Result<Option<V>, anyhow::Error> {
954    remote_system_parameters
955        .as_ref()
956        .and_then(|params| params.get(name))
957        .map(|x| {
958            parse(x).map_err(|err| anyhow!("failed to parse remote value for {}: {}", name, err))
959        })
960        .transpose()
961}
962
963/// A running `environmentd` server.
964pub struct Server {
965    // Drop order matters for these fields.
966    pub sql_listener_handles: BTreeMap<String, ListenerHandle>,
967    pub http_listener_handles: BTreeMap<String, ListenerHandle>,
968    #[cfg(feature = "test")]
969    adapter_client: AdapterClient,
970    _adapter_handle: mz_adapter::Handle,
971}
972
973impl Server {
974    /// A client for the adapter, letting tests drive the adapter API directly.
975    #[cfg(feature = "test")]
976    pub fn adapter_client(&self) -> &AdapterClient {
977        &self.adapter_client
978    }
979}
mz_environmentd/lib.rs

mz_environmentd/
lib.rs
