Vertex Pipelines are written in the Kubeflow Pipelines SDK. A fairly common pattern in any pipeline would be:

if job has failed
send failure notification

I Kubeflow Pipelines this would normally be specified as:

exit_task = send_notification(msg='{{workflow.name}} has status {{workflow.status}}.')

with dsl.ExitHandler(exit_task):
    download_task = gcs_download_op(url)
    echo_task = print_file(download_task.output)

Problem you have encountered:

The '{{workflow.name}}' and '{{workflow.status}}' would normally be resolved by Argo Workflows. In Vertex these placeholders are not resolved and so it seems impossible to get hold of the Pipeline status during execution for the purposes of sending alerts.

I am not sure if Vertex Pipelines uses Argo under the hood?

What you expected to happen:

I would expect '{{workflow.status}}' to be resolved or documentation to be provided on how to obtain the PipelineJob status during runtime.

Steps to reproduce:

Simple case:

from google.cloud import aiplatform
from kfp.v2 import dsl
from kfp.v2 import compiler


@dsl.component(kfp_package_path=_KFP_PACKAGE_PATH)
def hello_world(text: str) -> str:
    print(text)
    return text


@dsl.pipeline(name='hello-world', description='A simple intro pipeline')
def pipeline(text: str = 'hi there'):
    '''Pipeline that passes small pipeline parameter string to consumer op'''

    exit_task = hello_world('{{workflow.name}} has status {{workflow.status}}')

    with dsl.ExitHandler(exit_task):
        consume_task = hello_world(
            text
        )  # Passing pipeline parameter as argument to consumer op

Other information (workarounds you have tried, documentation consulted, etc):

Issue 203956058

Links (5)

Issue metadata