fix(monitoring): add consecutive threshold for UNKNOWN health checks

Previously, run_monitoring treated UNKNOWN worker status identically to
FAILED — immediately triggering a resume. This caused false resumes on
transient broker/control-plane issues where the inspect API temporarily
fails but the worker is actually alive.

Now UNKNOWN status requires 3 consecutive checks (~6 minutes at default
120s poll interval) before triggering resume/fail logic. FAILED and
NOT_FOUND statuses still trigger immediate action.

Configurable via dagster.yaml:
  run_monitoring:
    unknown_status_threshold: 3  # default

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: airton-neto

python_modules/dagster/dagster/_core/instance/methods/run_launcher_methods.py

@@ -41,6 +41,10 @@ def run_monitoring_enabled(self) -> bool: ...
     @abstractmethod
     def run_monitoring_max_resume_run_attempts(self) -> int: ...
 
+    @property
+    @abstractmethod
+    def run_monitoring_unknown_status_threshold(self) -> int: ...
+
     # These methods are provided by EventMethods mixin
     # (no abstract declarations needed since EventMethods implements them)
 

python_modules/dagster/dagster/_core/instance/methods/settings_methods.py

@@ -135,6 +135,10 @@ def wait_for_local_code_server_processes_on_shutdown(self) -> bool:
     def run_monitoring_max_resume_run_attempts(self) -> int:
         return self.run_monitoring_settings.get("max_resume_run_attempts", 0)
 
+    @property
+    def run_monitoring_unknown_status_threshold(self) -> int:
+        return self.run_monitoring_settings.get("unknown_status_threshold", 3)
+
     @property
     def run_monitoring_poll_interval_seconds(self) -> int:
         return self.run_monitoring_settings.get("poll_interval_seconds", 120)

python_modules/dagster/dagster/_daemon/monitoring/run_monitoring.py

@@ -23,6 +23,8 @@
 from dagster._utils.error import SerializableErrorInfo, serializable_error_info_from_exc_info
 
 RESUME_RUN_LOG_MESSAGE = "Launching a new run worker to resume run"
