#!/usr/lib64/linuxfabrik-monitoring-plugins/venv/bin/python
# -*- coding: utf-8; py-indent-offset: 4 -*-
#
# Author:  Linuxfabrik GmbH, Zurich, Switzerland
# Contact: info (at) linuxfabrik (dot) ch
#          https://www.linuxfabrik.ch/
# License: The Unlicense, see LICENSE file.

# https://github.com/Linuxfabrik/monitoring-plugins/blob/main/CONTRIBUTING.md

"""See the check's README for more details."""

import argparse
import os
import platform
import sys

import lib.args
import lib.base
import lib.cache
import lib.db_sqlite
import lib.disk
import lib.human
import lib.time
import lib.txt
import lib.version
from lib.globals import STATE_OK, STATE_UNKNOWN

try:
    import psutil
except ImportError:
    print('Python module "psutil" is not installed.')
    sys.exit(STATE_UNKNOWN)


__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
__version__ = '2026081001'

DESCRIPTION = """Checks disk I/O bandwidth over time and alerts on sustained saturation, not
short spikes. The check records per-disk read/write counters and then derives current (R1/W1)
and period averages (R{COUNT}/W{COUNT}). It compares the period's total bandwidth against the
maximum ever observed for that disk (RWmax). It raises a WARNING when the period average exceeds
--warning percent of RWmax. This bandwidth part only ever warns, never criticals: sustained I/O
is a signal to investigate, not an emergency you have to react to at night.

The check also reports per-disk I/O latency (await): the average time a read or write took to
complete over the period, in milliseconds. Unlike disk busy percentage, latency is robust
against device parallelism, so it is a meaningful "is my storage slow" signal on NVMe, SSD and
RAID as well. Optional --await-warning and --await-critical thresholds alert on sustained
latency; both are disabled by default. A critical latency threshold is the place to catch a
disk that is effectively hung.

Perfdata is emitted for each disk (read/write throughput per second and I/O latency, plus disk
busy percentage on Linux), so you can graph trends. On Linux the check focuses on block devices with a mounted
filesystem by default; use `--include-unmounted` to also include raw, unmounted devices such as
multipath SAN volumes. On Windows it uses psutil's disk counters. Optionally, `--top` lists the
processes that generated the most I/O traffic (read/write totals) to help identify offenders.

This check is cross-platform and works on Linux, Windows, and all psutil-supported systems.
The check stores its short trend state locally in an SQLite DB to evaluate sustained load across
runs."""


DEFAULT_CACHE_EXPIRE = 90
DEFAULT_COUNT = (
    5  # measurements; if check runs once per minute, this is a 5 minute interval
)
DEFAULT_INCLUDE_UNMOUNTED = False
DEFAULT_MATCH = ''
DEFAULT_NO_MATCH_SEVERITY = 'ok'
DEFAULT_TOP = 5
DEFAULT_WARN = 80  # %


