chore(v2): standardize MLMD data model. Fixes #5669 #6054
Conversation
|
/retest |
sdk/python/kfp/compiler/v2_compat.py
Outdated
| "--run_id", | ||
| "$(KFP_RUN_ID)", | ||
| "--run_resource", | ||
| "workflows.argoproj.io/$(WORKFLOW_NAME)", |
There was a problem hiding this comment.
Do we need this workflows.argoproj.io/ here?
There was a problem hiding this comment.
It's meant to make the field vendor-neutral, it can also be a tekton pipeline with KFP-tekton
| @@ -275,16 +281,21 @@ func (l *Launcher) executeWithoutCacheEnabled(ctx context.Context, executorInput | |||
| cmd := l.cmdArgs[0] | |||
| args := make([]string, len(l.cmdArgs)-1) | |||
| _ = copy(args, l.cmdArgs[1:]) | |||
| pipeline, err := l.metadataClient.GetPipeline(ctx, l.options.PipelineName, l.options.PipelineRunID) | |||
| pipeline, err := l.metadataClient.GetPipeline(ctx, l.options.PipelineName, l.options.RunID, l.options.Namespace, l.options.RunResource) | |||
There was a problem hiding this comment.
Is it more clear if we pass l.options as the parameter?
There was a problem hiding this comment.
I intentionally did not do so, because metadata package is a low level package. It will be reused by both v2 compatible launcher and v2 driver/publisher.
It should not use a type defined in launcher package.
| @@ -310,16 +321,22 @@ func (l *Launcher) executeWithCacheEnabled(ctx context.Context, executorInput *p | |||
| return fmt.Errorf("failure while getting executionCache: %w", err) | |||
| } | |||
|
|
|||
| pipeline, err := l.metadataClient.GetPipeline(ctx, l.options.PipelineName, l.options.PipelineRunID) | |||
| pipeline, err := l.metadataClient.GetPipeline(ctx, l.options.PipelineName, l.options.RunID, l.options.Namespace, l.options.RunResource) | |||
There was a problem hiding this comment.
same here, is it more clear if we pass l.options as the parameter?
| "namespace": stringValue(namespace), | ||
| "resource_name": stringValue(runResource), | ||
| } | ||
| pipelineRunContext, err := getOrInsertContext(ctx, c.svc, pipelineRunID, pipelineRunContextType, runMetadata) |
There was a problem hiding this comment.
Since we change the value of "pipelineRunID" here, from workflow id to the uuid of the pipelineRun, will user still be able to get info of the previous tasks(which stores the workflow id in MLMD) via UI?
There was a problem hiding this comment.
Good catch! This is a breaking change. No, they won't.
This is under the assumption that v2 compatible doesn't have many users, we kept saying it's experimental and we can make breaking changes.
| @@ -222,6 +223,20 @@ def _create_run(): | |||
| fire.Fire(main) | |||
|
|
|||
|
|
|||
| def simplify_proto_struct(data: dict) -> dict: | |||
| res = {} | |||
| for key, value in data.items(): | |||
There was a problem hiding this comment.
Do we have intValue here?
There was a problem hiding this comment.
not sure, we can add one when we have a sample
v2/metadata/client_test.go
Outdated
| @@ -84,9 +84,9 @@ func Test_GetPipeline(t *testing.T) { | |||
| mlmdClient, err := NewTestMlmdClient() | |||
| fatalIf(err) | |||
|
|
|||
| _, err = client.GetPipeline(ctx, "get-pipeline-test", runId) | |||
| _, err = client.GetPipeline(ctx, "get-pipeline-test", runId, "kubeflow", "workflows.argoproj.io/hello-world-abcd") | |||
There was a problem hiding this comment.
Do we need to write "kubeflow" and "workflows.argoproj.io" as constants?
There was a problem hiding this comment.
My gut feeling is we don't need to, because these values can be arbitrary in tests.
There was a problem hiding this comment.
rethinking over this, I think you are right, if I set up default values, they help reduce information noise in each test.
|
/approve |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: Bobgy The full list of commands accepted by this bot can be found here. The pull request process is described here
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
Description of your changes:
Fixes #5669
Fixes #5985
Fixes #5986
Checklist: