Fix: terminate hanging jobs

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 The changelog lists most feature changes between each release. 
 
+## Upcoming release
+* Fix: on startup, terminate runs still in started/starting state, as dagster doesn't terminate them cleanly on shutdown.
+* Fix: enable run monitoring to terminate jobs hanging on startup/cancellation (after 180s) or running for more than 6h
 
 ## 2025-01-28
 - Fix: [create primary key if missing](https://github.com/mobidata-bw/ipl-dagster-pipeline/pull/182)

dagster.Dockerfile

@@ -27,6 +27,10 @@ LABEL org.opencontainers.image.licenses="(EUPL-1.2)"
 
 EXPOSE 3000
 
+COPY scripts/ /opt/dagster/app/scripts/
+
+ENTRYPOINT ["/opt/dagster/app/scripts/start_runs_termination_script_in_background.sh"]
+
 CMD ["dagster-webserver", "-h", "0.0.0.0", "-p", "3000", "-w", "workspace.yaml"]
 
 FROM base AS daemon

dagster.docker.yaml

@@ -13,6 +13,17 @@ run_coordinator:
     dequeue_use_threads: true
     dequeue_interval_seconds: 1
 
+run_monitoring:
+  enabled: true
+  start_timeout_seconds: 180
+  cancel_timeout_seconds: 180
+  max_resume_run_attempts: 3
+  poll_interval_seconds: 120
+  # max_runtime_seconds 21600s = 6h
+  # to overwrite this for individual jobs, see 
+  # https://docs.dagster.io/guides/deploy/execution/run-monitoring#general-run-timeouts
+  max_runtime_seconds: 21600
+
 run_storage:
   module: dagster_postgres.run_storage
   class: PostgresRunStorage

scripts/start_runs_termination_script_in_background.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+set -eo pipefail
+set -x
+
+echo "Starting terminate_starting_and_started_runs in background to terminate orphaned ones."
+python /opt/dagster/app/scripts/terminate_starting_and_started_runs.py &
+
+# Hand off to the CMD
+exec "$@"

scripts/terminate_starting_and_started_runs.py

@@ -0,0 +1,59 @@
+import logging
+import socket
+import time
+
+from dagster import DagsterRunStatus
+from dagster_graphql import DagsterGraphQLClient, DagsterGraphQLClientError
+from requests.exceptions import ConnectionError
+
+logging.basicConfig()
+logger = logging.getLogger(__file__.rsplit('/', 1)[1])
+logger.setLevel(logging.INFO)
+
+client = DagsterGraphQLClient('localhost', port_number=3000)
+
+GET_RUNS_QUERY = '''
+query RunsQuery ($filter: RunsFilter) {
+  runsOrError(
+    filter: $filter
+  ) {
+    __typename
+    ... on Runs {
+      results {
+        runId
+        jobName
+        status
+        runConfigYaml
+        startTime
+        endTime
+      }
+    }
+  }
+}
+'''
+
+
+def get_run_ids_of_runs(status: list[str], timeout: int = 20) -> list[str]:
+    variables = {'filter': {'statuses': status}}
+
+    start_time = time.perf_counter()
+    while True:
+        try:
+            response = client._execute(GET_RUNS_QUERY, variables)
+            return [run['runId'] for run in response['runsOrError']['results']]
+        except DagsterGraphQLClientError as ex:
+            if isinstance(ex.__cause__, ConnectionError):
+                time.sleep(0.1)
+                if time.perf_counter() - start_time >= timeout:
+                    raise TimeoutError('Waited too long for the Dagster Webserver startup') from ex
+            else:
+                raise
+
+
+run_ids = get_run_ids_of_runs(['STARTED', 'STARTING'])
+
+if len(run_ids) > 0:
+    logger.info(f'Terminating runs {run_ids}')
+    client.terminate_runs(run_ids)
+else:
+    logger.info('No run in state STARTED or STARTING')