diff --git a/docs/design/DECISIONS_LOG.md b/docs/design/DECISIONS_LOG.md new file mode 100644 index 00000000..69451ee7 --- /dev/null +++ b/docs/design/DECISIONS_LOG.md @@ -0,0 +1,120 @@ +# Decision log — collector / measurement POC + +Records every non-trivial design or implementation decision, each with a +one-line rationale. **Bold** marks decisions I was unsure about and that are +worth revisiting. + +See `POC_BRIEF.md` for the goal and `resource-collectors.md` for the design. + +## Environment / setup + +- Container venv lives at `.venv` (created with `uv`, not the host's + `~/.venvs/duct`); installed `-e .[all]` plus `tox`, `pytest`, and `psutil` + so both the stdlib-only and optional-psutil paths can be exercised. Rationale: + the container ships `uv` only (no pip/tox/venv preinstalled). +- Tests live under `test/` (singular), matching the existing repo layout — the + brief/CLAUDE.md say `tests/`, but I follow the repo's actual convention. + Rationale: behavior-preserving, don't fork the test tree. + +## Decisions + + + +- **Scope = data collection e2e (incl. the end-of-run summary).** The + collector/measurement pipeline replaces the old sampling + per-sample-max + aggregation, emits new renamed measurement keys per report, AND writes the + end-of-run `execution_summary` from those same measurements (its keys update + to the new names). The other log consumers — `plot.py` / `ls.py` / + `pprint.py` — are explicitly NOT touched this round (user decision). +- Replace, don't reproduce: old fields become new namespaced keys carrying the + same data (e.g. `rss` → `ps_rss`, `total_rss` → `ps_rss_total`); there is no + legacy/back-compat field block (user decision — "only new renamed fields"). +- Per-report usage record uses a single `measurements` object keyed by + measurement name: per_pid keys hold a `{pid: value}` map, single keys hold a + scalar. (Design's "measurements are keys" framing; user deferred to judgment.) +- CLI: add a `--measurements` flag (default: all available keys). Requesting a + `psutil_*` key without psutil installed is a clean error. (user decision) +- Cgroup here is v2 (`/sys/fs/cgroup/memory.peak`, readable); v1 + `memory.max_usage_in_bytes` path is code-only/untested in this container. + +### Implementation architecture (after reading the code) + +- Module split: `_collectors.py` (pure I/O → raw readings: `Scope`/`Reduce`/ + `Derive` enums, `Measurement`, the ps/cgroup/psutil collectors, key registry) + and `_aggregation.py` (buffer raw readings, aggregate-once: derive → collapse + → reduce, with per-key seeds carried across report boundaries + a run-level + summary accumulator). Keeps pure collection separate from stateful reduction. +- The brief's ps keys are exactly `ps_rss`, `ps_rss_total`, `ps_pdcpu`, + `ps_cpu_seconds` — so the new model **drops `pmem`, `vsz`, and lifetime + `pcpu`** entirely (not in the design's key list). The ps `-o` columns shrink + to `pid,rss,cputime,cmd`. `cputime` is parsed by reusing + `etime_to_etimes` (same `[[DD-]HH:]MM:SS` grammar). +- **Added a non-brief `ps_cmd` key** (per_pid, reduce=last) to keep per-pid + command labels in the record — needed for plot labels / `ls` / the e2e + child-counting test, and it exercises a string-valued per_pid measurement. + Mild stretch of the "measurement = numeric" framing; revisit. +- Per-report usage record is replaced by `{timestamp, num_samples, + measurements:{key: scalar | {pid: value}}}`; old `processes`/`totals`/ + `averages` blocks are gone. Records carry only selected+available keys. +- `execution_summary` is rewritten from the run-level accumulator and **always + emits the full key set** (optional cgroup/psutil keys present as `null` when + unavailable) so info.json schema is environment-independent and `test_schema` + stays deterministic. Summary keys: meta (unchanged) + `peak_ps_rss_total`, + `ave_ps_rss_total`, `ps_cpu_seconds`, plus `peak_cgroup_rss_peak`, + `peak_psutil_pss_total` (null when absent). +- Renaming summary keys forces updating `ls.py`'s `LS_FIELD_CHOICES` registry + and bumping `__schema_version__` — this is the schema test's own sanctioned + workflow ("Fails when schema changes — bump schema version"), treated as a + field-registry update, NOT as reworking `ls`/`plot`/`pprint` behavior (still + out of scope; they keep reading their existing test fixtures, so their tests + stay green; live plot/ls on new-format logs is explicit follow-up). +- **cgroup scope = duct's own cgroup peak** (resolved from `/proc/self/cgroup`; + v2 `memory.peak`, v1 `memory.max_usage_in_bytes`), since under SLURM duct runs + inside the job cgroup. Reader-mode only; refuse cleanly if unreadable. +- **Summary keys fully replaced** (not additive): old `peak_rss`/`peak_vsz`/ + `peak_pmem`/`peak_pcpu`/`average_*` are gone, replaced by `peak_ps_rss_total`, + `ave_ps_rss_total`, `ps_cpu_seconds`, `peak_cgroup_rss_peak`, + `peak_psutil_pss_total` (meta keys unchanged). This honors "only new renamed + fields" for the summary too. The rename's unavoidable cascade: `ls.py`'s + `LS_FIELD_CHOICES` registry, the `ls --fields` default in `cli.py` (`peak_rss` + → `peak_ps_rss_total`, else argparse rejects its own default), and + `EXECUTION_SUMMARY_FORMAT`. Treated as the schema test's sanctioned + "bump schema version" workflow — `__schema_version__` 0.2.2 → 0.3.0 — NOT as + reworking ls/plot/pprint *behavior* (their logic + fixtures stay untouched and + green; `pprint`'s humanize map simply won't pretty-print the new keys, which + is fine for the POC). `ps_pdcpu` rate is generic Δfield/Δt = **cputime-seconds + per wall second (≈cores, 1.0 = one core), NOT percent** like the old `pcpu`; + kept generic so future io rates fit the same `derive=rate`. Revisit. +- `ps` is the liveness signal: when it returns no pids the whole sample is + dropped (not buffered, not counted) so a trailing empty reading can't skew a + delta/total. `add_sample()` returns whether a sample was recorded; reporting + is no longer gated on a nonzero ps pid count, so a non-ps-only selection + (e.g. just `cgroup_rss_peak`) still emits records. +- **`ps_cpu_seconds` delta clamps to 0 on a negative window** (a pid with + accrued cputime exits, so the summed cumulative total drops below the seed). + The lost counter can't be attributed without per-pid delta tracking; clamping + under-counts that window. Acceptable for the POC; a per-pid delta sum would + fix it. Revisit. +- Collectors expose `collect()` (per-sample) / `read()` (per-report) with no + `ts` arg — the aggregator stamps `time.monotonic()` (for rate Δt) and a wall + isoformat (for the record) centrally, keeping collectors pure. +- The first foundation commit was reformatted by pre-commit and aborted, so the + collector+aggregation modules landed inside the "wire the engine" commit + rather than as a separate commit. Harmless; history is clean. + +### Out of scope — interface fit (per brief) + +- **`io` collector:** fits cleanly. It is just another per-sample/per-pid + collector whose counters use `derive=rate` exactly like cputime + (`io_read_rate` = `rate(read_bytes)`), and node `iowait` is a single-read like + cgroup. No new machinery needed. +- **`/proc` sub-second CPU:** fits — same `derive=rate` measurement, different + collector reading `/proc//stat`. Only the source changes; the reducer is + identical. (Confirms the rate derive belongs to the measurement, not ps.) +- **Named key-groups / presets:** easy to add over the existing registry as an + alias layer expanding to keys before `resolve_selection`; deliberately not + built. +- **Differential (same metric, two collectors):** the model already supports it + — `ps_pdcpu` and `psutil_pdcpu` coexist and are recorded side by side, so + comparing two methods of the same quantity is just selecting both keys. The + awkward part is purely presentation (plot/ls), which is out of scope. diff --git a/docs/design/POC_BRIEF.md b/docs/design/POC_BRIEF.md new file mode 100644 index 00000000..e91a044e --- /dev/null +++ b/docs/design/POC_BRIEF.md @@ -0,0 +1,31 @@ +# POC implementation brief — collector / measurement resource stats + +This worktree is a throwaway **proof of concept**. Goal: make the design in `docs/design/resource-collectors.md` concrete enough to evaluate and to anchor a PR (the design doc and this implementation go up together). It does **not** need to be the final shape — it needs to work, be reasonably clean, and surface the real problems the design sketch can't. + +## What to build (one vertical slice that exercises the whole interface) + +Implement the collect → aggregate model from `resource-collectors.md`: + +- **Pure collect, aggregate once per report.** Each sample, collectors append raw readings to a buffer (collect does no derivation and holds no cross-sample state). At each report interval, aggregate once: **derive** (e.g. pdcpu from cputime) → **collapse** per-pid readings into single values → **reduce** over the interval (a level reduces by `max`; a counter by `delta`/`last`). Keep the previous window's last reading per key as the **seed** for deltas that cross a report boundary. +- **Measurements are keys.** A measurement = `(name, scope, derive?, reduce)` where `scope` is `per_pid` or `single`. The user selects which measurement keys to record (default: all available keys). Collectors are internal; a collector batches the I/O for all of its selected keys (one `ps` call serves every `ps_*` key). +- **Collectors:** + - **`ps`** (always available, per-pid). Keys: `ps_rss` (per_pid, max), `ps_rss_total` (single, max), `ps_pdcpu` (per_pid, derive rate from `cputime`, max), `ps_cpu_seconds` (single, delta of `cputime`). Add `cputime` to the `ps -o` columns (a one-line change). + - **`cgroup`** (available iff a memory cgroup is present; single value; read once per report). Key: `cgroup_rss_peak` (single, last) from `memory.max_usage_in_bytes` (cgroup v1) / `memory.peak` (v2). Reader-mode only — no sudo/setuid/privilege probes; refuse cleanly if the file isn't readable; never create a cgroup. + - **`psutil`** (optional, available iff `psutil` is importable; per-pid). Keys: `psutil_pss` (per_pid, max; Linux), `psutil_pss_total` (single, max), `psutil_pdcpu` (per_pid, rate from psutil's raw `cpu_times()` — **not** `cpu_percent()`, so the collector stays pure). psutil must be an **optional** dependency (`options.extras_require` "all"), never required for `con-duct run`; a clean error if a `psutil_*` key is requested without psutil installed. +- **Behavior-preserving where possible:** existing ps-based numbers and the existing tests should stay green; cputime/pdcpu/cgroup/psutil are additive. + +## Constraints (follow the repo `CLAUDE.md`) + +- `con-duct run` stays standard-library-only by default; psutil only via `extras_require`. +- Strict typing (`from __future__ import annotations`), `pathlib` over `os.path`, tests under `tests/`, run `tox` (lint, typing, py3) before declaring done. +- **Namespacing:** never overload an existing field name with new semantics; each new measurement is a distinct key. + +## Process requirements (IMPORTANT) + +- **Keep a decision log** in a git-tracked file: `docs/design/DECISIONS_LOG.md`. Record every non-trivial design or implementation decision, each with a one-line rationale. +- **Bold any decision you were unsure about** (`**like this**`) in that log, so they are easy to find and revisit. +- Commit as you go with clear messages. + +## Out of scope (note in the log if relevant, do NOT build) + +- The `io` collector, named key-groups/presets, a `/proc` sub-second CPU source, and running multiple collectors of the same metric for differential comparison. If the interface makes any of these obviously easy or awkward, note it in the decision log — but don't implement them here. diff --git a/docs/design/resource-collectors.md b/docs/design/resource-collectors.md new file mode 100644 index 00000000..04ab1788 --- /dev/null +++ b/docs/design/resource-collectors.md @@ -0,0 +1,140 @@ +# Resource statistics: correctness problems and a proposed collector/measurement model + +## Problems with the current resource statistics + +duct's per-process resource statistics (collected via `ps`) have several correctness problems: + +1. **CPU is a lifetime average, not a rate.** `ps -o pcpu` reports cumulative CPU% over the whole process lifetime, so summing it across a process tree overshoots the core ceiling (con/duct#399 shows 5363% on a 20-core machine) and cannot show *when* CPU was actually spent. The plot-time correction in con/duct#424 reconstructs an instantaneous rate, but only approximately: it works from the already-`max`-aggregated `pcpu`, so it is an upper bound whose error grows with run length. + +2. **Memory is summed, double-counting shared pages.** `total_rss` is the sum of per-process `rss`, so shared pages (shared libraries, etc.) are counted once per process, overstating real memory use. + +3. **Aggregation happens per sample, discarding intra-interval detail.** Each sample is immediately folded into a running `max` and dropped. Because the individual samples are gone, a per-interval CPU rate cannot be recovered after the fact, and short spikes between report points are lost. + +4. **Every field is reduced by `max`.** That is correct for an instantaneous level (memory, %CPU) but wrong for a cumulative counter (e.g. CPU-seconds), where the meaningful quantity is a difference between readings. + +5. **`pcpu` is not portable.** On Linux it is a lifetime average; on macOS it is a decayed moving average. The same field means different things on different platforms. + +## Proposal + +Two connected changes. + +### 1. Aggregate once per report, over kept readings + +Today each sample is reduced immediately into a running aggregate, then dropped. Instead, keep the raw readings for a report interval and aggregate them once, at report time. + +This is what makes a per-sample instantaneous CPU rate ("pdcpu") possible at all: `pdcpu = Δ(CPU time) / Δ(wall time)` between consecutive readings, which needs the individual readings that per-sample `max` currently discards. Each reported value is then the **max** of those per-sample rates over the interval, so a short CPU spike between report points is preserved rather than averaged away. + +It depends on collecting **cumulative CPU time** instead of `pcpu`. `cputime` is a one-line addition to the `ps` columns duct already requests; unlike `pcpu`, it is the same cumulative quantity on Linux and macOS, so the rate is identical on both. (Deriving the rate from it was prototyped in con/duct#423.) + +There is a resolution tradeoff, and it is the user's to make. `ps` cputime has whole-second resolution, so the shorter the interval a rate is computed over — the closer the sample interval gets to ~1s — the more a reported "spike" can be a quantization artifact rather than a real one; longer intervals are smoother but can hide real spikes. duct exposes the knob (sample interval) and can offer finer-resolution sources (e.g. a `/proc` reader on Linux) so the user chooses where to sit on the sensitivity/noise tradeoff. + +### 2. A collector / measurement interface + +Define two roles: + +- A **collector** does one I/O pass and produces a set of **measurements** — e.g. `ps` produces `rss`, `cputime`, …; a cgroup collector reads the kernel's peak-RAM counter. +- A **measurement** is a named, namespaced value that declares its scope (per-process or total) and how it reduces over the interval — an instantaneous level reduces by `max`; a cumulative counter is differenced, either as a per-interval total or as a per-sample rate whose peak (`max`) reveals spikes. + +The pipeline becomes: + +``` +per sample: collect (pure) → append raw readings to a buffer +per report: aggregate = derive (e.g. rate from cputime) → combine across processes → reduce over the interval → write +``` + +This makes new data sources modular and composable instead of special-cased. The motivating example is cgroup `memory.max_usage_in_bytes`: the kernel's high-water peak RAM, which cannot miss a between-sample spike, is job-scoped under SLURM, and is the number an HPC `--mem` request needs. As a collector/measurement it is just another entry; without the interface, a new total has to mutate the shared sample object — which is what made an earlier cgroup prototype (con/duct#415) awkward. Namespacing measurements (a cgroup peak is a distinct field, never overwriting the ps-summed total) keeps different sources or methods from silently colliding under one field name. + +Users select measurements by key (e.g. `ps_rss`, `cgroup_rss_peak`); the collector behind each key is internal, and a collector batches the I/O for all of its selected keys, so selecting several keys from one source is still a single pass. Named groups of keys are possible later but are not required. + +The same interface should absorb other collectors without special-casing. Sketching several is how we check the shape holds: + +
+How collectors map onto the interface — ps, cgroup, psutil, /proc, io + +``` +Collector: available() -> bool; measurements; collect() -> reading; read() -> reading # per-sample collect(); per-report read() +Measurement = (name, collector, field, scope, reduce, derive?) + scope = per_pid | single # single = one value; the collector decides how (sum of its pids, a kernel read, a system read) + derive = rate # counter -> per-sample rate Δfield/Δt; optional + reduce = max | delta | last # per-interval collapse (mean sketched, not built in the POC) + +# ps — per-pid, always available (the baseline) +collect: one `ps` call -> reading per pid {rss, cputime} + ps_rss per_pid reduce=max + ps_rss_total single reduce=max # the ps collector sums its pids per tick, then max + ps_pdcpu per_pid derive=rate(cputime) reduce=max + ps_cpu_seconds single reduce=delta(cputime) + +# cgroup — totals only, read once per report (kernel high-water mark) +available: memory cgroup present +report_read: memory.max_usage_in_bytes -> reading {mem_peak} + cgroup_rss_peak single reduce=last # kernel already took the peak; no per-pid pass + +# psutil — optional per-pid (PSS + finer cputime), only if installed +available: `import psutil` works +collect: iterate procs (getsid filter) -> reading per pid {pss, uss, cputime} + psutil_pss per_pid reduce=max # Linux; shared pages split across sharers + psutil_pss_total single reduce=max # sum = exact session footprint + psutil_pdcpu per_pid derive=rate(cputime) reduce=max # sub-second, cross-platform + +# /proc — Linux, per-pid, sub-second, stdlib +available: Linux + /proc +collect: read /proc//stat for session pids -> reading per pid {cputime, ...} + proc_pdcpu per_pid derive=rate(cputime) reduce=max # same derive, finer source + +# io — NEW / speculative: does it fit, and what does it reveal? +collect: /proc//io + /proc//stat per pid -> reading per pid {read_bytes, write_bytes, blkio_ticks} + io_read_rate per_pid derive=rate(read_bytes) reduce=max # counter -> rate, exactly like cputime + io_write_rate per_pid derive=rate(write_bytes) reduce=max + io_blocked_pct per_pid derive=rate(blkio_ticks) reduce=max # time blocked on block-IO (needs delay accounting) + io_wait_pct single reduce=max # node iowait (/proc/stat): a single read, same shape as cgroup +``` + +What the sketch shows: every collector is the same three pieces (`available` / `collect` / `measurements`); counters always reduce by `rate`/`delta` and levels by `max`. A measurement is either per-pid (rows) or a single value — and a single value is whatever the collector produces: a sum of its own pids (ps total), a kernel read (cgroup peak), or a system read (iowait). So a system-wide signal is not a special scope; it is a single-value collector like cgroup. The shape holds across all five. + +
+ +## What this enables + +- A per-sample instantaneous CPU rate that can reveal spikes, cross-platform, with a user-controlled resolution/noise tradeoff. +- Kernel-accurate peak memory (cgroup) for HPC sizing, added as a modular collector rather than special-cased. +- Running more than one collector at once — e.g. measuring the same quantity two ways to compare them directly. +- Future sources (psutil, `/proc`, I/O statistics) added as collectors without touching the core. + +## Other approaches considered + +- **Surgical fixes without the interface:** change `total_rss` to `max(per-process rss)` instead of the sum (removes the shared-page double-count, but undercounts a large private child), and keep the plot-time CPU correction (already shipped, approximate). Smaller, but does not generalize and does not let new sources compose. +- **Add `cputime` and the cgroup peak as standalone fields, no refactor:** cheaper and delivers the two numbers, but commits field names and record placement without the structure, and each additional source stays ad-hoc. + +## Implementation status (POC) + +A proof of concept of this model lives in `con_duct._collectors` (pure +collectors → raw readings) and `con_duct._aggregation` (buffer → +aggregate-once). It replaces the old per-sample `max` engine end to end. See +`docs/design/DECISIONS_LOG.md` for the per-decision rationale. What the POC +actually builds, and where it deviates from the sketch above: + +- **Collectors built:** `ps` (per-sample, per-pid: `ps_rss`, `ps_rss_total`, + `ps_pdcpu`, `ps_cpu_seconds`), `cgroup` (per-report single read: + `cgroup_rss_peak`, reading duct's *own* cgroup peak — v2 `memory.peak` / + v1 `memory.max_usage_in_bytes`), and optional `psutil` (`psutil_pss`, + `psutil_pss_total`, `psutil_pdcpu`). The `io` and `/proc` collectors are not + built. +- **`pmem`/`vsz`/lifetime-`pcpu` are dropped**, not just renamed: the model + records only the keys above. The old `pcpu` lifetime average is superseded by + `*_pdcpu`. +- **`ps_pdcpu`/`psutil_pdcpu` units are cputime-seconds per wall-second + (≈ cores; 1.0 = one fully-used core), not percent.** The `rate` derive is + generic `Δfield/Δt`, so the same machinery will serve future I/O byte-rates. +- **Extra `ps_cmd` key** (per_pid, `reduce=last`): a string command label, not + in the key list above, kept so per-pid records stay identifiable. +- **Selection:** `con-duct run --measurements k1,k2,...` (default: all available + keys; `DUCT_MEASUREMENTS` env). An unknown key, or a `psutil_*` key without + psutil installed, is a clean fail-fast error. +- **On-disk shapes** (schema 0.3.0): each `usage.jsonl` record is + `{timestamp, num_samples, measurements: {key: scalar | {pid: value}}}`; the + end-of-run `execution_summary` carries `peak_ps_rss_total`, + `ave_ps_rss_total`, `ps_cpu_seconds`, `peak_cgroup_rss_peak`, + `peak_psutil_pss_total` (always present, `null` when unselected/unavailable). +- **Out of scope / not migrated:** `plot`, `ls`, and `pprint` still read the + old field shapes and were intentionally not updated to the new records. diff --git a/src/con_duct/_aggregation.py b/src/con_duct/_aggregation.py new file mode 100644 index 00000000..057df24a --- /dev/null +++ b/src/con_duct/_aggregation.py @@ -0,0 +1,307 @@ +"""Buffer raw readings and aggregate them once per report. + +The collectors in :mod:`con_duct._collectors` are pure: each sample they append +raw readings to this aggregator's buffer. At report time +:meth:`Aggregator.report` runs the design's three-step aggregation once over the +buffered window: + + derive (e.g. a rate from a counter) + -> collapse per-pid readings into single values where the scope is single + -> reduce over the interval (max for a level, delta/last for a counter) + +The last raw reading of each window is kept as the **seed** for the next window +so a delta or rate that crosses a report boundary is not lost. A run-level +accumulator feeds the end-of-run summary. + +See ``docs/design/resource-collectors.md``. +""" + +from __future__ import annotations +from collections import defaultdict +from datetime import datetime +import logging +import subprocess +import time +from typing import Dict, List, Optional, Tuple +from con_duct._collectors import ( + Collector, + Derive, + Measurement, + PerPidReading, + Reduce, + Scope, + available_collectors, +) + +lgr = logging.getLogger("con-duct") + +# A reduced measurement value: a scalar (single scope) or a {pid: value} map. +ReducedValue = object + +# Measurement-derived keys always present in execution_summary, so info.json +# schema does not depend on which collectors happened to be available. +SUMMARY_KEYS = ( + "peak_ps_rss_total", + "ave_ps_rss_total", + "ps_cpu_seconds", + "peak_cgroup_rss_peak", + "peak_psutil_pss_total", +) + + +def null_summary() -> Dict[str, Optional[float]]: + """The measurement summary with every value ``None`` (no run happened).""" + return {key: None for key in SUMMARY_KEYS} + + +class Aggregator: + """Owns the per-collector buffers, the cross-boundary seeds, and the + run-level summary accumulator for one duct execution. + + :param measurements: the selected measurements (already resolved/validated). + :param session_id: tracked session id, used to construct collectors. + """ + + def __init__( + self, + measurements: List[Measurement], + session_id: int, + collectors: Optional[List[Collector]] = None, + ) -> None: + self.measurements = measurements + selected_collectors = {m.collector for m in measurements} + + # Only keep collectors that back a selected measurement. Tests may + # inject fakes; otherwise discover the environment's available ones. + if collectors is None: + collectors = available_collectors(session_id) + self._collectors: Dict[str, Collector] = { + c.name: c for c in collectors if c.name in selected_collectors + } + + # Current-window buffers (per per-sample collector): list of + # (monotonic_ts, reading). Wall timestamps kept in parallel. + self._samples: Dict[str, List[Tuple[float, PerPidReading]]] = defaultdict(list) + self._wall_times: List[str] = [] + + # Cross-report seeds. + # rate seeds: measurement_name -> {pid: (mono_ts, value)} + self._rate_seeds: Dict[str, Dict[int, Tuple[float, float]]] = defaultdict(dict) + # delta seeds: measurement_name -> last cumulative collapsed total + self._delta_seeds: Dict[str, float] = defaultdict(float) + + # Run-level accumulators for the summary. + self._run_peak: Dict[str, float] = {} + self._run_total: Dict[str, float] = defaultdict(float) + self._run_sum: Dict[str, float] = defaultdict(float) # for averages + self._run_count: Dict[str, int] = defaultdict(int) + self.num_samples = 0 + self.num_reports = 0 + + @property + def has_samples(self) -> bool: + """True when the current window holds at least one buffered sample.""" + return bool(self._wall_times) + + def add_sample(self) -> bool: + """Collect one sample from every per-sample collector into the buffer. + + :returns: True if a sample was buffered; False when ``ps`` found no + processes (the session is gone), letting the monitor loop stop. + + When ``ps`` finds no processes the sample is dropped entirely (not + buffered, not counted): there is nothing to record and an empty + trailing reading must not skew a delta/total. + """ + mono = time.monotonic() + wall = datetime.now().astimezone().isoformat() + readings: Dict[str, PerPidReading] = {} + ps_pid_count = 0 + for name, collector in self._collectors.items(): + if not collector.per_sample: # type: ignore[attr-defined] + continue + try: + reading = collector.collect() # type: ignore[attr-defined] + except subprocess.CalledProcessError as exc: + # ps exits non-zero when the session has no processes left. + lgr.debug("Collector %s found no processes: %s", name, exc) + reading = {} + readings[name] = reading + if name == "ps": + ps_pid_count = len(reading) + + # ps is the liveness signal: no pids => session gone, drop the sample. + if "ps" in self._collectors and ps_pid_count == 0: + return False + + for name, reading in readings.items(): + self._samples[name].append((mono, reading)) + self._wall_times.append(wall) + self.num_samples += 1 + return True + + def report(self) -> Optional[dict]: + """Aggregate the current window into one usage record, then reset it. + + :returns: the record ``{timestamp, num_samples, measurements}`` or + ``None`` when the window held no samples (nothing to write). + """ + if not self.has_samples: + return None + + window_samples = len(self._wall_times) + measurements: Dict[str, ReducedValue] = {} + for meas in self.measurements: + value = self._reduce_measurement(meas) + if value is not None: + measurements[meas.name] = value + self._accumulate_summary(meas, value, window_samples) + + record = { + "timestamp": self._wall_times[-1], + "num_samples": window_samples, + "measurements": measurements, + } + self.num_reports += 1 + self._reset_window() + return record + + def _reset_window(self) -> None: + self._samples = defaultdict(list) + self._wall_times = [] + + # -- reduction --------------------------------------------------------- + + def _reduce_measurement(self, meas: Measurement) -> ReducedValue: + collector = self._collectors.get(meas.collector) + if collector is None: + return None + + if not collector.per_sample: # type: ignore[attr-defined] + # Single value read once per report (e.g. cgroup peak), reduce=last. + single = collector.read() # type: ignore[attr-defined] + return single.get(meas.field) + + samples = self._samples[meas.collector] + if meas.scope is Scope.PER_PID: + if meas.derive is Derive.RATE: + return self._per_pid_rate(meas, samples) + if meas.reduce is Reduce.LAST: + return self._per_pid_last(meas, samples) + return self._per_pid_max(meas, samples) + # SINGLE: collapse across pids per sample, then reduce over the window. + return self._single(meas, samples) + + @staticmethod + def _per_pid_max( + meas: Measurement, samples: List[Tuple[float, PerPidReading]] + ) -> Dict[str, float]: + out: Dict[str, float] = {} + for _ts, reading in samples: + for pid, fields in reading.items(): + val = float(fields[meas.field]) # type: ignore[arg-type] + key = str(pid) + out[key] = val if key not in out else max(out[key], val) + return out + + @staticmethod + def _per_pid_last( + meas: Measurement, samples: List[Tuple[float, PerPidReading]] + ) -> Dict[str, object]: + out: Dict[str, object] = {} + for _ts, reading in samples: # ordered; last write wins + for pid, fields in reading.items(): + out[str(pid)] = fields[meas.field] + return out + + def _per_pid_rate( + self, meas: Measurement, samples: List[Tuple[float, PerPidReading]] + ) -> Dict[str, Optional[float]]: + # Build each pid's (ts, value) sequence, prepend the carried seed so a + # rate spanning the report boundary is included. + seq_by_pid: Dict[int, List[Tuple[float, float]]] = defaultdict(list) + for ts, reading in samples: + for pid, fields in reading.items(): + seq_by_pid[pid].append((ts, float(fields[meas.field]))) # type: ignore[arg-type] + + prev_seed = self._rate_seeds[meas.name] + out: Dict[str, Optional[float]] = {} + new_seed: Dict[int, Tuple[float, float]] = {} + for pid, seq in seq_by_pid.items(): + points = ([prev_seed[pid]] if pid in prev_seed else []) + seq + rates = [ + (v1 - v0) / (t1 - t0) + for (t0, v0), (t1, v1) in zip(points, points[1:]) + if t1 > t0 and v1 >= v0 # ignore non-monotonic counter (pid reuse) + ] + out[str(pid)] = max(rates) if rates else None + new_seed[pid] = seq[-1] + self._rate_seeds[meas.name] = new_seed # drop pids that vanished + return out + + def _single( + self, meas: Measurement, samples: List[Tuple[float, PerPidReading]] + ) -> Optional[float]: + per_sample_totals = [ + sum(float(fields[meas.field]) for fields in reading.values()) # type: ignore[arg-type] + for _ts, reading in samples + ] + if not per_sample_totals: + return None + if meas.reduce is Reduce.DELTA: + last = per_sample_totals[-1] + delta = last - self._delta_seeds[meas.name] + self._delta_seeds[meas.name] = last + if delta < 0: + # Cumulative total dropped because a pid exited mid-window; we + # cannot attribute the lost counter, so report 0 for the window. + lgr.debug("delta for %s went negative (pid churn); clamping", meas.name) + return 0.0 + return delta + # Reduce.MAX over the per-sample collapsed totals. + return max(per_sample_totals) + + # -- summary ----------------------------------------------------------- + + def _accumulate_summary( + self, meas: Measurement, value: ReducedValue, window_samples: int + ) -> None: + """Fold one report's value into the run-level summary accumulator.""" + if meas.scope is not Scope.SINGLE or value is None: + return + fval = float(value) # type: ignore[arg-type] + if meas.reduce is Reduce.DELTA: + self._run_total[meas.name] += fval + else: # MAX or LAST -> a level; track the run peak (+ mean for averages) + self._run_peak[meas.name] = ( + fval + if meas.name not in self._run_peak + else max(self._run_peak[meas.name], fval) + ) + self._run_sum[meas.name] += fval * window_samples + self._run_count[meas.name] += window_samples + + def summary(self) -> dict: + """Measurement-derived summary keys, always present (``None`` when the + backing measurement was not selected/available), so info.json schema is + environment-independent. + """ + + def peak(name: str) -> Optional[float]: + return self._run_peak.get(name) + + def average(name: str) -> Optional[float]: + count = self._run_count.get(name, 0) + return self._run_sum[name] / count if count else None + + return { + "peak_ps_rss_total": peak("ps_rss_total"), + "ave_ps_rss_total": average("ps_rss_total"), + "ps_cpu_seconds": ( + self._run_total["ps_cpu_seconds"] + if "ps_cpu_seconds" in self._run_total + else None + ), + "peak_cgroup_rss_peak": peak("cgroup_rss_peak"), + "peak_psutil_pss_total": peak("psutil_pss_total"), + } diff --git a/src/con_duct/_collectors.py b/src/con_duct/_collectors.py new file mode 100644 index 00000000..afb77b8c --- /dev/null +++ b/src/con_duct/_collectors.py @@ -0,0 +1,403 @@ +"""Collector / measurement model for con-duct resource statistics. + +A *collector* does one I/O pass and yields raw readings; it is pure (no +derivation, no cross-sample state). A *measurement* is a namespaced key that +selects one raw field from a collector and declares how it is scoped, derived, +and reduced over a report interval. See ``docs/design/resource-collectors.md``. + +Two collector flavors: + +- per-sample / per-pid (``ps``, ``psutil``): ``collect()`` is called every + sample and returns ``{pid: {field: value}}``. +- per-report / single (``cgroup``): ``read()`` is called once per report and + returns ``{field: value}`` (the kernel already took the peak). + +The aggregation that turns buffered raw readings into reduced measurement +values lives in :mod:`con_duct._aggregation`; collectors stay pure. +""" + +from __future__ import annotations +from dataclasses import dataclass +from enum import Enum +import logging +from pathlib import Path +import platform +import subprocess +import sys +from typing import Dict, List, Optional, Protocol, Union, runtime_checkable +from con_duct._utils import etime_to_etimes + +lgr = logging.getLogger("con-duct") + +SYSTEM = platform.system() + +_SUPPORTED_SYSTEMS = {"Linux", "Darwin"} +if SYSTEM not in _SUPPORTED_SYSTEMS: + sys.tracebacklimit = 0 + raise NotImplementedError( + f"`con_duct` does not currently support the detected operating system " + f"({SYSTEM}).\n\nIf you would like to request support, please open an " + f"issue at: https://github.com/con/duct/issues/new" + ) + +# A single raw field value: numeric for everything reduced arithmetically, str +# for the ``cmd`` label (reduce=last). +ReadingValue = Union[float, int, str] +# One per-pid collect() pass: pid -> {field: value}. +PerPidReading = Dict[int, Dict[str, ReadingValue]] +# One single read() pass: {field: value}. +SingleReading = Dict[str, float] + + +class Scope(str, Enum): + """Whether a measurement is one value per process or one value total.""" + + PER_PID = "per_pid" + SINGLE = "single" + + def __str__(self) -> str: + return self.value + + +class Derive(str, Enum): + """Optional per-sample derivation applied before reduction.""" + + NONE = "none" + RATE = "rate" # Δfield / Δt between consecutive readings of a pid + + def __str__(self) -> str: + return self.value + + +class Reduce(str, Enum): + """How per-sample values collapse over a report interval.""" + + MAX = "max" # instantaneous level (or peak per-sample rate) + DELTA = "delta" # cumulative counter: difference across the interval + LAST = "last" # already-reduced value (e.g. a kernel high-water mark) + + def __str__(self) -> str: + return self.value + + +@dataclass(frozen=True) +class Measurement: + """A user-selectable measurement key. + + :param name: the namespaced key the user selects (e.g. ``ps_rss``). + :param collector: the collector that produces the raw field. + :param field: the raw field read from that collector's readings. + :param scope: per-process or a single total. + :param reduce: how per-sample values collapse over the interval. + :param derive: optional per-sample derivation (e.g. a rate from a counter). + :param optional: True when the backing collector is an optional dependency + (psutil); requesting such a key without the dependency is a clean error. + """ + + name: str + collector: str + field: str + scope: Scope + reduce: Reduce + derive: Derive = Derive.NONE + optional: bool = False + + +@runtime_checkable +class Collector(Protocol): + """Structural type shared by every collector. + + A per-sample collector additionally implements ``collect()``; a per-report + collector implements ``read()`` (callers branch on ``per_sample``). + """ + + name: str + per_sample: bool + measurements: List[Measurement] + + +# --------------------------------------------------------------------------- +# Collectors +# --------------------------------------------------------------------------- + + +class PsCollector: + """``ps``-backed per-pid collector; always available (the baseline). + + One ``ps`` call per sample yields ``rss`` (bytes), ``cputime`` (seconds), + and ``cmd`` for every pid in the tracked session. + """ + + name = "ps" + per_sample = True + measurements = [ + Measurement("ps_rss", "ps", "rss", Scope.PER_PID, Reduce.MAX), + Measurement("ps_rss_total", "ps", "rss", Scope.SINGLE, Reduce.MAX), + Measurement( + "ps_pdcpu", "ps", "cputime", Scope.PER_PID, Reduce.MAX, Derive.RATE + ), + Measurement("ps_cpu_seconds", "ps", "cputime", Scope.SINGLE, Reduce.DELTA), + # Not in the design's key list: a string label kept so per-pid records + # stay identifiable (plot labels, ls, child-counting). reduce=last. + Measurement("ps_cmd", "ps", "cmd", Scope.PER_PID, Reduce.LAST), + ] + + def __init__(self, session_id: int) -> None: + self.session_id = session_id + + @staticmethod + def available() -> bool: + return SYSTEM in {"Linux", "Darwin"} + + def collect(self) -> PerPidReading: + if SYSTEM == "Darwin": + return self._collect_darwin() + return self._collect_linux() + + def _collect_linux(self) -> PerPidReading: + ps_command = [ + "ps", + "-w", + "-s", + str(self.session_id), + "-o", + "pid,rss,cputime,cmd", + ] + output = subprocess.check_output(ps_command, text=True) + reading: PerPidReading = {} + for line in output.splitlines()[1:]: + if not line: + continue + pid, rss_kib, cputime, cmd = line.split(maxsplit=3) + reading[int(pid)] = { + "rss": int(rss_kib) * 1024, + "cputime": etime_to_etimes(cputime), + "cmd": cmd, + } + return reading + + def _collect_darwin(self) -> PerPidReading: + # macOS ps cannot filter by session id, so list all and filter by + # getsid (mirrors _sampling._get_sample_mac). Untested in CI (Linux). + import os + + ps_command = ["ps", "-ax", "-o", "pid,rss,cputime,command"] + output = subprocess.check_output(ps_command, text=True) + reading: PerPidReading = {} + for line in output.splitlines()[1:]: + if not line: + continue + pid_s, rss_kb, cputime, cmd = line.split(maxsplit=3) + pid = int(pid_s) + try: + if os.getsid(pid) != self.session_id: + continue + except ProcessLookupError: + continue + reading[pid] = { + "rss": int(rss_kb) * 1024, + "cputime": etime_to_etimes(cputime), + "cmd": cmd, + } + return reading + + +class CgroupCollector: + """Memory-cgroup high-water peak; a single value read once per report. + + Reader-mode only: reads duct's own cgroup peak counter (duct runs inside + the job cgroup under SLURM). Never creates a cgroup, never escalates + privilege; refuses cleanly when the counter is not readable. + """ + + name = "cgroup" + per_sample = False + measurements = [ + Measurement("cgroup_rss_peak", "cgroup", "mem_peak", Scope.SINGLE, Reduce.LAST), + ] + + def __init__(self, peak_path: Optional[Path] = None) -> None: + self.peak_path = peak_path if peak_path is not None else self._find_peak_path() + + @staticmethod + def _find_peak_path() -> Optional[Path]: + """Resolve duct's own memory-cgroup peak file, or None if absent.""" + try: + cgroup_lines = Path("/proc/self/cgroup").read_text().splitlines() + except OSError: + return None + + # cgroup v2: a single "0::" line; peak is memory.peak. + for line in cgroup_lines: + if line.startswith("0::"): + rel = line[3:].lstrip("/") + candidate = Path("/sys/fs/cgroup", rel, "memory.peak") + if candidate.is_file(): + return candidate + root = Path("/sys/fs/cgroup/memory.peak") + return root if root.is_file() else None + + # cgroup v1: the "...:memory:" line; peak is max_usage_in_bytes. + for line in cgroup_lines: + parts = line.split(":") + if len(parts) == 3 and "memory" in parts[1].split(","): + rel = parts[2].lstrip("/") + candidate = Path( + "/sys/fs/cgroup/memory", rel, "memory.max_usage_in_bytes" + ) + if candidate.is_file(): + return candidate + return None + + def available(self) -> bool: + path = self.peak_path + if path is None: + return False + try: + with open(path): + pass + except OSError as exc: + lgr.debug("cgroup peak file %s not readable: %s", path, exc) + return False + return True + + def read(self) -> SingleReading: + assert self.peak_path is not None + return {"mem_peak": float(int(self.peak_path.read_text().strip()))} + + +class PsutilCollector: + """Optional per-pid collector: PSS (Linux) and a pure cputime for rate. + + Uses ``psutil`` if importable. Reads raw ``cpu_times()`` (user+system), + never ``cpu_percent()``, so the collect stays pure. PSS comes from + ``memory_full_info().pss`` (Linux only). + """ + + name = "psutil" + per_sample = True + measurements = [ + Measurement( + "psutil_pss", "psutil", "pss", Scope.PER_PID, Reduce.MAX, optional=True + ), + Measurement( + "psutil_pss_total", + "psutil", + "pss", + Scope.SINGLE, + Reduce.MAX, + optional=True, + ), + Measurement( + "psutil_pdcpu", + "psutil", + "cputime", + Scope.PER_PID, + Reduce.MAX, + Derive.RATE, + optional=True, + ), + ] + + def __init__(self, session_id: int) -> None: + self.session_id = session_id + + @staticmethod + def available() -> bool: + import importlib.util + + return importlib.util.find_spec("psutil") is not None + + def collect(self) -> PerPidReading: + import os + import psutil # type: ignore[import-untyped] + + reading: PerPidReading = {} + for proc in psutil.process_iter(["pid"]): + pid = proc.info["pid"] + try: + if os.getsid(pid) != self.session_id: + continue + with proc.oneshot(): + cpu = proc.cpu_times() + pss = proc.memory_full_info().pss + except (psutil.NoSuchProcess, psutil.AccessDenied, ProcessLookupError): + continue + reading[pid] = {"pss": pss, "cputime": cpu.user + cpu.system} + return reading + + +# --------------------------------------------------------------------------- +# Registry / selection +# --------------------------------------------------------------------------- + +# key -> Measurement, across every collector (selectable or not yet available). +ALL_MEASUREMENTS: Dict[str, Measurement] = { + m.name: m + for m in ( + *PsCollector.measurements, + *CgroupCollector.measurements, + *PsutilCollector.measurements, + ) +} + + +class UnknownMeasurementError(ValueError): + """A requested measurement key does not exist.""" + + +class UnavailableMeasurementError(RuntimeError): + """A requested key's collector is unavailable (e.g. psutil not installed).""" + + +def available_collectors(session_id: int) -> List[Collector]: + """Instantiate every collector whose ``available()`` holds (ps first).""" + instances: List[Collector] = [] + if PsCollector.available(): + instances.append(PsCollector(session_id)) + cgroup = CgroupCollector() + if cgroup.available(): + instances.append(cgroup) + if PsutilCollector.available(): + instances.append(PsutilCollector(session_id)) + return instances + + +def available_keys(session_id: int) -> List[str]: + """All measurement keys whose collector is available in this environment.""" + keys: List[str] = [] + for inst in available_collectors(session_id): + keys.extend(m.name for m in inst.measurements) + return keys + + +def resolve_selection( + requested: Optional[List[str]], session_id: int +) -> List[Measurement]: + """Resolve a user key selection into Measurements. + + ``None`` selects every available key. An explicit selection validates each + key: unknown keys raise :class:`UnknownMeasurementError`; keys whose + (optional) collector is unavailable raise + :class:`UnavailableMeasurementError` with an actionable message. + """ + avail = set(available_keys(session_id)) + if requested is None: + return [ALL_MEASUREMENTS[k] for k in available_keys(session_id)] + + resolved: List[Measurement] = [] + for key in requested: + meas = ALL_MEASUREMENTS.get(key) + if meas is None: + raise UnknownMeasurementError( + f"Unknown measurement key {key!r}. " + f"Available keys: {', '.join(sorted(ALL_MEASUREMENTS))}." + ) + if key not in avail: + hint = " (install con-duct[all] for psutil)" if meas.optional else "" + raise UnavailableMeasurementError( + f"Measurement {key!r} is unavailable: its collector " + f"({meas.collector}) is not available in this environment{hint}." + ) + resolved.append(meas) + return resolved diff --git a/src/con_duct/_constants.py b/src/con_duct/_constants.py index db84702f..07457347 100644 --- a/src/con_duct/_constants.py +++ b/src/con_duct/_constants.py @@ -1,6 +1,6 @@ """Constants used throughout con-duct.""" -__schema_version__ = "0.2.2" +__schema_version__ = "0.3.0" ENV_PREFIXES = ("PBS_", "SLURM_", "OSG") SUFFIXES = { diff --git a/src/con_duct/_duct_main.py b/src/con_duct/_duct_main.py index e2954927..d3bcbb28 100644 --- a/src/con_duct/_duct_main.py +++ b/src/con_duct/_duct_main.py @@ -6,7 +6,12 @@ import subprocess import threading import time -from typing import IO, TextIO +from typing import IO, List, Optional, TextIO +from con_duct._collectors import ( + UnavailableMeasurementError, + UnknownMeasurementError, + resolve_selection, +) from con_duct._models import LogPaths, Outputs, RecordTypes, SessionMode from con_duct._output import TailPipe, prepare_outputs, remove_files, safe_close_files from con_duct._signals import SigIntHandler @@ -23,14 +28,11 @@ "Command: {command}\n" "Log files location: {logs_prefix}\n" "Wall Clock Time: {wall_clock_time:.3f} sec\n" - "Memory Peak Usage (RSS): {peak_rss!S}\n" - "Memory Average Usage (RSS): {average_rss!S}\n" - "Virtual Memory Peak Usage (VSZ): {peak_vsz!S}\n" - "Virtual Memory Average Usage (VSZ): {average_vsz!S}\n" - "Memory Peak Percentage: {peak_pmem:.2f!N}%\n" - "Memory Average Percentage: {average_pmem:.2f!N}%\n" - "CPU Peak Usage: {peak_pcpu:.2f!N}%\n" - "Average CPU Usage: {average_pcpu:.2f!N}%\n" + "Memory Peak Usage (ps RSS total): {peak_ps_rss_total!S}\n" + "Memory Average Usage (ps RSS total): {ave_ps_rss_total!S}\n" + "CPU Seconds (ps): {ps_cpu_seconds!N}\n" + "Memory Peak (cgroup): {peak_cgroup_rss_peak!S}\n" + "Memory Peak (psutil PSS total): {peak_psutil_pss_total!S}\n" ) @@ -49,6 +51,7 @@ def execute( colors: bool, mode: SessionMode, message: str = "", + measurements: Optional[str] = None, ) -> int: """A wrapper to execute a command, monitor and log the process details. @@ -65,6 +68,20 @@ def execute( sample_interval, ) + # Resolve the measurement selection up front so an unknown/unavailable key + # fails fast with a clean message before the command runs. Availability is + # environment- (not session-) based, so resolve with our own session id. + requested: Optional[List[str]] = ( + [key.strip() for key in measurements.split(",") if key.strip()] + if measurements + else None + ) + try: + resolved_measurements = resolve_selection(requested, os.getsid(0)) + except (UnknownMeasurementError, UnavailableMeasurementError) as exc: + lgr.error("%s", exc) + return 1 + log_paths = LogPaths.create(output_prefix, pid=os.getpid()) try: log_paths.prepare_paths(clobber, capture_outputs) @@ -96,6 +113,7 @@ def execute( colors, clobber, message=message, + measurements=resolved_measurements, ) files_to_close.append(report.usage_file) @@ -166,9 +184,11 @@ def execute( monitoring_thread.join() lgr.debug("Monitoring thread finished") - # If we have any extra samples that haven't been written yet, do it now - if report.current_sample is not None: - report.write_subreport() + # Flush any buffered samples from the final, partial report window. + if report.aggregator is not None: + final_record = report.aggregator.report() + if final_record is not None: + report.write_record(final_record) report.process = process if env_thread is not None: diff --git a/src/con_duct/_models.py b/src/con_duct/_models.py index e83a9adf..5e38fc4e 100644 --- a/src/con_duct/_models.py +++ b/src/con_duct/_models.py @@ -1,18 +1,14 @@ """Data models and enums for con-duct.""" from __future__ import annotations -from collections import Counter from collections.abc import Iterator -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass from datetime import datetime from enum import Enum import logging import os -import re import string -from typing import Any, Optional from con_duct._constants import SUFFIXES -from con_duct._utils import assert_num lgr = logging.getLogger("con-duct") @@ -65,56 +61,6 @@ class SystemInfo: user: str | None -@dataclass -class ProcessStats: - pcpu: float # %CPU - pmem: float # %MEM - rss: int # Memory Resident Set Size in Bytes - vsz: int # Virtual Memory size in Bytes - timestamp: str - etime: str - stat: Counter - cmd: str - - def aggregate(self, other: ProcessStats) -> ProcessStats: - cmd = self.cmd - if self.cmd != other.cmd: - lgr.debug( - f"cmd has changed. Previous measurement was {self.cmd}, now {other.cmd}." - ) - # Brackets indicate that the kernel has substituted an abbreviation. - surrounded_by_brackets = r"^\[.+\]" - if re.search(surrounded_by_brackets, self.cmd): - lgr.debug(f"using {other.cmd}.") - cmd = other.cmd - lgr.debug(f"using {self.cmd}.") - - new_counter: Counter = Counter() - new_counter.update(self.stat) - new_counter.update(other.stat) - return ProcessStats( - pcpu=max(self.pcpu, other.pcpu), - pmem=max(self.pmem, other.pmem), - rss=max(self.rss, other.rss), - vsz=max(self.vsz, other.vsz), - timestamp=max(self.timestamp, other.timestamp), - etime=other.etime, # For the aggregate always take the latest - stat=new_counter, - cmd=cmd, - ) - - def for_json(self) -> dict: - ret = asdict(self) - ret["stat"] = dict(self.stat) - return ret - - def __post_init__(self) -> None: - self._validate() - - def _validate(self) -> None: - assert_num(self.pcpu, self.pmem, self.rss, self.vsz) - - @dataclass class LogPaths: stdout: str @@ -179,110 +125,3 @@ def prepare_paths(self, clobber: bool, capture_outputs: Outputs) -> None: # assistant monitoring new files to be created and committing # as soon as they are closed open(path, "w").close() - - -@dataclass -class Averages: - rss: Optional[float] = None - vsz: Optional[float] = None - pmem: Optional[float] = None - pcpu: Optional[float] = None - num_samples: int = 0 - - def update(self: Averages, other: Sample) -> None: - assert_num(other.total_rss, other.total_vsz, other.total_pmem, other.total_pcpu) - if not self.num_samples: - self.num_samples += 1 - self.rss = other.total_rss - self.vsz = other.total_vsz - self.pmem = other.total_pmem - self.pcpu = other.total_pcpu - else: - assert self.rss is not None - assert self.vsz is not None - assert self.pmem is not None - assert self.pcpu is not None - assert other.total_rss is not None - assert other.total_vsz is not None - assert other.total_pmem is not None - assert other.total_pcpu is not None - self.num_samples += 1 - self.rss += (other.total_rss - self.rss) / self.num_samples - self.vsz += (other.total_vsz - self.vsz) / self.num_samples - self.pmem += (other.total_pmem - self.pmem) / self.num_samples - self.pcpu += (other.total_pcpu - self.pcpu) / self.num_samples - - @classmethod - def from_sample(cls, sample: Sample) -> Averages: - assert_num( - sample.total_rss, sample.total_vsz, sample.total_pmem, sample.total_pcpu - ) - return cls( - rss=sample.total_rss, - vsz=sample.total_vsz, - pmem=sample.total_pmem, - pcpu=sample.total_pcpu, - num_samples=1, - ) - - -@dataclass -class Sample: - stats: dict[int, ProcessStats] = field(default_factory=dict) - averages: Averages = field(default_factory=Averages) - total_rss: Optional[int] = None - total_vsz: Optional[int] = None - total_pmem: Optional[float] = None - total_pcpu: Optional[float] = None - timestamp: str = "" # TS of last sample collected - - def add_pid(self, pid: int, stats: ProcessStats) -> None: - # We do not calculate averages when we add a pid because we require all pids first - assert ( - self.stats.get(pid) is None - ) # add_pid should only be called when pid not in Sample - self.total_rss = (self.total_rss or 0) + stats.rss - self.total_vsz = (self.total_vsz or 0) + stats.vsz - self.total_pmem = (self.total_pmem or 0.0) + stats.pmem - self.total_pcpu = (self.total_pcpu or 0.0) + stats.pcpu - self.stats[pid] = stats - self.timestamp = max(self.timestamp, stats.timestamp) - - def aggregate(self: Sample, other: Sample) -> Sample: - output = Sample() - for pid in self.stats.keys() | other.stats.keys(): - if (mine := self.stats.get(pid)) is not None: - if (theirs := other.stats.get(pid)) is not None: - output.add_pid(pid, mine.aggregate(theirs)) - else: - output.add_pid(pid, mine) - else: - output.add_pid(pid, other.stats[pid]) - assert other.total_pmem is not None - assert other.total_pcpu is not None - assert other.total_rss is not None - assert other.total_vsz is not None - output.total_pmem = max(self.total_pmem or 0.0, other.total_pmem) - output.total_pcpu = max(self.total_pcpu or 0.0, other.total_pcpu) - output.total_rss = max(self.total_rss or 0, other.total_rss) - output.total_vsz = max(self.total_vsz or 0, other.total_vsz) - output.averages = self.averages - output.averages.update(other) - return output - - def for_json(self) -> dict[str, Any]: - d = { - "timestamp": self.timestamp, - "num_samples": self.averages.num_samples, - "processes": { - str(pid): stats.for_json() for pid, stats in self.stats.items() - }, - "totals": { # total of all processes during this sample - "pmem": self.total_pmem, - "pcpu": self.total_pcpu, - "rss": self.total_rss, - "vsz": self.total_vsz, - }, - "averages": asdict(self.averages) if self.averages.num_samples >= 1 else {}, - } - return d diff --git a/src/con_duct/_sampling.py b/src/con_duct/_sampling.py deleted file mode 100644 index caa90517..00000000 --- a/src/con_duct/_sampling.py +++ /dev/null @@ -1,146 +0,0 @@ -"""Platform-specific process sampling for con-duct.""" - -from __future__ import annotations -from collections import Counter, deque -from datetime import datetime -import logging -import os -import platform -import subprocess -import sys -from typing import Callable, Optional -from con_duct._models import Averages, ProcessStats, Sample - -SYSTEM = platform.system() - -_SUPPORTED_SYSTEMS = {"Linux", "Darwin"} -if SYSTEM not in _SUPPORTED_SYSTEMS: - sys.tracebacklimit = 0 - message = ( - f"`con_duct` does not currently support the detected operating system ({SYSTEM}).\n\n" - "If you would like to request support, please open an issue at: " - "https://github.com/con/duct/issues/new" - ) - raise NotImplementedError(message) - -lgr = logging.getLogger("con-duct") - - -def _get_sample_linux(session_id: int) -> Sample: - sample = Sample() - - ps_command = [ - "ps", - "-w", - "-s", - str(session_id), - "-o", - "pid,pcpu,pmem,rss,vsz,etime,stat,cmd", - ] - output = subprocess.check_output(ps_command, text=True) - - for line in output.splitlines()[1:]: - if not line: - continue - - pid, pcpu, pmem, rss_kib, vsz_kib, etime, stat, cmd = line.split(maxsplit=7) - - sample.add_pid( - pid=int(pid), - stats=ProcessStats( - pcpu=float(pcpu), - pmem=float(pmem), - rss=int(rss_kib) * 1024, - vsz=int(vsz_kib) * 1024, - timestamp=datetime.now().astimezone().isoformat(), - etime=etime, - stat=Counter([stat]), - cmd=cmd, - ), - ) - sample.averages = Averages.from_sample(sample=sample) - return sample - - -def _try_to_get_sid(pid: int) -> int: - """ - It is possible that the `pid` returned by the top `ps` call no longer exists at time of `getsid` request. - """ - try: - return os.getsid(pid) - except ProcessLookupError as exc: - lgr.debug(f"Error fetching session ID for PID {pid}: {str(exc)}") - return -1 - - -def _get_ps_lines_mac() -> list[str]: - ps_command = [ - "ps", - "-ax", - "-o", - "pid,pcpu,pmem,rss,vsz,etime,stat,args", - ] - output = subprocess.check_output(ps_command, text=True) - - lines = [line for line in output.splitlines()[1:] if line] - return lines - - -def _add_pid_to_sample_from_line_mac( - line: str, pid_to_matching_sid: dict[int, int], sample: Sample -) -> None: - pid, pcpu, pmem, rss_kb, vsz_kb, etime, stat, cmd = line.split(maxsplit=7) - - if pid_to_matching_sid.get(int(pid)) is not None: - sample.add_pid( - pid=int(pid), - stats=ProcessStats( - pcpu=float(pcpu), - pmem=float(pmem), - rss=int(rss_kb) * 1024, - vsz=int(vsz_kb) * 1024, - timestamp=datetime.now().astimezone().isoformat(), - etime=etime, - stat=Counter([stat]), - cmd=cmd, - ), - ) - - -def _get_sample_mac(session_id: int) -> Optional[Sample]: - sample = Sample() - - lines = _get_ps_lines_mac() - pid_to_matching_sid = { - pid: sid - for line in lines - if (sid := _try_to_get_sid(pid=(pid := int(line.split(maxsplit=1)[0])))) - == session_id - } - - if not pid_to_matching_sid: - lgr.debug(f"No processes found for session ID {session_id}.") - return None - - # collections.deque with maxlen=0 is used to approximate the - # performance of list comprehension (superior to basic for-loop) - # and also does not store `None` (or other) return values - deque( - ( - _add_pid_to_sample_from_line_mac( # type: ignore[func-returns-value] - line=line, pid_to_matching_sid=pid_to_matching_sid, sample=sample - ) - for line in lines - ), - maxlen=0, - ) - - sample.averages = Averages.from_sample(sample=sample) - return sample - - -_get_sample_per_system = { - "Linux": _get_sample_linux, - "Darwin": _get_sample_mac, -} -_get_sample: Callable[[int], Optional[Sample]] = _get_sample_per_system[SYSTEM] # type: ignore[assignment] diff --git a/src/con_duct/_tracker.py b/src/con_duct/_tracker.py index 1d563b1a..7d9594e9 100644 --- a/src/con_duct/_tracker.py +++ b/src/con_duct/_tracker.py @@ -12,12 +12,13 @@ import subprocess import threading import time -from typing import Any, Optional, TextIO +from typing import Any, List, Optional, TextIO +from con_duct._aggregation import Aggregator, null_summary +from con_duct._collectors import Measurement from con_duct._constants import ENV_PREFIXES, __schema_version__ from con_duct._formatter import SummaryFormatter -from con_duct._models import LogPaths, Sample, SystemInfo +from con_duct._models import LogPaths, SystemInfo from con_duct._output import safe_close_files -from con_duct._sampling import _get_sample __version__ = version("con-duct") @@ -38,6 +39,7 @@ def __init__( clobber: bool = False, process: subprocess.Popen | None = None, message: str = "", + measurements: Optional[List[Measurement]] = None, ) -> None: self._command = command self.arguments = arguments @@ -46,6 +48,8 @@ def __init__( self.clobber = clobber self.colors = colors self.message = message + # The resolved measurement selection (empty => nothing to record). + self.measurements: List[Measurement] = measurements or [] # Defaults to be set later self.start_time: float | None = None self.process = process @@ -54,8 +58,9 @@ def __init__( self.env: dict[str, str] | None = None self.number = 1 self.system_info: SystemInfo | None = None - self.full_run_stats = Sample() - self.current_sample: Optional[Sample] = None + # The collect -> buffer -> aggregate-once engine; built once the tracked + # session id is known (see start_aggregator). + self.aggregator: Optional[Aggregator] = None self.end_time: float | None = None self.run_time_seconds: str | None = None self.usage_file: TextIO | None = None @@ -132,29 +137,16 @@ def get_system_info(self) -> None: lgr.warning("Error parsing gpu information: %s", str(e)) self.gpus = None - def collect_sample(self) -> Optional[Sample]: + def start_aggregator(self) -> None: + """Build the aggregation engine once the tracked session id is known.""" assert self.session_id is not None - try: - sample = _get_sample(self.session_id) - return sample - except subprocess.CalledProcessError as exc: # when session_id has no processes - lgr.debug("Error collecting sample: %s", str(exc)) - return None - - def update_from_sample(self, sample: Sample) -> None: - self.full_run_stats = self.full_run_stats.aggregate(sample) - if self.current_sample is None: - self.current_sample = Sample().aggregate(sample) - else: - assert self.current_sample.averages is not None - self.current_sample = self.current_sample.aggregate(sample) - assert self.current_sample is not None - - def write_subreport(self) -> None: - assert self.current_sample is not None + self.aggregator = Aggregator(self.measurements, self.session_id) + + def write_record(self, record: dict) -> None: + """Append one aggregated usage record as a JSON line.""" if self.usage_file is None: self.usage_file = open(self.log_paths.usage, "w") - self.usage_file.write(json.dumps(self.current_sample.for_json()) + "\n") + self.usage_file.write(json.dumps(record) + "\n") self.usage_file.flush() # Force flush immediately @property @@ -163,22 +155,22 @@ def execution_summary(self) -> dict[str, Any]: # https://pubs.opengroup.org/onlinepubs/9799919799/utilities/V3_chap02.html#tag_19_08_02 if self.process and self.process.returncode < 0: self.process.returncode = 128 + abs(self.process.returncode) - # prepare the base, but enrich if we did get process running + # Measurement-derived keys (always present; None when unselected) plus + # run/meta keys. Built from the aggregator, or nulls if monitoring was + # disabled so info.json schema stays stable. + measurement_summary = ( + self.aggregator.summary() if self.aggregator is not None else null_summary() + ) + num_samples = self.aggregator.num_samples if self.aggregator else 0 + num_reports = self.aggregator.num_reports if self.aggregator else 0 return { "exit_code": self.process.returncode if self.process else None, "command": self.command, "logs_prefix": self.log_paths.prefix if self.log_paths else "", "wall_clock_time": self.wall_clock_time, - "peak_rss": self.full_run_stats.total_rss, - "average_rss": self.full_run_stats.averages.rss, - "peak_vsz": self.full_run_stats.total_vsz, - "average_vsz": self.full_run_stats.averages.vsz, - "peak_pmem": self.full_run_stats.total_pmem, - "average_pmem": self.full_run_stats.averages.pmem, - "peak_pcpu": self.full_run_stats.total_pcpu, - "average_pcpu": self.full_run_stats.averages.pcpu, - "num_samples": self.full_run_stats.averages.num_samples, - "num_reports": self.number, + **measurement_summary, + "num_samples": num_samples, + "num_reports": num_reports, "start_time": self.start_time, "end_time": self.end_time, "working_directory": self.working_directory, @@ -221,17 +213,19 @@ def monitor_process( sample_interval, report_interval, ) + if report.aggregator is None: + report.start_aggregator() + assert report.aggregator is not None while True: if process.poll() is not None: lgr.debug( "Breaking out of the monitor since the passthrough command has finished" ) break - sample = report.collect_sample() - # Report averages should be updated prior to sample aggregation - if ( - sample is None - ): # passthrough has probably finished before sample could be collected + recorded = report.aggregator.add_sample() + if not recorded: + # ps found no processes -- passthrough has probably finished before + # a sample could be collected. if process.poll() is not None: lgr.debug( "Breaking out of the monitor since the passthrough command has finished " @@ -240,13 +234,13 @@ def monitor_process( break # process is still running, but we could not collect sample continue - report.update_from_sample(sample) if ( report.start_time and report.elapsed_time >= (report.number - 1) * report_interval ): - report.write_subreport() - report.current_sample = None + record = report.aggregator.report() + if record is not None: + report.write_record(record) report.number += 1 if stop_event.wait(timeout=sample_interval): lgr.debug("Breaking out because stop event was set") diff --git a/src/con_duct/cli.py b/src/con_duct/cli.py index e3cc655b..7d351bdd 100644 --- a/src/con_duct/cli.py +++ b/src/con_duct/cli.py @@ -353,6 +353,17 @@ def _create_run_parser() -> argparse.ArgumentParser: help="Record a descriptive message about the purpose of this execution. " "You can also provide value via DUCT_MESSAGE env variable.", ) + parser.add_argument( + "--measurements", + type=str, + default=os.getenv("DUCT_MEASUREMENTS"), + help="Comma-separated measurement keys to record (default: all keys " + "available in this environment). The collector behind each key is " + "internal; selecting several keys from one source is still a single " + "pass. Example: ps_rss,ps_pdcpu,ps_cpu_seconds,cgroup_rss_peak. " + "Optional psutil_* keys require con-duct[all]. " + "You can also provide value via DUCT_MEASUREMENTS env variable.", + ) parser.add_argument( "--mode", default="new-session", @@ -437,7 +448,7 @@ def _create_ls_parser() -> argparse.ArgumentParser: "command", "exit_code", "wall_clock_time", - "peak_rss", + "peak_ps_rss_total", ], ) parser.add_argument( diff --git a/src/con_duct/ls.py b/src/con_duct/ls.py index 403eed6a..08cb2374 100644 --- a/src/con_duct/ls.py +++ b/src/con_duct/ls.py @@ -26,17 +26,15 @@ lgr = logging.getLogger(__name__) VALUE_TRANSFORMATION_MAP: Dict[str, str] = { - "average_pcpu": "{value:.2f!N}%", - "average_pmem": "{value:.2f!N}%", - "average_rss": "{value!S}", - "average_vsz": "{value!S}", + # Measurement-derived summary keys (see con_duct._aggregation.SUMMARY_KEYS). + "peak_ps_rss_total": "{value!S}", + "ave_ps_rss_total": "{value!S}", + "ps_cpu_seconds": "{value:.2f!N}", + "peak_cgroup_rss_peak": "{value!S}", + "peak_psutil_pss_total": "{value!S}", "end_time": "{value:.2f!N}", "exit_code": "{value!E}", "memory_total": "{value!S}", - "peak_pcpu": "{value:.2f!N}%", - "peak_pmem": "{value:.2f!N}%", - "peak_rss": "{value!S}", - "peak_vsz": "{value!S}", "start_time": "{value:.2f!N}", "wall_clock_time": "{value:.3f} sec", } diff --git a/test/duct_main/test_aggregation.py b/test/duct_main/test_aggregation.py deleted file mode 100644 index fb6925ca..00000000 --- a/test/duct_main/test_aggregation.py +++ /dev/null @@ -1,391 +0,0 @@ -from collections import Counter -from copy import deepcopy -import os -from typing import cast -from unittest import mock -import pytest -from con_duct._duct_main import EXECUTION_SUMMARY_FORMAT -from con_duct._models import ProcessStats, Sample -from con_duct._tracker import Report - -stat0 = ProcessStats( - pcpu=0.0, - pmem=0, - rss=0, - vsz=0, - timestamp="2024-06-11T10:09:37-04:00", - etime="00:00", - cmd="cmd 0", - stat=Counter(["stat0"]), -) - -stat1 = ProcessStats( - pcpu=1.0, - pmem=1.0, - rss=1, - vsz=1, - timestamp="2024-06-11T10:13:23-04:00", - etime="00:02", - cmd="cmd 1", - stat=Counter(["stat1"]), -) - -stat2 = ProcessStats( - pcpu=2.0, - pmem=2.0, - rss=2, - vsz=2, - timestamp="2024-06-11T10:13:23-04:00", - etime="00:02", - cmd="cmd 2", - stat=Counter(["stat2"]), -) - -stat100 = ProcessStats( - pcpu=100.0, - pmem=100.0, - rss=2, - vsz=2, - timestamp="2024-06-11T10:13:23-04:00", - etime="00:02", - cmd="cmd 100", - stat=Counter(["stat100"]), -) -stat_big = ProcessStats( - pcpu=20000.0, - pmem=21234234.0, - rss=43645634562, - vsz=2345234523452342, - timestamp="2024-06-11T10:13:23-04:00", - etime="00:02", - cmd="cmd 2", - stat=Counter(["statbig"]), -) - - -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_num_samples_increment(mock_log_paths: mock.MagicMock) -> None: - ex0 = Sample() - ex0.add_pid(1, deepcopy(stat1)) - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - assert report.current_sample is None - assert report.full_run_stats.averages.num_samples == 0 - report.update_from_sample(ex0) - report.current_sample = cast( - Sample, report.current_sample - ) # So mypy is convcinced it is not None - assert report.current_sample is not None - assert report.current_sample.averages.num_samples == 1 - assert report.full_run_stats.averages.num_samples == 1 - report.update_from_sample(ex0) - assert report.current_sample.averages.num_samples == 2 - assert report.full_run_stats.averages.num_samples == 2 - report.update_from_sample(ex0) - assert report.current_sample.averages.num_samples == 3 - assert report.full_run_stats.averages.num_samples == 3 - - -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_single_sample_sanity(mock_log_paths: mock.MagicMock) -> None: - ex0 = Sample() - ex0.add_pid(0, deepcopy(stat0)) - ex0.add_pid(1, deepcopy(stat1)) - ex0.add_pid(2, deepcopy(stat2)) - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - assert report.current_sample is None - assert report.full_run_stats.averages.num_samples == 0 - report.update_from_sample(ex0) - # 3 pids in a single sample should still be "1" sample - report.current_sample = cast( - Sample, report.current_sample - ) # So mypy is convcinced it is not None - assert report.current_sample is not None - assert report.full_run_stats is not None - assert report.current_sample.averages.num_samples == 1 - assert report.full_run_stats.averages.num_samples == 1 - - # assert totals sanity - assert report.current_sample.total_rss == stat0.rss + stat1.rss + stat2.rss - assert report.current_sample.total_vsz == stat0.vsz + stat1.vsz + stat2.vsz - assert report.current_sample.total_pmem == stat0.pmem + stat1.pmem + stat2.pmem - assert report.current_sample.total_pcpu == stat0.pcpu + stat1.pcpu + stat2.pcpu - - # With one sample averages should be equal to totals - assert report.current_sample.averages.rss == report.current_sample.averages.rss - assert report.current_sample.averages.vsz == report.current_sample.averages.vsz - assert report.current_sample.averages.pmem == report.current_sample.averages.pmem - assert report.current_sample.averages.pcpu == report.current_sample.averages.pcpu - - -@pytest.mark.parametrize("stat", [stat0, stat1, stat2, stat_big]) -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_single_stat_multiple_samples_sanity( - mock_log_paths: mock.MagicMock, stat: ProcessStats -) -> None: - ex0 = Sample() - ex0.add_pid(1, deepcopy(stat)) - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - assert report.current_sample is None - assert report.full_run_stats.averages.num_samples == 0 - report.update_from_sample(ex0) - report.update_from_sample(ex0) - report.update_from_sample(ex0) - report.current_sample = cast( - Sample, report.current_sample - ) # So mypy is convcinced it is not None - assert report.current_sample is not None - assert report.current_sample.averages.num_samples == 3 - assert report.full_run_stats.averages.num_samples == 3 - - # With 3 identical samples, totals should be identical to 1 sample - assert report.current_sample.total_rss == stat.rss - assert report.current_sample.total_vsz == stat.vsz - assert report.current_sample.total_pmem == stat.pmem - assert report.current_sample.total_pcpu == stat.pcpu - - # Without resetting the current sample, full_run_stats should equal current_sample - assert report.full_run_stats.total_rss == report.current_sample.total_rss - assert report.full_run_stats.total_vsz == report.current_sample.total_vsz - assert report.full_run_stats.total_pmem == report.current_sample.total_pmem - assert report.full_run_stats.total_pcpu == report.current_sample.total_pcpu - # Averages too - assert report.current_sample.averages.rss == report.full_run_stats.averages.rss - assert report.current_sample.averages.vsz == report.full_run_stats.averages.vsz - assert report.current_sample.averages.pmem == report.full_run_stats.averages.pmem - assert report.current_sample.averages.pcpu == report.full_run_stats.averages.pcpu - - # With 3 identical samples, averages should be identical to 1 sample - assert report.current_sample.averages.rss == report.current_sample.total_rss - assert report.current_sample.averages.vsz == report.current_sample.total_vsz - assert report.current_sample.averages.pmem == report.current_sample.total_pmem - assert report.current_sample.averages.pcpu == report.current_sample.total_pcpu - - -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_averages(mock_log_paths: mock.MagicMock) -> None: - sample0 = Sample() - sample0.add_pid(1, deepcopy(stat0)) - sample1 = Sample() - sample1.add_pid(1, deepcopy(stat1)) - sample2 = Sample() - sample2.add_pid(1, deepcopy(stat2)) - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - assert report.current_sample is None - assert report.full_run_stats.averages.num_samples == 0 - report.update_from_sample(sample0) - report.update_from_sample(sample1) - report.update_from_sample(sample2) - report.current_sample = cast( - Sample, report.current_sample - ) # So mypy is convcinced it is not None - assert report.current_sample is not None - assert report.current_sample.averages.num_samples == 3 - assert report.full_run_stats.averages.num_samples == 3 - - # Assert that average calculation works as expected - assert ( - report.current_sample.averages.rss == (stat0.rss + stat1.rss + stat2.rss) / 3.0 - ) - assert ( - report.current_sample.averages.vsz == (stat0.vsz + stat1.vsz + stat2.vsz) / 3.0 - ) - assert ( - report.current_sample.averages.pmem - == (stat0.pmem + stat1.pmem + stat2.pmem) / 3.0 - ) - assert ( - report.current_sample.averages.pcpu - == (stat0.pcpu + stat1.pcpu + stat2.pcpu) / 3.0 - ) - # And full_run_stats.averages is still identical - assert report.current_sample.averages.rss == report.full_run_stats.averages.rss - assert report.current_sample.averages.vsz == report.full_run_stats.averages.vsz - assert report.current_sample.averages.pmem == report.full_run_stats.averages.pmem - assert report.current_sample.averages.pcpu == report.full_run_stats.averages.pcpu - - # Lets make the arithmetic a little less round - report.update_from_sample(sample2) - report.update_from_sample(sample2) - report.update_from_sample(sample2) - assert report.current_sample.averages.num_samples == 6 - assert report.full_run_stats.averages.num_samples == 6 - assert ( - report.current_sample.averages.rss - == (stat0.rss + stat1.rss + stat2.rss * 4) / 6.0 - ) - assert ( - report.current_sample.averages.vsz - == (stat0.vsz + stat1.vsz + stat2.vsz * 4) / 6.0 - ) - assert ( - report.current_sample.averages.pmem - == (stat0.pmem + stat1.pmem + stat2.pmem * 4) / 6.0 - ) - assert ( - report.current_sample.averages.pcpu - == (stat0.pcpu + stat1.pcpu + stat2.pcpu * 4) / 6.0 - ) - - -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_current_ave_diverges_from_total_ave( - mock_log_paths: mock.MagicMock, -) -> None: - sample0 = Sample() - sample0.add_pid(1, deepcopy(stat0)) - sample1 = Sample() - sample1.add_pid(1, deepcopy(stat1)) - sample2 = Sample() - sample2.add_pid(1, deepcopy(stat2)) - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - assert report.current_sample is None - assert report.full_run_stats.averages.num_samples == 0 - report.update_from_sample(sample0) - report.update_from_sample(sample1) - report.update_from_sample(sample2) - report.current_sample = cast( - Sample, report.current_sample - ) # So mypy is convcinced it is not None - assert report.current_sample is not None - assert report.current_sample.averages.num_samples == 3 - assert report.full_run_stats.averages.num_samples == 3 - # full_run_stats.averages is still identical to current_sample - assert report.current_sample.averages.rss == report.full_run_stats.averages.rss - assert report.current_sample.averages.vsz == report.full_run_stats.averages.vsz - assert report.current_sample.averages.pmem == report.full_run_stats.averages.pmem - assert report.current_sample.averages.pcpu == report.full_run_stats.averages.pcpu - - # Reset current_sample so averages will diverge from full_run_stats.averages - report.current_sample = None - report.update_from_sample(sample2) - report.update_from_sample(sample2) - report.update_from_sample(sample2) - report.current_sample = cast( - Sample, report.current_sample - ) # So mypy is convcinced it is not None - assert report.current_sample is not None - assert report.current_sample.averages.num_samples == 3 - assert report.full_run_stats.averages.num_samples == 6 - - # Current sample should only contain sample2 - assert report.current_sample.averages.rss == sample2.total_rss - assert report.current_sample.averages.vsz == sample2.total_vsz - assert report.current_sample.averages.pmem == sample2.total_pmem - assert report.current_sample.averages.pcpu == sample2.total_pcpu - - # Full sample average should == (samples_sum/num_samples) - assert ( - report.full_run_stats.averages.rss - == (stat0.rss + stat1.rss + stat2.rss * 4) / 6.0 - ) - assert ( - report.full_run_stats.averages.vsz - == (stat0.vsz + stat1.vsz + stat2.vsz * 4) / 6.0 - ) - assert ( - report.full_run_stats.averages.pmem - == (stat0.pmem + stat1.pmem + stat2.pmem * 4) / 6.0 - ) - assert ( - report.full_run_stats.averages.pcpu - == (stat0.pcpu + stat1.pcpu + stat2.pcpu * 4) / 6.0 - ) - - -@pytest.mark.parametrize("stat", [stat0, stat1, stat2, stat_big]) -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_many_samples( - mock_log_paths: mock.MagicMock, stat: ProcessStats -) -> None: - sample1 = Sample() - pid = 1 - sample1.add_pid(pid, deepcopy(stat)) - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - assert report.current_sample is None - assert report.full_run_stats.averages.num_samples == 0 - - # Ensure nothing strange happens after many updates - for _ in range(100): - report.update_from_sample(sample1) - - report.current_sample = cast( - Sample, report.current_sample - ) # So mypy is convcinced it is not None - assert report.current_sample is not None - # Assert that there is exactly 1 ProcessStat.stat count per update - assert ( - sum(report.current_sample.stats[pid].stat.values()) - == report.full_run_stats.averages.num_samples - == 100 - ) - assert report.full_run_stats.averages.rss == (stat.rss * 100) / 100.0 - assert report.full_run_stats.averages.vsz == (stat.vsz * 100) / 100.0 - assert report.full_run_stats.averages.pmem == (stat.pmem * 100) / 100.0 - assert report.full_run_stats.averages.pcpu == (stat.pcpu * 100) / 100.0 - - # Add a stat that is not 0 and check that the average is still correct - sample2 = Sample() - sample2.add_pid(1, deepcopy(stat2)) - report.update_from_sample(sample2) - assert report.full_run_stats.averages.num_samples == 101 - assert report.full_run_stats.averages.rss == (stat.rss * 100 + stat2.rss) / 101.0 - assert report.full_run_stats.averages.vsz == (stat.vsz * 100 + stat2.vsz) / 101.0 - assert report.full_run_stats.averages.pmem == (stat.pmem * 100 + stat2.pmem) / 101.0 - assert report.full_run_stats.averages.pcpu == (stat.pcpu * 100 + stat2.pcpu) / 101.0 - - -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_sample_no_pids(mock_log_paths: mock.MagicMock) -> None: - sample0 = Sample() - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - # When there are no pids, finalization should be triggered because the exe is finished, - # so a Sample with no PIDs should never be passed to update_from_sample. - with pytest.raises(AssertionError): - report.update_from_sample(sample0) - - -@mock.patch("con_duct._duct_main.LogPaths") -def test_aggregation_no_false_peak(mock_log_paths: mock.MagicMock) -> None: - sample1 = Sample() - sample2 = Sample() - mock_log_paths.prefix = "mock_prefix" - cwd = os.getcwd() - report = Report( - "_cmd", [], mock_log_paths, EXECUTION_SUMMARY_FORMAT, cwd, clobber=False - ) - sample1.add_pid(1, deepcopy(stat100)) - sample1.add_pid(2, deepcopy(stat0)) - report.update_from_sample(sample1) - sample2.add_pid(1, deepcopy(stat0)) - sample2.add_pid(2, deepcopy(stat100)) - report.update_from_sample(sample2) - assert report.current_sample is not None - assert report.current_sample.total_pcpu == 100 diff --git a/test/duct_main/test_aggregation_pipeline.py b/test/duct_main/test_aggregation_pipeline.py new file mode 100644 index 00000000..399ff81e --- /dev/null +++ b/test/duct_main/test_aggregation_pipeline.py @@ -0,0 +1,259 @@ +"""Unit tests for the collect -> buffer -> aggregate-once pipeline. + +A ``FakeCollector`` feeds canned per-sample readings so the derive/collapse/ +reduce logic and the cross-report seeds are exercised deterministically, +without depending on live ``ps`` timing. +""" + +from __future__ import annotations +from typing import Dict, List +import pytest +from con_duct._aggregation import Aggregator +from con_duct._collectors import ( + ALL_MEASUREMENTS, + Collector, + Measurement, + PerPidReading, +) + + +class FakeCollector: + """Per-sample collector that pops canned readings in order.""" + + per_sample = True + measurements: List[Measurement] = [] + + def __init__(self, name: str, readings: List[PerPidReading]) -> None: + self.name = name + self._readings = list(readings) + + def collect(self) -> PerPidReading: + return self._readings.pop(0) + + +class FakeSingleCollector: + """Read-once-per-report collector returning a fixed single reading.""" + + per_sample = False + measurements: List[Measurement] = [] + + def __init__(self, name: str, value: float) -> None: + self.name = name + self._value = value + + def read(self) -> Dict[str, float]: + return {"mem_peak": self._value} + + +def _agg(measurements: List[Measurement], collectors: List[Collector]) -> Aggregator: + return Aggregator(measurements, session_id=0, collectors=collectors) + + +def test_per_pid_max_level() -> None: + m = ALL_MEASUREMENTS["ps_rss"] + fake = FakeCollector( + "ps", + [ + {1: {"rss": 100.0}, 2: {"rss": 50.0}}, + {1: {"rss": 80.0}, 2: {"rss": 70.0}}, + ], + ) + agg = _agg([m], [fake]) + agg.add_sample() + agg.add_sample() + record = agg.report() + assert record is not None + assert record["measurements"]["ps_rss"] == {"1": 100.0, "2": 70.0} + assert record["num_samples"] == 2 + + +def test_single_max_total() -> None: + m = ALL_MEASUREMENTS["ps_rss_total"] + fake = FakeCollector( + "ps", + [ + {1: {"rss": 100.0}, 2: {"rss": 50.0}}, # sum 150 + {1: {"rss": 80.0}, 2: {"rss": 70.0}}, # sum 150 + {1: {"rss": 120.0}, 2: {"rss": 70.0}}, # sum 190 (peak) + ], + ) + agg = _agg([m], [fake]) + for _ in range(3): + agg.add_sample() + record = agg.report() + assert record is not None + assert record["measurements"]["ps_rss_total"] == 190.0 + + +def test_single_delta_counter() -> None: + m = ALL_MEASUREMENTS["ps_cpu_seconds"] + # cumulative cputime per pid: totals 10 -> 12 -> 15 across samples. + fake = FakeCollector( + "ps", + [ + {1: {"cputime": 6.0}, 2: {"cputime": 4.0}}, # 10 + {1: {"cputime": 7.0}, 2: {"cputime": 5.0}}, # 12 + {1: {"cputime": 9.0}, 2: {"cputime": 6.0}}, # 15 + ], + ) + agg = _agg([m], [fake]) + for _ in range(3): + agg.add_sample() + record = agg.report() + assert record is not None + # First window seeds delta from 0, so the whole cumulative total is counted. + assert record["measurements"]["ps_cpu_seconds"] == 15.0 + + +def test_delta_seed_crosses_report_boundary() -> None: + m = ALL_MEASUREMENTS["ps_cpu_seconds"] + fake = FakeCollector( + "ps", + [ + {1: {"cputime": 10.0}}, # window 1 sample 1 + {1: {"cputime": 13.0}}, # window 1 sample 2 -> total 13 + {1: {"cputime": 18.0}}, # window 2 sample 1 -> delta 18-13 = 5 + ], + ) + agg = _agg([m], [fake]) + agg.add_sample() + agg.add_sample() + r1 = agg.report() + assert r1 is not None and r1["measurements"]["ps_cpu_seconds"] == 13.0 + agg.add_sample() + r2 = agg.report() + assert r2 is not None and r2["measurements"]["ps_cpu_seconds"] == 5.0 + + +def test_delta_clamps_on_pid_churn() -> None: + m = ALL_MEASUREMENTS["ps_cpu_seconds"] + # A pid with large cputime exits between reports, so the summed cumulative + # total drops below the carried seed and the window delta would go negative. + fake = FakeCollector( + "ps", + [ + {1: {"cputime": 5.0}, 2: {"cputime": 100.0}}, # report 1: total 105 + {1: {"cputime": 6.0}}, # report 2: pid 2 gone -> total 6 (< seed 105) + ], + ) + agg = _agg([m], [fake]) + agg.add_sample() + r1 = agg.report() + assert r1 is not None and r1["measurements"]["ps_cpu_seconds"] == 105.0 + agg.add_sample() + r2 = agg.report() + assert r2 is not None + assert r2["measurements"]["ps_cpu_seconds"] == 0.0 # clamped, not negative + + +def test_per_pid_rate_uses_monotonic_dt(monkeypatch: pytest.MonkeyPatch) -> None: + m = ALL_MEASUREMENTS["ps_pdcpu"] + fake = FakeCollector( + "ps", + [ + {1: {"cputime": 0.0}}, + {1: {"cputime": 2.0}}, # +2 cputime + {1: {"cputime": 3.0}}, # +1 cputime + ], + ) + # Drive monotonic clock at a fixed 1s cadence: rates are 2.0 then 1.0. + ticks = iter([100.0, 101.0, 102.0]) + monkeypatch.setattr("con_duct._aggregation.time.monotonic", lambda: next(ticks)) + agg = _agg([m], [fake]) + for _ in range(3): + agg.add_sample() + record = agg.report() + assert record is not None + # reduce=max over per-sample rates -> 2.0 (cputime-seconds per wall second). + assert record["measurements"]["ps_pdcpu"] == {"1": 2.0} + + +def test_rate_seed_crosses_report_boundary(monkeypatch: pytest.MonkeyPatch) -> None: + m = ALL_MEASUREMENTS["ps_pdcpu"] + fake = FakeCollector( + "ps", + [ + {1: {"cputime": 0.0}}, # w1 s1 + {1: {"cputime": 1.0}}, # w1 s2 rate 1.0 + {1: {"cputime": 5.0}}, # w2 s1 rate (5-1)/1 = 4.0 via seed + ], + ) + ticks = iter([10.0, 11.0, 12.0]) + monkeypatch.setattr("con_duct._aggregation.time.monotonic", lambda: next(ticks)) + agg = _agg([m], [fake]) + agg.add_sample() + agg.add_sample() + r1 = agg.report() + assert r1 is not None and r1["measurements"]["ps_pdcpu"] == {"1": 1.0} + agg.add_sample() + r2 = agg.report() + # Without the seed there is only one point in window 2 and no rate; the + # carried seed lets the boundary-spanning rate be recovered. + assert r2 is not None and r2["measurements"]["ps_pdcpu"] == {"1": 4.0} + + +def test_single_collector_read_once_per_report() -> None: + m = ALL_MEASUREMENTS["cgroup_rss_peak"] + ps = FakeCollector( + "ps", [{1: {"rss": 1.0}}] + ) # provides a sample so window != empty + cg = FakeSingleCollector("cgroup", 4096.0) + agg = _agg([ALL_MEASUREMENTS["ps_rss"], m], [ps, cg]) + agg.add_sample() + record = agg.report() + assert record is not None + assert record["measurements"]["cgroup_rss_peak"] == 4096.0 + + +def test_empty_window_returns_none() -> None: + agg = _agg([ALL_MEASUREMENTS["ps_rss"]], [FakeCollector("ps", [])]) + assert agg.report() is None + + +def test_empty_ps_reading_drops_sample() -> None: + # ps returning no pids (session gone) must not buffer or count the sample. + fake = FakeCollector("ps", [{}, {1: {"rss": 5.0}}]) + agg = _agg([ALL_MEASUREMENTS["ps_rss"]], [fake]) + assert agg.add_sample() is False + assert agg.num_samples == 0 + assert agg.add_sample() is True + assert agg.num_samples == 1 + record = agg.report() + assert record is not None and record["measurements"]["ps_rss"] == {"1": 5.0} + + +def test_non_ps_only_selection_still_records() -> None: + # Selecting only a single-read collector (no ps liveness signal) still + # buffers samples so reports get written. + cg = FakeSingleCollector("cgroup", 4096.0) + agg = _agg([ALL_MEASUREMENTS["cgroup_rss_peak"]], [cg]) + assert agg.add_sample() is True + record = agg.report() + assert record is not None + assert record["measurements"]["cgroup_rss_peak"] == 4096.0 + + +def test_summary_accumulates_across_reports() -> None: + ms = [ALL_MEASUREMENTS["ps_rss_total"], ALL_MEASUREMENTS["ps_cpu_seconds"]] + fake = FakeCollector( + "ps", + [ + {1: {"rss": 100.0, "cputime": 2.0}}, + {1: {"rss": 200.0, "cputime": 5.0}}, # report 1 + {1: {"rss": 150.0, "cputime": 9.0}}, # report 2 + ], + ) + agg = _agg(ms, [fake]) + agg.add_sample() + agg.add_sample() + agg.report() + agg.add_sample() + agg.report() + summary = agg.summary() + assert summary["peak_ps_rss_total"] == 200.0 # run peak across both reports + assert summary["ps_cpu_seconds"] == 9.0 # total cputime over the run + assert agg.num_reports == 2 + assert agg.num_samples == 3 + # Optional collectors absent -> keys present but null (stable schema). + assert summary["peak_cgroup_rss_peak"] is None + assert summary["peak_psutil_pss_total"] is None diff --git a/test/duct_main/test_collectors.py b/test/duct_main/test_collectors.py new file mode 100644 index 00000000..5e1b67c2 --- /dev/null +++ b/test/duct_main/test_collectors.py @@ -0,0 +1,83 @@ +"""Unit tests for the collector / measurement registry and live collectors.""" + +from __future__ import annotations +import os +from pathlib import Path +import pytest +from con_duct._collectors import ( + ALL_MEASUREMENTS, + CgroupCollector, + Derive, + PsCollector, + PsutilCollector, + Reduce, + Scope, + UnavailableMeasurementError, + UnknownMeasurementError, + available_keys, + resolve_selection, +) + +SID = os.getsid(0) + + +def test_ps_keys_match_design() -> None: + names = {m.name for m in PsCollector.measurements} + assert {"ps_rss", "ps_rss_total", "ps_pdcpu", "ps_cpu_seconds"} <= names + # ps_pdcpu derives a rate; ps_cpu_seconds is a delta counter. + by_name = {m.name: m for m in PsCollector.measurements} + assert by_name["ps_pdcpu"].derive is Derive.RATE + assert by_name["ps_pdcpu"].scope is Scope.PER_PID + assert by_name["ps_cpu_seconds"].reduce is Reduce.DELTA + assert by_name["ps_cpu_seconds"].scope is Scope.SINGLE + assert by_name["ps_rss_total"].scope is Scope.SINGLE + + +def test_ps_collect_live() -> None: + """ps always available; collecting our own session yields readings.""" + assert PsCollector.available() + reading = PsCollector(SID).collect() + assert reading, "expected at least our own process" + fields = next(iter(reading.values())) + assert set(fields) == {"rss", "cputime", "cmd"} + assert isinstance(fields["rss"], int) and fields["rss"] > 0 + assert isinstance(fields["cputime"], float) and fields["cputime"] >= 0.0 + + +def test_cgroup_refuses_when_missing(tmp_path: Path) -> None: + missing = CgroupCollector(peak_path=tmp_path / "does_not_exist") + assert missing.available() is False + + +def test_cgroup_reads_value(tmp_path: Path) -> None: + peak = tmp_path / "memory.peak" + peak.write_text("4096\n") + cg = CgroupCollector(peak_path=peak) + assert cg.available() is True + assert cg.read() == {"mem_peak": 4096.0} + + +def test_psutil_optional_keys_are_marked() -> None: + assert all(m.optional for m in PsutilCollector.measurements) + + +def test_resolve_none_returns_available() -> None: + resolved = resolve_selection(None, SID) + assert [m.name for m in resolved] == available_keys(SID) + + +def test_resolve_unknown_key_raises() -> None: + with pytest.raises(UnknownMeasurementError): + resolve_selection(["not_a_real_key"], SID) + + +def test_resolve_unavailable_psutil_message(monkeypatch: pytest.MonkeyPatch) -> None: + # Force psutil unavailable and confirm a clean, actionable error. + monkeypatch.setattr(PsutilCollector, "available", staticmethod(lambda: False)) + with pytest.raises(UnavailableMeasurementError, match="psutil"): + resolve_selection(["psutil_pss"], SID) + + +def test_all_measurements_have_unique_names() -> None: + names = list(ALL_MEASUREMENTS) + assert len(names) == len(set(names)) diff --git a/test/duct_main/test_e2e.py b/test/duct_main/test_e2e.py index 62aefc39..6ffba73a 100644 --- a/test/duct_main/test_e2e.py +++ b/test/duct_main/test_e2e.py @@ -46,12 +46,12 @@ def test_spawn_children( with open(f"{duct_prefix}{SUFFIXES['usage']}") as usage_file: all_samples = [json.loads(line) for line in usage_file] - # Only count the child sleep processes + # Only count the child sleep processes (ps_cmd is a per-pid command label). all_child_pids = set( pid for sample in all_samples - for pid, proc in sample["processes"].items() - if "sleep" in proc["cmd"] + for pid, cmd in sample["measurements"].get("ps_cmd", {}).items() + if "sleep" in cmd ) # Add one pid for the hold-the-door process, see spawn_children.sh line 7 if mode == "setsid": @@ -84,8 +84,8 @@ def test_session_modes(temp_output_dir: str, duct_cmd: str, session_mode: str) - # Validate sample structure for sample in samples: assert "timestamp" in sample - assert "processes" in sample - assert "totals" in sample + assert "num_samples" in sample + assert "measurements" in sample # Read and validate info data with open(info_file) as f: @@ -135,16 +135,16 @@ def test_session_mode_behavior_difference(temp_output_dir: str, duct_cmd: str) - # Check for our unique background process new_session_has_marker = any( any( - "DUCT_TEST_MARKER" in str(proc.get("cmd", "")) - for proc in sample["processes"].values() + "DUCT_TEST_MARKER" in str(cmd) + for cmd in sample["measurements"].get("ps_cmd", {}).values() ) for sample in new_session_samples ) current_session_has_marker = any( any( - "DUCT_TEST_MARKER" in str(proc.get("cmd", "")) - for proc in sample["processes"].values() + "DUCT_TEST_MARKER" in str(cmd) + for cmd in sample["measurements"].get("ps_cmd", {}).values() ) for sample in current_session_samples ) diff --git a/test/duct_main/test_execution.py b/test/duct_main/test_execution.py index 72745c97..f94fdf6e 100644 --- a/test/duct_main/test_execution.py +++ b/test/duct_main/test_execution.py @@ -95,10 +95,10 @@ def test_execution_summary( info_dict = json.loads(info.read()) execution_summary = info_dict["execution_summary"] # Since resources used should be small lets make sure values are roughly sane - assert execution_summary["average_pmem"] < 10 - assert execution_summary["peak_pmem"] < 10 - assert execution_summary["average_pcpu"] < 10 - assert execution_summary["peak_pcpu"] < 10 + assert execution_summary["peak_ps_rss_total"] > 0 + assert execution_summary["ave_ps_rss_total"] > 0 + # A brief sleep accrues little/no whole-second ps cputime. + assert execution_summary["ps_cpu_seconds"] >= 0 assert execution_summary["exit_code"] == 0 assert execution_summary["working_directory"] == os.getcwd() diff --git a/test/duct_main/test_report.py b/test/duct_main/test_report.py index c4cb9206..4155a649 100644 --- a/test/duct_main/test_report.py +++ b/test/duct_main/test_report.py @@ -1,218 +1,10 @@ from __future__ import annotations -from collections import Counter -from copy import deepcopy -from datetime import datetime import os import subprocess from unittest import mock -import pytest from con_duct._duct_main import EXECUTION_SUMMARY_FORMAT -from con_duct._models import Averages, ProcessStats, Sample from con_duct._tracker import Report -stat0 = ProcessStats( - pcpu=0.0, - pmem=0, - rss=0, - vsz=0, - timestamp="2024-06-11T10:09:37-04:00", - etime="00:00", - cmd="cmd 1", - stat=Counter(["stat0"]), -) - -stat1 = ProcessStats( - pcpu=1.0, - pmem=0, - rss=0, - vsz=0, - timestamp="2024-06-11T10:13:23-04:00", - etime="00:02", - cmd="cmd 1", - stat=Counter(["stat1"]), -) - -stat2 = ProcessStats( - pcpu=1.1, - pmem=1.1, - rss=11, - vsz=11, - timestamp="2024-06-11T10:13:23-04:00", - etime="00:02", - cmd="cmd 1", - stat=Counter(["stat2"]), -) - - -def test_sample_max_initial_values_one_pid() -> None: - maxes = Sample() - ex0 = Sample() - ex0.add_pid(1, deepcopy(stat0)) - maxes = maxes.aggregate(ex0) - assert maxes.stats == {1: stat0} - - -def test_sample_max_one_pid() -> None: - maxes = Sample() - maxes.add_pid(1, deepcopy(stat0)) - ex1 = Sample() - ex1.add_pid(1, deepcopy(stat1)) - maxes = maxes.aggregate(ex1) - assert maxes.stats[1].rss == stat1.rss - assert maxes.stats[1].vsz == stat1.vsz - assert maxes.stats[1].pmem == stat1.pmem - assert maxes.stats[1].pcpu == stat1.pcpu - - -def test_sample_max_initial_values_two_pids() -> None: - maxes = Sample() - ex0 = Sample() - ex0.add_pid(1, deepcopy(stat0)) - ex0.add_pid(2, deepcopy(stat0)) - maxes = maxes.aggregate(ex0) - assert maxes.stats == {1: stat0, 2: stat0} - assert maxes.stats == {1: stat0, 2: stat0} - - -def test_sample_aggregate_two_pids() -> None: - maxes = Sample() - maxes.add_pid(1, deepcopy(stat0)) - maxes.add_pid(2, deepcopy(stat0)) - assert maxes.stats[1].stat["stat0"] == 1 - assert maxes.stats[2].stat["stat0"] == 1 - assert maxes.stats[1].stat["stat1"] == 0 - assert maxes.stats[2].stat["stat1"] == 0 - ex1 = Sample() - ex1.add_pid(1, deepcopy(stat1)) - maxes = maxes.aggregate(ex1) - assert maxes.stats[1].stat["stat0"] == 1 - assert maxes.stats[2].stat["stat0"] == 1 - assert maxes.stats[1].stat["stat1"] == 1 - assert maxes.stats[2].stat["stat1"] == 0 - ex2 = Sample() - ex2.add_pid(2, deepcopy(stat1)) - maxes = maxes.aggregate(ex2) - # Check the `stat` counts one of each for both pids - assert maxes.stats[1].stat["stat0"] == 1 - assert maxes.stats[2].stat["stat0"] == 1 - assert maxes.stats[1].stat["stat1"] == 1 - assert maxes.stats[2].stat["stat1"] == 1 - - # Each stat1 value > stat0 value - assert maxes.stats[1].pcpu == stat1.pcpu - assert maxes.stats[1].pmem == stat1.pmem - assert maxes.stats[1].rss == stat1.rss - assert maxes.stats[1].vsz == stat1.vsz - assert maxes.stats[2].pcpu == stat1.pcpu - assert maxes.stats[2].pmem == stat1.pmem - assert maxes.stats[2].rss == stat1.rss - assert maxes.stats[2].vsz == stat1.vsz - - -def test_average_no_samples() -> None: - averages = Averages() - assert averages.num_samples == 0 - sample = Sample() - sample.averages = averages - serialized = sample.for_json() - assert "averages" in serialized - assert not serialized["averages"] - - -def test_averages_one_sample() -> None: - sample = Sample() - sample.add_pid(1, deepcopy(stat0)) - averages = Averages.from_sample(sample) - assert averages.rss == sample.total_rss - assert averages.vsz == sample.total_vsz - assert averages.pmem == sample.total_pmem - assert averages.pcpu == sample.total_pcpu - assert averages.num_samples == 1 - - -def test_averages_two_samples() -> None: - sample = Sample() - sample.add_pid(1, deepcopy(stat0)) - averages = Averages.from_sample(sample) - sample2 = Sample() - sample2.add_pid(2, deepcopy(stat1)) - averages.update(sample2) - assert averages.pcpu == (stat0.pcpu + stat1.pcpu) / 2 - - -def test_averages_three_samples() -> None: - sample = Sample() - sample.add_pid(1, deepcopy(stat0)) - averages = Averages.from_sample(sample) - sample2 = Sample() - sample2.add_pid(2, deepcopy(stat1)) - averages.update(sample2) - averages.update(sample2) - assert averages.pcpu == (stat0.pcpu + (2 * stat1.pcpu)) / 3 - - -def test_sample_totals() -> None: - sample = Sample() - sample.add_pid(1, deepcopy(stat2)) - sample.add_pid(2, deepcopy(stat2)) - assert sample.total_rss == stat2.rss * 2 - assert sample.total_vsz == stat2.vsz * 2 - assert sample.total_pmem == stat2.pmem * 2 - assert sample.total_pcpu == stat2.pcpu * 2 - - -@pytest.mark.parametrize( - "pcpu, pmem, rss, vsz, etime, cmd", - [ - (1.0, 1.1, 1024, 1025, "00:00", "cmd"), - (0.5, 0.7, 20.48, 40.96, "00:01", "any"), - (1, 2, 3, 4, "100:1000", "string"), - (0, 0.0, 0, 0.0, "999:999:999", "can have spaces"), - (2.5, 3.5, 8192, 16384, "any", "for --this --kind of thing"), - (100.0, 99.9, 65536, 131072, "string", "cmd"), - ], -) -def test_process_stats_green( - pcpu: float, pmem: float, rss: int, vsz: int, etime: str, cmd: str -) -> None: - # Assert does not raise - ProcessStats( - pcpu=pcpu, - pmem=pmem, - rss=rss, - vsz=vsz, - timestamp=datetime.now().astimezone().isoformat(), - etime=etime, - cmd=cmd, - stat=Counter(["stat0"]), - ) - - -@pytest.mark.parametrize( - "pcpu, pmem, rss, vsz, etime, cmd", - [ - ("only", 1.1, 1024, 1025, "etime", "cmd"), - (0.5, "takes", 20.48, 40.96, "some", "str"), - (1, 2, "one", 4, "anything", "accepted"), - (1, 2, 3, "value", "etime", "cmd"), - ("2", "fail", "or", "more", "etime", "cmd"), - ], -) -def test_process_stats_red( - pcpu: float, pmem: float, rss: int, vsz: int, etime: str, cmd: str -) -> None: - with pytest.raises(AssertionError): - ProcessStats( - pcpu=pcpu, - pmem=pmem, - rss=rss, - vsz=vsz, - timestamp=datetime.now().astimezone().isoformat(), - etime=etime, - cmd=cmd, - stat=Counter(["stat0"]), - ) - @mock.patch("con_duct._tracker.LogPaths") def test_system_info_sanity(mock_log_paths: mock.MagicMock) -> None: @@ -230,6 +22,27 @@ def test_system_info_sanity(mock_log_paths: mock.MagicMock) -> None: assert report.system_info.user == os.environ.get("USER") +def test_execution_summary_keys_stable_without_run() -> None: + """Even with no monitoring, the measurement summary keys are present (None) + so info.json schema is environment-independent.""" + cwd = os.getcwd() + report = Report( + "_cmd", [], mock.MagicMock(), EXECUTION_SUMMARY_FORMAT, cwd, clobber=False + ) + summary = report.execution_summary + for key in ( + "peak_ps_rss_total", + "ave_ps_rss_total", + "ps_cpu_seconds", + "peak_cgroup_rss_peak", + "peak_psutil_pss_total", + ): + assert key in summary + assert summary[key] is None + assert summary["num_samples"] == 0 + assert summary["num_reports"] == 0 + + @mock.patch("con_duct._tracker.shutil.which") @mock.patch("con_duct._tracker.subprocess.check_output") @mock.patch("con_duct._tracker.LogPaths") diff --git a/test/test_windows.py b/test/test_windows.py index 5352e3f0..ed5d2a6c 100644 --- a/test/test_windows.py +++ b/test/test_windows.py @@ -7,9 +7,9 @@ @pytest.fixture -def _reload_sampling() -> Generator: - """Ensure _sampling module is restored after tests that reload it.""" - import con_duct._sampling as mod +def _reload_collectors() -> Generator: + """Ensure _collectors module is restored after tests that reload it.""" + import con_duct._collectors as mod yield if hasattr(sys, "tracebacklimit"): @@ -18,9 +18,9 @@ def _reload_sampling() -> Generator: @mock.patch("platform.system", return_value="Windows") -@pytest.mark.usefixtures("_reload_sampling") +@pytest.mark.usefixtures("_reload_collectors") def test_unsupported_system_raises(_mock_system: mock.MagicMock) -> None: - import con_duct._sampling as mod + import con_duct._collectors as mod with pytest.raises( NotImplementedError, match="does not currently support.*Windows"