add PipelineConfig to DSL to re-implement pipeline-level config

KFP v1 supported setting pipeline-level configuration via a
`PipelineConf` class. This class was deprecated and no replacement
was added to KFP v2.

add new PipelineConfig class to support setting pipeline-level
configuration in KFP v2.

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

Author: gregsheremeta

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

@@ -80,6 +80,9 @@ message PipelineSpec {
 
   // Optional field. The default root output directory of the pipeline.
   string default_pipeline_root = 10;
+
+  // Contains pipeline-level config options. See PipelineConfig DSL class.
+  PipelineConfig pipeline_config = 11;
 }
 
 // Definition of a component.
@@ -1084,9 +1087,13 @@ message SinglePlatformSpec {
   PlatformDeploymentConfig deployment_spec = 1;
 }
 
-
 message PlatformDeploymentConfig {
   // Map of executor label to executor-level config
   // Mirrors PipelineSpec.deployment_spec.executors structure
   map<string, google.protobuf.Struct> executors = 1;
 }
+
+// Spec for pipeline-level config options. See PipelineConfig DSL class.
+message PipelineConfig {
+  // TODO add pipeline-level configs
+}

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
+    pipeline_config_spec = pipeline_spec_pb2.PipelineConfig()
+    # TODO add pipeline-level config options
+
     pipeline_spec.root.CopyFrom(
         _build_component_spec_from_component_spec_structure(component_spec))
 
@@ -1951,6 +1958,9 @@ def create_pipeline_spec(
         dag_outputs=modified_pipeline_outputs_dict,
         structures_component_spec=component_spec)
 
+    # add pipeline-level config options
+    pipeline_spec.pipeline_config.CopyFrom(pipeline_config_spec)
+
     return pipeline_spec, platform_spec
 
 

sdk/python/kfp/compiler/pipeline_spec_builder.py

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

sdk/python/kfp/dsl/__init__.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/component_factory.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/graph_component.py

@@ -0,0 +1,23 @@
+# 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):
+        pass
+
+    # TODO add pipeline level configs

sdk/python/kfp/dsl/pipeline_config.py

@@ -17,17 +17,20 @@
 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
 
 
-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:
+def pipeline(
+        func: Optional[Callable] = None,
+        *,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        pipeline_root: Optional[str] = None,
+        display_name: Optional[str] = None,
+        pipeline_config: pipeline_config.PipelineConfig = None) -> Callable:
     """Decorator used to construct a pipeline.
 
     Example
@@ -49,6 +52,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.
+        pipeline_config: Pipeline-level config options.
     """
     if func is None:
         return functools.partial(
@@ -57,6 +61,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 +72,7 @@ def my_pipeline(a: str, b: int):
         name=name,
         description=description,
         display_name=display_name,
+        pipeline_config=pipeline_config,
     )