Slow podlatency

bugzooka/analysis/log_analyzer.py

@@ -1,5 +1,6 @@
 import logging
 import asyncio
+import tempfile
 from pydantic import BaseModel, Field
 
 from langchain_core.tools import StructuredTool
@@ -15,7 +16,12 @@
 from bugzooka.analysis.log_summarizer import (
     download_prow_logs,
     generate_prompt,
+    find_pod_latency_file,
+    download_pod_latency_file,
+    parse_slowest_pod,
+    download_node_journal,
 )
+from bugzooka.analysis.node_log_analyzer import analyze_node_journal, format_result_markdown
 from bugzooka.integrations.inference_client import (
     get_inference_client,
     analyze_with_agentic,
@@ -26,7 +32,8 @@
 from bugzooka.integrations.mcp_client import initialize_global_resources_async
 from bugzooka.core.config import get_prompt_config
 from bugzooka.analysis.prow_analyzer import analyze_prow_artifacts, ProwAnalysisResult
-from bugzooka.core.utils import extract_job_details
+from bugzooka.core.utils import extract_job_details, extract_gcs_path
+from bugzooka.analysis.log_summarizer import get_prow_inner_artifact_files
 
 logger = logging.getLogger(__name__)
 
@@ -228,3 +235,71 @@ def _run():
     result = _run()
     retry_count = _run.statistics.get("attempt_number", 1) - 1
     return result, retry_count
+
+
+def run_node_rca_analysis(job_url: str, step_hint: str = "") -> str:
+    """
+    Full pipeline: prow URL → slowest pod → node journal → RCA markdown.
+
+    Steps:
+      1. Extract GCS path from prow view URL
+      2. Discover log_folder via get_prow_inner_artifact_files
+      3. Find podLatencyMeasurement JSON in openshift-qe-* step dir
+      4. Download + parse JSON to find slowest pod and its node
+      5. Download + decompress node journal (gzipped, no .gz suffix)
+      6. Run deterministic parser, return formatted markdown
+
+    :param job_url: prow view URL (https://prow.ci.../view/gs/...)
+    :param step_hint: workload name (e.g. "node-density-cni") to select the right
+        openshift-qe-* artifacts directory when multiple workloads are present
+    :return: markdown RCA report string, or an error message string
+    """
+    try:
+        gcs_path = extract_gcs_path(job_url)
+    except Exception as exc:
+        return f"node-rca: could not extract GCS path from URL: {exc}"
+
+    log_folder_path, _ = get_prow_inner_artifact_files(gcs_path)
+    if not log_folder_path:
+        return "node-rca: could not find inner artifact folder (is this a valid prow URL?)."
+
+    # log_folder_path looks like "gs://bucket/.../artifacts/<log_folder>/"
+    # extract just the log_folder name
+    log_folder = log_folder_path.rstrip("/").split("/artifacts/", 1)[-1].rstrip("/")
+
+    tmp_dir = tempfile.mkdtemp(prefix="bugzooka-node-rca-")
+
+    metrics_url = find_pod_latency_file(gcs_path, log_folder, step_hint=step_hint or None)
+    if not metrics_url:
+        return (
+            "node-rca: podLatencyMeasurement JSON not found under "
+            f"artifacts/{log_folder}/openshift-qe-*. "
+            "Is this a node-density or similar perf job?"
+        )
+
+    json_path = download_pod_latency_file(metrics_url, tmp_dir)
+    if not json_path:
+        return f"node-rca: failed to download {metrics_url}."
+
+    pod_name, node_hostname, latency_ms = parse_slowest_pod(json_path)
+    if not pod_name or not node_hostname:
+        return "node-rca: could not determine slowest pod from podLatencyMeasurement JSON."
+
+    logger.info(
+        "node-rca: slowest pod=%s node=%s latency=%dms",
+        pod_name, node_hostname, latency_ms,
+    )
+
+    journal_path = download_node_journal(gcs_path, log_folder, node_hostname, tmp_dir)
+    if not journal_path:
+        return (
+            f"node-rca: could not download journal for node {node_hostname}. "
+            "Check that gather-extra artifacts are present for this run."
+        )
+
+    rca_result = analyze_node_journal(journal_path, pod_name=pod_name)
+    header = (
+        f"**Node RCA for `{pod_name}`** "
+        f"(slowest pod, {latency_ms / 1000:.1f}s podReadyLatency)\n\n"
+    )
+    return header + format_result_markdown(rca_result)

bugzooka/analysis/log_summarizer.py

@@ -1,3 +1,5 @@
+import gzip
+import json
 import logging
 import os
 import re
@@ -381,6 +383,167 @@ def generate_prompt(error_list):
     return messages
 
 
+# ---------------------------------------------------------------------------
+# Node journal RCA helpers
+# ---------------------------------------------------------------------------
+
+
+def find_pod_latency_file(
+    gcs_path: str, log_folder: str, step_hint: Optional[str] = None
+) -> Optional[str]:
+    """
+    Locate podLatencyMeasurement-*.json under a prow artifact directory by
+    listing GCS at each variable path segment.
+
+    Expected layout:
+        gs://<gcs_path>/artifacts/<log_folder>/openshift-qe-*/
+            artifacts/collected-metrics-<uuid>/podLatencyMeasurement-*.json
+
+    :param gcs_path: raw GCS path without gs:// prefix
+    :param log_folder: inner log folder name (from get_prow_inner_artifact_files)
+    :param step_hint: workload name (e.g. "node-density-cni") to prefer the matching
+        openshift-qe-<step_hint> directory; falls back to all openshift-qe-* dirs
+    :return: full gs:// URL of the JSON file, or None if not found
+    """
+    base = f"gs://{gcs_path}/artifacts/{log_folder}/"
+    try:
+        top_entries = list_gcs_files(base)
+    except Exception as exc:
+        logger.error("find_pod_latency_file: cannot list %s: %s", base, exc)
+        return None
+
+    # Find openshift-qe-* step directories (GCS lists dirs with trailing /)
+    qe_dirs = [e for e in top_entries if e.endswith("/")
+               and "openshift-qe-" in gcs_basename(e.rstrip("/"))]
+    if not qe_dirs:
+        logger.info("find_pod_latency_file: no openshift-qe-* dirs under %s", base)
+        return None
+
+    # When a step_hint is given (e.g. "node-density-cni"), search matching dirs
+    # first so we pick the right workload's metrics, not an alphabetically earlier one.
+    if step_hint:
+        preferred = [d for d in qe_dirs if step_hint in gcs_basename(d.rstrip("/"))]
+        rest = [d for d in qe_dirs if d not in preferred]
+        qe_dirs = preferred + rest
+
+    for qe_dir in qe_dirs:
+        artifacts_path = qe_dir.rstrip("/") + "/artifacts/"
+        try:
+            artifacts_entries = list_gcs_files(artifacts_path)
+        except Exception:
+            continue
+
+        metrics_dirs = [
+            e for e in artifacts_entries
+            if e.endswith("/") and "collected-metrics-" in gcs_basename(e.rstrip("/"))
+        ]
+        for metrics_dir in metrics_dirs:
+            try:
+                files = list_gcs_files(metrics_dir.rstrip("/") + "/")
+            except Exception:
+                continue
+            for f in files:
+                name = gcs_basename(f.rstrip("/"))
+                if name.startswith("podLatencyMeasurement-") and name.endswith(".json"):
+                    return f.rstrip()
+
+    logger.info("find_pod_latency_file: no podLatencyMeasurement JSON found under %s", base)
+    return None
+
+
+def download_pod_latency_file(gcs_url: str, output_dir: str) -> Optional[str]:
+    """
+    Download a podLatencyMeasurement JSON from GCS.
+
+    :param gcs_url: full gs:// URL
+    :param output_dir: local directory to write the file
+    :return: local file path, or None on failure
+    """
+    try:
+        download_file_from_gcs(gcs_url, output_dir)
+        local_path = os.path.join(output_dir, gcs_basename(gcs_url))
+        if os.path.exists(local_path):
+            return local_path
+    except Exception as exc:
+        logger.error("download_pod_latency_file failed: %s", exc)
+    return None
+
+
+def parse_slowest_pod(json_path: str) -> Tuple[Optional[str], Optional[str], int]:
+    """
+    Read a podLatencyMeasurement JSON and return the pod with the highest
+    podReadyLatency.
+
+    :param json_path: local path to the JSON file
+    :return: (pod_name, node_name, latency_ms) — all None/0 on failure
+    """
+    try:
+        with open(json_path, encoding="utf-8") as fh:
+            records = json.load(fh)
+        if not records:
+            return None, None, 0
+        slowest = max(records, key=lambda r: r.get("podReadyLatency", 0))
+        return (
+            slowest.get("podName"),
+            slowest.get("nodeName"),
+            slowest.get("podReadyLatency", 0),
+        )
+    except Exception as exc:
+        logger.error("parse_slowest_pod failed for %s: %s", json_path, exc)
+        return None, None, 0
+
+
+def download_node_journal(
+    gcs_path: str,
+    log_folder: str,
+    node_hostname: str,
+    output_dir: str,
+) -> Optional[str]:
+    """
+    Download the node's systemd journal (gzipped without .gz suffix) and
+    decompress it to a plain-text file.
+
+    GCS path:
+        gs://<gcs_path>/artifacts/<log_folder>/gather-extra/artifacts/nodes/<node>/journal
+
+    :param gcs_path: raw GCS path without gs:// prefix
+    :param log_folder: inner log folder name
+    :param node_hostname: full node hostname (e.g. ip-10-0-67-181.us-west-2...)
+    :param output_dir: local directory for downloaded + decompressed files
+    :return: path to decompressed journal file, or None on failure
+    """
+    journal_url = (
+        f"gs://{gcs_path}/artifacts/{log_folder}/"
+        f"gather-extra/artifacts/nodes/{node_hostname}/journal"
+    )
+    node_dir = os.path.join(output_dir, node_hostname)
+    os.makedirs(node_dir, exist_ok=True)
+
+    try:
+        download_file_from_gcs(journal_url, node_dir)
+    except Exception as exc:
+        logger.error("download_node_journal: gsutil failed for %s: %s", journal_url, exc)
+        return None
+
+    compressed_path = os.path.join(node_dir, "journal")
+    if not os.path.exists(compressed_path):
+        logger.error("download_node_journal: journal not found at %s", compressed_path)
+        return None
+
+    decompressed_path = compressed_path + ".decompressed"
+    try:
+        with gzip.open(compressed_path, "rt", errors="replace") as f_in, \
+                open(decompressed_path, "w", encoding="utf-8") as f_out:
+            f_out.write(f_in.read())
+        return decompressed_path
+    except Exception as exc:
+        logger.error(
+            "download_node_journal: decompression failed for %s: %s",
+            compressed_path, exc,
+        )
+        return None
+
+
 def classify_failure_type(errors_list, categorization_message, is_install_issue):
     """
     Map analysis outputs to a display label for failure type.