Enable TTL configuration

api/v2alpha1/pipeline_spec.proto

@@ -1079,6 +1079,8 @@ message PipelineStateEnum {
 message PlatformSpec {
   // Platform key to full platform config
   map<string, SinglePlatformSpec> platforms = 1;
+
+  optional int32 ttlStrategysecondsAfterCompletion = 2;
 }
 
 message SinglePlatformSpec {

sdk/python/kfp/cli/cli.py

@@ -92,7 +92,7 @@ def _install_completion(shell: str) -> None:
     show_default=True,
     help='The formatting style for command output.')
 @click.pass_context
-@click.version_option(version=kfp.__version__, message='%(prog)s %(version)s')
+@click.version_option(version='dev', message='%(prog)s %(version)s')
 def cli(ctx: click.Context, endpoint: str, iap_client_id: str, namespace: str,
         other_client_id: str, other_client_secret: str, output: OutputFormat,
         show_completion: str, install_completion: str):

sdk/python/kfp/compiler/pipeline_spec_builder.py

@@ -1879,6 +1879,7 @@ def create_pipeline_spec(
     pipeline: pipeline_context.Pipeline,
     component_spec: structures.ComponentSpec,
     pipeline_outputs: Optional[Any] = None,
+    ttl_strategy_seconds_after_completion: Optional[int] = None,
 ) -> Tuple[pipeline_spec_pb2.PipelineSpec, pipeline_spec_pb2.PlatformSpec]:
     """Creates a pipeline spec object.
 
@@ -1899,7 +1900,8 @@ def create_pipeline_spec(
     pipeline_spec = pipeline_spec_pb2.PipelineSpec()
 
     pipeline_spec.pipeline_info.name = pipeline.name
-    pipeline_spec.sdk_version = f'kfp-{kfp.__version__}'
+    # pipeline_spec.sdk_version = f'kfp-{kfp.__version__}'
+    pipeline_spec.sdk_version = f'kfp-dev'
     # Schema version 2.1.0 is required for kfp-pipeline-spec>0.1.13
     pipeline_spec.schema_version = '2.1.0'
 
@@ -1945,8 +1947,9 @@ def create_pipeline_spec(
         group_name_to_group=group_name_to_group,
         condition_channels=condition_channels,
     )
-
+    import pdb; pdb.set_trace()
     platform_spec = pipeline_spec_pb2.PlatformSpec()
+    platform_spec.ttlStrategysecondsAfterCompletion = ttl_strategy_seconds_after_completion
     for group in all_groups:
         build_spec_by_group(
             pipeline_spec=pipeline_spec,

sdk/python/kfp/dsl/component_decorator.py

@@ -28,7 +28,7 @@ def component(func: Optional[Callable] = None,
               output_component_file: Optional[str] = None,
               install_kfp_package: bool = True,
               kfp_package_path: Optional[str] = None,
-              pip_trusted_hosts: Optional[List[str]] = None):
+              ttl_active_seconds: Optional[int] = None):
     """Decorator for Python-function based components.
 
     A KFP component can either be a lightweight component or a containerized
@@ -116,7 +116,7 @@ def pipeline():
             output_component_file=output_component_file,
             install_kfp_package=install_kfp_package,
             kfp_package_path=kfp_package_path,
-            pip_trusted_hosts=pip_trusted_hosts)
+            ttl_active_seconds=ttl_active_seconds)
 
     return component_factory.create_component_from_func(
         func,
@@ -127,4 +127,4 @@ def pipeline():
         output_component_file=output_component_file,
         install_kfp_package=install_kfp_package,
         kfp_package_path=kfp_package_path,
-        pip_trusted_hosts=pip_trusted_hosts)
+        ttl_active_seconds=ttl_active_seconds)

sdk/python/kfp/dsl/component_factory.py

@@ -162,7 +162,8 @@ def _get_packages_to_install_command(
         else:
             kfp_pip_install_command = make_pip_install_command(
                 install_parts=[
-                    f'kfp=={kfp.__version__}',
+                    # f'kfp=={kfp.__version__}',
+                    f'kfp==dev',
                     '--no-deps',
                     'typing-extensions>=3.7.4,<5; python_version<"3.9"',
                 ],
@@ -533,6 +534,7 @@ def create_component_from_func(
     install_kfp_package: bool = True,
     kfp_package_path: Optional[str] = None,
     pip_trusted_hosts: Optional[List[str]] = None,
+    ttl_active_seconds: Optional[int] = None,
 ) -> python_component.PythonComponent:
     """Implementation for the @component decorator.
 
@@ -676,6 +678,7 @@ def create_graph_component_from_func(
     name: Optional[str] = None,
     description: Optional[str] = None,
     display_name: Optional[str] = None,
+    ttl_strategy_seconds_after_completion: Optional[int] = None,
 ) -> graph_component.GraphComponent:
     """Implementation for the @pipeline decorator.
 
@@ -692,6 +695,7 @@ def create_graph_component_from_func(
         component_spec=component_spec,
         pipeline_func=func,
         display_name=display_name,
+        ttl_strategy_seconds_after_completion=ttl_strategy_seconds_after_completion,
     )
 
 

sdk/python/kfp/dsl/graph_component.py

@@ -37,6 +37,7 @@ def __init__(
         component_spec: structures.ComponentSpec,
         pipeline_func: Callable,
         display_name: Optional[str] = None,
+        ttl_strategy_seconds_after_completion: Optional[int] = None,
     ):
         super().__init__(component_spec=component_spec)
         self.pipeline_func = pipeline_func
@@ -69,6 +70,7 @@ def __init__(
             pipeline=dsl_pipeline,
             component_spec=self.component_spec,
             pipeline_outputs=pipeline_outputs,
+            ttl_strategy_seconds_after_completion=ttl_strategy_seconds_after_completion,
         )
 
         pipeline_root = getattr(pipeline_func, 'pipeline_root', None)

sdk/python/kfp/dsl/pipeline_context.py

@@ -28,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,
+             ttl_active_seconds: Optional[int] = None) -> Callable:
     """Decorator used to construct a pipeline.
 
     Example
@@ -50,6 +51,7 @@ 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.
+        ttl_active_seconds: Delete completed workflows after a set time.
     """
     if func is None:
         return functools.partial(
@@ -58,6 +60,7 @@ def my_pipeline(a: str, b: int):
             description=description,
             pipeline_root=pipeline_root,
             display_name=display_name,
+            ttl_active_seconds=ttl_active_seconds,
         )
 
     if pipeline_root:
@@ -68,6 +71,7 @@ def my_pipeline(a: str, b: int):
         name=name,
         description=description,
         display_name=display_name,
+        ttl_strategy_seconds_after_completion=ttl_active_seconds,
     )