Add ability to register ADP as a remote agent to the Datadog Agent

Cargo.toml

@@ -142,6 +142,8 @@ chrono-tz = { version = "0.10", default-features = false }
 iana-time-zone = { version = "0.1", default-features = false }
 backon = { version = "1", default-features = false }
 http-serde-ext = { version = "1", default-features = false }
+uuid = { version = "1.11.0", default-features = false }
+rcgen = { version = "0.13.1", default-features = false }
 tikv-jemalloc-ctl = "0.6"
 tikv-jemallocator = { version = "0.6", default-features = false }
 axum-extra = { version = "0.9", default-features = false }

LICENSE-3rdparty.csv

@@ -221,6 +221,7 @@ rand_xorshift,https://github.com/rust-random/rngs,MIT OR Apache-2.0,"The Rand Pr
 raw-cpuid,https://github.com/gz/rust-cpuid,MIT,Gerd Zellweger <[email protected]>
 rawpointer,https://github.com/bluss/rawpointer,MIT OR Apache-2.0,bluss
 rayon,https://github.com/rayon-rs/rayon,MIT OR Apache-2.0,"Niko Matsakis <[email protected]>, Josh Stone <[email protected]>"
+rcgen,https://github.com/rustls/rcgen,MIT OR Apache-2.0,The rcgen Authors
 redox_syscall,https://gitlab.redox-os.org/redox-os/syscall,MIT,Jeremy Soller <[email protected]>
 regex,https://github.com/rust-lang/regex,MIT OR Apache-2.0,"The Rust Project Developers, Andrew Gallant <[email protected]>"
 regex-automata,https://github.com/BurntSushi/regex-automata,Unlicense OR MIT,Andrew Gallant <[email protected]>
@@ -319,6 +320,7 @@ utf16_iter,https://github.com/hsivonen/utf16_iter,Apache-2.0 OR MIT,Henri Sivone
 utf8-width,https://github.com/magiclen/utf8-width,MIT,Magic Len <[email protected]>
 utf8_iter,https://github.com/hsivonen/utf8_iter,Apache-2.0 OR MIT,Henri Sivonen <[email protected]>
 utf8parse,https://github.com/alacritty/vte,Apache-2.0 OR MIT,"Joe Wilm <[email protected]>, Christian Duerr <[email protected]>"
+uuid,https://github.com/uuid-rs/uuid,Apache-2.0 OR MIT,"Ashley Mannix<[email protected]>, Dylan DPC<[email protected]>, Hunar Roop Kahlon<[email protected]>"
 valuable,https://github.com/tokio-rs/valuable,MIT,The valuable Authors
 walkdir,https://github.com/BurntSushi/walkdir,Unlicense OR MIT,Andrew Gallant <[email protected]>
 want,https://github.com/seanmonstar/want,MIT,Sean McArthur <[email protected]>
@@ -338,6 +340,7 @@ winnow,https://github.com/winnow-rs/winnow,MIT,The winnow Authors
 write16,https://github.com/hsivonen/write16,Apache-2.0 OR MIT,The write16 Authors
 writeable,https://github.com/unicode-org/icu4x,Unicode-3.0,The ICU4X Project Developers
 yansi,https://github.com/SergioBenitez/yansi,MIT OR Apache-2.0,Sergio Benitez <[email protected]>
+yasna,https://github.com/qnighy/yasna.rs,MIT OR Apache-2.0,Masaki Hara <[email protected]>
 yoke,https://github.com/unicode-org/icu4x,Unicode-3.0,Manish Goregaokar <[email protected]>
 yoke-derive,https://github.com/unicode-org/icu4x,Unicode-3.0,Manish Goregaokar <[email protected]>
 zerocopy,https://github.com/google/zerocopy,BSD-2-Clause OR Apache-2.0 OR MIT,Joshua Liebow-Feeser <[email protected]>

bin/agent-data-plane/Cargo.toml

@@ -29,6 +29,7 @@ stringtheory = { workspace = true }
 tikv-jemalloc-ctl = { workspace = true, features = ["use_std"] }
 tikv-jemallocator = { workspace = true, features = ["background_threads", "unprefixed_malloc_on_supported_platforms", "stats"] }
 tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "signal"] }
+tokio-rustls = { workspace = true }
 tracing = { workspace = true }
 
 [build-dependencies]

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

