Add onhost remote config integration test (#708)

alvarocabanas · web-flow · commit 371e94856caf · 2024-06-25T09:48:35.000Z
diff --git a/super-agent/src/super_agent/defaults.rs b/super-agent/src/super_agent/defaults.rs
@@ -128,7 +128,7 @@ generate_const_getter!(STDOUT_LOG_PREFIX, "stdout.log");
 generate_const_getter!(STDERR_LOG_PREFIX, "stderr.log");
 
 #[cfg(debug_assertions)]
-pub(super) fn set_local_dir(path: &Path) {
+pub fn set_local_dir(path: &Path) {
     // The Err variant in `set` just contains the value we attempted to set,
     // so we can just ignore the Result
     _ = SUPER_AGENT_LOCAL_DATA_DIR_STATIC
@@ -139,7 +139,7 @@ pub(super) fn set_local_dir(path: &Path) {
 }
 
 #[cfg(debug_assertions)]
-pub(super) fn set_remote_dir(path: &Path) {
+pub fn set_remote_dir(path: &Path) {
     _ = SUPER_AGENT_DATA_DIR_STATIC
         .set(path.to_string_lossy().to_string())
         .inspect_err(|_| {
diff --git a/super-agent/src/super_agent/run.rs b/super-agent/src/super_agent/run.rs
@@ -21,10 +21,11 @@ use std::sync::Arc;
 use tokio::runtime::Runtime;
 use tracing::{debug, error, info, trace};
 
+// k8s and on_host need to be public to allow integration tests to access the fn run_super_agent.
 #[cfg(feature = "k8s")]
-mod k8s;
+pub mod k8s;
 #[cfg(feature = "onhost")]
-mod on_host;
+pub mod on_host;
 
 /// Structures for running the super-agent provided by CLI inputs
 pub struct SuperAgentRunConfig {
diff --git a/super-agent/tests/common/mod.rs b/super-agent/tests/common/mod.rs
@@ -3,3 +3,5 @@ pub(super) mod health;
 pub(super) mod opamp;
 /// Includes helpers to handle the _async_ code execution in non-tokio-tests.
 pub(super) mod runtime;
+
+pub(super) mod retry;
diff --git a/super-agent/tests/common/retry.rs b/super-agent/tests/common/retry.rs
diff --git a/super-agent/tests/k8s/no_opamp.rs b/super-agent/tests/k8s/no_opamp.rs
@@ -1,7 +1,7 @@
-use crate::common::runtime::block_on;
+use crate::common::{retry::retry, runtime::block_on};
 
 use super::tools::{
-    k8s_api::check_deployments_exist, k8s_env::K8sEnv, retry,
+    k8s_api::check_deployments_exist, k8s_env::K8sEnv,
     super_agent::start_super_agent_with_testdata_config,
 };
 use std::time::Duration;
diff --git a/super-agent/tests/k8s/opamp.rs b/super-agent/tests/k8s/opamp.rs
@@ -1,14 +1,14 @@
 use crate::common::{
     health::check_latest_health_status_was_healthy,
     opamp::{ConfigResponse, FakeServer},
+    retry::retry,
     runtime::block_on,
 };
 
 use super::tools::{
     instance_id,
     k8s_api::{check_deployments_exist, check_helmrelease_spec_values},
     k8s_env::K8sEnv,
-    retry,
     super_agent::{
         start_super_agent_with_testdata_config, wait_until_super_agent_with_opamp_is_started,
     },
diff --git a/super-agent/tests/k8s/tools/mod.rs b/super-agent/tests/k8s/tools/mod.rs
@@ -5,10 +5,7 @@ pub mod instance_id;
 pub mod k8s_api;
 /// Provides a k8s testing environment.
 pub mod k8s_env;
-mod retry;
 /// Contains helpers to execute the super-agent binary (compiled with the k8s feature)
 /// and specific initial configuration. Any helper receiving a `folder_name` assumes that the folder exists
 /// in the path `tests/k8s/data/`.
 pub mod super_agent;
-
-pub use retry::retry;
diff --git a/super-agent/tests/k8s/tools/super_agent.rs b/super-agent/tests/k8s/tools/super_agent.rs
@@ -12,9 +12,9 @@ use std::time::Duration;
 use std::{collections::BTreeMap, path::PathBuf};
 use std::{fs::File, io::Write};
 
-use crate::common::runtime::block_on;
+use crate::common::{retry::retry, runtime::block_on};
 
-use super::{k8s_api::check_config_map_exist, retry};
+use super::k8s_api::check_config_map_exist;
 
 pub const TEST_CLUSTER_NAME: &str = "minikube";
 
diff --git a/super-agent/tests/on_host/mod.rs b/super-agent/tests/on_host/mod.rs
@@ -5,6 +5,8 @@ mod consts;
 mod id;
 mod logging;
 mod opamp_auth;
+mod remote_config;
 mod restarting_processes;
 mod supervisor;
+mod tools;
 mod values_repository;
diff --git a/super-agent/tests/on_host/remote_config.rs b/super-agent/tests/on_host/remote_config.rs
@@ -0,0 +1,92 @@
+use crate::common::opamp::{ConfigResponse, FakeServer};
+use crate::common::retry::retry;
+use crate::on_host::tools::instance_id::get_instance_id;
+use crate::on_host::tools::super_agent::start_super_agent_with_custom_config;
+use newrelic_super_agent::super_agent::config::{AgentID, SuperAgentDynamicConfig};
+use newrelic_super_agent::super_agent::defaults::{set_local_dir, set_remote_dir};
+use std::error::Error;
+use std::fs::File;
+use std::io::Write;
+use std::thread;
+use std::time::Duration;
+use tempfile::tempdir;
+use url::Url;
+
+#[cfg(unix)]
+#[test]
+fn onhost_opamp_superagent_configuration_change() {
+    // Given a super-agent without agents and opamp configured.
+    let mut server = FakeServer::start_new();
+    let server_endpoint = Url::try_from(server.endpoint().as_str()).unwrap();
+
+    let local_dir = tempdir().expect("failed to create local temp dir");
+    let remote_dir = tempdir().expect("failed to create remote temp dir");
+
+    set_local_dir(local_dir.path());
+    set_remote_dir(remote_dir.path());
+
+    let config_file_path = local_dir.path().join("config.yaml");
+    let mut local_file =
+        File::create(config_file_path.clone()).expect("failed to create local config file");
+    let local_config = r#"
+host_id: integration-test
+fleet_id: integration
+opamp:
+  endpoint: http://127.0.0.1/v1/opamp
+agents: {}
+"#;
+    write!(local_file, "{}", local_config).unwrap();
+
+    // We won't join and wait for the thread to finish because we want the super_agent to exit
+    // if our assertions were not ok.
+    let _super_agent_join = thread::spawn(move || {
+        start_super_agent_with_custom_config(config_file_path.as_path(), server_endpoint)
+    });
+
+    let super_agent_instance_id = get_instance_id(&AgentID::new_super_agent_id());
+
+    // When a new config with two agents is received from OpAMP
+    server.set_config_response(
+        super_agent_instance_id,
+        ConfigResponse::from(
+            r#"
+agents:
+  nr-infra-agent:
+    agent_type: "newrelic/com.newrelic.infrastructure_agent:0.1.2"
+  otel-collector:
+    agent_type: "newrelic/io.opentelemetry.collector:0.0.1"
+"#,
+        ),
+    );
+
+    // Then the config should be updated in the remote filesystem.
+    let expected_config = r#"agents:
+  nr-infra-agent:
+    agent_type: newrelic/com.newrelic.infrastructure_agent:0.1.2
+  otel-collector:
+    agent_type: newrelic/io.opentelemetry.collector:0.0.1
+"#;
+    let expected_config_parsed =
+        serde_yaml::from_str::<SuperAgentDynamicConfig>(expected_config).unwrap();
+
+    retry(20, Duration::from_secs(5), || {
+        || -> Result<(), Box<dyn Error>> {
+            let remote_file = remote_dir.path().join("config.yaml");
+            let content =
+                std::fs::read_to_string(remote_file.as_path()).unwrap_or("agents:".to_string());
+            let content_parsed =
+                serde_yaml::from_str::<SuperAgentDynamicConfig>(content.as_str()).unwrap();
+            if content_parsed != expected_config_parsed {
+                return Err(format!(
+                    "Super agent config not as expected, Expected: {:?}, Found: {:?}",
+                    expected_config, content,
+                )
+                .into());
+            }
+            Ok(())
+        }()
+    });
+
+    // TODO: Then OpAMP receives applied (& applying?) AgentToServer (check state on the server).
+    // TODO: Then the two agent processes are running (we should create custom agent_types for custom binary).
+}
diff --git a/super-agent/tests/on_host/tools.rs b/super-agent/tests/on_host/tools.rs
@@ -0,0 +1,2 @@
+pub mod instance_id;
+pub mod super_agent;
diff --git a/super-agent/tests/on_host/tools/instance_id.rs b/super-agent/tests/on_host/tools/instance_id.rs
@@ -0,0 +1,96 @@
+use fs::directory_manager::{DirectoryManager, DirectoryManagerFs};
+use fs::file_reader::FileReader;
+use fs::writer_file::FileWriter;
+use fs::LocalFile;
+use newrelic_super_agent::opamp::instance_id::getter::{
+    DataStored, InstanceIDGetter, InstanceIDWithIdentifiersGetter,
+};
+use newrelic_super_agent::opamp::instance_id::storer::InstanceIDStorer;
+use newrelic_super_agent::opamp::instance_id::{IdentifiersProvider, InstanceID};
+use newrelic_super_agent::super_agent::config::AgentID;
+use newrelic_super_agent::super_agent::defaults::{
+    REMOTE_AGENT_DATA_DIR, SUPER_AGENT_IDENTIFIERS_PATH,
+};
+use std::path::PathBuf;
+use tracing::debug;
+
+pub struct Storer<F = LocalFile, D = DirectoryManagerFs>
+where
+    D: DirectoryManager,
+    F: FileWriter + FileReader,
+{
+    file_rw: F,
+    _dir_manager: D,
+}
+
+fn get_instance_id_path(agent_id: &AgentID) -> PathBuf {
+    if agent_id.is_super_agent_id() {
+        PathBuf::from(SUPER_AGENT_IDENTIFIERS_PATH())
+    } else {
+        PathBuf::from(format!(
+            "{}/{}/identifiers.yaml",
+            REMOTE_AGENT_DATA_DIR(),
+            agent_id.get()
+        ))
+    }
+}
+
+impl<F, D> InstanceIDStorer for Storer<F, D>
+where
+    D: DirectoryManager,
+    F: FileWriter + FileReader,
+{
+    fn set(
+        &self,
+        _agent_id: &AgentID,
+        _ds: &DataStored,
+    ) -> Result<(), newrelic_super_agent::opamp::instance_id::StorerError> {
+        Ok(())
+    }
+
+    fn get(
+        &self,
+        agent_id: &AgentID,
+    ) -> Result<Option<DataStored>, newrelic_super_agent::opamp::instance_id::StorerError> {
+        self.read_contents(agent_id)
+    }
+}
+
+impl<F, D> Storer<F, D>
+where
+    D: DirectoryManager,
+    F: FileWriter + FileReader,
+{
+    fn read_contents(
+        &self,
+        agent_id: &AgentID,
+    ) -> Result<Option<DataStored>, newrelic_super_agent::opamp::instance_id::StorerError> {
+        let dest_path = get_instance_id_path(agent_id);
+        let file_str = match self.file_rw.read(dest_path.as_path()) {
+            Ok(s) => s,
+            Err(e) => {
+                debug!("error reading file for agent {}: {}", agent_id, e);
+                return Ok(None);
+            }
+        };
+        match serde_yaml::from_str(&file_str) {
+            Ok(ds) => Ok(Some(ds)),
+            Err(e) => {
+                debug!("error deserializing data for agent {}: {}", agent_id, e);
+                Ok(None)
+            }
+        }
+    }
+}
+
+pub fn get_instance_id(agent_id: &AgentID) -> InstanceID {
+    let identifiers_provider = IdentifiersProvider::default()
+        .with_host_id("integration-test".to_string())
+        .with_fleet_id("integration".to_string());
+    let identifiers = identifiers_provider.provide().unwrap_or_default();
+
+    let instance_id_getter =
+        InstanceIDWithIdentifiersGetter::default().with_identifiers(identifiers);
+
+    instance_id_getter.get(agent_id).unwrap()
+}
diff --git a/super-agent/tests/on_host/tools/super_agent.rs b/super-agent/tests/on_host/tools/super_agent.rs
@@ -0,0 +1,51 @@
+use http::HeaderMap;
+use newrelic_super_agent::event::channel::pub_sub;
+use newrelic_super_agent::event::SuperAgentEvent;
+use newrelic_super_agent::opamp::auth::token_retriever::{TokenRetrieverImpl, TokenRetrieverNoop};
+use newrelic_super_agent::opamp::client_builder::DefaultOpAMPClientBuilder;
+use newrelic_super_agent::opamp::http::builder::UreqHttpClientBuilder;
+use newrelic_super_agent::super_agent::config::OpAMPClientConfig;
+use newrelic_super_agent::super_agent::config_storer::store::SuperAgentConfigStore;
+use newrelic_super_agent::super_agent::run::on_host::run_super_agent;
+use std::path::Path;
+use std::sync::Arc;
+use url::Url;
+
+/// Starts the super-agent through [start_super_agent] after setting up the corresponding configuration file
+/// and config map according to the provided `folder_name` and the provided `file_names`.
+pub fn start_super_agent_with_custom_config(config_path: &Path, opamp_endpoint: Url) {
+    // Create the Tokio runtime
+    let runtime = Arc::new(
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap(),
+    );
+
+    let token_retriever = Arc::new(TokenRetrieverImpl::Noop(TokenRetrieverNoop {}));
+
+    let opamp_config = OpAMPClientConfig {
+        endpoint: opamp_endpoint,
+        headers: HeaderMap::default(),
+        auth_config: None,
+    };
+
+    let http_builder = UreqHttpClientBuilder::new(opamp_config.clone(), token_retriever);
+    let builder = Some(DefaultOpAMPClientBuilder::new(
+        opamp_config.clone(),
+        http_builder,
+    ));
+
+    let (_application_event_publisher, application_event_consumer) = pub_sub();
+    let (super_agent_publisher, _super_agent_consumer) = pub_sub::<SuperAgentEvent>();
+
+    let config_storer = SuperAgentConfigStore::new(config_path);
+
+    _ = run_super_agent(
+        runtime.clone(),
+        config_storer,
+        application_event_consumer,
+        builder,
+        super_agent_publisher,
+    );
+}
