feat(agent-data-plane): emit liveness signals

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,
@@ -393,7 +393,8 @@ async fn create_topology(
     // 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.
+    // 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()
@@ -426,6 +427,12 @@ async fn create_topology(
         add_baseline_traces_pipeline_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.
     if dp_config.checks().enabled() {
         add_checks_pipeline_to_blueprint(&mut blueprint, &config_system.raw_map(), env_provider).await?;
@@ -445,6 +452,21 @@ async fn create_topology(
     Ok((blueprint, control_surfaces))
 }
 
+const LIVENESS_METRICS_DESTINATION: &str = "metrics_enrich";
+
+async fn add_liveness_source_to_blueprint(
+    blueprint: &mut TopologyBlueprint, env_provider: &ADPEnvironmentProvider,
+) -> Result<(), GenericError> {
+    let liveness_config = LivenessConfiguration::from_environment_provider(env_provider).await?;
+
+    blueprint
+        .add_source("liveness_in", liveness_config)?
+        .connect_components("liveness_in.metrics", LIVENESS_METRICS_DESTINATION)?
+        .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> {

bin/agent-data-plane/src/components/liveness/mod.rs

@@ -0,0 +1,206 @@
+use std::{sync::LazyLock, time::Duration};
+
+use async_trait::async_trait;
+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, MissedTickBehavior},
+};
+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,
+}
+
+impl LivenessConfiguration {
+    /// Creates a liveness source configuration using the configured environment.
+    pub async fn from_environment_provider(env_provider: &ADPEnvironmentProvider) -> Result<Self, GenericError> {
+        let hostname = env_provider
+            .host()
+            .get_hostname()
+            .await
+            .error_context("Failed to get hostname for liveness source.")?;
+        Ok(Self::from_hostname(hostname))
+    }
+
+    fn from_hostname(hostname: String) -> Self {
+        Self {
+            hostname: hostname.into(),
+            version: saluki_metadata::get_app_details().version().raw().into(),
+        }
+    }
+}
+
+#[async_trait]
+impl SourceBuilder for LivenessConfiguration {
+    async fn build(&self, _context: ComponentContext) -> Result<Box<dyn Source + Send>, GenericError> {
+        Ok(Box::new(Liveness::new(self.hostname.clone(), self.version.clone())))
+    }
+
+    fn outputs(&self) -> &[OutputDefinition<EventType>] {
+        static OUTPUTS: LazyLock<Vec<OutputDefinition<EventType>>> = 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>("liveness source");
+    }
+}
+
+struct Liveness {
+    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]
+impl Source for Liveness {
+    async fn run(self: Box<Self>, 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);
+        tick_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+        health.mark_ready();
+        debug!("Liveness source started.");
+
+        loop {
+            select! {
+                _ = &mut global_shutdown => {
+                    debug!("Received shutdown signal.");
+                    break;
+                },
+                _ = health.live() => continue,
+                _ = tick_interval.tick() => {
+                    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.");
+                    }
+
+                    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) -> (Event, Event) {
+    let metric_context = Context::from_static_name(RUNNING_METRIC_NAME)
+        .with_host(hostname.clone())
+        .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));
+
+    (metric, service_check)
+}
+
+#[cfg(test)]
+mod tests {
+    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::from_hostname("host-a".to_string());
+        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 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 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!(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");
+        };
+        assert_eq!(service_check.name(), UP_SERVICE_CHECK_NAME);
+        assert_eq!(service_check.status(), CheckStatus::Ok);
+        assert_eq!(service_check.hostname(), Some("host-a"));
+    }
+}

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;

bin/agent-data-plane/src/config.rs

@@ -98,16 +98,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.
+    /// 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 {
-        // 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.dogstatsd().enabled()
             || (self.otlp().enabled() && !self.otlp().proxy().enabled())
+            || (!self.standalone_mode() && self.data_pipelines_enabled())
     }
 
     /// Returns `true` if the logs pipeline is required.
@@ -131,10 +129,13 @@ 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.
+    /// 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.checks().enabled() || self.dogstatsd().enabled()
+        self.checks().enabled()
+            || self.dogstatsd().enabled()
+            || (!self.standalone_mode() && self.data_pipelines_enabled())
     }
 
     /// Returns `true` if the traces pipeline is required.
@@ -441,17 +442,17 @@ 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 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 } },
             },
@@ -467,9 +468,28 @@ mod tests {
     }
 
     #[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 connected_otlp_proxy_mode_requires_liveness_baseline_pipelines() {
+        let dp = dp_config_from(json!({
+            "data_plane": {
+                "enabled": 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 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 +503,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 topology_without_data_pipelines_requires_no_baseline_pipelines() {
         let dp = dp_config_from(json!({
             "data_plane": {
                 "enabled": true,