mz_adapter/
config.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::collections::{BTreeMap, BTreeSet};
11use std::path::PathBuf;
12
13use mz_build_info::BuildInfo;
14use mz_cluster_client::ReplicaId;
15use mz_controller_types::ClusterId;
16use mz_ore::metric;
17use mz_ore::metrics::{MetricsRegistry, UIntGauge};
18use mz_ore::now::NowFn;
19use mz_sql::catalog::EnvironmentId;
20use prometheus::IntCounter;
21
22mod backend;
23mod frontend;
24mod params;
25mod sync;
26
27pub use backend::SystemParameterBackend;
28pub use frontend::{
29    ClusterEvalContext, ClusterScopeContext, ReplicaEvalContext, ReplicaScopeContext,
30    SystemParameterFrontend,
31};
32pub use params::{ModifiedParameter, SynchronizedParameters};
33pub use sync::system_parameter_sync;
34
35/// Scoped (per-cluster and per-replica) system-parameter overrides, keyed by
36/// object id. Each value is the raw (unparsed) string for a parameter whose
37/// scoped value differs from the environment-wide value; an absent entry means
38/// no override. Empty maps mean no scoped overrides at all.
39///
40/// This is the in-memory mirror of the durable `cluster_system_configurations`
41/// and `replica_system_configurations` catalog collections.
42#[derive(Clone, Debug, Default, PartialEq, Eq)]
43pub struct ScopedParameters {
44    /// Cluster-coherent overrides, keyed by cluster id.
45    pub cluster: BTreeMap<ClusterId, BTreeMap<String, String>>,
46    /// Replica-local overrides, keyed by replica id.
47    pub replica: BTreeMap<ReplicaId, BTreeMap<String, String>>,
48}
49
50/// The set of objects a [`ScopedParameters`] update was evaluated for, used to
51/// bound which durable override rows the update may prune.
52///
53/// The update is authoritative only for objects in this set. The durable apply
54/// removes a row only when its owning object is in scope and the update no
55/// longer carries that override, so an object created after the update's
56/// evaluation snapshot, and the override it folded into its own create
57/// transaction, is not wiped by a concurrent full-state reconcile.
58#[derive(Clone, Debug, Default, PartialEq, Eq)]
59pub struct ScopedParametersScope {
60    /// Cluster ids whose rows the update may prune.
61    pub clusters: BTreeSet<ClusterId>,
62    /// Replica ids whose rows the update may prune.
63    pub replicas: BTreeSet<ReplicaId>,
64}
65
66impl ScopedParameters {
67    /// Returns `true` if there are no cluster or replica overrides.
68    pub fn is_empty(&self) -> bool {
69        self.cluster.is_empty() && self.replica.is_empty()
70    }
71
72    /// Returns a copy of `self` with `other`'s entries merged in, replacing any
73    /// existing entry for the same object. Expresses no removals.
74    pub fn merge(&self, other: &ScopedParameters) -> ScopedParameters {
75        let mut merged = self.clone();
76        merged
77            .cluster
78            .extend(other.cluster.iter().map(|(id, v)| (*id, v.clone())));
79        merged
80            .replica
81            .extend(other.replica.iter().map(|(id, v)| (*id, v.clone())));
82        merged
83    }
84}
85
86/// A factory for [SystemParameterFrontend] instances.
87#[derive(Clone, Debug)]
88pub struct SystemParameterSyncConfig {
89    /// The environment ID that should identify connected clients.
90    env_id: EnvironmentId,
91    /// Build info for the environment running this.
92    build_info: &'static BuildInfo,
93    /// Parameter sync metrics.
94    metrics: Metrics,
95    ///  /// A map from parameter names to LaunchDarkly feature keys
96    /// to use when populating the [SynchronizedParameters]
97    /// instance in [SystemParameterFrontend::pull].
98    key_map: BTreeMap<String, String>,
99    /// Configuration for the parameter backend that we're syncing with.
100    backend_config: SystemParameterSyncClientConfig,
101}
102
103#[derive(Clone, Debug)]
104pub enum SystemParameterSyncClientConfig {
105    File {
106        // Path to a JSON config file that contains system parameters.
107        path: PathBuf,
108    },
109    LaunchDarkly {
110        /// The LaunchDarkly SDK key
111        sdk_key: String,
112        /// Overrides the LaunchDarkly streaming, polling, and events endpoints
113        /// with a single base URL (as for a relay proxy). `None` uses
114        /// LaunchDarkly's default endpoints. Primarily for pointing the SDK at
115        /// a mock server in tests.
116        base_uri: Option<String>,
117        /// Function to return the current time.
118        now_fn: NowFn,
119    },
120}
121
122impl SystemParameterSyncClientConfig {
123    fn is_launch_darkly(&self) -> bool {
124        match &self {
125            Self::LaunchDarkly { .. } => true,
126            Self::File { .. } => false,
127        }
128    }
129}
130
131impl SystemParameterSyncConfig {
132    /// Construct a new [SystemParameterFrontend] instance.
133    pub fn new(
134        env_id: EnvironmentId,
135        build_info: &'static BuildInfo,
136        registry: &MetricsRegistry,
137        key_map: BTreeMap<String, String>,
138        backend_config: SystemParameterSyncClientConfig,
139    ) -> Self {
140        Self {
141            env_id,
142            build_info,
143            metrics: Metrics::register_into(registry),
144            key_map,
145            backend_config,
146        }
147    }
148}
149
150#[derive(Debug, Clone)]
151pub(super) struct Metrics {
152    pub last_cse_time_seconds: UIntGauge,
153    pub last_sse_time_seconds: UIntGauge,
154    pub params_changed: IntCounter,
155}
156
157impl Metrics {
158    pub(super) fn register_into(registry: &MetricsRegistry) -> Self {
159        Self {
160            last_cse_time_seconds: registry.register(metric!(
161                name: "mz_parameter_frontend_last_cse_time_seconds",
162                help: "The last known time when the LaunchDarkly client sent an event to the LaunchDarkly server (as unix timestamp).",
163            )),
164            last_sse_time_seconds: registry.register(metric!(
165                name: "mz_parameter_frontend_last_sse_time_seconds",
166                help: "The last known time when the LaunchDarkly client received an event from the LaunchDarkly server (as unix timestamp).",
167            )),
168            params_changed: registry.register(metric!(
169                name: "mz_parameter_frontend_params_changed",
170                help: "The number of parameter changes pulled from the LaunchDarkly frontend.",
171            )),
172        }
173    }
174}
175
176#[cfg(test)]
177mod tests {
178    use std::collections::BTreeMap;
179
180    use mz_cluster_client::ReplicaId;
181    use mz_controller_types::ClusterId;
182
183    use super::ScopedParameters;
184
185    fn cfg(name: &str, value: &str) -> BTreeMap<String, String> {
186        BTreeMap::from([(name.to_string(), value.to_string())])
187    }
188
189    #[mz_ore::test]
190    fn test_scoped_parameters_is_empty() {
191        assert!(ScopedParameters::default().is_empty());
192
193        let mut params = ScopedParameters::default();
194        params.cluster.insert(ClusterId::User(1), cfg("f", "true"));
195        assert!(!params.is_empty());
196
197        let mut params = ScopedParameters::default();
198        params.replica.insert(ReplicaId::User(1), cfg("f", "true"));
199        assert!(!params.is_empty());
200    }
201
202    #[mz_ore::test]
203    fn test_scoped_parameters_merge() {
204        let mut base = ScopedParameters::default();
205        base.cluster.insert(ClusterId::User(1), cfg("f", "old"));
206        base.cluster.insert(ClusterId::User(2), cfg("f", "keep"));
207        base.replica.insert(ReplicaId::User(1), cfg("g", "old"));
208
209        let mut incoming = ScopedParameters::default();
210        // Overrides the existing entry for the same object...
211        incoming.cluster.insert(ClusterId::User(1), cfg("f", "new"));
212        // ...and adds a new object, leaving others untouched.
213        incoming.replica.insert(ReplicaId::User(2), cfg("g", "new"));
214
215        let merged = base.merge(&incoming);
216
217        // Replaced.
218        assert_eq!(merged.cluster[&ClusterId::User(1)], cfg("f", "new"));
219        // Untouched object retained (merge does not express removals).
220        assert_eq!(merged.cluster[&ClusterId::User(2)], cfg("f", "keep"));
221        // Pre-existing replica retained, new replica added.
222        assert_eq!(merged.replica[&ReplicaId::User(1)], cfg("g", "old"));
223        assert_eq!(merged.replica[&ReplicaId::User(2)], cfg("g", "new"));
224
225        // The original is unchanged (merge returns a copy).
226        assert_eq!(base.cluster[&ClusterId::User(1)], cfg("f", "old"));
227    }
228}
mz_adapter/config.rs

mz_adapter/
config.rs
