From 290f3ca6d7f6d369e60f5637fec7e3efb69ae273 Mon Sep 17 00:00:00 2001 From: nashzhou Date: Sat, 10 Oct 2026 09:36:59 +0800 Subject: [PATCH] feat(metrics): add sha256 and zfile compression cost metrics Instrument the SHA-256 and LZ4 compression hot paths so their CPU cost and throughput are visible on the server's /metrics endpoint. New series ---------- agentenv_sha256_duration_seconds{op} histogram; pure Sha256::update time, reads and lock waits excluded agentenv_sha256_bytes_total{op} counter; bytes hashed op=file_describe file-level digests via src/digest.rs (snapshot repository publication, image cache, recontainerize) op=copy_verify copy_with_sha256_hex (posixfs publish/verify) op=rw_append guest writes absorbed into the LSMT RW-layer digest op=rw_rebuild full-file rehash when a non-empty RW layer is reopened (resume/restore path) agentenv_overlaybd_zfile_compress_duration_seconds{codec=lz4} agentenv_overlaybd_zfile_compress_input_bytes_total{codec=lz4} agentenv_overlaybd_zfile_compress_output_bytes_total{codec=lz4} Recorded at Compressor::compress, the choke point of every encoding path (ZFileBuilder, worker pool, compress_batch). The byte counters are exact. The duration covers only the codec call (CRC generation is excluded, mirroring the read path's decode-only timing) and is fed by a 1-in-1024 per-thread block sampler like the pread sampler, so its _count is a SAMPLE COUNT, not a block count. How to use ---------- Scrape the server's /metrics endpoint with the existing Prometheus exporter. Series register lazily on first activity, so an absent series means that path has not run yet. Per-op hash cost and throughput (bytes/s): sum by (op) (rate(agentenv_sha256_duration_seconds_sum[5m])) sum by (op) (rate(agentenv_sha256_bytes_total[5m])) / sum by (op) (rate(agentenv_sha256_duration_seconds_sum[5m])) Compression ratio from the exact counters (no sampling correction): sum(rate(agentenv_overlaybd_zfile_compress_input_bytes_total[5m])) / sum(rate(agentenv_overlaybd_zfile_compress_output_bytes_total[5m])) Estimated total codec time and throughput (scale the sampled _sum by 1024, like the decompress estimate below): sum(rate(agentenv_overlaybd_zfile_compress_duration_seconds_sum[5m])) * 1024 sum(rate(agentenv_overlaybd_zfile_compress_input_bytes_total[5m])) / (sum(rate(agentenv_overlaybd_zfile_compress_duration_seconds_sum[5m])) * 1024) Decompression stays covered by the existing read-path series (no code change here): agentenv_overlaybd_zfile_output_bytes_total and agentenv_overlaybd_zfile_encoded_bytes_total are exact counters, while agentenv_overlaybd_zfile_decompress_duration_seconds is fed by the 1-in-1024 pread sampler, so its _count is a sample count. Estimate the true decompress total and throughput as: sum(rate(agentenv_overlaybd_zfile_decompress_duration_seconds_sum[5m])) * 1024 rate(agentenv_overlaybd_zfile_output_bytes_total[5m]) / (sum(rate(agentenv_overlaybd_zfile_decompress_duration_seconds_sum[5m])) * 1024) Overhead: metric handles register lazily and are cached, so after the first record the hot paths only pay atomic increments. The sha256 trackers hash 128KiB stretches, so their clock pairs are negligible. The compression path runs once per 4KiB block and pause-time memory conversion is its high-rate caller, so its duration is sampled: an unsampled block pays one thread-local counter check plus the two exact byte-counter increments; the clock pair and histogram push run for 1 block in 1024 only. --- src/digest.rs | 23 +++ storage/overlaybd/src/compression/zfile.rs | 36 ++++- storage/overlaybd/src/lsmt/file/readwrite.rs | 4 + storage/overlaybd/src/lsmt/file/types.rs | 14 ++ storage/overlaybd/src/metrics.rs | 161 ++++++++++++++++++- 5 files changed, 236 insertions(+), 2 deletions(-) diff --git a/src/digest.rs b/src/digest.rs index 7b7aad6..5601a82 100644 --- a/src/digest.rs +++ b/src/digest.rs @@ -1,6 +1,7 @@ use std::fs; use std::io::{self, Read, Write}; use std::path::Path; +use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -8,6 +9,16 @@ use tokio::io::AsyncReadExt; const DIGEST_BUFFER_SIZE: usize = 128 * 1024; +/// Records pure SHA-256 hashing cost on the `agentenv_sha256_*` family also +/// fed by the overlaybd crate's digest trackers, so one query aggregates +/// every hash in the process. `elapsed` counts only `Sha256::update` +/// stretches, excluding file reads, so it isolates hash CPU from I/O wait. +fn record_sha256(op: &'static str, bytes: u64, elapsed: Duration) { + metrics::histogram!("agentenv_sha256_duration_seconds", "op" => op) + .record(elapsed.as_secs_f64()); + metrics::counter!("agentenv_sha256_bytes_total", "op" => op).increment(bytes); +} + /// Content identity and integrity information for a local file. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, Hash)] pub struct FileDigest { @@ -61,6 +72,7 @@ pub fn copy_with_sha256_hex( ) -> io::Result<(String, u64)> { let mut hasher = Sha256::new(); let mut copied = 0_u64; + let mut hashed = Duration::ZERO; let mut buffer = [0_u8; DIGEST_BUFFER_SIZE]; loop { @@ -69,16 +81,20 @@ pub fn copy_with_sha256_hex( break; } destination.write_all(&buffer[..read])?; + let start = Instant::now(); hasher.update(&buffer[..read]); + hashed += start.elapsed(); copied += read as u64; } + record_sha256("copy_verify", copied, hashed); Ok((format!("{:x}", hasher.finalize()), copied)) } fn hash_reader(mut reader: impl Read) -> io::Result<(Sha256, u64)> { let mut hasher = Sha256::new(); let mut size = 0_u64; + let mut hashed = Duration::ZERO; let mut buffer = [0_u8; DIGEST_BUFFER_SIZE]; loop { @@ -86,10 +102,13 @@ fn hash_reader(mut reader: impl Read) -> io::Result<(Sha256, u64)> { if read == 0 { break; } + let start = Instant::now(); hasher.update(&buffer[..read]); + hashed += start.elapsed(); size += read as u64; } + record_sha256("file_describe", size, hashed); Ok((hasher, size)) } @@ -98,6 +117,7 @@ async fn hash_reader_async( ) -> io::Result<(Sha256, u64)> { let mut hasher = Sha256::new(); let mut size = 0_u64; + let mut hashed = Duration::ZERO; let mut buffer = vec![0_u8; DIGEST_BUFFER_SIZE]; // Use a heap buffer to avoid large stack allocations in async contexts. loop { @@ -105,10 +125,13 @@ async fn hash_reader_async( if read == 0 { break; } + let start = Instant::now(); hasher.update(&buffer[..read]); + hashed += start.elapsed(); size += read as u64; } + record_sha256("file_describe", size, hashed); Ok((hasher, size)) } diff --git a/storage/overlaybd/src/compression/zfile.rs b/storage/overlaybd/src/compression/zfile.rs index 5219747..6b1cb2c 100644 --- a/storage/overlaybd/src/compression/zfile.rs +++ b/storage/overlaybd/src/compression/zfile.rs @@ -5,7 +5,9 @@ use crate::io::vfile_io::{read_exact, DirectRead, FileReader}; use crate::io::virtual_file::VirtualFile; #[cfg(feature = "io-uring")] use crate::io::virtual_file::{IoCtx, LocalBoxFuture}; -use crate::metrics::{ZFileCodec, ZFileReadMetrics, ZFileReadStats, ZFileReadStatus}; +use crate::metrics::{ + ZFileCodec, ZFileCompressMetrics, ZFileReadMetrics, ZFileReadStats, ZFileReadStatus, +}; use anyhow::{bail, ensure, Context, Result}; use async_trait::async_trait; use bytes::Bytes; @@ -48,6 +50,29 @@ fn sample_zfile_pread() -> bool { }) } +/// One compressed block in this many per thread is sampled for the +/// duration histogram, mirroring the pread sampler above. Pause-time memory +/// conversion calls `compress` once per 4KiB block, so the exact-rate clock +/// pair and histogram push must not run on every block; an unsampled block +/// pays only a `Cell` increment for its duration. +const ZFILE_COMPRESS_SAMPLE_INTERVAL: u64 = 1024; + +std::thread_local! { + /// Per-thread compressed-block counter driving [`sample_zfile_compress`]. + static ZFILE_COMPRESS_COUNTER: Cell = const { Cell::new(0) }; +} + +/// Return true for one block in [`ZFILE_COMPRESS_SAMPLE_INTERVAL`] on each +/// thread. The compress duration histogram exports a `_count` that is a +/// SAMPLE COUNT, not a block count. +fn sample_zfile_compress() -> bool { + ZFILE_COMPRESS_COUNTER.with(|counter| { + let next = counter.get().wrapping_add(1); + counter.set(next); + next % ZFILE_COMPRESS_SAMPLE_INTERVAL == 0 + }) +} + /// Observation collected over one logical [`ZFileRO::pread`]: the exact /// counters plus, for sampled preads only, the durations. Handed to the /// metrics recorder exactly once per pread by [`ZFileRO::pread_generic`]. @@ -424,6 +449,11 @@ impl Lz4Compressor { } } +/// `agentenv_overlaybd_zfile_compress_*` recorder. Every encoding path +/// (builder, worker pool, `compress_batch`) funnels through +/// [`Compressor::compress`], so this static sees every compressed block. +static LZ4_COMPRESS_METRICS: ZFileCompressMetrics = ZFileCompressMetrics::new(ZFileCodec::Lz4); + impl Compressor for Lz4Compressor { fn max_dst_size(&self) -> usize { self.max_dst_size @@ -442,6 +472,8 @@ impl Compressor for Lz4Compressor { ); // C liblz4 block format (identical to lz4_flex and upstream OverlayBD). // Block sizes are 4KiB-scale, so the c_int casts cannot overflow. + let sampled = sample_zfile_compress(); + let start = sampled.then(Instant::now); let n = unsafe { lz4_sys::LZ4_compress_default( src.as_ptr().cast(), @@ -450,7 +482,9 @@ impl Compressor for Lz4Compressor { dst.len() as i32, ) }; + let elapsed = start.map(|start| start.elapsed()); ensure!(n > 0, "lz4 compression failed"); + LZ4_COMPRESS_METRICS.record(src.len() as u64, n as u64, elapsed); Ok(n as usize) } diff --git a/storage/overlaybd/src/lsmt/file/readwrite.rs b/storage/overlaybd/src/lsmt/file/readwrite.rs index e1a2c26..0cfa5d2 100644 --- a/storage/overlaybd/src/lsmt/file/readwrite.rs +++ b/storage/overlaybd/src/lsmt/file/readwrite.rs @@ -287,6 +287,9 @@ impl LSMTFile { "rw data file is smaller than header while rebuilding append digest" ); let mut tracker = AppendDigestTracker::from_header(rw_data_file).await?; + // Attribute the reopen rehash to the rebuild series, then fall back + // to the steady-state append label for writes on this tracker. + tracker.metrics = &RW_REBUILD_DIGEST_METRICS; let mut offset = HEADER_SIZE; let mut buffer = vec![0u8; 128 * 1024]; while offset < data_file_size { @@ -301,6 +304,7 @@ impl LSMTFile { tracker.absorb(offset, &buffer[..read])?; offset += read as u64; } + tracker.metrics = &RW_APPEND_DIGEST_METRICS; Ok(tracker) } diff --git a/storage/overlaybd/src/lsmt/file/types.rs b/storage/overlaybd/src/lsmt/file/types.rs index fa8a274..d65f4a5 100644 --- a/storage/overlaybd/src/lsmt/file/types.rs +++ b/storage/overlaybd/src/lsmt/file/types.rs @@ -3,9 +3,11 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::mem::size_of; use std::sync::Arc; +use std::time::Instant; use uuid::Uuid; use crate::io::virtual_file::{VirtualFile, VirtualFileWriter}; +use crate::metrics::Sha256Metrics; use storage_util::CompactWriter; pub(super) const ALIGNMENT: u64 = 512; @@ -197,8 +199,17 @@ pub struct LayerDescriptor { pub(super) struct AppendDigestTracker { hasher: Sha256, pub(super) offset: u64, + /// Where absorb costs are recorded: `rw_rebuild` while a non-empty layer + /// is rehashed on reopen, `rw_append` otherwise. + pub(super) metrics: &'static Sha256Metrics, } +/// `agentenv_sha256_*` recorder for guest-write appends into the RW layer. +pub(super) static RW_APPEND_DIGEST_METRICS: Sha256Metrics = Sha256Metrics::new("rw_append"); +/// `agentenv_sha256_*` recorder for full-file rehashes when a non-empty RW +/// layer is reopened (resume/restore path). +pub(super) static RW_REBUILD_DIGEST_METRICS: Sha256Metrics = Sha256Metrics::new("rw_rebuild"); + impl AppendDigestTracker { pub(super) async fn from_header(file: &Arc) -> Result { let mut raw = vec![0u8; HEADER_SIZE as usize]; @@ -212,6 +223,7 @@ impl AppendDigestTracker { Ok(Self { hasher, offset: HEADER_SIZE, + metrics: &RW_APPEND_DIGEST_METRICS, }) } @@ -224,7 +236,9 @@ impl AppendDigestTracker { "append digest tracker saw out-of-order write at {offset}, expected {}", self.offset ); + let start = Instant::now(); self.hasher.update(data); + self.metrics.record(data.len() as u64, start.elapsed()); self.offset = self .offset .checked_add(data.len() as u64) diff --git a/storage/overlaybd/src/metrics.rs b/storage/overlaybd/src/metrics.rs index 4a545e9..b82e6df 100644 --- a/storage/overlaybd/src/metrics.rs +++ b/storage/overlaybd/src/metrics.rs @@ -408,10 +408,114 @@ impl ZFileReadMetrics { } } +// --------------------------------------------------------------------------- +// SHA-256 hash-path metrics. Streaming digest trackers (LSMT append digests) +// record here per absorbed stretch; the agentenv crate's file-digest helpers +// feed the same metric family, so one query aggregates every SHA-256 cost in +// the process. Handles register lazily and cache in `Sha256Metrics`, so after +// the first record the hot path only pays atomic increments. +// --------------------------------------------------------------------------- + +/// SHA-256 hash cost of one data path, labeled by `op`; keep the label set +/// small. +#[derive(Debug)] +pub(crate) struct Sha256Metrics { + op: &'static str, + duration: OnceLock, + bytes: OnceLock, +} + +impl Sha256Metrics { + pub(crate) const fn new(op: &'static str) -> Self { + Self { + op, + duration: OnceLock::new(), + bytes: OnceLock::new(), + } + } + + /// Record one hashing stretch: `bytes` hashed for `elapsed` of pure + /// `Sha256::update` time, excluding reads and lock waits. + pub(crate) fn record(&self, bytes: u64, elapsed: Duration) { + self.duration + .get_or_init( + || ::metrics::histogram!("agentenv_sha256_duration_seconds", "op" => self.op), + ) + .record(elapsed.as_secs_f64()); + self.bytes + .get_or_init(|| ::metrics::counter!("agentenv_sha256_bytes_total", "op" => self.op)) + .increment(bytes); + } +} + +// --------------------------------------------------------------------------- +// ZFile write-path (compression) metrics, mirroring the read-path series +// above: one record per compressed block, duration covering only the codec +// call (CRC generation excluded, like the read path times only the decode). +// The byte counters are exact; the duration histogram is fed only by the +// 1-in-1024 block sampler in the compression module, like the read path's +// pread sampler. +// --------------------------------------------------------------------------- + +/// Metric handles for one codec's ZFile compression path. +#[derive(Debug)] +pub(crate) struct ZFileCompressMetrics { + codec: ZFileCodec, + duration: OnceLock, + input_bytes: OnceLock, + output_bytes: OnceLock, +} + +impl ZFileCompressMetrics { + pub(crate) const fn new(codec: ZFileCodec) -> Self { + Self { + codec, + duration: OnceLock::new(), + input_bytes: OnceLock::new(), + output_bytes: OnceLock::new(), + } + } + + /// Record one compressed block: `input` raw bytes encoded to `output` + /// bytes. The counters are exact; `elapsed` is `Some` only for blocks + /// picked by the sampler, so the duration histogram's exported `_count` + /// is a SAMPLE COUNT, not a block count. + pub(crate) fn record(&self, input: u64, output: u64, elapsed: Option) { + let codec = self.codec.as_label(); + if let Some(elapsed) = elapsed { + self.duration + .get_or_init(|| { + ::metrics::histogram!( + "agentenv_overlaybd_zfile_compress_duration_seconds", + "codec" => codec, + ) + }) + .record(elapsed.as_secs_f64()); + } + self.input_bytes + .get_or_init(|| { + ::metrics::counter!( + "agentenv_overlaybd_zfile_compress_input_bytes_total", + "codec" => codec, + ) + }) + .increment(input); + self.output_bytes + .get_or_init(|| { + ::metrics::counter!( + "agentenv_overlaybd_zfile_compress_output_bytes_total", + "codec" => codec, + ) + }) + .increment(output); + } +} + #[cfg(test)] mod tests { use super::{normalize_host, registry_source, RemoteReadStatus, RemoteSource}; - use super::{ZFileCodec, ZFileReadMetrics, ZFileReadStats, ZFileReadStatus}; + use super::{Sha256Metrics, ZFileCodec, ZFileCompressMetrics}; + use super::{ZFileReadMetrics, ZFileReadStats, ZFileReadStatus}; use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use std::time::Duration; @@ -576,4 +680,59 @@ mod tests { "sampled histogram series with fixed names/labels and sample counts", ); } + + /// Same thread-local [`DebuggingRecorder`] pattern as the ZFile test: + /// assert the sha256 and compression series' names, labels, and values. + #[test] + fn sha256_and_compress_metrics_use_fixed_low_cardinality_series() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + ::metrics::with_local_recorder(&recorder, || { + let rw_append = Sha256Metrics::new("rw_append"); + rw_append.record(4096, Duration::from_micros(10)); + rw_append.record(4096, Duration::from_micros(12)); + Sha256Metrics::new("rw_rebuild").record(1_048_576, Duration::from_micros(500)); + ZFileCompressMetrics::new(ZFileCodec::Lz4).record( + 4096, + 2048, + Some(Duration::from_micros(3)), + ); + // Unsampled block: counters still advance, duration histogram + // does not, mirroring the 1-in-1024 compress sampler. + ZFileCompressMetrics::new(ZFileCodec::Lz4).record(4096, 2048, None); + }); + + let mut series: Vec = Vec::new(); + for (key, _unit, _description, value) in snapshotter.snapshot().into_vec() { + let mut labels: Vec = key + .key() + .labels() + .map(|label| format!("{}={}", label.key(), label.value())) + .collect(); + labels.sort_unstable(); + let name = key.key().name().to_owned(); + let value = match value { + DebugValue::Counter(recorded) => format!("{recorded}"), + DebugValue::Histogram(values) => format!("{} samples", values.len()), + DebugValue::Gauge(_) => panic!("metric set must not use gauges: {name}"), + }; + series.push(format!("{name}{{{}}} = {value}", labels.join(","))); + } + series.sort(); + + assert_eq!( + series, + vec![ + "agentenv_overlaybd_zfile_compress_duration_seconds{codec=lz4} = 1 samples", + "agentenv_overlaybd_zfile_compress_input_bytes_total{codec=lz4} = 8192", + "agentenv_overlaybd_zfile_compress_output_bytes_total{codec=lz4} = 4096", + "agentenv_sha256_bytes_total{op=rw_append} = 8192", + "agentenv_sha256_bytes_total{op=rw_rebuild} = 1048576", + "agentenv_sha256_duration_seconds{op=rw_append} = 2 samples", + "agentenv_sha256_duration_seconds{op=rw_rebuild} = 1 samples", + ], + "exact series with fixed names/labels", + ); + } } -- 2.43.0