WIP: implement setting ttl on pipelines

Signed-off-by: Greg Sheremeta <gshereme@redhat.com>

Author: gregsheremeta

api/v2alpha1/go/cachekey/cache_key.pb.go

@@ -679,7 +679,8 @@ message ValueOrRuntimeParameter {
 }
 
 // The definition of the deployment config of the pipeline. It contains the
-// the platform specific executor configs for KFP OSS.
+// the platform specific executor configs for KFP OSS, as well as platform level config
+// such as settings to pass to an Argo Workflows Workflow CR.
 message PipelineDeploymentConfig {
   // The specification on a container invocation.
   // The string fields of the message support string based placeholder contract
@@ -852,6 +853,11 @@ message PipelineDeploymentConfig {
   }
   // Map from executor label to executor spec.
   map<string, ExecutorSpec> executors = 1;
+
+  // begin platform level / workflow level config
+
+  // duration in seconds after which the pipeline platform will do garbage collection
+  optional int32 completed_pipeline_ttl = 2;
 }
 
 // Value is the value of the field.

api/v2alpha1/go/pipelinespec/pipeline_spec.pb.go

@@ -72,6 +72,13 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
 		}
 	}
 
+	// pipeline-level config options
+	var completedPipelineTtl int32 = 0
+
+	if deploy.CompletedPipelineTtl != nil {
+		completedPipelineTtl = *deploy.CompletedPipelineTtl
+	}
+
 	var kubernetesSpec *pipelinespec.SinglePlatformSpec
 	if kubernetesSpecArg != nil {
 		// clone kubernetesSpecArg, because we don't want to change it
@@ -112,6 +119,16 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
 			Entrypoint:         tmplEntrypoint,
 		},
 	}
