virtual_disk.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

// Copyright 2024 Oxide Computer Company

//! Types for tracking statistics about virtual disks.

use std::{
    num::NonZeroUsize,
    sync::{Arc, Mutex},
    time::Duration,
};

// NOTE: TOML definitions of timeseries are centralized in Omicron, so this file
// lives in that repo, at
// `./omicron/oximeter/oximeter/schema/virtual-machine.toml`.
oximeter::use_timeseries!("virtual-disk.toml");
use chrono::Utc;
use oximeter::{
    histogram::{Histogram, Record as _},
    types::Cumulative,
    MetricsError, Producer, Sample,
};
use propolis::block::{self, Operation};

pub use self::virtual_disk::VirtualDisk;
use self::virtual_disk::{
    BytesRead, BytesWritten, FailedFlushes, FailedReads, FailedWrites, Flushes,
    IoLatency, IoSize, Reads, Writes,
};

/// Type for tracking virtual disk stats.
///
/// This is shared between the oximeter producer registry and the actual block
/// device that implements the I/O processing. As I/Os are completed, this is
/// called back into in order to update the stats. Note that the actual shared
/// type is an Arc+Mutex around this, the [`VirtualDiskProducer`].
#[derive(Debug)]
struct VirtualDiskStats {
    /// The oximeter::Target representing this disk.
    disk: VirtualDisk,
    /// Cumulative number of reads.
    reads: Reads,
    /// Cumulative number of bytes read.
    bytes_read: BytesRead,
    /// Cumulative number of failed reads, by failure reason.
    failed_reads: [FailedReads; N_FAILURE_KINDS],
    /// Cumulative number of writes.
    writes: Writes,
    /// Cumulative number of bytes written.
    bytes_written: BytesWritten,
    /// Cumulative number of failed writes, by failure reason.
    failed_writes: [FailedWrites; N_FAILURE_KINDS],
    /// Cumulative number of flushes.
    flushes: Flushes,
    /// Cumulative number of failed flushes, by failure reason.
    failed_flushes: [FailedFlushes; N_FAILURE_KINDS],
    /// Histogram tracking I/O latency, by the I/O kind.
    io_latency: [IoLatency; N_IO_KINDS],
    /// Histogram tracking I/O sizes, by the I/O kind.
    ///
    /// Note that we have 1 fewer histogram here, since flush operations do not
    /// have a size.
    io_size: [IoSize; N_IO_KINDS - 1],
}

impl VirtualDiskStats {
    /// Update the tracked statistics with the result of an I/O completion.
    fn on_completion(&mut self, sample: BlockSample) {
        let BlockSample { op, result, duration } = sample;
        match op {
            Operation::Read(_, len) => {
                self.on_read_completion(result, len, duration)
            }
            Operation::Write(_, len) => {
                self.on_write_completion(result, len, duration)
            }
            Operation::Flush => self.on_flush_completion(result, duration),
            Operation::Discard => {
                // Discard is now wired up for local disks. We need to add support for it to the
                // schema in Omicron before we can report stats for it. For now, just ignore it.
            }
        }
    }

    fn on_read_completion(
        &mut self,
        result: block::Result,
        len: usize,
        duration: Duration,
    ) {
        let index = match result {
            block::Result::Success => {
                let _ = self.io_latency[READ_INDEX]
                    .datum
                    .sample(duration.as_nanos() as u64);
                let _ = self.io_size[READ_INDEX].datum.sample(len as u64);
                self.reads.datum += 1;
                self.bytes_read.datum += len as u64;
                return;
            }
            block::Result::Failure => FAILURE_INDEX,
            block::Result::ReadOnly => READONLY_INDEX,
            block::Result::Unsupported => UNSUPPORTED_INDEX,
        };
        self.failed_reads[index].datum.increment();
    }

    fn on_write_completion(
        &mut self,
        result: block::Result,
        len: usize,
        duration: Duration,
    ) {
        let index = match result {
            block::Result::Success => {
                let _ = self.io_latency[WRITE_INDEX]
                    .datum
                    .sample(duration.as_nanos() as u64);
                let _ = self.io_size[WRITE_INDEX].datum.sample(len as u64);
                self.writes.datum += 1;
                self.bytes_written.datum += len as u64;
                return;
            }
            block::Result::Failure => FAILURE_INDEX,
            block::Result::ReadOnly => READONLY_INDEX,
            block::Result::Unsupported => UNSUPPORTED_INDEX,
        };
        self.failed_writes[index].datum.increment();
    }

    fn on_flush_completion(
        &mut self,
        result: block::Result,
        duration: Duration,
    ) {
        let index = match result {
            block::Result::Success => {
                let _ = self.io_latency[FLUSH_INDEX]
                    .datum
                    .sample(duration.as_nanos() as u64);
                self.flushes.datum += 1;
                return;
            }
            block::Result::Failure => FAILURE_INDEX,
            block::Result::ReadOnly => READONLY_INDEX,
            block::Result::Unsupported => UNSUPPORTED_INDEX,
        };
        self.failed_flushes[index].datum.increment();
    }
}

