feat(dogstatsd): add dogstatsd top

.gitattributes

@@ -0,0 +1,2 @@
+# NDJSON fixtures are byte-compared with compressed streams and must not be translated to CRLF.
+*.ndjson text eol=lf

bin/agent-data-plane/Cargo.toml

@@ -55,6 +55,7 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 serde_with = { workspace = true }
 stringtheory = { workspace = true }
+tempfile = { workspace = true }
 tokio = { workspace = true, features = [
   "macros",
   "net",
@@ -67,6 +68,7 @@ tokio-util = { workspace = true }
 tonic = { workspace = true }
 tracing = { workspace = true }
 uuid = { workspace = true, features = ["v4"] }
+zstd = { workspace = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 tokio = { workspace = true, features = ["taskdump"] }
@@ -82,7 +84,8 @@ chrono = { workspace = true }
 [dev-dependencies]
 datadog-agent-config-testing = { workspace = true }
 derive-where = { workspace = true }
-saluki-components = { workspace = true }
+saluki-components = { workspace = true, features = ["test-util"] }
 saluki-config = { workspace = true, features = ["test-util"] }
 saluki-core = { workspace = true, features = ["test-util"] }
 saluki-metrics = { workspace = true, features = ["test"] }
+tower = { workspace = true, features = ["util"] }

bin/agent-data-plane/src/cli/dogstatsd.rs

@@ -17,9 +17,7 @@ use saluki_components::sources::DEFAULT_REPLAY_LOOPS;
 #[cfg(target_os = "linux")]
 use saluki_components::sources::REPLAY_CREDENTIALS_GID;
 use saluki_config::{DurationString, GenericConfiguration};
-#[cfg(any(target_os = "linux", test))]
-use saluki_error::generic_error;
-use saluki_error::{ErrorContext as _, GenericError};
+use saluki_error::{generic_error, ErrorContext as _, GenericError};
 #[cfg(target_os = "linux")]
 use saluki_io::net::{unix::uds_sendmsg_with_creds, ProcessCredentials};
 use serde::Deserialize;
@@ -34,6 +32,9 @@ use tracing::{error, info};
 
 use crate::cli::utils::DataPlaneAPIClient;
 
+mod top;
+use self::top::{handle_dogstatsd_dump_contexts, handle_dogstatsd_top, DumpContextsCommand, TopCommand};
+
 /// DogStatsD-specific debugging commands.
 #[derive(FromArgs, Debug)]
 #[argh(subcommand, name = "dogstatsd")]
@@ -48,6 +49,8 @@ enum DogstatsdSubcommand {
     Stats(StatsCommand),
     Capture(CaptureCommand),
     Replay(ReplayCommand),
+    Top(TopCommand),
+    DumpContexts(DumpContextsCommand),
 }
 
 /// Prints basic statistics about the metrics received by the data plane.
@@ -168,34 +171,35 @@ struct StatsResponse<'a> {
 
 /// Entrypoint for the `dogstatsd` commands.
 pub async fn handle_dogstatsd_command(bootstrap_config: &GenericConfiguration, cmd: DogstatsdCommand) {
-    let mut api_client = match DataPlaneAPIClient::from_config(bootstrap_config) {
-        Ok(client) => client,
-        Err(e) => {
-            error!("Failed to create data plane API client: {:#}", e);
-            std::process::exit(1);
-        }
-    };
+    if let Err(error) = run_dogstatsd_command(bootstrap_config, cmd).await {
+        error!("{:#}", error);
+        std::process::exit(1);
+    }
+}
 
+async fn run_dogstatsd_command(
+    bootstrap_config: &GenericConfiguration, cmd: DogstatsdCommand,
+) -> Result<(), GenericError> {
     match cmd.subcommand {
         DogstatsdSubcommand::Stats(config) => {
-            if let Err(e) = handle_dogstatsd_stats(&mut api_client, config).await {
-                error!("Failed to run stats subcommand: {:#}", e);
-                std::process::exit(1);
-            }
+            let mut api_client = data_plane_api_client(bootstrap_config)?;
+            handle_dogstatsd_stats(&mut api_client, config)
+                .await
+                .error_context("Failed to run stats subcommand")
         }
         DogstatsdSubcommand::Capture(config) => {
-            if let Err(e) = handle_dogstatsd_capture(&mut api_client, config).await {
-                error!("Failed to start DogStatsD capture: {:#}", e);
-                std::process::exit(1);
-            }
+            let mut api_client = data_plane_api_client(bootstrap_config)?;
+            handle_dogstatsd_capture(&mut api_client, config)
+                .await
+                .error_context("Failed to start DogStatsD capture")
         }
         DogstatsdSubcommand::Replay(config) => {
             #[cfg(target_os = "linux")]
             {
-                if let Err(e) = handle_dogstatsd_replay(&mut api_client, bootstrap_config, config).await {
-                    error!("Failed to replay DogStatsD traffic: {:#}", e);
-                    std::process::exit(1);
-                }
+                let mut api_client = data_plane_api_client(bootstrap_config)?;
+                handle_dogstatsd_replay(&mut api_client, bootstrap_config, config)
+                    .await
+                    .error_context("Failed to replay DogStatsD traffic")
             }
 
             #[cfg(not(target_os = "linux"))]
@@ -205,13 +209,31 @@ pub async fn handle_dogstatsd_command(bootstrap_config: &GenericConfiguration, c
                     loops,
                 } = config;
                 let _ = (replay_file_path, loops);
-                error!("DogStatsD replay is only supported on Linux.");
-                std::process::exit(1);
+                Err(generic_error!("DogStatsD replay is only supported on Linux."))
+            }
+        }
+        DogstatsdSubcommand::Top(config) => {
+            let config = config.validate();
+            let mut output = std::io::stdout();
+            if config.is_offline() {
+                handle_dogstatsd_top(None, config, &mut output).await
+            } else {
+                let mut api_client = data_plane_api_client(bootstrap_config)?;
+                handle_dogstatsd_top(Some(&mut api_client), config, &mut output).await
             }
         }
+        DogstatsdSubcommand::DumpContexts(_) => {
+            let mut api_client = data_plane_api_client(bootstrap_config)?;
+            let mut output = std::io::stdout();
+            handle_dogstatsd_dump_contexts(&mut api_client, &mut output).await
+        }
     }
 }
 
+fn data_plane_api_client(config: &GenericConfiguration) -> Result<DataPlaneAPIClient, GenericError> {
+    DataPlaneAPIClient::from_config(config).error_context("Failed to create data plane API client")
+}
+
 async fn handle_dogstatsd_stats(api_client: &mut DataPlaneAPIClient, cmd: StatsCommand) -> Result<(), GenericError> {
     // Trigger a statistics collection and wait for it to complete.
     info!(