def parse_args():
    """Parse command line arguments using argparse."""
    parser = argparse.ArgumentParser(
        description=DESCRIPTION,
        epilog=lib.args.epilog(__file__),
        formatter_class=lib.args.HelpFormatter,
    )

    parser.add_argument(
        '-V',
        '--version',
        action='version',
        version=f'%(prog)s: v{__version__} by {__author__}',
    )
    parser.add_argument(
        '--always-ok',
        help=lib.args.help('--always-ok'),
        dest='ALWAYS_OK',
        action='store_true',
        default=False,
    )
    parser.add_argument(
        '--await-critical',
        help='CRIT threshold for per-disk I/O latency (await) in milliseconds, '
        'averaged over the last `--count` runs. '
        'await is the average time a read or write took to complete. '
        'Use this to catch a disk that is effectively hung (sustained latency of '
        'seconds), not a merely busy one. '
        'Supports Nagios ranges. '
        'Disabled by default.',
        dest='AWAIT_CRIT',
        default=None,
    )
    parser.add_argument(
        '--await-warning',
        help='WARN threshold for per-disk I/O latency (await) in milliseconds, '
        'averaged over the last `--count` runs. '
        'await is the average time a read or write took to complete. '
        'A good value depends on the storage (an SSD is well below a millisecond, '
        'a busy HDD can sustain tens of milliseconds), so set it to what is '
        'abnormal for your disks. '
        'Supports Nagios ranges. '
        'Disabled by default.',
        dest='AWAIT_WARN',
        default=None,
    )
    parser.add_argument(
        '--count',
        help=lib.args.help('--count') + ' Default: %(default)s',
        dest='COUNT',
        type=int,
        default=DEFAULT_COUNT,
    )
    # Deprecated: this check is a saturation/trending check and only ever warns.
    # A busy disk is a signal to investigate, not to react to at night, so there
    # is no CRIT tier. Kept for backward compatibility, silently ignored.
    parser.add_argument(
        '--critical',
        help=argparse.SUPPRESS,
        dest='CRIT',
    )
    parser.add_argument(
        '--include-unmounted',
        help='Also monitor block devices that have no mounted filesystem (Linux only). '
        'By default only block devices with a mounted filesystem are monitored. '
        'Enable this to include raw, unmounted devices such as multipath SAN volumes or '
        'disks used directly by a database or storage layer. '
        'Combine with `--match`, otherwise every unmounted device shows up. '
        'Pseudo devices (loop, ram, zram, floppy, optical) are always excluded. '
        'Default: %(default)s',
        dest='INCLUDE_UNMOUNTED',
        action='store_true',
        default=DEFAULT_INCLUDE_UNMOUNTED,
    )
    # Deprecated and removed: the check no longer measures iowait at all. Linux
    # iowait is relabelled idle time and, per the kernel, "broken/meaningless" on
    # SMP, so it is neither a saturation measure nor a reliable signal. Both
    # thresholds are kept for backward compatibility and silently ignored.
    parser.add_argument(
        '--iowait-critical',
        help=argparse.SUPPRESS,
        dest='IOWAIT_CRIT',
    )
    parser.add_argument(
        '--iowait-warning',
        help=argparse.SUPPRESS,
        dest='IOWAIT_WARN',
    )
    parser.add_argument(
        '--match',
        # Written inline rather than taken from lib.args: this check takes a single
        # expression and has no --ignore, so neither the "can be specified multiple
        # times" nor the --ignore precedence sentence of the shared text applies here.
        help='Only check disks whose path or mountpoint matches this Python regular '
        'expression. '
        'Case-sensitive by default; use `(?i)` for case-insensitive matching. '
        'The regex is anchored at the start of the string (Python `re.match`) and matched '
        'against the full device path (e.g. `/dev/sda`), the device-mapper path '
        '(e.g. `/dev/mapper/vg-lv`) and the mountpoint, so prefix with `.*` to match anywhere '
        '(`.*sda$` instead of `^sda$`). '
        'On Linux only block devices with a mounted filesystem are considered by default; add '
        '`--include-unmounted` to also match raw, unmounted devices (multipath SAN volumes, raw '
        'LUNs). '
        'Default: %(default)s',
        dest='MATCH',
        default=DEFAULT_MATCH,
    )
    parser.add_argument(
        '--no-match-severity',
        help=lib.args.help('--no-match-severity') + ' Default: %(default)s',
        dest='NO_MATCH_SEVERITY',
        choices=['ok', 'warn', 'crit', 'unknown'],
        default=DEFAULT_NO_MATCH_SEVERITY,
    )
    parser.add_argument(
        '--no-perfdata',
        help=lib.args.help('--no-perfdata'),
        dest='NO_PERFDATA',
        action='store_true',
        default=False,
    )
    parser.add_argument(
        '--top',
        help='Number of top processes to list by I/O traffic. '
        'Use `--top=0` to disable. '
        'Default: %(default)s',
        dest='TOP',
        type=int,
        default=DEFAULT_TOP,
    )
    parser.add_argument(
        '--warning',
        help='WARN threshold for disk bandwidth saturation as a percentage of the observed maximum, '
        'measured over the last `--count` runs. '
        'Default: >= %(default)s',
        dest='WARN',
        type=int,
        default=DEFAULT_WARN,
    )

    args, _ = parser.parse_known_args()
    return args


def get_max_bandwidth(disk, current_bandwidth):
    """Store the maximum measured bandwidth for the secific disk in cache table."""
    historic_bandwidth = lib.cache.get(
        f'disk-io-{disk}-bandwidth-max',
        filename='linuxfabrik-monitoring-plugins-disk-io.db',
    )
    # Disk should be capable of at least 10 MB/sec (if no info is provided)
    max_bandwidth = max(
        int(historic_bandwidth),
        int(current_bandwidth),
        10 * 1024 * 1024,
    )
    lib.cache.set(
        f'disk-io-{disk}-bandwidth-max',
        max_bandwidth,
        filename='linuxfabrik-monitoring-plugins-disk-io.db',
    )
    return max_bandwidth