/// Number of I/O kinds we track.
const N_IO_KINDS: usize = 3;

/// Indices into arrays tracking operations broken out by I/O kind.
const READ_INDEX: usize = 0;
const WRITE_INDEX: usize = 1;
const FLUSH_INDEX: usize = 2;

/// String representations of I/O kinds we report to Oximeter.
const READ_KIND: &str = "read";
const WRITE_KIND: &str = "write";
const FLUSH_KIND: &str = "flush";

/// Number of failure kinds we track.
const N_FAILURE_KINDS: usize = 3;

/// Indices into arrays tracking operations broken out by failure kind.
const FAILURE_INDEX: usize = 0;
const READONLY_INDEX: usize = 1;
const UNSUPPORTED_INDEX: usize = 2;

/// String representations of failure kinds we report to Oximeter.
const FAILURE_KIND: &str = "failed";
const READONLY_KIND: &str = "read-only";
const UNSUPPORTED_KIND: &str = "unsupported";

/// Latency is measured in nanoseconds. We want to track between 1 microsecond,
/// which is 10 ** 3 nanos, and 10s, which is 10 * 1e9 == 10 ** 10 nanoseconds.
const LATENCY_POWERS: (u16, u16) = (3, 10);

/// Sizes are measured in powers of 2 from 512B to 1GiB.
///
/// We use 512B as the minimum since that is the minimum supported block size.
const SIZE_POWERS: (u16, u16) = (9, 30);

/// Maximum number of samples to buffer for a given device queue before
/// consolidating them into the totals for the device.
///
/// This value was arbitrarily chosen
const MAX_BUFFERED_SAMPLES: usize = 512;

/// A [`Producer`] that emits statistics about virtual disks.
///
/// This type is shared between the block devie that handles guest I/Os, and the
/// oximeter producer server, which publishes data to oximeter when it polls us.
/// As I/Os are completed, the [`VirtualDiskProducer::on_completion()`] method
/// is called, to update stats with the results of the I/O operation.
///
/// As oximeter polls us, the producer server also collects these updated
/// statistics.

#[derive(Debug)]
pub(crate) struct BlockMetrics {
    sample_buffer: Vec<Mutex<Vec<BlockSample>>>,
    stats: Mutex<VirtualDiskStats>,
}
impl BlockMetrics {
    pub fn new(disk: VirtualDisk, max_queues: NonZeroUsize) -> Arc<Self> {
        let now = Utc::now();
        let datum = Cumulative::with_start_time(now, 0);
        let latency_histogram = Self::latency_histogram();
        let size_histogram = Self::size_histogram();
        let stats = VirtualDiskStats {
            disk,
            reads: Reads { datum },
            bytes_read: BytesRead { datum },
            failed_reads: [
                FailedReads { failure_reason: FAILURE_KIND.into(), datum },
                FailedReads { failure_reason: READONLY_KIND.into(), datum },
                FailedReads { failure_reason: UNSUPPORTED_KIND.into(), datum },
            ],
            writes: Writes { datum },
            bytes_written: BytesWritten { datum },
            failed_writes: [
                FailedWrites { failure_reason: FAILURE_KIND.into(), datum },
                FailedWrites { failure_reason: READONLY_KIND.into(), datum },
                FailedWrites { failure_reason: UNSUPPORTED_KIND.into(), datum },
            ],
            flushes: Flushes { datum },
            failed_flushes: [
                FailedFlushes { failure_reason: FAILURE_KIND.into(), datum },
                FailedFlushes { failure_reason: READONLY_KIND.into(), datum },
                FailedFlushes {
                    failure_reason: UNSUPPORTED_KIND.into(),
                    datum,
                },
            ],
            io_latency: [
                IoLatency {
                    io_kind: READ_KIND.into(),
                    datum: latency_histogram.clone(),
                },
                IoLatency {
                    io_kind: WRITE_KIND.into(),
                    datum: latency_histogram.clone(),
                },
                IoLatency {
                    io_kind: FLUSH_KIND.into(),
                    datum: latency_histogram.clone(),
                },
            ],
            io_size: [
                IoSize {
                    io_kind: READ_KIND.into(),
                    datum: size_histogram.clone(),
                },
                IoSize {
                    io_kind: WRITE_KIND.into(),
                    datum: size_histogram.clone(),
                },
            ],
        };
        let mut sample_buffer = Vec::with_capacity(max_queues.get());
        sample_buffer.resize_with(max_queues.get(), Default::default);

        Arc::new(Self { sample_buffer, stats: Mutex::new(stats) })
    }

    pub(crate) fn producer(self: &Arc<Self>) -> VirtualDiskProducer {
        VirtualDiskProducer(self.clone())
    }

    /// Construct a histogram for tracking I/O latencies.
    ///
    /// This builds a "log-linear" histogram, which has 10 bins for each power
    /// of 10, spaced between 1 microsecond and 10s inclusive.
    fn latency_histogram() -> Histogram<u64> {
        // Safety: This only fails if the bins are not valid.
        Histogram::span_decades(LATENCY_POWERS.0, LATENCY_POWERS.1).unwrap()
    }

