[Test]Fix e2e test (#3471)

* Initial execution cache

This commit adds initial execution cache service. Including http service
and execution key generation.

* fix master

* fix go.sum

* Fix e2e test

* Add max_cache_staleness for flipA

* add comments

samples/core/recursion/recursion.py

@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# Notice: caching is tricky when recursion is involved. Please be careful and 
+# set proper max_cache_staleness in case of infinite loop.
 
 import kfp
 from kfp import dsl
@@ -45,6 +47,8 @@ def print_op(msg):
 def flip_component(flip_result):
     print_flip = print_op(flip_result)
     flipA = flip_coin_op().after(print_flip)
+    # set max_cache_staleness to 0 to prevent infinite loop due to caching
+    flipA.execution_options.caching_strategy.max_cache_staleness = "P0D"
     with dsl.Condition(flipA.output == 'heads'):
         # When the flip_component is called recursively, the flipA.output
         # from inside the graph component will be passed to the next flip_component
@@ -59,6 +63,8 @@ def flip_component(flip_result):
 )
 def flipcoin():
     first_flip = flip_coin_op()
+    # set max_cache_staleness to 0 to prevent infinite loop due to caching
+    first_flip.execution_options.caching_strategy.max_cache_staleness = "P0D"
     flip_loop = flip_component(first_flip.output)
     # flip_loop is a graph_component with the outputs field
     # filled with the returned dictionary.

sdk/python/kfp/dsl/_component.py

@@ -104,11 +104,14 @@ def graph_component(func):
 
   Usage:
   ```python
+  # Warning: caching is tricky when recursion is involved. Please be careful and 
+  # set proper max_cache_staleness in case of infinite loop.
   import kfp.dsl as dsl
   @dsl.graph_component
   def flip_component(flip_result):
     print_flip = PrintOp(flip_result)
     flipA = FlipCoinOp().after(print_flip)
+    flipA.execution_options.caching_strategy.max_cache_staleness = "P0D"
     with dsl.Condition(flipA.output == 'heads'):
       flip_component(flipA.output)
     return {'flip_result': flipA.output}

test/e2e_test_gke_v2.yaml

@@ -129,6 +129,7 @@ spec:
         withItems:
           - condition
           - exit_handler
+          - recursion
           - sequential
           - parallel_join
           - volume_ops