Add owner references check optional in watcher

This feature makes the owner reference check in watcher optional.

cmd/watcher/main.go

@@ -66,6 +66,7 @@ var (
 	labelSelector           = flag.String("label_selector", "", "Selector (label query) to filter objects to be deleted. Matching objects must satisfy all labels requirements to be eligible for deletion")
 	requeueInterval         = flag.Duration("requeue_interval", 10*time.Minute, "How long the Watcher waits to reprocess keys on certain events (e.g. an object doesn't match the provided selectors)")
 	namespace               = flag.String("namespace", corev1.NamespaceAll, "Should the Watcher only watch a single namespace, then this value needs to be set to the namespace name otherwise leave it empty.")
+	checkOwner              = flag.Bool("check_owner", true, "If enabled, owner references will be checked while deleting objects")
 )
 
 func main() {
@@ -97,6 +98,7 @@ func main() {
 		DisableAnnotationUpdate:      *disableCRDUpdate,
 		CompletedResourceGracePeriod: *completedRunGracePeriod,
 		RequeueInterval:              *requeueInterval,
+		CheckOwner:                   *checkOwner,
 	}
 
 	if selector := *labelSelector; selector != "" {

docs/watcher/README.md

@@ -57,3 +57,9 @@ metadata:
     results.tekton.dev/recordSummaryAnnotations: |-
       {"foo": "bar"}
 ```
+
+## Resource Deletion
+
+When the command line flag is `completed_run_grace_period` is set to any value other than `0`, resources will be deleted after the specified duration in the flag, calculated from the time of completion. If the value is < `0`, Runs will be deleted immediately after completion or failure.
+
+The flag `check_owner` allows additional check before deleting a resource. If set `true`, resources with any owner references set will not be deleted. When the flag is `false`, owner references will be not be checked before deletion. 

pkg/watcher/reconciler/config.go

@@ -38,6 +38,9 @@ type Config struct {
 	// How long the controller waits to reprocess keys on certain events
 	// (e.g. an object doesn't match the provided label selectors).
 	RequeueInterval time.Duration
+
+	// Check owner reference when deleting objects. By default, objects having owner references set won't be deleted.
+	CheckOwner bool
 }
 
 // GetDisableAnnotationUpdate returns whether annotation updates should be

pkg/watcher/reconciler/dynamic/dynamic.go

@@ -182,7 +182,7 @@ func (r *Reconciler) deleteUponCompletion(ctx context.Context, o results.Object)
 		return nil
 	}
 
-	if ownerReferences := o.GetOwnerReferences(); len(ownerReferences) > 0 {
+	if ownerReferences := o.GetOwnerReferences(); r.cfg.CheckOwner && len(ownerReferences) > 0 {
 		logger.Debugw("Resource is owned by another object, deferring deletion to parent resource(s)", zap.Any("results.tekton.dev/ownerReferences", ownerReferences))
 		return nil
 	}

pkg/watcher/reconciler/dynamic/dynamic_test.go

@@ -20,6 +20,8 @@ import (
 	"testing"
 	"time"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	watcherresults "github.com/tektoncd/results/pkg/watcher/results"
 
 	"github.com/google/uuid"
@@ -326,6 +328,55 @@ func TestReconcile_TaskRun(t *testing.T) {
 			t.Fatalf("Want NotFound, but got %v", err)
 		}
 	})
+
+	t.Run("delete object with owner references when owner check is disabled", func(t *testing.T) {
+		r.cfg.CheckOwner = false
+
+		taskrun.OwnerReferences = []metav1.OwnerReference{{
+			APIVersion: "v1",
+			Kind:       "test",
+			Name:       "test-parent",
+		}}
+
+		// Recreate the object to retest the deletion
+		if _, err := trclient.Create(ctx, taskrun, metav1.CreateOptions{}); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := r.Reconcile(ctx, taskrun); err != nil {
+			t.Fatal(err)
+		}
+
+		// Make sure that the resource no longer exists
+		if _, err := trclient.Get(ctx, taskrun.GetName(), metav1.GetOptions{}); !apierrors.IsNotFound(err) {
+			t.Fatalf("Want NotFound, but got %v", err)
+		}
+	})
+
+	t.Run("do not delete object with owner references when owner check is enabled", func(t *testing.T) {
+		r.cfg.CheckOwner = true
+
+		taskrun.OwnerReferences = []metav1.OwnerReference{{
+			APIVersion: "v1",
+			Kind:       "test",
+			Name:       "test-parent",
+			UID:        types.UID(uuid.NewString()),
+		}}
+
+		// Recreate the object to retest the deletion
+		if _, err := trclient.Create(ctx, taskrun, metav1.CreateOptions{}); err != nil {
+			t.Fatal(err)
+		}
+
+		if err := r.Reconcile(ctx, taskrun); err != nil {
+			t.Fatal(err)
+		}
+
+		// Make sure that the resource no longer exists
+		if _, err := trclient.Get(ctx, taskrun.GetName(), metav1.GetOptions{}); apierrors.IsNotFound(err) {
+			t.Fatalf("Want Found, but got %v", err)
+		}
+	})
 }
 
 // This is a simpler test than TaskRun, since most of this behavior is