Extract observer into new crate

Cargo.toml

@@ -8,6 +8,7 @@ members = [
   "lading_capture",
   "lading_fuzz",
   "lading_payload",
+  "lading_observer",
   "lading_throttle",
 ]
 
@@ -60,6 +61,11 @@ http-body-util = { version = "0.1" }
 http-serde = { version = "2.1" }
 hyper = { version = "1.7", default-features = false }
 hyper-util = { version = "0.1", default-features = false }
+heck = { version = "0.5", default-features = false }
+nix = { version = "0.30" }
+num-traits = { version = "0.2", default-features = false }
+num_cpus = "1.17"
+regex = { version = "1.12" }
 sysinfo = { version = "0.37", default-features = false }
 opentelemetry-proto = "0.31"
 ddsketch-agent = { git = "https://github.com/DataDog/saluki", rev = "f47a7ef588c53aa1da35dcfd93808595ebeb1291", features = [

lading/Cargo.toml

@@ -18,6 +18,7 @@ description = "A tool for load testing daemons."
 lading-capture = { version = "0.2", path = "../lading_capture" }
 lading-capture-schema = { path = "../lading_capture_schema" }
 lading-payload = { version = "0.1", path = "../lading_payload" }
+lading-observer = { version = "0.1", path = "../lading_observer" }
 lading-throttle = { version = "0.1", path = "../lading_throttle" }
 lading-signal = { version = "0.1", path = "../lading_signal" }
 
@@ -39,7 +40,6 @@ either = { version = "1.15.0" }
 flate2 = { version = "1.1.2" }
 futures = { version = "0.3.31" }
 fuser = { version = "0.16", optional = true, features = ["libfuse"] }
-heck = { version = "0.5", default-features = false }
 http = { workspace = true }
 http-body-util = { workspace = true }
 http-serde = { version = "2.1" }
@@ -53,14 +53,13 @@ k8s-openapi = { version = "0.26.0", default-features = false, features = [
 metrics = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
 metrics-util = { workspace = true }
-nix = { version = "0.30", features = ["fs", "signal"] }
+nix = { workspace = true, features = ["fs", "signal"] }
 num_cpus = { version = "1.17" }
-num-traits = { version = "0.2", default-features = false }
 once_cell = { workspace = true }
 opentelemetry-proto = { workspace = true }
 prost = { workspace = true }
 rand = { workspace = true, features = ["small_rng", "thread_rng"] }
-regex = { version = "1.12" }
+regex = { workspace = true }
 reqwest = { version = "0.12", default-features = false, features = [
   "default-tls",
   "json",
@@ -95,7 +94,6 @@ tracing-subscriber = { workspace = true, features = ["env-filter", "json"] }
 zstd = { version = "0.13.3" }
 
 [target.'cfg(target_os = "linux")'.dependencies]
-procfs = { version = "0.18", default-features = false, features = [] }
 async-pidfd = { version = "0.1" }
 
 [dev-dependencies]

lading/src/lib.rs

@@ -17,7 +17,8 @@ mod common;
 pub mod config;
 pub mod generator;
 pub mod inspector;
-pub mod observer;
+/// Target observation utilities.
+pub use lading_observer as observer;
 pub(crate) mod proto;
 pub mod target;
 pub mod target_metrics;

lading_observer/Cargo.toml

@@ -0,0 +1,34 @@
+[package]
+name = "lading-observer"
+version = "0.1.0"
+authors = [
+  "Brian L. Troutwine <brian.troutwine@datadoghq.com>",
+  "George Hahn <george.hahn@datadoghq.com>",
+]
+edition = "2024"
+license = "MIT"
+repository = "https://github.com/datadog/lading/"
+keywords = ["random_test", "generator"]
+categories = ["development-tools::profiling"]
+description = "A tool for load testing daemons."
+
+[dependencies]
+lading-signal = { version = "0.1", path = "../lading_signal" }
+metrics = { workspace = true }
+nix = { workspace = true, features = ["fs", "signal"] }
+num-traits = { workspace = true }
+num_cpus = { workspace = true }
+procfs = { version = "0.18", default-features = false, features = [] }
+regex = { workspace = true }
+rustc-hash = { workspace = true }
+serde = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["fs", "macros", "sync", "time"] }
+tracing = { workspace = true }
+heck = { workspace = true }
+
+[lib]
+doctest = false
+
+[lints]
+workspace = true

lading/src/observer.rs → lading_observer/src/lib.rs

@@ -1,20 +1,26 @@
-//! Manage the target observer
+//! Observe target resource usage for lading.
 //!
-//! The interogation that lading does of the target sub-process is intentionally
-//! limited to in-process concerns, for the most part. The 'inspector' does
-//! allow for a sub-process to do out-of-band inspection of the target but
-//! cannot incorporate whatever it's doing into the capture data that lading
-//! produces. This observer, on Linux, looks up the target process in procfs and
-//! writes out key details about memory and CPU consumption into the capture
-//! data. On non-Linux systems the observer, if enabled, will emit a warning.
+//! This library supports the lading binary found elsewhere in this project. It
+//! samples target resource usage, primarily on Linux via procfs, and is not
+//! intended for external use.
 
-use std::io;
+#![deny(clippy::cargo)]
+#![allow(clippy::cast_precision_loss)]
+#![allow(clippy::multiple_crate_versions)]
 
-use crate::target::TargetPidReceiver;
+use std::{io, time::Duration};
+
+#[cfg(target_os = "linux")]
+use crate::linux::Sampler;
 use serde::Deserialize;
+use tokio::sync::broadcast;
 
 #[cfg(target_os = "linux")]
-mod linux;
+/// Linux-specific observer functionality for cgroups and procfs.
+pub mod linux;
+
+/// Type used to receive the target PID once it is running.
+pub type TargetPidReceiver = broadcast::Receiver<Option<i32>>;
 
 #[derive(thiserror::Error, Debug)]
 /// Errors produced by [`Server`]
@@ -38,13 +44,13 @@ pub enum Error {
 pub struct Config {}
 
 #[derive(Debug)]
-/// The inspector sub-process server.
+/// The observer sub-process server.
 ///
 /// This struct manages a sub-process that can be used to do further examination
-/// of the [`crate::target::Server`] by means of operating system facilities. The
-/// sub-process is not created until [`Server::run`] is called. It is assumed
-/// that only one instance of this struct will ever exist at a time, although
-/// there are no protections for that.
+/// of a target process by means of operating system facilities. The sub-process
+/// is not created until [`Server::run`] is called. It is assumed that only one
+/// instance of this struct will ever exist at a time, although there are no
+/// protections for that.
 pub struct Server {
     #[allow(dead_code)] // config is not actively used, left as a stub
     config: Config,
@@ -93,10 +99,8 @@ impl Server {
     pub async fn run(
         self,
         mut pid_snd: TargetPidReceiver,
-        sample_period: std::time::Duration,
+        sample_period: Duration,
     ) -> Result<(), Error> {
-        use crate::observer::linux::Sampler;
-
         let target_pid = pid_snd
             .recv()
             .await
@@ -143,7 +147,7 @@ impl Server {
     pub async fn run(
         self,
         _pid_snd: TargetPidReceiver,
-        _sample_period: std::time::Duration,
+        _sample_period: Duration,
     ) -> Result<(), Error> {
         tracing::warn!("observer unavailable on non-Linux system");
         Ok(())

lading/src/observer/linux.rs → lading_observer/src/linux.rs

@@ -1,9 +1,11 @@
-mod cgroup;
-mod procfs;
+/// Cgroup v2 metrics collection.
+pub mod cgroup;
+/// Procfs metrics collection.
+pub mod procfs;
 mod utils;
 mod wss;
 
-use tracing::{error, warn};
+use tracing::warn;
 
 #[derive(thiserror::Error, Debug)]
 /// Errors produced by functions in this module

lading/src/observer/linux/cgroup.rs → lading_observer/src/linux/cgroup.rs

@@ -1,5 +1,5 @@
 /// Code to read cgroup information.
-pub(crate) mod v2;
+pub mod v2;
 
 use std::{collections::VecDeque, io, path::PathBuf};
 

lading/src/observer/linux/cgroup/v2.rs → lading_observer/src/linux/cgroup/v2.rs

@@ -13,15 +13,21 @@
 use tracing::{debug, error, warn};
 
 #[derive(thiserror::Error, Debug)]
+/// Errors that can occur during cgroup v2 operations.
 pub enum Error {
+    /// I/O error.
     #[error("IO error: {0}")]
     Io(#[from] std::io::Error),
+    /// Integer parsing error.
     #[error("Parse int error: {0}")]
     ParseInt(#[from] std::num::ParseIntError),
+    /// Float parsing error.
     #[error("Parse float error: {0}")]
     ParseFloat(#[from] std::num::ParseFloatError),
+    /// Cgroup v2 not found.
     #[error("Cgroup v2 not found")]
     CgroupV2NotFound,
+    /// PSI parsing error.
     #[error("Parsing PSI error: {0}")]
     ParsingPsi(String),
 }
@@ -52,7 +58,7 @@
 /// Polls for any cgroup metrics that can be read, v2 version.
 #[tracing::instrument(skip_all)]
 #[allow(clippy::too_many_lines)]
-pub(crate) async fn poll(file_path: &Path, labels: &[(String, String)]) -> Result<(), Error> {
+pub async fn poll(file_path: &Path, labels: &[(String, String)]) -> Result<(), Error> {
     // Read all files in the cgroup `path` and create metrics for them. If we
     // lack permissions to read we skip the file. We do not use ? to allow for
     // the maximal number of files to be read.

lading/src/observer/linux/procfs.rs → lading_observer/src/linux/procfs.rs

@@ -1,5 +1,5 @@
 /// Sampler implementation for procfs filesystems
-mod memory;
+pub mod memory;
 mod stat;
 mod uptime;
 mod vmstat;
@@ -14,7 +14,7 @@
 use rustc_hash::FxHashMap;
 use tracing::{error, warn};
 
-use crate::observer::linux::utils::process_descendents::ProcessDescendantsIterator;
+use crate::linux::utils::process_descendents::ProcessDescendantsIterator;
 
 const BYTES_PER_KIBIBYTE: u64 = 1024;
 
@@ -410,10 +410,10 @@
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use proptest::prelude::*;
 
     prop_compose! {
         /// Generate a valid executable path
         fn arb_exe_path()(
             components in prop::collection::vec("[a-zA-Z0-9_-]+", 1..5),
@@ -422,7 +422,7 @@
         }
     }
 
     prop_compose! {
         /// Generate a command line with arguments
         fn arb_cmdline()(
             cmd in "[a-zA-Z0-9_-]+",
@@ -436,7 +436,7 @@
         }
     }
 
     prop_compose! {
         /// Generate process info for testing
         fn arb_process_info()(
             exe in arb_exe_path(),
@@ -454,7 +454,7 @@
         }
     }
 
     proptest! {
         #[test]
         fn identical_processes_are_forked_but_not_execed(
             exe in arb_exe_path(),

lading/src/observer/linux/procfs/memory.rs → lading_observer/src/linux/procfs/memory.rs

@@ -1,5 +1,6 @@
 pub(crate) mod smaps;
-pub(crate) mod smaps_rollup;
+/// Memory metrics from /proc/[pid]/smaps_rollup.
+pub mod smaps_rollup;
 
 const BYTES_PER_KIBIBYTE: u64 = 1024;
 

lading/src/observer/linux/procfs/memory/smaps.rs → lading_observer/src/linux/procfs/memory/smaps.rs

@@ -3,7 +3,7 @@ use std::{fs, io::Read};
 use regex::Regex;
 use rustc_hash::FxHashMap;
 
-use crate::observer::linux::procfs::BYTES_PER_KIBIBYTE;
+use crate::linux::procfs::BYTES_PER_KIBIBYTE;
 
 use super::next_token;
 

lading/src/observer/linux/procfs/memory/smaps_rollup.rs → lading_observer/src/linux/procfs/memory/smaps_rollup.rs

@@ -6,24 +6,33 @@
 
 #[derive(thiserror::Error, Debug)]
 /// Errors produced by functions in this module
-pub(crate) enum Error {
+pub enum Error {
     /// Wrapper for [`std::io::Error`]
     #[error("IO error: {0}")]
     Io(#[from] std::io::Error),
+    /// Integer parsing error.
     #[error("Number Parsing: {0}")]
     ParseInt(#[from] std::num::ParseIntError),
+    /// Parsing error.
     #[error("Parsing: {0}")]
     Parsing(String),
 }
 
 #[derive(Debug, Clone, Copy, Default)]
-pub(crate) struct Aggregator {
-    pub(crate) rss: u64,
-    pub(crate) pss: u64,
+/// Aggregator for memory metrics from smaps_rollup.
+pub struct Aggregator {
+    /// Resident Set Size in bytes.
+    pub rss: u64,
+    /// Proportional Set Size in bytes.
+    pub pss: u64,
 }
 
-// Read `/proc/{pid}/smaps_rollup` and parse it directly into metrics.
-pub(crate) async fn poll(
+/// Read `/proc/{pid}/smaps_rollup` and parse it directly into metrics.
+///
+/// # Errors
+///
+/// Returns an error if the file cannot be read or parsed.
+pub async fn poll(
     pid: i32,
     labels: &[(&'static str, String)],
     aggr: &mut Aggregator,

lading/src/observer/linux/procfs/stat.rs → lading_observer/src/linux/procfs/stat.rs

@@ -1,7 +1,7 @@
 use metrics::gauge;
 use tokio::fs;
 
-use crate::observer::linux::cgroup;
+use crate::linux::cgroup;
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {

lading_observer/src/linux/utils.rs

@@ -0,0 +1 @@
+pub(in crate::linux) mod process_descendents;