rws_k8s/k8s/
config.rs

1//! ConfigMaps and Secrets from the Kubernetes API, kept current by a watch:
2//! no volume mount, no restart, and a change reaches the app within a
3//! second or two.
4//!
5//! ```rust,no_run
6//! # async fn example() -> rws_k8s::k8s::Result<()> {
7//! use rws_k8s::k8s::{client, config::ConfigWatch};
8//!
9//! let client = client::client().await?;
10//! let settings = ConfigWatch::config_map(client.clone(), "default", "app-settings").await?;
11//! let page_size: usize = settings.get("page_size").and_then(|v| v.parse().ok()).unwrap_or(20);
12//! settings.on_change(|data| println!("settings now: {data:?}"));
13//!
14//! // `RWS_CONFIG_*` keys straight into rws's own settings, re-read on change:
15//! settings.apply_to_env("RWS_CONFIG_");
16//!
17//! let password = rws_k8s::k8s::config::resolve(&client, "k8s://default/db-credentials#password").await?;
18//! # Ok(()) }
19//! ```
20//!
21//! The pod needs `get`, `list` and `watch` on the object's kind in its
22//! namespace ([`manifests::Rbac`](super::manifests::Rbac)).
23
24use std::collections::BTreeMap;
25use std::sync::{Arc, RwLock};
26
27use futures_util::StreamExt;
28use k8s_openapi::api::core::v1::{ConfigMap, Secret};
29use kube::api::Api;
30use kube::runtime::watcher;
31
32use super::{Error, Result};
33
34type Listener = Arc<dyn Fn(&BTreeMap<String, String>) + Send + Sync>;
35
36struct Shared {
37    data: RwLock<BTreeMap<String, String>>,
38    listeners: RwLock<Vec<Listener>>,
39    present: std::sync::atomic::AtomicBool,
40}
41
42/// One ConfigMap's or Secret's data, kept current. Cheap to clone; clones
43/// share it. The watch stops when the last clone is dropped.
44#[derive(Clone)]
45pub struct ConfigWatch {
46    shared: Arc<Shared>,
47    _task: Arc<AbortOnDrop>,
48}
49
50struct AbortOnDrop(tokio::task::JoinHandle<()>);
51
52impl Drop for AbortOnDrop {
53    fn drop(&mut self) {
54        self.0.abort();
55    }
56}
57
58/// A ConfigMap's data and binary data (as UTF-8 where it is).
59pub fn config_map_data(cm: &ConfigMap) -> BTreeMap<String, String> {
60    let mut out = cm.data.clone().unwrap_or_default();
61    for (k, v) in cm.binary_data.clone().unwrap_or_default() {
62        if let Ok(text) = String::from_utf8(v.0) {
63            out.insert(k, text);
64        }
65    }
66    out
67}
68
69/// A Secret's data (decoded) and string data, as UTF-8 where it is.
70pub fn secret_data(secret: &Secret) -> BTreeMap<String, String> {
71    let mut out = BTreeMap::new();
72    for (k, v) in secret.data.clone().unwrap_or_default() {
73        if let Ok(text) = String::from_utf8(v.0) {
74            out.insert(k, text);
75        }
76    }
77    out.extend(secret.string_data.clone().unwrap_or_default());
78    out
79}
80
81impl ConfigWatch {
82    /// Watch the ConfigMap `name` in `namespace`. Fails when it can't be
83    /// read now (missing, or no permission).
84    pub async fn config_map(client: kube::Client, namespace: &str, name: &str) -> Result<Self> {
85        Self::start(Api::<ConfigMap>::namespaced(client, namespace), name, config_map_data).await
86    }
87
88    /// Watch the Secret `name` in `namespace`.
89    pub async fn secret(client: kube::Client, namespace: &str, name: &str) -> Result<Self> {
90        Self::start(Api::<Secret>::namespaced(client, namespace), name, secret_data).await
91    }
92
93    async fn start<K>(api: Api<K>, name: &str, read: fn(&K) -> BTreeMap<String, String>) -> Result<Self>
94    where
95        K: kube::Resource + Clone + std::fmt::Debug + serde::de::DeserializeOwned + Send + Sync + 'static,
96    {
97        let initial = api.get(name).await?;
98        let shared = Arc::new(Shared {
99            data: RwLock::new(read(&initial)),
100            listeners: RwLock::new(Vec::new()),
101            present: std::sync::atomic::AtomicBool::new(true),
102        });
103        let config = watcher::Config::default().fields(&format!("metadata.name={name}"));
104        let task_shared = Arc::clone(&shared);
105        let task = tokio::spawn(async move {
106            let mut stream = watcher::watcher(api, config).boxed();
107            while let Some(event) = stream.next().await {
108                let next = match event {
109                    Ok(watcher::Event::Apply(obj)) | Ok(watcher::Event::InitApply(obj)) => Some(read(&obj)),
110                    Ok(watcher::Event::Delete(_)) => {
111                        task_shared.present.store(false, std::sync::atomic::Ordering::SeqCst);
112                        None
113                    }
114                    Ok(_) => None,
115                    Err(e) => {
116                        eprintln!("[k8s] config watch: {e}");
117                        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
118                        None
119                    }
120                };
121                if let Some(data) = next {
122                    task_shared.present.store(true, std::sync::atomic::Ordering::SeqCst);
123                    let changed = {
124                        let mut current = task_shared.data.write().unwrap_or_else(|e| e.into_inner());
125                        let changed = *current != data;
126                        *current = data.clone();
127                        changed
128                    };
129                    if changed {
130                        let listeners = task_shared.listeners.read().unwrap_or_else(|e| e.into_inner()).clone();
131                        for listener in listeners {
132                            listener(&data);
133                        }
134                    }
135                }
136            }
137        });
138        Ok(ConfigWatch { shared, _task: Arc::new(AbortOnDrop(task)) })
139    }
140
141    /// The value of `key` now.
142    pub fn get(&self, key: &str) -> Option<String> {
143        self.shared.data.read().unwrap_or_else(|e| e.into_inner()).get(key).cloned()
144    }
145
146    /// Every key now.
147    pub fn snapshot(&self) -> BTreeMap<String, String> {
148        self.shared.data.read().unwrap_or_else(|e| e.into_inner()).clone()
149    }
150
151    /// Whether the object exists (it may be deleted after the watch started;
152    /// the last data is kept).
153    pub fn exists(&self) -> bool {
154        self.shared.present.load(std::sync::atomic::Ordering::SeqCst)
155    }
156
157    /// Call `f` with the new data after each change.
158    pub fn on_change(&self, f: impl Fn(&BTreeMap<String, String>) + Send + Sync + 'static) {
159        self.shared.listeners.write().unwrap_or_else(|e| e.into_inner()).push(Arc::new(f));
160    }
161
162    /// Set every key starting with `prefix` (`RWS_CONFIG_`) as an
163    /// environment variable now and after each change, and have rws re-read
164    /// its settings (`server_config::invalidate`). Keys under the prefix
165    /// that disappear are left as they were.
166    pub fn apply_to_env(&self, prefix: &str) {
167        apply(&self.snapshot(), prefix);
168        let prefix = prefix.to_string();
169        self.on_change(move |data| apply(data, &prefix));
170    }
171}
172
173fn apply(data: &BTreeMap<String, String>, prefix: &str) {
174    let mut any = false;
175    for (k, v) in data.iter().filter(|(k, _)| k.starts_with(prefix)) {
176        // SAFETY of set_var: rws reads its settings through ServerConfig
177        // snapshots, rebuilt after `invalidate`.
178        unsafe { std::env::set_var(k, v) };
179        any = true;
180    }
181    if any {
182        rws_server::server_config::invalidate();
183    }
184}
185
186/// A `k8s://<namespace>/<secret>#<key>` reference.
187#[derive(Debug, Clone, PartialEq, Eq)]
188pub struct SecretRef {
189    pub namespace: String,
190    pub name: String,
191    pub key: String,
192}
193
194impl SecretRef {
195    /// Parse `k8s://namespace/name#key` (`k8s://name#key`: the current
196    /// namespace).
197    pub fn parse(reference: &str) -> Result<SecretRef> {
198        let rest = reference.strip_prefix("k8s://").ok_or_else(|| Error::Other(format!("{reference}: not a k8s:// reference")))?;
199        let (path, key) = rest.split_once('#').ok_or_else(|| Error::Other(format!("{reference}: no #key")))?;
200        let (namespace, name) = match path.split_once('/') {
201            Some((ns, name)) => (ns.to_string(), name.to_string()),
202            None => (super::client::current_namespace(), path.to_string()),
203        };
204        if namespace.is_empty() || name.is_empty() || key.is_empty() || name.contains('/') {
205            return Err(Error::Other(format!("{reference}: expected k8s://namespace/name#key")));
206        }
207        Ok(SecretRef { namespace, name, key: key.to_string() })
208    }
209}
210
211/// The value a `k8s://namespace/name#key` reference names: one key of a
212/// Secret, read now.
213pub async fn resolve(client: &kube::Client, reference: &str) -> Result<String> {
214    let r = SecretRef::parse(reference)?;
215    let secret = Api::<Secret>::namespaced(client.clone(), &r.namespace).get(&r.name).await?;
216    secret_data(&secret)
217        .remove(&r.key)
218        .ok_or_else(|| Error::Other(format!("secret {}/{} has no key {}", r.namespace, r.name, r.key)))
219}
220
221#[cfg(test)]
222mod tests;
rws_k8s/k8s/config.rs

rws_k8s/k8s/
config.rs
