feat(instrumentation): add spans and agent_id context (#1143)

Author: paologallinaharbur

Cargo.lock

@@ -27,7 +27,7 @@ use crossbeam::select;
 use opamp_client::StartedClient;
 use std::sync::Arc;
 use std::time::SystemTime;
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info, instrument, warn};
 
 pub struct AgentControl<S, O, HR, SL, DV>
 where
@@ -141,6 +141,7 @@ where
     //  * Recreate the Final Agent using the Agent Type and the latest persisted config
     //  * Build a Stopped Sub Agent
     //  * Run the Sub Agent and add it to the Running Sub Agents
+    #[instrument(skip_all)]
     pub(super) fn recreate_sub_agent(
         &self,
         agent_identity: &AgentIdentity,
@@ -149,12 +150,10 @@ where
         >,
     ) -> Result<(), AgentError> {
         running_sub_agents.stop_and_remove(&agent_identity.id)?;
-
         self.build_and_run_sub_agent(agent_identity, running_sub_agents)
     }
 
-    // build_sub_agents returns a collection of started sub agents given the corresponding
-    // EffectiveAgents
+    // build_sub_agents returns a collection of started sub agents
     fn build_and_run_sub_agents(
         &self,
         sub_agents: &SubAgentsMap,
@@ -250,6 +249,7 @@ where
         }
     }
 