@@ -13,7 +13,10 @@ use std::{
 use memory_accounting::ComponentRegistry;
 use saluki_app::{api::APIBuilder, logging::LoggingAPIHandler, prelude::*};
 use saluki_components::{
-    destinations::{DatadogEventsServiceChecksConfiguration, DatadogMetricsConfiguration, PrometheusConfiguration},
+    destinations::{
+        new_remote_agent_service, DatadogEventsServiceChecksConfiguration, DatadogMetricsConfiguration,
+        DatadogStatusFlareConfiguration, PrometheusConfiguration,
+    },
     sources::{DogStatsDConfiguration, InternalMetricsConfiguration},
     transforms::{
         AggregateConfiguration, ChainedConfiguration, DogstatsDPrefixFilterConfiguration, HostEnrichmentConfiguration,
@@ -106,17 +109,21 @@ async fn run(started: Instant, logging_api_handler: LoggingAPIHandler) -> Result
 
     // Create a simple pipeline that runs a DogStatsD source, an aggregation transform to bucket into 10 second windows,
     // and a Datadog Metrics destination that forwards aggregated buckets to the Datadog Platform.
-    let blueprint = create_topology(&configuration, env_provider, &component_registry)?;
+    let blueprint = create_topology(&configuration, env_provider, &component_registry).await?;
 
     // Build our administrative API server.
     let primary_api_listen_address = configuration
         .try_get_typed("api_listen_address")
         .error_context("Failed to get API listen address.")?
         .unwrap_or_else(|| ListenAddress::Tcp(([0, 0, 0, 0], 5100).into()));
 
+    let remote_agent_service = new_remote_agent_service()?;
+
     let primary_api = APIBuilder::new()
         .with_handler(health_registry.api_handler())
         .with_handler(component_registry.api_handler())
+        .with_grpc_service(remote_agent_service)
+        .with_self_signed_tls()
         .with_handler(logging_api_handler);
 
     // Run memory bounds validation to ensure that we can launch the topology with our configured memory limit, if any.
@@ -168,7 +175,7 @@ async fn run(started: Instant, logging_api_handler: LoggingAPIHandler) -> Result
     }
 }
 
-fn create_topology(
+async fn create_topology(
     configuration: &GenericConfiguration, env_provider: ADPEnvironmentProvider, component_registry: &ComponentRegistry,
 ) -> Result<TopologyBlueprint, GenericError> {
     // Create a simple pipeline that runs a DogStatsD source, an aggregation transform to bucket into 10 second windows,
@@ -196,11 +203,17 @@ fn create_topology(
         }
     }
 
+    let status_configuration = DatadogStatusFlareConfiguration::from_configuration(configuration).await?;
+
     let events_service_checks_config = DatadogEventsServiceChecksConfiguration::from_configuration(configuration)
         .error_context("Failed to configure Datadog Events/Service Checks destination.")?;
 
+    let int_metrics_config = InternalMetricsConfiguration;
+    let int_metrics_remap_config = AgentTelemetryRemapperConfiguration::new();
+
     let topology_registry = component_registry.get_or_create("topology");
     let mut blueprint = TopologyBlueprint::from_component_registry(topology_registry);
+
     blueprint
         .add_source("dsd_in", dsd_config)?
         .add_transform("dsd_agg", dsd_agg_config)?
@@ -214,17 +227,29 @@ fn create_topology(
         .connect_component("dd_metrics_out", ["enrich"])?
         .connect_component("dd_events_sc_out", ["dsd_in.events", "dsd_in.service_checks"])?;
 
-    // Insert a Prometheus scrape destination if we've been instructed to enable internal telemetry.
-    if configuration.get_typed_or_default::<bool>("telemetry_enabled") {
-        let int_metrics_config = InternalMetricsConfiguration;
-        let int_metrics_remap_config = AgentTelemetryRemapperConfiguration::new();
-        let prometheus_config = PrometheusConfiguration::from_configuration(configuration)?;
+    let use_prometheus = configuration.get_typed_or_default::<bool>("telemetry_enabled");
+    let use_status_flare_component = !configuration.get_typed_or_default::<bool>("adp_standalone_mode");
 
+    // Insert internal metrics source only if internal telemetry or status and flare component is enabled.
+    if use_prometheus || use_status_flare_component {
         blueprint
             .add_source("internal_metrics_in", int_metrics_config)?
             .add_transform("internal_metrics_remap", int_metrics_remap_config)?
+            .connect_component("internal_metrics_remap", ["internal_metrics_in"])?;
+    }
+
+    // Insert a Datadog Status Flare destination if we've been instructed to not do a no-op.
+    if use_status_flare_component {
+        blueprint
+            .add_destination("dd_status_flare_out", status_configuration)?
+            .connect_component("dd_status_flare_out", ["internal_metrics_remap"])?;
+    }
+
+    // Insert a Prometheus scrape destination if we've been instructed to enable internal telemetry.
+    if use_prometheus {
+        let prometheus_config = PrometheusConfiguration::from_configuration(configuration)?;
+        blueprint
             .add_destination("internal_metrics_out", prometheus_config)?
-            .connect_component("internal_metrics_remap", ["internal_metrics_in"])?
             .connect_component("internal_metrics_out", ["internal_metrics_remap"])?;
     }
 

lib/datadog-protos/build.rs

@@ -31,21 +31,15 @@ fn main() {
 
     // Handle code generation for gRPC service definitions.
     tonic_build::configure()
-        .build_server(false)
+        .build_server(true)
         .include_file("api.mod.rs")
         .compile_protos(
             &[
                 "proto/datadog/api/v1/api.proto",
                 "proto/datadog/workloadmeta/workloadmeta.proto",
+                "proto/datadog/remoteagent/remoteagent.proto",
             ],
             &["proto"],
         )
         .expect("failed to build gRPC service definitions for DCA");
-
-    // Handle code generation for gRPC service definitions.
-    tonic_build::configure()
-        .build_server(true)
-        .include_file("remoteagent.mod.rs")
-        .compile_protos(&["proto/datadog/remoteagent/remoteagent.proto"], &["proto"])
-        .expect("failed to build gRPC service definitions for RemoteAgent")
 }

lib/datadog-protos/src/lib.rs

@@ -12,10 +12,6 @@ mod agent_include {
     include!(concat!(env!("OUT_DIR"), "/api.mod.rs"));
 }
 
-mod remoteagent_include {
-    include!(concat!(env!("OUT_DIR"), "/remoteagent.mod.rs"));
-}
-
 /// Metrics-related definitions.
 pub mod metrics {
     pub use super::include::dd_metric::metric_payload::*;
@@ -33,11 +29,9 @@ pub mod traces {
 pub mod agent {
     pub use super::agent_include::datadog::api::v1::agent_client::AgentClient;
     pub use super::agent_include::datadog::api::v1::agent_secure_client::AgentSecureClient;
+    pub use super::agent_include::datadog::api::v1::remote_agent_server::RemoteAgent;
+    pub use super::agent_include::datadog::api::v1::remote_agent_server::RemoteAgentServer;
     pub use super::agent_include::datadog::model::v1::*;
+    pub use super::agent_include::datadog::remoteagent::*;
     pub use super::agent_include::datadog::workloadmeta::*;
 }
-
-// RemoteAgent definitions.
-pub mod remoteagent {
-    pub use super::remoteagent_include::datadog::remoteagent::*;
-}

lib/saluki-app/Cargo.toml

@@ -20,10 +20,14 @@ axum = { workspace = true, optional = true }
 bytesize = { workspace = true, optional = true }
 chrono = { workspace = true, optional = true }
 chrono-tz = { workspace = true, optional = true }
+http = { workspace = true }
 iana-time-zone = { workspace = true, optional = true }
 memory-accounting = { workspace = true, optional = true }
 metrics = { workspace = true, optional = true }
 metrics-util = { workspace = true, features = ["handles", "recency", "registry"], optional = true }
+rcgen = { workspace = true, features = ["crypto", "aws_lc_rs", "pem"] }
+rustls = { workspace = true, features = ["tls12"] }
+rustls-pemfile = { workspace = true, features = ["std"] }
 saluki-api = { workspace = true, optional = true }
 saluki-config = { workspace = true, optional = true }
 saluki-core = { workspace = true, optional = true }
@@ -33,6 +37,7 @@ saluki-metadata = { workspace = true }
 saluki-tls = { workspace = true, optional = true }
 serde = { workspace = true, optional = true }
 tokio = { workspace = true, features = ["macros", "sync"], optional = true }
+tonic = { workspace = true, features = ["transport"] }
 tower = { workspace = true, features = ["util"], optional = true }
 tracing = { workspace = true, optional = true }
 tracing-subscriber = { workspace = true, features = ["ansi", "env-filter", "fmt", "json", "local-time", "registry", "std", "tracing-log"], optional = true }