change default run coordinator

dagster-io · Jan 3, 2025 · 45ba09b · 45ba09b
1 parent 5d20559
commit 45ba09b
Show file tree

Hide file tree

Showing 15 changed files with 131 additions and 59 deletions.
diff --git a/helm/dagster/templates/configmap-instance.yaml b/helm/dagster/templates/configmap-instance.yaml
@@ -55,6 +55,10 @@ data:
       {{- else if eq $runCoordinatorType "CustomRunCoordinator" }}
         {{- include "dagsterYaml.runCoordinator.custom" . | indent 6 -}}
       {{- end }}
+    {{- else if (.Values.dagsterDaemon.enabled) }}
+    run_coordinator:
+      module: dagster._core.run_coordinator.immediately_launch_run_coordinator
+      class: ImmediatelyLaunchRunCoordinator
     {{- end }}
 
     {{- $computeLogManagerType := .Values.computeLogManager.type }}

diff --git a/python_modules/dagster-graphql/dagster_graphql_tests/graphql/graphql_context_test_suite.py b/python_modules/dagster-graphql/dagster_graphql_tests/graphql/graphql_context_test_suite.py
@@ -203,6 +203,10 @@ def _sqlite_instance_with_default_hijack():
                 with instance_for_test(
                     temp_dir=temp_dir,
                     overrides={
+                        "run_coordinator": {
+                            "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                            "class": "ImmediatelyLaunchRunCoordinator",
+                        },
                         "scheduler": {
                             "module": "dagster.utils.test",
                             "class": "FilesystemTestScheduler",
@@ -223,10 +227,14 @@ def postgres_instance_with_sync_run_launcher():
         def _postgres_instance():
             with graphql_postgres_instance(
                 overrides={
+                    "run_coordinator": {
+                        "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                        "class": "ImmediatelyLaunchRunCoordinator",
+                    },
                     "run_launcher": {
                         "module": "dagster._core.launcher.sync_in_memory_run_launcher",
                         "class": "SyncInMemoryRunLauncher",
-                    }
+                    },
                 }
             ) as instance:
                 yield instance
@@ -240,7 +248,14 @@ def _postgres_instance():
     def postgres_instance_with_default_run_launcher():
         @contextmanager
         def _postgres_instance_with_default_hijack():
-            with graphql_postgres_instance() as instance:
+            with graphql_postgres_instance(
+                overrides={
+                    "run_coordinator": {
+                        "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                        "class": "ImmediatelyLaunchRunCoordinator",
+                    },
+                }
+            ) as instance:
                 yield instance
 
         return MarkedManager(

diff --git a/python_modules/dagster/dagster/_core/instance/ref.py b/python_modules/dagster/dagster/_core/instance/ref.py
@@ -304,7 +304,9 @@ def config_defaults(base_dir: str) -> Mapping[str, Optional[ConfigurableClassDat
                 yaml.dump({}),
             ),
             "run_coordinator": ConfigurableClassData(
-                "dagster._core.run_coordinator", "DefaultRunCoordinator", yaml.dump({})
+                "dagster.core.run_coordinator",
+                "QueuedRunCoordinator",
+                yaml.dump({}),
             ),
             "run_launcher": ConfigurableClassData(
                 "dagster",

diff --git a/python_modules/dagster/dagster/_core/run_coordinator/default_run_coordinator.py b/python_modules/dagster/dagster/_core/run_coordinator/default_run_coordinator.py
@@ -1,52 +1,6 @@
-import logging
-from typing import Mapping, Optional
+from dagster._core.run_coordinator.immediately_launch_run_coordinator import (
+    ImmediatelyLaunchRunCoordinator,
+)
 
-from typing_extensions import Self
-
-import dagster._check as check
-from dagster._config.config_schema import UserConfigSchema
-from dagster._core.run_coordinator.base import RunCoordinator, SubmitRunContext
-from dagster._core.storage.dagster_run import DagsterRun, DagsterRunStatus
-from dagster._serdes import ConfigurableClass, ConfigurableClassData
-
-
-class DefaultRunCoordinator(RunCoordinator, ConfigurableClass):
-    """Immediately send runs to the run launcher."""
-
-    def __init__(self, inst_data: Optional[ConfigurableClassData] = None):
-        self._inst_data = check.opt_inst_param(inst_data, "inst_data", ConfigurableClassData)
-        self._logger = logging.getLogger("dagster.run_coordinator.default_run_coordinator")
-        super().__init__()
-
-    @property
-    def inst_data(self) -> Optional[ConfigurableClassData]:
-        return self._inst_data
-
-    @classmethod
-    def config_type(cls) -> UserConfigSchema:
-        return {}
-
-    @classmethod
-    def from_config_value(
-        cls, inst_data: Optional[ConfigurableClassData], config_value: Mapping[str, object]
-    ) -> Self:
-        return cls(inst_data=inst_data, **config_value)
-
-    def submit_run(self, context: SubmitRunContext) -> DagsterRun:
-        dagster_run = context.dagster_run
-
-        if dagster_run.status == DagsterRunStatus.NOT_STARTED:
-            self._instance.launch_run(dagster_run.run_id, context.workspace)
-        else:
-            self._logger.warning(
-                f"submit_run called for run {dagster_run.run_id} with status "
-                f"{dagster_run.status.value}, skipping launch."
-            )
-
-        run = self._instance.get_run_by_id(dagster_run.run_id)
-        if run is None:
-            check.failed(f"Failed to reload run {dagster_run.run_id}")
-        return run
-
-    def cancel_run(self, run_id: str) -> bool:
-        return self._instance.run_launcher.terminate(run_id)
+# for backwards compatibility, we need to keep the old DefaultRunCoordinator
+DefaultRunCoordinator = ImmediatelyLaunchRunCoordinator
diff --git a/python_modules/dagster/dagster/_core/run_coordinator/immediately_launch_run_coordinator.py b/python_modules/dagster/dagster/_core/run_coordinator/immediately_launch_run_coordinator.py
@@ -0,0 +1,52 @@
+import logging
+from typing import Mapping, Optional
+
+from typing_extensions import Self
+
+import dagster._check as check
+from dagster._config.config_schema import UserConfigSchema
+from dagster._core.run_coordinator.base import RunCoordinator, SubmitRunContext
+from dagster._core.storage.dagster_run import DagsterRun, DagsterRunStatus
+from dagster._serdes import ConfigurableClass, ConfigurableClassData
+
+
+class ImmediatelyLaunchRunCoordinator(RunCoordinator, ConfigurableClass):
+    """Immediately send runs to the run launcher."""
+
+    def __init__(self, inst_data: Optional[ConfigurableClassData] = None):
+        self._inst_data = check.opt_inst_param(inst_data, "inst_data", ConfigurableClassData)
+        self._logger = logging.getLogger("dagster.run_coordinator.default_run_coordinator")
+        super().__init__()
+
+    @property
+    def inst_data(self) -> Optional[ConfigurableClassData]:
+        return self._inst_data
+
+    @classmethod
+    def config_type(cls) -> UserConfigSchema:
+        return {}
+
+    @classmethod
+    def from_config_value(
+        cls, inst_data: Optional[ConfigurableClassData], config_value: Mapping[str, object]
+    ) -> Self:
+        return cls(inst_data=inst_data, **config_value)
+
+    def submit_run(self, context: SubmitRunContext) -> DagsterRun:
+        dagster_run = context.dagster_run
+
+        if dagster_run.status == DagsterRunStatus.NOT_STARTED:
+            self._instance.launch_run(dagster_run.run_id, context.workspace)
+        else:
+            self._logger.warning(
+                f"submit_run called for run {dagster_run.run_id} with status "
+                f"{dagster_run.status.value}, skipping launch."
+            )
+
+        run = self._instance.get_run_by_id(dagster_run.run_id)
+        if run is None:
+            check.failed(f"Failed to reload run {dagster_run.run_id}")
+        return run
+
+    def cancel_run(self, run_id: str) -> bool:
+        return self._instance.run_launcher.terminate(run_id)
diff --git a/python_modules/dagster/dagster_tests/core_tests/instance_tests/test_instance.py b/python_modules/dagster/dagster_tests/core_tests/instance_tests/test_instance.py
@@ -368,12 +368,14 @@ def test_get_required_daemon_types():
         SchedulerDaemon,
         SensorDaemon,
     )
+    from dagster._daemon.run_coordinator import QueuedRunCoordinatorDaemon
 
     with instance_for_test() as instance:
         assert instance.get_required_daemon_types() == [
             SensorDaemon.daemon_type(),
             BackfillDaemon.daemon_type(),
             SchedulerDaemon.daemon_type(),
+            QueuedRunCoordinatorDaemon.daemon_type(),
             AssetDaemon.daemon_type(),
         ]
 
@@ -390,6 +392,7 @@ def test_get_required_daemon_types():
             SensorDaemon.daemon_type(),
             BackfillDaemon.daemon_type(),
             SchedulerDaemon.daemon_type(),
+            QueuedRunCoordinatorDaemon.daemon_type(),
             MonitoringDaemon.daemon_type(),
             AssetDaemon.daemon_type(),
         ]
@@ -403,6 +406,7 @@ def test_get_required_daemon_types():
             SensorDaemon.daemon_type(),
             BackfillDaemon.daemon_type(),
             SchedulerDaemon.daemon_type(),
+            QueuedRunCoordinatorDaemon.daemon_type(),
         ]
 
 

diff --git a/python_modules/dagster/dagster_tests/daemon_sensor_tests/conftest.py b/python_modules/dagster/dagster_tests/daemon_sensor_tests/conftest.py
@@ -38,7 +38,11 @@ def submit_executor(request):
 def instance_module_scoped_fixture() -> Iterator[DagsterInstance]:
     with instance_for_test(
         overrides={
-            "run_launcher": {"module": "dagster._core.test_utils", "class": "MockedRunLauncher"}
+            "run_launcher": {"module": "dagster._core.test_utils", "class": "MockedRunLauncher"},
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
         }
     ) as instance:
         yield instance

diff --git a/python_modules/dagster/dagster_tests/daemon_sensor_tests/test_run_status_sensors.py b/python_modules/dagster/dagster_tests/daemon_sensor_tests/test_run_status_sensors.py
@@ -48,9 +48,15 @@
 @pytest.fixture(name="instance_module_scoped", scope="module")
 def instance_module_scoped_fixture() -> Iterator[DagsterInstance]:
     # Overridden from conftest.py, uses DefaultRunLauncher since we care about
-    # runs actually completing for run status sensors
+    # runs actually completing for run status sensors.
+    # Still uses DefaultRunCoordinator, since we don't need to check dequeuing logic.
     with instance_for_test(
-        overrides={},
+        overrides={
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
+        },
     ) as instance:
         yield instance
 

diff --git a/python_modules/dagster/dagster_tests/daemon_sensor_tests/test_sensor_run.py b/python_modules/dagster/dagster_tests/daemon_sensor_tests/test_sensor_run.py
@@ -1432,6 +1432,10 @@ def test_launch_failure(caplog, executor, workspace_context, remote_repo):
                 "module": "dagster._core.test_utils",
                 "class": "ExplodingRunLauncher",
             },
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
         },
     ) as instance:
         with freeze_time(freeze_datetime):

