mz_dyncfg_launchdarkly/
lib.rs1use 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
22pub async fn sync_launchdarkly_to_configset<F>(
34 set: ConfigSet,
35 build_info: &'static BuildInfo,
36 ctx_builder: F,
37 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 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 let transport = https_transport(Duration::from_secs(10), Duration::from_secs(300))
59 .expect("failed to create HTTPS transport");
60
61 let mut data_source = ld::StreamingDataSourceBuilder::new();
62 data_source.transport(transport.clone());
63
64 let mut event_processor = ld::EventProcessorBuilder::new();
65 event_processor.transport(transport);
66
67 let config = ld::ConfigBuilder::new(key)
68 .data_source(&data_source)
69 .event_processor(&event_processor)
70 .build()
71 .expect("valid config");
72 let client = ld::Client::build(config)?;
73 client.start_with_default_executor();
74 let init = async {
75 let max_backoff = Duration::from_secs(60);
76 let mut backoff = Duration::from_secs(5);
77
78 loop {
80 match client.wait_for_initialization(config_sync_timeout).await {
81 Some(true) => break,
82 Some(false) => tracing::warn!("SyncedConfigSet failed to initialize"),
83 None => {}
84 }
85
86 tokio::time::sleep(backoff).await;
87 backoff = (backoff * 2).min(max_backoff);
88 }
89 };
90 if tokio::time::timeout(config_sync_timeout, init)
91 .await
92 .is_err()
93 {
94 tracing::info!("SyncedConfigSet initialize on boot: initialize has timed out");
95 }
96 Some(client)
97 } else {
98 None
99 };
100
101 let synced = SyncedConfigSet {
102 set,
103 ld_client,
104 ld_ctx: ld_ctx(build_info, ctx_builder)?,
105 on_update,
106 };
107 synced.sync()?;
108 task::spawn(
109 || "SyncedConfigSet sync_loop",
110 synced.sync_loop(config_sync_loop_interval),
111 );
112 Ok(())
113}
114
115pub fn https_transport(
124 connect_timeout: Duration,
125 read_timeout: Duration,
126) -> Result<impl HttpTransport, std::io::Error> {
127 let connector = hyper_rustls::HttpsConnectorBuilder::new()
128 .with_provider_and_native_roots(rustls::crypto::aws_lc_rs::default_provider())?
129 .https_or_http()
130 .enable_http1()
131 .enable_http2()
132 .build();
133 launchdarkly_sdk_transport::HyperTransport::builder()
137 .disable_proxy()
138 .connect_timeout(connect_timeout)
139 .read_timeout(read_timeout)
140 .build_with_connector(connector)
141}
142
143fn ld_ctx<F>(build_info: &'static BuildInfo, ctx_builder: F) -> Result<ld::Context, anyhow::Error>
144where
145 F: FnOnce(&mut ld::MultiContextBuilder) -> Result<(), anyhow::Error>,
146{
147 let mut builder = ld::MultiContextBuilder::new();
154
155 builder.add_context(
156 ld::ContextBuilder::new(build_info.sha)
157 .kind("build")
158 .set_string("semver_version", build_info.semver_version().to_string())
159 .build()
160 .map_err(|e| anyhow::anyhow!(e))?,
161 );
162
163 ctx_builder(&mut builder)?;
164
165 builder.build().map_err(|e| anyhow::anyhow!(e))
166}
167
168struct SyncedConfigSet<F>
169where
170 F: Fn(&ConfigUpdates, &ConfigSet) + Send,
171{
172 set: ConfigSet,
173 ld_client: Option<ld::Client>,
174 ld_ctx: ld::Context,
175 on_update: F,
176}
177
178impl<F: Fn(&ConfigUpdates, &ConfigSet) + Send> SyncedConfigSet<F> {
179 async fn sync_loop(self, tick_interval: Option<Duration>) {
181 let Some(tick_interval) = tick_interval else {
182 tracing::info!("skipping SyncedConfigSet sync as tick_interval = None");
183 return;
184 };
185
186 let mut interval = time::interval(tick_interval);
187 interval.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
188
189 tracing::info!(
190 "synchronizing SyncedConfigSet values every {} seconds",
191 tick_interval.as_secs()
192 );
193
194 loop {
195 interval.tick().await;
196
197 if let Err(err) = self.sync() {
198 tracing::info!("SyncedConfigSet: {err}");
199 }
200 }
201 }
202
203 fn sync(&self) -> Result<(), anyhow::Error> {
205 let mut updates = ConfigUpdates::default();
206 let Some(ld_client) = &self.ld_client else {
207 (self.on_update)(&updates, &self.set);
208 return Ok(());
209 };
210 for entry in self.set.entries() {
211 let val = dyn_into_flag(entry.val()).expect("new() verifies all configs can convert");
212 let flag_var = ld_client.variation(&self.ld_ctx, entry.name(), val);
213 let update = match (entry.val(), flag_var) {
214 (ConfigVal::Bool(_), ld::FlagValue::Bool(flag)) => ConfigVal::Bool(flag),
215 (ConfigVal::U32(_), ld::FlagValue::Number(flag)) => {
216 ConfigVal::U32(u32::cast_lossy(flag))
217 }
218 (ConfigVal::Usize(_), ld::FlagValue::Number(flag)) => {
219 ConfigVal::Usize(usize::cast_lossy(flag))
220 }
221 (ConfigVal::F64(_), ld::FlagValue::Number(flag)) => ConfigVal::F64(flag),
222 (ConfigVal::String(_), ld::FlagValue::Str(flag)) => ConfigVal::String(flag),
223 (ConfigVal::Duration(_), ld::FlagValue::Str(flag)) => {
224 match humantime::parse_duration(&flag) {
225 Ok(d) => ConfigVal::Duration(d),
226 Err(e) => {
227 tracing::warn!("failed to parse Duration for {}: {}", entry.name(), e);
228 continue;
229 }
230 }
231 }
232 (ConfigVal::Json(_), ld::FlagValue::Json(flag)) => ConfigVal::Json(flag),
233
234 (ConfigVal::Bool(_), _)
237 | (ConfigVal::U32(_), _)
238 | (ConfigVal::Usize(_), _)
239 | (ConfigVal::F64(_), _)
240 | (ConfigVal::Duration(_), _)
241 | (ConfigVal::Json(_), _)
242 | (ConfigVal::OptUsize(_), _)
243 | (ConfigVal::String(_), _)
244 | (ConfigVal::OptString(_), _) => {
245 tracing::warn!("LD flag type mismatch for {}", entry.name());
246 continue;
247 }
248 };
249 tracing::debug!(
250 "updating config value {} from {:?} to {:?}",
251 &entry.name(),
252 &entry.val(),
253 update
254 );
255 updates.add_dynamic(entry.name(), update);
256 }
257 updates.apply(&self.set);
258 (self.on_update)(&updates, &self.set);
259 Ok(())
260 }
261}
262
263fn dyn_into_flag(val: ConfigVal) -> Result<ld::FlagValue, anyhow::Error> {
266 Ok(match val {
270 ConfigVal::Bool(v) => ld::FlagValue::Bool(v),
271 ConfigVal::U32(v) => ld::FlagValue::Number(v.into()),
272 ConfigVal::Usize(v) => ld::FlagValue::Number(f64::cast_lossy(v)),
273 ConfigVal::OptUsize(_) => anyhow::bail!("OptUsize None cannot be converted to a FlagValue"),
274 ConfigVal::F64(v) => ld::FlagValue::Number(v),
275 ConfigVal::String(v) => ld::FlagValue::Str(v),
276 ConfigVal::OptString(_) => {
277 anyhow::bail!("OptString None cannot be converted to a FlagValue")
278 }
279 ConfigVal::Duration(v) => ld::FlagValue::Str(humantime::format_duration(v).to_string()),
280 ConfigVal::Json(v) => ld::FlagValue::Json(v),
281 })
282}