import importlib.util import importlib.machinery import os import subprocess import tempfile import time import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] def load_plugin(name): path = ROOT / "netdata" / "plugins" / f"{name}.plugin" loader = importlib.machinery.SourceFileLoader(name, str(path)) spec = importlib.util.spec_from_loader(name, loader) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def write(path, text): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text) class CollectorTests(unittest.TestCase): def test_mem_trajectory_eta_uses_recent_falling_runway(self): mod = load_plugin("mem_trajectory") with tempfile.TemporaryDirectory() as td: proc = Path(td) write(proc / "meminfo", "\n".join([ "MemTotal: 1000000 kB", "MemAvailable: 180000 kB", "SwapTotal: 1000000 kB", "SwapFree: 180000 kB", ])) write(proc / "pressure" / "memory", "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=3.50 avg60=0.00 avg300=0.00 total=0\n") history = [(0.0, 600000), (60.0, 360000)] metrics = mod.collect(proc, history, now=90.0) self.assertEqual(metrics["runway_pct"], 18.0) self.assertEqual(metrics["eta_seconds"], 75) self.assertEqual(metrics["psi_full_avg10"], 3.5) def test_mem_trajectory_plateau_emits_eta_sentinel(self): mod = load_plugin("mem_trajectory") with tempfile.TemporaryDirectory() as td: proc = Path(td) write(proc / "meminfo", "\n".join([ "MemTotal: 1000000 kB", "MemAvailable: 500000 kB", "SwapTotal: 1000000 kB", "SwapFree: 500000 kB", ])) history = [(0.0, 1000000), (60.0, 1000000)] self.assertEqual(mod.collect(proc, history, now=90.0)["eta_seconds"], 999999) def test_tmpfs_guard_counts_used_inodes_and_stale_owned_files(self): mod = load_plugin("tmpfs_guard") with tempfile.TemporaryDirectory() as td: tmp = Path(td) / "tmp" tmp.mkdir() stale = tmp / "old.bin" stale.write_bytes(b"x" * 1024 * 1024) old = time.time() - 80 * 3600 os.utime(stale, (old, old)) fresh = tmp / "fresh.bin" fresh.write_bytes(b"x" * 128) metrics = mod.collect(tmp, stale_hours=72, uid=os.getuid()) self.assertGreaterEqual(metrics["stale_junk_mb"], 1) self.assertGreaterEqual(metrics["used_pct"], 0) self.assertGreaterEqual(metrics["inodes_pct"], 0) def test_cpu_runaway_excludes_build_groups_and_counts_nonbuild_hog(self): mod = load_plugin("cpu_runaway") with tempfile.TemporaryDirectory() as td: proc = Path(td) write(proc / "stat", "cpu 1000 0 0 1000 0 0 0 0 0 0\n") for pid, exe, cmd, cgroup, ticks in [ (101, "gcc", "gcc -c file.c", "0::/user.slice/builds.slice/x", 500), (202, "python3", "python3 loop.py", "0::/user.slice/app.slice/y", 600), ]: p = proc / str(pid) fields = ["S"] + ["0"] * 50 fields[11] = str(ticks) fields[12] = "0" write(p / "stat", f"{pid} ({exe}) " + " ".join(fields) + "\n") write(p / "cmdline", cmd.replace(" ", "\0") + "\0") write(p / "cgroup", cgroup + "\n") exe_path = p / "exe" exe_path.symlink_to(f"/usr/bin/{exe}") previous = {"total": 1000, "groups": {"gcc": 100, "python3": 100}} metrics, _state = mod.collect(proc, previous, interval=10.0, build_allowlist=["gcc"]) self.assertGreater(metrics["build_busy_pct"], 0) self.assertGreater(metrics["nonbuild_busy_pct"], 0) self.assertGreater(metrics["nonbuild_busy_pct"], metrics["build_busy_pct"]) def test_proc_fd_reports_file_process_and_inotify_pressure(self): mod = load_plugin("proc_fd") with tempfile.TemporaryDirectory() as td: proc = Path(td) write(proc / "sys" / "fs" / "file-nr", "80\t0\t100\n") write(proc / "sys" / "fs" / "inotify" / "max_user_watches", "200\n") write(proc / "1" / "fd" / "0", "") write(proc / "2" / "fd" / "0", "") metrics = mod.collect(proc, inotify_watches=50) self.assertEqual(metrics["open_fds_pct"], 80.0) self.assertEqual(metrics["proc_count"], 2) self.assertEqual(metrics["inotify_watch_pct"], 25.0) def test_disk_guard_reports_block_inode_and_reserved_capacity(self): mod = load_plugin("disk_guard") class Stat: f_blocks = 100 f_bfree = 20 f_bavail = 10 f_frsize = 1024**3 f_files = 1000 f_ffree = 750 metrics = mod.collect(Path("/"), statvfs=lambda _path: Stat()) self.assertEqual(metrics["used_pct"], 80.0) self.assertEqual(metrics["avail_gib"], 10.0) self.assertEqual(metrics["reserved_gib"], 10.0) self.assertEqual(metrics["inodes_pct"], 25.0) def test_run_build_rejects_low_disk_before_systemd_run(self): with tempfile.TemporaryDirectory() as td: bin_dir = Path(td) write( bin_dir / "df", "#!/bin/sh\nprintf 'Filesystem 1024-blocks Used Available Capacity Mounted on\\n/dev/test 104857600 99614720 5242880 95%% /\\n'\n", ) write(bin_dir / "systemd-run", "#!/bin/sh\nprintf called > \"$SM_CALLED\"\n") for path in bin_dir.iterdir(): path.chmod(0o755) called = bin_dir / "called" env = os.environ | { "PATH": f"{bin_dir}:{os.environ['PATH']}", "SM_CALLED": str(called), "SM_DISK_MIN_FREE_GIB": "10", "SM_DISK_MAX_USED_PCT": "85", } proc = subprocess.run( [str(ROOT / "slices" / "bin" / "run-build"), "true"], env=env, text=True, capture_output=True, check=False, ) self.assertEqual(proc.returncode, 75) self.assertIn("disk admission rejected", proc.stderr) self.assertFalse(called.exists()) def test_run_build_rejects_when_disk_probe_fails(self): with tempfile.TemporaryDirectory() as td: bin_dir = Path(td) write(bin_dir / "df", "#!/bin/sh\nexit 1\n") write(bin_dir / "systemd-run", "#!/bin/sh\nprintf called > \"$SM_CALLED\"\n") for path in bin_dir.iterdir(): path.chmod(0o755) called = bin_dir / "called" proc = subprocess.run( [str(ROOT / "slices" / "bin" / "run-build"), "true"], env=os.environ | {"PATH": f"{bin_dir}:{os.environ['PATH']}", "SM_CALLED": str(called)}, text=True, capture_output=True, check=False, ) self.assertEqual(proc.returncode, 75) self.assertIn("disk probe failed", proc.stderr) self.assertFalse(called.exists()) def test_disk_check_notifies_once_while_tier_is_unchanged(self): with tempfile.TemporaryDirectory() as td: root = Path(td) bin_dir = root / "bin" bin_dir.mkdir() write( bin_dir / "df", "#!/bin/sh\nprintf 'Filesystem 1024-blocks Used Available Capacity Mounted on\\n/dev/test 104857600 99614720 5242880 95%% /\\n'\n", ) write(bin_dir / "notify-send", "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$SM_NOTIFY_LOG\"\n") write(bin_dir / "logger", "#!/bin/sh\nexit 0\n") for path in bin_dir.iterdir(): path.chmod(0o755) notify_log = root / "notifications" env = os.environ | { "PATH": f"{bin_dir}:{os.environ['PATH']}", "XDG_STATE_HOME": str(root / "state"), "SM_NOTIFY_LOG": str(notify_log), } command = [str(ROOT / "slices" / "bin" / "disk-check")] subprocess.run(command, env=env, check=True, capture_output=True, text=True) subprocess.run(command, env=env, check=True, capture_output=True, text=True) self.assertEqual(len(notify_log.read_text().splitlines()), 1) if __name__ == "__main__": unittest.main()