expose add_op_transformer in the PipelineConf and add an example

samples/basic/pipeline_transformers.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+# Copyright 2019 Google LLC
+#
+# 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.
+
+
+import kfp
+from kfp import dsl
+
+
+def random_failure_op(exit_codes):
+  """A component that fails randomly."""
+  return dsl.ContainerOp(
+      name='random_failure',
+      image='python:alpine3.6',
+      command=['python', '-c'],
+      arguments=['import random; import sys; exit_code = random.choice(sys.argv[1].split(",")); print(exit_code); sys.exit(exit_code)', exit_codes]
+  )
+
+def add_retry(op):
+  op.set_retry(5)
+  return op
+
+@dsl.pipeline(
+    name='Retry random failures',
+    description='The pipeline includes two steps which fail randomly. It shows how to use ContainerOp(...).set_retry(...).'
+)
+def retry_sample_pipeline():
+  op1 = random_failure_op('0,1,2,3')
+  op2 = random_failure_op('0,1')
+  dsl.get_pipeline_conf().add_op_transformer(add_retry)
+
+if __name__ == '__main__':
+  kfp.compiler.Compiler().compile(retry_sample_pipeline, __file__ + '.zip')

sdk/python/kfp/dsl/_pipeline.py

@@ -104,6 +104,14 @@ def foo_pipeline(tag: str, pull_image_policy: str):
     self.artifact_location = artifact_location
     return self
 
+  def add_op_transformer(self, transformer):
+    """Configures the op_transformers which will be applied to all ops in the pipeline.
+
+    Args:
+      transformer: a function that takes a ContainOp as input and returns a ContainerOp
+    """
+    self.op_transformers.append(transformer)
+
 def get_pipeline_conf():
   """Configure the pipeline level setting to the current pipeline
     Note: call the function inside the user defined pipeline function.