add etcd, param slicing, paging fix in ssm

teller-providers/Cargo.toml

@@ -20,6 +20,7 @@ default = [
     "aws_secretsmanager",
     "google_secretmanager",
     "hashicorp_consul",
+    "etcd",
 ]
 
 ssm = ["aws", "dep:aws-sdk-ssm"]
@@ -29,6 +30,7 @@ hashicorp_vault = ["dep:vaultrs", "dep:rustify"]
 dotenv = ["dep:dotenvy"]
 hashicorp_consul = ["dep:rs-consul"]
 aws = ["dep:aws-config"]
+etcd = ["dep:etcd-client"]
 
 [dependencies]
 async-trait = { workspace = true }
@@ -44,6 +46,7 @@ fs-err = "2.9.0"
 home = "0.5.5"
 hyper = "0.14"
 base64 = "0.22.0"
+tokio = "1"
 # gcp
 google-secretmanager1 = { version = "5.0.2", optional = true }
 crc32c = { version = "0.6", optional = true }
@@ -61,7 +64,12 @@ rustify = { version = "0.5.3", optional = true }
 # HashiCorp Consul 
 rs-consul = { version = "0.6.0", optional = true }
 
+etcd-client = { version = "0.12", optional = true }
+
 [dev-dependencies]
 insta = { workspace = true }
 dockertest-server = { version = "0.1.7", features = ["hashi", "cloud"] }
+dockertest = "0.3.0"
 tokio = { workspace = true }
+test-log = "0.2"
+tracing = "0.1"

teller-providers/src/lib.rs

