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

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) and 'true' for VolumeOps. This allows the garbage collection of
PVCs upon workflow cleanup without any further change [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

@@ -32,6 +32,7 @@ class Resource(object):
         "merge_strategy": "str",
         "success_condition": "str",
         "failure_condition": "str",
+        "set_owner_reference": "bool",
         "manifest": "str",
         "flags": "list[str]"
     }
@@ -40,6 +41,7 @@ class Resource(object):
         "merge_strategy": "str",
         "success_condition": "str",
         "failure_condition": "str",
+        "set_owner_reference": "bool",
         "manifest": "str",
         "flags": "list[str]"
     }
@@ -48,6 +50,7 @@ class Resource(object):
         "merge_strategy": "mergeStrategy",
         "success_condition": "successCondition",
         "failure_condition": "failureCondition",
+        "set_owner_reference": "setOwnerReference",
         "manifest": "manifest",
         "flags": "flags"
     }
@@ -57,13 +60,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
 
@@ -100,6 +105,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):
@@ -118,7 +124,7 @@ def __init__(self,
 
         if merge_strategy and action != "apply":
             raise ValueError("You can't set merge_strategy when action != 'apply'")
-        
+
         # if action is delete, there should not be any outputs, success_condition, and failure_condition
         if action == "delete" and (success_condition or failure_condition or attribute_outputs):
             raise ValueError("You can't set success_condition, failure_condition, or attribute_outputs when action == 'delete'")
@@ -130,6 +136,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`