feat: add K8sSecretProvider (#1473)

* feat: add k8s secret provider

* chore: generalize retry

* nits

Author: gsanchezgavier

agent-control/src/agent_control/run/k8s.rs

@@ -15,6 +15,7 @@ use crate::agent_control::run::AgentControlRunner;
 use crate::agent_control::version_updater::k8s::K8sACUpdater;
 use crate::agent_type::render::renderer::TemplateRenderer;
 use crate::agent_type::variable::Variable;
+use crate::agent_type::variable::namespace::Namespace;
 #[cfg_attr(test, mockall_double::double)]
 use crate::k8s::client::SyncK8sClient;
 use crate::opamp::effective_config::loader::DefaultEffectiveConfigLoaderBuilder;
@@ -23,7 +24,8 @@ use crate::opamp::instance_id::k8s::getter::{Identifiers, get_identifiers};
 use crate::opamp::operations::build_opamp_with_channel;
 use crate::opamp::remote_config::validators::SupportedRemoteConfigValidator;
 use crate::opamp::remote_config::validators::regexes::RegexValidator;
-use crate::secrets_provider::SecretsProvidersRegistry;
+use crate::secrets_provider::k8s_secret::K8sSecretProvider;
+use crate::secrets_provider::{SecretsProviderType, SecretsProvidersRegistry};
 use crate::sub_agent::effective_agents_assembler::LocalEffectiveAgentsAssembler;
 use crate::sub_agent::identity::AgentIdentity;
 use crate::sub_agent::k8s::builder::SupervisorBuilderK8s;
@@ -133,7 +135,7 @@ impl AgentControlRunner {
         let template_renderer = TemplateRenderer::default()
             .with_agent_control_variables(agent_control_variables.clone().into_iter());
 
-        let secrets_providers = if let Some(config) = &agent_control_config.secrets_providers {
+        let mut secrets_providers = if let Some(config) = &agent_control_config.secrets_providers {
             SecretsProvidersRegistry::try_from(config.clone()).map_err(|e| {
                 AgentError::ConfigResolve(AgentControlConfigError::Load(format!(
                     "Failed to load secrets providers: {e}"
@@ -142,6 +144,10 @@ impl AgentControlRunner {
         } else {
             HashMap::default()
         };
+        secrets_providers.insert(
+            Namespace::K8sSecret,
+            SecretsProviderType::K8sSecret(K8sSecretProvider::new(k8s_client.clone())),
+        );
 
         let agents_assembler = Arc::new(LocalEffectiveAgentsAssembler::new(
             self.agent_type_registry.clone(),

agent-control/src/agent_type/variable/namespace.rs

@@ -15,6 +15,7 @@ pub enum Namespace {
     // These are loaded every time a remote config is received.
     EnvironmentVariable,
     Vault,
+    K8sSecret,
 }
 
 impl Namespace {
@@ -32,6 +33,8 @@ impl Namespace {
     const ENVIRONMENT_VARIABLE: &'static str = "env";
     /// Encapsulates the secrets retrieved from a HashiCorp Vault
     const VAULT_SECRET: &'static str = "vault";
+    /// Encapsulates the secrets retrieved from K8s Secrets
+    const K8S_SECRET: &'static str = "kubesec";
 
     pub fn namespaced_name(&self, name: &str) -> NamespacedVariableName {
         format!("{}{}{}", self, Self::PREFIX_NS_SEPARATOR, name)
@@ -52,6 +55,7 @@ impl Display for Namespace {
             Self::AgentControl => Self::AC,
             Self::EnvironmentVariable => Self::ENVIRONMENT_VARIABLE,
             Self::Vault => Self::VAULT_SECRET,
+            Self::K8sSecret => Self::K8S_SECRET,
         };
         write!(f, "{}{ns}", Self::PREFIX)
     }
@@ -78,5 +82,13 @@ mod tests {
             "nr-ac:test".to_string(),
             Namespace::AgentControl.namespaced_name("test")
         );
+        assert_eq!(
+            "nr-vault:test".to_string(),
+            Namespace::Vault.namespaced_name("test")
+        );
+        assert_eq!(
+            "nr-kubesec:test".to_string(),
+            Namespace::K8sSecret.namespaced_name("test")
+        );
     }
 }

agent-control/src/agent_type/variable/secret_variables.rs

@@ -84,6 +84,9 @@ impl SecretVariables {
                 SecretsProviderType::Vault(provider) => {
                     self.load_secrets_at(namespace, provider)?
                 }
+                SecretsProviderType::K8sSecret(provider) => {
+                    self.load_secrets_at(namespace, provider)?
+                }
             };
             result.extend(secrets_map);
         }

agent-control/src/cli/uninstall_agent_control.rs

@@ -3,10 +3,11 @@ use crate::agent_control::config::{
 };
 use crate::cli::errors::CliError;
 use crate::cli::install_agent_control::{RELEASE_NAME, REPOSITORY_NAME};
-use crate::cli::utils::{retry, try_new_k8s_client};
+use crate::cli::utils::try_new_k8s_client;
 #[cfg_attr(test, mockall_double::double)]
 use crate::k8s::client::SyncK8sClient;
 use crate::k8s::labels::Labels;
+use crate::utils::retry::retry;
 use clap::Parser;
 use either::Either;
 use kube::api::{DynamicObject, ObjectList, TypeMeta};

agent-control/src/cli/utils.rs

@@ -3,7 +3,6 @@ use crate::cli::errors::CliError;
 use crate::k8s::client::SyncK8sClient;
 use std::collections::BTreeMap;
 use std::sync::Arc;
-use std::time::Duration;
 use tracing::debug;
 
 /// Parses a string of key-value pairs separated by commas.
@@ -48,21 +47,6 @@ pub fn try_new_k8s_client() -> Result<SyncK8sClient, CliError> {
     SyncK8sClient::try_new(runtime).map_err(|err| CliError::K8sClient(err.to_string()))
 }
 
-pub fn retry<F>(max_attempts: usize, interval: Duration, mut f: F) -> Result<(), CliError>
-where
-    F: FnMut() -> Result<(), CliError>,
-{
-    let mut last_err = Ok(());
-    for _ in 0..max_attempts {
-        let Err(err) = f() else {
-            return Ok(());
-        };
-        last_err = Err(err);
-        std::thread::sleep(interval);
-    }
-    last_err
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

agent-control/src/k8s/client.rs

@@ -6,7 +6,7 @@ use crate::k8s::dynamic_object::TypeMetaNamespaced;
 use crate::k8s::utils::{get_namespace, get_type_meta};
 use either::Either;
 use k8s_openapi::api::apps::v1::{DaemonSet, Deployment, StatefulSet};
-use k8s_openapi::api::core::v1::ConfigMap;
+use k8s_openapi::api::core::v1::{ConfigMap, Secret};
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{APIResourceList, ObjectMeta};
 use kube::api::ObjectList;
 use kube::api::entry::Entry;
@@ -157,6 +157,17 @@ impl SyncK8sClient {
             .block_on(self.async_client.get_configmap_key(name, namespace, key))
     }
 
+    // Gets the decoded secret key assuming it contains a String.
+    pub fn get_secret_key(
+        &self,
+        name: &str,
+        namespace: &str,
+        key: &str,
+    ) -> Result<Option<String>, K8sError> {
+        self.runtime
+            .block_on(self.async_client.get_secret_key(name, namespace, key))
+    }
+
     pub fn set_configmap_key(
         &self,
         name: &str,
@@ -256,6 +267,35 @@ impl AsyncK8sClient {
         Ok(list)
     }
 
+    /// Gets the decoded secret key assuming it contains a String.
+    pub async fn get_secret_key(
+        &self,
+        name: &str,
+        namespace: &str,
+        key: &str,
+    ) -> Result<Option<String>, K8sError> {
+        let secret_client = Api::<Secret>::namespaced(self.client.clone(), namespace);
+
+        let Some(secret) = secret_client.get_opt(name).await? else {
+            debug!("Secret {}:{} not found", namespace, name);
+            return Ok(None);
+        };
+
+        let Some(data) = secret.data else {
+            debug!("Secret {}:{} missing data", namespace, name);
+            return Ok(None);
+        };
+
+        let Some(value) = data.get(key) else {
+            debug!("Secret {}:{} missing key {}", namespace, name, key);
+            return Ok(None);
+        };
+
+        let v = std::str::from_utf8(&value.0)
+            .map_err(|e| K8sError::Generic(format!("decoding secret key: {}", e)))?;
+        Ok(Some(v.to_string()))
+    }
+
     pub async fn delete_configmap_collection(
         &self,
         namespace: &str,
@@ -276,20 +316,22 @@ impl AsyncK8sClient {
         let cm_client: Api<ConfigMap> =
             Api::<ConfigMap>::namespaced(self.client.clone(), namespace);
 
-        if let Some(cm) = cm_client.get_opt(name).await? {
-            if let Some(data) = cm.data {
-                if let Some(key) = data.get(key) {
-                    return Ok(Some(key.clone()));
-                }
-                debug!("ConfigMap {} missing key {}", name, key)
-            } else {
-                debug!("ConfigMap {} missing data", name)
-            }
-        } else {
-            debug!("ConfigMap {} not found", name)
-        }
+        let Some(cm) = cm_client.get_opt(name).await? else {
+            debug!("ConfigMap {}:{} not found", namespace, name);
+            return Ok(None);
+        };
+
+        let Some(data) = cm.data else {
+            debug!("ConfigMap {}:{} missing data", namespace, name);
+            return Ok(None);
+        };
+
+        let Some(value) = data.get(key) else {
+            debug!("ConfigMap {}:{} missing key {}", namespace, name, key);
+            return Ok(None);
+        };
 
-        Ok(None)
+        Ok(Some(value.clone()))
     }
 
     pub async fn set_configmap_key(

agent-control/src/k8s/error.rs

@@ -6,8 +6,11 @@ use kube::{api, config::KubeconfigError};
 
 #[derive(thiserror::Error, Debug)]
 pub enum K8sError {
+    #[error("{0}")]
+    Generic(String),
+
     #[error("the kube client returned an error: `{0}`")]
-    Generic(#[from] kube::Error),
+    KubeRs(#[from] kube::Error),
 
     #[error("it is not possible to read kubeconfig: `{0}`")]
     UnableToSetupClientKubeconfig(#[from] KubeconfigError),

agent-control/src/k8s/store.rs

@@ -225,7 +225,7 @@ pub mod tests {
         k8s_client
             .expect_get_configmap_key()
             .once()
-            .returning(move |_, _, _| Err(K8sError::Generic(kube::Error::TlsRequired)));
+            .returning(move |_, _, _| Err(K8sError::KubeRs(kube::Error::TlsRequired)));
 
         let k8s_store = K8sStore::new(Arc::new(k8s_client), TEST_NAMESPACE.to_string());
 
@@ -288,7 +288,7 @@ pub mod tests {
         k8s_client
             .expect_set_configmap_key()
             .once()
-            .returning(move |_, _, _, _, _| Err(K8sError::Generic(kube::Error::TlsRequired)));
+            .returning(move |_, _, _, _, _| Err(K8sError::KubeRs(kube::Error::TlsRequired)));
         let k8s_store = K8sStore::new(Arc::new(k8s_client), TEST_NAMESPACE.to_string());
 
         let id = k8s_store.set_opamp_data(

agent-control/src/opamp/remote_config/validators/signature/certificate_fetcher.rs

@@ -79,7 +79,7 @@ mod tests {
     use crate::http::config::ProxyConfig;
     use crate::http::tls::install_rustls_default_crypto_provider;
     use crate::opamp::remote_config::validators::signature::certificate_store::tests::TestSigner;
-    use crate::utils::tests::retry;
+    use crate::utils::retry::retry;
     use assert_matches::assert_matches;
 
     const DEFAULT_CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
@@ -108,13 +108,12 @@ mod tests {
                         CertificateFetcher::Https(Url::parse(self.url).unwrap(), client.clone())
                             .fetch()
                     {
-                        return Err(
-                            format!("fetching cert err '{}', case: '{}'", e, self.name).into()
-                        );
+                        return Err(format!("fetching cert err '{}', case: '{}'", e, self.name));
                     }
 
                     Ok(())
-                });
+                })
+                .unwrap();
             }
         }
         let test_cases = vec![