Как использовать ExitHandler с Kubeflow Pipelines SDK v2

Question

Я пытаюсь перевести все свои конвейеры Kubeflow с использования предыдущего SDK v1 ( kfp), до более новой версии Pipelines SDK v2 (). я использую версию 1.8.12Этот рефакторинг оказался успешным почти для всего кода, за исключением , который все еще существует; from kfp.v2.dsl import ExitHandler. Похоже, что предыдущий способ компиляции объекта конвейера в tar.gz-файл с использованием kfp.compiler.Compiler().compile(pipeline, 'basic_pipeline.tar.gz')файл сохранил некоторые типы заполнителей Argo, в то время как новый .jsonтрубопроводы с использованием compiler.Compiler().compile(pipeline_func=pipeline, package_path="basic-pipeline.json")не работает так же. Ниже я подробно расскажу, что работает в Pipelines SDK v1 и как я пытался реализовать это в v2.

Раньше , используя Kubeflow Pipelines v1, я мог использовать ExitHandler, как показано в этом вопросе StackOverflow , например. отправить сообщение в Slack, когда один из компонентов конвейера вышел из строя. Я бы определил конвейер как

      import kfp.dsl as dsl

@dsl.pipeline(
    name='Basic-pipeline'
)
def pipeline(...):
    exit_task = dsl.ContainerOp(
        name='Exit handler that catches errors and post them in Slack',
        image='eu.gcr.io/.../send-error-msg-to-slack',
        arguments=[
                    'python3', 'main.py',
                    '--message', 'Basic-pipeline failed'
                    '--status', "{{workflow.status}}"
                  ]
    )
    with dsl.ExitHandler(exit_task):
        step_1 = dsl.ContainerOp(...)
        step_2 = dsl.ContainerOp(...) \
            .after(step_1)

if __name__ == '__main__':
    import kfp.compiler as compiler
    compiler.Compiler().compile(pipeline, 'basic_pipeline.tar.gz')

куда бы отправить messageв наш Slack, если какой-либо из шагов конвейера не удался. Код для exit_taskизображение выглядит как

      import argparse

def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--message', type=str)
    parser.add_argument('--status', type=str)
    return parser.parse_known_args()

def main(FLAGS):
    def post_to_slack(msg):
        ...

    if FLAGS.status == "Failed":
        post_to_slack(FLAGS.message)
    else:
        pass

if __name__ == '__main__':
    FLAGS, unparsed = get_args()
    main(FLAGS)

Это сработало, потому что базовый рабочий процесс Argo каким-то образом мог понять "{{workflow.status}}"понятие.

Однако сейчас я пытаюсь использовать Vertex AI для запуска конвейера, используя Kubeflow Pipelines SDK v2, kfp.v2. Используя то же изображение обработчика выхода, что и раньше, 'eu.gcr.io/.../send-error-msg-to-slack', теперь я определяю файл компонента yaml ( exit_handler.yaml) вместо,

      name: Exit handler
description: Prints to Slack if any step of the pipeline fails

inputs:
  - {name: message, type: String}
  - {name: status, type: String}

implementation:
  container:
    image: eu.gcr.io/.../send-error-msg-to-slack
    command: [
      python3,
      main.py,
      --message, {inputValue: message},
      --status, {inputValue: status}
    ]

Код конвейера теперь выглядит так:

      from google.cloud import aiplatform
from google.cloud.aiplatform import pipeline_jobs
from kfp.v2 import compiler
from kfp.v2.dsl import pipeline, ExitHandler
from kfp.components import load_component_from_file

@pipeline(name="Basic-pipeline",
          pipeline_root='gs://.../basic-pipeline')
def pipeline():
    exit_handler_spec = load_component_from_file('./exit_handler.yaml')
    exit_handler = exit_handler_spec(
        message="Basic pipeline failed.",
        status="{{workflow.status}}"
    )
    with ExitHandler(exit_handler):
        step_0_spec = load_component_from_file('./comp_0.yaml')
        step0 = step_0_spec(...)

        step_1_spec = load_component_from_file('./comp_1.yaml')
        step1 = step_1_spec(...) \
            .after(step0)

if __name__ == '__main__':
    compiler.Compiler().compile(
        pipeline_func=pipeline,
        package_path="basic-pipeline.json"
    )
    from google.oauth2 import service_account
    credentials = service_account.Credentials.from_service_account_file("./my-key.json")
    aiplatform.init(project='bsg-personalization',
                    location='europe-west4',
                    credentials=credentials)

    job = pipeline_jobs.PipelineJob(
        display_name="basic-pipeline",
        template_path="basic-pipeline.json",
        parameter_values={...}
    )
    job.run()

Это «работает» (без исключений) для компиляции и запуска, но код ExitHandler интерпретирует statusв виде строки со значением {{workflow.status}}, на что также указывает скомпилированный конвейер json, сгенерированный из приведенного выше кода ( basic-pipeline.json), который вы можете увидеть ниже ( "stringValue": "{{workflow.status}}"):

      ...
         "exit-handler": {
            "componentRef": {
              "name": "comp-exit-handler"
            },
            "dependentTasks": [
              "exit-handler-1"
            ],
            "inputs": {
              "parameters": {
                "message": {
                  "runtimeValue": {
                    "constantValue": {
                      "stringValue": "Basic pipeline failed."
                    }
                  }
                },
                "status": {
                  "runtimeValue": {
                    "constantValue": {
                      "stringValue": "{{workflow.status}}"
                    }
                  }
                }
              }
            },
            "taskInfo": {
              "name": "exit-handler"
            },
            "triggerPolicy": {
              "strategy": "ALL_UPSTREAM_TASKS_COMPLETED"
            }
          }
...

Любая идея о том, как я могу реорганизовать свой старый ExitHandlerкод, использующий v1, в новый SDK v2, чтобы обработчик выхода понял , не выполнен ли статус моего конвейера или нет?

Answer 1

Замена"{{workflow.status}}"в KFP SDK v2 это аннотация специального типаPipelineTaskFinalStatusкак упомянутый выше IronPan.

Его использование описано в https://www.kubeflow.org/docs/components/pipelines/v2/author-a-pipeline/pipelines/#dslexithandler .

Answer 2

Вероятно, это еще не полностью задокументировано, но в V2 мы ввели другую переменнуюPipelineTaskFinalStatusкоторый может быть автоматически заполнен, чтобы вы могли отправить его на свой канал Slack.

Вот пример обработчика выхода в официальном документе https://cloud.google.com/vertex-ai/docs/pipelines/email-notifications#sending_a_notification_from_a_pipeline.

А вот соответствующий компонент уведомления по электронной почте https://github.com/kubeflow/pipelines/blob/master/components/google-cloud/google_cloud_pipeline_components/v1/vertex_notification_email/component.yaml

Вы можете написать свой собственный компонент со следующим параметром, который будет автоматически заполняться при запуске обработчика выхода.

      inputs:
...
  - name: pipeline_task_final_status
    type: PipelineTaskFinalStatus

(Обратите внимание, что эта функция в настоящее время недоступна в дистрибутиве с открытым исходным кодом Kubeflow Pipelines и будет доступна в KFP V2. Она доступна только в дистрибутиве Vertex Pipelines)