iotop 能看出是哪个进程在烧磁盘

(
  capture=$(mktemp -d) || exit 1
  printf 'Evidence directory: %s\n' "$capture"
  iotop --version || exit $?
  iostat -V || exit $?
  LC_ALL=C iostat -x -y -t 2 5 > "$capture/iostat.txt" 2> "$capture/iostat.err" &
  stat_pid=$!
  if LC_ALL=C XDG_CONFIG_HOME="$capture" iotop -b -P -k -t -d 2 -n 5 \
      --no-only --no-accumulated --no-accum-bw \
      > "$capture/iotop.txt" 2> "$capture/iotop.err"; then
    task_status=0
  else
    task_status=$?
  fi
  if wait "$stat_pid"; then stat_status=0; else stat_status=$?; fi
  printf 'iotop_status=%s iostat_status=%s\n' "$task_status" "$stat_status"
  cat "$capture/iotop.err" "$capture/iostat.err"
  [ "$task_status" -eq 0 ] && [ "$stat_status" -eq 0 ] || exit 1
  cat "$capture/iotop.txt" "$capture/iostat.txt"
)

"""Read-only Linux identity snapshot; requires Python 3.9+ and pidfd support."""
import json
import os
import re
import select
import sys


def snapshot(token, proc_root="/proc", ops=os, poll_factory=select.poll):
    if not re.fullmatch(r"[1-9][0-9]*", token):
        raise ValueError("PID must be a positive decimal process ID")
    pid = int(token)
    pidfd = ops.pidfd_open(pid, 0)
    directory = None
    try:
        directory = ops.open(f"{proc_root}/{pid}", os.O_RDONLY | os.O_DIRECTORY)
        watcher = poll_factory()
        watcher.register(pidfd, select.POLLIN)

        def read(name, limit):
            fd = ops.open(name, os.O_RDONLY, dir_fd=directory)
            with ops.fdopen(fd, "rb") as source:
                data = source.read(limit + 1)
            if len(data) > limit:
                raise ValueError(f"{name} exceeds the snapshot limit")
            return data

        def identity():
            head, tail = read("stat", 65536).rsplit(b")", 1)
            if int(head.split(b" ", 1)[0]) != pid:
                raise ValueError("stat PID mismatch")
            fields = tail.split()  # field 3 is index 0; field 22 is index 19
            if fields[0] in (b"Z", b"X", b"x"):
                raise ProcessLookupError("process has exited")
            return int(fields[19])

        before = identity()
        executable = ops.readlink("exe", dir_fd=directory)
        raw_command = read("cmdline", 65536)
        # Remove one final NUL; empty input means no argument data.
        command = raw_command[:-1] if raw_command.endswith(b"\0") else raw_command
        command = command.split(b"\0") if raw_command else []
        after = identity()
        if before != after or watcher.poll(0):
            raise ProcessLookupError("identity changed or process exited")
        return {"pid": pid, "starttime_ticks": before, "executable": executable,
                "argv": [arg.decode("utf-8", "backslashreplace") for arg in command]}
    finally:
        if directory is not None:
            ops.close(directory)
        ops.close(pidfd)


if __name__ == "__main__":
    try:
        if len(sys.argv) != 2:
            raise ValueError("usage: inspect_io_process.py PID")
        print(json.dumps(snapshot(sys.argv[1]), ensure_ascii=True))
    except (OSError, ValueError, IndexError, AttributeError) as error:
        sys.exit(f"Identity snapshot failed: {error}")

python3 inspect_io_process.py "${PID:?Set PID from the reviewed process row}"