Hi Dagster team! We have a run status sensor set up to perform operations on graph success, running every 10s with

minimum_interval_seconds=10

. We noticed that each time the sensor runs, it only processes one succeeded graph, even if 50 have succeeded over the past 10s. This is heavily problematic because the sensor requires 10 minutes to process 60 completed jobs, which doesn't scale when we need to handle hundreds. Is there a way we can get the sensor to process more than one pipeline in one go?

What version of Dagster are you running? We improved throughput for the sensor daemon in 0.15.5 with the introduction of a a threadpool in the sensor daemon

@rex This is on Dagster 0.15.8

How have you deployed Dagster? The threadpool feature is currently opt-in, so it’s not enabled by default. If you’re using helm, you can use

.Values.sensors.useThreads

to enable the feature

Oh, let me give it a try immediately. Thanks for the heads up!

is your daemon also reading from that dagster.yaml?

It should? The file is present in the

DAGSTER_HOME

folder of the Daemon and the value is shown in Dagit's Status>Configuration page.

you can check to see if the daemon process is spawning thread workers: https://sourcegraph.com/github.com/dagster-io/dagster/-/blob/python_modules/dagster/dagster/_daemon/sensor.py?L205%3A63=

Yes I can see 8 threads in the output of

pstree -a

/app/.runtimes/python-3.10.0-2/bin/poetry run dagster-daemon run
    └─python /app/.runtimes/python-3.10.0-2/bin/dagster-daemon run
        └─8*[{python}]

Yeah, I think the issue here is the fact that we don’t duplicate run_status_sensor invocations even if it’s used for multiple jobs. The sensor will fire just once per interval.

Thanks for looking into it!

The main thing I can think of is to bypass the

run_status_sensor

machinery and run your own sensor that queries the event log for run success events after a particular query… the body of your sensor could then execute your logic for each event that happens.

Oh I wasn't aware there was a way to get succeeded jobs from the event log; that's something to explore on our end yes, thanks!

well, there are success events for every run

Sounds like we may want to wait for Dagster to do it natively. 😅

Want to call out that we just ran into this as well at Whatnot, when I switched the cursor logic from a "query the run records via GraphQL every 30 seconds" to a more "native" implementation using

run_status_sensor

, and it caused our workflows to get out of sync by several hours (and was pretty unclear to me why). our situation was resolved by removing

minimum_interval_seconds

@VxD this is what our sensors looked like previously and it worked fine at our scale (not huge)

@sensor(job=my_yielded_job)
def my_job_checking_sensor(context):
    job_name_to_check = "foo_job"
 
    time_window_start = datetime.now() - timedelta(seconds=1800)
    run_records = context.instance.get_run_records(
        filters=RunsFilter(
            job_name=job_name_to_check,
            statuses=[PipelineRunStatus.SUCCESS, PipelineRunStatus.FAILURE],
            updated_after=time_window_start,
        ),
        order_by="update_timestamp",
        ascending=False,
    )
    for run_record in run_records:
        yield RunRequest(run_key=str(run_record.storage_id))

Ah, thanks Stephen! That looks like an elegant work around.

Yeah, I think we need to add better support for processing multiple runs per sensor iteration in the run status sensor, but @Stephen Bailey’s workaround of managing your own sensor is a good one. I’ll create a separate issue to track this framework improvement.

Created issue at: https://github.com/dagster-io/dagster/issues/9458

Thanks for looking into it! Dagster has been working great for us so far! Getting it to scale horizontally is definitely an interesting challenge.