monitored_jobs doesn't seem to work on sensors

[eliot1785]

I am new to Dagster. I've created a couple of sensors to watch for failures. I want 2 different failure sensors, looking at distinct groups of jobs using the monitored_jobs sensor parameter. Something like this:

severity=1
monitored_jobs=sev1_jobs
@run_failure_sensor(
    name=f"sev{str(severity)}_failure_sensor",
    description=f"Detects failures in jobs whose failures are Sev{severity}",
    monitored_jobs=monitored_jobs,
):
def my_sensor1(context: RunFailureSensorContext):
    job_name = context.dagster_run.job_name
    context.log.info(f"Found failure for {job_name}")
    ...

severity=2
monitored_jobs=sev2_jobs
@run_failure_sensor(
    name=f"sev{str(severity)}_failure_sensor",
    description=f"Detects failures in jobs whose failures are Sev{severity}",
    monitored_jobs=monitored_jobs,
):
def my_sensor2(context: RunFailureSensorContext):
    job_name = context.dagster_run.job_name
    context.log.info(f"Found failure for {job_name}")

I've also tried this with using @run_status_sensor while additionally specifying run_status=DagsterRunStatus.FAILURE.

Regardless of what I do, both sensors pick up a failure in any job, as if monitored_jobs is not being used to reduce the scope of what is monitored. I can clearly see that a failure in a job which is in sev1_jobs also causes the other sensor intended for sev2_jobs to execute, based on looking at the logs for that sensor.

One thing that is also making this harder to diagnose is that I cannot find the monitored_jobs property anywhere, whether in the UI or by printing for example my_sensor2.monitored_jobs. So I cannot absolutely verify that the monitored_jobs properties are correct, but I believe that they are.

monitored_jobs for me is a list of UnresolvedAssetJobDefinition objects, each individually determined by define_asset_job() and placed in a Python list before being sent over.

I'm using 1.9.9. Thanks for your help.

[eliot1785]

Does anybody at least have any ideas for how to further diagnose this?

[ion-elgreco]

This guide shows how to create run status sensor context: https://docs.dagster.io/guides/automate/sensors/testing-run-status-sensors

You could use that and then put breakpoints in the sensor code and run it through a debugger