Add resource op owner reference mapping

sdk/python/kfp_tekton/compiler/compiler.py

@@ -775,13 +775,16 @@ def _process_resourceOp(self, task_refs, pipeline):
         merge_strategy = op.resource.get('merge_strategy')
         success_condition = op.resource.get('successCondition')
         failure_condition = op.resource.get('failureCondition')
+        set_owner_reference = op.resource.get('setOwnerReference')
         task['params'] = [tp for tp in task.get('params', []) if tp.get('name') != "image"]
         if not merge_strategy:
           task['params'] = [tp for tp in task.get('params', []) if tp.get('name') != 'merge-strategy']
         if not success_condition:
           task['params'] = [tp for tp in task.get('params', []) if tp.get('name') != 'success-condition']
         if not failure_condition:
           task['params'] = [tp for tp in task.get('params', []) if tp.get('name') != "failure-condition"]
+        if not set_owner_reference:
+          task['params'] = [tp for tp in task.get('params', []) if tp.get('name') != "set-ownerreference"]
         for tp in task.get('params', []):
           if tp.get('name') == "action" and action:
             tp['value'] = action
@@ -791,6 +794,8 @@ def _process_resourceOp(self, task_refs, pipeline):
             tp['value'] = success_condition
           if tp.get('name') == "failure-condition" and failure_condition:
             tp['value'] = failure_condition
+          if tp.get('name') == "set-ownerreference" and set_owner_reference:
+            tp['value'] = set_owner_reference
           if tp.get('name') == "output":
             output_values = ''
             for value in sorted(list(op.attribute_outputs.items()), key=lambda x: x[0]):

sdk/python/tests/compiler/testdata/resourceop_basic.py

@@ -57,7 +57,8 @@ def resourceop_basic():
     op = dsl.ResourceOp(
         name='test-step',
         k8s_resource=json.loads(_CONTAINER_MANIFEST),
-        action='create'
+        action='create',
+        set_owner_reference='true'
         # success_condition='status.succeeded > 0',
         # failure_condition='status.failed > 0'
     )

sdk/python/tests/compiler/testdata/resourceop_basic.yaml

@@ -41,7 +41,7 @@ spec:
           - name: name
             valueFrom: '{.metadata.name}'
       - name: set-ownerreference
-        value: "false"
+        value: "true"
       taskSpec:
         params:
         - description: Action on the resource

sdk/python/tests/compiler/testdata/volume_op.yaml

@@ -50,8 +50,6 @@ spec:
             valueFrom: '{.metadata.name}'
           - name: size
             valueFrom: '{.status.capacity.storage}'
-      - name: set-ownerreference
-        value: "false"
       - name: size
         value: $(params.size)
       taskSpec: