https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Write an automations.yaml for an auto...
# ask-marvin
r
@Marvin Write an automations.yaml for an automation to notify when a flow run fails, crashes or timesout in either the default or another workpool called backup. The subject and body of the notification should use go template to show the reader if the problem was a fail, crash or timeout. Assume the notification block_id will be inserted by yq later. Also tell me how to ensure that the automation stays current if the workpool id (so not the name but the unique hash) changes over time.
m
✅ thought for 147.8 seconds
A small but important correction: Prefect automations use Jinja templates, not Go templates. Here’s an 
automations.yaml
example for Prefect 3.x that notifies when a flow run fails, crashes, or times out, limited to flow runs associated with either the 
default
or 
backup
work pool.
Copy code
yaml
automations:
  - name: notify-on-flow-run-failure-crash-timeout
    description: Notify when a flow run fails, crashes, or times out in default or backup work pools
    enabled: true

    trigger:
      type: event
      posture: Reactive
      expect:
        - prefect.flow-run.Failed
        - prefect.flow-run.Crashed
        - prefect.flow-run.TimedOut
      threshold: 1
      match:
        prefect.resource.id: prefect.flow-run.*
      match_related:
        - prefect.resource.role: work-pool
          prefect.resource.name:
            - default
            - backup

    actions:
      - type: send-notification
        block_document_id: REPLACE_ME_WITH_YQ

        subject: >-
          {% if event.event == "prefect.flow-run.Failed" -%}
          Flow run FAILED: {{ flow_run.name }}
          {%- elif event.event == "prefect.flow-run.Crashed" -%}
          Flow run CRASHED: {{ flow_run.name }}
          {%- elif event.event == "prefect.flow-run.TimedOut" -%}
          Flow run TIMED OUT: {{ flow_run.name }}
          {%- else -%}
          Flow run issue: {{ flow_run.name }}
          {%- endif %}

        body: |-
          {% if event.event == "prefect.flow-run.Failed" -%}
          A flow run *failed*.
          {%- elif event.event == "prefect.flow-run.Crashed" -%}
          A flow run *crashed*.
          {%- elif event.event == "prefect.flow-run.TimedOut" -%}
          A flow run *timed out*.
          {%- else -%}
          A flow run encountered a problem.
          {%- endif %}

          *Flow run:* {{ flow_run.name }}
          *Flow run ID:* {{ flow_run.id }}
          *Flow:* {{ flow.name }}
          *State:* {{ flow_run.state.name }}
          *Message:* {{ flow_run.state.message or "No state message" }}

          *Work pool:* {{ work_pool.name if work_pool else "Unknown" }}
          *Deployment:* {{ deployment.name if deployment else "Ad hoc / unknown" }}

          *UI:* {{ flow_run | ui_url }}
Best way to keep this current if the work pool ID changes: Use work pool names, not IDs, in 
match_related
, exactly like above:
Copy code
yaml
match_related:
  - prefect.resource.role: work-pool
    prefect.resource.name:
      - default
      - backup
That way, if the underlying work pool UUID/hash changes over time, the automation still matches as long as the pool names remain 
default
and 
backup
. If you instead matched on:
Copy code
yaml
prefect.resource.id: prefect.work-pool.<uuid>
then you’d need to update the automation whenever that ID changes. So the practical answer is: Use 
prefect.resource.name
for work pools if you want the automation to stay current across work pool ID changes. Relevant docs: - Automations concepts - Create automations If you want, I can also give you: - a version formatted specifically for 
prefect deploy
consumption, or - a 
yq
command snippet to patch in the 
block_document_id
.
r
@Marvin Do I need to add a time element? i.e. I want this to trigger as soon as any of these events happen, so it the current yaml good enough to do that?
m
✅ thought for 9.5 seconds
Yes — your current YAML is good enough for that. You do not need to add a time element if you want the automation to fire as soon as a flow run enters 
Failed
, 
Crashed
, or 
TimedOut
. Why: - 
type: event
- 
posture: Reactive
- 
threshold: 1
means: trigger immediately when one matching event is observed. So this part is sufficient:
Copy code
yaml
trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Failed
    - prefect.flow-run.Crashed
    - prefect.flow-run.TimedOut
  threshold: 1
  match:
    prefect.resource.id: prefect.flow-run.*
  match_related:
    - prefect.resource.role: work-pool
      prefect.resource.name:
        - default
        - backup
