mz_adapter/
flags.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
10use std::time::Duration;
11
12use mz_adapter_types::dyncfgs::PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT;
13use mz_compute_client::protocol::command::ComputeParameters;
14use mz_orchestrator::scheduling_config::{ServiceSchedulingConfig, ServiceTopologySpreadConfig};
15use mz_ore::cast::CastFrom;
16use mz_ore::error::ErrorExt;
17use mz_service::params::GrpcClientParameters;
18use mz_sql::session::vars::SystemVars;
19use mz_storage_types::parameters::{
20    PgSourceSnapshotConfig, StorageMaxInflightBytesConfig, StorageParameters,
21};
22use mz_tracing::params::TracingParameters;
23
24use mz_timestamp_oracle::postgres_oracle::TimestampOracleParameters;
25
26/// Return the current compute configuration, derived from the system configuration.
27pub fn compute_config(config: &SystemVars) -> ComputeParameters {
28    ComputeParameters {
29        workload_class: None,
30        max_result_size: Some(config.max_result_size()),
31        tracing: tracing_config(config),
32        grpc_client: grpc_client_config(config),
33        dyncfg_updates: config.dyncfg_updates(),
34    }
35}
36
37/// Return the current storage configuration, derived from the system configuration.
38pub fn storage_config(config: &SystemVars) -> StorageParameters {
39    StorageParameters {
40        pg_source_connect_timeout: Some(config.pg_source_connect_timeout()),
41        pg_source_tcp_keepalives_retries: Some(config.pg_source_tcp_keepalives_retries()),
42        pg_source_tcp_keepalives_idle: Some(config.pg_source_tcp_keepalives_idle()),
43        pg_source_tcp_keepalives_interval: Some(config.pg_source_tcp_keepalives_interval()),
44        pg_source_tcp_user_timeout: Some(config.pg_source_tcp_user_timeout()),
45        pg_source_tcp_configure_server: config.pg_source_tcp_configure_server(),
46        pg_source_snapshot_statement_timeout: config.pg_source_snapshot_statement_timeout(),
47        pg_source_wal_sender_timeout: config.pg_source_wal_sender_timeout(),
48        mysql_source_timeouts: mz_mysql_util::TimeoutConfig::build(
49            config.mysql_source_snapshot_max_execution_time(),
50            config.mysql_source_snapshot_lock_wait_timeout(),
51            config.mysql_source_tcp_keepalive(),
52            config.mysql_source_connect_timeout(),
53        ),
54        keep_n_source_status_history_entries: config.keep_n_source_status_history_entries(),
55        keep_n_sink_status_history_entries: config.keep_n_sink_status_history_entries(),
56        keep_n_privatelink_status_history_entries: config
57            .keep_n_privatelink_status_history_entries(),
58        replica_status_history_retention_window: config.replica_status_history_retention_window(),
59        upsert_rocksdb_tuning_config: {
60            match mz_rocksdb_types::RocksDBTuningParameters::from_parameters(
61                config.upsert_rocksdb_compaction_style(),
62                config.upsert_rocksdb_optimize_compaction_memtable_budget(),
63                config.upsert_rocksdb_level_compaction_dynamic_level_bytes(),
64                config.upsert_rocksdb_universal_compaction_ratio(),
65                config.upsert_rocksdb_parallelism(),
66                config.upsert_rocksdb_compression_type(),
67                config.upsert_rocksdb_bottommost_compression_type(),
68                config.upsert_rocksdb_batch_size(),
69                config.upsert_rocksdb_retry_duration(),
70                config.upsert_rocksdb_stats_log_interval_seconds(),
71                config.upsert_rocksdb_stats_persist_interval_seconds(),
72                config.upsert_rocksdb_point_lookup_block_cache_size_mb(),
73                config.upsert_rocksdb_shrink_allocated_buffers_by_ratio(),
74                config.upsert_rocksdb_write_buffer_manager_memory_bytes(),
75                config
76                    .upsert_rocksdb_write_buffer_manager_cluster_memory_fraction()
77                    .and_then(|d| match d.try_into() {
78                        Err(e) => {
79                            tracing::error!(
80                                "Couldn't convert upsert_rocksdb_write_buffer_manager_cluster_memory_fraction {:?} to f64, so defaulting to `None`: {e:?}",
81                                config
82                                    .upsert_rocksdb_write_buffer_manager_cluster_memory_fraction()
83                            );
84                            None
85                        }
86                        Ok(o) => Some(o),
87                    }),
88                config.upsert_rocksdb_write_buffer_manager_allow_stall(),
89            ) {
90                Ok(u) => u,
91                Err(e) => {
92                    tracing::warn!(
93                        "Failed to deserialize upsert_rocksdb parameters \
94                            into a `RocksDBTuningParameters`, \
95                            failing back to reasonable defaults: {}",
96                        e.display_with_causes()
97                    );
98                    mz_rocksdb_types::RocksDBTuningParameters::default()
99                }
100            }
101        },
102        finalize_shards: config.enable_storage_shard_finalization(),
103        tracing: tracing_config(config),
104        storage_dataflow_max_inflight_bytes_config: StorageMaxInflightBytesConfig {
105            max_inflight_bytes_default: config.storage_dataflow_max_inflight_bytes(),
106            // Interpret the `Numeric` as a float here, we don't need perfect
107            // precision for a percentage. Unfortunately `Decimal` makes us handle errors.
108            max_inflight_bytes_cluster_size_fraction: config
109                .storage_dataflow_max_inflight_bytes_to_cluster_size_fraction()
110                .and_then(|d| match d.try_into() {
111                    Err(e) => {
112                        tracing::error!(
113                            "Couldn't convert {:?} to f64, so defaulting to `None`: {e:?}",
114                            config.storage_dataflow_max_inflight_bytes_to_cluster_size_fraction()
115                        );
116                        None
117                    }
118                    Ok(o) => Some(o),
119                }),
120            disk_only: config.storage_dataflow_max_inflight_bytes_disk_only(),
121        },
122        grpc_client: grpc_client_config(config),
123        shrink_upsert_unused_buffers_by_ratio: config
124            .storage_shrink_upsert_unused_buffers_by_ratio(),
125        record_namespaced_errors: config.storage_record_source_sink_namespaced_errors(),
126        ssh_timeout_config: mz_ssh_util::tunnel::SshTimeoutConfig {
127            check_interval: config.ssh_check_interval(),
128            connect_timeout: config.ssh_connect_timeout(),
129            keepalives_idle: config.ssh_keepalives_idle(),
130        },
131        kafka_timeout_config: mz_kafka_util::client::TimeoutConfig::build(
132            config.kafka_socket_keepalive(),
133            config.kafka_socket_timeout(),
134            config.kafka_transaction_timeout(),
135            config.kafka_socket_connection_setup_timeout(),
136            config.kafka_fetch_metadata_timeout(),
137            config.kafka_progress_record_fetch_timeout(),
138        ),
139        statistics_interval: config.storage_statistics_interval(),
140        statistics_collection_interval: config.storage_statistics_collection_interval(),
141        pg_snapshot_config: PgSourceSnapshotConfig {
142            collect_strict_count: config.pg_source_snapshot_collect_strict_count(),
143        },
144        user_storage_managed_collections_batch_duration: config
145            .user_storage_managed_collections_batch_duration(),
146        dyncfg_updates: config.dyncfg_updates(),
147    }
148}
149
150pub fn tracing_config(config: &SystemVars) -> TracingParameters {
151    TracingParameters {
152        log_filter: Some(config.logging_filter()),
153        opentelemetry_filter: Some(config.opentelemetry_filter()),
154        log_filter_defaults: config.logging_filter_defaults(),
155        opentelemetry_filter_defaults: config.opentelemetry_filter_defaults(),
156        sentry_filters: config.sentry_filters(),
157    }
158}
159
160pub fn caching_config(config: &SystemVars) -> mz_secrets::CachingPolicy {
161    let ttl_secs = config.webhooks_secrets_caching_ttl_secs();
162    mz_secrets::CachingPolicy {
163        enabled: ttl_secs > 0,
164        ttl: Duration::from_secs(u64::cast_from(ttl_secs)),
165    }
166}
167
168pub fn timestamp_oracle_config(config: &SystemVars) -> TimestampOracleParameters {
169    TimestampOracleParameters {
170        pg_connection_pool_max_size: Some(config.pg_timestamp_oracle_connection_pool_max_size()),
171        pg_connection_pool_max_wait: Some(config.pg_timestamp_oracle_connection_pool_max_wait()),
172        pg_connection_pool_ttl: Some(config.pg_timestamp_oracle_connection_pool_ttl()),
173        pg_connection_pool_ttl_stagger: Some(
174            config.pg_timestamp_oracle_connection_pool_ttl_stagger(),
175        ),
176        // We use a shared set of crdb flags for the basics, but the above flags
177        // for the connection pool are specific to the postgres/crdb timestamp
178        // oracle.
179        pg_connection_pool_connect_timeout: Some(config.crdb_connect_timeout()),
180        pg_connection_pool_tcp_user_timeout: Some(config.crdb_tcp_user_timeout()),
181        pg_connection_pool_keepalives_idle: Some(config.crdb_keepalives_idle()),
182        pg_connection_pool_keepalives_interval: Some(config.crdb_keepalives_interval()),
183        pg_connection_pool_keepalives_retries: Some(config.crdb_keepalives_retries()),
184        pg_statement_timeout: Some(PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT.get(config.dyncfgs())),
185    }
186}
187
188fn grpc_client_config(config: &SystemVars) -> GrpcClientParameters {
189    GrpcClientParameters {
190        connect_timeout: Some(config.grpc_connect_timeout()),
191        http2_keep_alive_interval: Some(config.grpc_client_http2_keep_alive_interval()),
192        http2_keep_alive_timeout: Some(config.grpc_client_http2_keep_alive_timeout()),
193    }
194}
195
196pub fn orchestrator_scheduling_config(config: &SystemVars) -> ServiceSchedulingConfig {
197    ServiceSchedulingConfig {
198        multi_pod_az_affinity_weight: config.cluster_multi_process_replica_az_affinity_weight(),
199        soften_replication_anti_affinity: config.cluster_soften_replication_anti_affinity(),
200        soften_replication_anti_affinity_weight: config
201            .cluster_soften_replication_anti_affinity_weight(),
202        topology_spread: ServiceTopologySpreadConfig {
203            enabled: config.cluster_enable_topology_spread(),
204            ignore_non_singular_scale: config.cluster_topology_spread_ignore_non_singular_scale(),
205            max_skew: config.cluster_topology_spread_max_skew(),
206            min_domains: config.cluster_topology_spread_set_min_domains(),
207            soft: config.cluster_topology_spread_soft(),
208        },
209        soften_az_affinity: config.cluster_soften_az_affinity(),
210        soften_az_affinity_weight: config.cluster_soften_az_affinity_weight(),
211        security_context_enabled: config.cluster_security_context_enabled(),
212    }
213}
mz_adapter/flags.rs

mz_adapter/
flags.rs