    /// Construct a histogram for tracking I/O sizes.
    ///
    /// This creates a power-of-2 histogram for tracking I/O sizes between 512B
    /// and 1GiB.
    fn size_histogram() -> Histogram<u64> {
        let bins: Vec<_> =
            (SIZE_POWERS.0..=SIZE_POWERS.1).map(|p| 1u64 << p).collect();

        // Safety: This only fails if the bins are not valid.
        Histogram::new(&bins).unwrap()
    }

    fn consolidate_one(&self, idx: usize) {
        let mut stats = self.stats.lock().unwrap();
        let mut buf = self.sample_buffer.get(idx).unwrap().lock().unwrap();
        for sample in buf.drain(..) {
            stats.on_completion(sample);
        }
    }
    fn consolidate_all(&self) {
        let mut stats = self.stats.lock().unwrap();
        for buf in self.sample_buffer.iter() {
            let mut buf = buf.lock().unwrap();
            for sample in buf.drain(..) {
                stats.on_completion(sample);
            }
        }
    }
}

#[derive(Debug)]
struct BlockSample {
    op: block::Operation,
    result: block::Result,
    duration: Duration,
}

#[derive(Clone, Debug)]
pub struct VirtualDiskProducer(Arc<BlockMetrics>);

impl block::MetricConsumer for BlockMetrics {
    fn request_completed(
        &self,
        queue_id: block::QueueId,
        op: Operation,
        result: block::Result,
        _time_queued: Duration,
        time_processed: Duration,
    ) {
        let idx = usize::from(queue_id);
        let buf = self
            .sample_buffer
            .get(idx)
            .expect("queue ID should be within maximum");
        let mut guard = buf.lock().unwrap();
        guard.push(BlockSample { op, result, duration: time_processed });

        // Do not let an unbounded number of samples accumulate
        if guard.len() > MAX_BUFFERED_SAMPLES {
            drop(guard);
            self.consolidate_one(idx);
        }
    }
}

impl Producer for VirtualDiskProducer {
    fn produce(
        &mut self,
    ) -> Result<Box<dyn Iterator<Item = Sample>>, MetricsError> {
        // Consolidate any buffer samples first
        self.0.consolidate_all();

        // 5 scalar samples (reads, writes, flushes, bytes read / written)
        // 3 scalars broken out by failure kind
        // 2 histograms broken out by I/O kind
        const N_SAMPLES: usize = 5 + 3 * N_FAILURE_KINDS + 2 * N_IO_KINDS;
        let mut out = Vec::with_capacity(N_SAMPLES);
        let stats = self.0.stats.lock().unwrap();

        // Read statistics.
        out.push(Sample::new(&stats.disk, &stats.reads)?);
        out.push(Sample::new(&stats.disk, &stats.bytes_read)?);
        for failed in stats.failed_reads.iter() {
            out.push(Sample::new(&stats.disk, failed)?);
        }

        // Write statistics.
        out.push(Sample::new(&stats.disk, &stats.writes)?);
        out.push(Sample::new(&stats.disk, &stats.bytes_written)?);
        for failed in stats.failed_writes.iter() {
            out.push(Sample::new(&stats.disk, failed)?);
        }

        // Flushes
        out.push(Sample::new(&stats.disk, &stats.flushes)?);
        for failed in stats.failed_flushes.iter() {
            out.push(Sample::new(&stats.disk, failed)?);
        }

        // Histograms for latency and size.
        for hist in stats.io_latency.iter() {
            out.push(Sample::new(&stats.disk, hist)?);
        }
        for hist in stats.io_size.iter() {
            out.push(Sample::new(&stats.disk, hist)?);
        }
        drop(stats);
        Ok(Box::new(out.into_iter()))
    }
}

#[cfg(test)]
mod test {
    use super::BlockMetrics;
    use super::LATENCY_POWERS;
    use super::SIZE_POWERS;

    #[test]
    fn test_latency_histogram() {
        let hist = BlockMetrics::latency_histogram();
        println!("{:#?}", hist.iter().map(|bin| bin.range).collect::<Vec<_>>());
        // The math here is a bit silly, but we end up with 9 bins in each
        // "interior" power of 10, plus one more bin on the right and left for
        // the bins from [0, 1us) and [10s, inf)
        assert_eq!(
            hist.n_bins(),
            (LATENCY_POWERS.1 - LATENCY_POWERS.0) as usize * 9 + 1 + 1
        );
    }

    #[test]
    fn test_size_histogram() {
        let hist = BlockMetrics::size_histogram();
        println!("{:#?}", hist.iter().map(|bin| bin.range).collect::<Vec<_>>());
        // 1 extra left bin for [0, 512), and 1 because the range is inclusive.
        assert_eq!(
            hist.n_bins(),
            (SIZE_POWERS.1 - SIZE_POWERS.0) as usize + 1 + 1
        );
    }
}