Introduces PipelineConfig class with TTL configuration

Signed-off-by: Ricardo M. Oliveira <[email protected]>

api/v2alpha1/pipeline_spec.proto

@@ -1079,6 +1079,8 @@ message PipelineStateEnum {
 message PlatformSpec {
   // Platform key to full platform config
   map<string, SinglePlatformSpec> platforms = 1;
+
+
 }
 
 message SinglePlatformSpec {
@@ -1095,6 +1097,11 @@ message SinglePlatformSpec {
   PipelineConfig pipelineConfig = 4;
 }
 
+message PipelineConfig {
+  // Set TTL at pipeline-level
+  optional int32 pipelineTtl = 1;
+}
+
 message PlatformDeploymentConfig {
   // Map of executor label to executor-level config
   // Mirrors PipelineSpec.deployment_spec.executors structure

backend/src/apiserver/template/v2_template.go

@@ -23,6 +23,7 @@ import (
 	"regexp"
 	"strings"
 
+	"github.com/golang/glog"
 	structpb "github.com/golang/protobuf/ptypes/struct"
 	"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/model"
@@ -77,9 +78,17 @@ func (t *V2Spec) ScheduledWorkflow(modelJob *model.Job) (*scheduledworkflow.Sche
 		}
 	}
 
+	var pipeline_options argocompiler.Options
+	if t.platformSpec.PipelineConfig.Ttl != nil {
+		glog.Info("Found pipeline config")
+		pipeline_options = argocompiler.Options{
+			TtlSeconds: *t.platformSpec.PipelineConfig.Ttl,
+		}
+	}
+
 	var obj interface{}
 	if util.CurrentExecutionType() == util.ArgoWorkflow {
-		obj, err = argocompiler.Compile(job, kubernetesSpec, nil)
+		obj, err = argocompiler.Compile(job, kubernetesSpec, &pipeline_options)
 	} else if util.CurrentExecutionType() == util.TektonPipelineRun {
 		obj, err = tektoncompiler.Compile(job, kubernetesSpec, &tektoncompiler.Options{LauncherImage: Launcher})
 	}
@@ -300,9 +309,17 @@ func (t *V2Spec) RunWorkflow(modelRun *model.Run, options RunWorkflowOptions) (u
 		}
 	}
 
+	var pipeline_options *argocompiler.Options
+	if t.platformSpec.PipelineConfig.Ttl != nil {
+		glog.Info("Found pipeline config")
+		pipeline_options = &argocompiler.Options{
+			TtlSeconds: *t.platformSpec.PipelineConfig.Ttl,
+		}
+	}
+
 	var obj interface{}
 	if util.CurrentExecutionType() == util.ArgoWorkflow {
-		obj, err = argocompiler.Compile(job, kubernetesSpec, nil)
+		obj, err = argocompiler.Compile(job, kubernetesSpec, pipeline_options)
 	} else if util.CurrentExecutionType() == util.TektonPipelineRun {
 		obj, err = tektoncompiler.Compile(job, kubernetesSpec, nil)
 	}
@@ -344,7 +361,7 @@ func IsPlatformSpecWithKubernetesConfig(template []byte) bool {
 		return false
 	}
 	_, ok := platformSpec.Platforms["kubernetes"]
-	return ok
+	return ok || platformSpec.PipelineConfig != nil
 }
 
 func (t *V2Spec) validatePipelineJobInputs(job *pipelinespec.PipelineJob) error {

backend/src/v2/compiler/argocompiler/argo.go

@@ -40,8 +40,13 @@ type Options struct {
 	// optional
 	PipelineRoot string
 	// TODO(Bobgy): add an option -- dev mode, ImagePullPolicy should only be Always in dev mode.
+	TtlSeconds int32
 }
 
+const (
+	pipeline_default_ttlSeconds = int32(30)
+)
+
 func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.SinglePlatformSpec, opts *Options) (*wfapi.Workflow, error) {
 	// clone jobArg, because we don't want to change it
 	jobMsg := proto.Clone(jobArg)
@@ -86,6 +91,11 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
 		}
 	}
 
+	pipeline_ttlseconds := pipeline_default_ttlSeconds
+	if &opts.TtlSeconds != nil {
+		pipeline_ttlseconds = opts.TtlSeconds
+	}
+
 	// initialization
 	wf := &wfapi.Workflow{
 		TypeMeta: k8smeta.TypeMeta{
@@ -117,6 +127,9 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
 			},
 			ServiceAccountName: "pipeline-runner",
 			Entrypoint:         tmplEntrypoint,
+			TTLStrategy: &wfapi.TTLStrategy{
+				SecondsAfterCompletion: &pipeline_ttlseconds,
+			},
 		},
 	}
 	c := &workflowCompiler{

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -1953,7 +1953,10 @@ def create_pipeline_spec(
     if pipeline_config is not None:
         _merge_pipeline_config(
             pipelineConfig=pipeline_config, platformSpec=platform_spec)
+<<<<<<< HEAD
 
+=======
+>>>>>>> cc9690e2b (Introduces PipelineConfig class with TTL configuration)
     for group in all_groups:
         build_spec_by_group(
             pipeline_spec=pipeline_spec,
@@ -2076,6 +2079,11 @@ def _merge_pipeline_config(pipelineConfig: pipeline_config.PipelineConfig,
     #     {'pipelineConfig': {
     #         '<some pipeline config option>': pipelineConfig.<get that value>,
     #     }}, platformSpec.platforms['kubernetes'])
+
+    pipeline_config_json = json_format.ParseDict(
+        {'pipelineConfig': {
+            'pipelineTtl': pipelineConfig.get_ttl(),
+        }}, platformSpec.platforms['kubernetes'])
 
     return platformSpec
 

sdk/python/kfp/dsl/component_decorator.py

@@ -17,6 +17,7 @@
 import warnings
 
 from kfp.dsl import component_factory
+from kfp.dsl import pipeline_config
 
 
 def component(func: Optional[Callable] = None,

sdk/python/kfp/dsl/pipeline_config.py

@@ -21,3 +21,8 @@ def __init__(self):
         pass
 
     # TODO add pipeline level configs
+    def set_ttl(self, ttl: int):
+        self.__ttl = ttl
+
+    def get_ttl(self) -> int:
+        return int(self.__ttl)

sdk/python/kfp/dsl/pipeline_context.py

@@ -53,7 +53,11 @@ def my_pipeline(a: str, b: int):
         pipeline_root: The root directory from which to read input and output
             parameters and artifacts.
         display_name: A human-readable name for the pipeline.
+<<<<<<< HEAD
         pipeline_config: Pipeline-level config options.
+=======
+        pipeline_config: Pipeline-level configuration.
+>>>>>>> cc9690e2b (Introduces PipelineConfig class with TTL configuration)
     """
     if func is None:
         return functools.partial(