Repository: tektoncd/results
Pull request #255 (2022-10-18 13:56:07)

Fix resource deletion and refactor watcher
## What

This pull request resolves https://github.com/tektoncd/results/issues/254. The
following fixes and improvements were made:

- Fix object deletion upon completion: apparently, the main cause of the
instability was the use of the record to determine the time at which the Run
object had been updated. I changed the watcher to use the
`Status.CompletionTime` since this approach seems to be more
reliable. Currently, if the Tekton
pipelines controller resyncs the object and transitions the status, the object
will change, the record will be updated and this will cause discrepancies in the
verification of the grace period expiration.
- The aforementioned change fixes end-to-end tests. Now, the test suite is
passing consistently in my machine (this
[pr](https://github.com/tektoncd/results/pull/253) should be merged as well
since it fixes other issues in the current e2e setup).
- Improve logs emitted by the watcher: favor structured logging since it
simplifies a lot the debugging with log indexer tools such as Splunk and others.
- Refactor the controller's code to make it simpler and hopefully, easier to
reason about. I tried to make things more uniform and meaningful for new
collaborators. One of the most relevant changes was removing the enqueue
function that was being passed around and replacing the approach with the
knative's `controller.NewRequeueAfter` function which returns an error that
tells knative to requeue the key after a given duration.

tekton-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **vdemeester** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @vdemeester` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://prow.tekton.dev/command-help?repo=tektoncd%2Fresults).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/tektoncd/results/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["vdemeester"]} -->

Patch:
diff --git a/pkg/api/server/v1alpha2/records.go b/pkg/api/server/v1alpha2/records.go
index 6aad5a20..f136c0aa 100644
--- a/pkg/api/server/v1alpha2/records.go
+++ b/pkg/api/server/v1alpha2/records.go
@@ -137,7 +137,11 @@ func (s *Server) ListRecords(ctx context.Context, req *pb.ListRecordsRequest) (*
 	if req.GetParent() == "" {
 		return nil, status.Error(codes.InvalidArgument, "parent missing")
 	}
-	if err := s.auth.Check(ctx, req.GetParent(), auth.ResourceRecords, auth.PermissionList); err != nil {
+	parent, _, err := result.ParseName(req.GetParent())
+	if err != nil {
+		return nil, err
+	}
+	if err := s.auth.Check(ctx, parent, auth.ResourceRecords, auth.PermissionList); err != nil {
 		return nil, err
 	}
 
diff --git a/pkg/watcher/reconciler/dynamic/dynamic.go b/pkg/watcher/reconciler/dynamic/dynamic.go
index 5b35c7e5..f368c3ad 100644
--- a/pkg/watcher/reconciler/dynamic/dynamic.go
+++ b/pkg/watcher/reconciler/dynamic/dynamic.go
@@ -16,18 +16,22 @@ package dynamic
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/jonboulle/clockwork"
+	pipelinev1beta1 "github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
 	"github.com/tektoncd/results/pkg/watcher/convert"
 	"github.com/tektoncd/results/pkg/watcher/reconciler"
 	"github.com/tektoncd/results/pkg/watcher/reconciler/annotation"
 	"github.com/tektoncd/results/pkg/watcher/results"
 	pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto"
+	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/pkg/apis"
+	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 )
 
@@ -41,23 +45,21 @@ type Reconciler struct {
 	resultsClient *results.Client
 	objectClient  ObjectClient
 	cfg           *reconciler.Config
-	enqueue       func(interface{}, time.Duration)
 }
 
 // NewDynamicReconciler creates a new dynamic Reconciler.
-func NewDynamicReconciler(rc pb.ResultsClient, oc ObjectClient, cfg *reconciler.Config, enqueue func(interface{}, time.Duration)) *Reconciler {
+func NewDynamicReconciler(rc pb.ResultsClient, oc ObjectClient, cfg *reconciler.Config) *Reconciler {
 	return &Reconciler{
 		resultsClient: &results.Client{ResultsClient: rc},
 		objectClient:  oc,
 		cfg:           cfg,
-		enqueue:       enqueue,
 	}
 }
 
 // Reconcile handles result/record uploading for the given Run object.
 // If enabled, the object may be deleted upon successful result upload.
 func (r *Reconciler) Reconcile(ctx context.Context, o results.Object) error {
-	log := logging.FromContext(ctx)
+	logger := logging.FromContext(ctx)
 
 	if o.GetObjectKind().GroupVersionKind().Empty() {
 		gvk, err := convert.InferGVK(o)
@@ -65,55 +67,101 @@ func (r *Reconciler) Reconcile(ctx context.Context, o results.Object) error {
 			return err
 		}
 		o.GetObjectKind().SetGroupVersionKind(gvk)
-		log.Infof("Post-GVK Object: %v", o)
+		logger.Debugf("Post SetGroupVersionKind: %s", o.GetObjectKind().GroupVersionKind().String())
 	}
 
 	// Update record.
 	result, record, err := r.resultsClient.Put(ctx, o)
 	if err != nil {
-		log.Errorf("error updating Record: %v", err)
+		logger.Errorw("Error updating Record", zap.Error(err))
 		return err
 	}
 
-	if a := o.GetAnnotations(); !r.cfg.GetDisableAnnotationUpdate() && (result.GetName() != a[annotation.Result] || record.GetName() != a[annotation.Record]) {
+	logger = logger.With(zap.String("results.tekton.dev/result", result.Name),
+		zap.String("results.tekton.dev/record", record.Name))
+
+	if err := r.addResultsAnnotations(logging.WithLogger(ctx, logger), o, result, record); err != nil {
+		return err
+	}
+
+	return r.deleteUponCompletion(logging.WithLogger(ctx, logger), o)
+}
+
+// addResultsAnnotations adds Results annotations to the object in question if
+// annotation patching is enabled.
+func (r *Reconciler) addResultsAnnotations(ctx context.Context, o results.Object, result *pb.Result, record *pb.Record) error {
+	logger := logging.FromContext(ctx)
+
+	objectAnnotations := o.GetAnnotations()
+	if r.cfg.GetDisableAnnotationUpdate() {
+		logger.Info("Skipping CRD annotation patch: annotation update is disabled")
+	} else if result.GetName() == objectAnnotations[annotation.Result] && record.GetName() == objectAnnotations[annotation.Record] {
+		logger.Info("Skipping CRD annotation patch: Result annotations are already set")
+	} else {
 		// Update object with Result Annotations.
 		patch, err := annotation.Add(result.GetName(), record.GetName())
 		if err != nil {
-			log.Errorf("error adding Result annotations: %v", err)
+			logger.Errorw("Error adding Result annotations", zap.Error(err))
 			return err
 		}
 		if err := r.objectClient.Patch(ctx, o.GetName(), types.MergePatchType, patch, metav1.PatchOptions{}); err != nil {
-			log.Errorf("Patch: %v", err)
+			logger.Errorw("Error patching object", zap.Error(err))
 			return err
 		}
-	} else {
-		log.Infof("skipping CRD patch - annotation patching disabled [%t] or annotations already match", r.cfg.GetDisableAnnotationUpdate())
+	}
+	return nil
+}
+
+// deleteUponCompletion deletes the object in question when the following
+// conditions are met:
+// * The resource deletion is enabled in the config (the grace period is greater
+// than 0).
+// * The object is done and it isn't owned by other object.
+// * The configured grace period has elapsed since the object's completion.
+func (r *Reconciler) deleteUponCompletion(ctx context.Context, o results.Object) error {
+	logger := logging.FromContext(ctx)
+
+	gracePeriod := r.cfg.GetCompletedResourceGracePeriod()
+	logger = logger.With(zap.Duration("results.tekton.dev/gracePeriod", gracePeriod))
+	if gracePeriod == 0 {
+		logger.Info("Skipping resource deletion: deletion is disabled")
+		return nil
 	}
 
-	// If the Object is complete and not yet marked for deletion, cleanup the run resource from the cluster.
-	done := isDone(o)
-	log.Infof("should skipping resource deletion?  - done: %t, delete enabled: %t", done, r.cfg.GetCompletedResourceGracePeriod() != 0)
-	if done && r.cfg.GetCompletedResourceGracePeriod() != 0 {
-		if o := o.GetOwnerReferences(); len(o) > 0 {
-			log.Infof("resource is owned by another object, defering deletion to parent resource(s): %v", o)
-			return nil
-		}
+	if !isDone(o) {
+		logger.Info("Skipping resource deletion: object is not done yet")
+		return nil
+	}
 
-		// We haven't hit the grace period yet - reenqueue the key for processing later.
-		if s := clock.Since(record.GetUpdatedTime().AsTime()); s < r.cfg.GetCompletedResourceGracePeriod() {
-			log.Infof("object is not ready for deletion - grace period: %v, time since completion: %v", r.cfg.GetCompletedResourceGracePeriod(), s)
-			r.enqueue(o, r.cfg.GetCompletedResourceGracePeriod())
-			return nil
-		}
-		log.Infof("deleting PipelineRun UID %s", o.GetUID())
-		if err := r.objectClient.Delete(ctx, o.GetName(), metav1.DeleteOptions{
-			Preconditions: metav1.NewUIDPreconditions(string(o.GetUID())),
-		}); err != nil && !errors.IsNotFound(err) {
-			log.Errorf("PipelineRun.Delete: %v", err)
-			return err
-		}
-	} else {
-		log.Infof("skipping resource deletion - done: %t, delete enabled: %t, %v", done, r.cfg.GetCompletedResourceGracePeriod() != 0, r.cfg.GetCompletedResourceGracePeriod())
+	if ownerReferences := o.GetOwnerReferences(); len(ownerReferences) > 0 {
+		logger.Infow("Resource is owned by another object, deferring deletion to parent resource(s)", zap.Any("results.tekton.dev/ownerReferences", ownerReferences))
+		return nil
+	}
+
+	completionTime, err := getCompletionTime(o)
+	if err != nil {
+		return err
+	}
+
+	// This isn't probable since the object is done, but defensive
+	// programming never hurts.
+	if completionTime == nil {
+		logger.Info("Object's completion time isn't set yet - requeuing to process later")
+		return controller.NewRequeueAfter(gracePeriod)
+	}
+
+	if timeSinceCompletion := clock.Since(*completionTime); timeSinceCompletion < gracePeriod {
+		requeueAfter := gracePeriod - timeSinceCompletion
+		logger.Infow("Object is not ready for deletion yet - requeuing to process later", zap.Duration("results.tekton.dev/requeueAfter", requeueAfter))
+		return controller.NewRequeueAfter(requeueAfter)
+	}
+
+	logger.Infow("Deleting object", zap.String("results.tekton.dev/uid", string(o.GetUID())))
+	if err := r.objectClient.Delete(ctx, o.GetName(), metav1.DeleteOptions{
+		Preconditions: metav1.NewUIDPreconditions(string(o.GetUID())),
+	}); err != nil && !errors.IsNotFound(err) {
+		logger.Errorw("Error deleting object", zap.Error(err))
+		return err
 	}
 
 	return nil
@@ -122,3 +170,26 @@ func (r *Reconciler) Reconcile(ctx context.Context, o results.Object) error {
 func isDone(o results.Object) bool {
 	return o.GetStatusCondition().GetCondition(apis.ConditionSucceeded).IsTrue()
 }
+
+// getCompletionTime returns the completion time of the object (PipelineRun or
+// TaskRun) in question.
+func getCompletionTime(object results.Object) (*time.Time, error) {
+	var completionTime *time.Time
+
+	switch o := object.(type) {
+
+	case *pipelinev1beta1.PipelineRun:
+		if o.Status.CompletionTime != nil {
+			completionTime = &o.Status.CompletionTime.Time
+		}
+
+	case *pipelinev1beta1.TaskRun:
+		if o.Status.CompletionTime != nil {
+			completionTime = &o.Status.CompletionTime.Time
+		}
+
+	default:
+		return nil, controller.NewPermanentError(fmt.Errorf("error getting completion time from incoming object: unrecognized type %T", o))
+	}
+	return completionTime, nil
+}
diff --git a/pkg/watcher/reconciler/pipelinerun/controller.go b/pkg/watcher/reconciler/pipelinerun/controller.go
index a5761463..c660445b 100644
--- a/pkg/watcher/reconciler/pipelinerun/controller.go
+++ b/pkg/watcher/reconciler/pipelinerun/controller.go
@@ -48,7 +48,6 @@ func NewControllerWithConfig(ctx context.Context, client pb.ResultsClient, cfg *
 		Logger:        logging.FromContext(ctx),
 		WorkQueueName: "PipelineRuns",
 	})
-	c.enqueue = impl.EnqueueAfter
 
 	informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    impl.Enqueue,
diff --git a/pkg/watcher/reconciler/pipelinerun/reconciler.go b/pkg/watcher/reconciler/pipelinerun/reconciler.go
index 8c0a4c24..a964a0fa 100644
--- a/pkg/watcher/reconciler/pipelinerun/reconciler.go
+++ b/pkg/watcher/reconciler/pipelinerun/reconciler.go
@@ -2,7 +2,6 @@ package pipelinerun
 
 import (
 	"context"
-	"time"
 
 	"github.com/tektoncd/pipeline/pkg/client/clientset/versioned"
 	"github.com/tektoncd/pipeline/pkg/client/listers/pipeline/v1beta1"
@@ -25,15 +24,14 @@ type Reconciler struct {
 	lister    v1beta1.PipelineRunLister
 	k8sclient versioned.Interface
 	cfg       *reconciler.Config
-	enqueue   func(interface{}, time.Duration)
 }
 
 // Check that our Reconciler is LeaderAware.
 var _ knativereconciler.LeaderAware = (*Reconciler)(nil)
 
 func (r *Reconciler) Reconcile(ctx context.Context, key string) error {
-	log := logging.FromContext(ctx)
-	log.With(zap.String("key", key))
+	log := logging.FromContext(ctx).With(zap.String("results.tekton.dev/kind", "PipelineRun"))
+	log.Info("Reconciling PipelineRun")
 
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
@@ -42,7 +40,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, key string) error {
 	}
 
 	if !r.IsLeaderFor(types.NamespacedName{Namespace: namespace, Name: name}) {
-		log.Info("Skipping PipelineRun key because this instance isn't its leader")
+		log.Debug("Skipping PipelineRun key because this instance isn't its leader")
 		return controller.NewSkipKey(key)
 	}
 
@@ -55,8 +53,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, key string) error {
 		PipelineRunInterface: r.k8sclient.TektonV1beta1().PipelineRuns(namespace),
 	}
 
-	dyn := dynamic.NewDynamicReconciler(r.client, k8sclient, r.cfg, r.enqueue)
-	if err := dyn.Reconcile(ctx, pr); err != nil {
+	dyn := dynamic.NewDynamicReconciler(r.client, k8sclient, r.cfg)
+	if err := dyn.Reconcile(logging.WithLogger(ctx, log), pr); err != nil {
 		return err
 	}
 
diff --git a/pkg/watcher/reconciler/taskrun/controller.go b/pkg/watcher/reconciler/taskrun/controller.go
index b1c92628..76ee66d4 100644
--- a/pkg/watcher/reconciler/taskrun/controller.go
+++ b/pkg/watcher/reconciler/taskrun/controller.go
@@ -48,7 +48,6 @@ func NewControllerWithConfig(ctx context.Context, client pb.ResultsClient, cfg *
 		Logger:        logging.FromContext(ctx),
 		WorkQueueName: "TaskRuns",
 	})
-	c.enqueue = impl.EnqueueAfter
 
 	informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    impl.Enqueue,
diff --git a/pkg/watcher/reconciler/taskrun/reconciler.go b/pkg/watcher/reconciler/taskrun/reconciler.go
index c25ff272..d14dcfbd 100644
--- a/pkg/watcher/reconciler/taskrun/reconciler.go
+++ b/pkg/watcher/reconciler/taskrun/reconciler.go
@@ -2,7 +2,6 @@ package taskrun
 
 import (
 	"context"
-	"time"
 
 	"knative.dev/pkg/controller"
 	knativereconciler "knative.dev/pkg/reconciler"
@@ -26,15 +25,14 @@ type Reconciler struct {
 	lister    v1beta1.TaskRunLister
 	k8sclient versioned.Interface
 	cfg       *reconciler.Config
-	enqueue   func(interface{}, time.Duration)
 }
 
 // Check that our Reconciler is LeaderAware.
 var _ knativereconciler.LeaderAware = (*Reconciler)(nil)
 
 func (r *Reconciler) Reconcile(ctx context.Context, key string) error {
-	log := logging.FromContext(ctx)
-	log.With(zap.String("key", key))
+	log := logging.FromContext(ctx).With(zap.String("results.tekton.dev/kind", "TaskRun"))
+	log.Info("Reconciling TaskRun")
 
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
@@ -43,7 +41,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, key string) error {
 	}
 
 	if !r.IsLeaderFor(types.NamespacedName{Namespace: namespace, Name: name}) {
-		log.Info("Skipping TaskRun key because this instance isn't its leader")
+		log.Debug("Skipping TaskRun key because this instance isn't its leader")
 		return controller.NewSkipKey(key)
 	}
 
@@ -56,8 +54,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, key string) error {
 		TaskRunInterface: r.k8sclient.TektonV1beta1().TaskRuns(namespace),
 	}
 
-	dyn := dynamic.NewDynamicReconciler(r.client, k8sclient, r.cfg, r.enqueue)
-	if err := dyn.Reconcile(ctx, tr); err != nil {
+	dyn := dynamic.NewDynamicReconciler(r.client, k8sclient, r.cfg)
+	if err := dyn.Reconcile(logging.WithLogger(ctx, log), tr); err != nil {
 		return err
 	}
 


Test patch:
diff --git a/pkg/watcher/reconciler/dynamic/dynamic_test.go b/pkg/watcher/reconciler/dynamic/dynamic_test.go
index 422339a2..ead722ff 100644
--- a/pkg/watcher/reconciler/dynamic/dynamic_test.go
+++ b/pkg/watcher/reconciler/dynamic/dynamic_test.go
@@ -136,7 +136,7 @@ func TestReconcile_TaskRun(t *testing.T) {
 		DisableAnnotationUpdate: true,
 	}
 
-	r := NewDynamicReconciler(results, trclient, cfg, nil)
+	r := NewDynamicReconciler(results, trclient, cfg)
 	if err := r.Reconcile(ctx, taskrun); err != nil {
 		t.Fatal(err)
 	}
@@ -179,19 +179,27 @@ func TestReconcile_TaskRun(t *testing.T) {
 		// Enable object deletion, re-reconcile
 		cfg.CompletedResourceGracePeriod = 1 * time.Second
 
-		reenqueued := false
-		r.enqueue = func(i interface{}, d time.Duration) {
-			reenqueued = true
+		isRequeueKey := func(err error) bool {
+			ok, _ := controller.IsRequeueKey(err)
+			return ok
 		}
-		if err := r.Reconcile(ctx, taskrun); err != nil {
-			t.Fatal(err)
+
+		// The controller must requeue the object since the
+		// CompletionTime field is nil.
+		if err := r.Reconcile(ctx, taskrun); !isRequeueKey(err) {
+			t.Fatalf("Want a controller.RequeueKey error, but got %v", err)
 		}
 
-		if !reenqueued {
-			t.Fatal("expected object to be reenqueued")
+		taskrun.Status.CompletionTime = &metav1.Time{Time: fakeclock.Now()}
+		// The controller must requeue the object since the grace period
+		// hasn't elapsed yet.
+		if err := r.Reconcile(ctx, taskrun); !isRequeueKey(err) {
+			t.Fatalf("Want a controller.RequeueKey error, but got %v", err)
 		}
 
-		fakeclock.Advance(1 * time.Minute)
+		// Advance the clock to force the grace period to elapse. The
+		// TaskRun must be processed and deleted from the cluster.
+		fakeclock.Advance(2 * time.Second)
 		if err := r.Reconcile(ctx, taskrun); err != nil {
 			t.Fatal(err)
 		}
@@ -219,7 +227,7 @@ func TestReconcile_PipelineRun(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	r := NewDynamicReconciler(results, prclient, nil, nil)
+	r := NewDynamicReconciler(results, prclient, nil)
 	if err := r.Reconcile(ctx, pipelinerun); err != nil {
 		t.Fatal(err)
 	}
diff --git a/test/e2e/01-install.sh b/test/e2e/01-install.sh
index df1c4d08..4c2bc6df 100755
--- a/test/e2e/01-install.sh
+++ b/test/e2e/01-install.sh
@@ -30,41 +30,50 @@ kubectl create secret generic tekton-results-postgres --namespace="tekton-pipeli
 
 echo "Generating TLS key pair..."
 set +e
-  openssl req -x509 \
-     -newkey rsa:4096 \
-     -keyout "/tmp/tekton-results-key.pem" \
-     -out "/tmp/tekton-results-cert.pem" \
-     -days 365 \
-     -nodes \
-     -subj "/CN=tekton-results-api-service.tekton-pipelines.svc.cluster.local" \
-     -addext "subjectAltName = DNS:tekton-results-api-service.tekton-pipelines.svc.cluster.local"
+openssl req -x509 \
+        -newkey rsa:4096 \
+        -keyout "/tmp/tekton-results-key.pem" \
+        -out "/tmp/tekton-results-cert.pem" \
+        -days 365 \
+        -nodes \
+        -subj "/CN=tekton-results-api-service.tekton-pipelines.svc.cluster.local" \
+        -addext "subjectAltName = DNS:tekton-results-api-service.tekton-pipelines.svc.cluster.local"
 
-  if [ $? -ne 0 ] ; then
+if [ $? -ne 0 ] ; then
     # LibreSSL didn't support the -addext flag until version 3.1.0 but
     # version 2.8.3 ships with MacOS Big Sur. So let's try a different way...
     echo "Falling back to legacy libressl cert generation"
     openssl req -x509 \
-      -verbose \
-      -config <(cat /etc/ssl/openssl.cnf <(printf "[SAN]\nsubjectAltName = DNS:tekton-results-api-service.tekton-pipelines.svc.cluster.local")) \
-      -extensions SAN \
-      -newkey rsa:4096 \
-      -keyout "/tmp/tekton-results-key.pem" \
-      -out "/tmp/tekton-results-cert.pem" \
-      -days 365 \
-      -nodes \
-      -subj "/CN=tekton-results-api-service.tekton-pipelines.svc.cluster.local"
+            -verbose \
+            -config <(cat /etc/ssl/openssl.cnf <(printf "[SAN]\nsubjectAltName = DNS:tekton-results-api-service.tekton-pipelines.svc.cluster.local")) \
+            -extensions SAN \
+            -newkey rsa:4096 \
+            -keyout "/tmp/tekton-results-key.pem" \
+            -out "/tmp/tekton-results-cert.pem" \
+            -days 365 \
+            -nodes \
+            -subj "/CN=tekton-results-api-service.tekton-pipelines.svc.cluster.local"
 
     if [ $? -ne 0 ] ; then
-      echo "There was an error generating certificates"
-      exit 1
+        echo "There was an error generating certificates"
+        exit 1
     fi
-  fi
+fi
 set -e
 kubectl create secret tls -n tekton-pipelines tekton-results-tls --cert="/tmp/tekton-results-cert.pem" --key="/tmp/tekton-results-key.pem" || true
 
 echo "Installing Tekton Results..."
 kubectl kustomize "${ROOT}/test/e2e/kustomize" | ko apply -f -
 
+echo "Fetching access tokens..."
+tokens_dir=/tmp/tekton-results/tokens
+mkdir -p $tokens_dir
+service_accounts=(all-namespaces-read-access single-namespace-read-access)
+for service_account in ${service_accounts[@]}; do
+    kubectl get secrets -o jsonpath="{.items[?(@.metadata.annotations['kubernetes\.io/service-account\.name']=='$service_account')].data.token}"|base64 --decode > $tokens_dir/$service_account
+    echo "Created $tokens_dir/$service_account"
+done
+
 echo "Waiting for deployments to be ready..."
 kubectl wait deployment "tekton-results-api" --namespace="tekton-pipelines" --for="condition=available" --timeout="60s"
 kubectl wait deployment "tekton-results-watcher" --namespace="tekton-pipelines" --for="condition=available" --timeout="60s"
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
index 8f9b9e7c..951ca387 100644
--- a/test/e2e/e2e_test.go
+++ b/test/e2e/e2e_test.go
@@ -21,6 +21,13 @@ import (
 	"context"
 	"io/ioutil"
 	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	resultsv1alpha2 "github.com/tektoncd/results/proto/v1alpha2/results_go_proto"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/testing/protocmp"
+
 	"time"
 
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
@@ -138,3 +145,149 @@ func client(t *testing.T) *clientset.TektonV1beta1Client {
 	}
 	return clientset.NewForConfigOrDie(config)
 }
+
+func TestListResults(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("list results under the default parent", func(t *testing.T) {
+		client := newResultsClient(t, allNamespacesReadAccess)
+		resp, err := client.ListResults(ctx, &resultsv1alpha2.ListResultsRequest{Parent: "default"})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if length := len(resp.Results); length == 0 {
+			t.Error("No results returned by the API server")
+		} else {
+			t.Logf("Found %d results\n", length)
+		}
+	})
+
+	t.Run("list results across parents", func(t *testing.T) {
+		client := newResultsClient(t, allNamespacesReadAccess)
+
+		// For the purposes of this test suite, listing results under
+		// the `default` parent or using the `-` symbol must return the
+		// same items. Therefore, let's run both queries and make sure
+		// that results are identical.
+
+		want, err := client.ListResults(ctx, &resultsv1alpha2.ListResultsRequest{
+			Parent:  "default",
+			OrderBy: "created_time",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		got, err := client.ListResults(ctx, &resultsv1alpha2.ListResultsRequest{
+			Parent:  "-",
+			OrderBy: "created_time",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if diff := cmp.Diff(want.Results, got.Results, protocmp.Transform()); diff != "" {
+			t.Errorf("Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("return an error because the identity isn't authorized to access all namespaces", func(t *testing.T) {
+		client := newResultsClient(t, singleNamespaceReadAccess)
+		_, err := client.ListResults(ctx, &resultsv1alpha2.ListResultsRequest{Parent: "-"})
+		if err == nil {
+			t.Fatal("Want an unauthenticated error, but the request succeeded")
+		}
+
+		if status.Code(err) != codes.Unauthenticated {
+			t.Errorf("API server returned an unexpected error: %v", err)
+		}
+	})
+
+	t.Run("list results under the default parent using the identity with more limited access", func(t *testing.T) {
+		client := newResultsClient(t, singleNamespaceReadAccess)
+		resp, err := client.ListResults(ctx, &resultsv1alpha2.ListResultsRequest{Parent: "default"})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if length := len(resp.Results); length == 0 {
+			t.Error("No results returned by the API server")
+		} else {
+			t.Logf("Found %d results\n", length)
+		}
+	})
+}
+
+func TestListRecords(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("list records by omitting the result name", func(t *testing.T) {
+		client := newResultsClient(t, allNamespacesReadAccess)
+		resp, err := client.ListRecords(ctx, &resultsv1alpha2.ListRecordsRequest{Parent: "default/results/-"})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if length := len(resp.Records); length == 0 {
+			t.Error("No records returned by the API server")
+		} else {
+			t.Logf("Found %d records\n", length)
+		}
+	})
+
+	t.Run("list records by omitting the parent and result names", func(t *testing.T) {
+		client := newResultsClient(t, allNamespacesReadAccess)
+
+		// For the purposes of this test suite, listing records under
+		// the `default/results/-` result or using the `-/results/-`
+		// form must return the same items. Therefore, let's run both
+		// queries and make sure that results are identical.
+
+		want, err := client.ListRecords(ctx, &resultsv1alpha2.ListRecordsRequest{
+			Parent:  "default/results/-",
+			OrderBy: "created_time",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		got, err := client.ListRecords(ctx, &resultsv1alpha2.ListRecordsRequest{
+			Parent:  "-/results/-",
+			OrderBy: "created_time",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if diff := cmp.Diff(want.Records, got.Records, protocmp.Transform()); diff != "" {
+			t.Errorf("Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("return an error because the identity isn't authorized to access all namespaces", func(t *testing.T) {
+		client := newResultsClient(t, singleNamespaceReadAccess)
+		_, err := client.ListRecords(ctx, &resultsv1alpha2.ListRecordsRequest{Parent: "-/results/-"})
+		if err == nil {
+			t.Fatal("Want an unauthenticated error, but the request succeeded")
+		}
+
+		if status.Code(err) != codes.Unauthenticated {
+			t.Errorf("API server returned an unexpected error: %v", err)
+		}
+	})
+
+	t.Run("list records using the identity with more limited access", func(t *testing.T) {
+		client := newResultsClient(t, singleNamespaceReadAccess)
+		resp, err := client.ListRecords(ctx, &resultsv1alpha2.ListRecordsRequest{Parent: "default/results/-"})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if length := len(resp.Records); length == 0 {
+			t.Error("No records returned by the API server")
+		} else {
+			t.Logf("Found %d records\n", length)
+		}
+	})
+}
diff --git a/test/e2e/grpc_client.go b/test/e2e/grpc_client.go
new file mode 100644
index 00000000..43f3da2e
--- /dev/null
+++ b/test/e2e/grpc_client.go
@@ -0,0 +1,90 @@
+// Copyright 2022 The Tekton Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package e2e
+
+import (
+	"context"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc/credentials/oauth"
+	"k8s.io/client-go/transport"
+
+	resultsv1alpha2 "github.com/tektoncd/results/proto/v1alpha2/results_go_proto"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+const (
+	certFile                  = "/tmp/tekton-results-cert.pem"
+	apiServerName             = "tekton-results-api-service.tekton-pipelines.svc.cluster.local"
+	apiServerAddress          = "localhost:50051"
+	allNamespacesReadAccess   = "/tmp/tekton-results/tokens/all-namespaces-read-access"
+	singleNamespaceReadAccess = "/tmp/tekton-results/tokens/single-namespace-read-access"
+)
+
+// newResultsClient creates a new Results GRPC client to talk to the Results API
+// server.
+func newResultsClient(t *testing.T, accessTokenPath string) resultsv1alpha2.ResultsClient {
+	t.Helper()
+
+	certPool, err := loadCertificates()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := openConnection(certPool, accessTokenPath)
+	if err != nil {
+		t.Fatalf("Error connecting to the Results API server: %v", err)
+	}
+
+	return resultsv1alpha2.NewResultsClient(conn)
+}
+
+func loadCertificates() (*x509.CertPool, error) {
+	certPool, err := x509.SystemCertPool()
+	if err != nil {
+		return nil, fmt.Errorf("error loading system cert pool: %w", err)
+	}
+
+	cert, err := os.ReadFile(certFile)
+	if err != nil {
+		return nil, err
+	}
+
+	if !certPool.AppendCertsFromPEM(cert) {
+		return nil, errors.New("unable to append certificate to cert pool")
+	}
+
+	return certPool, nil
+}
+
+func openConnection(certPool *x509.CertPool, accessTokenPath string) (*grpc.ClientConn, error) {
+	transportCredentials := credentials.NewClientTLSFromCert(certPool, apiServerName)
+	tokenSource := transport.NewCachedFileTokenSource(accessTokenPath)
+	opts := []grpc.DialOption{
+		grpc.WithBlock(),
+		grpc.WithDefaultCallOptions(grpc.PerRPCCredentials(oauth.TokenSource{TokenSource: tokenSource})),
+		grpc.WithTransportCredentials(transportCredentials),
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	return grpc.DialContext(ctx, apiServerAddress, opts...)
+}
diff --git a/test/e2e/kind-cluster.yaml b/test/e2e/kind-cluster.yaml
index 59590bf7..ac3b4a98 100644
--- a/test/e2e/kind-cluster.yaml
+++ b/test/e2e/kind-cluster.yaml
@@ -16,7 +16,17 @@
 kind: Cluster
 apiVersion: kind.x-k8s.io/v1alpha4
 nodes:
-- role: control-plane
-- role: worker
-- role: worker
-- role: worker
\ No newline at end of file
+- &node
+  role: control-plane
+  # Set a fixed Kubernetes version.
+  image: kindest/node:v1.22.13@sha256:4904eda4d6e64b402169797805b8ec01f50133960ad6c19af45173a27eadf959
+
+# Merge the same definition as the control-plane to reuse the same Kindest image
+# in the data-plane. We'll set specific values of the data-plane afterwards.
+# For further details on merge keys, please refer to https://learnxinyminutes.com/docs/yaml/.
+- <<: *node
+  role: worker
+  extraPortMappings:
+    # API GRPC server
+    - containerPort: 30000
+      hostPort: 50051
diff --git a/test/e2e/kustomize/api-server.yaml b/test/e2e/kustomize/api-server.yaml
new file mode 100644
index 00000000..49ce7fd5
--- /dev/null
+++ b/test/e2e/kustomize/api-server.yaml
@@ -0,0 +1,23 @@
+# Copyright 2022 The Tekton Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+- op: replace
+  path: /spec/ports/0
+  value:
+    name: grpc
+    port: 50051
+    nodePort: 30000
+- op: add
+  path: /spec/type
+  value: NodePort
diff --git a/test/e2e/kustomize/kustomization.yaml b/test/e2e/kustomize/kustomization.yaml
index 3be4ef10..8d57600c 100644
--- a/test/e2e/kustomize/kustomization.yaml
+++ b/test/e2e/kustomize/kustomization.yaml
@@ -14,11 +14,19 @@
 
 bases:
 - ../../../config/
+resources:
+- rbac.yaml
 patchesJson6902:
 - target:
     group: apps
-    version: v1 # apiVersion
+    version: v1
     kind: Deployment
     name: tekton-results-watcher
     namespace: tekton-pipelines
-  path: watcher.yaml
\ No newline at end of file
+  path: watcher.yaml
+- target:
+    version: v1
+    kind: Service
+    name: tekton-results-api-service
+    namespace: tekton-pipelines
+  path: api-server.yaml
diff --git a/test/e2e/kustomize/rbac.yaml b/test/e2e/kustomize/rbac.yaml
new file mode 100644
index 00000000..47d26781
--- /dev/null
+++ b/test/e2e/kustomize/rbac.yaml
@@ -0,0 +1,52 @@
+# Copyright 2022 The Tekton Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: all-namespaces-read-access
+  namespace: default
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: single-namespace-read-access
+  namespace: default
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: all-namespaces-read-access
+subjects:
+  - kind: ServiceAccount
+    name: all-namespaces-read-access
+    namespace: default
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: tekton-results-readonly
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: single-namespace-read-access
+  namespace: default
+subjects:
+  - kind: ServiceAccount
+    name: single-namespace-read-access
+    namespace: default
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: tekton-results-readonly