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>`), each value the resource's JSON. Every gateway
8//! instance loads from the same prefix and a background poll task keeps the
9//! cluster converged (see [`spawn_watch`]).
10//!
11//! # Route ordering
12//! etcd returns keys in lexicographic order, so in etcd mode routes are matched
13//! **by name order**, not file declaration order. Name routes accordingly when
14//! precedence matters.
15//!
16//! # v1 limitations (documented follow-ups)
17//! - Plaintext / no TLS to etcd (loopback or private-network etcd).
18//! - Poll-based convergence (default 2 s) rather than a streaming watch.
19//! - Uses the first endpoint; multi-endpoint failover is a later addition.
20
21use std::path::PathBuf;
22use std::sync::Arc;
23use std::time::Duration;
24
25use async_trait::async_trait;
26use base64::engine::general_purpose::STANDARD as BASE64;
27use base64::Engine;
28use bytes::Bytes;
29use serde_json::{json, Value};
30use tokio::sync::Mutex;
31
32use crate::config::{EtcdConfig, GatewayConfig, SystemConfig};
33use crate::config::{PolicyConfig, RouteConfig};
34use crate::config_store::ConfigStore;
35use crate::consumers::ConsumerConfig;
36use crate::outbound::{OutboundClient, OutboundRequest};
37use crate::state::SharedState;
38
39/// etcd config store over the v3 HTTP/JSON gateway.
40pub struct EtcdConfigStore {
41    client: Arc<OutboundClient>,
42    /// etcd base URLs (v1 uses the first).
43    endpoints: Vec<String>,
44    /// Key prefix (no trailing slash), e.g. `/featherbit`.
45    prefix: String,
46    auth: Option<(String, String)>,
47    /// Cached auth token, refreshed on 401.
48    token: Mutex<Option<String>>,
49    timeout: Duration,
50}
51
52impl EtcdConfigStore {
53    /// Builds the store from `system.yaml`'s etcd settings.
54    pub fn new(cfg: &EtcdConfig) -> Self {
55        Self {
56            client: Arc::new(OutboundClient::new()),
57            endpoints: cfg.endpoints.clone(),
58            prefix: cfg.prefix.trim_end_matches('/').to_string(),
59            auth: match (&cfg.user, &cfg.password) {
60                (Some(u), Some(p)) => Some((u.clone(), p.clone())),
61                _ => None,
62            },
63            token: Mutex::new(None),
64            timeout: Duration::from_millis(cfg.timeout_ms),
65        }
66    }
67
68    fn base(&self) -> Result<&str, String> {
69        self.endpoints
70            .first()
71            .map(String::as_str)
72            .ok_or_else(|| "no etcd endpoints configured".to_string())
73    }
74
75    /// POSTs `body` to an etcd JSON endpoint, attaching the auth token and
76    /// re-authenticating once on 401.
77    async fn call(&self, path: &str, body: &Value) -> Result<Value, String> {
78        match self.call_once(path, body).await {
79            Err(EtcdCallError::Unauthorized) if self.auth.is_some() => {
80                self.authenticate().await?;
81                self.call_once(path, body).await.map_err(|e| e.to_string())
82            }
83            other => other.map_err(|e| e.to_string()),
84        }
85    }
86
87    async fn call_once(&self, path: &str, body: &Value) -> Result<Value, EtcdCallError> {
88        let url = format!("{}{}", self.base().map_err(EtcdCallError::Other)?, path);
89        let mut headers = vec![("content-type".to_string(), "application/json".to_string())];
90        if let Some(token) = self.token.lock().await.clone() {
91            headers.push(("authorization".to_string(), token));
92        }
93        let req = OutboundRequest {
94            method: http::Method::POST,
95            url,
96            headers,
97            body: Bytes::from(serde_json::to_vec(body).unwrap_or_default()),
98            timeout: self.timeout,
99            ssl_verify: true,
100            tls: None,
101        };
102        let resp = self
103            .client
104            .request(req)
105            .await
106            .map_err(|e| EtcdCallError::Other(e.to_string()))?;
107        if resp.status == 401 {
108            return Err(EtcdCallError::Unauthorized);
109        }
110        if resp.status != 200 {
111            return Err(EtcdCallError::Other(format!(
112                "etcd {} returned status {}: {}",
113                path,
114                resp.status,
115                String::from_utf8_lossy(&resp.body)
116            )));
117        }
118        serde_json::from_slice(&resp.body)
119            .map_err(|e| EtcdCallError::Other(format!("invalid etcd response: {}", e)))
120    }
121
122    /// Authenticates and caches the token.
123    async fn authenticate(&self) -> Result<(), String> {
124        let (user, pass) = match &self.auth {
125            Some(c) => c,
126            None => return Ok(()),
127        };
128        let resp = self
129            .call_once(
130                "/v3/auth/authenticate",
131                &json!({ "name": user, "password": pass }),
132            )
133            .await
134            .map_err(|e| e.to_string())?;
135        let token = resp
136            .get("token")
137            .and_then(|v| v.as_str())
138            .ok_or("etcd auth response missing token")?;
139        *self.token.lock().await = Some(token.to_string());
140        Ok(())
141    }
142
143    /// Ranges all keys under the prefix, returning `(key, value_bytes)` pairs.
144    async fn range_prefix(&self) -> Result<Vec<(String, Vec<u8>)>, String> {
145        let key = format!("{}/", self.prefix);
146        let range_end = prefix_range_end(key.as_bytes());
147        let resp = self
148            .call(
149                "/v3/kv/range",
150                &json!({
151                    "key": BASE64.encode(key.as_bytes()),
152                    "range_end": BASE64.encode(range_end),
153                }),
154            )
155            .await?;
156        let mut out = Vec::new();
157        if let Some(kvs) = resp.get("kvs").and_then(|v| v.as_array()) {
158            for kv in kvs {
159                let k = kv.get("key").and_then(|v| v.as_str()).unwrap_or("");
160                let v = kv.get("value").and_then(|v| v.as_str()).unwrap_or("");
161                let key = BASE64
162                    .decode(k)
163                    .ok()
164                    .and_then(|b| String::from_utf8(b).ok())
165                    .ok_or("etcd key not valid base64/utf8")?;
166                let value = BASE64
167                    .decode(v)
168                    .map_err(|_| "etcd value not valid base64")?;
169                out.push((key, value));
170            }
171        }
172        Ok(out)
173    }
174
175    async fn put(&self, key: &str, value: &[u8]) -> Result<(), String> {
176        self.call(
177            "/v3/kv/put",
178            &json!({ "key": BASE64.encode(key.as_bytes()), "value": BASE64.encode(value) }),
179        )
180        .await
181        .map(|_| ())
182    }
183
184    async fn delete(&self, key: &str) -> Result<(), String> {
185        self.call(
186            "/v3/kv/deleterange",
187            &json!({ "key": BASE64.encode(key.as_bytes()) }),
188        )
189        .await
190        .map(|_| ())
191    }
192
193    fn route_key(&self, name: &str) -> String {
194        format!("{}/routes/{}", self.prefix, name)
195    }
196    fn policy_key(&self, name: &str) -> String {
197        format!("{}/policies/{}", self.prefix, name)
198    }
199    fn consumer_key(&self, name: &str) -> String {
200        format!("{}/consumers/{}", self.prefix, name)
201    }
202
203    /// Writes every resource in `gw` to etcd (used to seed an empty prefix).
204    async fn write_all(&self, gw: &GatewayConfig) -> Result<(), String> {
205        for r in &gw.routes {
206            self.put(&self.route_key(&r.name), &serde_json::to_vec(r).unwrap())
207                .await?;
208        }
209        for p in &gw.policies {
210            self.put(&self.policy_key(&p.name), &serde_json::to_vec(p).unwrap())
211                .await?;
212        }
213        for c in &gw.consumers {
214            self.put(&self.consumer_key(&c.name), &serde_json::to_vec(c).unwrap())
215                .await?;
216        }
217        Ok(())
218    }
219}
220
221#[async_trait]
222impl ConfigStore for EtcdConfigStore {
223    async fn load_all(&self) -> Result<GatewayConfig, String> {
224        let kvs = self.range_prefix().await?;
225        gateway_from_kvs(&self.prefix, kvs)
226    }
227
228    async fn commit(&self, state: &SharedState, candidate: GatewayConfig) -> Result<(), String> {
229        // 1. Reject invalid config before touching etcd (synchronous 400).
230        state.validate_gateway(&candidate)?;
231
232        // 2. Reconcile etcd to match the candidate: put all desired keys, then
233        //    delete any keys under the prefix that the candidate dropped.
234        let current: std::collections::HashSet<String> = self
235            .range_prefix()
236            .await?
237            .into_iter()
238            .map(|(k, _)| k)
239            .collect();
240        let mut desired = std::collections::HashSet::new();
241
242        for r in &candidate.routes {
243            let key = self.route_key(&r.name);
244            self.put(&key, &serde_json::to_vec(r).unwrap()).await?;
245            desired.insert(key);
246        }
247        for p in &candidate.policies {
248            let key = self.policy_key(&p.name);
249            self.put(&key, &serde_json::to_vec(p).unwrap()).await?;
250            desired.insert(key);
251        }
252        for c in &candidate.consumers {
253            let key = self.consumer_key(&c.name);
254            self.put(&key, &serde_json::to_vec(c).unwrap()).await?;
255            desired.insert(key);
256        }
257        for stale in current.difference(&desired) {
258            self.delete(stale).await?;
259        }
260
261        // 3. Apply locally so the writing node reflects the change immediately;
262        //    other nodes converge on their next poll. Idempotent with the poll.
263        state.apply_gateway(candidate).await
264    }
265}
266
267/// Assembles a [`GatewayConfig`] from the etcd key/value pairs under `prefix`.
268///
269/// Keys are `<prefix>/{routes,policies,consumers}/<name>`; values are the
270/// resource JSON. Unknown key shapes are skipped. Malformed resource JSON is an
271/// error (so a bad write surfaces rather than silently dropping config).
272fn gateway_from_kvs(prefix: &str, kvs: Vec<(String, Vec<u8>)>) -> Result<GatewayConfig, String> {
273    let mut gw = GatewayConfig {
274        routes: Vec::new(),
275        policies: Vec::new(),
276        consumers: Vec::new(),
277    };
278    for (key, value) in kvs {
279        let rest = match key.strip_prefix(&format!("{}/", prefix)) {
280            Some(r) => r,
281            None => continue,
282        };
283        let (category, _name) = match rest.split_once('/') {
284            Some(p) => p,
285            None => continue,
286        };
287        match category {
288            "routes" => {
289                let r: RouteConfig = serde_json::from_slice(&value)
290                    .map_err(|e| format!("bad route '{}': {}", key, e))?;
291                gw.routes.push(r);
292            }
293            "policies" => {
294                let p: PolicyConfig = serde_json::from_slice(&value)
295                    .map_err(|e| format!("bad policy '{}': {}", key, e))?;
296                gw.policies.push(p);
297            }
298            "consumers" => {
299                let c: ConsumerConfig = serde_json::from_slice(&value)
300                    .map_err(|e| format!("bad consumer '{}': {}", key, e))?;
301                gw.consumers.push(c);
302            }
303            _ => {}
304        }
305    }
306    Ok(gw)
307}
308
309/// Computes etcd's prefix range-end: the prefix with its last non-`0xff` byte
310/// incremented (so a Range covers all keys starting with the prefix). An
311/// all-`0xff` prefix ranges to the end of the keyspace (`[0]`).
312fn prefix_range_end(prefix: &[u8]) -> Vec<u8> {
313    let mut end = prefix.to_vec();
314    while let Some(&last) = end.last() {
315        if last < 0xff {
316            *end.last_mut().unwrap() = last + 1;
317            return end;
318        }
319        end.pop();
320    }
321    vec![0]
322}
323
324fn is_empty(gw: &GatewayConfig) -> bool {
325    gw.routes.is_empty() && gw.policies.is_empty() && gw.consumers.is_empty()
326}
327
328/// Builds the etcd store and the initial gateway config.
329///
330/// Seeds etcd from the local `gateway.yaml` (`seed_path`) when the etcd prefix
331/// is empty, then loads from etcd. Returns `config_path = None` (etcd mode does
332/// not reload from disk).
333pub async fn build_source(
334    system: &SystemConfig,
335    seed_path: &std::path::Path,
336) -> Result<(Arc<dyn ConfigStore>, GatewayConfig, Option<PathBuf>), String> {
337    let cfg = system
338        .config
339        .etcd
340        .as_ref()
341        .ok_or("config.source is 'etcd' but no 'config.etcd' block is set")?;
342    let store = Arc::new(EtcdConfigStore::new(cfg));
343    if store.auth.is_some() {
344        store.authenticate().await?;
345    }
346
347    let mut gateway = store.load_all().await?;
348    if is_empty(&gateway) {
349        if let Ok(local) = crate::config::load_yaml_with_env::<GatewayConfig>(seed_path) {
350            if !is_empty(&local) {
351                tracing::info!("etcd prefix empty — seeding from {}", seed_path.display());
352                store.write_all(&local).await?;
353                gateway = store.load_all().await?;
354            }
355        }
356    }
357
358    let store: Arc<dyn ConfigStore> = store;
359    Ok((store, gateway, None))
360}
361
362/// Spawns the poll-based convergence task: every `poll_interval` (default 2 s)
363/// it reloads from etcd and, when the config changed, applies it. Errors are
364/// logged and retried on the next tick — a transient etcd outage leaves the
365/// last-good config serving.
366pub fn spawn_watch(state: Arc<SharedState>, system: &SystemConfig) {
367    let interval = Duration::from_secs(2);
368    let store = state.config_store.clone();
369    tokio::spawn(async move {
370        let mut last: Option<String> = None;
371        loop {
372            tokio::time::sleep(interval).await;
373            match store.load_all().await {
374                Ok(gw) => {
375                    let fingerprint = serde_json::to_string(&gw).unwrap_or_default();
376                    if last.as_deref() != Some(fingerprint.as_str()) {
377                        match state.apply_gateway(gw).await {
378                            Ok(_) => last = Some(fingerprint),
379                            Err(e) => tracing::error!("etcd config apply failed: {}", e),
380                        }
381                    }
382                }
383                Err(e) => tracing::warn!("etcd poll failed (keeping last-good config): {}", e),
384            }
385        }
386    });
387    let _ = system; // reserved for future per-source poll-interval config
388}
389
390/// Internal call-error type distinguishing a 401 (triggers re-auth) from other
391/// failures.
392enum EtcdCallError {
393    Unauthorized,
394    Other(String),
395}
396
397impl std::fmt::Display for EtcdCallError {
398    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
399        match self {
400            Self::Unauthorized => write!(f, "etcd unauthorized"),
401            Self::Other(m) => write!(f, "{}", m),
402        }
403    }
404}
405
406#[cfg(test)]
407mod tests {
408    use super::*;
409
410    #[test]
411    fn test_prefix_range_end() {
412        assert_eq!(prefix_range_end(b"/featherbit/"), b"/featherbit0".to_vec()); // '/'+1 = '0'
413        assert_eq!(prefix_range_end(b"ab"), b"ac".to_vec());
414        assert_eq!(prefix_range_end(&[0xff, 0xff]), vec![0]);
415        assert_eq!(prefix_range_end(&[0x01, 0xff]), vec![0x02]);
416    }
417
418    #[test]
419    fn test_gateway_from_kvs() {
420        let prefix = "/featherbit";
421        let route = serde_json::to_vec(&json!({
422            "name": "r", "match": { "path": "/api/*" }, "policy": "p"
423        }))
424        .unwrap();
425        let policy = serde_json::to_vec(&json!({
426            "name": "p",
427            "nodes": [{ "id": "listener", "type": "listener" }, { "id": "client", "type": "client" }],
428            "edges": [{ "from": "listener.out", "to": "client.in" }]
429        }))
430        .unwrap();
431        let consumer = serde_json::to_vec(&json!({ "name": "alice" })).unwrap();
432
433        let kvs = vec![
434            ("/featherbit/routes/r".to_string(), route),
435            ("/featherbit/policies/p".to_string(), policy),
436            ("/featherbit/consumers/alice".to_string(), consumer),
437            ("/featherbit/unknown/x".to_string(), b"{}".to_vec()), // skipped
438            ("/other/routes/z".to_string(), b"{}".to_vec()),       // wrong prefix, skipped
439        ];
440        let gw = gateway_from_kvs(prefix, kvs).unwrap();
441        assert_eq!(gw.routes.len(), 1);
442        assert_eq!(gw.routes[0].name, "r");
443        assert_eq!(gw.policies.len(), 1);
444        assert_eq!(gw.consumers.len(), 1);
445        assert_eq!(gw.consumers[0].name, "alice");
446    }
447
448    #[test]
449    fn test_gateway_from_kvs_rejects_bad_json() {
450        let kvs = vec![("/featherbit/routes/r".to_string(), b"not json".to_vec())];
451        assert!(gateway_from_kvs("/featherbit", kvs).is_err());
452    }
453
454    #[test]
455    fn test_is_empty() {
456        assert!(is_empty(&GatewayConfig {
457            routes: vec![],
458            policies: vec![],
459            consumers: vec![]
460        }));
461    }
462}
featherbit/config_store/etcd.rs

featherbit/config_store/
etcd.rs
