Tate Wilks
09/17/2024, 7:02 PMMarvin
09/17/2024, 7:02 PMon_failure
hooks, within the Prefect Flow that called them. Here are some useful resources:
### Prefect 2.x
1. Viewing Logs in Prefect UI:
You can see logs for a flow run in the Prefect UI by navigating to the Flow Runs page and selecting a specific flow run to inspect. This will include logs for task runs as well.
Link: Prefect 2.x Logging Guide
2. Using State Change Hooks:
State change hooks execute code in response to changes in flow or task run states. You can use state change hooks to handle specific state transitions, such as logging failures.
Code Example:
python
from prefect import flow
from prefect.blocks.core import Block
from prefect.settings import PREFECT_API_URL
def notify_slack(flow, flow_run, state):
slack_webhook_block = Block.load("slack-webhook/my-slack-webhook")
slack_webhook_block.notify(f"Your job {flow_run.name} entered {state.name} with message: See the flow run in the UI: {PREFECT_API_URL.value()}/flow-runs/flow-run/{flow_run.id}")
@flow(on_failure=[notify_slack], retries=1)
def failing_flow():
raise ValueError("oops!")
if __name__ == "__main__":
failing_flow()
Link: State Change Hooks in Prefect 2.x
### Prefect 3.x
1. Logging Configuration:
By default, task runs and nested flow runs inherit the log_prints
setting from their parent flow run. You can customize logging behavior to ensure that logs are captured correctly.
Code Example:
python
from prefect import task, flow
@task
def my_task():
print("we're logging print statements from a task")
@flow(log_prints=True)
def my_flow():
print("we're logging print statements from a flow")
my_task()
my_flow()
Link: Logging in Prefect 3.x
2. Using State Change Hooks:
Similar to Prefect 2.x, Prefect 3.x also allows you to use state change hooks to handle task or flow state transitions.
Code Example:
python
from prefect import flow
from prefect.blocks.core import Block
from prefect.settings import PREFECT_API_URL
def notify_slack(flow, flow_run, state):
slack_webhook_block = Block.load("slack-webhook/my-slack-webhook")
slack_webhook_block.notify(f"Your job {flow_run.name} entered {state.name}")
@flow(on_failure=[notify_slack], retries=1)
def failing_flow():
raise ValueError("oops!")
if __name__ == "__main__":
failing_flow()
Link: State Management in Prefect 3.x
These resources should help you configure your Prefect flows to capture and view logs from tasks that have failed and triggered on_failure
hooks.