@@ -65,6 +65,9 @@ pub enum Error {
 
     #[error("LIST {path}: {msg}")]
     ListError { path: String, msg: String },
+
+    #[error("{0}")]
+    CreateProviderError(String),
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;

teller-providers/src/providers/etcd.rs

@@ -0,0 +1,241 @@
+//! Hashicorp Consul
+//!
+//!
+//! ## Example configuration
+//!
+//! ```yaml
+//! providers:
+//!  consul1:
+//!    kind: hashicorp_consul
+//!    # options: ...
+//! ```
+//! ## Options
+//!
+//! See [`EtcdOptions`] for more.
+//!
+
+use async_trait::async_trait;
+use etcd_client::{Client, DeleteOptions, GetOptions, KvClient};
+use serde_derive::{Deserialize, Serialize};
+use tokio::sync::Mutex;
+
+use super::ProviderKind;
+use crate::{
+    config::{PathMap, ProviderInfo, KV},
+    Error, Provider, Result,
+};
+
+#[allow(clippy::module_name_repetitions)]
+#[derive(Default, Serialize, Deserialize, Debug, Clone)]
+pub struct EtcdOptions {
+    /// Etcd address.
+    pub address: Option<String>,
+}
+
+pub struct Etcd {
+    pub client: Mutex<Client>,
+    pub name: String,
+}
+
+fn to_err(_pm: &PathMap, err: etcd_client::Error) -> Error {
+    Error::Any(Box::new(err))
+}
+async fn create_client() -> Result<Client> {
+    Ok(Client::connect(["127.0.0.1:2379"], None)
+        .await
+        .map_err(|err| Error::CreateProviderError(err.to_string()))?)
+}
+
+impl Etcd {
+    /// Create a new hashicorp Consul
+    ///
+    /// # Errors
+    ///
+    /// This function will return an error if cannot create a provider
+    pub async fn new(name: &str, opts: Option<EtcdOptions>) -> Result<Self> {
+        let opts = opts.unwrap_or_default();
+
+        let address = opts
+            .address
+            .as_ref()
+            .ok_or_else(|| Error::Message("address not present.".to_string()))?;
+
+        Ok(Self {
+            client: Mutex::new(
+                Client::connect([address], None)
+                    .await
+                    .map_err(|err| Error::CreateProviderError(err.to_string()))?,
+            ),
+            name: name.to_string(),
+        })
+    }
+}
+
+#[async_trait]
+impl Provider for Etcd {
+    fn kind(&self) -> ProviderInfo {
+        ProviderInfo {
+            kind: ProviderKind::Etcd,
+            name: self.name.clone(),
+        }
+    }
+
+    async fn get(&self, pm: &PathMap) -> Result<Vec<KV>> {
+        let mut client = create_client().await?;
+
+        let res = if pm.keys.is_empty() {
+            client
+                .get(pm.path.as_str(), Some(GetOptions::new().with_prefix()))
+                .await
+                .map_err(|err| to_err(pm, err))?
+                .kvs()
+                .to_vec()
+        } else {
+            let mut res = Vec::new();
+            for key in pm.keys.keys() {
+                let fetched = client
+                    .get(format!("{}/{}", pm.path.as_str(), key), None)
+                    .await
+                    .map_err(|err| to_err(pm, err))?
+                    .kvs()
+                    .to_vec();
+                res.extend(fetched);
+            }
+            res
+        };
+
+        drop(client);
+
+        if res.is_empty() {
+            return Err(Error::NotFound {
+                msg: "not found".to_string(),
+                path: pm.path.clone(),
+            });
+        }
+
+        let mut results = vec![];
+        for kv_pair in res {
+            let key = kv_pair.key_str().map_err(|err| to_err(pm, err))?;
+
+            // strip path pref
+            let key = key
+                .strip_prefix(&pm.path)
+                .map_or(key, |s| s.trim_start_matches('/'));
+
+            let val = kv_pair.value_str().map_err(|err| to_err(pm, err))?;
+
+            results.push(KV::from_value(val, key, key, pm, self.kind()));
+        }
+
+        Ok(results)
+    }
+
+    async fn put(&self, pm: &PathMap, kvs: &[KV]) -> Result<()> {
+        let mut client = create_client().await?;
+        for kv in kvs {
+            client
+                .put(
+                    format!("{}/{}", pm.path, kv.key).as_str(),
+                    kv.value.as_bytes().to_vec(),
+                    None,
+                )
+                .await
+                .map_err(|e| to_err(pm, e))?;
+        }
+        drop(client);
+
+        Ok(())
+    }
+
+    async fn del(&self, pm: &PathMap) -> Result<()> {
+        let mut client = create_client().await?;
+        if pm.keys.is_empty() {
+            client
+                .delete(
+                    pm.path.as_str(),
+                    Some(DeleteOptions::default().with_prefix()),
+                )
+                .await
+                .map_err(|err| to_err(pm, err))?;
+        } else {
+            for key in pm.keys.keys().map(|kv| format!("{}/{kv}", &pm.path)) {
+                client
+                    .delete(key, None)
+                    .await
+                    .map_err(|err| to_err(pm, err))?;
+            }
+        };
+        drop(client);
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+    use crate::providers::test_utils;
+
+    const PORT: u32 = 2379;
+
+    #[test_log::test]
+    #[cfg(not(windows))]
+    fn sanity_test() {
+        use std::{collections::HashMap, env, time::Duration};
+
+        use dockertest::{waitfor, Composition, DockerTest, Image};
+
+        if env::var("RUNNER_OS").unwrap_or_default() == "macOS" {
+            return;
+        }
+        let mut test = DockerTest::new();
+        let wait = Box::new(waitfor::MessageWait {
+            message: "serving client traffic insecurely".to_string(),
+            source: waitfor::MessageSource::Stderr,
+            timeout: 20,
+        });
+
+        let mut env = HashMap::new();
+
+        env.insert("ALLOW_NONE_AUTHENTICATION".to_string(), "yes".to_string());
+
+        #[cfg(target_arch = "aarch64")]
+        env.insert("ETCD_UNSUPPORTED_ARCH".to_string(), "arm64".to_string());
+
+        #[cfg(target_arch = "aarch64")]
+        let image_name = "bitnami/etcd";
+        #[cfg(not(target_arch = "aarch64"))]
+        let image_name = "bitnami/etcd";
+
+        let image = Image::with_repository(image_name)
+            .pull_policy(dockertest::PullPolicy::IfNotPresent)
+            .source(dockertest::Source::DockerHub);
+        let mut etcd_container = Composition::with_image(image)
+            .with_container_name("etcd-server")
+            .with_env(env)
+            .with_wait_for(wait);
+        etcd_container.port_map(PORT, PORT);
+
+        test.add_composition(etcd_container);
+
+        test.run(|ops| async move {
+            let _instance = ops.handle("etcd-server");
+            let address = format!("localhost:{PORT}");
+            // banner is not enough, we have to wait for the image to stabilize
+
+            let p = Box::new(
+                super::Etcd::new(
+                    "etcd",
+                    Some(EtcdOptions {
+                        address: Some(address),
+                    }),
+                )
+                .await
+                .unwrap(),
+            ) as Box<dyn Provider + Send + Sync>;
+
+            test_utils::ProviderTest::new(p).run().await;
+        });
+    }
+}

teller-providers/src/providers/hashicorp_consul.rs

@@ -133,7 +133,10 @@ impl Provider for HashiCorpConsul {
 
             let (_, key) = kv_pair.key.rsplit_once('/').unwrap_or(("", &kv_pair.key));
 
-            results.push(KV::from_value(&val, key, key, pm, self.kind()));
+            // take all or slice the requested keys
+            if pm.keys.is_empty() || pm.keys.contains_key(key) {
+                results.push(KV::from_value(&val, key, key, pm, self.kind()));
+            }
         }
 
         Ok(results)

teller-providers/src/providers/hashicorp_vault.rs

@@ -98,7 +98,6 @@ fn parse_path(pm: &PathMap) -> Result<(&str, &str, &str)> {
 }
 
 fn xerr(pm: &PathMap, e: ClientError) -> Error {
-    println!("{e:?}");
     match e {
         ClientError::RestClientError { source } => match source {
             rustify::errors::ClientError::ServerResponseError { code, content } => {

teller-providers/src/providers/mod.rs

@@ -27,6 +27,9 @@ pub mod google_secretmanager;
 #[cfg(feature = "hashicorp_consul")]
 pub mod hashicorp_consul;
 
+#[cfg(feature = "etcd")]
+pub mod etcd;
+
 lazy_static! {
     pub static ref PROVIDER_KINDS: String = {
         let providers: Vec<String> = ProviderKind::iter()
@@ -66,6 +69,10 @@ pub enum ProviderKind {
     #[cfg(feature = "google_secretmanager")]
     #[serde(rename = "google_secretmanager")]
     GoogleSecretManager,
+
+    #[cfg(feature = "etcd")]
+    #[serde(rename = "etcd")]
+    Etcd,
 }
 
 impl std::fmt::Display for ProviderKind {

teller-providers/src/providers/snapshots/teller_providers__providers__test_utils__[get-secret_development].snap

@@ -4,9 +4,9 @@ expression: res
 ---
 [
     KV {
-        value: "DEBUG",
-        key: "log_level",
-        from_key: "log_level",
+        value: "Teller",
+        key: "app",
+        from_key: "app",
         path: Some(
             PathInfo {
                 id: "",
@@ -29,9 +29,9 @@ expression: res
         ),
     },
     KV {
-        value: "Teller",
-        key: "app",
-        from_key: "app",
+        value: "{\"DB_PASS\": \"1234\",\"DB_NAME\": \"FOO\"}",
+        key: "db",
+        from_key: "db",
         path: Some(
             PathInfo {
                 id: "",
@@ -54,9 +54,9 @@ expression: res
         ),
     },
     KV {
-        value: "{\"DB_PASS\": \"1234\",\"DB_NAME\": \"FOO\"}",
-        key: "db",
-        from_key: "db",
+        value: "DEBUG",
+        key: "log_level",
+        from_key: "log_level",
         path: Some(
             PathInfo {
                 id: "",