grafos_profile/

summary.rs

//! Aggregated statistics computed from a profile recording.

use alloc::collections::BTreeMap;
use alloc::string::String;
use alloc::vec::Vec;

use grafos_observe::span::ResourceSpan;

use crate::recording::ProfileRecording;

/// Per-resource-type aggregated statistics.
#[derive(Debug, Clone, Default)]
pub struct ResourceTypeSummary {
    /// Total bytes leased (approximated from lease_cost / duration).
    pub total_lease_cost_byte_secs: u64,
    /// Total bytes read across all spans of this type.
    pub total_bytes_read: u64,
    /// Total bytes written across all spans of this type.
    pub total_bytes_written: u64,
    /// Total operation count.
    pub total_ops: u64,
    /// Total lease acquire wait (microseconds).
    pub total_acquire_wait_us: u64,
}

/// Aggregated profile statistics.
#[derive(Debug, Clone)]
pub struct ProfileSummary {
    /// Total recording duration in microseconds.
    pub total_duration_us: u64,
    /// Number of spans.
    pub span_count: usize,
    /// Number of unique lease IDs.
    pub unique_lease_count: usize,
    /// Total lease cost across all spans (byte-seconds).
    pub total_lease_cost_byte_secs: u64,
    /// Per-resource-type breakdown.
    pub by_resource_type: BTreeMap<String, ResourceTypeSummary>,
    /// Top N spans by lease cost (name, cost).
    pub top_by_cost: Vec<(String, u64)>,
    /// Top N spans by bytes (name, bytes_read + bytes_written).
    pub top_by_bytes: Vec<(String, u64)>,
    /// Top N spans by duration (name, duration_us).
    pub top_by_duration: Vec<(String, u64)>,
    /// Top N spans by acquire wait (name, wait_us).
    pub top_by_acquire_wait: Vec<(String, u64)>,
}

impl ProfileSummary {
    /// Compute summary from a recording with default top-N = 10.
    pub fn from_recording(rec: &ProfileRecording) -> Self {
        Self::from_recording_top_n(rec, 10)
    }

    /// Compute summary from a recording with specified top-N.
    pub fn from_recording_top_n(rec: &ProfileRecording, top_n: usize) -> Self {
        let total_duration_us = rec.header.duration_us.unwrap_or_else(|| {
            if rec.spans.is_empty() {
                0
            } else {
                let min_start = rec
                    .spans
                    .iter()
                    .map(|s| s.start_time_unix_us)
                    .min()
                    .unwrap_or(0);
                let max_end = rec
                    .spans
                    .iter()
                    .map(|s| s.end_time_unix_us)
                    .max()
                    .unwrap_or(0);
                max_end.saturating_sub(min_start)
            }
        });

        // Unique lease IDs
        let mut lease_ids: Vec<u128> = Vec::new();
        for span in &rec.spans {
            for &lid in &span.lease_ids {
                if !lease_ids.contains(&lid) {
                    lease_ids.push(lid);
                }
            }
        }

        // Per-resource-type aggregation
        let mut by_type: BTreeMap<String, ResourceTypeSummary> = BTreeMap::new();
        for span in &rec.spans {
            for &(ref op_key, count) in &span.ops {
                let key = alloc::format!("{}", op_key.resource_type);
                let entry = by_type.entry(key).or_default();
                entry.total_ops += count;
            }
        }

        // Attribute bytes and cost to resource types based on ops
        for span in &rec.spans {
            if span.ops.is_empty() {
                continue;
            }
            // Distribute cost proportionally by op count
            let total_ops: u64 = span.ops.iter().map(|(_, c)| c).sum();
            if total_ops == 0 {
                continue;
            }
            for &(ref op_key, count) in &span.ops {
                let key = alloc::format!("{}", op_key.resource_type);
                let entry = by_type.entry(key).or_default();
                let fraction_num = count;
                let fraction_denom = total_ops;
                entry.total_lease_cost_byte_secs +=
                    span.lease_cost_byte_secs * fraction_num / fraction_denom;
                entry.total_bytes_read += span.bytes_read * fraction_num / fraction_denom;
                entry.total_bytes_written += span.bytes_written * fraction_num / fraction_denom;
                entry.total_acquire_wait_us +=
                    span.lease_acquire_wait_us * fraction_num / fraction_denom;
            }
        }

        let total_lease_cost_byte_secs: u64 =
            rec.spans.iter().map(|s| s.lease_cost_byte_secs).sum();

        // Top-N lists
        let top_by_cost = top_n_by(&rec.spans, top_n, |s| s.lease_cost_byte_secs);
        let top_by_bytes = top_n_by(&rec.spans, top_n, |s| s.bytes_read + s.bytes_written);
        let top_by_duration = top_n_by(&rec.spans, top_n, |s| s.duration_us());
        let top_by_acquire_wait = top_n_by(&rec.spans, top_n, |s| s.lease_acquire_wait_us);

        ProfileSummary {
            total_duration_us,
            span_count: rec.spans.len(),
            unique_lease_count: lease_ids.len(),
            total_lease_cost_byte_secs,
            by_resource_type: by_type,
            top_by_cost,
            top_by_bytes,
            top_by_duration,
            top_by_acquire_wait,
        }
    }
}

