Passing and retrieving data in the `on_failure` hook

Hi, I have some questions about using the on_failure hook that was recently released: prefect/RELEASE-NOTES.md at main · PrefectHQ/prefect · GitHub. Apologies for the upcoming lengthy post:

1. How can we pass task parameters to the hook?

I have tried it with this sample code:

from prefect import flow, task
from pprint import pprint

def task_hook(task, task_run, state):
    # pprint(task.__dict__, indent=2)
    pprint(task_run.__dict__, indent=2)
    # pprint(state.__dict__, indent=2)

@task(on_failure=[task_hook])
def failing_task(message):
    raise ValueError(message)

@flow
def failing_flow(message):
    failing_task(message)

failing_flow("foobars")

In this case, I would like to retrieve the message param from within task_hook. After going through the output of task, task_run, and state, it looks like task_run has what I need:

{
  ...,
  'task_inputs': {'message': []},
  ...
}

But it is empty. Is there a way to retrieve the actual message that was passed to the task? It seems possible with flows - if I were to use a flow hook instead, flow_run will look like this:

{
  ...,
  'parameters': {'message': 'foobars'},
  ...
}

This leads to a second related question:

2. Is it possible to retrieve the value of a Secret Block in a hook?

With some updated code that passes a Secret instead of a plain string as a param:

from prefect import flow, task
from prefect.blocks.system import Secret

def flow_hook(flow, flow_run, state):
    print(flow_run.parameters)

@task
def failing_task(message: Secret):
    raise ValueError(message)

@flow(on_failure=[flow_hook])
def failing_flow(message: Secret):
    failing_task(message)

failing_flow(Secret(value="foobars"))

The output of flow_run.parameters in flow_hook looks like this:

{'message': {'value': '**********', 'block_type_slug': 'secret'}}

Normally, we’d use .get() to retrieve the value of a Secret block, but in this case, how can we get the unobfuscated value?

Last question:

3. How can we retrieve more of the stack trace from within the hook?

It looks like the error message itself is accessible, by printing state.message, which gives:

Task run encountered an exception: ValueError: foobars

But can we also get part of the stack trace as well? The Prefect error logs are useful to pinpoint the source of the error since they have file names and line numbers:

...
  File "/FILE_PATH/test.py", line 44, in failing_task
    raise ValueError(message)
ValueError: foobars

But it does not seem to be present in the data that is sent to the hook.

Thanks for your help!

2 Likes

I believe the state.data has some of the result of the task run.

I have a similar question - in particular I want to get the stack trace of the error that caused the failure and send that to another service

Regarding the first part of the question, one can use prefect.states.get_state_exception to retrieve the exception from the state. Then one can get the arguments of the exception as usual.

from prefect import flow, task
from prefect.states import get_state_exception
from pprint import pprint

def task_hook(task, task_run, state):
    # pprint(task.__dict__, indent=2)
    # pprint(task_run.__dict__, indent=2)
    # pprint(state.__dict__, indent=2)
    e = get_state_exception(state)
    pprint(e.args)

@task(on_failure=[task_hook])
def failing_task(message):
    raise ValueError(message)

@flow
def failing_flow(message):
    failing_task(message)

failing_flow("foobars")
1 Like