From 39b21c357956a7a0e8c786fead0f2be609e6f488 Mon Sep 17 00:00:00 2001 From: Travis Thieman Date: Mon, 27 Jul 2026 12:39:06 -0400 Subject: [PATCH 1/6] feat(agent-data-plane): emit liveness signals Add a periodic liveness source that emits Core Agent-compatible running metrics and service checks through the existing metric and service-check pipelines.\n\nKeep the liveness baseline pipelines available whenever ADP is enabled, and fetch/cache Core Agent system host tags without suppressing signals when tag retrieval fails. --- bin/agent-data-plane/src/cli/run.rs | 26 +- .../src/components/liveness/mod.rs | 230 ++++++++++++++++++ bin/agent-data-plane/src/components/mod.rs | 1 + bin/agent-data-plane/src/config.rs | 39 +-- 4 files changed, 273 insertions(+), 23 deletions(-) create mode 100644 bin/agent-data-plane/src/components/liveness/mod.rs diff --git a/bin/agent-data-plane/src/cli/run.rs b/bin/agent-data-plane/src/cli/run.rs index 697d5d274e..481e6e9742 100644 --- a/bin/agent-data-plane/src/cli/run.rs +++ b/bin/agent-data-plane/src/cli/run.rs @@ -45,8 +45,8 @@ use crate::{ apm_onboarding::ApmOnboardingConfiguration, dogstatsd_post_aggregate_filter::DogStatsDPostAggregateFilterConfiguration, dogstatsd_prefix_filter::DogStatsDPrefixFilterConfiguration, host_tags::HostTagsConfiguration, - ottl_filter_processor::OttlFilterConfiguration, ottl_transform_processor::OttlTransformConfiguration, - tag_filterlist::TagFilterlistConfiguration, + liveness::LivenessConfiguration, ottl_filter_processor::OttlFilterConfiguration, + ottl_transform_processor::OttlTransformConfiguration, tag_filterlist::TagFilterlistConfiguration, }, internal::{ create_internal_supervisor, logging::LoggingConfigurationTranslator, remote_agent::RemoteAgentBootstrap, @@ -380,8 +380,9 @@ async fn create_topology( let mut control_surfaces = TopologyControlSurfaces::default(); - // If no data pipelines are enabled, then there's nothing for us to do. - if !dp_config.data_pipelines_enabled() { + // A running data plane always emits liveness signals, even if every data pipeline is disabled. Standalone mode + // can still invoke this setup while disabled, where no topology is useful. + if !dp_config.enabled() && !dp_config.data_pipelines_enabled() { return Err(generic_error!("No data pipelines are enabled. Exiting.")); } @@ -426,6 +427,10 @@ async fn create_topology( add_baseline_traces_pipeline_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?; } + if dp_config.enabled() { + add_liveness_source_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?; + } + // Now we move on to our actual data pipelines. if dp_config.checks().enabled() { add_checks_pipeline_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?; @@ -445,6 +450,19 @@ async fn create_topology( Ok((blueprint, control_surfaces)) } +async fn add_liveness_source_to_blueprint( + blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, env_provider: &ADPEnvironmentProvider, +) -> Result<(), GenericError> { + let liveness_config = LivenessConfiguration::from_environment_provider(config, env_provider).await?; + + blueprint + .add_source("liveness_in", liveness_config)? + .connect_components("liveness_in.metrics", "metrics_enrich")? + .connect_components("liveness_in.service_checks", "dd_service_checks_encode")?; + + Ok(()) +} + async fn add_checks_pipeline_to_blueprint( blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, env_provider: &ADPEnvironmentProvider, ) -> Result<(), GenericError> { diff --git a/bin/agent-data-plane/src/components/liveness/mod.rs b/bin/agent-data-plane/src/components/liveness/mod.rs new file mode 100644 index 0000000000..cca8fc630e --- /dev/null +++ b/bin/agent-data-plane/src/components/liveness/mod.rs @@ -0,0 +1,230 @@ +use std::{sync::LazyLock, time::Duration}; + +use async_trait::async_trait; +use datadog_agent_commons::ipc::client::RemoteAgentClient; +use saluki_context::{ + tags::{Tag, TagSet}, + Context, +}; +use saluki_core::{ + accounting::{MemoryBounds, MemoryBoundsBuilder}, + components::{sources::*, ComponentContext}, + data_model::event::{ + metric::Metric, + service_check::{CheckStatus, ServiceCheck}, + Event, EventType, + }, + topology::OutputDefinition, +}; +use saluki_env::{EnvironmentProvider as _, HostProvider as _}; +use saluki_error::{ErrorContext as _, GenericError}; +use stringtheory::MetaString; +use tokio::{pin, select, time::interval}; +use tracing::{debug, warn}; + +use crate::internal::env::ADPEnvironmentProvider; + +const LIVENESS_INTERVAL: Duration = Duration::from_secs(15); +const RUNNING_METRIC_NAME: &str = "datadog.agent_data_plane.running"; +const UP_SERVICE_CHECK_NAME: &str = "datadog.agent_data_plane.up"; + +/// Periodically emits Agent Data Plane liveness signals. +pub struct LivenessConfiguration { + hostname: MetaString, + version: MetaString, + client: Option, +} + +impl LivenessConfiguration { + /// Creates a liveness source configuration using the configured environment and Agent IPC client. + pub async fn from_environment_provider( + config: &saluki_config::GenericConfiguration, env_provider: &ADPEnvironmentProvider, + ) -> Result { + let hostname = env_provider + .host() + .get_hostname() + .await + .error_context("Failed to get hostname for liveness source.")?; + let client = match RemoteAgentClient::from_configuration(config).await { + Ok(client) => Some(client), + Err(error) => { + warn!(error = %error, "Failed to create Agent IPC client for liveness host tags; continuing without tags."); + None + } + }; + + Ok(Self { + hostname: hostname.into(), + version: saluki_metadata::get_app_details().version().raw().into(), + client, + }) + } + + #[cfg(test)] + fn for_tests(hostname: &str, version: &str) -> Self { + Self { + hostname: hostname.into(), + version: version.into(), + client: None, + } + } +} + +#[async_trait] +impl SourceBuilder for LivenessConfiguration { + async fn build(&self, _context: ComponentContext) -> Result, GenericError> { + Ok(Box::new(Liveness { + hostname: self.hostname.clone(), + version: self.version.clone(), + client: self.client.clone(), + system_tags: None, + })) + } + + fn outputs(&self) -> &[OutputDefinition] { + static OUTPUTS: LazyLock>> = LazyLock::new(|| { + vec![ + OutputDefinition::named_output("metrics", EventType::Metric), + OutputDefinition::named_output("service_checks", EventType::ServiceCheck), + ] + }); + &OUTPUTS + } +} + +impl MemoryBounds for LivenessConfiguration { + fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) { + builder.minimum().with_single_value::("liveness source"); + } +} + +struct Liveness { + hostname: MetaString, + version: MetaString, + client: Option, + system_tags: Option, +} + +impl Liveness { + async fn system_tags(&mut self) -> TagSet { + let Some(client) = self.client.as_ref() else { + return self.system_tags.clone().unwrap_or_default(); + }; + + match client.get_host_tags().await { + Ok(response) => { + let tags = response.into_inner().system.into_iter().map(Tag::from).collect(); + self.system_tags = Some(tags); + } + Err(error) => { + warn!(error = %error, "Failed to fetch liveness host tags; continuing with cached or empty tags."); + } + } + + self.system_tags.clone().unwrap_or_default() + } +} + +#[async_trait] +impl Source for Liveness { + async fn run(mut self: Box, mut context: SourceContext) -> Result<(), GenericError> { + let global_shutdown = context.take_shutdown_handle(); + pin!(global_shutdown); + + let mut health = context.take_health_handle(); + let mut tick_interval = interval(LIVENESS_INTERVAL); + + health.mark_ready(); + debug!("Liveness source started."); + + loop { + select! { + _ = &mut global_shutdown => { + debug!("Received shutdown signal."); + break; + }, + _ = health.live() => continue, + _ = tick_interval.tick() => { + let system_tags = self.system_tags().await; + let (metric, service_check) = create_liveness_events(self.hostname.clone(), self.version.clone(), &system_tags); + + if let Err(error) = context.dispatcher().dispatch_one_named("metrics", metric).await { + warn!(error = %error, "Failed to dispatch liveness metric."); + } + if let Err(error) = context.dispatcher().dispatch_one_named("service_checks", service_check).await { + warn!(error = %error, "Failed to dispatch liveness service check."); + } + }, + } + } + + debug!("Liveness source stopped."); + Ok(()) + } +} + +fn create_liveness_events(hostname: MetaString, version: MetaString, system_tags: &TagSet) -> (Event, Event) { + let mut metric_tags = system_tags.clone(); + metric_tags.insert_tag(format!("version:{version}")); + let metric_context = Context::from_static_name(RUNNING_METRIC_NAME) + .with_host(hostname.clone()) + .with_tags(metric_tags); + let metric = Event::Metric(Metric::gauge(metric_context, 1.0)); + + let service_check = Event::ServiceCheck( + ServiceCheck::new(UP_SERVICE_CHECK_NAME, CheckStatus::Ok) + .with_hostname(hostname) + .with_tags(system_tags.clone()), + ); + + (metric, service_check) +} + +#[cfg(test)] +mod tests { + use saluki_context::tags::TagSet; + use saluki_core::{ + components::sources::SourceBuilder as _, + data_model::event::{metric::MetricValues, service_check::CheckStatus, Event, EventType}, + }; + + use super::*; + + #[test] + fn declares_named_metric_and_service_check_outputs() { + let configuration = LivenessConfiguration::for_tests("host-a", "1.2.3"); + let outputs = configuration.outputs(); + + assert_eq!(outputs.len(), 2); + assert_eq!(outputs[0].output_name(), Some("metrics")); + assert_eq!(outputs[0].data_ty(), EventType::Metric); + assert_eq!(outputs[1].output_name(), Some("service_checks")); + assert_eq!(outputs[1].data_ty(), EventType::ServiceCheck); + } + + #[test] + fn emits_core_agent_compatible_liveness_events() { + let system_tags = TagSet::from_iter(["system:linux".into(), "region:us-east-1".into()]); + let (metric, service_check) = create_liveness_events("host-a".into(), "1.2.3".into(), &system_tags); + + let Event::Metric(metric) = metric else { + panic!("expected metric event"); + }; + assert_eq!(metric.context().name(), RUNNING_METRIC_NAME); + assert_eq!(metric.context().host(), Some("host-a")); + assert!(metric.context().tags().has_tag("system:linux")); + assert!(metric.context().tags().has_tag("region:us-east-1")); + assert!(metric.context().tags().has_tag("version:1.2.3")); + assert_eq!(metric.values(), &MetricValues::gauge((0, 1.0))); + + let Event::ServiceCheck(service_check) = service_check else { + panic!("expected service check event"); + }; + assert_eq!(service_check.name(), UP_SERVICE_CHECK_NAME); + assert_eq!(service_check.status(), CheckStatus::Ok); + assert_eq!(service_check.hostname(), Some("host-a")); + assert!(service_check.tags().has_tag("system:linux")); + assert!(service_check.tags().has_tag("region:us-east-1")); + assert!(!service_check.tags().has_tag("version:1.2.3")); + } +} diff --git a/bin/agent-data-plane/src/components/mod.rs b/bin/agent-data-plane/src/components/mod.rs index af6500911b..81b41c8504 100644 --- a/bin/agent-data-plane/src/components/mod.rs +++ b/bin/agent-data-plane/src/components/mod.rs @@ -3,6 +3,7 @@ mod dogstatsd_filterlist; pub mod dogstatsd_post_aggregate_filter; pub mod dogstatsd_prefix_filter; pub mod host_tags; +pub mod liveness; pub mod ottl_filter_processor; pub mod ottl_transform_processor; pub mod tag_filterlist; diff --git a/bin/agent-data-plane/src/config.rs b/bin/agent-data-plane/src/config.rs index 72be9dac86..4984e18c6c 100644 --- a/bin/agent-data-plane/src/config.rs +++ b/bin/agent-data-plane/src/config.rs @@ -99,13 +99,14 @@ impl DataPlaneConfiguration { /// Returns `true` if the metrics pipeline is required. /// /// This indicates that the "baseline" metrics pipeline (aggregation, enrichment, encoding, forwarding) is required - /// by higher-level data pipelines, such as DogStatsD. + /// by liveness signals or higher-level data pipelines, such as DogStatsD. pub const fn metrics_pipeline_required(&self) -> bool { // We consider the metrics pipeline to be enabled if: // - Checks is enabled // - DogStatsD is enabled // - OTLP is enabled and not in proxy mode - self.checks().enabled() + self.enabled + || self.checks().enabled() || self.dogstatsd().enabled() || (self.otlp().enabled() && !self.otlp().proxy().enabled()) } @@ -131,10 +132,10 @@ impl DataPlaneConfiguration { /// Returns `true` if the service checks pipeline is required. /// - /// This indicates that the "baseline" service checks pipeline (encoding, forwarding) is required by higher-level - /// data pipelines, such as Checks or DogStatsD. + /// This indicates that the "baseline" service checks pipeline (encoding, forwarding) is required by liveness + /// signals or higher-level data pipelines, such as Checks or DogStatsD. pub const fn service_checks_pipeline_required(&self) -> bool { - self.checks().enabled() || self.dogstatsd().enabled() + self.enabled || self.checks().enabled() || self.dogstatsd().enabled() } /// Returns `true` if the traces pipeline is required. @@ -441,14 +442,14 @@ mod tests { assert!(dp.metrics_pipeline_required()); assert!(dp.logs_pipeline_required()); assert!(!dp.events_pipeline_required()); - assert!(!dp.service_checks_pipeline_required()); + assert!(dp.service_checks_pipeline_required()); assert!(dp.traces_pipeline_required()); } #[tokio::test] - async fn otlp_proxy_mode_proxying_all_signals_requires_no_baseline_pipelines() { - // With proxy mode enabled and traces still proxied to the Core Agent (the default), ADP handles no signals - // itself, so no baseline pipeline is required even though a data pipeline (OTLP) is enabled. + async fn otlp_proxy_mode_proxying_all_signals_requires_liveness_baseline_pipelines() { + // With proxy mode enabled and traces still proxied to the Core Agent (the default), ADP handles no OTLP + // signals itself, but it still needs the liveness metric and service-check baseline pipelines. let dp = dp_config_from(json!({ "data_plane": { "enabled": true, @@ -459,17 +460,17 @@ mod tests { .await; assert!(dp.data_pipelines_enabled()); - assert!(!dp.metrics_pipeline_required()); + assert!(dp.metrics_pipeline_required()); assert!(!dp.logs_pipeline_required()); assert!(!dp.events_pipeline_required()); - assert!(!dp.service_checks_pipeline_required()); + assert!(dp.service_checks_pipeline_required()); assert!(!dp.traces_pipeline_required()); } #[tokio::test] - async fn otlp_proxy_mode_with_local_traces_requires_traces_pipeline() { - // Proxy mode is enabled but trace proxying is turned off, so ADP must handle traces locally and the traces - // pipeline becomes required again while the other baseline pipelines stay off. + async fn otlp_proxy_mode_with_local_traces_requires_liveness_and_traces_pipelines() { + // Proxy mode is enabled but trace proxying is turned off, so ADP must handle traces locally. The liveness + // metric and service-check baselines remain required regardless of the OTLP routing. let dp = dp_config_from(json!({ "data_plane": { "enabled": true, @@ -483,15 +484,15 @@ mod tests { .await; assert!(dp.data_pipelines_enabled()); - assert!(!dp.metrics_pipeline_required()); + assert!(dp.metrics_pipeline_required()); assert!(!dp.logs_pipeline_required()); assert!(!dp.events_pipeline_required()); - assert!(!dp.service_checks_pipeline_required()); + assert!(dp.service_checks_pipeline_required()); assert!(dp.traces_pipeline_required()); } #[tokio::test] - async fn no_pipelines_enabled_requires_no_baseline_pipelines() { + async fn liveness_requires_metrics_and_service_checks_without_data_pipelines() { let dp = dp_config_from(json!({ "data_plane": { "enabled": true, @@ -503,10 +504,10 @@ mod tests { .await; assert!(!dp.data_pipelines_enabled()); - assert!(!dp.metrics_pipeline_required()); + assert!(dp.metrics_pipeline_required()); assert!(!dp.logs_pipeline_required()); assert!(!dp.events_pipeline_required()); - assert!(!dp.service_checks_pipeline_required()); + assert!(dp.service_checks_pipeline_required()); assert!(!dp.traces_pipeline_required()); } From 3dcda7b16fa628970ea70f4b68fc8b18ea62a558 Mon Sep 17 00:00:00 2001 From: Travis Thieman Date: Mon, 27 Jul 2026 12:52:45 -0400 Subject: [PATCH 2/6] fix(agent-data-plane): keep liveness independent of host tags Defer liveness IPC client construction until the source tick so host tag failures cannot block topology startup. Bound and cancel tag lookups and dispatches, cache successful system tags, and skip missed ticks.\n\nRoute liveness metrics directly to the Datadog metrics encoder so optional host-tag enrichment cannot suppress liveness signals. --- bin/agent-data-plane/src/cli/run.rs | 15 +- .../src/components/liveness/mod.rs | 151 ++++++++++++++---- 2 files changed, 137 insertions(+), 29 deletions(-) diff --git a/bin/agent-data-plane/src/cli/run.rs b/bin/agent-data-plane/src/cli/run.rs index 481e6e9742..d1ebcc0d4b 100644 --- a/bin/agent-data-plane/src/cli/run.rs +++ b/bin/agent-data-plane/src/cli/run.rs @@ -450,6 +450,8 @@ async fn create_topology( Ok((blueprint, control_surfaces)) } +const LIVENESS_METRICS_DESTINATION: &str = "dd_metrics_encode"; + async fn add_liveness_source_to_blueprint( blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, env_provider: &ADPEnvironmentProvider, ) -> Result<(), GenericError> { @@ -457,7 +459,7 @@ async fn add_liveness_source_to_blueprint( blueprint .add_source("liveness_in", liveness_config)? - .connect_components("liveness_in.metrics", "metrics_enrich")? + .connect_components("liveness_in.metrics", LIVENESS_METRICS_DESTINATION)? .connect_components("liveness_in.service_checks", "dd_service_checks_encode")?; Ok(()) @@ -909,3 +911,14 @@ fn write_sizing_guide(bounds: ComponentBounds) -> Result<(), GenericError> { Ok(()) } + +#[cfg(test)] +mod liveness_topology_tests { + use super::LIVENESS_METRICS_DESTINATION; + + #[test] + fn liveness_metrics_bypass_optional_host_tag_enrichment() { + assert_eq!(LIVENESS_METRICS_DESTINATION, "dd_metrics_encode"); + assert_ne!(LIVENESS_METRICS_DESTINATION, "metrics_enrich"); + } +} diff --git a/bin/agent-data-plane/src/components/liveness/mod.rs b/bin/agent-data-plane/src/components/liveness/mod.rs index cca8fc630e..e41a5dce26 100644 --- a/bin/agent-data-plane/src/components/liveness/mod.rs +++ b/bin/agent-data-plane/src/components/liveness/mod.rs @@ -1,7 +1,10 @@ -use std::{sync::LazyLock, time::Duration}; +use std::{ + sync::{Arc, LazyLock}, + time::Duration, +}; use async_trait::async_trait; -use datadog_agent_commons::ipc::client::RemoteAgentClient; +use datadog_agent_commons::ipc::{client::RemoteAgentClient, config::RemoteAgentClientConfiguration}; use saluki_context::{ tags::{Tag, TagSet}, Context, @@ -19,12 +22,17 @@ use saluki_core::{ use saluki_env::{EnvironmentProvider as _, HostProvider as _}; use saluki_error::{ErrorContext as _, GenericError}; use stringtheory::MetaString; -use tokio::{pin, select, time::interval}; +use tokio::{ + pin, select, + time::{interval, timeout, MissedTickBehavior}, +}; use tracing::{debug, warn}; use crate::internal::env::ADPEnvironmentProvider; const LIVENESS_INTERVAL: Duration = Duration::from_secs(15); +const LIVENESS_IPC_TIMEOUT: Duration = Duration::from_secs(2); +const LIVENESS_DISPATCH_TIMEOUT: Duration = Duration::from_secs(2); const RUNNING_METRIC_NAME: &str = "datadog.agent_data_plane.running"; const UP_SERVICE_CHECK_NAME: &str = "datadog.agent_data_plane.up"; @@ -32,11 +40,11 @@ const UP_SERVICE_CHECK_NAME: &str = "datadog.agent_data_plane.up"; pub struct LivenessConfiguration { hostname: MetaString, version: MetaString, - client: Option, + client_config: Option>, } impl LivenessConfiguration { - /// Creates a liveness source configuration using the configured environment and Agent IPC client. + /// Creates a liveness source configuration using the configured environment and Agent IPC client settings. pub async fn from_environment_provider( config: &saluki_config::GenericConfiguration, env_provider: &ADPEnvironmentProvider, ) -> Result { @@ -45,18 +53,16 @@ impl LivenessConfiguration { .get_hostname() .await .error_context("Failed to get hostname for liveness source.")?; - let client = match RemoteAgentClient::from_configuration(config).await { - Ok(client) => Some(client), - Err(error) => { - warn!(error = %error, "Failed to create Agent IPC client for liveness host tags; continuing without tags."); - None - } - }; + Self::from_hostname(config, hostname) + } + + fn from_hostname(config: &saluki_config::GenericConfiguration, hostname: String) -> Result { + let client_config = RemoteAgentClientConfiguration::from_configuration(config)?; Ok(Self { hostname: hostname.into(), version: saluki_metadata::get_app_details().version().raw().into(), - client, + client_config: Some(Arc::new(client_config)), }) } @@ -65,7 +71,7 @@ impl LivenessConfiguration { Self { hostname: hostname.into(), version: version.into(), - client: None, + client_config: None, } } } @@ -76,7 +82,8 @@ impl SourceBuilder for LivenessConfiguration { Ok(Box::new(Liveness { hostname: self.hostname.clone(), version: self.version.clone(), - client: self.client.clone(), + client_config: self.client_config.clone(), + client: None, system_tags: None, })) } @@ -101,24 +108,52 @@ impl MemoryBounds for LivenessConfiguration { struct Liveness { hostname: MetaString, version: MetaString, + client_config: Option>, client: Option, system_tags: Option, } impl Liveness { async fn system_tags(&mut self) -> TagSet { - let Some(client) = self.client.as_ref() else { - return self.system_tags.clone().unwrap_or_default(); - }; + if let Some(system_tags) = &self.system_tags { + return system_tags.clone(); + } - match client.get_host_tags().await { - Ok(response) => { + if self.client.is_none() { + let Some(client_config) = &self.client_config else { + return TagSet::default(); + }; + + match timeout( + LIVENESS_IPC_TIMEOUT, + RemoteAgentClient::from_client_configuration(client_config), + ) + .await + { + Ok(Ok(client)) => self.client = Some(client), + Ok(Err(error)) => { + warn!(error = %error, "Failed to create Agent IPC client for liveness host tags; continuing without tags."); + return TagSet::default(); + } + Err(_) => { + warn!("Timed out creating Agent IPC client for liveness host tags; continuing without tags."); + return TagSet::default(); + } + } + } + + let client = self.client.as_ref().expect("client is initialized above"); + match timeout(LIVENESS_IPC_TIMEOUT, client.get_host_tags()).await { + Ok(Ok(response)) => { let tags = response.into_inner().system.into_iter().map(Tag::from).collect(); self.system_tags = Some(tags); } - Err(error) => { + Ok(Err(error)) => { warn!(error = %error, "Failed to fetch liveness host tags; continuing with cached or empty tags."); } + Err(_) => { + warn!("Timed out fetching liveness host tags; continuing with cached or empty tags."); + } } self.system_tags.clone().unwrap_or_default() @@ -133,11 +168,12 @@ impl Source for Liveness { let mut health = context.take_health_handle(); let mut tick_interval = interval(LIVENESS_INTERVAL); + tick_interval.set_missed_tick_behavior(MissedTickBehavior::Skip); health.mark_ready(); debug!("Liveness source started."); - loop { + 'run: loop { select! { _ = &mut global_shutdown => { debug!("Received shutdown signal."); @@ -145,14 +181,53 @@ impl Source for Liveness { }, _ = health.live() => continue, _ = tick_interval.tick() => { - let system_tags = self.system_tags().await; - let (metric, service_check) = create_liveness_events(self.hostname.clone(), self.version.clone(), &system_tags); + let system_tags = { + let system_tags = self.system_tags(); + pin!(system_tags); + select! { + _ = &mut global_shutdown => { + debug!("Received shutdown signal."); + break 'run; + }, + _ = health.live() => continue 'run, + system_tags = &mut system_tags => system_tags, + } + }; + let (metric, service_check) = + create_liveness_events(self.hostname.clone(), self.version.clone(), &system_tags); - if let Err(error) = context.dispatcher().dispatch_one_named("metrics", metric).await { - warn!(error = %error, "Failed to dispatch liveness metric."); + let metric_dispatch = + timeout(LIVENESS_DISPATCH_TIMEOUT, context.dispatcher().dispatch_one_named("metrics", metric)); + pin!(metric_dispatch); + select! { + _ = &mut global_shutdown => { + debug!("Received shutdown signal."); + break 'run; + }, + _ = health.live() => continue 'run, + result = &mut metric_dispatch => match result { + Ok(Ok(())) => {}, + Ok(Err(error)) => warn!(error = %error, "Failed to dispatch liveness metric."), + Err(_) => warn!("Timed out dispatching liveness metric."), + }, } - if let Err(error) = context.dispatcher().dispatch_one_named("service_checks", service_check).await { - warn!(error = %error, "Failed to dispatch liveness service check."); + + let service_check_dispatch = timeout( + LIVENESS_DISPATCH_TIMEOUT, + context.dispatcher().dispatch_one_named("service_checks", service_check), + ); + pin!(service_check_dispatch); + select! { + _ = &mut global_shutdown => { + debug!("Received shutdown signal."); + break 'run; + }, + _ = health.live() => continue 'run, + result = &mut service_check_dispatch => match result { + Ok(Ok(())) => {}, + Ok(Err(error)) => warn!(error = %error, "Failed to dispatch liveness service check."), + Err(_) => warn!("Timed out dispatching liveness service check."), + }, } }, } @@ -182,11 +257,13 @@ fn create_liveness_events(hostname: MetaString, version: MetaString, system_tags #[cfg(test)] mod tests { + use saluki_config::ConfigurationLoader; use saluki_context::tags::TagSet; use saluki_core::{ components::sources::SourceBuilder as _, data_model::event::{metric::MetricValues, service_check::CheckStatus, Event, EventType}, }; + use serde_json::json; use super::*; @@ -202,6 +279,24 @@ mod tests { assert_eq!(outputs[1].data_ty(), EventType::ServiceCheck); } + #[tokio::test] + async fn source_configuration_parses_ipc_settings_without_creating_a_client() { + let (config, _) = ConfigurationLoader::for_tests( + Some(json!({ + "auth_token_file_path": "/path/that/does/not/exist", + "cmd_port": 9, + })), + None, + false, + ) + .await; + + let configuration = LivenessConfiguration::from_hostname(&config, "host-a".to_string()) + .expect("IPC settings should parse without connecting to the Agent"); + + assert!(configuration.client_config.is_some()); + } + #[test] fn emits_core_agent_compatible_liveness_events() { let system_tags = TagSet::from_iter(["system:linux".into(), "region:us-east-1".into()]); From cd938ee460a0447e2fb2a2c2a1ba1f2de22d3eeb Mon Sep 17 00:00:00 2001 From: Travis Thieman Date: Mon, 27 Jul 2026 14:28:21 -0400 Subject: [PATCH 3/6] fix(agent-data-plane): restore no-pipeline exit Keep liveness tied to a running topology rather than allowing it to create a liveness-only topology. Require the metric and service-check baselines whenever a data pipeline runs so proxy-only OTLP topologies can emit liveness, while no-pipeline configurations retain the established early exit. --- bin/agent-data-plane/src/cli/run.rs | 14 ++++++++------ bin/agent-data-plane/src/config.rs | 25 +++++++++---------------- 2 files changed, 17 insertions(+), 22 deletions(-) diff --git a/bin/agent-data-plane/src/cli/run.rs b/bin/agent-data-plane/src/cli/run.rs index d1ebcc0d4b..f29e866224 100644 --- a/bin/agent-data-plane/src/cli/run.rs +++ b/bin/agent-data-plane/src/cli/run.rs @@ -380,9 +380,9 @@ async fn create_topology( let mut control_surfaces = TopologyControlSurfaces::default(); - // A running data plane always emits liveness signals, even if every data pipeline is disabled. Standalone mode - // can still invoke this setup while disabled, where no topology is useful. - if !dp_config.enabled() && !dp_config.data_pipelines_enabled() { + // Without a data pipeline, there is no useful topology to run. In particular, liveness alone must not keep ADP + // alive. + if !dp_config.data_pipelines_enabled() { return Err(generic_error!("No data pipelines are enabled. Exiting.")); } @@ -393,8 +393,8 @@ async fn create_topology( // we additionally create metrics- and logs-specific components connected to that forwarder depending on which of // the baseline pipelines are required. // - // Notably, we _don't_ need either of these if all we're doing is running the OTLP pipeline in proxy mode, which - // is the only reason we're differentiating here. + // Liveness requires the metrics and service-check baseline pipelines for every running topology, including an + // OTLP proxy-only topology. if dp_config.metrics_pipeline_required() || dp_config.logs_pipeline_required() || dp_config.events_pipeline_required() @@ -427,7 +427,9 @@ async fn create_topology( add_baseline_traces_pipeline_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?; } - if dp_config.enabled() { + // Every running topology has at least one data pipeline and emits liveness through the metric and service-check + // baselines created above. + if dp_config.data_pipelines_enabled() { add_liveness_source_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?; } diff --git a/bin/agent-data-plane/src/config.rs b/bin/agent-data-plane/src/config.rs index 4984e18c6c..9be6c36f0a 100644 --- a/bin/agent-data-plane/src/config.rs +++ b/bin/agent-data-plane/src/config.rs @@ -98,17 +98,10 @@ impl DataPlaneConfiguration { /// Returns `true` if the metrics pipeline is required. /// - /// This indicates that the "baseline" metrics pipeline (aggregation, enrichment, encoding, forwarding) is required - /// by liveness signals or higher-level data pipelines, such as DogStatsD. + /// A running topology needs this pipeline whenever it has a data pipeline so liveness signals can be encoded and + /// forwarded, including when the only data pipeline is an OTLP proxy. pub const fn metrics_pipeline_required(&self) -> bool { - // We consider the metrics pipeline to be enabled if: - // - Checks is enabled - // - DogStatsD is enabled - // - OTLP is enabled and not in proxy mode - self.enabled - || self.checks().enabled() - || self.dogstatsd().enabled() - || (self.otlp().enabled() && !self.otlp().proxy().enabled()) + self.data_pipelines_enabled() } /// Returns `true` if the logs pipeline is required. @@ -132,10 +125,10 @@ impl DataPlaneConfiguration { /// Returns `true` if the service checks pipeline is required. /// - /// This indicates that the "baseline" service checks pipeline (encoding, forwarding) is required by liveness - /// signals or higher-level data pipelines, such as Checks or DogStatsD. + /// A running topology needs this pipeline whenever it has a data pipeline so liveness signals can be encoded and + /// forwarded, including when the only data pipeline is an OTLP proxy. pub const fn service_checks_pipeline_required(&self) -> bool { - self.enabled || self.checks().enabled() || self.dogstatsd().enabled() + self.data_pipelines_enabled() } /// Returns `true` if the traces pipeline is required. @@ -492,7 +485,7 @@ mod tests { } #[tokio::test] - async fn liveness_requires_metrics_and_service_checks_without_data_pipelines() { + async fn topology_without_data_pipelines_requires_no_baseline_pipelines() { let dp = dp_config_from(json!({ "data_plane": { "enabled": true, @@ -504,10 +497,10 @@ mod tests { .await; assert!(!dp.data_pipelines_enabled()); - assert!(dp.metrics_pipeline_required()); + assert!(!dp.metrics_pipeline_required()); assert!(!dp.logs_pipeline_required()); assert!(!dp.events_pipeline_required()); - assert!(dp.service_checks_pipeline_required()); + assert!(!dp.service_checks_pipeline_required()); assert!(!dp.traces_pipeline_required()); } From 41c9389e9a6bac8c372ba4cda80d2af1c8693e76 Mon Sep 17 00:00:00 2001 From: Travis Thieman Date: Tue, 28 Jul 2026 14:35:47 -0400 Subject: [PATCH 4/6] fix(agent-data-plane): route liveness through common enrichment Remove liveness-specific IPC host-tag retrieval, caching, retries, and dispatch timeouts. Route its metric through the common enrichment pipeline so host tags remain a pipeline concern, while service checks continue through the standard encoder path.\n\nKeep connected topologies provisioned for liveness, including OTLP proxy-only, but leave standalone OTLP proxy mode free of liveness and otherwise-unneeded output paths. --- bin/agent-data-plane/src/cli/run.rs | 24 +-- .../src/components/liveness/mod.rs | 196 +++--------------- bin/agent-data-plane/src/config.rs | 44 +++- 3 files changed, 79 insertions(+), 185 deletions(-) diff --git a/bin/agent-data-plane/src/cli/run.rs b/bin/agent-data-plane/src/cli/run.rs index f29e866224..c9100b4f69 100644 --- a/bin/agent-data-plane/src/cli/run.rs +++ b/bin/agent-data-plane/src/cli/run.rs @@ -393,8 +393,8 @@ async fn create_topology( // we additionally create metrics- and logs-specific components connected to that forwarder depending on which of // the baseline pipelines are required. // - // Liveness requires the metrics and service-check baseline pipelines for every running topology, including an - // OTLP proxy-only topology. + // Connected liveness requires the metrics and service-check baseline pipelines for every running topology, + // including an OTLP proxy-only topology. if dp_config.metrics_pipeline_required() || dp_config.logs_pipeline_required() || dp_config.events_pipeline_required() @@ -427,10 +427,10 @@ async fn create_topology( add_baseline_traces_pipeline_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?; } - // Every running topology has at least one data pipeline and emits liveness through the metric and service-check - // baselines created above. - if dp_config.data_pipelines_enabled() { - add_liveness_source_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?; + // Connected topologies emit liveness through the metric and service-check baselines created above. Standalone + // OTLP proxy mode must not construct liveness or output paths that it does not otherwise need. + if !dp_config.standalone_mode() { + add_liveness_source_to_blueprint(&mut blueprint, env_provider).await?; } // Now we move on to our actual data pipelines. @@ -452,12 +452,12 @@ async fn create_topology( Ok((blueprint, control_surfaces)) } -const LIVENESS_METRICS_DESTINATION: &str = "dd_metrics_encode"; +const LIVENESS_METRICS_DESTINATION: &str = "metrics_enrich"; async fn add_liveness_source_to_blueprint( - blueprint: &mut TopologyBlueprint, config: &GenericConfiguration, env_provider: &ADPEnvironmentProvider, + blueprint: &mut TopologyBlueprint, env_provider: &ADPEnvironmentProvider, ) -> Result<(), GenericError> { - let liveness_config = LivenessConfiguration::from_environment_provider(config, env_provider).await?; + let liveness_config = LivenessConfiguration::from_environment_provider(env_provider).await?; blueprint .add_source("liveness_in", liveness_config)? @@ -919,8 +919,8 @@ mod liveness_topology_tests { use super::LIVENESS_METRICS_DESTINATION; #[test] - fn liveness_metrics_bypass_optional_host_tag_enrichment() { - assert_eq!(LIVENESS_METRICS_DESTINATION, "dd_metrics_encode"); - assert_ne!(LIVENESS_METRICS_DESTINATION, "metrics_enrich"); + fn liveness_metrics_use_common_host_tag_enrichment() { + assert_eq!(LIVENESS_METRICS_DESTINATION, "metrics_enrich"); + assert_ne!(LIVENESS_METRICS_DESTINATION, "dd_metrics_encode"); } } diff --git a/bin/agent-data-plane/src/components/liveness/mod.rs b/bin/agent-data-plane/src/components/liveness/mod.rs index e41a5dce26..5ad6da27b8 100644 --- a/bin/agent-data-plane/src/components/liveness/mod.rs +++ b/bin/agent-data-plane/src/components/liveness/mod.rs @@ -1,10 +1,6 @@ -use std::{ - sync::{Arc, LazyLock}, - time::Duration, -}; +use std::{sync::LazyLock, time::Duration}; use async_trait::async_trait; -use datadog_agent_commons::ipc::{client::RemoteAgentClient, config::RemoteAgentClientConfiguration}; use saluki_context::{ tags::{Tag, TagSet}, Context, @@ -24,15 +20,13 @@ use saluki_error::{ErrorContext as _, GenericError}; use stringtheory::MetaString; use tokio::{ pin, select, - time::{interval, timeout, MissedTickBehavior}, + time::{interval, MissedTickBehavior}, }; use tracing::{debug, warn}; use crate::internal::env::ADPEnvironmentProvider; const LIVENESS_INTERVAL: Duration = Duration::from_secs(15); -const LIVENESS_IPC_TIMEOUT: Duration = Duration::from_secs(2); -const LIVENESS_DISPATCH_TIMEOUT: Duration = Duration::from_secs(2); const RUNNING_METRIC_NAME: &str = "datadog.agent_data_plane.running"; const UP_SERVICE_CHECK_NAME: &str = "datadog.agent_data_plane.up"; @@ -40,38 +34,23 @@ const UP_SERVICE_CHECK_NAME: &str = "datadog.agent_data_plane.up"; pub struct LivenessConfiguration { hostname: MetaString, version: MetaString, - client_config: Option>, } impl LivenessConfiguration { - /// Creates a liveness source configuration using the configured environment and Agent IPC client settings. - pub async fn from_environment_provider( - config: &saluki_config::GenericConfiguration, env_provider: &ADPEnvironmentProvider, - ) -> Result { + /// Creates a liveness source configuration using the configured environment. + pub async fn from_environment_provider(env_provider: &ADPEnvironmentProvider) -> Result { let hostname = env_provider .host() .get_hostname() .await .error_context("Failed to get hostname for liveness source.")?; - Self::from_hostname(config, hostname) + Ok(Self::from_hostname(hostname)) } - fn from_hostname(config: &saluki_config::GenericConfiguration, hostname: String) -> Result { - let client_config = RemoteAgentClientConfiguration::from_configuration(config)?; - - Ok(Self { - hostname: hostname.into(), - version: saluki_metadata::get_app_details().version().raw().into(), - client_config: Some(Arc::new(client_config)), - }) - } - - #[cfg(test)] - fn for_tests(hostname: &str, version: &str) -> Self { + fn from_hostname(hostname: String) -> Self { Self { hostname: hostname.into(), - version: version.into(), - client_config: None, + version: saluki_metadata::get_app_details().version().raw().into(), } } } @@ -82,9 +61,6 @@ impl SourceBuilder for LivenessConfiguration { Ok(Box::new(Liveness { hostname: self.hostname.clone(), version: self.version.clone(), - client_config: self.client_config.clone(), - client: None, - system_tags: None, })) } @@ -108,61 +84,11 @@ impl MemoryBounds for LivenessConfiguration { struct Liveness { hostname: MetaString, version: MetaString, - client_config: Option>, - client: Option, - system_tags: Option, -} - -impl Liveness { - async fn system_tags(&mut self) -> TagSet { - if let Some(system_tags) = &self.system_tags { - return system_tags.clone(); - } - - if self.client.is_none() { - let Some(client_config) = &self.client_config else { - return TagSet::default(); - }; - - match timeout( - LIVENESS_IPC_TIMEOUT, - RemoteAgentClient::from_client_configuration(client_config), - ) - .await - { - Ok(Ok(client)) => self.client = Some(client), - Ok(Err(error)) => { - warn!(error = %error, "Failed to create Agent IPC client for liveness host tags; continuing without tags."); - return TagSet::default(); - } - Err(_) => { - warn!("Timed out creating Agent IPC client for liveness host tags; continuing without tags."); - return TagSet::default(); - } - } - } - - let client = self.client.as_ref().expect("client is initialized above"); - match timeout(LIVENESS_IPC_TIMEOUT, client.get_host_tags()).await { - Ok(Ok(response)) => { - let tags = response.into_inner().system.into_iter().map(Tag::from).collect(); - self.system_tags = Some(tags); - } - Ok(Err(error)) => { - warn!(error = %error, "Failed to fetch liveness host tags; continuing with cached or empty tags."); - } - Err(_) => { - warn!("Timed out fetching liveness host tags; continuing with cached or empty tags."); - } - } - - self.system_tags.clone().unwrap_or_default() - } } #[async_trait] impl Source for Liveness { - async fn run(mut self: Box, mut context: SourceContext) -> Result<(), GenericError> { + async fn run(self: Box, mut context: SourceContext) -> Result<(), GenericError> { let global_shutdown = context.take_shutdown_handle(); pin!(global_shutdown); @@ -173,7 +99,7 @@ impl Source for Liveness { health.mark_ready(); debug!("Liveness source started."); - 'run: loop { + loop { select! { _ = &mut global_shutdown => { debug!("Received shutdown signal."); @@ -181,53 +107,14 @@ impl Source for Liveness { }, _ = health.live() => continue, _ = tick_interval.tick() => { - let system_tags = { - let system_tags = self.system_tags(); - pin!(system_tags); - select! { - _ = &mut global_shutdown => { - debug!("Received shutdown signal."); - break 'run; - }, - _ = health.live() => continue 'run, - system_tags = &mut system_tags => system_tags, - } - }; - let (metric, service_check) = - create_liveness_events(self.hostname.clone(), self.version.clone(), &system_tags); + let (metric, service_check) = create_liveness_events(self.hostname.clone(), self.version.clone()); - let metric_dispatch = - timeout(LIVENESS_DISPATCH_TIMEOUT, context.dispatcher().dispatch_one_named("metrics", metric)); - pin!(metric_dispatch); - select! { - _ = &mut global_shutdown => { - debug!("Received shutdown signal."); - break 'run; - }, - _ = health.live() => continue 'run, - result = &mut metric_dispatch => match result { - Ok(Ok(())) => {}, - Ok(Err(error)) => warn!(error = %error, "Failed to dispatch liveness metric."), - Err(_) => warn!("Timed out dispatching liveness metric."), - }, + if let Err(error) = context.dispatcher().dispatch_one_named("metrics", metric).await { + warn!(error = %error, "Failed to dispatch liveness metric."); } - let service_check_dispatch = timeout( - LIVENESS_DISPATCH_TIMEOUT, - context.dispatcher().dispatch_one_named("service_checks", service_check), - ); - pin!(service_check_dispatch); - select! { - _ = &mut global_shutdown => { - debug!("Received shutdown signal."); - break 'run; - }, - _ = health.live() => continue 'run, - result = &mut service_check_dispatch => match result { - Ok(Ok(())) => {}, - Ok(Err(error)) => warn!(error = %error, "Failed to dispatch liveness service check."), - Err(_) => warn!("Timed out dispatching liveness service check."), - }, + if let Err(error) = context.dispatcher().dispatch_one_named("service_checks", service_check).await { + warn!(error = %error, "Failed to dispatch liveness service check."); } }, } @@ -238,38 +125,30 @@ impl Source for Liveness { } } -fn create_liveness_events(hostname: MetaString, version: MetaString, system_tags: &TagSet) -> (Event, Event) { - let mut metric_tags = system_tags.clone(); - metric_tags.insert_tag(format!("version:{version}")); +fn create_liveness_events(hostname: MetaString, version: MetaString) -> (Event, Event) { let metric_context = Context::from_static_name(RUNNING_METRIC_NAME) .with_host(hostname.clone()) - .with_tags(metric_tags); + .with_tags(TagSet::from_iter([Tag::from(format!("version:{version}"))])); let metric = Event::Metric(Metric::gauge(metric_context, 1.0)); - let service_check = Event::ServiceCheck( - ServiceCheck::new(UP_SERVICE_CHECK_NAME, CheckStatus::Ok) - .with_hostname(hostname) - .with_tags(system_tags.clone()), - ); + let service_check = + Event::ServiceCheck(ServiceCheck::new(UP_SERVICE_CHECK_NAME, CheckStatus::Ok).with_hostname(hostname)); (metric, service_check) } #[cfg(test)] mod tests { - use saluki_config::ConfigurationLoader; - use saluki_context::tags::TagSet; use saluki_core::{ components::sources::SourceBuilder as _, data_model::event::{metric::MetricValues, service_check::CheckStatus, Event, EventType}, }; - use serde_json::json; use super::*; #[test] fn declares_named_metric_and_service_check_outputs() { - let configuration = LivenessConfiguration::for_tests("host-a", "1.2.3"); + let configuration = LivenessConfiguration::from_hostname("host-a".to_string()); let outputs = configuration.outputs(); assert_eq!(outputs.len(), 2); @@ -279,36 +158,27 @@ mod tests { assert_eq!(outputs[1].data_ty(), EventType::ServiceCheck); } - #[tokio::test] - async fn source_configuration_parses_ipc_settings_without_creating_a_client() { - let (config, _) = ConfigurationLoader::for_tests( - Some(json!({ - "auth_token_file_path": "/path/that/does/not/exist", - "cmd_port": 9, - })), - None, - false, - ) - .await; - - let configuration = LivenessConfiguration::from_hostname(&config, "host-a".to_string()) - .expect("IPC settings should parse without connecting to the Agent"); - - assert!(configuration.client_config.is_some()); + #[test] + fn source_configuration_requires_only_a_hostname_and_version() { + let configuration = LivenessConfiguration::from_hostname("host-a".to_string()); + + assert_eq!(configuration.hostname, "host-a"); + assert_eq!( + configuration.version, + saluki_metadata::get_app_details().version().raw() + ); } #[test] - fn emits_core_agent_compatible_liveness_events() { - let system_tags = TagSet::from_iter(["system:linux".into(), "region:us-east-1".into()]); - let (metric, service_check) = create_liveness_events("host-a".into(), "1.2.3".into(), &system_tags); + fn emits_liveness_events_without_direct_system_host_tags() { + let (metric, service_check) = create_liveness_events("host-a".into(), "1.2.3".into()); let Event::Metric(metric) = metric else { panic!("expected metric event"); }; assert_eq!(metric.context().name(), RUNNING_METRIC_NAME); assert_eq!(metric.context().host(), Some("host-a")); - assert!(metric.context().tags().has_tag("system:linux")); - assert!(metric.context().tags().has_tag("region:us-east-1")); + assert_eq!(metric.context().tags().len(), 1); assert!(metric.context().tags().has_tag("version:1.2.3")); assert_eq!(metric.values(), &MetricValues::gauge((0, 1.0))); @@ -318,8 +188,6 @@ mod tests { assert_eq!(service_check.name(), UP_SERVICE_CHECK_NAME); assert_eq!(service_check.status(), CheckStatus::Ok); assert_eq!(service_check.hostname(), Some("host-a")); - assert!(service_check.tags().has_tag("system:linux")); - assert!(service_check.tags().has_tag("region:us-east-1")); - assert!(!service_check.tags().has_tag("version:1.2.3")); + assert!(service_check.tags().is_empty()); } } diff --git a/bin/agent-data-plane/src/config.rs b/bin/agent-data-plane/src/config.rs index 9be6c36f0a..2d543b55e7 100644 --- a/bin/agent-data-plane/src/config.rs +++ b/bin/agent-data-plane/src/config.rs @@ -98,10 +98,14 @@ impl DataPlaneConfiguration { /// Returns `true` if the metrics pipeline is required. /// - /// A running topology needs this pipeline whenever it has a data pipeline so liveness signals can be encoded and - /// forwarded, including when the only data pipeline is an OTLP proxy. + /// Connected topologies need this pipeline whenever they have a data pipeline so the liveness metric can be + /// enriched and forwarded, including when the only data pipeline is an OTLP proxy. Standalone mode only creates + /// the pipeline for data sources that use it directly. pub const fn metrics_pipeline_required(&self) -> bool { - self.data_pipelines_enabled() + self.checks().enabled() + || self.dogstatsd().enabled() + || (self.otlp().enabled() && !self.otlp().proxy().enabled()) + || (!self.standalone_mode() && self.data_pipelines_enabled()) } /// Returns `true` if the logs pipeline is required. @@ -125,10 +129,13 @@ impl DataPlaneConfiguration { /// Returns `true` if the service checks pipeline is required. /// - /// A running topology needs this pipeline whenever it has a data pipeline so liveness signals can be encoded and - /// forwarded, including when the only data pipeline is an OTLP proxy. + /// Connected topologies need this pipeline whenever they have a data pipeline so the liveness service check can + /// be encoded and forwarded, including when the only data pipeline is an OTLP proxy. Standalone mode only creates + /// the pipeline for data sources that use it directly. pub const fn service_checks_pipeline_required(&self) -> bool { - self.data_pipelines_enabled() + self.checks().enabled() + || self.dogstatsd().enabled() + || (!self.standalone_mode() && self.data_pipelines_enabled()) } /// Returns `true` if the traces pipeline is required. @@ -440,9 +447,28 @@ mod tests { } #[tokio::test] - async fn otlp_proxy_mode_proxying_all_signals_requires_liveness_baseline_pipelines() { - // With proxy mode enabled and traces still proxied to the Core Agent (the default), ADP handles no OTLP - // signals itself, but it still needs the liveness metric and service-check baseline pipelines. + async fn standalone_otlp_proxy_mode_does_not_require_liveness_baseline_pipelines() { + // Standalone OTLP proxy mode must only construct the local proxy path, which avoids resolving output endpoints. + let dp = dp_config_from(json!({ + "data_plane": { + "enabled": true, + "standalone_mode": true, + "dogstatsd": { "enabled": false }, + "otlp": { "enabled": true, "proxy": { "enabled": true } }, + }, + })) + .await; + + assert!(dp.data_pipelines_enabled()); + assert!(!dp.metrics_pipeline_required()); + assert!(!dp.logs_pipeline_required()); + assert!(!dp.events_pipeline_required()); + assert!(!dp.service_checks_pipeline_required()); + assert!(!dp.traces_pipeline_required()); + } + + #[tokio::test] + async fn connected_otlp_proxy_mode_requires_liveness_baseline_pipelines() { let dp = dp_config_from(json!({ "data_plane": { "enabled": true, From c90a1a2c7b444933a149f9063fb085626babc0cc Mon Sep 17 00:00:00 2001 From: Travis Thieman Date: Tue, 28 Jul 2026 15:36:21 -0400 Subject: [PATCH 5/6] fix(agent-data-plane): prebuild liveness payloads Construct the liveness metric and service check with the source, then clone them for each dispatch tick. Preserve explicit hostnames because the shared metric path and direct service-check encoder do not add them.\n\nRestore the original topology comments while documenting the connected liveness exception for proxy-only OTLP deployments, and replace source-tag implementation assertions with payload contract coverage. --- bin/agent-data-plane/src/cli/run.rs | 8 ++--- .../src/components/liveness/mod.rs | 35 +++++++++++++------ 2 files changed, 28 insertions(+), 15 deletions(-) diff --git a/bin/agent-data-plane/src/cli/run.rs b/bin/agent-data-plane/src/cli/run.rs index c9100b4f69..e21f66a045 100644 --- a/bin/agent-data-plane/src/cli/run.rs +++ b/bin/agent-data-plane/src/cli/run.rs @@ -380,8 +380,7 @@ async fn create_topology( let mut control_surfaces = TopologyControlSurfaces::default(); - // Without a data pipeline, there is no useful topology to run. In particular, liveness alone must not keep ADP - // alive. + // If no data pipelines are enabled, then there's nothing for us to do. if !dp_config.data_pipelines_enabled() { return Err(generic_error!("No data pipelines are enabled. Exiting.")); } @@ -393,8 +392,9 @@ async fn create_topology( // we additionally create metrics- and logs-specific components connected to that forwarder depending on which of // the baseline pipelines are required. // - // Connected liveness requires the metrics and service-check baseline pipelines for every running topology, - // including an OTLP proxy-only topology. + // Notably, we _don't_ need either of these if all we're doing is running the OTLP pipeline in proxy mode, which + // is the only reason we're differentiating here. Connected mode is the exception: liveness deliberately + // provisions metric and service-check output paths even for proxy-only topologies. if dp_config.metrics_pipeline_required() || dp_config.logs_pipeline_required() || dp_config.events_pipeline_required() diff --git a/bin/agent-data-plane/src/components/liveness/mod.rs b/bin/agent-data-plane/src/components/liveness/mod.rs index 5ad6da27b8..df1f35dbe0 100644 --- a/bin/agent-data-plane/src/components/liveness/mod.rs +++ b/bin/agent-data-plane/src/components/liveness/mod.rs @@ -58,10 +58,7 @@ impl LivenessConfiguration { #[async_trait] impl SourceBuilder for LivenessConfiguration { async fn build(&self, _context: ComponentContext) -> Result, GenericError> { - Ok(Box::new(Liveness { - hostname: self.hostname.clone(), - version: self.version.clone(), - })) + Ok(Box::new(Liveness::new(self.hostname.clone(), self.version.clone()))) } fn outputs(&self) -> &[OutputDefinition] { @@ -82,8 +79,19 @@ impl MemoryBounds for LivenessConfiguration { } struct Liveness { - hostname: MetaString, - version: MetaString, + metric: Event, + service_check: Event, +} + +impl Liveness { + fn new(hostname: MetaString, version: MetaString) -> Self { + let (metric, service_check) = create_liveness_events(hostname, version); + Self { metric, service_check } + } + + fn events(&self) -> (Event, Event) { + (self.metric.clone(), self.service_check.clone()) + } } #[async_trait] @@ -107,7 +115,7 @@ impl Source for Liveness { }, _ = health.live() => continue, _ = tick_interval.tick() => { - let (metric, service_check) = create_liveness_events(self.hostname.clone(), self.version.clone()); + let (metric, service_check) = self.events(); if let Err(error) = context.dispatcher().dispatch_one_named("metrics", metric).await { warn!(error = %error, "Failed to dispatch liveness metric."); @@ -170,17 +178,23 @@ mod tests { } #[test] - fn emits_liveness_events_without_direct_system_host_tags() { - let (metric, service_check) = create_liveness_events("host-a".into(), "1.2.3".into()); + fn prebuilt_metric_payload_has_required_contract() { + let liveness = Liveness::new("host-a".into(), "1.2.3".into()); + let (metric, _) = liveness.events(); let Event::Metric(metric) = metric else { panic!("expected metric event"); }; assert_eq!(metric.context().name(), RUNNING_METRIC_NAME); assert_eq!(metric.context().host(), Some("host-a")); - assert_eq!(metric.context().tags().len(), 1); assert!(metric.context().tags().has_tag("version:1.2.3")); assert_eq!(metric.values(), &MetricValues::gauge((0, 1.0))); + } + + #[test] + fn prebuilt_service_check_payload_has_required_contract() { + let liveness = Liveness::new("host-a".into(), "1.2.3".into()); + let (_, service_check) = liveness.events(); let Event::ServiceCheck(service_check) = service_check else { panic!("expected service check event"); @@ -188,6 +202,5 @@ mod tests { assert_eq!(service_check.name(), UP_SERVICE_CHECK_NAME); assert_eq!(service_check.status(), CheckStatus::Ok); assert_eq!(service_check.hostname(), Some("host-a")); - assert!(service_check.tags().is_empty()); } } From d94bbb262d99a4de7edcf2e1718966006962378a Mon Sep 17 00:00:00 2001 From: Travis Thieman Date: Tue, 28 Jul 2026 15:52:38 -0400 Subject: [PATCH 6/6] test(agent-data-plane): remove redundant liveness topology test Remove the constant-only liveness topology test module after review feedback.\n\nLiveness routing remains covered by the production topology configuration and existing component tests. --- bin/agent-data-plane/src/cli/run.rs | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/bin/agent-data-plane/src/cli/run.rs b/bin/agent-data-plane/src/cli/run.rs index e21f66a045..e51a04cfc4 100644 --- a/bin/agent-data-plane/src/cli/run.rs +++ b/bin/agent-data-plane/src/cli/run.rs @@ -913,14 +913,3 @@ fn write_sizing_guide(bounds: ComponentBounds) -> Result<(), GenericError> { Ok(()) } - -#[cfg(test)] -mod liveness_topology_tests { - use super::LIVENESS_METRICS_DESTINATION; - - #[test] - fn liveness_metrics_use_common_host_tag_enrichment() { - assert_eq!(LIVENESS_METRICS_DESTINATION, "metrics_enrich"); - assert_ne!(LIVENESS_METRICS_DESTINATION, "dd_metrics_encode"); - } -}