featherbit/config_store/
etcd.rs

1//! etcd-backed config store for HA clustering.
2//!
3//! Talks to **etcd's v3 HTTP/JSON gateway** (the `/v3/kv/*` and
4//! `/v3/auth/authenticate` endpoints) through the shared [`OutboundClient`] —
5//! no gRPC, `protoc`, or `tonic` dependency. Config lives under per-resource
6//! keys (`<prefix>/routes/<name>`, `<prefix>/policies/<name>`,
7//! `<prefix>/consumers/<name>`, `<prefix>/supernodes/<name>`, `<prefix>/plugin_configs/<name>`,
8//! `<prefix>/stores/<name>`),
9//! each value the resource's JSON. Every gateway instance loads from the same prefix and a
10//! background poll task keeps the cluster converged (see [`spawn_watch`]).
11//! Note: an older build sharing the same prefix garbage-collects unknown key
12//! families on its next commit.
13//!
14//! # Route ordering
15//! etcd returns keys in lexicographic order, so in etcd mode routes are matched
16//! **by name order**, not file declaration order. Name routes accordingly when
17//! precedence matters.
18//!
19//! # v1 limitations (documented follow-ups)
20//! - Plaintext / no TLS to etcd (loopback or private-network etcd).
21//! - Poll-based convergence (default 2 s) rather than a streaming watch.
22//! - Uses the first endpoint; multi-endpoint failover is a later addition.
23
24use std::path::PathBuf;
25use std::sync::Arc;
26use std::time::Duration;
27
28use async_trait::async_trait;
29use base64::engine::general_purpose::STANDARD as BASE64;
30use base64::Engine;
31use bytes::Bytes;
32use serde_json::{json, Value};
33use tokio::sync::Mutex;
34
35use crate::config::{EtcdConfig, GatewayConfig, PluginConfigDef, StoreConfig, SystemConfig};
36use crate::config::{PolicyConfig, RouteConfig, SupernodeConfig};
37use crate::config_store::ConfigStore;
38use crate::consumers::ConsumerConfig;
39use crate::outbound::{OutboundClient, OutboundRequest};
40use crate::state::SharedState;
41
42/// etcd config store over the v3 HTTP/JSON gateway.
43pub struct EtcdConfigStore {
44    client: Arc<OutboundClient>,
45    /// etcd base URLs (v1 uses the first).
46    endpoints: Vec<String>,
47    /// Key prefix (no trailing slash), e.g. `/featherbit`.
48    prefix: String,
49    auth: Option<(String, String)>,
50    /// Cached auth token, refreshed on 401.
51    token: Mutex<Option<String>>,
52    timeout: Duration,
53}
54
55impl EtcdConfigStore {
56    /// Builds the store from `system.yaml`'s etcd settings.
57    pub fn new(cfg: &EtcdConfig) -> Self {
58        Self {
59            client: Arc::new(OutboundClient::new()),
60            endpoints: cfg.endpoints.clone(),
61            prefix: cfg.prefix.trim_end_matches('/').to_string(),
62            auth: match (&cfg.user, &cfg.password) {
63                (Some(u), Some(p)) => Some((u.clone(), p.clone())),
64                _ => None,
65            },
66            token: Mutex::new(None),
67            timeout: Duration::from_millis(cfg.timeout_ms),
68        }
69    }
70
71    fn base(&self) -> Result<&str, String> {
72        self.endpoints
73            .first()
74            .map(String::as_str)
75            .ok_or_else(|| "no etcd endpoints configured".to_string())
76    }
77
78    /// POSTs `body` to an etcd JSON endpoint, attaching the auth token and
79    /// re-authenticating once on 401.
80    async fn call(&self, path: &str, body: &Value) -> Result<Value, String> {
81        match self.call_once(path, body).await {
82            Err(EtcdCallError::Unauthorized) if self.auth.is_some() => {
83                self.authenticate().await?;
84                self.call_once(path, body).await.map_err(|e| e.to_string())
85            }
86            other => other.map_err(|e| e.to_string()),
87        }
88    }
89
90    async fn call_once(&self, path: &str, body: &Value) -> Result<Value, EtcdCallError> {
91        let url = format!("{}{}", self.base().map_err(EtcdCallError::Other)?, path);
92        let mut headers = vec![("content-type".to_string(), "application/json".to_string())];
93        if let Some(token) = self.token.lock().await.clone() {
94            headers.push(("authorization".to_string(), token));
95        }
96        let req = OutboundRequest {
97            method: http::Method::POST,
98            url,
99            headers,
100            body: Bytes::from(serde_json::to_vec(body).unwrap_or_default()),
101            timeout: self.timeout,
102            ssl_verify: true,
103            tls: None,
104        };
105        let resp = self
106            .client
107            .request(req)
108            .await
109            .map_err(|e| EtcdCallError::Other(e.to_string()))?;
110        if resp.status == 401 {
111            return Err(EtcdCallError::Unauthorized);
112        }
113        if resp.status != 200 {
114            return Err(EtcdCallError::Other(format!(
115                "etcd {} returned status {}: {}",
116                path,
117                resp.status,
118                String::from_utf8_lossy(&resp.body)
119            )));
120        }
121        serde_json::from_slice(&resp.body)
122            .map_err(|e| EtcdCallError::Other(format!("invalid etcd response: {}", e)))
123    }
124
125    /// Authenticates and caches the token.
126    async fn authenticate(&self) -> Result<(), String> {
127        let (user, pass) = match &self.auth {
128            Some(c) => c,
129            None => return Ok(()),
130        };
131        let resp = self
132            .call_once(
133                "/v3/auth/authenticate",
134                &json!({ "name": user, "password": pass }),
135            )
136            .await
137            .map_err(|e| e.to_string())?;
138        let token = resp
139            .get("token")
140            .and_then(|v| v.as_str())
141            .ok_or("etcd auth response missing token")?;
142        *self.token.lock().await = Some(token.to_string());
143        Ok(())
144    }
145
146    /// Ranges all keys under the prefix, returning `(key, value_bytes)` pairs.
147    async fn range_prefix(&self) -> Result<Vec<(String, Vec<u8>)>, String> {
148        let key = format!("{}/", self.prefix);
149        let range_end = prefix_range_end(key.as_bytes());
150        let resp = self
151            .call(
152                "/v3/kv/range",
153                &json!({
154                    "key": BASE64.encode(key.as_bytes()),
155                    "range_end": BASE64.encode(range_end),
156                }),
157            )
158            .await?;
159        let mut out = Vec::new();
160        if let Some(kvs) = resp.get("kvs").and_then(|v| v.as_array()) {
161            for kv in kvs {
162                let k = kv.get("key").and_then(|v| v.as_str()).unwrap_or("");
163                let v = kv.get("value").and_then(|v| v.as_str()).unwrap_or("");
164                let key = BASE64
165                    .decode(k)
166                    .ok()
167                    .and_then(|b| String::from_utf8(b).ok())
168                    .ok_or("etcd key not valid base64/utf8")?;
169                let value = BASE64
170                    .decode(v)
171                    .map_err(|_| "etcd value not valid base64")?;
172                out.push((key, value));
173            }
174        }
175        Ok(out)
176    }
177
178    async fn put(&self, key: &str, value: &[u8]) -> Result<(), String> {
179        self.call(
180            "/v3/kv/put",
181            &json!({ "key": BASE64.encode(key.as_bytes()), "value": BASE64.encode(value) }),
182        )
183        .await
184        .map(|_| ())
185    }
186
187    async fn delete(&self, key: &str) -> Result<(), String> {
188        self.call(
189            "/v3/kv/deleterange",
190            &json!({ "key": BASE64.encode(key.as_bytes()) }),
191        )
192        .await
193        .map(|_| ())
194    }
195
196    fn route_key(&self, name: &str) -> String {
197        format!("{}/routes/{}", self.prefix, name)
198    }
199    fn policy_key(&self, name: &str) -> String {
200        format!("{}/policies/{}", self.prefix, name)
201    }
202    fn consumer_key(&self, name: &str) -> String {
203        format!("{}/consumers/{}", self.prefix, name)
204    }
205    fn supernode_key(&self, name: &str) -> String {
206        format!("{}/supernodes/{}", self.prefix, name)
207    }
208    fn plugin_config_key(&self, name: &str) -> String {
209        format!("{}/plugin_configs/{}", self.prefix, name)
210    }
211    fn store_key(&self, name: &str) -> String {
212        format!("{}/stores/{}", self.prefix, name)
213    }
214
215    /// Writes every resource in `gw` to etcd (used to seed an empty prefix).
216    async fn write_all(&self, gw: &GatewayConfig) -> Result<(), String> {
217        for r in &gw.routes {
218            self.put(&self.route_key(&r.name), &serde_json::to_vec(r).unwrap())
219                .await?;
220        }
221        for p in &gw.policies {
222            self.put(&self.policy_key(&p.name), &serde_json::to_vec(p).unwrap())
223                .await?;
224        }
225        for c in &gw.consumers {
226            self.put(&self.consumer_key(&c.name), &serde_json::to_vec(c).unwrap())
227                .await?;
228        }
229        for s in &gw.supernodes {
230            self.put(
231                &self.supernode_key(&s.name),
232                &serde_json::to_vec(s).unwrap(),
233            )
234            .await?;
235        }
236        for pc in &gw.plugin_configs {
237            self.put(
238                &self.plugin_config_key(&pc.name),
239                &serde_json::to_vec(pc).unwrap(),
240            )
241            .await?;
242        }
243        for s in &gw.stores {
244            self.put(&self.store_key(&s.name), &serde_json::to_vec(s).unwrap())
245                .await?;
246        }
247        Ok(())
248    }
249}
250
251#[async_trait]
252impl ConfigStore for EtcdConfigStore {
253    async fn load_all(&self) -> Result<GatewayConfig, String> {
254        let kvs = self.range_prefix().await?;
255        gateway_from_kvs(&self.prefix, kvs)
256    }
257
258    async fn commit(&self, state: &SharedState, candidate: GatewayConfig) -> Result<(), String> {
259        // 1. Reject invalid config before touching etcd (synchronous 400).
260        state.validate_gateway(&candidate)?;
261
262        // 2. Reconcile etcd to match the candidate: put all desired keys, then
263        //    delete any keys under the prefix that the candidate dropped.
264        let current: std::collections::HashSet<String> = self
265            .range_prefix()
266            .await?
267            .into_iter()
268            .map(|(k, _)| k)
269            .collect();
270        let mut desired = std::collections::HashSet::new();
271
272        for r in &candidate.routes {
273            let key = self.route_key(&r.name);
274            self.put(&key, &serde_json::to_vec(r).unwrap()).await?;
275            desired.insert(key);
276        }
277        for p in &candidate.policies {
278            let key = self.policy_key(&p.name);
279            self.put(&key, &serde_json::to_vec(p).unwrap()).await?;
280            desired.insert(key);
281        }
282        for c in &candidate.consumers {
283            let key = self.consumer_key(&c.name);
284            self.put(&key, &serde_json::to_vec(c).unwrap()).await?;
285            desired.insert(key);
286        }
287        for s in &candidate.supernodes {
288            let key = self.supernode_key(&s.name);
289            self.put(&key, &serde_json::to_vec(s).unwrap()).await?;
290            desired.insert(key);
291        }
292        for pc in &candidate.plugin_configs {
293            let key = self.plugin_config_key(&pc.name);
294            self.put(&key, &serde_json::to_vec(pc).unwrap()).await?;
295            desired.insert(key);
296        }
297        for s in &candidate.stores {
298            let key = self.store_key(&s.name);
299            self.put(&key, &serde_json::to_vec(s).unwrap()).await?;
300            desired.insert(key);
301        }
302        for stale in current.difference(&desired) {
303            self.delete(stale).await?;
304        }
305
306        // 3. Apply locally so the writing node reflects the change immediately;
307        //    other nodes converge on their next poll. Idempotent with the poll.
308        state.apply_gateway(candidate).await
309    }
310}
311
312/// Assembles a [`GatewayConfig`] from the etcd key/value pairs under `prefix`.
313///
314/// Keys are `<prefix>/{routes,policies,consumers,supernodes,plugin_configs,stores}/<name>`; values
315/// are the resource JSON. Unknown key shapes are skipped. Malformed resource
316/// JSON is an error (so a bad write surfaces rather than silently dropping
317/// config).
318fn gateway_from_kvs(prefix: &str, kvs: Vec<(String, Vec<u8>)>) -> Result<GatewayConfig, String> {
319    let mut gw = GatewayConfig {
320        routes: Vec::new(),
321        policies: Vec::new(),
322        consumers: Vec::new(),
323        supernodes: Vec::new(),
324        plugin_configs: Vec::new(),
325        stores: Vec::new(),
326    };
327    for (key, value) in kvs {
328        let rest = match key.strip_prefix(&format!("{}/", prefix)) {
329            Some(r) => r,
330            None => continue,
331        };
332        let (category, name) = match rest.split_once('/') {
333            Some(p) => p,
334            None => continue,
335        };
336        match category {
337            "routes" => {
338                let r: RouteConfig = serde_json::from_slice(&value)
339                    .map_err(|e| format!("bad route '{}': {}", key, e))?;
340                gw.routes.push(r);
341            }
342            "policies" => {
343                let p: PolicyConfig = serde_json::from_slice(&value)
344                    .map_err(|e| format!("bad policy '{}': {}", key, e))?;
345                gw.policies.push(p);
346            }
347            "consumers" => {
348                let c: ConsumerConfig = serde_json::from_slice(&value)
349                    .map_err(|e| format!("bad consumer '{}': {}", key, e))?;
350                gw.consumers.push(c);
351            }
352            "supernodes" => {
353                let s: SupernodeConfig = serde_json::from_slice(&value)
354                    .map_err(|e| format!("bad supernode '{}': {}", key, e))?;
355                gw.supernodes.push(s);
356            }
357            "plugin_configs" => {
358                let pc: PluginConfigDef = serde_json::from_slice(&value)
359                    .map_err(|e| format!("bad plugin config '{}': {}", key, e))?;
360                gw.plugin_configs.push(pc);
361            }
362            "stores" => {
363                let s: StoreConfig = serde_json::from_str(&String::from_utf8_lossy(&value))
364                    .map_err(|e| format!("bad store '{}': {}", name, e))?;
365                gw.stores.push(s);
366            }
367            _ => {}
368        }
369    }
370    Ok(gw)
371}
372
373/// Computes etcd's prefix range-end: the prefix with its last non-`0xff` byte
374/// incremented (so a Range covers all keys starting with the prefix). An
375/// all-`0xff` prefix ranges to the end of the keyspace (`[0]`).
376fn prefix_range_end(prefix: &[u8]) -> Vec<u8> {
377    let mut end = prefix.to_vec();
378    while let Some(&last) = end.last() {
379        if last < 0xff {
380            *end.last_mut().unwrap() = last + 1;
381            return end;
382        }
383        end.pop();
384    }
385    vec![0]
386}
387
388fn is_empty(gw: &GatewayConfig) -> bool {
389    gw.routes.is_empty()
390        && gw.policies.is_empty()
391        && gw.consumers.is_empty()
392        && gw.supernodes.is_empty()
393        && gw.plugin_configs.is_empty()
394        && gw.stores.is_empty()
395}
396
397/// Builds the etcd store and the initial gateway config.
398///
399/// Seeds etcd from the local `gateway.yaml` (`seed_path`) when the etcd prefix
400/// is empty, then loads from etcd. Returns `config_path = None` (etcd mode does
401/// not reload from disk).
402///
403/// The seed candidate is **dry-run compiled** before anything is written (see
404/// [`crate::state::validate_gateway_config`]). A local config that does not
405/// compile is logged and skipped, leaving the prefix empty: startup then
406/// proceeds with no routes, and the next boot re-seeds once the file is fixed.
407/// Writing it anyway would publish a config the whole cluster then fails to
408/// apply — and, because the prefix would no longer be empty, no later boot
409/// would ever re-seed it.
410pub async fn build_source(
411    system: &SystemConfig,
412    seed_path: &std::path::Path,
413) -> Result<(Arc<dyn ConfigStore>, GatewayConfig, Option<PathBuf>), String> {
414    let cfg = system
415        .config
416        .etcd
417        .as_ref()
418        .ok_or("config.source is 'etcd' but no 'config.etcd' block is set")?;
419    let store = Arc::new(EtcdConfigStore::new(cfg));
420    if store.auth.is_some() {
421        store.authenticate().await?;
422    }
423
424    let mut gateway = store.load_all().await?;
425    if is_empty(&gateway) {
426        // Raw load: seeding must publish the `${VAR}` placeholder form, never
427        // locally-resolved secrets — every cluster node resolves its own env
428        // at compile time.
429        if let Ok(local) = crate::config::load_yaml::<GatewayConfig>(seed_path) {
430            if !is_empty(&local) {
431                // Never publish a config the cluster cannot apply: compile it
432                // locally first and skip the seed on failure, so the prefix
433                // stays empty and a later boot can re-seed a fixed file.
434                match crate::state::validate_gateway_config(&local) {
435                    Ok(()) => {
436                        tracing::info!("etcd prefix empty — seeding from {}", seed_path.display());
437                        store.write_all(&local).await?;
438                        gateway = store.load_all().await?;
439                    }
440                    Err(e) => tracing::error!(
441                        "etcd prefix empty but the seed config at {} does not compile — \
442                         NOT seeding (fix it and restart; the prefix stays empty so the \
443                         next boot re-seeds): {}",
444                        seed_path.display(),
445                        e
446                    ),
447                }
448            }
449        }
450    }
451
452    let store: Arc<dyn ConfigStore> = store;
453    Ok((store, gateway, None))
454}
455
456/// Spawns the poll-based convergence task: every `poll_interval` (default 2 s)
457/// it reloads from etcd and, when the config changed, applies it. Errors are
458/// logged and retried on the next tick — a transient etcd outage leaves the
459/// last-good config serving.
460pub fn spawn_watch(state: Arc<SharedState>, system: &SystemConfig) {
461    let interval = Duration::from_secs(2);
462    let store = state.config_store.clone();
463    tokio::spawn(async move {
464        let mut last: Option<String> = None;
465        loop {
466            tokio::time::sleep(interval).await;
467            match store.load_all().await {
468                Ok(gw) => {
469                    let fingerprint = serde_json::to_string(&gw).unwrap_or_default();
470                    if last.as_deref() != Some(fingerprint.as_str()) {
471                        match state.apply_gateway(gw).await {
472                            Ok(_) => last = Some(fingerprint),
473                            Err(e) => tracing::error!("etcd config apply failed: {}", e),
474                        }
475                    }
476                }
477                Err(e) => tracing::warn!("etcd poll failed (keeping last-good config): {}", e),
478            }
479        }
480    });
481    let _ = system; // reserved for future per-source poll-interval config
482}
483
484/// Internal call-error type distinguishing a 401 (triggers re-auth) from other
485/// failures.
486enum EtcdCallError {
487    Unauthorized,
488    Other(String),
489}
490
491impl std::fmt::Display for EtcdCallError {
492    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
493        match self {
494            Self::Unauthorized => write!(f, "etcd unauthorized"),
495            Self::Other(m) => write!(f, "{}", m),
496        }
497    }
498}
499
500#[cfg(test)]
501mod tests {
502    use super::*;
503
504    #[test]
505    fn test_prefix_range_end() {
506        assert_eq!(prefix_range_end(b"/featherbit/"), b"/featherbit0".to_vec()); // '/'+1 = '0'
507        assert_eq!(prefix_range_end(b"ab"), b"ac".to_vec());
508        assert_eq!(prefix_range_end(&[0xff, 0xff]), vec![0]);
509        assert_eq!(prefix_range_end(&[0x01, 0xff]), vec![0x02]);
510    }
511
512    #[test]
513    fn test_gateway_from_kvs() {
514        let prefix = "/featherbit";
515        let route = serde_json::to_vec(&json!({
516            "name": "r", "match": { "path": "/api/*" }, "policy": "p"
517        }))
518        .unwrap();
519        let policy = serde_json::to_vec(&json!({
520            "name": "p",
521            "nodes": [{ "id": "listener", "type": "listener" }, { "id": "client", "type": "client" }],
522            "edges": [{ "from": "listener.out", "to": "client.in" }]
523        }))
524        .unwrap();
525        let consumer = serde_json::to_vec(&json!({ "name": "alice" })).unwrap();
526
527        let kvs = vec![
528            ("/featherbit/routes/r".to_string(), route),
529            ("/featherbit/policies/p".to_string(), policy),
530            ("/featherbit/consumers/alice".to_string(), consumer),
531            ("/featherbit/unknown/x".to_string(), b"{}".to_vec()), // skipped
532            ("/other/routes/z".to_string(), b"{}".to_vec()),       // wrong prefix, skipped
533        ];
534        let gw = gateway_from_kvs(prefix, kvs).unwrap();
535        assert_eq!(gw.routes.len(), 1);
536        assert_eq!(gw.routes[0].name, "r");
537        assert_eq!(gw.policies.len(), 1);
538        assert_eq!(gw.consumers.len(), 1);
539        assert_eq!(gw.consumers[0].name, "alice");
540    }
541
542    #[test]
543    fn test_gateway_from_kvs_rejects_bad_json() {
544        let kvs = vec![("/featherbit/routes/r".to_string(), b"not json".to_vec())];
545        assert!(gateway_from_kvs("/featherbit", kvs).is_err());
546    }
547
548    #[test]
549    fn test_is_empty() {
550        assert!(is_empty(&GatewayConfig {
551            routes: vec![],
552            policies: vec![],
553            consumers: vec![],
554            supernodes: vec![],
555            plugin_configs: vec![],
556            stores: vec![],
557        }));
558    }
559
560    #[test]
561    fn test_gateway_from_kvs_parses_supernodes() {
562        let sn = serde_json::json!({
563            "name": "secured-call",
564            "nodes": [ { "id": "input", "type": "input", "config": {} } ],
565            "edges": []
566        });
567        let kvs = vec![(
568            "gw/supernodes/secured-call".to_string(),
569            serde_json::to_vec(&sn).unwrap(),
570        )];
571        let gw = gateway_from_kvs("gw", kvs).unwrap();
572        assert_eq!(gw.supernodes.len(), 1);
573        assert_eq!(gw.supernodes[0].name, "secured-call");
574    }
575
576    #[test]
577    fn test_gateway_from_kvs_bad_supernode_json_is_error() {
578        let kvs = vec![("gw/supernodes/x".to_string(), b"not json".to_vec())];
579        let err = gateway_from_kvs("gw", kvs).unwrap_err();
580        assert!(err.contains("bad supernode"), "{err}");
581    }
582
583    #[test]
584    fn test_is_empty_counts_supernodes() {
585        let mut gw: GatewayConfig = serde_yaml::from_str("{}").unwrap();
586        assert!(is_empty(&gw));
587        gw.supernodes.push(SupernodeConfig {
588            name: "s".into(),
589            description: None,
590            nodes: vec![],
591            edges: vec![],
592        });
593        assert!(!is_empty(&gw));
594    }
595
596    #[test]
597    fn test_gateway_from_kvs_parses_plugin_configs() {
598        let def = serde_json::json!({ "name": "corp", "type": "cors", "config": {} });
599        let kvs = vec![(
600            "gw/plugin_configs/corp".to_string(),
601            serde_json::to_vec(&def).unwrap(),
602        )];
603        let gw = gateway_from_kvs("gw", kvs).unwrap();
604        assert_eq!(gw.plugin_configs.len(), 1);
605        assert_eq!(gw.plugin_configs[0].name, "corp");
606    }
607
608    #[test]
609    fn test_gateway_from_kvs_bad_plugin_config_json_is_error() {
610        let kvs = vec![("gw/plugin_configs/x".to_string(), b"not json".to_vec())];
611        let err = gateway_from_kvs("gw", kvs).unwrap_err();
612        assert!(err.contains("bad plugin config"), "{err}");
613    }
614
615    #[test]
616    fn test_is_empty_counts_plugin_configs() {
617        let mut gw: GatewayConfig = serde_yaml::from_str("{}").unwrap();
618        assert!(is_empty(&gw));
619        gw.plugin_configs.push(PluginConfigDef {
620            name: "c".into(),
621            plugin_type: "cors".into(),
622            description: None,
623            config: Default::default(),
624        });
625        assert!(!is_empty(&gw));
626    }
627
628    #[test]
629    fn test_gateway_from_kvs_parses_stores() {
630        let kvs = vec![(
631            "/fb/stores/s1".to_string(),
632            br#"{"name":"s1","type":"redis","url":"redis://127.0.0.1:6379","key_prefix":"fb","connect_timeout_ms":2000}"#.to_vec(),
633        )];
634        let gw = gateway_from_kvs("/fb", kvs).unwrap();
635        assert_eq!(gw.stores.len(), 1);
636        assert_eq!(gw.stores[0].name, "s1");
637        assert_eq!(gw.stores[0].store_type, "redis");
638        assert!(!is_empty(&gw));
639
640        let err = gateway_from_kvs(
641            "/fb",
642            vec![("/fb/stores/bad".to_string(), b"{notjson".to_vec())],
643        )
644        .unwrap_err();
645        assert!(err.contains("bad store 'bad'"), "{err}");
646    }
647}
featherbit/config_store/etcd.rs

featherbit/config_store/
etcd.rs