def get_rate(ts1, ts2, r1, r2, w1, w2):
    """Given two read-, write- and timestamp-values, return the read- and write-rate plus bandwidth."""
    timediff = abs(ts1 - ts2)  # in seconds
    if timediff == 0:
        return 0, 0, 0, 0
    r = abs(int(float(r1 - r2) / timediff))
    w = abs(int(float(w1 - w2) / timediff))
    return timediff, r, w, r + w


def get_await(time1, time2, count1, count2):
    """Average latency of the completed I/Os between two snapshots, in milliseconds.

    This is iostat's await: the time counter (ms an I/O spent between issue and
    completion, a cumulative counter) divided by the number of I/Os completed in
    the interval. Unlike disk-busy percent (%util), latency is robust against
    device parallelism: it is real time per I/O on any device, whereas %util
    only tells whether at least one I/O was in flight. Returns 0.0 when no I/O
    completed in the interval (no latency to report).
    """
    count_delta = abs(count1 - count2)
    if count_delta <= 0:
        return 0.0
    return round(abs(time1 - time2) / count_delta, 1)


def top(count):
    """Get top X processes that generated the most I/O traffic."""
    # Fast path: nothing to print, so nothing to scan
    if count <= 0:
        return ''

    totals = {}  # name -> {'r': bytes, 'w': bytes}
    msg = ''

    # Prefer attrs path (psutil >= 5.3.0): fewer syscalls, fewer exceptions
    if lib.version.version(psutil.__version__) >= lib.version.version('5.3.0'):
        try:
            for p in psutil.process_iter(attrs=['name', 'io_counters'], ad_value=None):
                try:
                    info = p.info
                    name = info.get('name') or ''
                    ioc = info.get('io_counters')
                    if not ioc:
                        continue
                    entry = totals.setdefault(name, {'r': 0, 'w': 0})
                    # accumulate read/write bytes; guard against None
                    entry['r'] += getattr(ioc, 'read_bytes', 0) or 0
                    entry['w'] += getattr(ioc, 'write_bytes', 0) or 0
                except (
                    psutil.NoSuchProcess,
                    psutil.AccessDenied,
                    psutil.ZombieProcess,
                ):
                    # process vanished or denied: skip and continue
                    continue
        except Exception:
            # Defensive: if attrs/ad_value path misbehaves anywhere, fall back below.
            pass

    # Legacy / fallback path
    if not totals:
        try:
            for proc in psutil.process_iter():
                try:
                    info = proc.as_dict(attrs=['name', 'io_counters'])
                except (
                    psutil.NoSuchProcess,
                    psutil.AccessDenied,
                    psutil.ZombieProcess,
                ):
                    continue
                name = info.get('name') or ''
                ioc = info.get('io_counters')
                if not ioc:
                    continue
                entry = totals.setdefault(name, {'r': 0, 'w': 0})
                entry['r'] += getattr(ioc, 'read_bytes', 0) or 0
                entry['w'] += getattr(ioc, 'write_bytes', 0) or 0
        except psutil.NoSuchProcess:
            pass

    if not totals:
        return msg

    # Sort by total bytes (read+write) desc and show the top N
    ranked = sorted(
        totals.items(), key=lambda kv: kv[1]['r'] + kv[1]['w'], reverse=True
    )[:count]

    # If everything is truly zero, keep output empty
    if ranked and (ranked[0][1]['r'] + ranked[0][1]['w'] > 0):
        lines = [f'\nTop {count} processes that generate the most I/O traffic (r/w):']
        for i, (name, io) in enumerate(ranked, start=1):
            lines.append(
                f'{i}. {name}: '
                f'{lib.human.bytes2human(io["r"])}/'
                f'{lib.human.bytes2human(io["w"])}'
            )
        msg = '\n'.join(lines) + '\n'
    return msg


