Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

WIP: feat(api/sdk/backend): implement PipelineConfig and setting ttl on pipelines. Fixes: #10899 #10942

Closed
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions api/v2alpha1/go/cachekey/cache_key.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

820 changes: 423 additions & 397 deletions api/v2alpha1/go/pipelinespec/pipeline_spec.pb.go

Large diffs are not rendered by default.

8 changes: 7 additions & 1 deletion api/v2alpha1/pipeline_spec.proto
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

as per @droctothorpe comment from yesterday's meeting, what do we think about providing a comment here about logging? something a long the lines of:

"Note: when persisted logging is unavailable, any tasks that are garbage collected will result in the loss of the associated logs for those tasks in the UI."

adding this here will surface them in api docs, fyi

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

would suggest changing this type to duration. Duration explicitly represents time intervals, making the intent of the field more apparent

}

// Value is the value of the field.
Expand Down
17 changes: 17 additions & 0 deletions backend/src/v2/compiler/argocompiler/argo.go
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,13 @@
}
}

// pipeline-level config options
var completedPipelineTtl int32 = 0

if deploy.CompletedPipelineTtl != nil {

Check failure on line 78 in backend/src/v2/compiler/argocompiler/argo.go

View workflow job for this annotation

GitHub Actions / backend-tests

deploy.CompletedPipelineTtl undefined (type *pipelinespec.PipelineDeploymentConfig has no field or method CompletedPipelineTtl)

Check failure on line 78 in backend/src/v2/compiler/argocompiler/argo.go

View workflow job for this annotation

GitHub Actions / run-go-unittests

deploy.CompletedPipelineTtl undefined (type *pipelinespec.PipelineDeploymentConfig has no field or method CompletedPipelineTtl)
completedPipelineTtl = *deploy.CompletedPipelineTtl

Check failure on line 79 in backend/src/v2/compiler/argocompiler/argo.go

View workflow job for this annotation

GitHub Actions / backend-tests

deploy.CompletedPipelineTtl undefined (type *pipelinespec.PipelineDeploymentConfig has no field or method CompletedPipelineTtl)

Check failure on line 79 in backend/src/v2/compiler/argocompiler/argo.go

View workflow job for this annotation

GitHub Actions / run-go-unittests

deploy.CompletedPipelineTtl undefined (type *pipelinespec.PipelineDeploymentConfig has no field or method CompletedPipelineTtl)
}

var kubernetesSpec *pipelinespec.SinglePlatformSpec
if kubernetesSpecArg != nil {
// clone kubernetesSpecArg, because we don't want to change it
Expand Down Expand Up @@ -112,6 +119,16 @@
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),
Expand Down
7 changes: 7 additions & 0 deletions sdk/python/kfp/compiler/pipeline_spec_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

May we raise an error if <= 0?
How will the user know that the argument was ignored?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good point. I think the behavior should be:
-1 = 0 = infinite timeout = same as not specifying anything (per https://cloud.google.com/apis/design/design_patterns#integer_types)
< -1 = should return an invalid argument error

Copy link
Contributor

@DharmitD DharmitD Jul 15, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Some error handling message would be nice to have here. For instance,

if ttl_seconds < =0:
            raise ValueError("Invalid TTL value: {}. The value must be 1 or greater.".format(ttl_seconds))

deployment_config.completed_pipeline_ttl = pipeline_config.completed_pipeline_ttl_seconds

pipeline_spec.root.CopyFrom(
_build_component_spec_from_component_spec_structure(component_spec))

Expand Down
2 changes: 2 additions & 0 deletions sdk/python/kfp/dsl/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -292,4 +293,5 @@ def my_pipeline():
'IfPresentPlaceholder',
'ConcatPlaceholder',
'PipelineTask',
'PipelineConfig',
])
3 changes: 3 additions & 0 deletions sdk/python/kfp/dsl/component_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.

Expand All @@ -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,
)


Expand Down
4 changes: 4 additions & 0 deletions sdk/python/kfp/dsl/graph_component.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
34 changes: 34 additions & 0 deletions sdk/python/kfp/dsl/pipeline_config.py
Original file line number Diff line number Diff line change
@@ -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():
gregsheremeta marked this conversation as resolved.
Show resolved Hide resolved
"""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
6 changes: 5 additions & 1 deletion sdk/python/kfp/dsl/pipeline_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Optional nitpick: Add the new pipeline_config parameter to the docstring?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

"""Decorator used to construct a pipeline.

Example
Expand Down Expand Up @@ -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:
Expand All @@ -67,6 +70,7 @@ def my_pipeline(a: str, b: int):
name=name,
description=description,
display_name=display_name,
pipeline_config=pipeline_config,
)


Expand Down
Loading