[flyteadmin] Show diff structure when re-registration two different task with same ids

[austin362667]

Tracking issue

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

Why are the changes needed?

To highlight diff(delta part) error message for NewTaskExistsDifferentStructureError()

What changes were proposed in this pull request?

1. Modify task.proto to contain error reasoning info.

// The task id is already used and the structure is different
message TaskErrorExistsDifferentStructure {
    core.Identifier id = 1;
    core.TaskTemplate old_spec = 2;
    core.TaskTemplate new_spec = 3;
}

// The task id is already used with an identical sctructure
message TaskErrorExistsIdenticalStructure {
    core.Identifier id = 1;
}

// When a CreateTaskRequest fails due to matching id
message CreateTaskFailureReason {
    oneof reason {
        TaskErrorExistsDifferentStructure exists_different_structure = 1;
        TaskErrorExistsIdenticalStructure exists_identical_structure = 2;
    }
}

Actually it may not be necessary here, if we only return the diff structure as error msg string.

2. Refactor NewFlyteAdminErrorf() in task_manager.go to errors.go

• NewTaskExistsDifferentStructureError()
• NewTaskExistsIdenticalStructureError()

3. Add GetTask() to get existed/old task template with same id.

4. Add "wI2L/jsondiff" go package for computing structure diff.

How was this patch tested?

Setup process

1. register two different task code file, but with the same project, domain, name, version.

Screenshots

• [ ] I updated the documentation accordingly.
• [x] All new and existing tests passed.
• [x] All commits are signed-off.

Related PRs

Docs link

[welcome[bot]]

Thank you for opening this pull request! 🙌

These tips will help get your PR across the finish line:

• Most of the repos have a PR template; if not, fill it out to the best of your knowledge.
• Sign off your commits (Reference: DCO Guide).

[austin362667]

• successful register task
• re-register same task with same id (pass)
• re-register different(modified) task with same id (failed)
• re-register different(modified) task with same id (failed), with diff error message

[austin362667]

What about Workflow, LaunchPlan?

e.g., Modify a task name(function name) used by another workflow

[austin362667]

2. extend the error message to include structured data about what in the spec has changed

What if we just show diff as string, in order to prevent modify too much task proto buff?

[Future-Outlier]

I think overall looks well, however I am not familiar with flyteadmin's codebase. cc @katrogan, can you help review? thank you.

[codecov[bot]]

Codecov Report

Attention: Patch coverage is 69.56522% with 21 lines in your changes are missing coverage. Please review.

Project coverage is 59.03%. Comparing base (ec0bc4c) to head (04c34ba).

Files	Patch %	Lines
flyteadmin/pkg/manager/impl/task_manager.go	11.11%	8 Missing :warning:
flyteadmin/pkg/manager/impl/launch_plan_manager.go	0.00%	7 Missing :warning:
flyteadmin/pkg/manager/impl/workflow_manager.go	33.33%	3 Missing and 1 partial :warning:
flyteadmin/pkg/errors/errors.go	95.74%	2 Missing :warning:

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #4924      +/-   ##
==========================================
+ Coverage   59.00%   59.03%   +0.02%     
==========================================
  Files         645      645              
  Lines       55672    55726      +54     
==========================================
+ Hits        32850    32896      +46     
- Misses      20226    20233       +7     
- Partials     2596     2597       +1     

Flag	Coverage Δ
unittests	59.03% <69.56%> (+0.02%)	:arrow_up:

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

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.

[austin362667]

from flytekit import task, workflow
import time

@task(container_image="image1")
def say_hello() -> str:
    time.sleep(1)
    return "Hello, World!"

@workflow
def wf() -> str:
    res = say_hello()
    return res

if __name__ == "__main__":
    print(f"Running hello_world_wf() {wf()}")

from flytekit import task, workflow
import time

@task(container_image="image2")
def say_hello() -> str:
    time.sleep(1)
    return "Hello, World!"

@workflow
def wf() -> str:
    res = say_hello()
    return res

if __name__ == "__main__":
    print(f"Running hello_world_wf() {wf()}")

pyflyte register workflow.py --image basics:v2 --version v1

[austin362667]

1. Task: macos to ubuntu

from flytekit import task, workflow
import time
from flytekit.image_spec import ImageSpec

image_spec = "macos"

@task(container_image=image_spec)
def say_hi() -> str:
    time.sleep(1)
    return "Hello, World!"


@workflow
def wf() -> str:
    res = say_hi()
    return res

if __name__ == "__main__":
    print(f"Running hello_world_wf() {wf()}")

from flytekit import task, workflow
import time
from flytekit.image_spec import ImageSpec

image_spec = "ubuntu"

@task(container_image=image_spec)
def say_hi() -> str:
    time.sleep(1)
    return "Hello, World!"
    

@workflow
def wf() -> str:
    res = say_hi()
    return res

if __name__ == "__main__":
    print(f"Running hello_world_wf() {wf()}")

2. Workflow: say_hi() to say_hey()

from flytekit import task, workflow
import time
from flytekit.image_spec import ImageSpec

image_spec = "ubuntu"

@task(container_image=image_spec)
def say_hi() -> str:
    time.sleep(1)
    return "Hello, World!"


@workflow
def wf() -> str:
    res = say_hi()
    return res

if __name__ == "__main__":
    print(f"Running hello_world_wf() {wf()}")

from flytekit import task, workflow
import time
from flytekit.image_spec import ImageSpec

image_spec = "ubuntu"

@task(container_image=image_spec)
def say_hey() -> str:
    time.sleep(1)
    return "Hello, World!"
    

@workflow
def wf() -> str:
    res = say_hey()
    return res

if __name__ == "__main__":
    print(f"Running hello_world_wf() {wf()}")

[welcome[bot]]

Congrats on merging your first pull request! 🎉