+
+	// set pipeline-level config options
+
+	// completed pipeline ttl
+	if completedPipelineTtl > 0 {
+		wf.Spec.TTLStrategy = &wfapi.TTLStrategy{
+			SecondsAfterCompletion: &completedPipelineTtl,
+		}
+	}
+
 	c := &workflowCompiler{
 		wf:        wf,
 		templates: make(map[string]*wfapi.Template),

api/v2alpha1/pipeline_spec.proto

@@ -28,6 +28,7 @@
 from kfp.dsl import component_factory
 from kfp.dsl import for_loop
 from kfp.dsl import pipeline_channel
+from kfp.dsl import pipeline_config
 from kfp.dsl import pipeline_context
 from kfp.dsl import pipeline_task
 from kfp.dsl import placeholders
@@ -1850,13 +1851,15 @@ def create_pipeline_spec(
     pipeline: pipeline_context.Pipeline,
     component_spec: structures.ComponentSpec,
     pipeline_outputs: Optional[Any] = None,
+    pipeline_config: pipeline_config.PipelineConfig = None,
 ) -> Tuple[pipeline_spec_pb2.PipelineSpec, pipeline_spec_pb2.PlatformSpec]:
     """Creates a pipeline spec object.
 
     Args:
         pipeline: The instantiated pipeline object.
         component_spec: The component spec structures.
         pipeline_outputs: The pipeline outputs via return.
+        pipeline_config: The pipeline config object.
 
     Returns:
         A PipelineSpec proto representing the compiled pipeline.
@@ -1874,6 +1877,10 @@ def create_pipeline_spec(
     # Schema version 2.1.0 is required for kfp-pipeline-spec>0.1.13
     pipeline_spec.schema_version = '2.1.0'
 
+    # pipeline-level config options
+    if pipeline_config.completed_pipeline_ttl_seconds is not None and pipeline_config.completed_pipeline_ttl_seconds > 0:
+        deployment_config.completed_pipeline_ttl = pipeline_config.completed_pipeline_ttl_seconds
+
     pipeline_spec.root.CopyFrom(
         _build_component_spec_from_component_spec_structure(component_spec))
 

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

@@ -266,6 +266,7 @@ def my_pipeline():
     from kfp.dsl.pipeline_channel import OneOf
     from kfp.dsl.pipeline_context import pipeline
     from kfp.dsl.pipeline_task import PipelineTask
+    from kfp.dsl.pipeline_config import PipelineConfig
     from kfp.dsl.placeholders import ConcatPlaceholder
     from kfp.dsl.placeholders import IfPresentPlaceholder
     from kfp.dsl.structures import ContainerSpec
@@ -292,4 +293,5 @@ def my_pipeline():
         'IfPresentPlaceholder',
         'ConcatPlaceholder',
         'PipelineTask',
+        'PipelineConfig',
     ])

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -27,6 +27,7 @@
 from kfp.dsl import container_component_artifact_channel
 from kfp.dsl import container_component_class
 from kfp.dsl import graph_component
+from kfp.dsl import pipeline_config
 from kfp.dsl import placeholders
 from kfp.dsl import python_component
 from kfp.dsl import structures
@@ -658,6 +659,7 @@ def create_graph_component_from_func(
     name: Optional[str] = None,
     description: Optional[str] = None,
     display_name: Optional[str] = None,
+    pipeline_config: pipeline_config.PipelineConfig = None,
 ) -> graph_component.GraphComponent:
     """Implementation for the @pipeline decorator.
 
@@ -674,6 +676,7 @@ def create_graph_component_from_func(
         component_spec=component_spec,
         pipeline_func=func,
         display_name=display_name,
+        pipeline_config=pipeline_config,
     )
 
 

sdk/python/kfp/dsl/__init__.py

@@ -20,6 +20,7 @@
 from kfp.compiler import pipeline_spec_builder as builder
 from kfp.dsl import base_component
 from kfp.dsl import pipeline_channel
+from kfp.dsl import pipeline_config
 from kfp.dsl import pipeline_context
 from kfp.dsl import structures
 from kfp.pipeline_spec import pipeline_spec_pb2
@@ -37,9 +38,11 @@ def __init__(
         component_spec: structures.ComponentSpec,
         pipeline_func: Callable,
         display_name: Optional[str] = None,
+        pipeline_config: pipeline_config.PipelineConfig = None,
     ):
         super().__init__(component_spec=component_spec)
         self.pipeline_func = pipeline_func
+        self.pipeline_config = pipeline_config
 
         args_list = []
         signature = inspect.signature(pipeline_func)
@@ -69,6 +72,7 @@ def __init__(
             pipeline=dsl_pipeline,
             component_spec=self.component_spec,
             pipeline_outputs=pipeline_outputs,
+            pipeline_config=pipeline_config,
         )
 
         pipeline_root = getattr(pipeline_func, 'pipeline_root', None)

sdk/python/kfp/dsl/component_factory.py

@@ -0,0 +1,34 @@
+# Copyright 2024 The Kubeflow Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Pipeline-level config options."""
+
+
+class PipelineConfig():
+    """PipelineConfig contains pipeline-level config options."""
+
+    def __init__(self):
+        self.completed_pipeline_ttl_seconds = None
+
+    def set_completed_pipeline_ttl_seconds(self, seconds: int):
+        """Configures the duration in seconds after which the pipeline platform will
+        do garbage collection. This is dependent on the platform's implementation, but
+        typically involves deleting pods and higher level resources associated with the
+        pods.
+
+        Args:
+          seconds: number of seconds for the platform to do garbage collection after
+            the pipeline run is completed.
+        """
+        self.completed_pipeline_ttl_seconds = seconds
+        return self

sdk/python/kfp/dsl/graph_component.py

@@ -17,6 +17,7 @@
 from typing import Callable, Optional
 
 from kfp.dsl import component_factory
+from kfp.dsl import pipeline_config
 from kfp.dsl import pipeline_task
 from kfp.dsl import tasks_group
 from kfp.dsl import utils
@@ -27,7 +28,8 @@ def pipeline(func: Optional[Callable] = None,
              name: Optional[str] = None,
              description: Optional[str] = None,
              pipeline_root: Optional[str] = None,
-             display_name: Optional[str] = None) -> Callable:
+             display_name: Optional[str] = None,
+             pipeline_config: pipeline_config.PipelineConfig = None) -> Callable:
     """Decorator used to construct a pipeline.
 
     Example
@@ -57,6 +59,7 @@ def my_pipeline(a: str, b: int):
             description=description,
             pipeline_root=pipeline_root,
             display_name=display_name,
+            pipeline_config=pipeline_config,
         )
 
     if pipeline_root:
@@ -67,6 +70,7 @@ def my_pipeline(a: str, b: int):
         name=name,
         description=description,
         display_name=display_name,
+        pipeline_config=pipeline_config,
     )