Improve Execution Notifications plugins to allow passing custom payload

[wrussell1999]

Feature description

Add the ability to be able to get the {{ errorLogs() }} function inside of a Flow Trigger that runs when flows fail. This allows for instance wide error handling with useful errors inside of them.

From a user in Slack:

I'm using a triggered Slack alert flow for all failed flows in my workspace as described in best practices here. Is there any way to get the new {{ errorLogs() }} pebble function from the v0.20 release into the trigger attributtes, so that my Slack alert flow can display them? (without having to push it into outputs for every single flow)

[loicmathieu]

Better to use the io.kestra.plugin.core.log.Fetch task for that, the errorLog() function has been defined to load logs for the current execution only. There are RBAC considerations when loading logs from a different flow.

[Ben8t]

Agree with Loic, this one sounds like we need more info too. @wrussell1999 do you have the user name raising this need? I would like to have more details on the flow and corresponding trigger involved - reading the issue title "Improve Execution Notifications plugins to allow passing custom payload" it sounds like a broader subject around notification isn't?

[MarthaScheffler]

@Ben8t it's me :)
here is the Slack link: https://kestra-io.slack.com/archives/C03FQKXRK3K/p1734105888969879
wanna take the discussion here or via Slack?

[Ben8t]

Ah great :)

Let's keep the discussion here as we have the proper issue!
Can you send me a sample flow definition where you have your Slack alert/trigger please? Just want to be sure to have understood the point here first

[MarthaScheffler]

My setup: I get a Slack alert for all flows that end up with either FAILED or WARNED in the respective namespace. prod alerts go to the prod alert channel, dev alerts go to the dev alert channel. However, the error message is a bit crude at the moment, I only managed to include in the Slack message which flow, execution_id, and where the link is to access the logs.

I would love to additionally display part of the error logs in the Slack message. I haven't yet managed to look more into the log.Fetch task, and if it would help me here. I thought that the new {{ errorLogs() }} pebble function could cover my needs, but it's not available via trigger.

my slack alert task:

id: slack_alert
namespace: dev.slack
description: Send an alert to a Slack channel

variables:
  env: "{{ envs.env == 'prod' ? '' : '(dev)' }}"

tasks:
      - id: send_alert
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}" # https://hooks.slack.com/services/xzy/xyz/xyz
        # https://app.slack.com/block-kit-builder
        payload: |
          {
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": ":warning: {{ trigger.namespace ?? 'manual' }}.{{ trigger.flowId ?? 'triggered' }} - {{ trigger.state ?? 'MANUAL' }} {{ vars.env }} :warning:",
                  "emoji": true
                }
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "The _{{ trigger.flowId ?? flow.id }}_ flow has ended with {{ trigger.state ?? 'MANUAL' }} on execution_id <http://localhost:8080/ui/executions/{{ trigger.namespace ?? flow.namespace }}/{{ trigger.flowId ?? flow.id }}/{{ trigger.executionId ?? execution.id }}/logs|*{{ trigger.executionId ?? execution.id }}*>"
                }
              },
              {
                "type": "divider"
              },
              {
                "type": "context",
                "elements": [
                  {
                    "type": "plain_text",
                    "text": "Kestra {{ envs.env }} pod needs to be forwarded to localhost to be able to access the logs.",
                    "emoji": true
                  }
                ]
              }
            ]
          }

triggers:
  - id: failed_workflows
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionStatus
        in:
          - FAILED
          - WARNING
      - type: io.kestra.plugin.core.condition.Or
        conditions:
          - type: io.kestra.plugin.core.condition.ExecutionNamespace
            # namespace will be substituted with prod by git flows
            namespace: dev
            prefix: true
          - type: io.kestra.plugin.core.condition.ExecutionNamespace
            namespace: system
            prefix: true

[Ben8t]

Thanks for the details, very clear now :)

The good thing with the fetch.log task is that you can provide the trigger.executionId to fetch corresponding logs (the task outputs to file uri in internal storage). Then you can probably use a {{ read(outputs.fetch_log_task.uri) }} inside you Slack payload. Let us know if that works for you

[MarthaScheffler]

It kind of works.
I had to add a transformation from ion to json, and then I could read the first error message. not the others though, as my conversion onto something readable only catches the first error. I will add my outputs as well, hoping for a hint, how I can do a better conversion :)

adjusted workflow:

id: slack_alert
namespace: dev.slack
description: Send an alert to a Slack channel

variables:
  env: "{{ envs.env == 'prod' ? '' : '(dev)' }}"

tasks:
      - id: get_error_logs
        type: io.kestra.plugin.core.log.Fetch
        level: WARN
        executionId: "{{ trigger.executionId ?? execution.id }}"
        namespace: "{{ trigger.namespace ?? 'manual' }}"
        flowId: "{{ trigger.flowId ?? 'triggered' }}"

      - id: read_error_logs
        type: io.kestra.plugin.serdes.json.IonToJson
        from: "{{ outputs.get_error_logs.uri }}"

      - id: send_alert
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}" # https://hooks.slack.com/services/xzy/xyz/xyz
        # https://app.slack.com/block-kit-builder
        payload: |
          {
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": ":warning: {{ trigger.namespace ?? 'manual' }}.{{ trigger.flowId ?? 'triggered' }} - {{ trigger.state ?? 'MANUAL' }} {{ vars.env }} :warning:",
                  "emoji": true
                }
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "The _{{ trigger.flowId ?? flow.id }}_ flow has ended with {{ trigger.state ?? 'MANUAL' }} on execution_id <http://localhost:8080/ui/executions/{{ trigger.namespace ?? flow.namespace }}/{{ trigger.flowId ?? flow.id }}/{{ trigger.executionId ?? execution.id }}/logs|*{{ trigger.executionId ?? execution.id }}*>"
                }
              },
              {
                "type": "divider"
              },
              {
                "type": "context",
                "elements": [
                  {
                    "type": "plain_text",
                    "text": "Kestra {{ envs.env }} pod needs to be forwarded to localhost to be able to access the logs.",
                    "emoji": true
                  }
                ]
              },
              {
                "type": "divider"
              },
              {
                "type": "context",
                "elements": [
                  {
                    "type": "plain_text",
                    # this only gets the first error
                    "text": "{{ outputs.get_error_logs.size == 0 ? 'details see logs' : json(read(outputs.read_error_logs.uri)).message }}",
                    "emoji": true
                  }
                ]
              }
            ]
          }

triggers:
  - id: failed_workflows
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionStatus
        in:
          - FAILED
          - WARNING
      - type: io.kestra.plugin.core.condition.Or
        conditions:
          - type: io.kestra.plugin.core.condition.ExecutionNamespace
            # namespace will be substituted with prod by git flows
            namespace: dev
            prefix: true
          - type: io.kestra.plugin.core.condition.ExecutionNamespace
            namespace: system
            prefix: true

output of the get_error_logs task: