diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 758d11616..12f9dee2c 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -88,7 +88,7 @@ "name": "ci", "source": "./plugins/ci", "description": "A plugin to work with OpenShift CI and analyze Prow job results", - "version": "0.0.71", + "version": "0.0.72", "category": "ci", "keywords": [ "prow", diff --git a/docs/index.html b/docs/index.html index 9b44d6ff2..82e2b4010 100644 --- a/docs/index.html +++ b/docs/index.html @@ -541,7 +541,7 @@

ai-helpers

{ "name": "ci", "description": "Tools for working with OpenShift CI and analyzing Prow job results", - "version": "0.0.71", + "version": "0.0.72", "has_readme": true, "commands": [ { diff --git a/plugins/ci/.claude-plugin/plugin.json b/plugins/ci/.claude-plugin/plugin.json index f6f69c424..3b6cf30e8 100644 --- a/plugins/ci/.claude-plugin/plugin.json +++ b/plugins/ci/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "ci", "description": "Tools for working with OpenShift CI and analyzing Prow job results", - "version": "0.0.71", + "version": "0.0.72", "author": { "name": "github.com/openshift-eng" } diff --git a/plugins/ci/skills/payload-analysis/SKILL.md b/plugins/ci/skills/payload-analysis/SKILL.md index 2b0da9cdd..d1d147e9b 100644 --- a/plugins/ci/skills/payload-analysis/SKILL.md +++ b/plugins/ci/skills/payload-analysis/SKILL.md @@ -106,6 +106,22 @@ From `summary.json` top-level fields: **Record `phase` verbatim** from the `summary.json` metadata (`Accepted`, `Rejected`, or `Ready`). Never infer the phase from the job results or from whether failures exist — a payload can be `Accepted` *with* blocking failures (force-accepted) or `Ready` while jobs are still running. The stored phase drives the force-accept decision (Step 6.4) and the executive summary (Step 7.1), so an inferred phase silently corrupts both. +#### 3.1b: If the Snapshot Is Incomplete, Collect the Data Yourself + +Check `summary.json` → `data_complete`. An absent `test_failure_count` means +*unknown*, not zero — never conclude a job had no test failures, and therefore +failed for some other reason, from missing data. + +When data is missing, collect it yourself from the job's `gcs_url` artifacts +rather than analyzing around the gap. Do the same for any payload in the chain +whose per-test data is missing. Report a gap as a limitation only when the +artifacts themselves are unreachable. + +An aggregated job with no per-test results at all is **unclassified**, not part +of a regression streak — aggregation also fails when too few child runs +completed or infrastructure killed them. Check the child runs: one that died +before the test phase cannot have failed a test. + #### 3.2: Failed Blocking Jobs From `summary.json` → `blocking_jobs.failed_jobs[]`, each entry contains: @@ -123,7 +139,7 @@ For each failed job's `streak.originating_payload`, find the matching entry in ` - `url`, `component`, `number`, `description` - Paths to local artifacts: `diff`, `comments`, `jobs` -These PRs are the **candidates** for failures that started in that originating payload. +Treat this as a **preliminary** list only. The job-level streak merges unrelated failure modes, so its originating payload is frequently earlier than the regression being investigated — and candidates gathered from it can omit the causal PR entirely. Before scoring, re-derive the originating payload **per failure mode** from `test_failures.blocking[].first_failed_in` (Step 5) and collect the candidates from *that* payload. #### 3.4: Test Failure Details @@ -291,6 +307,12 @@ Compare the subagent's root cause analysis for the target payload against previo If a job fails in two consecutive payloads but for **different reasons**, treat each as a separate streak=1 failure with its own originating payload and candidate PRs. Re-split the streak and re-assign originating payloads before proceeding to scoring. +**The job-level streak is not a failure mode's originating payload.** `streak.originating_payload` tracks when the *job* started failing, which merges unrelated modes — an infrastructure blip, a flake, and a real regression all read as one streak. Scoring candidates from a payload that predates the actual regression guarantees misattribution: the causal PR is not even in the candidate set. + +For each failure mode, take the originating payload from the matching `test_failures.blocking[]` entry's **`first_failed_in`**. When that is later than the job-level streak's, the job's earlier failures are a different mode — score from `first_failed_in`. Confirm the test passed in the preceding payload; where that payload has no per-test data, check its child runs rather than assuming it was failing. + +Establish this before enumerating candidate PRs (Step 6.1). When the two onsets differ, record both and state which drove scoring. + ### Step 5b: Adjudicate Conflicting Root Causes When two or more investigations reach **contradictory root causes for the same failure signature** (same test, same operation, or same error class — across jobs, across retries, or between a subagent and a previous analysis), the analysis is **UNRESOLVED**. It is *not* a tie to be broken by whichever explanation feels more plausible. Resolve it only with discriminating evidence, applying these rules: diff --git a/plugins/ci/skills/payload-snapshot/SKILL.md b/plugins/ci/skills/payload-snapshot/SKILL.md index c3d276650..4a13372e7 100644 --- a/plugins/ci/skills/payload-snapshot/SKILL.md +++ b/plugins/ci/skills/payload-snapshot/SKILL.md @@ -28,8 +28,10 @@ Use this skill when you need to: 3. **Google Cloud SDK (`gcloud`)** — for JUnit test result download - Install: `brew install google-cloud-sdk` (macOS) or see https://cloud.google.com/sdk - - Authenticate: `gcloud auth login` - - Without `gcloud`, JUnit data is skipped; job directories still created + - **Authentication is not required** — the CI artifact buckets are public and + are read anonymously when no account is configured + - Without `gcloud` entirely, JUnit data is skipped and the snapshot is + reported as incomplete (see [Data completeness](#data-completeness)) 4. **Container runtime (`podman`)** — for RHCOS RPMDB extraction - Install: available in most Linux distributions; `brew install podman` on macOS @@ -165,6 +167,7 @@ Options: --workers N Parallel workers for API calls (default: 8) --no-junit Skip JUnit download and regression tracking --no-rpmdb Skip RHCOS RPMDB extraction + --fail-on-incomplete Exit 1 if any requested data could not be collected ``` ## Output Files @@ -183,6 +186,50 @@ Comprehensive stream-level triage data — start here. Contains: - `test_failures.blocking[]` — `test_name`, `jobs`, `first_failed_in`, `payloads_failing`, `failure_message`, `failure_text` (full, not truncated) - `payloads[]` — per-payload entries with `tag`, `phase`, relative file paths, `prs[]` with component/diff/comments paths, and `rhcos_changes[]` with RPM diffs per RHCOS variant - `rhcos_rpms[]` — RPMDB metadata for the target payload's RHCOS variants: `tag`, `name`, `pullspec`, `rpmdb` (relative path to rpmdb.sqlite) +- `data_complete` — `true` when all requested data was ultimately collected, including via a fallback after an initial read failed. `false` means some requested data could not be read at all. +- `collection_errors[]` — every read failure encountered. Each entry has `reason`, `command`, and optionally `detail`, `stage`, `job`, `payload_tag`, `recovered`. Reasons: `auth`, `timeout`, `gcloud_missing`, `command_failed`, `junit_unavailable` (nothing readable), `junit_missing` (nothing discovered), `junit_unparseable` (corrupt XML), `junit_partial` (some files unread), `build_log_unavailable`. + - `recovered: true` means a fallback subsequently obtained the data. These entries are diagnostic only (useful for spotting a timeout that needs tuning) and do **not** make `data_complete` false. + - `data_complete` is `false` only when at least one error was **not** recovered. + + +#### gcloud credentials are not required + +The CI artifact buckets are public. When gcloud has no active account it is +run in anonymous mode automatically, so an unauthenticated environment still +produces a complete snapshot. Authenticate only if you also need private +buckets. + +#### Missing data is absent, never empty + +When a collection step fails, the affected file is **not written** and the +corresponding summary field is **omitted** — it is never emitted as an empty +list or a zero count. Specifically: + +- If JUnit could not be read for a job, `results.json` is not created, the + job entry has **no** `test_failure_count` and `junit_results`, and instead + carries `junit_collection_failed: true`. +- An absent `test_failure_count` therefore means *unknown*, whereas `0` means + *verified clean*. + +- If **some** JUnit files were read and others were not, `results.json` holds + the real results that were obtained and the job entry adds + `junit_collection_partial: true`. `test_failure_count` is then a **lower + bound**, not a total. +- A failed job with **no** readable JUnit at all — none discovered + (`junit_missing`), or nothing that parsed (`junit_unparseable`) — is left + without a count entirely. A job whose tests may never have run is unknown, + not clean. A parse failure alongside other readable files is the partial + case above, not this one. + +Consumers **must** distinguish these. Treating unreadable data as "no test +failures" makes a broken job look like it failed for some other reason, which +misdirects root-cause analysis. Check `data_complete` before drawing any +conclusion from an absence of test failures. Use `--fail-on-incomplete` in +automation to exit non-zero rather than emit a partial snapshot. + +State is persisted to `collection_errors.json` beside `summary.json`, so a +later process can tell that an existing `results.json` came from an incomplete +read. ### `AGENTS.md` / `CLAUDE.md` @@ -268,9 +315,14 @@ Aggregated jobs run the same underlying test multiple times with statistical ana - **Tag not found**: Exits with code 2 and a descriptive error - **Release controller unreachable**: Exits with code 1 - **`gh` not authenticated**: Prints a warning and continues without PR data -- **`gcloud` not available**: Prints a warning and skips JUnit download -- **Individual job/PR fetch failure**: Logs a warning and continues -- **Idempotent**: Re-running skips files that already exist +- **`gcloud` not available**: Warns, skips JUnit download, and records + `gcloud_missing` so the snapshot is reported incomplete +- **`gcloud` not authenticated**: Reads the public buckets anonymously +- **Individual job/PR fetch failure**: Logs a warning, records a collection + error, and continues +- **Idempotent**: Re-running skips files that already exist — except JUnit + output a previous run recorded as incomplete, which is discarded and + re-collected so a partial snapshot cannot be inherited as complete ## Notes diff --git a/plugins/ci/skills/payload-snapshot/scripts/payload_snapshot.py b/plugins/ci/skills/payload-snapshot/scripts/payload_snapshot.py index 42941a460..85fb9282a 100755 --- a/plugins/ci/skills/payload-snapshot/scripts/payload_snapshot.py +++ b/plugins/ci/skills/payload-snapshot/scripts/payload_snapshot.py @@ -17,6 +17,7 @@ import subprocess import sys import tempfile +import threading import time import urllib.error import urllib.parse @@ -716,7 +717,48 @@ def collect(self) -> bool: os.makedirs(self.output_dir, exist_ok=True) all_results: list[_TestResult] = [] - junit_files = self._list_junit_files() + with _error_scope() as own_errors: + junit_files = self._list_junit_files() + downloaded, failed_downloads = self._download_and_parse( + junit_files, all_results + ) + gcloud_failed = any( + not e.get("recovered") for e in own_errors + ) + + # Nothing was discovered at all. These collectors only run for + # jobs that FAILED, so "no JUnit anywhere" is not evidence of a + # clean run — it is an unknown. Publishing [] here would report a + # verified zero for a job whose tests may never have run. + if not junit_files: + _record_collection_error( + "junit_missing", + ["gcloud", "storage", "ls", ""], + detail=( + "no JUnit artifacts discovered for a failed job; " + "test results are unknown, not zero" + ), + stage="junit", + job=self.job.name, + payload_tag=self.payload_tag, + ) + _log( + f" ERROR: no JUnit discovered for {self.job.name} — " + f"leaving results.json absent (unknown, not zero)" + ) + return False + + return self._publish( + junit_files, all_results, downloaded, failed_downloads, + gcloud_failed, + ) + + def _download_and_parse( + self, junit_files: list[str], all_results: list + ) -> tuple[int, int]: + """Fetch each JUnit file and parse it; count successes/failures.""" + downloaded = 0 + failed_downloads = 0 for gcs_uri in junit_files: filename = os.path.basename(gcs_uri) local_path = os.path.join(self.output_dir, filename) @@ -730,7 +772,72 @@ def collect(self) -> bool: if os.path.exists(local_path): results = _parse_junit_xml(local_path, source_name=filename) + if results is None: + # Downloaded but unparseable. Treated as unread, not as + # "no failures" — corrupt XML must never become a zero. + failed_downloads += 1 + _record_collection_error( + "junit_unparseable", + ["parse", filename], + detail=f"{filename} is not valid JUnit XML", + stage="junit", + job=self.job.name, + payload_tag=self.payload_tag, + ) + continue + downloaded += 1 all_results.extend(results) + else: + failed_downloads += 1 + return downloaded, failed_downloads + + def _publish( + self, junit_files: list[str], all_results: list, + downloaded: int, failed_downloads: int, gcloud_failed: bool, + ) -> bool: + """Write results.json unless doing so would misreport the data.""" + # Refuse to publish an empty result set that was caused by a read + # failure. Writing `[]` here makes the job look clean, and + # downstream consumers report "0 test failures" for a job that + # actually failed. Leaving results.json absent means "unknown", + # which the summary and the analysis skill treat as such. + if gcloud_failed and downloaded == 0: + _record_collection_error( + "junit_unavailable", + ["gcloud", "storage", "cat", ""], + detail=( + "no JUnit XML could be read; results.json intentionally " + "not written so this is not mistaken for zero failures" + ), + stage="junit", + job=self.job.name, + payload_tag=self.payload_tag, + ) + _log( + f" ERROR: could not read JUnit for {self.job.name} — " + f"leaving results.json absent (unknown, not zero)" + ) + return False + + # Some files read, others not. The results are real but partial, so + # publish them and mark them partial — a count derived from half the + # JUnit is not authoritative and must not read as one that is. + if failed_downloads: + _record_collection_error( + "junit_partial", + ["gcloud", "storage", "cat", ""], + detail=( + f"{failed_downloads} of {len(junit_files)} JUnit file(s) " + f"could not be read; test_failure_count is a lower bound" + ), + stage="junit", + job=self.job.name, + payload_tag=self.payload_tag, + ) + _log( + f" WARNING: partial JUnit for {self.job.name} — " + f"{failed_downloads}/{len(junit_files)} file(s) unread" + ) failures = _test_results_to_json(all_results) _write_json(self.output_path, failures) @@ -748,18 +855,40 @@ def collect(self) -> bool: return True def _list_junit_files(self) -> list[str]: - """Discover JUnit XML files in GCS for this job.""" + """Discover JUnit XML files in GCS for this job. + + Tries a recursive glob first, then falls back to bounded-depth + probes. Jobs that emit very large artifact trees (Hypershift e2e + can produce 10,000+ cluster resource dumps) make the ``**`` glob + slow enough to time out, and a timeout here previously produced a + silently empty result set. + """ bucket_path = self.job.gcs_bucket_path base = f"gs://{bucket_path}" - output = _run_gcloud( - ["gcloud", "storage", "ls", f"{base}/artifacts/**/junit*.xml"], - timeout=30, - ) - if not output: - return [] + with _error_scope() as glob_errors: + output = _run_gcloud( + ["gcloud", "storage", "ls", + f"{base}/artifacts/**/junit*.xml"], + timeout=120, + ) + + files = [line.strip() for line in (output or "").strip().splitlines() + if line.strip()] + + if not files: + _log( + f" warn: recursive glob found no JUnit for {self.job.name} " + f"— trying bounded-depth fallback" + ) + files = self._list_junit_files_fallback(base) + if files: + # The glob's failure is recovered; any probe failures the + # fallback itself hit are NOT — they may be missing data. + _mark_errors_recovered(glob_errors) - files = [l.strip() for l in output.strip().splitlines() if l.strip()] + if not files: + return [] if self.job.is_aggregated: # For aggregated jobs, keep junit_operator.xml and the @@ -783,6 +912,71 @@ def _list_junit_files(self) -> list[str]: keep = operator + (preferred if preferred else others[:1]) return keep if keep else files + def _list_junit_files_fallback(self, base: str) -> list[str]: + """Targeted JUnit discovery that avoids a full recursive glob. + + Lists the top-level step directories under ``artifacts/`` and + probes each at the depths JUnit files actually appear, rather + than enumerating every object in the tree: + + - ``{step}/junit*.xml`` + - ``{step}/artifacts/junit*.xml`` + - ``{step}/*/artifacts/junit*.xml`` + + Aggregated jobs keep ``junit-aggregated.xml`` far deeper than + that, so the aggregator subtree gets its own scoped recursive + probe — scoped to one small directory, it stays fast. + """ + dir_output = _run_gcloud( + ["gcloud", "storage", "ls", f"{base}/artifacts/"], + timeout=60, + ) + if not dir_output: + return [] + + # Each line is a directory like gs://bucket/path/artifacts/step-name/ + step_dirs = [ + d.strip().rstrip("/").split("/")[-1] + for d in dir_output.strip().splitlines() + if d.strip() + ] + + # junit_operator.xml sits directly in artifacts/, not in a step dir. + found: list[str] = [] + top = _run_gcloud( + ["gcloud", "storage", "ls", f"{base}/artifacts/junit*.xml"], + timeout=60, + ) + if top: + found.extend(line.strip() for line in top.strip().splitlines() + if line.strip()) + + for step in step_dirs: + patterns = [ + f"{base}/artifacts/{step}/junit*.xml", + f"{base}/artifacts/{step}/artifacts/junit*.xml", + f"{base}/artifacts/{step}/*/artifacts/junit*.xml", + ] + # The aggregated report lives several levels below the + # aggregator step; a scoped ** over that one subtree is cheap. + if "aggregator" in step: + patterns.append(f"{base}/artifacts/{step}/**/junit*.xml") + + for pattern in patterns: + probe = _run_gcloud( + ["gcloud", "storage", "ls", pattern], timeout=60 + ) + if probe: + found.extend( + line.strip() for line in probe.strip().splitlines() + if line.strip() + ) + + found = sorted(set(found)) + if found: + _log(f" fallback found {len(found)} JUnit file(s)") + return found + def _detect_underlying_job_name(self) -> Optional[str]: """For aggregated jobs, extract the underlying job name from GCS paths.""" if not self.job.is_aggregated: @@ -826,10 +1020,19 @@ def _fetch(self) -> Optional[dict]: return None gcs_uri = f"gs://{self.job.gcs_bucket_path}/build-log.txt" + errors_before = _unrecovered_error_count() raw = _run_gcloud_bytes( ["gcloud", "storage", "cat", gcs_uri], timeout=120 ) if not raw: + if _unrecovered_error_count() > errors_before: + _record_collection_error( + "build_log_unavailable", + ["gcloud", "storage", "cat", gcs_uri], + detail="build-log.txt could not be read", + stage="build_log", + job=self.job.name, + ) return None try: @@ -1264,7 +1467,14 @@ def generate(self) -> None: for entry in target_rpms ] + # Surface any data that could not be collected. An empty snapshot + # section must never be silently indistinguishable from a clean one. + if _COLLECTION_ERRORS: + summary["collection_errors"] = list(_COLLECTION_ERRORS) + summary["data_complete"] = not _unrecovered_errors() + _write_json(os.path.join(self.base_dir, "summary.json"), summary) + _write_collection_state(self.base_dir) self._write_agents_md(summary) _log(" Generated summary.json, AGENTS.md, and CLAUDE.md") @@ -1316,6 +1526,25 @@ def _write_agents_md(self, summary: dict) -> None: "", ] + if not summary.get("data_complete", True): + errs = [e for e in summary.get("collection_errors", []) + if not e.get("recovered")] + reasons = sorted({e.get("reason", "unknown") for e in errs}) + lines.extend([ + "## ⚠️ INCOMPLETE SNAPSHOT", + "", + f"{len(errs)} collection error(s) occurred: " + f"{', '.join(reasons)}.", + "See `collection_errors` in `summary.json`.", + "", + "Data that could not be read is **absent, not empty**. A job", + "with `junit_collection_failed: true` and no", + "`test_failure_count` has an *unknown* number of test", + "failures — do NOT report it as zero, and do not conclude a", + "job failed for reasons other than its tests on this basis.", + "", + ]) + if failed_names: lines.append("### Failed blocking jobs") lines.append("") @@ -1469,6 +1698,18 @@ def _build_failed_job_details(self, lifecycle: str) -> list[dict]: ) results_data = _read_json(results_path) or [] entry["test_failure_count"] = len(results_data) + if _job_junit_state( + job_name, self.target_tag, "junit_partial" + ): + # Real results, but not all of them: the count is a + # lower bound, not an authoritative total. + entry["junit_collection_partial"] = True + elif _job_junit_state( + job_name, self.target_tag, "junit_unavailable" + ): + # Data could not be read. Do NOT emit test_failure_count — + # its absence means "unknown", where 0 would mean "clean". + entry["junit_collection_failed"] = True build_log_path = os.path.join( lifecycle_dir, job_name, "build_log.json" @@ -1621,6 +1862,16 @@ def run(self) -> None: ) os.makedirs(base_dir, exist_ok=True) + # A previous run over this directory may have left JUnit output it + # recorded as incomplete. Drop it so it is re-collected rather than + # inherited as trustworthy — but only when we will actually + # re-collect, otherwise we would delete the only copy and then + # report the result as complete. + if self.collect_junit: + _invalidate_suspect_junit(base_dir) + else: + _carry_forward_junit_errors(base_dir) + self._collect_streams(base_dir) chain = self._build_chain() @@ -2162,29 +2413,358 @@ def _check_gh_auth() -> bool: return False +# Records every gcloud failure that could hide real data. A "no match" +# result is normal (the object simply does not exist) and is NOT recorded. +# Anything else — a missing binary, a timeout, an auth failure — means the +# snapshot could not read data that may well exist, and callers must be able +# to tell that apart from "there was nothing to find". +_COLLECTION_ERRORS: list[dict] = [] +_COLLECTION_ERRORS_LOCK = threading.Lock() + +# Per-thread stack of "operation scopes". Every recorded error is appended +# to each active scope as well as the global ledger, so a caller can act on +# exactly the errors *its own* calls produced. Positional indexes into the +# global list are not safe here: collectors run concurrently, so another +# worker can append between one collector's start and end offsets. +_ERROR_SCOPES = threading.local() + +# stderr fragments that mean "the object does not exist", not "I failed". +_GCLOUD_NO_MATCH_PATTERNS = ( + "matched no objects", + "matched no such objects", + "not found", +) + +# stderr fragments that indicate a credentials/permission problem. +_GCLOUD_AUTH_PATTERNS = ( + "does not have storage.objects", + "anonymous caller", + "reauthentication", + "credentials", + "unauthorized", + "forbidden", + "403", + "401", +) + + +def _error_scope(): + """Collect errors recorded by this thread inside the ``with`` body.""" + return _ErrorScope() + + +class _ErrorScope: + """Context manager yielding the list of errors recorded inside it.""" + + def __init__(self) -> None: + self.entries: list[dict] = [] + + def __enter__(self) -> list[dict]: + stack = getattr(_ERROR_SCOPES, "stack", None) + if stack is None: + stack = [] + _ERROR_SCOPES.stack = stack + stack.append(self.entries) + return self.entries + + def __exit__(self, *exc) -> None: + getattr(_ERROR_SCOPES, "stack", []).pop() + return None + + +def _sanitize_detail(text: str) -> str: + """Make tool stderr safe to persist in a shareable summary. + + Strips control characters (which can spoof terminal output when the + summary is read back) and collapses whitespace. Diagnostics are + truncated: they exist to identify a failure class, not to carry full + tool output. + """ + cleaned = "".join( + ch if ch.isprintable() else " " for ch in (text or "") + ) + # These diagnostics are persisted into a summary this skill exists to + # hand to agents and share. gcloud auth errors routinely name the + # active account, and URLs can carry signed-request parameters; neither + # should leave the local environment. + cleaned = re.sub( + r"[\w.+-]+@[\w-]+\.[\w.-]+", "", cleaned + ) + cleaned = re.sub(r"\?\S+", "?", cleaned) + return re.sub(r"\s+", " ", cleaned).strip()[:300] + + +def _classify_gcloud_stderr(stderr: str) -> str: + """Classify a failed gcloud invocation from its stderr.""" + low = (stderr or "").lower() + if any(p in low for p in _GCLOUD_NO_MATCH_PATTERNS): + return "no_match" + if any(p in low for p in _GCLOUD_AUTH_PATTERNS): + return "auth" + return "command_failed" + + +def _record_collection_error( + reason: str, + command: list[str], + detail: str = "", + stage: str = "", + job: str = "", + payload_tag: str = "", +) -> None: + """Record a data-collection failure so it can be surfaced, not swallowed. + + Silently returning empty data makes "I could not read this" look + identical to "there is nothing here", which invites downstream + consumers to report zero failures for a job that actually failed. + """ + entry: dict = {"reason": reason, "command": " ".join(command[:4])} + if detail: + entry["detail"] = _sanitize_detail(detail) + if stage: + entry["stage"] = stage + if job: + entry["job"] = job + if payload_tag: + entry["payload_tag"] = payload_tag + with _COLLECTION_ERRORS_LOCK: + _COLLECTION_ERRORS.append(entry) + for scope in getattr(_ERROR_SCOPES, "stack", []): + scope.append(entry) + + +def _collection_error_count() -> int: + """Number of collection errors recorded so far.""" + return len(_COLLECTION_ERRORS) + + +COLLECTION_STATE_FILE = "collection_errors.json" + +# Reasons that make a job's persisted JUnit output untrustworthy. +_JUNIT_SUSPECT_REASONS = ( + "junit_unavailable", "junit_partial", "junit_missing", + "junit_unparseable", +) + + +def _is_within(base: str, target: str) -> bool: + """True when ``target`` resolves inside ``base``.""" + base_r = os.path.realpath(base) + target_r = os.path.realpath(target) + return target_r == base_r or target_r.startswith(base_r + os.sep) + + +def _safe_component(value: str) -> bool: + """True when ``value`` is usable as a single path component.""" + return bool(value) and value not in (".", "..") and not any( + sep in value for sep in (os.sep, "/", "\\") + ) + + +def _invalidate_suspect_junit(base_dir: str) -> int: + """Discard JUnit output that a previous run recorded as incomplete. + + Collection errors live in process memory but ``results.json`` persists, + and collectors skip any job whose output already exists. Without this, + re-running over the same directory regenerates the summary from an empty + ledger and promotes a partial snapshot to "complete". Dropping the + suspect output forces it to be re-collected and re-judged. + """ + state_path = os.path.join(base_dir, COLLECTION_STATE_FILE) + previous = _read_json(state_path) + if not previous: + return 0 + + invalidated = 0 + for entry in previous: + if entry.get("recovered"): + continue + if entry.get("reason") not in _JUNIT_SUSPECT_REASONS: + continue + tag, job = entry.get("payload_tag"), entry.get("job") + # The ledger is data read back from disk; never let it aim a + # recursive delete at a path outside the snapshot directory. + if not _safe_component(tag or "") or not _safe_component(job or ""): + _log(f" warn: ignoring unsafe collection-state entry " + f"(tag={tag!r}, job={job!r})") + continue + for lifecycle in ("blocking", "informing"): + junit_dir = os.path.join( + base_dir, tag, "jobs", lifecycle, job, "junit" + ) + if not _is_within(base_dir, junit_dir): + continue + if os.path.isdir(junit_dir): + shutil.rmtree(junit_dir, ignore_errors=True) + invalidated += 1 + if invalidated: + _log(f" re-collecting {invalidated} job(s) whose previous JUnit " + f"collection was incomplete") + return invalidated + + +def _carry_forward_junit_errors(base_dir: str) -> int: + """Re-record persisted JUnit failures when JUnit is not re-collected. + + With ``--no-junit`` nothing re-reads the artifacts, so a previous run's + partial or unreadable JUnit is still exactly as incomplete as it was. + Without carrying those errors forward, the ledger would be rewritten + from an empty in-memory list and the stale output would be reported as + a complete snapshot. + """ + previous = _read_json( + os.path.join(base_dir, COLLECTION_STATE_FILE) + ) or [] + carried = 0 + for entry in previous: + if entry.get("recovered"): + continue + if entry.get("reason") not in _JUNIT_SUSPECT_REASONS: + continue + forwarded = dict(entry) + forwarded["carried_forward"] = True + with _COLLECTION_ERRORS_LOCK: + _COLLECTION_ERRORS.append(forwarded) + carried += 1 + if carried: + _log(f" carrying forward {carried} unresolved JUnit collection " + f"error(s) from a previous run (JUnit collection is disabled)") + return carried + + +def _write_collection_state(base_dir: str) -> None: + """Persist the error ledger so a later process can see it.""" + state_path = os.path.join(base_dir, COLLECTION_STATE_FILE) + if _COLLECTION_ERRORS: + _write_json(state_path, list(_COLLECTION_ERRORS)) + elif os.path.exists(state_path): + # Everything was collected this time; drop the stale ledger. + os.remove(state_path) + + +def _mark_errors_recovered(entries: list[dict]) -> None: + """Mark the given error entries as recovered. + + A first-choice read can fail (e.g. a glob times out) and a fallback + can then succeed. The failure is still worth reporting — it is how + you learn a timeout needs tuning — but it must not make the snapshot + look incomplete when the data was in fact obtained. + + Entries are identified by object, not by position: only the specific + failures handed in are recovered, never a concurrent collector's. + """ + for entry in entries: + entry["recovered"] = True + + +def _unrecovered_errors() -> list[dict]: + """Collection errors that were not resolved by a fallback.""" + return [e for e in _COLLECTION_ERRORS if not e.get("recovered")] + + +def _unrecovered_error_count() -> int: + """Number of unrecovered collection errors.""" + return len(_unrecovered_errors()) + + +def _job_junit_state(job_name: str, payload_tag: str, reason: str) -> bool: + """True when this payload's collection of this job hit ``reason``. + + Scoped by payload tag: the same job name recurs in every payload of + the chain, so an unscoped match would stamp one payload's job entry + with another payload's failure. + """ + return any( + e.get("job") == job_name + and e.get("payload_tag") == payload_tag + and e.get("reason") == reason + and not e.get("recovered") + for e in _COLLECTION_ERRORS + ) + + +# Whether gcloud must be run in anonymous mode. Resolved once, on first +# use, under a lock — collectors run in a thread pool. +_GCLOUD_ANONYMOUS: Optional[bool] = None +_GCLOUD_ANONYMOUS_LOCK = threading.Lock() + + +def _gcloud_env() -> dict: + """Environment for gcloud calls, enabling anonymous access if needed. + + The CI artifact buckets are public, but ``gcloud storage`` refuses + client-side when no account is configured ("You do not currently have + an active account selected") — it never even issues the request. + Setting ``CLOUDSDK_AUTH_DISABLE_CREDENTIALS`` makes it read public + objects anonymously, so an unauthenticated environment still gets a + complete snapshot instead of an empty one. + """ + global _GCLOUD_ANONYMOUS + if _GCLOUD_ANONYMOUS is None: + with _GCLOUD_ANONYMOUS_LOCK: + if _GCLOUD_ANONYMOUS is None: + anonymous = not _check_gcloud_credentials() + if anonymous: + _log(" note: no gcloud credentials found — reading " + "public artifact buckets anonymously") + _GCLOUD_ANONYMOUS = anonymous + env = os.environ.copy() + if _GCLOUD_ANONYMOUS: + env["CLOUDSDK_AUTH_DISABLE_CREDENTIALS"] = "true" + return env + + def _run_gcloud(args: list[str], timeout: int = 120) -> Optional[str]: - """Run a gcloud CLI command, returning stdout or None on error.""" + """Run a gcloud CLI command, returning stdout or None on error. + + Failures other than "no such object" are recorded in + ``_COLLECTION_ERRORS`` so callers can distinguish an unreadable + source from an empty one. + """ try: result = subprocess.run( - args, capture_output=True, text=True, timeout=timeout + args, capture_output=True, text=True, timeout=timeout, + env=_gcloud_env(), ) if result.returncode != 0: + reason = _classify_gcloud_stderr(result.stderr) + if reason != "no_match": + _record_collection_error(reason, args, detail=result.stderr) return None return result.stdout - except (subprocess.TimeoutExpired, FileNotFoundError): + except subprocess.TimeoutExpired: + _record_collection_error( + "timeout", args, detail=f"exceeded {timeout}s" + ) + return None + except FileNotFoundError: + _record_collection_error("gcloud_missing", args) return None def _run_gcloud_bytes(args: list[str], timeout: int = 120) -> Optional[bytes]: - """Run a gcloud CLI command, returning raw stdout bytes or None.""" + """Run a gcloud CLI command, returning raw stdout bytes or None. + + Records failures the same way as :func:`_run_gcloud`. + """ try: result = subprocess.run( - args, capture_output=True, timeout=timeout + args, capture_output=True, timeout=timeout, env=_gcloud_env() ) if result.returncode != 0: + stderr = (result.stderr or b"").decode("utf-8", errors="replace") + reason = _classify_gcloud_stderr(stderr) + if reason != "no_match": + _record_collection_error(reason, args, detail=stderr) return None return result.stdout - except (subprocess.TimeoutExpired, FileNotFoundError): + except subprocess.TimeoutExpired: + _record_collection_error( + "timeout", args, detail=f"exceeded {timeout}s" + ) + return None + except FileNotFoundError: + _record_collection_error("gcloud_missing", args) return None @@ -2200,6 +2780,26 @@ def _check_gcloud() -> bool: return False +def _check_gcloud_credentials() -> bool: + """Report whether gcloud has an active credential. + + ``gcloud --version`` only proves the binary exists. An installed but + unauthenticated gcloud passes that check and then fails every read, + which previously produced a snapshot full of silently empty JUnit + data. Public buckets can still be read anonymously, so a negative + result here is a warning rather than a hard failure. + """ + try: + result = subprocess.run( + ["gcloud", "auth", "list", + "--filter=status:ACTIVE", "--format=value(account)"], + capture_output=True, text=True, timeout=15, + ) + return result.returncode == 0 and bool(result.stdout.strip()) + except (FileNotFoundError, subprocess.TimeoutExpired): + return False + + def _check_podman() -> bool: """Verify that podman is available and functional. @@ -2355,12 +2955,17 @@ def _parse_system_out_yaml(text: str) -> dict: return result -def _parse_junit_xml(source, source_name: str = "") -> list[_TestResult]: +def _parse_junit_xml( + source, source_name: str = "" +) -> Optional[list[_TestResult]]: """Parse JUnit XML, returning a list of _TestResult.""" try: tree = ET.parse(source) except (ET.ParseError, OSError): - return [] + # Return None, not []: an unreadable or corrupt file is not the same + # as a file that legitimately contains no failures. Callers must be + # able to tell those apart or corruption becomes a verified zero. + return None root = tree.getroot() if root.tag == "testsuites": @@ -2490,6 +3095,12 @@ def main() -> None: "--no-junit", action="store_true", help="Skip JUnit download, regression tracking", ) + parser.add_argument( + "--fail-on-incomplete", action="store_true", + help=("Exit non-zero if any data could not be collected. Use in " + "automation so an incomplete snapshot is not analyzed as " + "though it were complete."), + ) parser.add_argument( "--no-rpmdb", action="store_true", help="Skip RHCOS RPMDB extraction", @@ -2522,6 +3133,20 @@ def main() -> None: _log("Warning: 'gcloud' CLI not found. JUnit data will not be fetched.") _log("Install gcloud SDK to enable JUnit download and regression tracking.\n") collect_junit = False + # JUnit was requested and cannot be collected, so the snapshot is + # incomplete by construction. Record it: otherwise the completeness + # gate sees an empty ledger and reports a complete snapshot. + _record_collection_error( + "gcloud_missing", + ["gcloud", "storage"], + detail=("gcloud CLI not available; JUnit, build logs and " + "regression tracking were skipped"), + stage="preflight", + ) + elif collect_junit and not _check_gcloud_credentials(): + _log("Note: 'gcloud' has no active credentials. CI artifact buckets") + _log("are public, so they will be read anonymously. Run") + _log("'gcloud auth login' if you need private buckets too.\n") collect_rpmdb = not args.no_rpmdb if collect_rpmdb and not _check_podman(): @@ -2543,6 +3168,39 @@ def main() -> None: collect_rpmdb=collect_rpmdb, ) snapshotter.run() + + unrecovered = _unrecovered_errors() + recovered = len(_COLLECTION_ERRORS) - len(unrecovered) + if recovered: + _log("") + status = ("data is complete" if not unrecovered + else "see the warning below for what is still missing") + _log(f"Note: {recovered} read failure(s) were recovered by a " + f"fallback; {status}. See 'collection_errors' " + f"(recovered: true) in summary.json.") + if unrecovered: + by_reason: dict[str, int] = {} + for e in unrecovered: + r = e.get("reason", "unknown") + by_reason[r] = by_reason.get(r, 0) + 1 + _log("") + _log("=" * 68) + _log(f"WARNING: SNAPSHOT IS INCOMPLETE " + f"({len(unrecovered)} unrecovered collection error(s))") + for reason, count in sorted(by_reason.items()): + _log(f" {reason}: {count}") + affected = sorted({ + e["job"] for e in unrecovered if e.get("job") + }) + if affected: + _log(f" affected jobs: {', '.join(affected)}") + _log("") + _log("Missing data is ABSENT, not empty. Do not interpret a job") + _log("without test_failure_count as having zero test failures.") + _log("Details: 'collection_errors' in summary.json") + _log("=" * 68) + if args.fail_on_incomplete: + sys.exit(1) except urllib.error.HTTPError as e: print(f"Error: HTTP {e.code}: {e.reason}", file=sys.stderr) sys.exit(1) diff --git a/plugins/ci/skills/payload-snapshot/scripts/test_collection_completeness.py b/plugins/ci/skills/payload-snapshot/scripts/test_collection_completeness.py new file mode 100644 index 000000000..a359c76cd --- /dev/null +++ b/plugins/ci/skills/payload-snapshot/scripts/test_collection_completeness.py @@ -0,0 +1,354 @@ +#!/usr/bin/env python3 +"""Regression tests for the snapshot completeness contract. + +The contract under test: data the snapshot could not read must never be +published as an authoritative zero. Each test below corresponds to a way +that guarantee was previously broken. + +Run with: pytest test_collection_completeness.py +""" + +import importlib.util +import json +import os +import sys +import threading + +import pytest + +_HERE = os.path.dirname(os.path.abspath(__file__)) + + +def _load_module(): + spec = importlib.util.spec_from_file_location( + "payload_snapshot", os.path.join(_HERE, "payload_snapshot.py") + ) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +ps = _load_module() + + +@pytest.fixture(autouse=True) +def clean_ledger(): + """Each test starts with an empty global error ledger.""" + ps._COLLECTION_ERRORS.clear() + ps._GCLOUD_ANONYMOUS = True # never probe real credentials in tests + yield + ps._COLLECTION_ERRORS.clear() + + +def _job(name="job-a", bucket="bucket/path"): + return ps.JobInfo( + name=name, state="Failed", lifecycle="blocking", url="", + retries=0, previous_attempt_urls=[], is_aggregated=False, + gcs_bucket_path=bucket, + ) + + +def _collector(tmp_path, job=None, tag="4.99.0-0.ci-2026-01-01-000000"): + return ps.JUnitCollector(str(tmp_path / "junit"), job or _job(), tag) + + +VALID_JUNIT = ( + '' + 'boom' + "" +) + + +# -------------------------------------------------------------------------- +# Error classification +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize("stderr,expected", [ + ("ERROR: One or more URLs matched no objects.", "no_match"), + ("ERROR: 403 does not have storage.objects.list access", "auth"), + ("ERROR: Anonymous caller does not have access", "auth"), + ("ERROR: something unexpected", "command_failed"), +]) +def test_stderr_classification(stderr, expected): + assert ps._classify_gcloud_stderr(stderr) == expected + + +def test_no_match_is_not_recorded_as_error(monkeypatch): + """A missing object is absence, not failure — probes must stay quiet.""" + monkeypatch.setattr(ps.subprocess, "run", lambda *a, **k: type( + "R", (), {"returncode": 1, "stdout": "", + "stderr": "ERROR: matched no objects"})()) + assert ps._run_gcloud(["gcloud", "storage", "ls", "x"]) is None + assert ps._COLLECTION_ERRORS == [] + + +def test_auth_failure_is_recorded(monkeypatch): + monkeypatch.setattr(ps.subprocess, "run", lambda *a, **k: type( + "R", (), {"returncode": 1, "stdout": "", + "stderr": "ERROR: 403 forbidden"})()) + assert ps._run_gcloud(["gcloud", "storage", "ls", "x"]) is None + assert [e["reason"] for e in ps._COLLECTION_ERRORS] == ["auth"] + + +def test_detail_is_sanitized(): + """Persisted diagnostics must not carry control characters.""" + ps._record_collection_error( + "auth", ["gcloud"], detail="bad\x1b[31mred\x00\nthing" + ) + detail = ps._COLLECTION_ERRORS[0]["detail"] + assert "\x1b" not in detail and "\x00" not in detail + assert "\n" not in detail + + +# -------------------------------------------------------------------------- +# Recovery bookkeeping +# -------------------------------------------------------------------------- + +def test_recovery_is_scoped_to_its_own_operation(): + """A recovered fallback must not clear another operation's error.""" + with ps._error_scope() as mine: + ps._record_collection_error("timeout", ["gcloud"], job="mine") + # A concurrent collector records its own, unrelated failure. + ps._record_collection_error("auth", ["gcloud"], job="theirs") + + # Only the entry captured for *this* operation is recovered. + ps._mark_errors_recovered([e for e in mine if e.get("job") == "mine"]) + + by_job = {e["job"]: e for e in ps._COLLECTION_ERRORS} + assert by_job["mine"].get("recovered") is True + assert by_job["theirs"].get("recovered") is None + assert len(ps._unrecovered_errors()) == 1 + + +def test_concurrent_collectors_do_not_recover_each_others_errors(): + """Threads racing on the shared ledger must stay isolated.""" + started = threading.Barrier(2) + + def worker(tag): + with ps._error_scope() as own: + ps._record_collection_error("timeout", ["gcloud"], job=tag) + started.wait(timeout=5) # force interleaving + if tag == "recovers": + ps._mark_errors_recovered(own) + + threads = [threading.Thread(target=worker, args=(t,)) + for t in ("recovers", "keeps")] + for t in threads: + t.start() + for t in threads: + t.join(timeout=10) + + by_job = {e["job"]: e for e in ps._COLLECTION_ERRORS} + assert by_job["recovers"].get("recovered") is True + assert by_job["keeps"].get("recovered") is None + + +def test_data_complete_ignores_recovered_errors(): + with ps._error_scope() as own: + ps._record_collection_error("timeout", ["gcloud"]) + ps._mark_errors_recovered(own) + assert ps._unrecovered_errors() == [] + + +# -------------------------------------------------------------------------- +# Parsing +# -------------------------------------------------------------------------- + +def test_malformed_xml_returns_none_not_empty(tmp_path): + """Corrupt XML must be distinguishable from "no failures".""" + bad = tmp_path / "junit_bad.xml" + bad.write_text('trunc') + assert ps._parse_junit_xml(str(bad)) is None + + +def test_valid_xml_with_no_failures_returns_empty_list(tmp_path): + good = tmp_path / "junit_ok.xml" + good.write_text('' + "") + results = ps._parse_junit_xml(str(good)) + assert results is not None + assert ps._test_results_to_json(results) == [] + + +# -------------------------------------------------------------------------- +# Publishing decisions +# -------------------------------------------------------------------------- + +def _stub_discovery(monkeypatch, files): + monkeypatch.setattr( + ps.JUnitCollector, "_list_junit_files", lambda self: list(files) + ) + + +def test_unreadable_junit_leaves_results_absent(tmp_path, monkeypatch): + """Read failure => results.json absent (unknown), never [].""" + _stub_discovery(monkeypatch, ["gs://b/junit_operator.xml"]) + monkeypatch.setattr(ps, "_run_gcloud_bytes", lambda *a, **k: ( + ps._record_collection_error("auth", ["gcloud"]) or None)) + + c = _collector(tmp_path) + assert c.collect() is False + assert not os.path.exists(c.output_path) + assert any(e["reason"] == "junit_unavailable" + for e in ps._COLLECTION_ERRORS) + + +def test_no_junit_discovered_is_unknown_not_zero(tmp_path, monkeypatch): + """A failed job with no JUnit at all is unknown, not verified clean.""" + _stub_discovery(monkeypatch, []) + + c = _collector(tmp_path) + assert c.collect() is False + assert not os.path.exists(c.output_path) + assert any(e["reason"] == "junit_missing" + for e in ps._COLLECTION_ERRORS) + + +def test_partial_read_is_published_but_flagged(tmp_path, monkeypatch): + """Half the files read => real results, explicitly marked partial.""" + _stub_discovery(monkeypatch, ["gs://b/junit_a.xml", "gs://b/junit_b.xml"]) + + def fake_cat(args, **kwargs): + if args[-1].endswith("junit_a.xml"): + return VALID_JUNIT.encode() + ps._record_collection_error("auth", ["gcloud"]) + return None + + monkeypatch.setattr(ps, "_run_gcloud_bytes", fake_cat) + + c = _collector(tmp_path) + assert c.collect() is True + assert json.load(open(c.output_path)) # real failures published + assert any(e["reason"] == "junit_partial" + for e in ps._COLLECTION_ERRORS) + + +def test_unparseable_download_counts_as_unread(tmp_path, monkeypatch): + """Corrupt XML must not be published as zero failures.""" + _stub_discovery(monkeypatch, ["gs://b/junit_bad.xml"]) + monkeypatch.setattr( + ps, "_run_gcloud_bytes", lambda *a, **k: b"