Get failed asset from dagster run with @run_status_sensor/@run_failure_sensor

[ssillaots-boku]

I want to come up with a @run_status_sensor/@run_failure_sensor that will only alert about the certain asset that failed in the run. An example (👇) where there are 3 assets in a chain will make this much more clear.

As you can see in the middle asset there's an intentional error.

If I run these assets in one run then upstream gets materialized, middle fails and the downstream doesn't start (grey in run view, but technically failed).

Is there a way to find out from the run which exact asset failed. Why I want to do this? Lets say that every asset is in a different code location or in a different group like in the example. And every group/code location has an owner. If a run fails then I want to send a slack alert only to the asset's owner which means I need to figure out from the run which asset failed.

Right now haven't figured out a way to do that. I can see the run that failed but I wouldn't want to send an alert to all of the owners of those assets. Only to the owner of the asset that actually failed.

So in this case I would like to a slack alert only to engineering group.

@asset(group='finance')
def upstream() -> None:
   return

@asset(deps=[upstream],
      group='engineering')
def middle() -> None:
   wow
   return

@asset(deps=[middle],
      group='infra')
def downstream() -> None:
   return

[ssillaots-boku]

There is a workaround (or this) but that solves only half the problem.

Based on that example middle and downstream assets would be filtered in with DagsterEventType.ASSET_MATERIALIZATION_PLANNED. So I would end up sending slack alerts about middle and downstream assets. But I only want to send the alert about the asset which failed and that's middle.

I was able to get asset naming from event_type_value='STEP_FAILURE' where in logging_tags dict there's a step_key. But if in the lineage there's a dbt asset which is "imported" to Dagster project with @dbt_assets where there are many models in select then this approach goes out of the window... In that case step_key value would be @dbt_assets definition name not individual asset's name.

[OwenKephart]

At the moment, there is unfortunately no general way to detect the specific asset within a multi-asset that caused the failure (as with @dbt_assets). When the step fails, that simply marks all assets that did not get materialized as failed. No information about the specific dbt model that caused the failure is extracted from the command. For dbt specifically, the best option here would likely involve manually emitting some sort of specific log (e.g. an AssetObservation) during that step in the case of failure that would let you detect the specific dbt key that caused the failure.