Request support notification for generic webhook on failure LaunchPlan or Workflows

[HKisCoding]

Hello, I am developing some base ML workflow for training and inference for my team. It worked well but I got some problems in sending notification when my workflow failed due to error such as Task Out of Memory.

I noticed that Flyte supports NotificationTask for LaunchPlan but it not support sending notification through Google Chat webhook. And for fl.workflow() on_failure options, I have to create another task with exactly the same input args of workflow, which make it difficult to create a generic Notification Task when any workflow fail.

Is there any work around on this or any update for this case ?

For example:

from flytekit.extras.webhook import WebhookTask

def get_notification_on_failure_task(
    url: str = GoogleChatWebhook.MLOPS_INTERNAL_ALERT.value,
):
    return WebhookTask(
        name="task-failure-notification",
        url=url,
        method="POST",
        headers={"Content-Type": "application/json"},
        data={
            "text": "<{inputs.user_ids}>",
            "cardsV2": [
                {
                    "cardId": "createCardMessage",
                    "card": {
                        "sections": [
                            {
                                "widgets": [
                                    {
                                        "textParagraph": {
                                            "text": "Owner: <b>{inputs.user_name}</b>\nTask: <b>{inputs.task_name}</b>\nException: <i>{inputs.exception}</i>\n",
                                            "maxLines": 6,
                                        }
                                    },
                                    {
                                        "buttonList": {
                                            "buttons": [
                                                {
                                                    "text": "View Flyte Execution",
                                                    "onClick": {
                                                        "openLink": {
                                                            "url": "{inputs.workflow_url}"
                                                        }
                                                    },
                                                }
                                            ]
                                        }
                                    },
                                ]
                            }
                        ]
                    },
                }
            ],
        },
        dynamic_inputs={
            "user_ids": str,
            "user_name": str,
            "task_name": str,
            "exception": str,
            "workflow_url": str,
        },
        show_data=True,
        show_url=True,
        description="Send notification on workflow failure",
    )

Then I create a generic task

@fl.task
def alert_ggchat_on_failure_task(
    username: str,
    func_name: str,
    exception: str,
    group_name: GoogleChatWebhook | None = None,
) -> None:
    """Alert on failure task with given username and group name

    Args:
        username (str): name of the user to be notified
        group_name (GoogleChatWebhook): group name to be notified
    """
    user_id = get_ggchat_id(username)
    execution_id = fl.current_context().execution_id.name
    domain = fl.current_context().execution_id.domain

    if group_name:
        url = group_name.value
    elif domain == "production":
        url = GoogleChatWebhook.PRODUCTION.value
    else:
        url = GoogleChatWebhook.DEVELOPMENT.value

    notification_on_failure_task = get_notification_on_failure_task(url=url)
    notification_on_failure_task(
        user_ids=user_id,
        user_name=username,
        task_name=func_name,
        exception=exception,
        workflow_url=WORKFLOW_URL.format(domain=domain, execution_id=execution_id),
    )

@fl.workflow(on_failure: alert_ggchat_on_failure_task)
def some_wf(
  username: str,
  func_name: str,
  exception: str,
  .... # some other *args
):
   ...