mz_dyncfg_launchdarkly/
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//! A dyncfg::ConfigSet backed by LaunchDarkly.
11
12use std::time::Duration;
13
14use launchdarkly_sdk_transport::HttpTransport;
15use launchdarkly_server_sdk as ld;
16use mz_build_info::BuildInfo;
17use mz_dyncfg::{ConfigSet, ConfigUpdates, ConfigVal};
18use mz_ore::cast::CastLossy;
19use mz_ore::task;
20use tokio::time;
21
22/// Start a background task that syncs to a ConfigSet from LaunchDarkly. `ctx_builder` can be used
23/// to add additional LD contexts. A `build` context is added automatically. Returns `Ok` after the
24/// LD client has been initialized and an initial sync completed. If the initialization takes longer
25/// than `config_sync_timeout`, an error is returned.
26///
27/// A successful initialization can take up to `config_sync_timeout`, preventing the calling service
28/// from starting for possibly up to that duration. Its value should be chosen based on the needs of
29/// the service in the case that LaunchDarkly is down.
30///
31/// If the caller chooses to continue if this function returns an error, the ConfigSet will retain
32/// its default values. Those should be chosen with this risk in mind.
33pub async fn sync_launchdarkly_to_configset<F>(
34    set: ConfigSet,
35    build_info: &'static BuildInfo,
36    ctx_builder: F,
37    // Use an option so that local dev where this is disabled still uses the same validation logic
38    // for the ConfigSet.
39    launchdarkly_sdk_key: Option<&str>,
40    config_sync_timeout: Duration,
41    config_sync_loop_interval: Option<Duration>,
42    on_update: impl Fn(&ConfigUpdates, &ConfigSet) + Send + 'static,
43) -> Result<(), anyhow::Error>
44where
45    F: FnOnce(&mut ld::MultiContextBuilder) -> Result<(), anyhow::Error>,
46{
47    // Ensure that all the ConfigVals in the set support FlagValue conversion, even if LD is
48    // disabled (preventing error skew in local vs prod settings).
49    for entry in set.entries() {
50        let _ = dyn_into_flag(entry.val())?;
51    }
52    let ld_client = if let Some(key) = launchdarkly_sdk_key {
53        // The 300s streaming read timeout must stay above LaunchDarkly's
54        // streaming heartbeat interval (roughly 3 minutes per LD's
55        // documentation), or a healthy idle stream would trip the timeout and
56        // reconnect spuriously. The same constant lives in the adapter's
57        // `SystemParameterFrontend`.
58        //
59        // NOTE: `HyperTransport` auto-detects the `HTTP_PROXY`/`HTTPS_PROXY`/
60        // `NO_PROXY` env vars and routes through a configured proxy. No
61        // exposure today (balancerd's cloud pods set no proxy vars), but worth
62        // knowing if proxy vars ever appear on a pod.
63        let transport = https_transport(Duration::from_secs(10), Duration::from_secs(300))
64            .expect("failed to create HTTPS transport");
65
66        let mut data_source = ld::StreamingDataSourceBuilder::new();
67        data_source.transport(transport.clone());
68
69        let mut event_processor = ld::EventProcessorBuilder::new();
70        event_processor.transport(transport);
71
72        let config = ld::ConfigBuilder::new(key)
73            .data_source(&data_source)
74            .event_processor(&event_processor)
75            .build()
76            .expect("valid config");
77        let client = ld::Client::build(config)?;
78        client.start_with_default_executor();
79        let init = async {
80            let max_backoff = Duration::from_secs(60);
81            let mut backoff = Duration::from_secs(5);
82
83            // TODO(materialize#38055): fix retry logic
84            loop {
85                match client.wait_for_initialization(config_sync_timeout).await {
86                    Some(true) => break,
87                    Some(false) => tracing::warn!("SyncedConfigSet failed to initialize"),
88                    None => {}
89                }
90
91                tokio::time::sleep(backoff).await;
92                backoff = (backoff * 2).min(max_backoff);
93            }
94        };
95        if tokio::time::timeout(config_sync_timeout, init)
96            .await
97            .is_err()
98        {
99            tracing::info!("SyncedConfigSet initialize on boot: initialize has timed out");
100        }
101        Some(client)
102    } else {
103        None
104    };
105
106    let synced = SyncedConfigSet {
107        set,
108        ld_client,
109        ld_ctx: ld_ctx(build_info, ctx_builder)?,
110        on_update,
111    };
112    synced.sync()?;
113    task::spawn(
114        || "SyncedConfigSet sync_loop",
115        synced.sync_loop(config_sync_loop_interval),
116    );
117    Ok(())
118}
119
120/// Builds a LaunchDarkly [`HttpTransport`] whose TLS runs on the aws-lc-rs rustls provider,
121/// trusting the platform's native root certificates.
122///
123/// Use this instead of `HyperTransport::builder().build_https()`, which hardcodes the `ring`
124/// provider. Like `build_https`, the connector also accepts plain `http` URIs, which a relay
125/// proxy or test mock may use.
126///
127/// NOTE: `HyperTransport` still honors `HTTP_PROXY`/`HTTPS_PROXY`/`NO_PROXY`, but only for
128/// proxies reached over plain HTTP. TLS to the proxy itself requires a transport crate feature
129/// that would pull `ring` back in.
130pub fn https_transport(
131    connect_timeout: Duration,
132    read_timeout: Duration,
133) -> Result<impl HttpTransport, std::io::Error> {
134    let connector = hyper_rustls::HttpsConnectorBuilder::new()
135        .with_provider_and_native_roots(rustls::crypto::aws_lc_rs::default_provider())?
136        .https_or_http()
137        .enable_http1()
138        .enable_http2()
139        .build();
140    launchdarkly_sdk_transport::HyperTransport::builder()
141        .connect_timeout(connect_timeout)
142        .read_timeout(read_timeout)
143        .build_with_connector(connector)
144}
145
146fn ld_ctx<F>(build_info: &'static BuildInfo, ctx_builder: F) -> Result<ld::Context, anyhow::Error>
147where
148    F: FnOnce(&mut ld::MultiContextBuilder) -> Result<(), anyhow::Error>,
149{
150    // Register multiple contexts for this client.
151    //
152    // Unfortunately, it seems that the order in which conflicting targeting
153    // rules are applied depends on the definition order of feature flag
154    // variations rather than on the order in which context are registered with
155    // the multi-context builder.
156    let mut builder = ld::MultiContextBuilder::new();
157
158    builder.add_context(
159        ld::ContextBuilder::new(build_info.sha)
160            .kind("build")
161            .set_string("semver_version", build_info.semver_version().to_string())
162            .build()
163            .map_err(|e| anyhow::anyhow!(e))?,
164    );
165
166    ctx_builder(&mut builder)?;
167
168    builder.build().map_err(|e| anyhow::anyhow!(e))
169}
170
171struct SyncedConfigSet<F>
172where
173    F: Fn(&ConfigUpdates, &ConfigSet) + Send,
174{
175    set: ConfigSet,
176    ld_client: Option<ld::Client>,
177    ld_ctx: ld::Context,
178    on_update: F,
179}
180
181impl<F: Fn(&ConfigUpdates, &ConfigSet) + Send> SyncedConfigSet<F> {
182    /// Returns a future that periodically polls LaunchDarkly and updates the ConfigSet.
183    async fn sync_loop(self, tick_interval: Option<Duration>) {
184        let Some(tick_interval) = tick_interval else {
185            tracing::info!("skipping SyncedConfigSet sync as tick_interval = None");
186            return;
187        };
188
189        let mut interval = time::interval(tick_interval);
190        interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
191
192        tracing::info!(
193            "synchronizing SyncedConfigSet values every {} seconds",
194            tick_interval.as_secs()
195        );
196
197        loop {
198            interval.tick().await;
199
200            if let Err(err) = self.sync() {
201                tracing::info!("SyncedConfigSet: {err}");
202            }
203        }
204    }
205
206    /// Reads current values from LaunchDarkly and updates the ConfigSet.
207    fn sync(&self) -> Result<(), anyhow::Error> {
208        let mut updates = ConfigUpdates::default();
209        let Some(ld_client) = &self.ld_client else {
210            (self.on_update)(&updates, &self.set);
211            return Ok(());
212        };
213        for entry in self.set.entries() {
214            let val = dyn_into_flag(entry.val()).expect("new() verifies all configs can convert");
215            let flag_var = ld_client.variation(&self.ld_ctx, entry.name(), val);
216            let update = match (entry.val(), flag_var) {
217                (ConfigVal::Bool(_), ld::FlagValue::Bool(flag)) => ConfigVal::Bool(flag),
218                (ConfigVal::U32(_), ld::FlagValue::Number(flag)) => {
219                    ConfigVal::U32(u32::cast_lossy(flag))
220                }
221                (ConfigVal::Usize(_), ld::FlagValue::Number(flag)) => {
222                    ConfigVal::Usize(usize::cast_lossy(flag))
223                }
224                (ConfigVal::F64(_), ld::FlagValue::Number(flag)) => ConfigVal::F64(flag),
225                (ConfigVal::String(_), ld::FlagValue::Str(flag)) => ConfigVal::String(flag),
226                (ConfigVal::Duration(_), ld::FlagValue::Str(flag)) => {
227                    match humantime::parse_duration(&flag) {
228                        Ok(d) => ConfigVal::Duration(d),
229                        Err(e) => {
230                            tracing::warn!("failed to parse Duration for {}: {}", entry.name(), e);
231                            continue;
232                        }
233                    }
234                }
235                (ConfigVal::Json(_), ld::FlagValue::Json(flag)) => ConfigVal::Json(flag),
236
237                // Hardcode all others so that if ConfigVal gets new types this match block will
238                // compile error.
239                (ConfigVal::Bool(_), _)
240                | (ConfigVal::U32(_), _)
241                | (ConfigVal::Usize(_), _)
242                | (ConfigVal::F64(_), _)
243                | (ConfigVal::Duration(_), _)
244                | (ConfigVal::Json(_), _)
245                | (ConfigVal::OptUsize(_), _)
246                | (ConfigVal::String(_), _)
247                | (ConfigVal::OptString(_), _) => {
248                    tracing::warn!("LD flag type mismatch for {}", entry.name());
249                    continue;
250                }
251            };
252            tracing::debug!(
253                "updating config value {} from {:?} to {:?}",
254                &entry.name(),
255                &entry.val(),
256                update
257            );
258            updates.add_dynamic(entry.name(), update);
259        }
260        updates.apply(&self.set);
261        (self.on_update)(&updates, &self.set);
262        Ok(())
263    }
264}
265
266/// Converts a dyncfg ConfigVal into a LaunchDarkly FlagValue. Returns an error if the ConfigVal
267/// type isn't supported by the FlagValue format.
268fn dyn_into_flag(val: ConfigVal) -> Result<ld::FlagValue, anyhow::Error> {
269    // Note that errors must only (and always) occur when the ConfigVal type isn't fully supported.
270    // That is, don't error only if the current value isn't supported (like None in an Opt type):
271    // error always for an Opt value because it might be None.
272    Ok(match val {
273        ConfigVal::Bool(v) => ld::FlagValue::Bool(v),
274        ConfigVal::U32(v) => ld::FlagValue::Number(v.into()),
275        ConfigVal::Usize(v) => ld::FlagValue::Number(f64::cast_lossy(v)),
276        ConfigVal::OptUsize(_) => anyhow::bail!("OptUsize None cannot be converted to a FlagValue"),
277        ConfigVal::F64(v) => ld::FlagValue::Number(v),
278        ConfigVal::String(v) => ld::FlagValue::Str(v),
279        ConfigVal::OptString(_) => {
280            anyhow::bail!("OptString None cannot be converted to a FlagValue")
281        }
282        ConfigVal::Duration(v) => ld::FlagValue::Str(humantime::format_duration(v).to_string()),
283        ConfigVal::Json(v) => ld::FlagValue::Json(v),
284    })
285}
mz_dyncfg_launchdarkly/lib.rs

mz_dyncfg_launchdarkly/
lib.rs
