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

mz_dyncfg_launchdarkly/
lib.rs