diff --git a/python_modules/dagster/dagster_tests/daemon_tests/conftest.py b/python_modules/dagster/dagster_tests/daemon_tests/conftest.py
@@ -29,6 +29,10 @@ def instance_module_scoped_fixture() -> Iterator[DagsterInstance]:
                     "module": "dagster._core.launcher.sync_in_memory_run_launcher",
                     "class": "SyncInMemoryRunLauncher",
                 },
+                "run_coordinator": {
+                    "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                    "class": "ImmediatelyLaunchRunCoordinator",
+                },
                 "event_log_storage": {
                     "module": "dagster._core.storage.event_log",
                     "class": "ConsolidatedSqliteEventLogStorage",

diff --git a/...ules/dagster/dagster_tests/declarative_automation_tests/daemon_tests/test_asset_daemon.py b/...ules/dagster/dagster_tests/declarative_automation_tests/daemon_tests/test_asset_daemon.py
@@ -86,6 +86,10 @@ def get_daemon_instance(
                 "module": "dagster._core.launcher.sync_in_memory_run_launcher",
                 "class": "SyncInMemoryRunLauncher",
             },
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
             **(extra_overrides or {}),
         }
     ) as instance:

diff --git a/python_modules/dagster/dagster_tests/declarative_automation_tests/daemon_tests/test_e2e.py b/python_modules/dagster/dagster_tests/declarative_automation_tests/daemon_tests/test_e2e.py
@@ -97,6 +97,10 @@ def get_workspace_request_context(
 ):
     with instance_for_test(
         overrides={
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
             "run_launcher": {
                 "module": "dagster._core.launcher.sync_in_memory_run_launcher",
                 "class": "SyncInMemoryRunLauncher",

diff --git a/python_modules/dagster/dagster_tests/scheduler_tests/conftest.py b/python_modules/dagster/dagster_tests/scheduler_tests/conftest.py
@@ -28,7 +28,11 @@ def submit_executor(request):
 def instance_session_scoped_fixture() -> Iterator[DagsterInstance]:
     with instance_for_test(
         overrides={
-            "run_launcher": {"module": "dagster._core.test_utils", "class": "MockedRunLauncher"}
+            "run_launcher": {"module": "dagster._core.test_utils", "class": "MockedRunLauncher"},
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
         }
     ) as instance:
         yield instance

diff --git a/python_modules/dagster/dagster_tests/scheduler_tests/test_scheduler_run.py b/python_modules/dagster/dagster_tests/scheduler_tests/test_scheduler_run.py
@@ -1313,6 +1313,10 @@ def test_launch_failure(
                 "module": "dagster._core.test_utils",
                 "class": "ExplodingRunLauncher",
             },
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
         },
     ) as scheduler_instance:
         schedule = remote_repo.get_schedule("simple_schedule")

diff --git a/python_modules/libraries/dagster-dask/dagster_dask_tests/test_graphql.py b/python_modules/libraries/dagster-dask/dagster_dask_tests/test_graphql.py
@@ -10,7 +10,14 @@
 
 
 def test_execute_hammer_through_webserver():
-    with instance_for_test() as instance:
+    with instance_for_test(
+        overrides={
+            "run_coordinator": {
+                "module": "dagster._core.run_coordinator.immediately_launch_run_coordinator",
+                "class": "ImmediatelyLaunchRunCoordinator",
+            },
+        }
+    ) as instance:
         with get_workspace_process_context_from_kwargs(
             instance,
             version="",