You’d only add a time window like 
within
if you wanted behavior such as: - “2 failures within 10 minutes” - “fire only if something does not happen within X time” - “deduplicate/group events over a period” For your use case, no extra time field is needed. If you want, I can also sanity-check the full YAML structure against Prefect’s expected 
automations.yaml
format.
r
@Marvin what jinja would need to be added to the body to show just the main stack trace. I just want to show an extra line or two and not the entire stack trace
m
✅ thought for 13.2 seconds
Yes — you can add a small Jinja snippet, but there’s an important limitation: Prefect notification templates don’t provide a built-in “stack trace parser” filter. In practice, the traceback usually appears inside 
flow_run.state.message
, so the best you can do in-template is truncate that message or extract a small leading portion. A simple approach is to show only the first couple of lines of the state message:
Copy code
yaml
body: |-
  {% if event.event == "prefect.flow-run.Failed" -%}
  A flow run *failed*.
  {%- elif event.event == "prefect.flow-run.Crashed" -%}
  A flow run *crashed*.
  {%- elif event.event == "prefect.flow-run.TimedOut" -%}
  A flow run *timed out*.
  {%- else -%}
  A flow run encountered a problem.
  {%- endif %}

  *Flow run:* {{ flow_run.name }}
  *Flow run ID:* {{ flow_run.id }}
  *Flow:* {{ flow.name }}
  *State:* {{ flow_run.state.name }}

  *Main error excerpt:*
  {{ (flow_run.state.message or "No state message").split('\n')[:3] | join('\n') }}

  *Work pool:* {{ work_pool.name if work_pool else "Unknown" }}
  *Deployment:* {{ deployment.name if deployment else "Ad hoc / unknown" }}

  *UI:* {{ flow_run | ui_url }}
That will print only the first 3 lines of the message. If you want it even shorter, use just 1–2 lines:
Copy code
yaml
{{ (flow_run.state.message or "No state message").split('\n')[:2] | join('\n') }}
If instead you want to limit by character count rather than lines:
Copy code
yaml
{{ (flow_run.state.message or "No state message")[:300] }}...
A couple caveats: - For some failures, the most useful error text is at the top of `state.message`; for others, the exception summary may be at the bottom. - If the traceback is not present in 
flow_run.state.message
, the template won’t be able to pull it from logs automatically. If you want, I can rewrite your full 
body:
block with a clean “short error excerpt” section included.
r
@Marvin how to add a jinja if statement with OR statement. i.e. I only want the state message to display if it's failed or crashed and not if it's cancelled.
m
✅ thought for 9.8 seconds
Use normal Jinja boolean syntax with 
or
. For your case:
Copy code
yaml
{% if event.event == "prefect.flow-run.Failed" or event.event == "prefect.flow-run.Crashed" %}
*State message:* {{ flow_run.state.message or "No state message" }}
{% endif %}
If you want it a little cleaner, use 
in
instead:
Copy code
yaml
{% if event.event in ["prefect.flow-run.Failed", "prefect.flow-run.Crashed"] %}
*State message:* {{ flow_run.state.message or "No state message" }}
{% endif %}
That will show the state message only for Failed or Crashed, and not for Cancelled. Example in a body block:
Copy code
yaml
body: |-
  *Flow run:* {{ flow_run.name }}
  *State:* {{ flow_run.state.name }}

  {% if event.event in ["prefect.flow-run.Failed", "prefect.flow-run.Crashed"] %}
  *State message:* {{ flow_run.state.message or "No state message" }}
  {% endif %}

  *UI:* {{ flow_run | ui_url }}
If you also want to include 
TimedOut
, just add it:
Copy code
yaml
{% if event.event in ["prefect.flow-run.Failed", "prefect.flow-run.Crashed", "prefect.flow-run.TimedOut"] %}
Jinja docs: - Jinja if statements
Open in Slack
PreviousNext
Powered by