fn top_n_by(
    spans: &[ResourceSpan],
    n: usize,
    key_fn: fn(&ResourceSpan) -> u64,
) -> Vec<(String, u64)> {
    let mut items: Vec<(String, u64)> = spans.iter().map(|s| (s.name.clone(), key_fn(s))).collect();
    items.sort_by(|a, b| b.1.cmp(&a.1));
    items.truncate(n);
    items
}

#[cfg(test)]
mod tests {
    use super::*;
    use grafos_observe::event::{OpType, ResourceType};
    use grafos_observe::trace::TraceContext;

    fn test_ctx() -> TraceContext {
        let mut bytes = [0u8; 24];
        for (i, b) in bytes.iter_mut().enumerate() {
            *b = (i as u8).wrapping_add(0x42);
        }
        TraceContext::new_root(&bytes)
    }

    fn make_span(name: &str, start: u64, end: u64, cost: u64) -> ResourceSpan {
        let mut child_bytes = [0xAA_u8; 8];
        let src = name.as_bytes();
        let len = src.len().min(8);
        child_bytes[..len].copy_from_slice(&src[..len]);
        let ctx = test_ctx().child(&child_bytes);
        let mut span = ResourceSpan::new(name, ctx);
        span.start_time_unix_us = start;
        span.end_time_unix_us = end;
        span.lease_cost_byte_secs = cost;
        span
    }

    #[test]
    fn summary_basic() {
        let mut s1 = make_span("read_op", 1000, 2000, 100);
        s1.bytes_read = 4096;
        s1.record_op(ResourceType::Mem, OpType::Read, 10);
        s1.add_lease_id(1);

        let mut s2 = make_span("write_op", 1500, 3000, 200);
        s2.bytes_written = 8192;
        s2.record_op(ResourceType::Block, OpType::WriteBlock, 5);
        s2.add_lease_id(2);

        let rec = ProfileRecording::from_spans(vec![s1, s2]);
        let summary = ProfileSummary::from_recording(&rec);

        assert_eq!(summary.span_count, 2);
        assert_eq!(summary.unique_lease_count, 2);
        assert_eq!(summary.total_lease_cost_byte_secs, 300);
        assert_eq!(summary.total_duration_us, 2000); // 3000 - 1000

        assert!(summary.by_resource_type.contains_key("mem"));
        assert!(summary.by_resource_type.contains_key("block"));
        assert_eq!(summary.by_resource_type["mem"].total_ops, 10);
        assert_eq!(summary.by_resource_type["block"].total_ops, 5);
    }

    #[test]
    fn top_n_ordering() {
        let s1 = make_span("cheap", 0, 100, 10);
        let s2 = make_span("expensive", 0, 100, 1000);
        let s3 = make_span("medium", 0, 100, 500);

        let rec = ProfileRecording::from_spans(vec![s1, s2, s3]);
        let summary = ProfileSummary::from_recording_top_n(&rec, 2);

        assert_eq!(summary.top_by_cost.len(), 2);
        assert_eq!(summary.top_by_cost[0].0, "expensive");
        assert_eq!(summary.top_by_cost[1].0, "medium");
    }
}