Create default launch plan when executing WorkflowBase

[pingsutw]

Signed-off-by: Kevin Su [email protected]

TL;DR

• Register default launch plan when executing WorkflowBase
• Register task and workflow if not found
• Throw error immediately when entity not found

Running a workflow

@task
def base_greet(name: str) -> (str, str):
    return name, name


@task
def base_add_question(greeting: str) -> str:
    return f"{greeting} How are you?"


@workflow
def base_wf(name: str) -> str:
    n1, n2 = base_greet(name=name)
    return base_add_question(greeting=n1)

Before:

exe = remote.register(base_greet, version=version)
exe = remote.register(base_add_question, version=version)
exe = remote.register(base_wf)
ctx = context_manager.FlyteContext.current_context()
default_lp = LaunchPlan.get_default_launch_plan(ctx, base_wf)
exe = remote.execute(base_wf, inputs={"name": "kevin"}, wait=True)

After:

exe = remote.execute(base_wf, inputs={"name": "kevin"}, wait=True)

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

How did you fix the bug, make the feature etc. Link to any design docs etc

Tracking Issue

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

Follow-up issue

NA

[wild-endeavor]

add some logging in here to say that we're trying to register because it wasn't found?

[wild-endeavor]

also doesn't register currently just raise NotImplementedError

[wild-endeavor]

how do we know it's always a task? might it be a subworkflow? or a launch plan node?

[wild-endeavor]

And if it's a subworkflow, or a launch plan, then those underlying entities will also need to be registered. take a look at the translator.py file if you can. that file has logic that will recursively aggregate up everything. you're basically running a compilation step here.

[codecov]

Codecov Report

Merging #707 (8093ff6) into master (782bb34) will decrease coverage by 0.00%.
The diff coverage is 53.73%.

@@            Coverage Diff             @@
##           master     #707      +/-   ##
==========================================
- Coverage   85.76%   85.76%   -0.01%     
==========================================
  Files         358      359       +1     
  Lines       29778    29853      +75     
  Branches     2428     2434       +6     
==========================================
+ Hits        25538    25602      +64     
- Misses       3601     3611      +10     
- Partials      639      640       +1     

Impacted Files	Coverage Δ
flytekit/core/type_engine.py	88.47% <ø> (ø)
flytekit/clients/raw.py	75.00% <36.84%> (+0.31%)	⬆️
flytekit/remote/remote.py	69.88% <38.70%> (-2.65%)	⬇️
tests/flytekit/integration/remote/test_remote.py	92.27% <100.00%> (+0.64%)	⬆️
flytekit/models/common.py	89.09% <0.00%> (-7.31%)	⬇️
flytekit/models/admin/common.py	97.47% <0.00%> (-2.53%)	⬇️
flytekit/models/core/types.py	98.67% <0.00%> (-1.33%)	⬇️
flytekit/remote/component_nodes.py	80.00% <0.00%> (-0.33%)	⬇️
flytekit/remote/nodes.py	72.30% <0.00%> (-0.22%)	⬇️
... and 83 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 782bb34...8093ff6. Read the comment docs.

[pingsutw]

I got the below error in the integration test when I register the same version of the task twice.

The integration will run make register first to register all the workflow and task in mock_flyte_repo
, and then remote.execute(my_wf) will register the task again in the test_execute_python_workflow_list_of_floats
I found out that the error only happened when we register the task by the remote client after make register.

1. (task register) flytectl register -> remote.register -> StatusCode.INVALID_ARGUMENT
2. (task register) remote.register -> remote.register -> FlyteEntityAlreadyExistsException
3. (workflow register) flytectl register -> remote.register -> FlyteEntityAlreadyExistsException
4. (workflow register) remote.register -> remote.register -> FlyteEntityAlreadyExistsException

  File "/Users/kevin/git/flytekit/flytekit/clients/raw.py", line 173, in handler
    raise _user_exceptions.FlyteEntityAlreadyExistsException(e)
flytekit.common.exceptions.user.FlyteEntityAlreadyExistsException: <_InactiveRpcError of RPC that terminated with:
	status = StatusCode.INVALID_ARGUMENT
	details = "task with different structure already exists with id resource_type:TASK project:"flyteexamples" domain:"development" name:"dataclass.workflows.example.creat_file" version:"v11" "
	debug_error_string = "{"created":"@1634913163.363030000","description":"Error received from peer ipv6:[::1]:30081","file":"src/core/lib/surface/call.cc","file_line":1070,"grpc_message":"task with different structure already exists with id resource_type:TASK project:"flyteexamples" domain:"development" name:"dataclass.workflows.example.creat_file" version:"v11" ","grpc_status":3}"

To workaround to issue, catch an exception here.

[pingsutw]

To quickly reproduce this issue.

1. use flytectl to register a workflow
2. and use FlyteRemote to register the task again.

from flytekit.remote import FlyteRemote
from dataclass.workflows.example import create_file

remote = FlyteRemote.from_config("flyteexamples", "development")
remote.register(create_file, version=VERSION)

[wild-endeavor]

can you add a link to flyteorg/flyte#1359 as a comment here?

[pingsutw]

@wild-endeavor we will register parent workflow here