[BUG] LaunchPlan created as part of a workflow that is called from a reference workflow uses incorrect project and domain

[ggydush]

Describe the bug

When calling a workflow that contains a LaunchPlan with a reference workflow in another project, the project/domain of the LaunchPlan matches the reference workflow rather than the project/domain of the originally registered workflow.

Expected behavior

The LaunchPlan being called by the reference workflow would have its project and domain be consistent with the originally registered

Additional context to reproduce

Let's say I have 2 projects (Project A and Project B). In Project A I have a workflow that uses a LaunchPlan, which executes fine if being called from Project A. However, if I try to call this workflow via a ReferenceLaunchPlan in Project B, Flyte attempts to find the LaunchPlan in Project B instead of Project A.

Project A:

└── projecta
    └── workflows
        └── test_workflow.py (shown below)

Where test__workflow.py

from flytekit import LaunchPlan, dynamic, task, workflow


@task
def dummy_task(a: int) -> int:
    return a


@workflow
def wrapper_workflow(a: int) -> int:
    return dummy_task(a=a)


@dynamic
def dynamic_lp(a: int) -> int:
    # In our real-world example, we are using LaunchPlan here rather than subworkflow to handle large fan out
    lp = LaunchPlan.get_or_create(wrapper_workflow)
    return lp(a=a)


@workflow
def workflow_test(a: int) -> int:
    return dynamic_lp(a=a)

Project B

from flytekit import workflow
from flytekit.core.launch_plan import reference_launch_plan


def interface(a: int) -> int:
    ...


rlp = reference_launch_plan(
    project="projecta",
    domain="development",
    name="projecta.workflows.test_workflow.workflow_test",
    version="1",
)(interface)


@workflow
def workflow_calling_reference(a: int) -> int:
    return rlp(a=a)

Screenshots

No response

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes

[welcome]

Thank you for opening your first issue here! 🛠

[wild-endeavor]

The registration flow being used here is pyflyte serialize and then flyte-cli register-files. If you run pyflyte register this does not happen.

Some background here... the registration step is split into two stages since

1. It is actually two steps. There is the compilation step, which takes user code and converts it into Flyte protobuf files, and the control plane step, which takes those files and registers them with a Flyte backend.
2. Separating out the steps theoretically produces "portable" assets. That is, the protobuf files generated from the first step should be registrable to any project, on any Flyte backend. Someone from company A can register entities generated by company Z, if Z were to send those protobuf files over to A.

Where it gets a bit tricky is with dynamic tasks. A dynamic task is a task that compiles down to a workflow, at run time, rather than at compile time, which is how it's able to do things like range() over an input. The spec that it produces has a list of tasks and subworkflows. It gets these because typically tasks called by a dynamic task are defined in the same python project. So the code will just import them. These tasks are recompiled down to new TaskTemplates and from there sent over to Propeller, which executes them.

This means it doesn't matter typically what the names/project/domain of these tasks are. Since the definition for these tasks is in-lined, Propeller has all the information it needs - it doesn't need to call Admin to fetch their definitions again so it doesn't matter what the name in the Task id is.

Note the dynamic job spec does not include launchplans, so for launch plans, it does matter what the id is. Any Launch Plans referenced in the dynamic job spec will be pulled by Propeller. Because of this, the ID actually matters.

[side note - this bundling of subworkflows and tasks into the spec is also why reference_task and reference_workflow do not work... there's no definition there to compile. To do these, you have to use fetched tasks & workflows.]

When a dynamic task runs, it's compiled and then converted into protobuf objects, much like the pyflyte serialize step. However at this step, it uses the ctx.serialization_settings. This is where the id gets pulled from.

For this reason (and some others) when you pyflyte register a dynamic task, it will serialize the context in with the task itself, and store it as the environment variable _F_SS_C.

But in the multi-step pyflyte serialize -> flyte-cli register path, this doesn't work. Btw, we've deprecated flyte-cli in favor of flytectl, which is another control plane CLI tool for Flyte meant to mirror the kubectl experience better.

[wild-endeavor]

additional use-case for fixing this: https://discuss.flyte.org/t/8421373/hi-there-i-am-wondering-if-there-is-a-way-to-dynamically-set (https://flyte-org.slack.com/archives/CP2HDHKE1/p1673646315622989)

[wild-endeavor]

For the record, this was an incomplete patch that we did (flyteorg/flytekit#1378) that was later reverted (flyteorg/flytekit#1460) because it caused another issue. We'll put in another patch that's hopefully a bit more correct today. And hopefully we can have time soon to implement the correct fix.