Harden H1 trace source completeness

Author: lzy-edu

flagscale/runner/tracing/analyzer.py

@@ -209,6 +209,7 @@ def scan(
         *,
         now_monotonic_s: float,
         now_unix_ns: int,
+        trace_source_issues: dict[tuple[int, int], tuple[str, ...]] | None = None,
     ) -> list[Finding]:
         findings: list[Finding] = []
         if self.detect_heartbeat_timeouts:
@@ -242,6 +243,7 @@ def scan(
                 elapsed_s=elapsed_s,
                 detection_phase=detection_phase,
                 timeout_s=timeout_s,
+                trace_source_issues=trace_source_issues,
             )
             self._emit_once(("collective_missing_enter", *key), missing, findings)
 
@@ -366,22 +368,23 @@ def _detect_missing_enter(
         elapsed_s: float,
         detection_phase: str,
         timeout_s: float,
+        trace_source_issues: dict[tuple[int, int], tuple[str, ...]] | None,
     ) -> Finding:
         entered = sorted(collective.enters)
         missing = sorted(set(range(collective.expected_nranks)) - set(entered))
         members = self._comm_members.get((collective.run_id, collective.comm_uid_hash), {})
         rank_status: list[dict[str, Any]] = []
         known_states: list[str] = []
-        trace_event_loss_possible = False
+        source_states: list[str] = []
         for comm_rank in missing:
             global_rank = members.get(comm_rank)
             status: dict[str, Any] = {"comm_rank": comm_rank, "rank": global_rank}
             heartbeat = self._heartbeats.get(global_rank) if global_rank is not None else None
-            probe_status = self._probe_status.get(global_rank) if global_rank is not None else None
-            dropped_events = _as_int(
-                probe_status.get("dropped_events") if probe_status else 0, default=0
+            source = self._trace_source_status(
+                global_rank,
+                trace_source_issues=trace_source_issues,
             )
-            trace_event_loss_possible = trace_event_loss_possible or dropped_events > 0
+            source_states.append(str(source["status"]))
             if heartbeat is None:
                 status["heartbeat"] = "unknown"
             else:
@@ -395,21 +398,35 @@ def _detect_missing_enter(
                     }
                 )
                 known_states.append(state)
-            status["probe_dropped_events"] = dropped_events
+            status["probe_dropped_events"] = source["dropped_events"]
+            status["trace_source"] = source
             rank_status.append(status)
 
-        if trace_event_loss_possible:
+        if "unavailable" in source_states:
+            source_coverage = "insufficient"
+            reason = "trace_source_unavailable"
+            confidence = "unknown"
+        elif "incomplete" in source_states:
+            source_coverage = "insufficient"
+            reason = "trace_source_incomplete"
+            confidence = "unknown"
+        elif "lossy" in source_states:
+            source_coverage = "degraded"
             reason = "probe_event_loss_possible"
             confidence = "suspected"
-        elif "stale" in known_states:
-            reason = "rank_exit_or_crash_suspected"
-            confidence = "suspected"
-        elif known_states and all(state == "alive" for state in known_states):
-            reason = "rank_alive_but_not_entered"
-            confidence = "observed"
         else:
-            reason = "missing_rank_status_unknown"
-            confidence = "observed"
+            source_coverage = "sufficient"
+            if "stale" in known_states:
+                reason = "rank_exit_or_crash_suspected"
+                confidence = "suspected"
+            elif known_states and all(state == "alive" for state in known_states):
+                reason = "rank_alive_but_not_entered"
+                confidence = "observed"
+            else:
+                reason = "missing_rank_status_unknown"
+                confidence = "observed"
+
+        trace_event_loss_possible = source_coverage != "sufficient"
 
         first_event = next(iter(collective.enters.values()))
         return Finding(
@@ -430,10 +447,77 @@ def _detect_missing_enter(
                 "threshold_reason": self._threshold_reason(detection_phase),
                 "reason": reason,
                 "confidence": confidence,
+                "trace_source_coverage": source_coverage,
                 "trace_event_loss_possible": trace_event_loss_possible,
             },
         )
 
