feat(sdk): Enable setting OwnerReference on ResourceOps. Fixes #1779 (#…

…4831)

Argo supports a field in the ResourceTemplate that makes the controller
add an owner reference of the workflow to the created resource since
v2.4.0 [1].
With the upgrade of Argo client [2] and deployment [3] we are now able
to exploit it.

We set it to 'false' by default on all ResourceOps (actually, leave it
empty).

Setting the field to 'true' for VolumeOps allows the garbage collection
of PVCs upon workflow cleanup [4].

[1] https://github.com/argoproj/argo/blob/v2.4.0/pkg/apis/workflow/v1alpha1/workflow_types.go#L1044-L1045
[2] #4498
[3] #3537
[4] #1779

Signed-off-by: Ilias Katsakioris <elikatsis@arrikto.com>

sdk/python/kfp/dsl/_resource_op.py

@@ -31,6 +31,7 @@ class Resource(object):
       "merge_strategy": "str",
       "success_condition": "str",
       "failure_condition": "str",
+      "set_owner_reference": "bool",
       "manifest": "str",
       "flags": "list[str]"
   }
@@ -39,6 +40,7 @@ class Resource(object):
       "merge_strategy": "str",
       "success_condition": "str",
       "failure_condition": "str",
+      "set_owner_reference": "bool",
       "manifest": "str",
       "flags": "list[str]"
   }
@@ -47,6 +49,7 @@ class Resource(object):
       "merge_strategy": "mergeStrategy",
       "success_condition": "successCondition",
       "failure_condition": "failureCondition",
+      "set_owner_reference": "setOwnerReference",
       "manifest": "manifest",
       "flags": "flags"
   }
@@ -56,13 +59,15 @@ def __init__(self,
                merge_strategy: str = None,
                success_condition: str = None,
                failure_condition: str = None,
+               set_owner_reference: bool = None,
                manifest: str = None,
                flags: Optional[List[str]] = None):
     """Create a new instance of Resource"""
     self.action = action
     self.merge_strategy = merge_strategy
     self.success_condition = success_condition
     self.failure_condition = failure_condition
+    self.set_owner_reference = set_owner_reference
     self.manifest = manifest
     self.flags = flags
 
@@ -98,6 +103,7 @@ def __init__(self,
                merge_strategy: str = None,
                success_condition: str = None,
                failure_condition: str = None,
+               set_owner_reference: bool = None,
                attribute_outputs: Optional[Dict[str, str]] = None,
                flags: Optional[List[str]] = None,
                **kwargs):
@@ -135,6 +141,7 @@ def __init__(self,
         "merge_strategy": merge_strategy,
         "success_condition": success_condition,
         "failure_condition": failure_condition,
+        "set_owner_reference": set_owner_reference,
         "flags": flags
     }
     # `resource` prop in `io.argoproj.workflow.v1alpha1.Template`

sdk/python/tests/compiler/compiler_tests.py

@@ -72,7 +72,8 @@ def my_pipeline(msg1, json, kind, msg2='value2'):
             name="resource"
           )
         ),
-        attribute_outputs={"out": json}
+        attribute_outputs={"out": json},
+        set_owner_reference=True
       )
       golden_output = {
         'container': {
@@ -147,7 +148,8 @@ def my_pipeline(msg1, json, kind, msg2='value2'):
             "kind: '{{inputs.parameters.kind}}'\n"
             "metadata:\n"
             "  name: resource\n"
-          )
+          ),
+          'setOwnerReference': True
         }
       }