Rudimentary iostat in Python to demo sysfs block device stats

pyiostat.py

# pyiostat

import os
import time

from pprint import pprint

# https://www.kernel.org/doc/html/latest/block/stat.html
stat_labels_extended = [
    # reads
    "read I/Os",  # number of read I/Os processed
    "read merges",  # number of read I/Os merged with in-queue I/O
    "read sectors",  # number of sectors read
    "read ticks",  # total wait time for read requests
    # writes
    "write I/Os",
    "write merges",
    "write sectors",
    "write ticks",
    # overall
    "in_flight",  # number of I/Os currently in flight
    "io_ticks",  # total time this block device has been active
    "time_in_queue",  # total wait time for all requests
    # discards
    "discard I/Os",
    "discard merges",
    "discard sectors",
    "discard ticks",
    # flushes
    "flush I/Os",  # number of flush I/Os processed
    "flush ticks",  # total wait time for flush request
]

stat_labels_basic = [
    "read I/Os",
    "read sectors",
    "write I/Os",
    "write sectors",
]

def get_stats():
    all_stats = {}
    for dev in filter(lambda d: any(d.startswith(prefix) for prefix in ("nvme", "sd")), os.listdir("/sys/block")):
        with open(os.path.join("/sys/block", dev, "stat"), "r") as f:
            stats = [float(s) for s in f.read().strip().split()]
            labels = stat_labels_basic if len(stats) == len(stat_labels_basic) else stat_labels_extended
            stat_table = dict(zip(labels[0:len(stats)], stats))
            all_stats[dev] = stat_table
    return all_stats


def diff_stats(left, right, drop_empty=True):
    all_diff = {}
    for dev in left:
        if dev not in right:
            print(f"{dev} missing")
            continue
        all_diff[dev] = {}
        for attr in left[dev]:
            if attr not in right[dev]:
                print(f"{dev} missing {attr}")
                continue
            all_diff[dev][attr] = right[dev][attr] - left[dev][attr]
        if drop_empty and all(attr == 0 for attr in all_diff[dev].values()):
            del all_diff[dev]
    return all_diff

def compare(interval=1):
    first = get_stats()
    time.sleep(interval)
    second = get_stats()
    return diff_stats(first, second)


for _ in range(10):
    pprint(compare())