+    def _trace_source_status(
+        self,
+        global_rank: int | None,
+        *,
+        trace_source_issues: dict[tuple[int, int], tuple[str, ...]] | None,
+    ) -> dict[str, Any]:
+        """Describe whether absence of a trace event is usable evidence."""
+
+        issues: list[str] = []
+        if global_rank is None:
+            return {
+                "status": "unavailable",
+                "pid": None,
+                "dropped_events": 0,
+                "issues": ["communicator_rank_not_mapped"],
+            }
+
+        process = self._processes.get(global_rank)
+        probe_status = self._probe_status.get(global_rank)
+        process_pid = _as_int(process.event.get("pid"), default=-1) if process is not None else -1
+        status_pid = (
+            _as_int(probe_status.get("pid"), default=-1) if probe_status is not None else -1
+        )
+        if process is None:
+            issues.append("process_start_not_observed")
+        if probe_status is None:
+            issues.append("probe_status_not_observed")
+        if process_pid >= 0 and status_pid >= 0 and process_pid != status_pid:
+            issues.append("probe_status_pid_mismatch")
+
+        pid = process_pid if process_pid >= 0 else status_pid
+        if trace_source_issues is not None and pid >= 0:
+            source_key = (global_rank, pid)
+            if source_key not in trace_source_issues:
+                issues.append("trace_file_not_observed")
+            else:
+                issues.extend(trace_source_issues[source_key])
+
+        dropped_events = _as_int(
+            probe_status.get("dropped_events") if probe_status else 0,
+            default=0,
+        )
+        unavailable_issues = {
+            "process_start_not_observed",
+            "probe_status_not_observed",
+            "probe_status_pid_mismatch",
+            "trace_file_not_observed",
+        }
+        if unavailable_issues.intersection(issues):
+            source_status = "unavailable"
+        elif issues:
+            source_status = "incomplete"
+        elif dropped_events > 0:
+            source_status = "lossy"
+        else:
+            # This proves that the source was loaded and that the collector has
+            # observed no concrete loss signal. It intentionally does not claim
+            # continuous writer liveness.
+            source_status = "available"
+        return {
+            "status": source_status,
+            "pid": pid if pid >= 0 else None,
+            "dropped_events": dropped_events,
+            "issues": sorted(set(issues)),
+        }
+
     def _collective_detection_phase(self, collective: _CollectiveRound) -> str:
         """Return a stable phase for one collective round.
 

flagscale/runner/tracing/monitor.py

@@ -20,6 +20,7 @@
 import json
 import logging
 import os
+import re
 import signal
 import time
 from pathlib import Path
@@ -29,6 +30,8 @@
 
 logger = logging.getLogger("flagscale.tracing")
 
+_TRACE_SOURCE_PATTERN = re.compile(r"^rank_(\d+)_pid_(\d+)\.jsonl$")
+
 
 class JsonlTailer:
     """Incrementally read complete JSON lines from per-process trace files."""
@@ -37,22 +40,55 @@ def __init__(self, trace_dir: Path) -> None:
         self.trace_dir = trace_dir
         self._offsets: dict[Path, int] = {}
         self._remainders: dict[Path, str] = {}
+        self._source_paths: dict[tuple[int, int], Path] = {}
+        self._source_issues: dict[tuple[int, int], set[str]] = {}
+
+    @staticmethod
+    def _source_key(path: Path) -> tuple[int, int] | None:
+        match = _TRACE_SOURCE_PATTERN.fullmatch(path.name)
+        if match is None:
+            return None
+        return int(match.group(1)), int(match.group(2))
+
+    def source_issues(self) -> dict[tuple[int, int], tuple[str, ...]]:
+        """Return collector-visible source files and permanent/transient issues."""
+
+        return {
+            source: tuple(sorted(self._source_issues.get(source, set())))
+            for source in self._source_paths
+        }
 
     def poll(self) -> list[dict]:
         events: list[dict] = []
-        for path in sorted(self.trace_dir.glob("rank_*_pid_*.jsonl")):
+        paths = sorted(self.trace_dir.glob("rank_*_pid_*.jsonl"))
+        visible_paths = set(paths)
+        for source, path in self._source_paths.items():
+            if path not in visible_paths:
+                self._source_issues.setdefault(source, set()).add("trace_file_missing")
+
+        for path in paths:
+            source = self._source_key(path)
+            if source is not None:
+                self._source_paths[source] = path
+                issues = self._source_issues.setdefault(source, set())
+                issues.discard("trace_file_missing")
+                issues.discard("trace_read_error")
             offset = self._offsets.get(path, 0)
             try:
                 size = path.stat().st_size
                 if size < offset:
                     offset = 0
                     self._remainders.pop(path, None)
+                    if source is not None:
+                        issues.add("trace_file_truncated")
                 with path.open("r", encoding="utf-8", errors="replace") as file_obj:
                     file_obj.seek(offset)
                     chunk = file_obj.read()
                     self._offsets[path] = file_obj.tell()
             except OSError as exc:
                 logger.debug("Could not read %s: %s", path, exc)
+                if source is not None:
+                    issues.add("trace_read_error")
                 continue
 
             if not chunk:
@@ -62,14 +98,20 @@ def poll(self) -> list[dict]:
             for line in lines:
                 if not line.endswith(("\n", "\r")):
                     self._remainders[path] = line
+                    if source is not None:
+                        issues.add("partial_json_record")
                     continue
                 try:
                     event = json.loads(line)
                 except json.JSONDecodeError:
                     logger.warning("Ignored malformed trace line in %s", path)
+                    if source is not None:
+                        issues.add("malformed_json_record")
                     continue
                 if isinstance(event, dict):
                     events.append(event)
+            if source is not None and path not in self._remainders:
+                issues.discard("partial_json_record")
         return events
 
 
@@ -112,7 +154,8 @@ def run_monitor(args: argparse.Namespace) -> int:
         delayed_enter_threshold_s=args.delayed_enter_threshold,
         checkpoint_timeout_s=args.checkpoint_timeout,
     )
-    tailers = [JsonlTailer(trace_dir)]
+    trace_tailer = JsonlTailer(trace_dir)
+    tailers = [trace_tailer]
     if args.heartbeat_dir:
         tailers.append(JsonlTailer(Path(args.heartbeat_dir)))
     stopping = False
@@ -140,6 +183,7 @@ def request_stop(_signum, _frame) -> None:
             findings = analyzer.scan(
                 now_monotonic_s=time.monotonic(),
                 now_unix_ns=time.time_ns(),
+                trace_source_issues=trace_tailer.source_issues(),
             )
             _append_findings(output, findings)
 
@@ -174,6 +218,7 @@ def request_stop(_signum, _frame) -> None:
                     analyzer.scan(
                         now_monotonic_s=time.monotonic(),
                         now_unix_ns=time.time_ns(),
+                        trace_source_issues=trace_tailer.source_issues(),
                     ),
                 )
                 break

tests/unit_tests/runner/tracing/test_analyzer.py

@@ -52,6 +52,11 @@ def _ingest(analyzer, event, observed=1.0, unix_ns=1_000_000_000):
     )
 
 
+def _ingest_probe_source(analyzer, rank, pid=10):
+    _ingest(analyzer, _event("process_start", rank=rank, pid=pid))
+    _ingest(analyzer, _event("probe_status", rank=rank, pid=pid, dropped_events=0))
+
+
 def test_missing_enter_reports_live_missing_rank_without_claiming_crash():
     analyzer = TraceAnalyzer(
         run_id=RUN_ID,
@@ -60,6 +65,7 @@ def test_missing_enter_reports_live_missing_rank_without_claiming_crash():
     )
     _ingest(analyzer, _event("comm_init", rank=0, comm_rank=0))
     _ingest(analyzer, _event("comm_init", rank=1, comm_rank=1))
+    _ingest_probe_source(analyzer, rank=1)
     _ingest(
         analyzer,
         _event(
@@ -82,7 +88,9 @@ def test_missing_enter_reports_live_missing_rank_without_claiming_crash():
     assert finding.hang_type == "collective_missing_enter"
     assert finding.details["missing_comm_ranks"] == [1]
     assert finding.details["reason"] == "rank_alive_but_not_entered"
+    assert finding.details["trace_source_coverage"] == "sufficient"
     assert finding.details["missing_rank_status"][0]["heartbeat"] == "alive"
+    assert finding.details["missing_rank_status"][0]["trace_source"]["status"] == "available"
 
 
 def test_missing_enter_correlates_a_stale_heartbeat_as_suspected_crash():
@@ -92,6 +100,7 @@ def test_missing_enter_correlates_a_stale_heartbeat_as_suspected_crash():
         collective_timeout_s=2,
     )
     _ingest(analyzer, _event("comm_init", rank=1, comm_rank=1))
+    _ingest_probe_source(analyzer, rank=1)
     _ingest(analyzer, _event("heartbeat", rank=1), observed=1.0)
     _ingest(analyzer, _event("nccl_call", rank=0, comm_rank=0), observed=1.0)
 
@@ -228,6 +237,7 @@ def test_missing_enter_is_downgraded_when_probe_dropped_events():
         collective_timeout_s=2,
     )
     _ingest(analyzer, _event("comm_init", rank=1, comm_rank=1))
+    _ingest(analyzer, _event("process_start", rank=1))
     _ingest(
         analyzer,
         _event("heartbeat", rank=1),
@@ -240,3 +250,53 @@ def test_missing_enter_is_downgraded_when_probe_dropped_events():
 
     assert findings[0].details["reason"] == "probe_event_loss_possible"
     assert findings[0].details["confidence"] == "suspected"
+    assert findings[0].details["trace_source_coverage"] == "degraded"
+
+
+def test_missing_enter_is_unknown_when_probe_source_was_not_observed():
+    analyzer = TraceAnalyzer(
+        run_id=RUN_ID,
+        heartbeat_timeout_s=10,
+        collective_timeout_s=2,
+    )
+    _ingest(analyzer, _event("comm_init", rank=1, comm_rank=1))
+    _ingest(analyzer, _event("heartbeat", rank=1), observed=1)
+    _ingest(analyzer, _event("nccl_call", rank=0, comm_rank=0), observed=1)
+
+    findings = analyzer.scan(now_monotonic_s=5, now_unix_ns=5_000_000_000)
+
+    assert findings[0].details["reason"] == "trace_source_unavailable"
+    assert findings[0].details["confidence"] == "unknown"
+    assert findings[0].details["trace_source_coverage"] == "insufficient"
+    source = findings[0].details["missing_rank_status"][0]["trace_source"]
+    assert source["status"] == "unavailable"
+    assert source["issues"] == ["probe_status_not_observed", "process_start_not_observed"]
+
+
+def test_missing_enter_is_unknown_when_trace_file_is_incomplete():
+    analyzer = TraceAnalyzer(
+        run_id=RUN_ID,
+        heartbeat_timeout_s=10,
+        collective_timeout_s=2,
+    )
+    _ingest(analyzer, _event("comm_init", rank=1, comm_rank=1))
+    _ingest_probe_source(analyzer, rank=1)
+    _ingest(analyzer, _event("heartbeat", rank=1), observed=1)
+    _ingest(analyzer, _event("nccl_call", rank=0, comm_rank=0), observed=1)
+
+    findings = analyzer.scan(
+        now_monotonic_s=5,
+        now_unix_ns=5_000_000_000,
+        trace_source_issues={(1, 10): ("partial_json_record",)},
+    )
+
+    assert findings[0].details["reason"] == "trace_source_incomplete"
+    assert findings[0].details["confidence"] == "unknown"
+    assert findings[0].details["trace_source_coverage"] == "insufficient"
+    source = findings[0].details["missing_rank_status"][0]["trace_source"]
+    assert source == {
+        "status": "incomplete",
+        "pid": 10,
+        "dropped_events": 0,
+        "issues": ["partial_json_record"],
+    }