def main():
    """The main function. This is where the magic happens."""

    # parse the command line
    try:
        args = parse_args()
    except SystemExit:
        sys.exit(STATE_UNKNOWN)

    # On Windows we can work with what psutil returns, but on Linux psutil returns too much noise
    # from devices of all kinds. There we use a different approach, but therefore we have
    # to handle both platforms separately. :-(
    # Kernel 5.5 added 2 more fields to /proc/diskstats, requiring another
    # change after the one for 4.18, which recently added 4 fields.
    # To prevent "ValueError: not sure how to interpret line",
    # we check the version of psutil first.
    if lib.base.LINUX and all(
        [
            lib.version.version(platform.release()) >= lib.version.version('4.18.0'),
            lib.version.version(psutil.__version__) < lib.version.version('5.7.0'),
        ]
    ):
        lib.base.oao(
            'Nothing checked. '
            'Running Kernel >= 4.18, this check needs the Python module '
            f'psutil v5.7.0+ (installed {psutil.__version__}; have a look at '
            'https://github.com/giampaolo/psutil/pull/1665 '
            'for details).',
            STATE_OK,
            always_ok=args.ALWAYS_OK,
        )

    # bd: block device; dmd: device mapper device, mp: mountpoint

    # create the perfdata table
    conn = lib.base.coe(
        lib.db_sqlite.connect(filename='linuxfabrik-monitoring-plugins-disk-io.db')
    )

    # Best-effort: reduce IO stalls and file locking on Windows without changing outputs
    try:
        conn.execute('PRAGMA journal_mode=WAL')
        conn.execute('PRAGMA synchronous=NORMAL')
    except Exception:
        pass

    # same structure for Linux and Windows, makes life easier.
    # read_count/write_count are the number of completed I/Os and feed the await
    # (latency) calculation together with read_time/write_time. On an existing
    # cache DB from an older release these columns are missing; lib.db_sqlite
    # detects the schema mismatch on insert and rebuilds the DB, so the check
    # self-heals after a single "Waiting for more data." run.
    definition = """
        bd TEXT NOT NULL,
        dmd TEXT,
        mp TEXT,
        busy_time INT DEFAULT 0,
        read_bytes INT DEFAULT 0,
        read_count INT DEFAULT 0,
        read_merged_count INT DEFAULT 0,
        read_time INT DEFAULT 0,
        write_bytes INT DEFAULT 0,
        write_count INT DEFAULT 0,
        write_merged_count INT DEFAULT 0,
        write_time INT DEFAULT 0,
        timestamp INT DEFAULT 0
    """
    lib.base.coe(lib.db_sqlite.create_table(conn, definition, drop_table_first=False))
    lib.base.coe(lib.db_sqlite.create_index(conn, 'bd'))

    # init some vars
    msg = f'No I/O on `{args.MATCH}`.' if args.MATCH else 'No I/O.'
    perfdata = ''
    state = STATE_OK
    table_values = []
    compiled_regex = lib.base.coe(lib.txt.compile_regex(args.MATCH))
    now = lib.time.now()
    busiest_disk = 0  # disk with the highest sum of r/w: show this on top later on
    alert_state = STATE_OK  # most severe per-disk finding, for the first line
    alert_msg = ''
    disks = []

    # fetch data
    try:
        disk_io_counters = psutil.disk_io_counters(perdisk=True)
    except ValueError:
        lib.base.cu('psutil raised an error')

    # analyze and enrich data, store it to database
    if lib.base.WINDOWS:
        for disk, values in disk_io_counters.items():
            # filter devices that do not match
            if args.MATCH and not lib.base.coe(
                lib.txt.match_regex(compiled_regex, disk)
            ):
                continue

            data = {}
            data['bd'] = disk
            data['dmd'] = ''
            data['mp'] = ''
            data['busy_time'] = getattr(values, 'busy_time', 0)
            data['read_bytes'] = getattr(values, 'read_bytes', 0)
            data['read_count'] = getattr(values, 'read_count', 0)
            data['read_merged_count'] = getattr(values, 'read_merged_count', 0)
            data['read_time'] = getattr(values, 'read_time', 0)
            data['write_bytes'] = getattr(values, 'write_bytes', 0)
            data['write_count'] = getattr(values, 'write_count', 0)
            data['write_merged_count'] = getattr(values, 'write_merged_count', 0)
            data['write_time'] = getattr(values, 'write_time', 0)
            data['timestamp'] = now
            disks.append({'bd': disk, 'dmd': '', 'mp': ''})

            # store it to database
            lib.base.coe(lib.db_sqlite.insert(conn, data))
    else:
        # by default only mounted filesystems; opt in to raw/unmounted block devices
        if args.INCLUDE_UNMOUNTED:
            real_disks = lib.disk.get_block_devices()
        else:
            real_disks = lib.disk.get_real_disks()
        for disk in real_disks:
            # filter devices that do not match
            if args.MATCH and not any(
                (
                    lib.base.coe(lib.txt.match_regex(compiled_regex, disk['bd'])),
                    lib.base.coe(lib.txt.match_regex(compiled_regex, disk['dmd'])),
                    lib.base.coe(lib.txt.match_regex(compiled_regex, disk['mp'])),
                )
            ):
                continue

            psutil_name = os.path.basename(disk['bd'])
            if psutil_name not in disk_io_counters:
                continue

            data = {}
            data['bd'] = disk['bd']
            data['dmd'] = disk['dmd']
            data['mp'] = disk['mp']
            data['busy_time'] = getattr(disk_io_counters[psutil_name], 'busy_time', 0)
            data['read_bytes'] = getattr(disk_io_counters[psutil_name], 'read_bytes', 0)
            data['read_count'] = getattr(disk_io_counters[psutil_name], 'read_count', 0)
            data['read_merged_count'] = getattr(
                disk_io_counters[psutil_name],
                'read_merged_count',
                0,
            )
            data['read_time'] = getattr(disk_io_counters[psutil_name], 'read_time', 0)
            data['write_bytes'] = getattr(
                disk_io_counters[psutil_name], 'write_bytes', 0
            )
            data['write_count'] = getattr(
                disk_io_counters[psutil_name], 'write_count', 0
            )
            data['write_merged_count'] = getattr(
                disk_io_counters[psutil_name],
                'write_merged_count',
                0,
            )
            data['write_time'] = getattr(disk_io_counters[psutil_name], 'write_time', 0)
            data['timestamp'] = now
            disks.append(disk)

            # store it to database
            lib.base.coe(lib.db_sqlite.insert(conn, data))

    if not disks:
        lib.db_sqlite.close(conn)
        lib.base.oao(
            'No disks matched.' if args.MATCH else 'No disks found.',
            lib.base.str2state(args.NO_MATCH_SEVERITY),
            '',
            always_ok=args.ALWAYS_OK,
            no_perfdata=args.NO_PERFDATA,
        )

    # truncate old data (just keep args.COUNT for each disk) and commit
    lib.base.coe(lib.db_sqlite.cut(conn, _max=args.COUNT * len(disks)))
    lib.base.coe(lib.db_sqlite.commit(conn))

    # from here on just working on the database
    # warn about a "count" period/amount of time, not about the current situation above
    # (what might be a peak only)
    for disk in disks:
        # get all historical data rows for a specific disk, newest item first
        data = lib.base.coe(
            lib.db_sqlite.select(
                conn,
                """
            SELECT *
            FROM perfdata
            WHERE bd = :name
            ORDER BY timestamp DESC
            """,
                {'name': disk['bd']},
            )
        )

        if len(data) < 2:
            lib.db_sqlite.close(conn)
            lib.base.oao('Waiting for more data.', state)

        # calculate current rates (like "load1")
        timediff1, read_bytes_per_second1, write_bytes_per_second1, bandwidth1 = (
            get_rate(
                data[0]['timestamp'],
                data[1]['timestamp'],
                data[0]['read_bytes'],
                data[1]['read_bytes'],
                data[0]['write_bytes'],
                data[1]['write_bytes'],
            )
        )
        if timediff1 <= 0:  # often happens after a reboot
            lib.db_sqlite.close(conn)
            lib.base.oao('Waiting for more data.', state)

        # get the maximum disk bandwidth in disks' history
        bandwidth_max = get_max_bandwidth(disk['bd'], bandwidth1)

        if bandwidth1 > busiest_disk:
            # get the current busiest disk for the first line of the message
            msg = (
                f'{disk["bd"]}: '
                f'{lib.human.bytes2human(read_bytes_per_second1)}/s read1, '
                f'{lib.human.bytes2human(write_bytes_per_second1)}/s write1, '
                f'{lib.human.bytes2human(bandwidth1)}/s total, '
                f'{lib.human.bytes2human(bandwidth_max)}/s max'
            )
            if args.MATCH:
                msg += f' (disks matching `{args.MATCH}`).'
            busiest_disk = bandwidth1

        # calculate read/write rate over the entire period (like "load15")
        if len(data) != args.COUNT:  # not enough data yet
            continue

        timediff15, read_bytes_per_second15, write_bytes_per_second15, bandwidth15 = (
            get_rate(
                data[0]['timestamp'],
                data[args.COUNT - 1]['timestamp'],
                data[0]['read_bytes'],
                data[args.COUNT - 1]['read_bytes'],
                data[0]['write_bytes'],
                data[args.COUNT - 1]['write_bytes'],
            )
        )
        if timediff15 <= 0:  # often happens after a reboot
            lib.db_sqlite.close(conn)
            lib.base.oao('Waiting for more data.', state)

        # bandwidth state, WARN-only (crit=None): sustained bandwidth near the
        # disk's own observed maximum is a saturation/trending signal to
        # investigate, not a wake-up-at-night event.
        local_state = lib.base.get_state(
            bandwidth15,
            bandwidth_max * args.WARN / 100,
            None,
        )
        state = lib.base.get_worst(local_state, state)

        # I/O latency (await) over the period, averaged across reads and writes.
        # Unlike bandwidth this CAN go critical: --await-critical is meant for a
        # disk that is effectively hung (sustained multi-second latency). Both
        # await thresholds are off by default, so with no thresholds set
        # get_state() returns OK and await is reported for trending only.
        await_ms = get_await(
            data[0]['read_time'] + data[0]['write_time'],
            data[args.COUNT - 1]['read_time'] + data[args.COUNT - 1]['write_time'],
            data[0]['read_count'] + data[0]['write_count'],
            data[args.COUNT - 1]['read_count'] + data[args.COUNT - 1]['write_count'],
        )
        read_await_ms = get_await(
            data[0]['read_time'],
            data[args.COUNT - 1]['read_time'],
            data[0]['read_count'],
            data[args.COUNT - 1]['read_count'],
        )
        write_await_ms = get_await(
            data[0]['write_time'],
            data[args.COUNT - 1]['write_time'],
            data[0]['write_count'],
            data[args.COUNT - 1]['write_count'],
        )
        await_state = lib.base.get_state(
            await_ms, args.AWAIT_WARN, args.AWAIT_CRIT, _operator='range'
        )
        state = lib.base.get_worst(await_state, state)

        # Remember the most severe disk for the first line: notifications only
        # show the first line, and a hung disk (await CRIT) is often not the
        # busiest one, so it must not get buried under a benign busy disk.
        disk_state = lib.base.get_worst(local_state, await_state)
        if (
            lib.base.get_worst(disk_state, alert_state) == disk_state
            and disk_state != alert_state
        ):
            alert_state = disk_state
            reasons = []
            if local_state != STATE_OK:
                reasons.append(
                    f'{lib.human.bytes2human(bandwidth15)}/s of '
                    f'{lib.human.bytes2human(bandwidth_max)}/s max'
                )
            if await_state != STATE_OK:
                reasons.append(f'{await_ms:.1f}ms latency')
            alert_msg = (
                f'{disk["bd"]}: '
                + ', '.join(reasons)
                + lib.base.state2str(disk_state, prefix=' ')
            )
            if args.MATCH:
                alert_msg += f' (disks matching `{args.MATCH}`)'
            alert_msg += '.'

        bd = disk['bd'].replace('/dev/', '')
        table_values.append(
            {
                'bd': bd,
                'dmd': disk['dmd'].replace('/dev/mapper/', ''),
                'mp': disk['mp'],
                'max': lib.human.bytes2human(bandwidth_max),
                'r1': lib.human.bytes2human(read_bytes_per_second1),
                'w1': lib.human.bytes2human(write_bytes_per_second1),
                'r15': lib.human.bytes2human(read_bytes_per_second15),
                'w15': lib.human.bytes2human(write_bytes_per_second15),
                't15': lib.human.bytes2human(bandwidth15)
                + lib.base.state2str(local_state, prefix=' '),
                'await': f'{await_ms:.1f}ms'
                + lib.base.state2str(await_state, prefix=' '),
            }
        )

        # perfdata: emit per-second rates as gauges, never the raw cumulative
        # counters. "throughput" is read + write per second; its _15 average
        # carries the per-disk alerting threshold (a percentage of the observed
        # maximum bandwidth).
        perfdata += lib.base.get_perfdata(
            f'{bd}_read_bytes_per_second1',
            read_bytes_per_second1,
            uom='B',
            _min=0,
        )
        perfdata += lib.base.get_perfdata(
            f'{bd}_read_bytes_per_second15',
            read_bytes_per_second15,
            uom='B',
            _min=0,
        )
        perfdata += lib.base.get_perfdata(
            f'{bd}_write_bytes_per_second1',
            write_bytes_per_second1,
            uom='B',
            _min=0,
        )
        perfdata += lib.base.get_perfdata(
            f'{bd}_write_bytes_per_second15',
            write_bytes_per_second15,
            uom='B',
            _min=0,
        )
        perfdata += lib.base.get_perfdata(
            f'{bd}_throughput1',
            bandwidth1,
            uom='B',
            _min=0,
        )
        perfdata += lib.base.get_perfdata(
            f'{bd}_throughput15',
            bandwidth15,
            uom='B',
            warn=int(bandwidth_max * args.WARN / 100),
            _min=0,
        )
        # I/O latency (await) in ms, cross-platform: read_time/write_time are
        # exposed on Windows too, unlike busy_time. warn/crit carry the optional
        # latency thresholds and are simply omitted when unset (None).
        perfdata += lib.base.get_perfdata(
            f'{bd}_await',
            await_ms,
            uom='ms',
            warn=args.AWAIT_WARN,
            crit=args.AWAIT_CRIT,
            _min=0,
        )
        perfdata += lib.base.get_perfdata(
            f'{bd}_read_await',
            read_await_ms,
            uom='ms',
            _min=0,
        )
        perfdata += lib.base.get_perfdata(
            f'{bd}_write_await',
            write_await_ms,
            uom='ms',
            _min=0,
        )
        # disk utilization (iostat's %util), Linux only: psutil exposes busy_time
        # (the underlying /proc/diskstats field 13 "io_ticks", in ms) only on
        # Linux and FreeBSD, so on Windows it is a constant 0 and must not be
        # emitted. busy_time is a cumulative counter, so derive the value from the
        # delta rather than emitting the raw counter (see CONTRIBUTING: no
        # continuous counters in perfdata).
        # Grounding (busy_time and %util are the same underlying counter):
        # - psutil: busy_time is /proc/diskstats field 13 "io_ticks" in ms,
        #   Linux/FreeBSD only (psutil sdiskio in _ntuples.py).
        # - sysstat: iostat %util = delta(io_ticks)/interval*100, identical
        #   formula (compute_ext_disk_stats in rd_stats.c, field "tot_ticks").
        # - kernel: io_ticks counts wall-clock time with >=1 request in flight,
        #   NOT queue depth (update_io_ticks in block/blk-core.c), so %util is
        #   only an idle detector, not a saturation measure on parallel devices
        #   (NVMe, SSD, dm/md, ZFS). See Documentation/admin-guide/iostats.rst.
        if lib.base.LINUX:
            busy_delta = data[0]['busy_time'] - data[1]['busy_time']
            busy_percent = round(
                max(0.0, min(100.0, busy_delta / (timediff1 * 1000) * 100)),
                1,
            )
            perfdata += lib.base.get_perfdata(
                f'{bd}_busy_percent',
                busy_percent,
                uom='%',
                _min=0,
                _max=100,
            )

    lib.db_sqlite.close(conn)

    # first line: lead with the most severe disk when something is in WARN/CRIT
    # (a hung disk on await is often not the busiest one, so it must not be
    # buried); otherwise the busiest disk set in the loop above stays.
    if alert_state != STATE_OK and alert_msg:
        msg = alert_msg

    # build the message
    msg = msg + '\n\n'
    if table_values:
        msg += lib.base.get_table(
            table_values,
            [
                'bd',
                'mp',
                'dmd',
                'max',
                'r1',
                'w1',
                'r15',
                'w15',
                't15',
                'await',
            ],
            header=[
                'Name',
                'MntPnts',
                'DvMppr',
                'RWmax/s',
                'R1/s',
                'W1/s',
                f'R{args.COUNT}/s',
                f'W{args.COUNT}/s',
                f'RW{args.COUNT}/s',
                'Await',
            ],
        )

    # Top X processes that generated the most I/O traffic
    msg += top(args.TOP)

    # over and out
    lib.base.oao(
        msg.replace('\n\n\n', '\n\n'),
        state,
        perfdata,
        always_ok=args.ALWAYS_OK,
        no_perfdata=args.NO_PERFDATA,
    )


if __name__ == '__main__':
    try:
        main()
    except Exception:
        lib.base.cu()