+    #[instrument(skip_all)]
     // apply an agent control remote config
     pub(super) fn apply_remote_agent_control_config(
         &self,

agent-control/src/agent_control/agent_control.rs

@@ -10,7 +10,6 @@ use crate::opamp::instance_id;
 use crate::opamp::remote_config::RemoteConfigError;
 use crate::sub_agent::effective_agents_assembler::EffectiveAgentsAssemblerError;
 use crate::sub_agent::error::{SubAgentBuilderError, SubAgentCollectionError, SubAgentError};
-use crate::utils::thread_context::ThreadContextStopperError;
 use crate::values::yaml_config::YAMLConfigError;
 use crate::values::yaml_config_repository::YAMLConfigRepositoryError;
 use fs::file_reader::FileReaderError;

agent-control/src/agent_control/error.rs

@@ -1,6 +1,3 @@
-use opamp_client::StartedClient;
-use tracing::{error, info};
-
 use crate::agent_control::config_storer::loader_storer::{
     AgentControlDynamicConfigDeleter, AgentControlDynamicConfigLoader,
     AgentControlDynamicConfigStorer,
@@ -13,6 +10,8 @@ use crate::{
     opamp::{hash_repository::HashRepository, remote_config::RemoteConfig},
     sub_agent::{collection::StartedSubAgents, NotStartedSubAgent, SubAgentBuilder},
 };
+use opamp_client::StartedClient;
+use tracing::{error, info};
 
 impl<S, O, HR, SL, DV> AgentControl<S, O, HR, SL, DV>
 where
@@ -38,7 +37,7 @@ where
             unreachable!("got remote config without OpAMP being enabled");
         };
 
-        info!(agent_id=%remote_config.agent_id, "Applying remote config");
+        info!("Applying remote config");
         OpampRemoteConfigStatus::Applying.report(opamp_client, &remote_config.hash)?;
 
         match self.apply_remote_agent_control_config(&remote_config, sub_agents) {

agent-control/src/agent_control/event_handler/opamp/remote_config.rs

@@ -19,8 +19,6 @@ const DEFAULT_BATCH_SCHEDULED_DELAY: Duration = Duration::from_secs(30);
 const TRACES_SUFFIX: &str = "/v1/traces";
 /// Metrics suffix for the OpenTelemetry endpoint
 const METRICS_SUFFIX: &str = "/v1/metrics";
-/// Logs suffix for the OpenTelemetry endpoint
-const LOGS_SUFFIX: &str = "/v1/logs";
 
 /// Represents the OpenTelemetry configuration
 #[derive(Debug, Deserialize, Serialize, PartialEq, Clone)]

agent-control/src/instrumentation/config/otel.rs

@@ -18,7 +18,7 @@ use crate::utils::thread_context::{NotStartedThreadContext, StartedThreadContext
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
 use kube::api::TypeMeta;
 use std::{sync::Arc, time::Duration};
-use tracing::{debug, error, info, trace, warn};
+use tracing::{debug, error, info, instrument, trace, warn};
 
 const THREAD_NAME: &str = "Garbage collector";
 
@@ -106,6 +106,7 @@ where
     /// Garbage collect all resources managed by the SA associated to removed sub-agents.
     /// Collection is stateful, only happens when the list of active agents has changed from
     /// the previous execution.
+    #[instrument(skip_all, name = "garbage_collector_collect")]
     pub fn collect(&mut self) -> Result<(), GarbageCollectorK8sError> {
         // check if current active agents differs from previous execution.
         if !self.update_active_config()? {

agent-control/src/k8s/garbage_collector.rs

@@ -92,14 +92,13 @@ where
     ) -> Result<Self::Client, OpAMPClientBuilderError> {
         let http_client = self.http_client_builder.build()?;
         let effective_config_loader = self.effective_config_loader_builder.build(agent_id.clone());
-        let callbacks =
-            AgentCallbacks::new(agent_id.clone(), opamp_publisher, effective_config_loader);
+        let callbacks = AgentCallbacks::new(agent_id, opamp_publisher, effective_config_loader);
         let not_started_client = NotStartedHttpClient::new(http_client, callbacks, start_settings)?;
         let mut not_started_client = not_started_client.with_interval(self.poll_interval);
         if self.disable_startup_check {
             not_started_client = not_started_client.with_startup_check_disabled();
         }
-        info!(%agent_id,"OpAMP client started");
+        info!("OpAMP client started");
         Ok(not_started_client.start()?)
     }
 }

agent-control/src/opamp/client_builder.rs

@@ -42,16 +42,15 @@ where
 {
     fn get(&self, agent_id: &AgentID) -> Result<InstanceID, GetterError> {
         let storer = self.storer.lock().expect("failed to acquire the lock");
-        debug!(%agent_id, "retrieving instance id");
+        debug!("retrieving instance id");
         let data = storer.get(agent_id)?;
 
         match data {
             None => {
-                debug!(%agent_id, "storer returned no data");
+                debug!("storer returned no data");
             }
             Some(d) if d.identifiers == self.identifiers => return Ok(d.instance_id),
             Some(d) => debug!(
-                %agent_id,
                 "stored data had different identifiers {:?}!={:?}",
                 d.identifiers, self.identifiers
             ),
@@ -62,7 +61,7 @@ where
             identifiers: self.identifiers.clone(),
         };
 
-        debug!(%agent_id, "persisting instance id {}", new_data.instance_id);
+        debug!("persisting instance id {}", new_data.instance_id);
         storer.set(agent_id, &new_data)?;
 
         Ok(new_data.instance_id)

agent-control/src/opamp/instance_id/getter.rs

@@ -3,7 +3,6 @@ use crate::opamp::instance_id::on_host::storer::StorerError;
 use resource_detection::cloud::aws::detector::{
     AWSDetector, AWS_IPV4_METADATA_ENDPOINT, AWS_IPV4_METADATA_TOKEN_ENDPOINT,
 };
-use resource_detection::cloud::aws::http_client::AWSHttpClient;
 use resource_detection::cloud::azure::detector::{AzureDetector, AZURE_IPV4_METADATA_ENDPOINT};
 use resource_detection::cloud::cloud_id::detector::CloudIdDetector;
 use resource_detection::cloud::gcp::detector::{GCPDetector, GCP_IPV4_METADATA_ENDPOINT};

agent-control/src/opamp/instance_id/on_host/getter.rs

@@ -11,6 +11,7 @@ impl<S> StartedSubAgents<S>
 where
     S: StartedSubAgent,
 {
+    #[tracing::instrument(skip_all)]
     pub(crate) fn stop_and_remove(
         &mut self,
         agent_id: &AgentID,
@@ -22,8 +23,8 @@ where
                     agent_id.to_string(),
                 ))?;
 
-        info!(%agent_id, "Stopping sub agent");
-        Self::stop_sub_agent(sub_agent, agent_id);
+        info!("Stopping sub agent");
+        Self::stop_sub_agent(sub_agent);
 
         Ok(())
     }
@@ -34,16 +35,17 @@ where
     }
 
     pub(crate) fn stop(self) {
-        self.0.into_iter().for_each(|(agent_id, sub_agent)| {
-            info!(%agent_id, "Stopping sub agent");
-            Self::stop_sub_agent(sub_agent, &agent_id);
+        self.0.into_iter().for_each(|(_, sub_agent)| {
+            info!("Stopping sub agent");
+            Self::stop_sub_agent(sub_agent);
         })
     }
 
-    fn stop_sub_agent(sub_agent: S, agent_id: &AgentID) {
+    #[tracing::instrument(skip_all)]
+    fn stop_sub_agent(sub_agent: S) {
         let _ = sub_agent
             .stop()
-            .inspect_err(|err| error!(%agent_id, %err, "Error stopping sub agent"));
+            .inspect_err(|err| error!(%err, "Error stopping sub agent"));
     }
 }