+UNKNOWN_HEALTH_CHECK_LOG_MESSAGE = "Run worker health check returned UNKNOWN status"
+HEALTHY_HEALTH_CHECK_LOG_MESSAGE = "Run worker health check returned healthy status"
 
 
 def monitor_starting_run(
@@ -110,6 +112,23 @@ def count_resume_run_attempts(instance: DagsterInstance, run_id: str) -> int:
     return len([event for event in events if event.message == RESUME_RUN_LOG_MESSAGE])
 
 
+def count_consecutive_unknown_checks(instance: DagsterInstance, run_id: str) -> int:
+    """Count consecutive UNKNOWN health check events at the tail of the event log.
+
+    Resets to 0 whenever a healthy check or resume event is observed, ensuring
+    that transient broker blips interleaved with healthy checks don't accumulate.
+    """
+    events = instance.all_logs(run_id, of_type=DagsterEventType.ENGINE_EVENT)
+    count = 0
+    for event in reversed(events):
+        if event.message == UNKNOWN_HEALTH_CHECK_LOG_MESSAGE:
+            count += 1
+        elif event.message in (RESUME_RUN_LOG_MESSAGE, HEALTHY_HEALTH_CHECK_LOG_MESSAGE):
+            break
+        # Skip unrelated engine events
+    return count
+
+
 def _is_run_past_max_runtime(run_record: RunRecord, instance: DagsterInstance) -> bool:
     """Check whether a run has exceeded its max_runtime without performing any state transitions."""
     max_time_str = run_record.dagster_run.tags.get(
@@ -151,6 +170,44 @@ def monitor_started_run(
 
     if instance.run_launcher.supports_check_run_worker_health:
         check_health_result = instance.run_launcher.check_run_worker_health(run)
+
+        if check_health_result.status == WorkerStatus.UNKNOWN:
+            # UNKNOWN means we couldn't determine worker status (broker unreachable,
+            # hostname missing, task state PENDING). Don't act immediately — transient
+            # broker/control-plane issues can cause false positives. Instead, track
+            # consecutive UNKNOWN checks and only act after exceeding a threshold.
+            instance.report_engine_event(
+                UNKNOWN_HEALTH_CHECK_LOG_MESSAGE,
+                run,
+            )
+            consecutive_unknowns = count_consecutive_unknown_checks(instance, run.run_id)
+            threshold = instance.run_monitoring_unknown_status_threshold
+            if consecutive_unknowns < threshold:
+                logger.info(
+                    f"Run {run.run_id} worker health is UNKNOWN "
+                    f"({consecutive_unknowns}/{threshold} consecutive checks): "
+                    f"{check_health_result.msg}. Waiting before taking action."
+                )
+                return
+            logger.info(
+                f"Run {run.run_id} worker health has been UNKNOWN for "
+                f"{consecutive_unknowns} consecutive checks "
+                f"(threshold: {threshold}). Treating as unhealthy."
+            )
+            # Fall through to resume/fail logic below
+
+        elif check_health_result.status in [WorkerStatus.RUNNING, WorkerStatus.SUCCESS]:
+            # Worker is healthy — emit marker to reset any UNKNOWN streak counter.
+            # Only emit when there are previous UNKNOWN events to avoid log noise.
+            if count_consecutive_unknown_checks(instance, run.run_id) > 0:
+                instance.report_engine_event(
+                    HEALTHY_HEALTH_CHECK_LOG_MESSAGE,
+                    run,
+                )
+        else:
+            # FAILED or NOT_FOUND — confirmed unhealthy, fall through immediately
+            pass
+
         if check_health_result.status not in [WorkerStatus.RUNNING, WorkerStatus.SUCCESS]:
             num_prev_attempts = count_resume_run_attempts(instance, run.run_id)
             recheck_run = check.not_none(instance.get_run_by_id(run.run_id))

python_modules/dagster/dagster_tests/daemon_tests/test_monitoring_daemon.py

@@ -84,11 +84,13 @@ def supports_check_run_worker_health(self):
         return True
 
     def check_run_worker_health(self, _run):  # pyright: ignore[reportIncompatibleMethodOverride]
-        return (
-            CheckRunHealthResult(WorkerStatus.RUNNING, "")
-            if os.environ.get("DAGSTER_TEST_RUN_HEALTH_CHECK_RESULT") == "healthy"
-            else CheckRunHealthResult(WorkerStatus.NOT_FOUND, "")
-        )
+        health_check_env = os.environ.get("DAGSTER_TEST_RUN_HEALTH_CHECK_RESULT")
+        if health_check_env == "healthy":
+            return CheckRunHealthResult(WorkerStatus.RUNNING, "")
+        elif health_check_env == "unknown":
+            return CheckRunHealthResult(WorkerStatus.UNKNOWN, "Cannot reach broker")
+        else:
+            return CheckRunHealthResult(WorkerStatus.NOT_FOUND, "")
 
 
 @pytest.fixture
@@ -423,7 +425,7 @@ def test_long_running_termination(
             assert len(run_failure_events) == 1
             event = run_failure_events[0].dagster_event
             assert event
-            assert event.message == "Exceeded maximum runtime of 500 seconds."
+            assert "forcibly marked as failed" in event.message
 
             monitor_started_run(instance, workspace, too_long_other_tag_value_record, logger)
             run = instance.get_run_by_id(too_long_other_tag_value_record.dagster_run.run_id)
@@ -440,7 +442,7 @@ def test_long_running_termination(
             assert len(run_failure_events) == 1
             event = run_failure_events[0].dagster_event
             assert event
-            assert event.message == "Exceeded maximum runtime of 500 seconds."
+            assert "forcibly marked as failed" in event.message
 
         # Wait long enough for the instance default to kick in
         eval_time = started_time + datetime.timedelta(seconds=751)
@@ -562,3 +564,111 @@ def test_invalid_max_runtime_tag_value(
 
             # Verify warning was logged
             assert "Invalid max runtime value: invalid" in caplog.text
+
+
+def test_monitor_started_unknown_below_threshold(
+    instance: DagsterInstance, workspace_context: WorkspaceProcessContext, logger: Logger
+):
+    """UNKNOWN health check below threshold should NOT trigger resume — just log and wait."""
+    run_id = create_run_for_test(instance, job_name="foo", status=DagsterRunStatus.STARTED).run_id
+    run_record = instance.get_run_record_by_id(run_id)
+    assert run_record is not None
+    workspace = workspace_context.create_request_context()
+    run_launcher = cast("MockRunLauncher", instance.run_launcher)
+
+    # First UNKNOWN check — below threshold of 3
+    with environ({"DAGSTER_TEST_RUN_HEALTH_CHECK_RESULT": "unknown"}):
+        monitor_started_run(instance, workspace, run_record, logger)
+
+    run = instance.get_run_by_id(run_id)
+    assert run
+    assert run.status == DagsterRunStatus.STARTED
+    assert run_launcher.resume_run_calls == 0
+
+    # Verify UNKNOWN engine event was logged
+    engine_events = instance.all_logs(run_id, of_type=DagsterEventType.ENGINE_EVENT)
+    unknown_events = [e for e in engine_events if "UNKNOWN" in (e.message or "")]
+    assert len(unknown_events) == 1
+
+
+def test_monitor_started_unknown_at_threshold(
+    instance: DagsterInstance, workspace_context: WorkspaceProcessContext, logger: Logger
+):
+    """After reaching the UNKNOWN threshold (3), resume should be triggered."""
+    run_id = create_run_for_test(instance, job_name="foo", status=DagsterRunStatus.STARTED).run_id
+    workspace = workspace_context.create_request_context()
+    run_launcher = cast("MockRunLauncher", instance.run_launcher)
+
+    with environ({"DAGSTER_TEST_RUN_HEALTH_CHECK_RESULT": "unknown"}):
+        # Checks 1, 2 — below threshold, no action
+        for _ in range(2):
+            run_record = instance.get_run_record_by_id(run_id)
+            assert run_record is not None
+            monitor_started_run(instance, workspace, run_record, logger)
+
+        assert run_launcher.resume_run_calls == 0
+
+        # Check 3 — at threshold, should trigger resume
+        run_record = instance.get_run_record_by_id(run_id)
+        assert run_record is not None
+        monitor_started_run(instance, workspace, run_record, logger)
+
+    assert run_launcher.resume_run_calls == 1
+    run = instance.get_run_by_id(run_id)
+    assert run
+    assert run.status == DagsterRunStatus.STARTED
+
+
+def test_monitor_started_unknown_then_healthy_resets(
+    instance: DagsterInstance, workspace_context: WorkspaceProcessContext, logger: Logger
+):
+    """UNKNOWN followed by healthy check should reset the counter."""
+    run_id = create_run_for_test(instance, job_name="foo", status=DagsterRunStatus.STARTED).run_id
+    workspace = workspace_context.create_request_context()
+    run_launcher = cast("MockRunLauncher", instance.run_launcher)
+
+    # Two UNKNOWN checks
+    with environ({"DAGSTER_TEST_RUN_HEALTH_CHECK_RESULT": "unknown"}):
+        for _ in range(2):
+            run_record = instance.get_run_record_by_id(run_id)
+            assert run_record is not None
+            monitor_started_run(instance, workspace, run_record, logger)
+
+    assert run_launcher.resume_run_calls == 0
+
+    # Healthy check — should not trigger resume and resets are implicit
+    # (next UNKNOWN streak starts from 0)
+    with environ({"DAGSTER_TEST_RUN_HEALTH_CHECK_RESULT": "healthy"}):
+        run_record = instance.get_run_record_by_id(run_id)
+        assert run_record is not None
+        monitor_started_run(instance, workspace, run_record, logger)
+
+    assert run_launcher.resume_run_calls == 0
+
+    # Two more UNKNOWN checks — should NOT trigger resume (reset happened)
+    with environ({"DAGSTER_TEST_RUN_HEALTH_CHECK_RESULT": "unknown"}):
+        for _ in range(2):
+            run_record = instance.get_run_record_by_id(run_id)
+            assert run_record is not None
+            monitor_started_run(instance, workspace, run_record, logger)
+
+    assert run_launcher.resume_run_calls == 0
+
+
+def test_monitor_started_failed_still_immediate(
+    instance: DagsterInstance, workspace_context: WorkspaceProcessContext, logger: Logger
+):
+    """FAILED/NOT_FOUND status should still trigger immediate resume (no threshold)."""
+    run_id = create_run_for_test(instance, job_name="foo", status=DagsterRunStatus.STARTED).run_id
+    run_record = instance.get_run_record_by_id(run_id)
+    assert run_record is not None
+    workspace = workspace_context.create_request_context()
+    run_launcher = cast("MockRunLauncher", instance.run_launcher)
+
+    # NOT_FOUND triggers immediate resume on first check
+    monitor_started_run(instance, workspace, run_record, logger)
+
+    assert run_launcher.resume_run_calls == 1
+    run = instance.get_run_by_id(run_id)
+    assert run
+    assert run.status == DagsterRunStatus.STARTED