From 57f9354e62fc05e4d906f64a8e30b8df57d8f019 Mon Sep 17 00:00:00 2001 From: CongkaiTan Date: Thu, 16 Jul 2026 12:55:41 -0700 Subject: [PATCH] Introduce AperfStatsCollector and extend collection of APerf stats --- src/data.rs | 71 ------- src/data/aperf_stats.rs | 57 +++++- src/data/common/time_series_data_processor.rs | 29 ++- src/data_collection.rs | 187 ++++++++++++------ src/lib.rs | 161 +++++++++++++-- src/record.rs | 9 + tests/test_aperf_stats.rs | 18 +- 7 files changed, 373 insertions(+), 159 deletions(-) diff --git a/src/data.rs b/src/data.rs index 58da9cdf..c8d59862 100644 --- a/src/data.rs +++ b/src/data.rs @@ -379,79 +379,8 @@ pub trait AnalyzeData { #[cfg(test)] mod tests { - #[cfg(target_os = "linux")] - use crate::data_collection::DataCollector; - use super::TimeEnum; use chrono::prelude::*; - #[cfg(target_os = "linux")] - use { - super::cpu_utilization::CpuUtilizationRaw, super::Data, crate::data_file_path, std::fs, - std::path::PathBuf, - }; - - #[cfg(target_os = "linux")] - #[test] - fn test_data_type_init() { - let run_data_dir = PathBuf::from("./performance_data_init_test"); - fs::DirBuilder::new() - .recursive(true) - .create(&run_data_dir) - .unwrap(); - - // Constructing a DataCollector creates and opens the data file. - let data = CpuUtilizationRaw::new(); - let _dc = DataCollector::new( - "cpu_utilization", - Data::CpuUtilizationRaw(data), - &run_data_dir, - ); - - let expected_path = data_file_path("cpu_utilization", &run_data_dir); - assert!(expected_path.exists()); - - fs::remove_dir_all(&run_data_dir).unwrap(); - } - - #[cfg(target_os = "linux")] - #[test] - fn test_print() { - let run_data_dir = PathBuf::from("./performance_data_print_test"); - fs::DirBuilder::new() - .recursive(true) - .create(&run_data_dir) - .unwrap(); - - let data = CpuUtilizationRaw::new(); - let mut dc = DataCollector::new( - "cpu_utilization", - Data::CpuUtilizationRaw(data), - &run_data_dir, - ); - - let data_file_path = data_file_path("cpu_utilization", &run_data_dir); - assert!(data_file_path.exists()); - dc.write_to_file().unwrap(); - - // Re-open the file to read back what was serialized (the collector's own handle is in - // append mode). - let read_handle = fs::File::open(&data_file_path).unwrap(); - loop { - match bincode::deserialize_from::<_, Data>(&read_handle) { - Ok(v) => match v { - Data::CpuUtilizationRaw(ref value) => assert!(value.data.is_empty()), - _ => unreachable!(), - }, - Err(e) => match *e { - bincode::ErrorKind::Io(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { - break - } - _ => unreachable!(), - }, - }; - } - fs::remove_dir_all(&run_data_dir).unwrap(); - } #[test] fn test_time_diff_second() { diff --git a/src/data/aperf_stats.rs b/src/data/aperf_stats.rs index bb8fd518..69c87394 100644 --- a/src/data/aperf_stats.rs +++ b/src/data/aperf_stats.rs @@ -24,6 +24,14 @@ impl AperfStat { } } + pub fn for_time(time: TimeEnum) -> Self { + AperfStat { + time, + name: String::new(), + data: HashMap::new(), + } + } + pub fn measure(&mut self, name: String, mut func: F) -> Result<()> where F: FnMut() -> Result<()>, @@ -69,20 +77,49 @@ impl ProcessData for AperfStat { }; } + let mut collection_started = false; for value in values { - time_series_data_processor.proceed_to_time(value.time); + // Ignore the time diff for data before the collection started, as time_diff + // computation would have been corrupted and these data are not time-series anyway. + if !collection_started + && report_params + .collection_start + .map_or(true, |collection_start_time| { + value.time >= collection_start_time + }) + { + collection_started = true; + } + if collection_started { + time_series_data_processor.proceed_to_time(value.time); + } - for (name, stat) in value.data { - let datatype: Vec<&str> = name.split('-').collect(); - let metric_name = datatype[0]; - let mut series_name = datatype.get(1).unwrap_or(&metric_name).to_string(); - // Make the series name easier to understand - since it's essentially to write the - // collected data to disk - if series_name == "print" { - series_name = "write".to_string(); + for (stat_key, stat_value) in value.data { + let stat_key_components: Vec<&str> = stat_key.split('-').collect(); + let data_name = stat_key_components[0]; + let mut stat_name = stat_key_components.get(1).unwrap_or(&data_name).to_string(); + // Backward compatibility - the previous stat name was "print" + if stat_name == "print" { + stat_name = "write".to_string(); } - time_series_data_processor.add_data_point(metric_name, &series_name, stat as f64); + if stat_name == "collect" || stat_name == "write" { + // The stats for APerf collecting time-series data, which should be + // processed as regular time-series data as well. + time_series_data_processor.add_data_point( + data_name, + &stat_name, + stat_value as f64, + ); + } else { + // For non-time-series stats, group them in a dummy metric by stat_name as different + // series by data_name. + time_series_data_processor.add_data_point( + &stat_name, + data_name, + stat_value as f64, + ); + } } } diff --git a/src/data/common/time_series_data_processor.rs b/src/data/common/time_series_data_processor.rs index 4b46717f..95e2e145 100644 --- a/src/data/common/time_series_data_processor.rs +++ b/src/data/common/time_series_data_processor.rs @@ -164,12 +164,7 @@ impl TimeSeriesDataProcessor { series.values.push(metric_value); // Every series value in the metric accounts for its value range - let (min, max) = self - .per_metric_value_range - .entry(metric_name.to_string()) - .or_insert((metric_value, metric_value)); - *min = (*min).min(metric_value); - *max = (*max).max(metric_value); + update_metric_value_range(&mut self.per_metric_value_range, metric_name, metric_value); if self.aggregate_mode == TimeSeriesDataAggregateMode::Average || self.aggregate_mode == TimeSeriesDataAggregateMode::Sum @@ -243,7 +238,7 @@ impl TimeSeriesDataProcessor { _ => "aggregate", }); - for (metric_name, (sum, count)) in &mut self.per_metric_sum_count { + for (metric_name, (sum, count)) in &self.per_metric_sum_count { let aggregate_series = self .per_metric_series .entry(metric_name.clone()) @@ -259,7 +254,10 @@ impl TimeSeriesDataProcessor { 0.0 } } - TimeSeriesDataAggregateMode::Sum => *sum, + TimeSeriesDataAggregateMode::Sum => { + update_metric_value_range(&mut self.per_metric_value_range, metric_name, *sum); + *sum + } _ => break, }; aggregate_series.values.push(aggregate_value); @@ -397,6 +395,18 @@ impl TimeSeriesDataProcessor { } } +fn update_metric_value_range( + per_metric_value_range: &mut HashMap, + metric_name: &str, + metric_value: f64, +) { + let (min, max) = per_metric_value_range + .entry(metric_name.to_string()) + .or_insert_with(|| (metric_value, metric_value)); + *min = (*min).min(metric_value); + *max = (*max).max(metric_value); +} + fn compress_all_zero_time_series_metric(time_series_metric: &mut TimeSeriesMetric) { if time_series_metric.stats.min == 0.0 && time_series_metric.stats.max == 0.0 { for series in &mut time_series_metric.series { @@ -806,6 +816,9 @@ mod tests { // Sum: 10+20+30=60, 40+50+60=150 assert_eq!(agg.values, vec![60.0, 150.0]); assert_eq!(agg.series_name.as_str(), "sum"); + // The value range must account for the aggregate sum (max 150), not just the + // individual series data points (max 60). + assert_eq!(ts.metrics["m"].value_range, (10, 150)); } #[test] diff --git a/src/data_collection.rs b/src/data_collection.rs index 2e415992..51a4998d 100644 --- a/src/data_collection.rs +++ b/src/data_collection.rs @@ -1,20 +1,20 @@ -use crate::data::aperf_stats::AperfStat; use crate::data::TimeEnum; use crate::PDError; -#[cfg(target_os = "linux")] -use crate::{aperf_runlog_file_path, data_file_path, AperfStatsWriter}; use crate::{APERF_TMP, GROUPED_PMU_MODE}; use anyhow::Result; use chrono::prelude::*; use log::{debug, error, info}; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::fs; use std::path::PathBuf; use std::time::Instant; #[cfg(target_os = "linux")] use { + crate::data::processes::ProcessesRaw, crate::data::Data, + crate::{aperf_runlog_file_path, data_file_path, get_data_name_from_type}, + crate::{aperf_stats_add, aperf_stats_measure, aperf_stats_proceed_to_next_stats}, nix::poll::{poll, PollFd, PollFlags, PollTimeout}, nix::sys::{ signal, @@ -22,7 +22,7 @@ use { }, std::fs::{File, OpenOptions}, std::os::unix::io::AsFd, - std::{process, time}, + std::time, timerfd::{SetTimeFlags, TimerFd, TimerState}, }; @@ -30,18 +30,14 @@ use { pub struct DataCollectionEngine { init_params: InitParams, data_collectors: HashMap, - aperf_stats_writer: AperfStatsWriter, } #[cfg(target_os = "linux")] impl DataCollectionEngine { pub fn new(init_params: InitParams) -> Self { - let aperf_stats_writer = AperfStatsWriter::new(&init_params.run_data_dir).unwrap(); - DataCollectionEngine { init_params, data_collectors: HashMap::new(), - aperf_stats_writer, } } @@ -65,7 +61,7 @@ impl DataCollectionEngine { // (perf_profile, java_profile) so they start as close to collect_data_serial as // possible and stay in sync with the collection period. for is_profile_pass in [false, true] { - for (name, data_collector) in self.data_collectors.iter_mut() { + for (data_name, data_collector) in self.data_collectors.iter_mut() { if data_collector.is_static() || data_collector.is_profile() != is_profile_pass { continue; } @@ -75,28 +71,23 @@ impl DataCollectionEngine { self.init_params.expected_end_time = Instant::now() + time::Duration::from_secs(self.init_params.period); - match data_collector.prepare_data_collector(&self.init_params) { - Err(e) => { - if data_collector.is_profile() { - error!("{}", e.to_string()); - error!("Aperf exiting..."); - process::exit(1); - } - let msg = format!( - "Excluding {} from collection, data preparation failed: {:?}", - name, e - ); - if matches!( - e.downcast_ref::(), - Some(PDError::IgnoredDataPreparationError(_)) - ) { - debug!("{}", msg); - } else { - error!("{}", msg); - } - remove_entries.push(name.clone()); + if let Err(e) = data_collector.prepare_data_collector(&self.init_params) { + if data_collector.is_profile() { + panic!("{}", e.to_string()); + } + let msg = format!( + "Excluding {} from collection, data preparation failed: {:?}", + data_name, e + ); + if matches!( + e.downcast_ref::(), + Some(PDError::IgnoredDataPreparationError(_)) + ) { + debug!("{}", msg); + } else { + error!("{}", msg); } - _ => continue, + remove_entries.push(data_name.clone()); } } } @@ -122,7 +113,9 @@ impl DataCollectionEngine { pub fn collect_data_serial(&mut self) -> Result<()> { let start_time = time::Instant::now(); - self.init_params.collection_start = Some(TimeEnum::DateTime(Utc::now())); + let collection_start_time = TimeEnum::DateTime(Utc::now()); + self.init_params.collection_start = Some(collection_start_time); + aperf_stats_proceed_to_next_stats(collection_start_time); let end_time = start_time + time::Duration::from_secs(self.init_params.period); self.init_params.expected_end_time = end_time; @@ -148,7 +141,6 @@ impl DataCollectionEngine { let mut poll_fds = [timer_pollfd, signal_pollfd]; let mut end_signal = String::new(); - let mut aperf_stats = AperfStat::new(); let mut current_time = start_time; while current_time <= end_time { @@ -165,33 +157,23 @@ impl DataCollectionEngine { debug!("Time elapsed: {:?}", start_time.elapsed()); let cur_collection_start = time::Instant::now(); - aperf_stats.time = TimeEnum::DateTime(Utc::now()); - aperf_stats.data = HashMap::new(); - for (name, data_collector) in self.data_collectors.iter_mut() { + for data_collector in self.data_collectors.values_mut() { if data_collector.is_static() { continue; } - - aperf_stats.measure(name.clone() + "-collect", || -> Result<()> { - data_collector.collect_data(&self.init_params)?; - Ok(()) - })?; - aperf_stats.measure(name.clone() + "-print", || -> Result<()> { - data_collector.write_to_file()?; - Ok(()) - })?; + data_collector.collect_data(&self.init_params)?; + data_collector.write_to_file()?; } let cur_collection_end = time::Instant::now(); let cur_collection_time = cur_collection_end - cur_collection_start; - aperf_stats - .data - .insert("aperf".to_string(), cur_collection_time.as_micros() as u64); + aperf_stats_add( + "aperf-collect".to_string(), + cur_collection_time.as_micros() as u64, + ); debug!("Collection time: {:?}", cur_collection_time); - self.aperf_stats_writer.write(&aperf_stats)?; - current_time = cur_collection_end; } } @@ -238,6 +220,16 @@ impl DataCollectionEngine { error!("Failed to save run metadata: {e}"); } + // Conduct another round of processes data collection to collect APerf performance + // data during the finish stage. + if let Some(processes_data_collector) = self + .data_collectors + .get_mut(get_data_name_from_type::()) + { + processes_data_collector.collect_data(&self.init_params)?; + processes_data_collector.write_to_file()?; + }; + Ok(()) } } @@ -283,22 +275,44 @@ impl DataCollector { } pub fn prepare_data_collector(&mut self, init_params: &InitParams) -> Result<()> { - self.data.prepare_data_collector(init_params)?; + aperf_stats_measure(format!("{}-prepare", self.data_name), || -> Result<()> { + self.data.prepare_data_collector(init_params)?; + Ok(()) + })?; Ok(()) } pub fn collect_data(&mut self, init_params: &InitParams) -> Result<()> { - self.data.collect_data(init_params)?; + let aperf_stat_name = format!( + "{}-{}collect", + self.data_name, + if self.is_static() { "static_" } else { "" } + ); + aperf_stats_measure(aperf_stat_name, || -> Result<()> { + self.data.collect_data(init_params)?; + Ok(()) + })?; Ok(()) } pub fn write_to_file(&mut self) -> Result<()> { - bincode::serialize_into(&mut self.data_file, &self.data)?; + let aperf_stat_name = format!( + "{}-{}write", + self.data_name, + if self.is_static() { "static_" } else { "" } + ); + aperf_stats_measure(aperf_stat_name, || -> Result<()> { + bincode::serialize_into(&mut self.data_file, &self.data)?; + Ok(()) + })?; Ok(()) } pub fn finish_data_collection(&mut self, init_params: &InitParams) -> Result<()> { - self.data.finish_data_collection(init_params)?; + aperf_stats_measure(format!("{}-finish", self.data_name), || -> Result<()> { + self.data.finish_data_collection(init_params)?; + Ok(()) + })?; Ok(()) } } @@ -339,6 +353,8 @@ pub struct InitParams { /// for archives produced by versions of aperf that did not record this. #[serde(default)] pub pid: Option, + /// The PIDs of all processes launched during data collection. + pub sub_process_pids: HashSet, /// The signal that ends the collection. An empty string means the collection /// followed the specified period and ended naturally. pub end_signal: String, @@ -369,6 +385,7 @@ impl InitParams { collection_start: None, collection_end: None, pid: Some(std::process::id()), + sub_process_pids: HashSet::new(), end_signal: String::new(), expected_end_time: Instant::now(), } @@ -403,7 +420,12 @@ impl Default for InitParams { #[cfg(test)] mod tests { #[cfg(target_os = "linux")] - use super::{DataCollectionEngine, InitParams}; + use { + super::{DataCollectionEngine, DataCollector, InitParams}, + crate::data::cpu_utilization::CpuUtilizationRaw, + crate::data::Data, + crate::data_file_path, + }; #[cfg(target_os = "linux")] #[test] @@ -421,4 +443,59 @@ mod tests { run_data_dir ); } + + #[cfg(target_os = "linux")] + #[test] + fn test_data_collector_init() { + let temp_dir = tempfile::tempdir().unwrap(); + let run_data_dir = temp_dir.path().to_path_buf(); + + // Constructing a DataCollector creates and opens the data file. + let data = CpuUtilizationRaw::new(); + let _dc = DataCollector::new( + "cpu_utilization", + Data::CpuUtilizationRaw(data), + &run_data_dir, + ); + + let expected_path = data_file_path("cpu_utilization", &run_data_dir); + assert!(expected_path.exists()); + } + + #[cfg(target_os = "linux")] + #[test] + fn test_write() { + let temp_dir = tempfile::tempdir().unwrap(); + let run_data_dir = temp_dir.path().to_path_buf(); + + let data = CpuUtilizationRaw::new(); + let mut dc = DataCollector::new( + "cpu_utilization", + Data::CpuUtilizationRaw(data), + &run_data_dir, + ); + + let data_file_path = data_file_path("cpu_utilization", &run_data_dir); + assert!(data_file_path.exists()); + + dc.write_to_file().unwrap(); + + // Re-open the file to read back what was serialized (the collector's own handle is in + // append mode). + let read_handle = std::fs::File::open(&data_file_path).unwrap(); + loop { + match bincode::deserialize_from::<_, Data>(&read_handle) { + Ok(v) => match v { + Data::CpuUtilizationRaw(ref value) => assert!(value.data.is_empty()), + _ => unreachable!(), + }, + Err(e) => match *e { + bincode::ErrorKind::Io(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { + break + } + _ => unreachable!(), + }, + }; + } + } } diff --git a/src/lib.rs b/src/lib.rs index 4cfa8cad..2c3113e7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,12 +17,14 @@ pub mod report; pub mod server; use crate::data::aperf_runlog::AperfRunlog; -use crate::data::aperf_stats::AperfStat; +use crate::data::TimeEnum; use anyhow::{bail, Result}; use regex::Regex; -use std::fs::{self, File}; +use std::fs; use std::path::PathBuf; use thiserror::Error; +#[cfg(target_os = "linux")] +use {crate::data::aperf_stats::AperfStat, chrono::Utc, log::error, std::cell::RefCell}; pub const APERF_FILE_FORMAT: &str = "bin"; @@ -152,16 +154,134 @@ pub fn aperf_runlog_file_path(run_data_dir: &PathBuf) -> PathBuf { run_data_dir.join(get_data_name_from_type::()) } -pub struct AperfStatsWriter { - aperf_stats_file: File, +#[cfg(target_os = "linux")] +thread_local! { + static APERF_STATS_COLLECTOR: RefCell = RefCell::new(AperfStatsCollector::new()); +} + +#[cfg(target_os = "linux")] +pub fn aperf_stats_initialize(run_data_dir: PathBuf) { + APERF_STATS_COLLECTOR.with(|aperf_stats_collector| { + aperf_stats_collector.borrow_mut().initialize(run_data_dir); + }); +} + +#[cfg(target_os = "linux")] +pub fn aperf_stats_proceed_to_next_stats(next_stats_time: TimeEnum) { + APERF_STATS_COLLECTOR.with(|aperf_stats_collector| { + aperf_stats_collector + .borrow_mut() + .proceed_to_next_stats(next_stats_time); + }); +} + +#[cfg(target_os = "linux")] +pub fn aperf_stats_measure(stat_name: String, func: F) -> Result<()> +where + F: FnMut() -> Result<()>, +{ + APERF_STATS_COLLECTOR.with(|aperf_stats_collector| { + aperf_stats_collector.borrow_mut().measure(stat_name, func) + })?; + + Ok(()) +} + +#[cfg(target_os = "linux")] +pub fn aperf_stats_add(stat_key: String, stat_value: u64) { + APERF_STATS_COLLECTOR.with(|aperf_stats_collector| { + aperf_stats_collector + .borrow_mut() + .add_stat(stat_key, stat_value); + }); } -impl AperfStatsWriter { - pub fn new(run_data_dir: &PathBuf) -> Result { - let aperf_stats_file_path = - data_file_path(get_data_name_from_type::(), run_data_dir); - let aperf_stats_file = match fs::OpenOptions::new() + +#[cfg(target_os = "linux")] +pub fn aperf_stats_flush() -> Result<()> { + APERF_STATS_COLLECTOR + .with(|aperf_stats_collector| aperf_stats_collector.borrow_mut().flush())?; + + Ok(()) +} + +/// Encapsulate all logics of collecting and writing APerf stats. +/// The collected stats will be saved in memory in time order and be written to +/// disk when flush() is called. +#[cfg(target_os = "linux")] +pub struct AperfStatsCollector { + cur_aperf_stats: AperfStat, + time_series_aperf_stats: Vec, + run_data_dir: Option, +} + +#[cfg(target_os = "linux")] +impl AperfStatsCollector { + pub fn new() -> Self { + Self { + cur_aperf_stats: AperfStat::new(), + time_series_aperf_stats: Vec::new(), + run_data_dir: None, + } + } + + pub fn initialize(&mut self, run_data_dir: PathBuf) { + self.run_data_dir = Some(run_data_dir); + } + + /// Check current time and if at next second, save the current stats and + /// proceed with a new empty stats. + /// If we get to a point where the stats is big and we want to limit APerf's + /// memory usage, we can also flush here. + fn update_time_series(&mut self) { + let cur_time = TimeEnum::DateTime(Utc::now()); + let cur_time_diff = match cur_time - self.cur_aperf_stats.time { + TimeEnum::TimeDiff(time_diff) => time_diff, + _ => return, + }; + if cur_time_diff >= 1 { + self.proceed_to_next_stats(cur_time); + } + } + + /// Save current stats and proceed to the next new stats. + fn proceed_to_next_stats(&mut self, next_stats_time: TimeEnum) { + let cur_aperf_stats = std::mem::replace( + &mut self.cur_aperf_stats, + AperfStat::for_time(next_stats_time), + ); + self.time_series_aperf_stats.push(cur_aperf_stats); + } + + /// Measure the wall-clock time of executing a function and save as a stat. + pub fn measure(&mut self, stat_name: String, func: F) -> Result<()> + where + F: FnMut() -> Result<()>, + { + self.update_time_series(); + + self.cur_aperf_stats.measure(stat_name, func) + } + + /// Save a stat. + pub fn add_stat(&mut self, stat_key: String, stat_value: u64) { + self.update_time_series(); + + self.cur_aperf_stats.data.insert(stat_key, stat_value); + } + + /// Write all saved stats to disk file. + pub fn flush(&mut self) -> Result<()> { + if self.run_data_dir.is_none() { + bail!("Failed to flush APerf stat since the run data directory path is uninitialized."); + } + + let aperf_stats_file_path = data_file_path( + get_data_name_from_type::(), + self.run_data_dir.as_ref().unwrap(), + ); + let mut aperf_stats_file = match fs::OpenOptions::new() .create(true) - .write(true) + .append(true) .open(&aperf_stats_file_path) { Ok(aperf_stats_file) => aperf_stats_file, @@ -172,16 +292,29 @@ impl AperfStatsWriter { ), }; - Ok(Self { aperf_stats_file }) - } + for aperf_stats in &self.time_series_aperf_stats { + bincode::serialize_into(&mut aperf_stats_file, aperf_stats)?; + } + if !self.cur_aperf_stats.data.is_empty() { + bincode::serialize_into(&mut aperf_stats_file, &self.cur_aperf_stats)?; + } - pub fn write(&mut self, aperf_stats: &AperfStat) -> Result<()> { - bincode::serialize_into(&mut self.aperf_stats_file, aperf_stats)?; + self.time_series_aperf_stats.clear(); + self.cur_aperf_stats = AperfStat::new(); Ok(()) } } +#[cfg(target_os = "linux")] +impl Drop for AperfStatsCollector { + fn drop(&mut self) { + if let Err(e) = self.flush() { + error!("Failed to flush APerf stats on drop: {e}"); + } + } +} + #[cfg(test)] mod test { use super::find_file; diff --git a/src/record.rs b/src/record.rs index fad41a4f..66ed6e52 100644 --- a/src/record.rs +++ b/src/record.rs @@ -1,5 +1,7 @@ #![cfg(target_os = "linux")] +use crate::aperf_stats_flush; +use crate::aperf_stats_initialize; use crate::data; use crate::data::java_profile::JavaProfile; use crate::data_collection::DataCollectionEngine; @@ -173,6 +175,8 @@ pub fn record(record: &Record, tmp_dir: &Path, runlog: &Path) -> Result<()> { } }; + aperf_stats_initialize(run_data_dir.clone()); + let mut init_params = InitParams::new(run_name, run_data_dir.clone()); init_params.period = record.period; init_params.interval = record.interval; @@ -235,6 +239,11 @@ pub fn record(record: &Record, tmp_dir: &Path, runlog: &Path) -> Result<()> { info!("Finishing data collection..."); data_collection_engine.finish_data_collection()?; info!("Data collection complete."); + + if let Err(e) = aperf_stats_flush() { + error!("Failed to write APerf stats: {e}"); + } + info!("Creating run data archive..."); create_run_data_archive(&run_data_dir)?; diff --git a/tests/test_aperf_stats.rs b/tests/test_aperf_stats.rs index dc5b2be0..e8e74b4e 100644 --- a/tests/test_aperf_stats.rs +++ b/tests/test_aperf_stats.rs @@ -216,6 +216,11 @@ mod aperf_stats_tests { stats .stats .insert("aperf".to_string(), 5000 + (sample * 500)); + // A non-collect/write stat (e.g. prepare) is grouped into a dummy metric named + // after the stat name, with a series per data name. + stats + .stats + .insert("cpu_utilization-prepare".to_string(), 100 + (sample * 10)); expected_stats.push(stats); } @@ -231,7 +236,8 @@ mod aperf_stats_tests { let result = aperf_stats.process_raw_data(¶ms, vec![]).unwrap(); if let AperfData::TimeSeries(time_series_data) = result { - assert_eq!(time_series_data.metrics.len(), 3); // aperf + cpu_utilization + diskstats + // aperf + cpu_utilization + diskstats + the "prepare" dummy metric + assert_eq!(time_series_data.metrics.len(), 4); // Validate time progression (1-second intervals) let aperf_metric = &time_series_data.metrics["aperf"]; @@ -239,6 +245,16 @@ mod aperf_stats_tests { // Validate values assert_eq!(aperf_metric.series[0].values, vec![5000.0, 5500.0, 6000.0]); + + // The prepare stat is grouped under a "prepare" metric, keyed by data name as the + // series name (single series, so the aggregate is stripped). + let prepare_metric = &time_series_data.metrics["prepare"]; + assert_eq!(prepare_metric.series.len(), 1); + assert_eq!( + prepare_metric.series[0].series_name.as_str(), + "cpu_utilization" + ); + assert_eq!(prepare_metric.series[0].values, vec![100.0, 110.0, 120.0]); } else { panic!("Expected TimeSeries data"); }