changed argo template

kubeflow · Apr 22, 2022 · 7f01604 · 7f01604
1 parent 4ebab0a
commit 7f01604
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 1 deletion.
diff --git a/backend/src/apiserver/template/v2_template.go b/backend/src/apiserver/template/v2_template.go
@@ -7,6 +7,7 @@ import (
 	structpb "github.com/golang/protobuf/ptypes/struct"
 
 	"github.com/ghodss/yaml"
+	"github.com/golang/glog"
 	"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
 	api "github.com/kubeflow/pipelines/backend/api/go_client"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
@@ -36,6 +37,7 @@ func (t *V2Spec) ScheduledWorkflow(apiJob *api.Job) (*scheduledworkflow.Schedule
 	}
 	job.RuntimeConfig = jobRuntimeConfig
 	wf, err := argocompiler.Compile(job, nil)
+	glog.Infof("wf value: %+v", wf)
 	if err != nil {
 		return nil, util.Wrap(err, "Failed to compile job")
 	}
@@ -147,6 +149,7 @@ func (t *V2Spec) RunWorkflow(apiRun *api.Run, options RunWorkflowOptions) (*util
 	}
 	job.RuntimeConfig = jobRuntimeConfig
 	wf, err := argocompiler.Compile(job, nil)
+	glog.Infof("wf value: %+v", wf)
 	if err != nil {
 		return nil, util.Wrap(err, "Failed to compile job")
 	}

diff --git a/backend/src/v2/cmd/compiler/main.go b/backend/src/v2/cmd/compiler/main.go
@@ -42,6 +42,9 @@ func main() {
 	flag.Parse()
 	noSpec := specPath == nil || *specPath == ""
 	noJob := jobPath == nil || *jobPath == ""
+	glog.Info("Hiiiiiiiiiiiiiiii!!!!!!!!!!!!!!!!!!!!!!!!!!!")
+	noSpec = true
+	noJob = true
 	if noSpec && noJob {
 		glog.Exitf("spec or job must be specified")
 	}
@@ -70,6 +73,7 @@ func compile(job *pipelinespec.PipelineJob) error {
 		LauncherImage: *launcher,
 		PipelineRoot:  *pipelineRoot,
 	})
+	glog.Infof("wf value: %+v", wf)
 	if err != nil {
 		return err
 	}

diff --git a/backend/src/v2/compiler/argocompiler/argo.go b/backend/src/v2/compiler/argocompiler/argo.go
@@ -19,6 +19,7 @@ import (
 	"strings"
 
 	wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
+	"github.com/golang/glog"
 	"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
 	"github.com/kubeflow/pipelines/backend/src/v2/compiler"
 	"google.golang.org/protobuf/proto"
@@ -73,6 +74,16 @@ func Compile(jobArg *pipelinespec.PipelineJob, opts *Options) (*wfapi.Workflow,
 			}
 		}
 	}
+	// fill in exit handler
+	var exit_task string
+	all_root_tasks := spec.GetRoot().GetDag().GetTasks()
+	for task_name, task_spec := range all_root_tasks {
+		glog.Infof("task name, task spec: %s, %+v", task_name, task_spec)
+		if task_spec.GetTriggerPolicy().GetStrategy().String() == "ALL_UPSTREAM_TASKS_COMPLETED" {
+			exit_task = task_name
+			glog.Infof("exit task: %s", exit_task)
+		}
+	}
 	// initialization
 	wf := &wfapi.Workflow{
 		TypeMeta: k8smeta.TypeMeta{
@@ -101,13 +112,15 @@ func Compile(jobArg *pipelinespec.PipelineJob, opts *Options) (*wfapi.Workflow,
 			},
 			ServiceAccountName: "pipeline-runner",
 			Entrypoint:         tmplEntrypoint,
+			OnExit:             exit_task,
 		},
 	}
 	c := &workflowCompiler{
 		wf:        wf,
 		templates: make(map[string]*wfapi.Template),
 		// TODO(chensun): release process and update the images.
-		driverImage:   "gcr.io/ml-pipeline-test/dev/kfp-driver:latest",
+		// driverImage:   "gcr.io/ml-pipeline-test/dev/kfp-driver:latest",
+		driverImage:   "gcr.io/ling-kfp/dev/kfp-driver:latest",
 		launcherImage: "gcr.io/ml-pipeline-test/dev/kfp-launcher-v2:latest",
 		job:           job,
 		spec:          spec,

diff --git a/backend/src/v2/driver/driver.go b/backend/src/v2/driver/driver.go
@@ -122,6 +122,7 @@ func RootDAG(ctx context.Context, opts Options, mlmd *metadata.Client) (executio
 	}
 	// TODO(Bobgy): fill in run resource.
 	pipeline, err := mlmd.GetPipeline(ctx, opts.PipelineName, opts.RunID, opts.Namespace, "run-resource", pipelineRoot)
+	glog.Infof("metadata pipeline: %+v", pipeline)
 	if err != nil {
 		return nil, err
 	}