Limiting task run time in Dagster

[yuval-airis]

I see that occasionally, unclear to me exactly when, certain runs in my flow take forever (Possibly corrupted somehow).

Can I cap the runtime of each task somehow in dagster and make every computation that didn't finish by that time stop and fail?

[salazarm]

Hello please take a look at https://docs.dagster.io/deployment/run-monitoring#general-run-timeouts for configuring a timeout on a job.

Also see this discussion for another approach using a sensor to detect the long running job: #12463

Finally this is another possible approach that may work for your use case using signal:

def some_op():
      sub_process = subprocess.Popen(
        [my_script_path],
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        cwd=cwd,
        start_new_session=True,
    )
    def kill_process_group():
        sub_process.kill()
        pgid = os.getpgid(sub_process.pid)
        os.killpg(pgid, signal.SIGKILL)

    def timeout_handler(signum, frame):
        kill_process_group()
        raise TimeoutException("Script execution timed out")

    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(300)

    if sub_process.stdout:
        for line_bytes in sub_process.stdout:
            # Reset alarm whenever there is output
            signal.alarm(300)
            line = line_bytes.decode("utf-8")
            context.log.info(line.rstrip())

    sub_process.wait()
    return_code = sub_process.returncode

    if return_code:
        raise Failure(description="Shell command execution failed")

In the code above we reset the timeout if there is any log output, so we only kill the process after 5 minutes of no output.