From 7f1ba1f8b9873809b727830a8b1726d55411559e Mon Sep 17 00:00:00 2001 From: Sandro Wenzel Date: Sun, 20 Sep 2026 05:18:01 +0200 Subject: [PATCH] Learn the file-IO graph without root, using strace This makes the file-IO-graph learning pluggable and adds an strace backend, so a pilot run no longer needs a binary with CAP_SYS_ADMIN. The graph it produces is the one --remove-files-early already reads back. - The runner gains --filegraph-backends. Naming several at once runs them side by side, which is how they are compared. O2DPG_PRODUCE_FILEGRAPH still selects fanotify and names its monitor. - The strace backend wraps each task command, so a file access is attributed by the trace it lands in rather than by walking /proc after the event. --seccomp-bpf keeps the cost at about 59 us per traced open; the backend probes for it. - filegraph_report.py holds the exclusion rules, the ./tfN -> ./tfX templating, the JSON schema and the graphviz rendering, so the two analysers cannot drift. analyse_FileIO_v2.py reproduces its previous output byte for byte. - compare_reports.py grades one report against another. A missing edge deletes a file a later task still reads; an extra edge only delays the deletion. The verdicts EXACT, SAFE and UNSAFE follow that asymmetry. - tests/equivalence_test.py runs a workflow whose graph is known by construction and grades every backend against it, in seconds and with no ALICE software. strace comes out EXACT. - monitor_fileaccess_v2.cpp spun forever on a queue overflow, because the overflow branch skipped FAN_EVENT_NEXT and re-tested the same event. - 33 offline tests come with it, plain unittest so they also run on a worker node, and a CI job runs them. Part of the o2dpg_workflow_runner.py refactoring. --- .github/workflows/syntax-checks.yml | 14 + MC/workflow_runner/o2dpg_runner/cli.py | 60 ++--- MC/workflow_runner/o2dpg_runner/config.py | 1 + MC/workflow_runner/o2dpg_runner/executor.py | 20 +- MC/workflow_runner/o2dpg_runner/filegraph.py | 239 ++++++++++++++++++ UTILS/FileIOGraph/README.md | 97 +++++-- UTILS/FileIOGraph/analyse_FileIO_strace.py | 152 +++++++++++ UTILS/FileIOGraph/analyse_FileIO_v2.py | 138 ++++++++++ UTILS/FileIOGraph/compare_reports.py | 156 ++++++++++++ UTILS/FileIOGraph/filegraph_report.py | 173 +++++++++++++ UTILS/FileIOGraph/monitor_fileaccess_v2.cpp | 7 +- UTILS/FileIOGraph/tests/equivalence_test.py | 205 +++++++++++++++ .../tests/filegraph_test_support.py | 20 ++ .../FileIOGraph/tests/test_analyse_strace.py | 131 ++++++++++ UTILS/FileIOGraph/tests/test_filegraph.py | 237 +++++++++++++++++ 15 files changed, 1581 insertions(+), 69 deletions(-) create mode 100644 MC/workflow_runner/o2dpg_runner/filegraph.py create mode 100755 UTILS/FileIOGraph/analyse_FileIO_strace.py create mode 100644 UTILS/FileIOGraph/analyse_FileIO_v2.py create mode 100755 UTILS/FileIOGraph/compare_reports.py create mode 100644 UTILS/FileIOGraph/filegraph_report.py create mode 100755 UTILS/FileIOGraph/tests/equivalence_test.py create mode 100644 UTILS/FileIOGraph/tests/filegraph_test_support.py create mode 100644 UTILS/FileIOGraph/tests/test_analyse_strace.py create mode 100644 UTILS/FileIOGraph/tests/test_filegraph.py diff --git a/.github/workflows/syntax-checks.yml b/.github/workflows/syntax-checks.yml index 942d7b1dd..bd9b5de6e 100644 --- a/.github/workflows/syntax-checks.yml +++ b/.github/workflows/syntax-checks.yml @@ -120,6 +120,20 @@ jobs: working-directory: MC/workflow_runner run: pytest o2dpg_runner/tests -q + filegraph-tests: + name: File-IO-graph unit tests + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Install prerequisites + run: pip install psutil + + - name: Run the FileIOGraph test suite + run: python3 -m unittest discover -s UTILS/FileIOGraph/tests -t UTILS/FileIOGraph/tests + pylint: name: Pylint runs-on: ubuntu-latest diff --git a/MC/workflow_runner/o2dpg_runner/cli.py b/MC/workflow_runner/o2dpg_runner/cli.py index defab9ad2..222780782 100644 --- a/MC/workflow_runner/o2dpg_runner/cli.py +++ b/MC/workflow_runner/o2dpg_runner/cli.py @@ -8,17 +8,16 @@ from __future__ import annotations import argparse -import json import logging import os import shutil -import subprocess import sys from typing import Optional, Tuple import psutil from .config import RunnerConfig +from .filegraph import BACKENDS as FILEGRAPH_BACKENDS, FileGraphManager from .workflow import build_workflow, load_json from .executor import WorkflowExecutor @@ -100,6 +99,11 @@ def build_parser() -> argparse.ArgumentParser: p.add_argument("--retry-on-failure", type=int, default=0) p.add_argument("--no-rootinit-speedup", action="store_true") p.add_argument("--remove-files-early", type=str, default="") + p.add_argument("--filegraph-backends", type=str, + default=os.getenv("O2DPG_FILEGRAPH_BACKENDS", ""), + help="comma-separated file-IO-graph backends to learn the " + "file dependencies with: " + + ", ".join(sorted(FILEGRAPH_BACKENDS))) # Accept-and-ignore for backward compatibility of call sites # that still pass these flags. They have no effect. @@ -154,6 +158,7 @@ def _args_to_config(ns: argparse.Namespace) -> RunnerConfig: retry_on_failure=ns.retry_on_failure, no_rootinit_speedup=ns.no_rootinit_speedup, remove_files_early=ns.remove_files_early, + filegraph_backends=ns.filegraph_backends, stdout_on_failure=ns.stdout_on_failure, production_mode=ns.production_mode, action_logfile=ns.action_logfile, @@ -341,22 +346,6 @@ def _maybe_draw_workflow(raw_spec): dot.render("workflow.gv") -def _launch_fileaccess_sidecar(actionlogger_file: str): - """Start the fanotify-based file-IO graph sidecar if requested.""" - exe = os.getenv("O2DPG_PRODUCE_FILEGRAPH") - if not exe: - return None, None, None - env = os.environ.copy() - env["FILEACCESS_MON_ROOTPATH"] = os.getcwd() - env["MAXMOTHERPID"] = f"{os.getpid()}" - log_file = f"pipeline_fileaccess_{os.getpid()}.log" - fh = open(log_file, "w") - proc = subprocess.Popen( - [exe], stdout=fh, stderr=subprocess.STDOUT, env=env, - ) - return proc, fh, log_file - - def main(argv=None) -> int: ns = build_parser().parse_args(argv) _maybe_reexec_in_slice(ns) # may replace this process; returns only if not re-execing @@ -409,6 +398,7 @@ def main(argv=None) -> int: "systemd_run_spec": cfg.systemd_run_spec, "in_systemd_slice": cfg.in_systemd_slice, "monitor_interval_cpu": cfg.monitor_interval_cpu, + "filegraph_backends": cfg.filegraph_backends, }) metric_logger.info(meta) @@ -429,37 +419,19 @@ def main(argv=None) -> int: for k, v in wf.global_env.items(): os.environ.setdefault(k, str(v)) - # Optional file-access sidecar - fileaccess_proc, fileaccess_fh, fileaccess_log_file = _launch_fileaccess_sidecar(action_log) + filegraph = FileGraphManager.from_config( + cfg.filegraph_backends, os.getcwd(), os.getpid(), action_log, action_logger) + filegraph.start() rc = 0 try: - execer = WorkflowExecutor(cfg, wf, action_logger, metric_logger) + execer = WorkflowExecutor(cfg, wf, action_logger, metric_logger, + filegraph=filegraph) rc = int(execer.execute()) finally: - if fileaccess_proc is not None: - fileaccess_proc.terminate() - try: - fileaccess_proc.wait(timeout=5) - except subprocess.TimeoutExpired: - fileaccess_proc.kill() - if fileaccess_fh is not None: - fileaccess_fh.close() - o2dpg_root = os.getenv("O2DPG_ROOT") - if o2dpg_root and fileaccess_log_file: - analyse_cmd = [ - sys.executable, - f"{o2dpg_root}/UTILS/FileIOGraph/analyse_FileIO_v2.py", - "--actionFile", action_log, - "--monitorFile", fileaccess_log_file, - "-o", f"pipeline_fileaccess_report_{os.getpid()}.json", - "--basedir", os.getcwd(), - ] - print(f"Producing FileIOGraph with command {analyse_cmd}") - try: - subprocess.run(analyse_cmd, check=True) - except subprocess.CalledProcessError as e: - print(f"FileIOGraph analysis failed: {e}", file=sys.stderr) + filegraph.stop() + for backend, path in filegraph.analyse().items(): + print(f"FileIOGraph[{backend}] -> {path}") return rc diff --git a/MC/workflow_runner/o2dpg_runner/config.py b/MC/workflow_runner/o2dpg_runner/config.py index d12bf73c8..5769d0223 100644 --- a/MC/workflow_runner/o2dpg_runner/config.py +++ b/MC/workflow_runner/o2dpg_runner/config.py @@ -54,6 +54,7 @@ class RunnerConfig: retry_on_failure: int = 0 no_rootinit_speedup: bool = False remove_files_early: str = "" + filegraph_backends: str = "" stdout_on_failure: bool = False production_mode: bool = False diff --git a/MC/workflow_runner/o2dpg_runner/executor.py b/MC/workflow_runner/o2dpg_runner/executor.py index 10dc73b45..ca2d178ab 100644 --- a/MC/workflow_runner/o2dpg_runner/executor.py +++ b/MC/workflow_runner/o2dpg_runner/executor.py @@ -34,10 +34,11 @@ from .graph import descendants, longest_path_length, kahn_topological_order from .resources import ResourceManager, ResourceLimitExceeded from .monitoring import MonitorThread, PsutilBackend, _read_cgroup_v2_dir +from .filegraph import FileGraphManager from .scheduler import get_policy from .scheduler.base import SchedulerState from .scheduler.timeframe import TimeframeFirstPolicy -from .cache import TaskCache, compute_fingerprint, remove_done_flag, done_path +from .cache import TaskCache, compute_fingerprint, remove_done_flag from .alienv import get_alienv_software_environment from .cleanup import EarlyFileRemover, archive_task_logs @@ -90,8 +91,10 @@ def __init__( workflow: Workflow, action_logger: logging.Logger, metric_logger: logging.Logger, + filegraph=None, ): self.cfg = config + self.filegraph = filegraph or FileGraphManager([], os.getpid(), action_logger) self.wf = workflow self.actionlog = action_logger self.metriclog = metric_logger @@ -366,16 +369,23 @@ def submit(self, tid: int, nice: int) -> Optional[psutil.Popen]: slice_name if slice_name.endswith(".slice") else f"{slice_name}.slice" ) unit = _unit_name(task["name"], tid) - launch_argv = [ + prefix = [ "systemd-run", "--user", "--scope", "--collect", "--expand-environment=no", # suppress the $VAR warning; bash handles expansion - f"--unit={unit}", f"--slice={systemd_slice}", - "--", "/bin/bash", "-c", cmd, + f"--unit={unit}", f"--slice={systemd_slice}", "--", ] + else: + prefix = [] + + # a tracer has to sit inside any systemd scope, or it would only ever + # see systemd-run itself + inner_argv = self.filegraph.wrap(["/bin/bash", "-c", cmd], task["name"], tid) + launch_argv = prefix + inner_argv + + if use_scope: p = psutil.Popen(launch_argv, cwd=workdir, env=env, stderr=subprocess.PIPE) _start_stderr_drainer(p.stderr, self.actionlog, task["name"]) else: - launch_argv = ["/bin/bash", "-c", cmd] p = psutil.Popen(launch_argv, cwd=workdir, env=env) try: p.nice(nice) diff --git a/MC/workflow_runner/o2dpg_runner/filegraph.py b/MC/workflow_runner/o2dpg_runner/filegraph.py new file mode 100644 index 000000000..27da41bf1 --- /dev/null +++ b/MC/workflow_runner/o2dpg_runner/filegraph.py @@ -0,0 +1,239 @@ +"""Pluggable observation of task file IO during a pilot run. + +A backend records which task produces and which tasks consume each +intermediate file and writes ``filegraph__.json``, which a +later production run replays with ``--remove-files-early``. Several +backends can be active at once so they can be compared on one run; see +UTILS/FileIOGraph/README.md. +""" + +from __future__ import annotations + +import logging +import os +import shutil +import subprocess +import sys +from typing import Dict, List, Optional + + +def filegraph_dir() -> str: + """The UTILS/FileIOGraph directory holding the monitors and analysers.""" + root = os.getenv("O2DPG_ROOT") + if not root: + root = os.path.abspath(__file__) + while root != "/" and not os.path.isdir( + os.path.join(root, "UTILS", "FileIOGraph")): + root = os.path.dirname(root) + return os.path.join(root, "UTILS", "FileIOGraph") + + +class FileGraphBackend: + """One way of observing which task reads and writes which file.""" + + name = "none" + analyser = "" + #: backends that replace the original fanotify sidecar also write its + #: report under the name the runner has always used + legacy_report = False + + def __init__(self, workdir: str, runner_pid: int, action_log: str, + logger: logging.Logger): + self.workdir = workdir + self.runner_pid = runner_pid + self.action_log = action_log + self.log = logger + + def start(self) -> None: + pass + + def stop(self) -> None: + pass + + def wrap(self, argv: List[str], taskname: str, tid: int) -> List[str]: + """The argv actually used to launch a task.""" + return argv + + # -- result -------------------------------------------------------- + def recorded(self) -> bool: + return False + + def analyser_args(self) -> List[str]: + return [] + + def analyse(self, out_json: str) -> bool: + if not self.recorded(): + return False + argv = [sys.executable, os.path.join(filegraph_dir(), self.analyser), + "--basedir", self.workdir, "-o", out_json] + self.analyser_args() + self.log.info("FileIOGraph analysis: %s", " ".join(argv)) + try: + subprocess.run(argv, check=True) + return True + except (subprocess.CalledProcessError, OSError) as e: + self.log.error("FileIOGraph analysis failed: %s", e) + return False + + +class FanotifyBackend(FileGraphBackend): + """Mount-wide sidecar; needs CAP_SYS_ADMIN on the monitor binary.""" + + name = "fanotify" + analyser = "analyse_FileIO_v2.py" + legacy_report = True + + def __init__(self, workdir, runner_pid, action_log, logger, + exe: Optional[str] = None): + super().__init__(workdir, runner_pid, action_log, logger) + # O2DPG_PRODUCE_FILEGRAPH is the original spelling and still names + # the monitor to run + self.exe = (exe or os.getenv("O2DPG_PRODUCE_FILEGRAPH") + or os.path.join(filegraph_dir(), "monitor_fileaccess_v2.exe")) + self.logfile = f"pipeline_fileaccess_{self.name}_{runner_pid}.log" + self._proc: Optional[subprocess.Popen] = None + self._fh = None + + def start(self) -> None: + if not os.path.exists(self.exe): + self.log.error("filegraph fanotify: no such executable %s", self.exe) + return + env = dict(os.environ, FILEACCESS_MON_ROOTPATH=self.workdir, + MAXMOTHERPID=str(self.runner_pid)) + self._fh = open(self.logfile, "w") + try: + self._proc = subprocess.Popen([self.exe], stdout=self._fh, + stderr=subprocess.STDOUT, env=env) + except OSError as e: + self.log.error("filegraph fanotify: could not start %s: %s", self.exe, e) + self._fh.close() + self._fh = None + return + self.log.info("filegraph fanotify: %s pid=%d -> %s", + self.exe, self._proc.pid, self.logfile) + + def stop(self) -> None: + if self._proc is not None: + self._proc.terminate() + try: + self._proc.wait(timeout=10) + except subprocess.TimeoutExpired: + self._proc.kill() + self._proc = None + if self._fh is not None: + self._fh.close() + self._fh = None + + def recorded(self) -> bool: + return os.path.exists(self.logfile) + + def analyser_args(self) -> List[str]: + return ["--actionFile", self.action_log, "--monitorFile", self.logfile] + + +class StraceBackend(FileGraphBackend): + """Wraps every task in its own strace; attribution is by construction.""" + + name = "strace" + analyser = "analyse_FileIO_strace.py" + #: must stay in step with what analyse_FileIO_strace.py parses + SYSCALLS = "openat,openat2,open,creat,rename,renameat,renameat2" + + def __init__(self, workdir, runner_pid, action_log, logger, + exe: str = "strace"): + super().__init__(workdir, runner_pid, action_log, logger) + self.exe = exe + self.tracedir = os.path.join(workdir, f"strace_{runner_pid}") + self._argv: List[str] = [] + + def start(self) -> None: + if not shutil.which(self.exe): + self.log.error("filegraph strace: %s not found on PATH", self.exe) + return + argv = [self.exe, "-f", "-qq", "-y", + "-e", f"trace={self.SYSCALLS}", "-e", "status=successful"] + # without --seccomp-bpf every read() costs two ptrace stops + if self._seccomp_works(argv): + argv.insert(1, "--seccomp-bpf") + os.makedirs(self.tracedir, exist_ok=True) + self._argv = argv + self.log.info("filegraph strace: %s -> %s", " ".join(argv), self.tracedir) + + def _seccomp_works(self, argv: List[str]) -> bool: + try: + r = subprocess.run(argv[:1] + ["--seccomp-bpf"] + argv[1:] + + ["-o", os.devnull, "true"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + timeout=60) + return r.returncode == 0 + except (OSError, subprocess.SubprocessError): + return False + + def wrap(self, argv, taskname, tid): + if not self._argv: + return argv + trace = os.path.join(self.tracedir, f"trace_{tid}_{taskname}.log") + return self._argv + ["-o", trace, "--"] + list(argv) + + def recorded(self) -> bool: + return os.path.isdir(self.tracedir) + + def analyser_args(self) -> List[str]: + return ["--straceDir", self.tracedir] + + +BACKENDS = {b.name: b for b in (FanotifyBackend, StraceBackend)} + + +class FileGraphManager: + """Drives zero or more backends over the runner's lifecycle.""" + + def __init__(self, backends: List[FileGraphBackend], runner_pid: int, + logger: logging.Logger): + self.backends = backends + self.runner_pid = runner_pid + self.log = logger + + @classmethod + def from_config(cls, spec: str, workdir: str, runner_pid: int, + action_log: str, logger: logging.Logger) -> "FileGraphManager": + names = [n.strip() for n in spec.split(",") if n.strip()] + if not names and os.getenv("O2DPG_PRODUCE_FILEGRAPH"): + names = ["fanotify"] + backends = [] + for n in names: + factory = BACKENDS.get(n) + if factory is None: + logger.error("unknown filegraph backend %r (known: %s)", + n, ", ".join(sorted(BACKENDS))) + continue + backends.append(factory(workdir, runner_pid, action_log, logger)) + return cls(backends, runner_pid, logger) + + def start(self) -> None: + for b in self.backends: + b.start() + + def stop(self) -> None: + for b in self.backends: + b.stop() + + def wrap(self, argv: List[str], taskname: str, tid: int) -> List[str]: + for b in self.backends: + argv = b.wrap(argv, taskname, tid) + return argv + + def analyse(self) -> Dict[str, str]: + """Produce one report per backend; returns backend name -> path.""" + produced: Dict[str, str] = {} + for b in self.backends: + out = f"filegraph_{b.name}_{self.runner_pid}.json" + if not b.analyse(out): + continue + produced[b.name] = out + if b.legacy_report: + legacy = f"pipeline_fileaccess_report_{self.runner_pid}.json" + try: + shutil.copyfile(out, legacy) + except OSError as e: + self.log.warning("could not write %s: %s", legacy, e) + return produced diff --git a/UTILS/FileIOGraph/README.md b/UTILS/FileIOGraph/README.md index 48443b310..8450e09af 100644 --- a/UTILS/FileIOGraph/README.md +++ b/UTILS/FileIOGraph/README.md @@ -1,38 +1,97 @@ -This is a small custom tool to monitor file access -and to produce graphs of file production and file consumption -by O2DPG Monte Carlo tasks. Such information can be useful for +# FileIOGraph — learning which task produces and which tasks consume each file -(a) verification of data paths -(b) early removal of files as soon as they are not needed anymore +An O2DPG Monte Carlo workflow declares task dependencies, but not which +file each task reads or writes. The tools here observe a pilot run and +write that missing relation down, so a production run can +(a) verify the data paths, and +(b) delete an intermediate file the moment its last consumer is finished + (`o2dpg_workflow_runner.py --remove-files-early`). -In more detail, core elements of this directory are +## Backends -* monitor_fileaccess: +The observation is pluggable: -A tool, useable by root, providing reports about -read and write events to files and which process is involved. -The tool is based on the efficient fanotify kernel system and reporting -can be restricted to certain shells (by giving a mother PID). +```bash +ALIEN_O2DPG_WORKFLOW_RUNNER=new $O2DPG_ROOT/MC/bin/o2_dpg_workflow_runner.py \ + -f workflow.json --filegraph-backends fanotify +``` -The tool is standalone and can be compiled, if needed, by running +`--filegraph-backends` exists only in the runner under +`MC/workflow_runner`, hence the environment variable. -`g++ monitor_fileaccess.cpp -O2 -o monitor_fileaccess.exe` +Each named backend writes `filegraph__.json`; fanotify also +writes `pipeline_fileaccess_report_.json`, the name the runner has +always used. Naming several at once is how they are compared. -The tool can be run simply by +| Backend | How it observes | How it attributes | Privilege | +|---|---|---|---| +| `fanotify` | `monitor_fileaccess_v2.exe`, a mount-wide sidecar | walks `/proc/` up to the runner, after the event | `CAP_SYS_ADMIN` | +| `strace` | each task command wrapped in its own `strace` | by construction, from the trace file it lands in | none | -``` -sudo MAXMOTHERPID=689584 ./monitor.exe | tee /tmp/fileaccess -``` +`O2DPG_PRODUCE_FILEGRAPH=` still selects fanotify and names the +monitor to run. + +### fanotify + +The original. It needs `CAP_SYS_ADMIN`, granted per machine with +`setcap cap_sys_admin+ep monitor_fileaccess_v2.exe`, so a pilot run is +only possible where somebody with root has prepared the binary — and file +capabilities do not survive a `nosuid` mount, so the copy distributed +through CVMFS can never carry one. + +Unprivileged fanotify does not help: since Linux 5.13 `fanotify_init()` +needs no `CAP_SYS_ADMIN`, but such a group may not use `FAN_MARK_MOUNT` +and does not receive the pid, and the pid is the whole attribution +mechanism. -to monitor file events happening by child processes of shell 689584. +Two further properties are worth knowing. A mount-wide mark sees +everything, so in a two-timeframe pilot only 4 301 of 563 611 records were +inside the working directory. And because the process chain is resolved +from `/proc` after the event, an access made by a process that has already +exited yields the chain `;0` and is dropped. +### strace -* analyse_FileIO.py: +Wraps each task command, so attribution needs no inference. It sees the +`open` itself rather than the close, and `-y` gives the path the kernel +resolved the descriptor to, so relative paths, `chdir` and directory +descriptors all come out right. +`--seccomp-bpf` is what makes this affordable; the backend probes for it +and uses it when present. Without it every `read()` costs two ptrace +stops: on a read-heavy task, 22.5 s against a 0.95 s baseline, versus +1.00 s with it. The residual is about 59 µs per traced `open`. +## Checking one backend against another +```bash +python3 compare_reports.py --reference filegraph_fanotify_123.json \ + --candidate filegraph_strace_123.json +``` + +`EXACT` means edge for edge, `SAFE` that the candidate is a superset, +`UNSAFE` that it misses an edge the reference has. The asymmetry is the +point — an extra reader only keeps a file on disc longer, a missing one +deletes a file a later task still needs. + +`tests/equivalence_test.py` runs a synthetic workflow whose graph is known +by construction under several backends at once and grades all of them, in +seconds and without any ALICE software: +```bash +python3 tests/equivalence_test.py --backends strace --reference fanotify \ + --ntf 8 --cpu-limit 8 --sleep 0.05 +``` + +Drop `--reference fanotify` where no privileged monitor exists; the +synthetic workflow still carries its own analytic truth. +Offline tests: `python3 -m unittest discover -s tests -t tests`. +## Building +```bash +g++ monitor_fileaccess_v2.cpp -O2 -o monitor_fileaccess_v2.exe +sudo setcap cap_sys_admin+ep monitor_fileaccess_v2.exe # fanotify only +``` diff --git a/UTILS/FileIOGraph/analyse_FileIO_strace.py b/UTILS/FileIOGraph/analyse_FileIO_strace.py new file mode 100755 index 000000000..2ea49e99e --- /dev/null +++ b/UTILS/FileIOGraph/analyse_FileIO_strace.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +"""Turn per-task strace logs into the file-task dependency report. + +The runner wraps each task in its own strace, so the trace file name says +which task made the call and nothing has to be attributed after the fact. +Traces are taken with -y, so the path on the *return* value is the one the +kernel resolved -- which is what makes relative paths, chdir and a +non-AT_FDCWD directory descriptor all come out right. +""" +from __future__ import annotations + +import argparse +import os +import re +import sys +from typing import Dict, List, Set, Tuple + +from filegraph_report import ( + add_common_arguments, basedir_prefix, emit, keep_file, relative_to_basedir, +) + +#: trace__.log, written by the runner's strace backend +TRACEFILE_RE = re.compile(r'^trace_(?P\d+)_(?P.+)\.log$') + +OPEN_CALLS = ("openat", "openat2", "open", "creat") +#: longest first, or 'openat' would match the head of 'openat2' +_OPEN_ALT = "|".join(sorted(OPEN_CALLS, key=len, reverse=True)) + +_OPEN_RE = re.compile( + rf'(?:^|\s)(?:{_OPEN_ALT})\((?P.*?)\)\s*=\s*\d+<(?P[^>]*)>') +_RENAME_RE = re.compile(r'(?:^|\s)rename(?:at2?)?\((?P.*?)\)\s*=\s*0') +_UNFINISHED_RE = re.compile(r'^(?P\d+)\s+(?P\w+)\((?P.*?)\d+)\s+<\.\.\.\s+(?P\w+)\s+resumed>(?P.*)$') +_RESULT_RE = re.compile(r'=\s*\d+<(?P[^>]*)>') +_QUOTED_RE = re.compile(r'"((?:[^"\\]|\\.)*)"') + +#: opendir reaches the kernel as an open with O_DIRECTORY; fanotify reports +#: directories only with FAN_ONDIR, so they are in neither graph +_DIRECTORY = "O_DIRECTORY" +_WRITE_FLAGS = ("O_WRONLY", "O_RDWR", "O_CREAT", "O_TRUNC", "O_APPEND") + + +def _kind(args: str) -> str: + return "write" if any(f in args for f in _WRITE_FLAGS) else "read" + + +def parse_trace(path: str) -> Tuple[Set[Tuple[str, str]], int]: + """Distinct (absolute path, read|write) pairs for one task, and the + number of calls they were distilled from.""" + seen: Set[Tuple[str, str]] = set() + calls = 0 + # strace splits a call around another process's entry; the flags are on + # the half that was interrupted + pending: Dict[Tuple[str, str], str] = {} + + with open(path, errors="replace") as fh: + for line in fh: + if "" in line: + m = _RESUMED_RE.match(line) + if m is None or m.group("call") not in OPEN_CALLS: + continue + res = _RESULT_RE.search(m.group("rest")) + if res is None or not res.group("path"): + continue + args = (pending.pop((m.group("pid"), m.group("call")), "") + + m.group("rest").split("=", 1)[0]) + calls += 1 + if _DIRECTORY not in args: + seen.add((res.group("path"), _kind(args))) + continue + + m = _OPEN_RE.search(line) + if m is not None: + calls += 1 + if m.group("path") and _DIRECTORY not in m.group("args"): + seen.add((m.group("path"), _kind(m.group("args")))) + continue + + if "rename" in line: + m = _RENAME_RE.search(line) + if m is None: + continue + names = _QUOTED_RE.findall(m.group("args")) + if names: + calls += 1 + seen.add((names[-1], "write")) # the destination is produced here + + return seen, calls + + +def collect(tracedir: str, basedir: str, file_filters): + written: Dict[str, Set[str]] = {} + read: Dict[str, Set[str]] = {} + tasks: List[str] = [] + prefix = basedir_prefix(basedir) + stats = {"traces": 0, "calls": 0, "kept": 0} + + for name in sorted(os.listdir(tracedir)): + m = TRACEFILE_RE.match(name) + if m is None: + continue + task = m.group("task") + tasks.append(task) + stats["traces"] += 1 + + accesses, calls = parse_trace(os.path.join(tracedir, name)) + stats["calls"] += calls + for target, kind in accesses: + rel = relative_to_basedir(target, prefix) + if rel is None or not keep_file(rel, file_filters): + continue + stats["kept"] += 1 + (written if kind == "write" else read).setdefault(rel, set()).add(task) + + return written, read, sorted(set(tasks)), stats + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--straceDir", required=True, + help="directory of trace__.log files") + add_common_arguments(p) + return p + + +def main(argv=None) -> int: + a = build_parser().parse_args(argv) + if not os.path.isdir(a.straceDir): + print(f"no such directory: {a.straceDir}", file=sys.stderr) + return 2 + + written, read, tasks, stats = collect( + a.straceDir, a.basedir, [re.compile(f) for f in a.file_filters]) + print(f"strace: {stats['traces']} task trace(s), {stats['calls']} call(s), " + f"{stats['kept']} access(es) kept under {a.basedir}") + if not stats["traces"]: + print(f"WARNING: no trace_*.log in {a.straceDir}", file=sys.stderr) + + emit(a, written, read, tasks) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/UTILS/FileIOGraph/analyse_FileIO_v2.py b/UTILS/FileIOGraph/analyse_FileIO_v2.py new file mode 100644 index 000000000..67951f183 --- /dev/null +++ b/UTILS/FileIOGraph/analyse_FileIO_v2.py @@ -0,0 +1,138 @@ +#!/usr/bin/env python3 +"""Turn a fanotify file-access log plus the runner action log into the +file-task dependency report. + +Attribution comes from the process chain the monitor records: an access is +a task's if any pid in the chain is the pid the action log gives that task. +Both action-log formats are accepted: + + Old runner: "... INFO Task : finished with status 0" + New runner: "... INFO Task pid= tid= finished rc=0" +""" +from __future__ import annotations + +import argparse +import re +import sys +from typing import Dict, List, Set, Tuple + +from filegraph_report import ( + add_common_arguments, basedir_prefix, emit, keep_file, relative_to_basedir, +) + + +# ── action-log patterns ─────────────────────────────────────────────────────── + +# Old runner: "... INFO Task : finished with status 0" +_PAT_OLD = re.compile(r'.*INFO Task (\d+)[^:]*:(\w+) finished with status 0') +# New runner: "... INFO Task pid= tid= finished rc=0" +_PAT_NEW = re.compile(r'.*INFO Task pid=(\d+) tid=\d+ (\S+) finished rc=0') + + +def parse_action_log(path: str) -> Tuple[Dict[str, str], Dict[str, str]]: + """Parse task-PID associations from an action log. + + Returns (pid_to_task, task_to_pid). Only successfully completed tasks + are included (rc=0 / status 0) so failed retries don't pollute the map. + """ + pid_to_task: Dict[str, str] = {} + task_to_pid: Dict[str, str] = {} + with open(path) as fh: + for line in fh: + m = _PAT_OLD.match(line) or _PAT_NEW.match(line) + if m: + pid, name = m.group(1), m.group(2) + pid_to_task[pid] = name + task_to_pid[name] = pid + return pid_to_task, task_to_pid + + +# ── monitor-log parsing ─────────────────────────────────────────────────────── + +_RECORD_RE = re.compile(r'"?([^"]+)"?,(read|write),(.*)') + + +def parse_monitor_log( + path: str, + pid_to_task: Dict[str, str], + basedir: str, + file_filters: List[re.Pattern], +) -> Tuple[Dict[str, Set[str]], Dict[str, Set[str]]]: + """Parse the fanotify raw log and map files to the tasks that touched them. + + Returns (file_written_by, file_read_by) where each value is a set of + task names. Only files inside *basedir* that pass *file_filters* and + are not excluded by the built-in exclude pattern are included. + + A file access is attributed to a task when any PID in the process-chain + column of the monitor log appears in *pid_to_task*. This works for both + direct children and children-of-children of the task process because the + fanotify monitor records the full ancestor chain up to the root PID. + """ + file_written: Dict[str, Set[str]] = {} + file_read: Dict[str, Set[str]] = {} + prefix = basedir_prefix(basedir) + + with open(path) as fh: + for line in fh: + m = _RECORD_RE.match(line) + if not m: + continue + fname, mode, chain = m.group(1), m.group(2), m.group(3) + + rel = relative_to_basedir(fname, prefix) + if rel is None or not keep_file(rel, file_filters): + continue + + for pid in chain.split(";"): + task = pid_to_task.get(pid) + if task is None: + continue + if mode == "write": + file_written.setdefault(rel, set()).add(task) + else: + file_read.setdefault(rel, set()).add(task) + + return file_written, file_read + + +# ── CLI ─────────────────────────────────────────────────────────────────────── + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + p.add_argument("--actionFile", required=True, + help="O2DPG pipeline runner action log") + p.add_argument("--monitorFile", required=True, + help="fanotify raw log from monitor_fileaccess_v2.exe") + add_common_arguments(p) + return p + + +def main(argv=None) -> None: + args = build_parser().parse_args(argv) + + file_filters = [re.compile(f) for f in args.file_filters] + + pid_to_task, task_to_pid = parse_action_log(args.actionFile) + if not pid_to_task: + print( + f"WARNING: no task completions found in {args.actionFile}.\n" + "Check that the action log is from a completed run and that\n" + "its format matches either the old or new O2DPG runner.", + file=sys.stderr, + ) + else: + print(f"Action log: {len(pid_to_task)} completed task(s) found") + + file_written, file_read = parse_monitor_log( + args.monitorFile, pid_to_task, args.basedir, file_filters, + ) + + emit(args, file_written, file_read, sorted(task_to_pid)) + + +if __name__ == "__main__": + main() diff --git a/UTILS/FileIOGraph/compare_reports.py b/UTILS/FileIOGraph/compare_reports.py new file mode 100755 index 000000000..588abae78 --- /dev/null +++ b/UTILS/FileIOGraph/compare_reports.py @@ -0,0 +1,156 @@ +#!/usr/bin/env python3 +"""Grade one file-graph report against another. + + EXACT candidate and reference agree edge for edge + SAFE candidate is a superset: nothing missing, some extra + UNSAFE candidate misses an edge the reference has + +The asymmetry is the point: the runner deletes a file once every task +listed against it has finished, so a missing edge deletes a file a later +task still reads, while an extra one only keeps it on disc longer. +""" +from __future__ import annotations + +import argparse +import json +import sys +from typing import Dict, List, Set, Tuple + +SECTIONS = ("file_report", "file_template_report") +Edges = Dict[Tuple[str, str], Set[str]] + + +def _edges(report: Dict, section: str) -> Edges: + """(file, 'written_by'|'read_by') -> set of task names.""" + out: Edges = {} + for entry in report.get(section) or []: + f = entry.get("file") + if not f: + continue + for kind in ("written_by", "read_by"): + out[(f, kind)] = set(entry.get(kind) or []) + return out + + +def _diff(a: Edges, b: Edges) -> List[Tuple[str, str, List[str]]]: + """Edges present in a and not in b, as (file, kind, tasks).""" + out = [] + for (f, kind), tasks in a.items(): + extra = tasks - b.get((f, kind), set()) + if extra: + out.append((f, kind, sorted(extra))) + return sorted(out) + + +class SectionDiff: + def __init__(self, section: str, ref: Dict, cand: Dict): + self.section = section + re_, ce = _edges(ref, section), _edges(cand, section) + self.ref_files = {f for f, _ in re_} + self.cand_files = {f for f, _ in ce} + self.only_ref = sorted(self.ref_files - self.cand_files) + self.only_cand = sorted(self.cand_files - self.ref_files) + self.missing = _diff(re_, ce) + self.extra = _diff(ce, re_) + self.n_ref_edges = sum(len(v) for v in re_.values()) + self.n_cand_edges = sum(len(v) for v in ce.values()) + self.n_missing = sum(len(t) for _, _, t in self.missing) + self.n_extra = sum(len(t) for _, _, t in self.extra) + + @property + def verdict(self) -> str: + if self.n_missing: + return "UNSAFE" + return "SAFE" if self.n_extra else "EXACT" + + @property + def recall(self) -> float: + if not self.n_ref_edges: + return 1.0 + return (self.n_ref_edges - self.n_missing) / self.n_ref_edges + + def report(self, limit: int) -> str: + def block(header, items, fmt): + if not items: + return [] + out = [f" {header} ({len(items)}):"] + out += [f" {fmt(i)}" for i in items[:limit]] + if len(items) > limit: + out.append(f" ... {len(items) - limit} more") + return out + + lines = [ + f"[{self.section}]", + f" files reference={len(self.ref_files)} " + f"candidate={len(self.cand_files)} only-ref={len(self.only_ref)} " + f"only-cand={len(self.only_cand)}", + f" edges reference={self.n_ref_edges} " + f"candidate={self.n_cand_edges} missing={self.n_missing} " + f"extra={self.n_extra} recall={self.recall * 100:.2f}%", + f" verdict {self.verdict}", + ] + lines += block("files the candidate never saw", self.only_ref, str) + lines += block("MISSING edges", self.missing, + lambda e: f"- {e[0]} {e[1]}: {', '.join(e[2])}") + lines += block("extra edges, harmless", self.extra, + lambda e: f"+ {e[0]} {e[1]}: {', '.join(e[2])}") + return "\n".join(lines) + + +def compare(ref: Dict, cand: Dict, limit: int = 20) -> Tuple[str, Dict]: + diffs = [SectionDiff(s, ref, cand) for s in SECTIONS] + verdicts = [d.verdict for d in diffs] + overall = ("UNSAFE" if "UNSAFE" in verdicts + else "SAFE" if "SAFE" in verdicts else "EXACT") + summary = { + "verdict": overall, + "sections": { + d.section: { + "verdict": d.verdict, + "reference_edges": d.n_ref_edges, + "candidate_edges": d.n_cand_edges, + "missing_edges": d.n_missing, + "extra_edges": d.n_extra, + "recall": d.recall, + "files_only_in_reference": d.only_ref, + "files_only_in_candidate": d.only_cand, + "missing": [list(e) for e in d.missing], + "extra": [list(e) for e in d.extra], + } for d in diffs + }, + } + return "\n".join(d.report(limit) for d in diffs), summary + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--reference", required=True, help="the report to be reproduced") + ap.add_argument("--candidate", required=True, help="the report under test") + ap.add_argument("--limit", type=int, default=20, help="differences to print per class") + ap.add_argument("--json", default=None, help="also write the full diff here") + ap.add_argument("--allow", choices=["EXACT", "SAFE"], default="SAFE", + help="worst verdict that still exits 0 (default: SAFE)") + a = ap.parse_args(argv) + + with open(a.reference) as f: + ref = json.load(f) + with open(a.candidate) as f: + cand = json.load(f) + + text, summary = compare(ref, cand, a.limit) + print(f"reference: {a.reference}") + print(f"candidate: {a.candidate}") + print(text) + print(f"\nOVERALL: {summary['verdict']}") + if a.json: + with open(a.json, "w") as f: + json.dump(summary, f, indent=2) + print(f"wrote {a.json}") + + return 0 if summary["verdict"] in ("EXACT", a.allow) else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/UTILS/FileIOGraph/filegraph_report.py b/UTILS/FileIOGraph/filegraph_report.py new file mode 100644 index 000000000..952110771 --- /dev/null +++ b/UTILS/FileIOGraph/filegraph_report.py @@ -0,0 +1,173 @@ +"""Build the file-task dependency report from two file->tasks maps. + +Holds the exclusion rules, the ./tfN -> ./tfX templating, the JSON schema +that --remove-files-early reads back, and the graphviz rendering. +""" +from __future__ import annotations + +import json +import re +import sys +from typing import Dict, Optional, Sequence, Set + +#: log files, the CCDB log and the DPL config are noise, not data flow +EXCLUDE_RE = re.compile(r'(.*\.log.*|ccdb/log|.*dpl-config\.json)') +TF_PATH_RE = re.compile(r'^\./tf(?P\d+)/') + + +def basedir_prefix(basedir: str) -> str: + return basedir.rstrip("/") + "/" + + +def relative_to_basedir(fname: str, prefix: str) -> Optional[str]: + """'./x/y' for a path under the prefix from basedir_prefix(), else None.""" + if not fname.startswith(prefix): + return None + return "./" + fname[len(prefix):] + + +def keep_file(rel: str, file_filters: Sequence) -> bool: + if EXCLUDE_RE.match(rel): + return False + return any(r.match(rel) for r in file_filters) + + +def task_template_for_timeframe(task_name: str, source_tf: int) -> str: + suffix = f"_{source_tf}" + if task_name.endswith(suffix): + return f"{task_name[:-len(suffix)]}_X" + return task_name + + +def build_report(file_written: Dict[str, Set[str]], + file_read: Dict[str, Set[str]], + tasks: Sequence[str]) -> Dict: + """Assemble the JSON document from the two file->tasks maps.""" + all_files = sorted(set(file_written) | set(file_read)) + file_report = [ + { + "file": f, + "written_by": sorted(file_written.get(f, set())), + "read_by": sorted(file_read.get(f, set())), + } + for f in all_files + ] + + templates: Dict[str, Dict] = {} + for entry in file_report: + match = TF_PATH_RE.match(entry["file"]) + if match is None: + continue + source_tf = int(match.group("tf")) + merged = templates.setdefault( + TF_PATH_RE.sub("./tfX/", entry["file"], count=1), + {"written_by": set(), "read_by": set(), "source_timeframes": set()}, + ) + merged["source_timeframes"].add(source_tf) + for kind in ("written_by", "read_by"): + for task in entry[kind]: + merged[kind].add(task_template_for_timeframe(task, source_tf)) + + file_template_report = [ + { + "file": f, + "written_by": sorted(v["written_by"]), + "read_by": sorted(v["read_by"]), + "source_timeframes": sorted(v["source_timeframes"]), + } + for f, v in sorted(templates.items()) + ] + + task_reads: Dict[str, Set[str]] = {} + task_writes: Dict[str, Set[str]] = {} + for f, ts in file_read.items(): + for t in ts: + task_reads.setdefault(t, set()).add(f) + for f, ts in file_written.items(): + for t in ts: + task_writes.setdefault(t, set()).add(f) + task_report = [ + { + "task": t, + "writes": sorted(task_writes.get(t, set())), + "reads": sorted(task_reads.get(t, set())), + } + for t in sorted(tasks) + ] + + return { + "file_report": file_report, + "file_template_report": file_template_report, + "task_report": task_report, + } + + +def draw_graph(filename: str, file_written: Dict[str, Set[str]], + file_read: Dict[str, Set[str]], tasks: Sequence[str]) -> None: + try: + from graphviz import Digraph + except ImportError: + print("graphviz not installed, skipping graph", file=sys.stderr) + return + + ccdb_re = re.compile(r"ccdb(.*)/snapshot\.root") + dot = Digraph(comment="O2DPG file-task network") + idx: Dict[str, int] = {} + + # CCDB snapshots are labelled by object path; a real workflow pulls + # dozens and the full names swamp the picture + ccdb, normal = [], [] + for f in set(file_written) | set(file_read): + m = ccdb_re.match(f) + (ccdb if m else normal).append((f, m.group(1) if m else f)) + + with dot.subgraph(name="CCDB") as sg: + sg.attr(color="blue") + for f, label in ccdb: + idx[f] = len(idx) + sg.node(str(idx[f]), label, color="blue") + + with dot.subgraph(name="normal") as sg: + sg.attr(color="black") + for f, label in normal: + idx[f] = len(idx) + sg.node(str(idx[f]), label, color="red") + for t in tasks: + idx[t] = len(idx) + sg.node(str(idx[t]), t, shape="box", color="green", style="filled") + + for f, ts in file_read.items(): + for t in ts: + dot.edge(str(idx[f]), str(idx[t])) + for f, ts in file_written.items(): + for t in ts: + dot.edge(str(idx[t]), str(idx[f])) + + dot.render(filename, format="pdf") + dot.render(filename, format="gv") + print(f"Wrote {filename}.pdf and {filename}.gv") + + +def add_common_arguments(parser) -> None: + parser.add_argument("--basedir", default="/", + help="Workflow working directory (default: /)") + parser.add_argument("--file-filters", nargs="+", default=[r".*"], + help="Regex filters to select file paths (default: all)") + parser.add_argument("--graphviz", default=None, + help="also render the file/task network with this " + "base filename") + parser.add_argument("-o", "--output", required=True, + help="Output JSON report path") + + +def emit(args, file_written: Dict[str, Set[str]], file_read: Dict[str, Set[str]], + tasks: Sequence[str]) -> None: + """Render and write what add_common_arguments() asked for.""" + if args.graphviz: + draw_graph(args.graphviz, file_written, file_read, tasks) + doc = build_report(file_written, file_read, tasks) + with open(args.output, "w") as fh: + json.dump(doc, fh, indent=2) + print(f"Wrote {args.output}: {len(doc['file_report'])} file(s) referenced, " + f"{len(doc['file_template_report'])} timeframe file template(s), " + f"{len(doc['task_report'])} task(s) mapped") diff --git a/UTILS/FileIOGraph/monitor_fileaccess_v2.cpp b/UTILS/FileIOGraph/monitor_fileaccess_v2.cpp index 2f3750bc6..b3d89ae8b 100644 --- a/UTILS/FileIOGraph/monitor_fileaccess_v2.cpp +++ b/UTILS/FileIOGraph/monitor_fileaccess_v2.cpp @@ -180,7 +180,12 @@ int main(int argc, char **argv) { if (metadata->mask & FAN_Q_OVERFLOW) { - fprintf(stderr, "Queue overflow!\n"); + // 'continue' here would re-test the same event forever, because it + // skips FAN_EVENT_NEXT; report and advance instead. + printf("#OVERFLOW\n"); + fflush(stdout); + fprintf(stderr, "Queue overflow! events were dropped\n"); + metadata = FAN_EVENT_NEXT(metadata, buflen); continue; } snprintf(fdpath, sizeof(fdpath), "/proc/self/fd/%d", metadata->fd); diff --git a/UTILS/FileIOGraph/tests/equivalence_test.py b/UTILS/FileIOGraph/tests/equivalence_test.py new file mode 100755 index 000000000..04541ee5b --- /dev/null +++ b/UTILS/FileIOGraph/tests/equivalence_test.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +"""Build a workflow whose file-IO graph is known, run it under one or more +backends, and grade every report against that graph and against a reference. + + equivalence_test.py --backends strace --reference fanotify --ntf 8 +""" +from __future__ import annotations + +import argparse +import glob +import json +import os +import subprocess +import sys +import time + +from filegraph_test_support import REPO, filegraph + +from compare_reports import compare # noqa: E402 +from filegraph_report import build_report # noqa: E402 + +RESOURCES = {"cpu": 1, "mem": 100, "relative_cpu": 1.0} + + +def build_workflow(ntf: int, sleep: str): + """A workflow plus the file->tasks maps its commands imply. + + Covers what makes attribution hard: a global task consuming every + timeframe, a file with two readers, a subdirectory created at run time, + four different libc entry points, and tasks that overlap in time. + """ + stages, written, read = [], {}, {} + + def w(f, t): + written.setdefault(f, set()).add(t) + + def r(f, t): + read.setdefault(f, set()).add(t) + + def stage(name, cmd, needs, cwd, tf, label): + stages.append({"name": name, "cmd": f"sleep {sleep}; {cmd}", + "needs": needs, "cwd": cwd, "timeframe": tf, + "labels": [label], "resources": RESOURCES}) + + stage("bkg", "echo bkgdata > bkg.dat", [], "./", -1, "SIM") + w("./bkg.dat", "bkg") + + for i in range(1, ntf + 1): + tf = f"tf{i}" + # shell redirection and Python open() in one task + stage(f"sgnsim_{i}", + f"cat ../bkg.dat > sgn.dat; " + f"python3 -c \"open('kine.dat','w').write('kine{i}')\"", + ["bkg"], f"./{tf}", i, "SIM") + r("./bkg.dat", f"sgnsim_{i}") + w(f"./{tf}/sgn.dat", f"sgnsim_{i}") + w(f"./{tf}/kine.dat", f"sgnsim_{i}") + + # cp, and a subdirectory created while the workflow is running + stage(f"digi_{i}", "cp sgn.dat digi.dat; mkdir -p sub; " + "cp digi.dat sub/extra.dat", + [f"sgnsim_{i}"], f"./{tf}", i, "DIGI") + r(f"./{tf}/sgn.dat", f"digi_{i}") + w(f"./{tf}/digi.dat", f"digi_{i}") + r(f"./{tf}/digi.dat", f"digi_{i}") + w(f"./{tf}/sub/extra.dat", f"digi_{i}") + + # awk opens its output file itself, through fopen + stage(f"reco_{i}", "awk '{print > \"reco.dat\"}' digi.dat; " + "cat kine.dat >> reco.dat", + [f"digi_{i}"], f"./{tf}", i, "RECO") + r(f"./{tf}/digi.dat", f"reco_{i}") + r(f"./{tf}/kine.dat", f"reco_{i}") + w(f"./{tf}/reco.dat", f"reco_{i}") + + stage(f"aod_{i}", "cat reco.dat sub/extra.dat > AO2D.dat", + [f"reco_{i}"], f"./{tf}", i, "AOD") + r(f"./{tf}/reco.dat", f"aod_{i}") + r(f"./{tf}/sub/extra.dat", f"aod_{i}") + w(f"./{tf}/AO2D.dat", f"aod_{i}") + r(f"./tf{i}/AO2D.dat", "aodmerge") + + inputs = " ".join(f"tf{i}/AO2D.dat" for i in range(1, ntf + 1)) + stage("aodmerge", f"cat {inputs} > AO2D_merged.dat", + [f"aod_{i}" for i in range(1, ntf + 1)], "./", -1, "AOD") + w("./AO2D_merged.dat", "aodmerge") + + truth = build_report(written, read, [s["name"] for s in stages]) + return {"stages": stages}, truth + + +def run(argv, **kw): + print("+ " + " ".join(argv), flush=True) + return subprocess.run(argv, **kw) + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--workdir", default=None, + help="scratch directory (default: a fresh one under $TMPDIR)") + ap.add_argument("--backends", default=None, + help="comma-separated backends to test (default: all but " + f"the reference; known: {', '.join(sorted(filegraph.BACKENDS))})") + ap.add_argument("--reference", default="fanotify", + help="backend to compare the others against ('' to skip)") + ap.add_argument("--ntf", type=int, default=4) + ap.add_argument("--cpu-limit", type=int, default=4) + ap.add_argument("--sleep", default="0.4") + ap.add_argument("--python", default=sys.executable) + a = ap.parse_args(argv) + + runner = os.path.join(REPO, "MC", "workflow_runner", + "o2dpg_workflow_runner.py") + if not os.path.exists(runner): + print(f"no runner at {runner}", file=sys.stderr) + return 2 + + ref = a.reference.strip() + candidates = ([b.strip() for b in a.backends.split(",") if b.strip()] + if a.backends is not None + else [b for b in sorted(filegraph.BACKENDS) if b != ref]) + active = list(dict.fromkeys(([ref] if ref else []) + candidates)) + if not active: + print("nothing to run", file=sys.stderr) + return 2 + + workdir = a.workdir or os.path.join( + os.getenv("TMPDIR", "/tmp"), f"filegraph_equiv_{os.getpid()}") + os.makedirs(workdir, exist_ok=True) + print(f"workdir: {workdir}") + + wf, truth = build_workflow(a.ntf, a.sleep) + for name, doc in (("workflow.json", wf), ("truth.json", truth)): + with open(os.path.join(workdir, name), "w") as f: + json.dump(doc, f, indent=2) + print(f"{len(wf['stages'])} stages, {len(truth['file_report'])} files expected") + + t0 = time.time() + r = run([a.python, runner, "-f", "workflow.json", + "--cpu-limit", str(a.cpu_limit), "--mem-limit", "8000", + "--filegraph-backends", ",".join(active)], + cwd=workdir, env=dict(os.environ, O2DPG_ROOT=REPO)) + if r.returncode != 0: + print(f"runner failed with rc={r.returncode}", file=sys.stderr) + return 2 + print(f"workflow finished in {time.time() - t0:.1f}s") + + reports = {} + for b in active: + found = glob.glob(os.path.join(workdir, f"filegraph_{b}_*.json")) + if found: + with open(found[0]) as f: + reports[b] = json.load(f) + + failures = [] + + def grade(title, reference, candidate, backend, why): + print(f"\n{'=' * 72}\n== {title}\n{'=' * 72}") + text, summary = compare(reference, candidate, limit=10) + print(text) + print(f"OVERALL: {summary['verdict']}") + if summary["verdict"] == "UNSAFE": + failures.append((backend, why)) + return summary + + for backend in active: + if backend not in reports: + print(f"\nNO REPORT from {backend}", file=sys.stderr) + if backend != ref: + failures.append((backend, "no report")) + continue + + if backend == ref: + # fanotify resolves the process chain from /proc after the event, + # so on a fast machine it loses accesses by processes that exited + print(f"\n{'=' * 72}\n== {backend} (reference) vs truth\n{'=' * 72}") + text, summary = compare(truth, reports[backend], limit=10) + print(text) + print(f"OVERALL: {summary['verdict']}") + if summary["verdict"] == "UNSAFE": + print(f"NOTE: the reference {backend} itself misses edges") + continue + + grade(f"{backend} vs truth", truth, reports[backend], backend, + "misses edges the truth has") + if ref in reports: + summary = grade(f"{backend} vs {ref} (reference)", reports[ref], + reports[backend], backend, f"misses edges {ref} has") + with open(os.path.join(workdir, f"diff_{backend}_vs_{ref}.json"), "w") as f: + json.dump(summary, f, indent=2) + + print(f"\n{'=' * 72}") + for b, why in failures: + print(f"FAIL {b}: {why}") + if not failures: + print("PASS every backend is at least SAFE against truth" + + (f" and against {ref}" if ref else "")) + print(f"workdir kept at {workdir}") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/UTILS/FileIOGraph/tests/filegraph_test_support.py b/UTILS/FileIOGraph/tests/filegraph_test_support.py new file mode 100644 index 000000000..69ee796c8 --- /dev/null +++ b/UTILS/FileIOGraph/tests/filegraph_test_support.py @@ -0,0 +1,20 @@ +"""Put the FileIOGraph tools and the runner package on sys.path. + +Imported first by every test module here, so there is one bootstrap +instead of one per file. +""" +import os +import sys + +_HERE = os.path.dirname(os.path.abspath(__file__)) +FILEGRAPH_DIR = os.path.dirname(_HERE) +REPO = os.path.dirname(os.path.dirname(FILEGRAPH_DIR)) +RUNNER_BIN = os.path.join(REPO, "MC", "workflow_runner") + +for _p in (FILEGRAPH_DIR, RUNNER_BIN): + if _p not in sys.path: + sys.path.insert(0, _p) + +from o2dpg_runner import filegraph # noqa: E402 + +__all__ = ["FILEGRAPH_DIR", "REPO", "RUNNER_BIN", "filegraph"] diff --git a/UTILS/FileIOGraph/tests/test_analyse_strace.py b/UTILS/FileIOGraph/tests/test_analyse_strace.py new file mode 100644 index 000000000..66a105e5b --- /dev/null +++ b/UTILS/FileIOGraph/tests/test_analyse_strace.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +"""Offline tests for the strace analyser.""" +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +import tempfile +import unittest + +from filegraph_test_support import FILEGRAPH_DIR # noqa: F401 + +import analyse_FileIO_strace as A # noqa: E402 + +ALL = [re.compile(r".*")] + +# What `strace -f -qq -y -e status=successful` really writes. The last two +# lines of the open block are a call split around another process's entry. +TRACE = '''\ +101 openat(AT_FDCWD, "sgn.dat", O_WRONLY|O_CREAT|O_TRUNC, 0666) = 3 +101 openat(AT_FDCWD, "../bkg.dat", O_RDONLY) = 4 +102 openat(AT_FDCWD, "kine.dat", O_RDONLY|O_CLOEXEC) = 3 +102 openat(AT_FDCWD, "/usr/lib/libc.so.6", O_RDONLY|O_CLOEXEC) = 3 +102 openat(AT_FDCWD, ".", O_RDONLY|O_NONBLOCK|O_CLOEXEC|O_DIRECTORY) = 8 +101 openat(AT_FDCWD, "reco.dat", O_WRONLY|O_CREAT|O_APPEND +102 openat(AT_FDCWD, "digi.dat", O_RDONLY) = 5 +101 <... openat resumed>, 0666) = 6 +101 renameat2(AT_FDCWD, "tmp.dat", AT_FDCWD, "final.dat", 0) = 0 +''' + + +class TestParseTrace(unittest.TestCase): + @classmethod + def setUpClass(cls): + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "trace_0_reco_1.log") + with open(p, "w") as f: + f.write(TRACE) + cls.seen, cls.calls = A.parse_trace(p) + + def test_resolved_path_wins_over_the_argument(self): + # the argument said '../bkg.dat'; -y resolved it + self.assertIn(("/w/bkg.dat", "read"), self.seen) + + def test_write_flags_classify_as_write(self): + self.assertIn(("/w/tf1/sgn.dat", "write"), self.seen) + self.assertIn(("/w/tf1/kine.dat", "read"), self.seen) + + def test_split_call_keeps_its_flags(self): + # O_WRONLY appeared on the '' half only + self.assertIn(("/w/tf1/reco.dat", "write"), self.seen) + + def test_interleaved_call_is_not_swallowed(self): + self.assertIn(("/w/tf1/digi.dat", "read"), self.seen) + + def test_rename_destination_counts_as_written(self): + self.assertIn(("final.dat", "write"), self.seen) + + def test_directory_open_is_not_part_of_the_graph(self): + self.assertNotIn("/w/tf1", [p for p, _ in self.seen]) + + def test_repeated_accesses_collapse(self): + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "trace_0_t.log") + line = ('9 openat(AT_FDCWD, "a.dat", O_RDONLY) = 3\n') + with open(p, "w") as f: + f.write(line * 500) + seen, calls = A.parse_trace(p) + self.assertEqual(seen, {("/w/a.dat", "read")}) + self.assertEqual(calls, 500) + + +class TestCollect(unittest.TestCase): + def test_attribution_comes_from_the_file_name(self): + with tempfile.TemporaryDirectory() as d: + with open(os.path.join(d, "trace_3_reco_1.log"), "w") as f: + f.write(TRACE) + with open(os.path.join(d, "trace_4_aod_1.log"), "w") as f: + f.write('7 openat(AT_FDCWD, "reco.dat", O_RDONLY) ' + '= 3\n') + written, read, tasks, stats = A.collect(d, "/w", ALL) + + self.assertEqual(tasks, ["aod_1", "reco_1"]) + self.assertEqual(written["./tf1/sgn.dat"], {"reco_1"}) + self.assertEqual(read["./tf1/reco.dat"], {"aod_1"}) + # outside the working directory, so not part of the graph + self.assertNotIn("/usr/lib/libc.so.6", written) + self.assertNotIn("/usr/lib/libc.so.6", read) + + def test_non_trace_files_are_ignored(self): + with tempfile.TemporaryDirectory() as d: + with open(os.path.join(d, "notes.txt"), "w") as f: + f.write("nothing\n") + _, _, tasks, stats = A.collect(d, "/w", ALL) + self.assertEqual(tasks, []) + self.assertEqual(stats["traces"], 0) + + +class TestCli(unittest.TestCase): + def test_end_to_end(self): + with tempfile.TemporaryDirectory() as d: + td = os.path.join(d, "traces") + os.makedirs(td) + with open(os.path.join(td, "trace_1_sgnsim_1.log"), "w") as f: + f.write('9 openat(AT_FDCWD, "sgn.dat", O_WRONLY|O_CREAT, ' + '0666) = 3\n') + with open(os.path.join(td, "trace_2_digi_1.log"), "w") as f: + f.write('9 openat(AT_FDCWD, "sgn.dat", O_RDONLY) ' + '= 3\n') + out = os.path.join(d, "report.json") + subprocess.run( + [sys.executable, + os.path.join(FILEGRAPH_DIR, "analyse_FileIO_strace.py"), + "--straceDir", td, "--basedir", "/w", "-o", out], + check=True, stdout=subprocess.DEVNULL) + with open(out) as f: + doc = json.load(f) + + entry = doc["file_report"][0] + self.assertEqual(entry["file"], "./tf1/sgn.dat") + self.assertEqual(entry["written_by"], ["sgnsim_1"]) + self.assertEqual(entry["read_by"], ["digi_1"]) + tpl = doc["file_template_report"][0] + self.assertEqual(tpl["file"], "./tfX/sgn.dat") + self.assertEqual(tpl["written_by"], ["sgnsim_X"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/UTILS/FileIOGraph/tests/test_filegraph.py b/UTILS/FileIOGraph/tests/test_filegraph.py new file mode 100644 index 000000000..960c452f9 --- /dev/null +++ b/UTILS/FileIOGraph/tests/test_filegraph.py @@ -0,0 +1,237 @@ +#!/usr/bin/env python3 +"""Offline tests for the report machinery, the comparison and the backends. + +Plain unittest: this has to run on the bare interpreter of a GRID worker +or a CVMFS O2 environment, neither of which has pytest. + + python3 -m unittest discover -s UTILS/FileIOGraph/tests -t UTILS/FileIOGraph/tests +""" +from __future__ import annotations + +import json +import logging +import os +import re +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +from filegraph_test_support import FILEGRAPH_DIR, filegraph + +import equivalence_test # noqa: E402 +from compare_reports import compare # noqa: E402 +from filegraph_report import ( # noqa: E402 + basedir_prefix, build_report, keep_file, relative_to_basedir, +) + +ALL = [re.compile(r".*")] +LOG = logging.getLogger("test") + + +class TestPathHelpers(unittest.TestCase): + def test_relative_to_basedir(self): + for base in ("/w", "/w/"): + prefix = basedir_prefix(base) + self.assertEqual(relative_to_basedir("/w/tf1/a.root", prefix), + "./tf1/a.root") + self.assertIsNone(relative_to_basedir("/usr/lib/libc.so", prefix)) + + def test_logs_and_dpl_config_are_noise(self): + self.assertFalse(keep_file("./tf1/sgnsim_1.log", ALL)) + self.assertFalse(keep_file("./tf1/sgnsim_1.log_done", ALL)) + self.assertFalse(keep_file("./dpl-config.json", ALL)) + self.assertTrue(keep_file("./tf1/AO2D.root", ALL)) + + def test_filters_apply(self): + self.assertTrue(keep_file("./tf1/a.root", [re.compile(r".*\.root")])) + self.assertFalse(keep_file("./tf1/a.dat", [re.compile(r".*\.root")])) + + +class TestBuildReport(unittest.TestCase): + def setUp(self): + self.doc = build_report( + {"./tf1/a.root": {"digi_1"}, "./tf2/a.root": {"digi_2"}, + "./AO2D.root": {"aodmerge"}}, + {"./tf1/a.root": {"reco_1", "aodmerge"}, + "./tf2/a.root": {"reco_2", "aodmerge"}}, + ["digi_1", "digi_2", "reco_1", "reco_2", "aodmerge"]) + self.tpl = {e["file"]: e for e in self.doc["file_template_report"]} + + def test_timeframe_tasks_become_templates(self): + self.assertEqual(self.tpl["./tfX/a.root"]["written_by"], ["digi_X"]) + self.assertEqual(self.tpl["./tfX/a.root"]["source_timeframes"], [1, 2]) + + def test_global_reader_is_not_templated(self): + # a naive "strip the trailing _N" would corrupt aodmerge + self.assertIn("aodmerge", self.tpl["./tfX/a.root"]["read_by"]) + self.assertIn("reco_X", self.tpl["./tfX/a.root"]["read_by"]) + + def test_non_timeframe_file_absent_from_templates(self): + self.assertNotIn("./AO2D.root", self.tpl) + + def test_task_report_covers_every_task(self): + self.assertEqual([t["task"] for t in self.doc["task_report"]], + ["aodmerge", "digi_1", "digi_2", "reco_1", "reco_2"]) + + +class TestCompare(unittest.TestCase): + def _verdict(self, cand_written, cand_read): + ref = build_report({"./tf1/a.root": {"digi_1"}}, + {"./tf1/a.root": {"reco_1"}}, ["digi_1", "reco_1"]) + cand = build_report(cand_written, cand_read, ["digi_1", "reco_1"]) + return compare(ref, cand)[1] + + def test_identical_is_exact(self): + s = self._verdict({"./tf1/a.root": {"digi_1"}}, {"./tf1/a.root": {"reco_1"}}) + self.assertEqual(s["verdict"], "EXACT") + + def test_extra_reader_is_safe(self): + s = self._verdict({"./tf1/a.root": {"digi_1"}}, + {"./tf1/a.root": {"reco_1", "qc_1"}}) + self.assertEqual(s["verdict"], "SAFE") + self.assertEqual(s["sections"]["file_report"]["missing_edges"], 0) + + def test_missing_reader_is_unsafe(self): + s = self._verdict({"./tf1/a.root": {"digi_1"}}, {}) + self.assertEqual(s["verdict"], "UNSAFE") + self.assertEqual(s["sections"]["file_report"]["missing_edges"], 1) + self.assertAlmostEqual(s["sections"]["file_report"]["recall"], 0.5) + + def test_missing_file_is_unsafe(self): + self.assertEqual(self._verdict({}, {})["verdict"], "UNSAFE") + + +class TestAnalyseFanotifyLog(unittest.TestCase): + """End-to-end on the fanotify analyser, so the shared module stays honest.""" + + ACTION = ( + "2026-05-06 16:49:18,267 INFO Task pid=101 tid=0 bkg finished rc=0\n" + "2026-05-06 16:49:19,267 INFO Task pid=102 tid=1 sgnsim_1 finished rc=0\n" + "2026-05-06 16:49:20,267 INFO Task pid=103 tid=2 reco_1 finished rc=0\n" + ) + + def test_report(self): + with tempfile.TemporaryDirectory() as d: + action = os.path.join(d, "action.log") + monitor = os.path.join(d, "monitor.log") + out = os.path.join(d, "report.json") + with open(action, "w") as f: + f.write(self.ACTION) + with open(monitor, "w") as f: + f.write(f'"{d}/bkg.dat",write,101\n' + f'"{d}/tf1/sgn.dat",write,201;102\n' + f'"{d}/bkg.dat",read,201;102\n' + f'"{d}/tf1/sgn.dat",read,103\n' + f'"{d}/tf1/reco_1.log",write,103\n' + f'"/usr/lib/libc.so.6",read,103\n') + subprocess.run( + [sys.executable, os.path.join(FILEGRAPH_DIR, "analyse_FileIO_v2.py"), + "--actionFile", action, "--monitorFile", monitor, + "--basedir", d, "-o", out], + check=True, stdout=subprocess.DEVNULL) + with open(out) as f: + doc = json.load(f) + + got = {e["file"]: (e["written_by"], e["read_by"]) for e in doc["file_report"]} + self.assertEqual(got["./bkg.dat"], (["bkg"], ["sgnsim_1"])) + # a grandchild's access is attributed through the parent chain + self.assertEqual(got["./tf1/sgn.dat"], (["sgnsim_1"], ["reco_1"])) + self.assertNotIn("./tf1/reco_1.log", got) + self.assertNotIn("/usr/lib/libc.so.6", got) + + +class StubBackend(filegraph.FileGraphBackend): + name = "stub" + + def wrap(self, argv, taskname, tid): + return ["stub", taskname, str(tid), "--"] + argv + + +def manager(spec, action_log="action.log"): + return filegraph.FileGraphManager.from_config(spec, "/w", 4242, action_log, LOG) + + +class TestBackendSelection(unittest.TestCase): + def setUp(self): + patch = mock.patch.dict(os.environ, {}, clear=False) + patch.start() + self.addCleanup(patch.stop) + for k in ("O2DPG_PRODUCE_FILEGRAPH", "O2DPG_FILEGRAPH_BACKENDS"): + os.environ.pop(k, None) + filegraph.BACKENDS["stub"] = StubBackend + self.addCleanup(filegraph.BACKENDS.pop, "stub", None) + + def test_nothing_requested_means_no_backend(self): + self.assertEqual(manager("").backends, []) + + def test_named_backends_are_built_in_order(self): + self.assertEqual([b.name for b in manager("stub,fanotify").backends], + ["stub", "fanotify"]) + + def test_legacy_variable_selects_fanotify_with_that_exe(self): + os.environ["O2DPG_PRODUCE_FILEGRAPH"] = "/opt/mon.exe" + m = manager("") + self.assertEqual([b.name for b in m.backends], ["fanotify"]) + self.assertEqual(m.backends[0].exe, "/opt/mon.exe") + + def test_unknown_backend_is_skipped_not_fatal(self): + self.assertEqual([b.name for b in manager("stub,nonsense").backends], ["stub"]) + + def test_the_action_log_reaches_the_backend_that_needs_it(self): + b = manager("fanotify", action_log="pipeline_action_7.log").backends[0] + self.assertIn("pipeline_action_7.log", b.analyser_args()) + + def test_only_fanotify_writes_the_legacy_report_name(self): + self.assertTrue(filegraph.FanotifyBackend.legacy_report) + self.assertFalse(StubBackend.legacy_report) + + +class TestBackendWrapping(unittest.TestCase): + ARGV = ["/bin/bash", "-c", "echo hi"] + + def setUp(self): + filegraph.BACKENDS["stub"] = StubBackend + self.addCleanup(filegraph.BACKENDS.pop, "stub", None) + + def test_a_backend_that_does_not_wrap_leaves_the_command_alone(self): + self.assertEqual(manager("fanotify").wrap(list(self.ARGV), "sgnsim_1", 3), + self.ARGV) + + def test_a_wrapping_backend_keeps_the_command_as_the_tail(self): + argv = manager("stub").wrap(list(self.ARGV), "sgnsim_1", 3) + self.assertEqual(argv[:4], ["stub", "sgnsim_1", "3", "--"]) + self.assertEqual(argv[4:], self.ARGV) + + def test_an_empty_manager_is_a_no_op(self): + m = manager("") + self.assertEqual(m.wrap(list(self.ARGV), "sgnsim_1", 3), self.ARGV) + m.start() + m.stop() + self.assertEqual(m.analyse(), {}) + + +class TestFilegraphDir(unittest.TestCase): + def test_resolves_without_o2dpg_root(self): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("O2DPG_ROOT", None) + d = filegraph.filegraph_dir() + self.assertTrue(os.path.exists(os.path.join(d, "analyse_FileIO_v2.py")), d) + + +class TestSyntheticWorkflow(unittest.TestCase): + def test_the_truth_matches_the_commands(self): + wf, truth = equivalence_test.build_workflow(2, "0") + files = {e["file"]: e for e in truth["file_report"]} + self.assertEqual(files["./tf1/AO2D.dat"]["read_by"], ["aodmerge"]) + # a run-time subdirectory is part of the expected graph + self.assertEqual(files["./tf1/sub/extra.dat"]["written_by"], ["digi_1"]) + for s in wf["stages"]: + self.assertIn("cwd", s) + self.assertIn("resources", s) + self.assertIn("timeframe", s) + + +if __name__ == "__main__": + unittest.main()