Hydrate failure node

[pingsutw]

TL;DR

Hydrate failure nodes during registration, so that we can successfully register a workflow with failure nodes.

Type

• Bug Fix
• Feature
• Plugin

Are all requirements met?

• Code completed
• Smoke tested
• Unit tests added
• Code documentation added
• Any pending items have an associated Issue

Complete description

command:

pyflyte --pkgs failure_node package -o union-airflow-package.tgz --force
flytectl register files union-airflow-package.tgz --archive -p union-airflow -d development --version v4 --logger.level 4

Workflow:

import typing
from click.testing import CliRunner

from flytekit import task, workflow, ImageSpec, WorkflowFailurePolicy, reference_task, reference_launch_plan
from flytekit.clis.sdk_in_container import pyflyte
from flytekit.types.error.error import FlyteError

new_flytekit = "git+https://github.com/flyteorg/flytekit.git@master"
image_spec = ImageSpec(packages=[new_flytekit], apt_packages=["git"], registry="pingsutw")


@task
def create_cluster(name: str):
    print(f"Creating cluster: {name}")


@task()
def t1(name: str):
    print(f"{name}")
    raise ValueError("Fail!")

@workflow()
def wff():
    t1(name="2")


@task()
def delete_cluster(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name}")
    print(err)


@task()
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name} due to {err}")
    print(err)


@workflow(on_failure=clean_up, failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def subwf(name: str = "kevin"):
    c = create_cluster(name=name)
    t = t1(name="2")
    d = delete_cluster(name=name)
    c >> t >> d


@workflow(on_failure=clean_up, failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def wf1(name: str = "kevin"):
    c = create_cluster(name=name)
    subwf(name="pingsutw")
    t = t1(name="2")
    d = delete_cluster(name=name)
    c >> t >> d


@workflow(on_failure=clean_up, failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def wf2(name: str = "kevin"):
    c = create_cluster(name=name)
    t = t1(name="2")
    d = delete_cluster(name=name)
    c >> t >> d


@workflow
def clean_up_wf(name: str = "kevin"):
    return create_cluster(name=name)


@workflow(on_failure=clean_up_wf, failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def wf3(name: str = "Kevin"):
    c = create_cluster(name=name)
    t = t1(name="2")
    d = delete_cluster(name=name)
    c >> t >> d

Tracking Issue

https://github.com/flyteorg/flyte/issues/

Follow-up issue

NA

[codecov]

Codecov Report

Attention: 6 lines in your changes are missing coverage. Please review.

Comparison is base (54ddddb) 67.75% compared to head (faa901c) 67.70%.

Files	Patch %	Lines
cmd/register/register_util.go	25.00%	4 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #456      +/-   ##
==========================================
- Coverage   67.75%   67.70%   -0.06%     
==========================================
  Files         148      148              
  Lines        6633     6641       +8     
==========================================
+ Hits         4494     4496       +2     
- Misses       1851     1855       +4     
- Partials      288      290       +2     

Flag	Coverage Δ
unittests	67.70% <33.33%> (-0.06%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.