fix(api,watcher): cascade delete Results on namespace deletion

AGENTS.md

@@ -89,6 +89,8 @@ Handles all data mutations and queries.
 **Watcher** (`cmd/watcher`, `pkg/watcher/`): Kubernetes controller that watches
 TaskRun, PipelineRun, and CustomRun resources. Creates or updates corresponding
 Records via the Results API. Annotates original CRDs with result identifiers.
+Also watches Namespace deletions and cascade-deletes associated Results/Records
+via a dedicated namespace reconciler.
 
 **Retention Policy Agent** (`cmd/retention-policy-agent`, `pkg/retention/`):
 Deletes old data from the database based on configured retention policies.
@@ -106,6 +108,7 @@ Logs are read from the external log store, not stored in the Results database.
 - **New API handler**: Follow the pattern in `pkg/api/server/<proto_version>/results.go`
 - **Watcher shared reconciler logic**: See `pkg/watcher/reconciler/dynamic/`
 - **Watcher resource-specific reconciler**: See `pkg/watcher/reconciler/pipelinerun/`, `taskrun/`, or `customrun/`
+- **Watcher namespace cleanup reconciler**: See `pkg/watcher/reconciler/namespace/`
 - **Proto definition changes**: Follow `proto/<proto_version>/results.proto`
 - **Database migrations**: See `tools/tkn-results-migrator/`
 - **Integration tests**: Follow examples in `test/e2e/`

cmd/watcher/main.go

@@ -32,6 +32,7 @@ import (
 	creds "github.com/tektoncd/results/pkg/watcher/grpc"
 	"github.com/tektoncd/results/pkg/watcher/reconciler"
 	"github.com/tektoncd/results/pkg/watcher/reconciler/customrun"
+	nsreconciler "github.com/tektoncd/results/pkg/watcher/reconciler/namespace"
 	"github.com/tektoncd/results/pkg/watcher/reconciler/pipelinerun"
 	"github.com/tektoncd/results/pkg/watcher/reconciler/taskrun"
 	v1alpha2pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto"
@@ -156,6 +157,8 @@ func main() {
 			return taskrun.NewControllerWithConfig(ctx, results, cfg, cmw)
 		}, func(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
 			return customrun.NewControllerWithConfig(ctx, results, cfg, cmw)
+		}, func(ctx context.Context, _ configmap.Watcher) *controller.Impl {
+			return nsreconciler.NewController(ctx, results)
 		},
 	}
 

config/base/100-watcher-serviceaccount.yaml

@@ -22,10 +22,10 @@ kind: ClusterRole
 metadata:
   name: watcher
 rules:
-  # Watcher needs to be able to create new and update existing results.
+  # Watcher needs to be able to create, update, list, and delete results.
   - apiGroups: ["results.tekton.dev"]
     resources: ["logs", "results", "records"]
-    verbs: ["create", "get", "update"]
+    verbs: ["create", "get", "list", "update", "delete"]
   # Needed to read results and update annotations with Result ID.
   - apiGroups: ["tekton.dev"]
     resources: ["pipelineruns", "taskruns", "customruns"]
@@ -43,6 +43,10 @@ rules:
   - apiGroups: [""]
     resources: ["events"]
     verbs: ["get", "list", "create", "update", "delete", "patch", "watch"]
+  # Required to watch namespace deletions for cascading result cleanup.
+  - apiGroups: [""]
+    resources: ["namespaces"]
+    verbs: ["get", "list", "watch"]
   - apiGroups: ["tekton.dev"]
     resources: ["pipelines"]
     verbs: ["get"]

docs/README.md

@@ -39,7 +39,8 @@ Tekton Results is composed of 3 main components:
 - A [queryable gRPC API server](api/) backed by persistent storage (see
   [proto/v1alpha2](../proto/v1alpha2) for the latest API spec).
 - A [controller to watch and report](watcher/) TaskRun, PipelineRun, and
-  CustomRun updates to the API server.
+  CustomRun updates to the API server. It also watches for Namespace
+  deletions and cascade-deletes associated Results and Records.
 - A [retention policy agent](retention-policy-agent/), an agent which deletes older data from DB.
 
 ### Life of a Result

docs/logging-support.md

@@ -59,6 +59,22 @@ These are the common configuration options for all third party logging APIs.
 - `LOGGING_PLUGIN_QUERY_PARAMS`: Sets the query params for Third Party Logging API, these can be direction/sort order.Specify them in this format: "foo=bar&direction=backward"
 - `LOGGING_PLUGIN_MULTIPART_REGEX`: Sets a Regex for matching parts of the same log. Some log backends (e.g S3) store objects immutably, once stored, you can't append. For long running TaskRun steps, it's not effective to keep such log in memory until the step completes. Instead one can store the log in multiple parts with a name suffix (e.g `-1743932245` seconds since the Epoch) and set a regex to match the parts of the same log (e.g `-\d{10}$`). (optional)
 
+## Error Handling
+
+When Loki or Splunk returns an HTTP error, the API server maps it to the
+appropriate gRPC status code instead of returning a generic Internal error:
+
+| HTTP Status | gRPC Code |
+|---|---|
+| 400 Bad Request | `InvalidArgument` |
+| 401 Unauthorized | `Unauthenticated` |
+| 403 Forbidden | `PermissionDenied` |
+| 404 Not Found | `NotFound` |
+| 429 Too Many Requests | `ResourceExhausted` |
+| Other / 5xx | `Internal` |
+
+The HTTP status code is included in the error message for debuggability.
+
 ## Loki-specific Configuration
 - `LOGGING_PLUGIN_JSON_MAP`: Define a map for the fields in Loki logs to be extracted like `{"timestamp": "@timestamp"}` to extract the timestamp as example.
 - `LOGGING_PLUGIN_LINE_FORMAT`: Define the format of the log line returned when using Loki. To use the Timestamp from the `LOGGING_PLUGIN_JSON_MAP` and the actual message, define `"{{.timestamp}}: {{.message}}"`. Only fields extracted by `LOGGING_PLUGIN_JSON_MAP` can be used here.

docs/watcher/README.md

@@ -22,6 +22,10 @@ The Watcher currently supports the following types:
 - `tekton.dev/v1 TaskRun`
 - `tekton.dev/v1 PipelineRun`
 
+The Watcher also watches core `Namespace` resources to cascade-delete
+associated Results when a namespace is deleted (see
+[Namespace Cleanup](#namespace-cleanup)).
+
 ## Result Grouping
 
 The Watcher uses Object data to automatically detect and group related Records
@@ -77,6 +81,25 @@ Watcher implements a finalizer to block deletion by an external pruner when obje
 When deletion request comes, it will block until completion time + `completed_run_grace_period` period is passed. A hard limit could be set as `store_deadline` (default 10m), after which the object will be removed from the cluster even without confirmation it's been stored in the DB.
 
 
+## Namespace Cleanup
+
+The Watcher watches for Kubernetes Namespace deletions and automatically
+cascade-deletes all Results associated with the deleted namespace via the
+Results API. Records are removed automatically by the database foreign key
+constraint (`ON DELETE CASCADE`).
+
+This ensures that no orphaned data remains in the database after a namespace
+is removed from the cluster. The cleanup is handled by a dedicated namespace
+reconciler (`pkg/watcher/reconciler/namespace/`) that paginates through all
+Results for the deleted namespace and issues individual `DeleteResult` calls.
+
+The watcher's ClusterRole requires the following additional permissions for
+this feature:
+
+- `namespaces` (get, list, watch) to receive namespace deletion events.
+- `results.tekton.dev` resources (list, delete) to query and remove Results
+  via the API.
+
 ## Disabling Incomplete Runs storage
 
 The `disable_storing_incomplete_runs` flag controls whether the Watcher should store PipelineRuns, TaskRuns, and CustomRuns that are still in progress (i.e., not yet completed, cancelled or failed).

pkg/api/server/v1alpha2/plugin/export_test.go

@@ -5,3 +5,4 @@ package plugin
 
 var MergeLogParts = mergeLogParts
 var GetLokiLogs = getLokiLogs
+var HTTPStatusToGRPCCode = httpStatusToGRPCCode

pkg/api/server/v1alpha2/plugin/plugin_logs.go

@@ -64,6 +64,23 @@ const (
 	splunkOutputFormat        = "?output_mode=json"
 )
 
+func httpStatusToGRPCCode(httpStatus int) codes.Code {
+	switch httpStatus {
+	case http.StatusBadRequest:
+		return codes.InvalidArgument
+	case http.StatusUnauthorized:
+		return codes.Unauthenticated
+	case http.StatusForbidden:
+		return codes.PermissionDenied
+	case http.StatusNotFound:
+		return codes.NotFound
+	case http.StatusTooManyRequests:
+		return codes.ResourceExhausted
+	default:
+		return codes.Internal
+	}
+}
+
 var (
 	openBucket = func(ctx context.Context, urlString string) (*blob.Bucket, error) {
 		bucket, err := blob.OpenBucket(ctx, urlString)
@@ -109,6 +126,7 @@ func (s *LogServer) GetLog(req *pb3.GetLogRequest, srv pb3.Logs_GetLogServer) er
 	err = s.getLog(s, writer, parent, rec)
 	if err != nil {
 		s.logger.Error(err)
+		return err
 	}
 
 	_, err = writer.Flush()
@@ -267,6 +285,11 @@ func getLokiLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record)
 		s.logger.Debugf("loki request url:%s", URL.String())
 		return status.Error(codes.Internal, "Error streaming log")
 	}
+	defer func() {
+		if err := resp.Body.Close(); err != nil {
+			s.logger.Errorf("error closing response body: %s", err)
+		}
+	}()
 
 	if resp.StatusCode != http.StatusOK {
 		s.logger.Errorf("Loki API request failed with HTTP status code: %d", resp.StatusCode)
@@ -278,7 +301,7 @@ func getLokiLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record)
 		if err == nil {
 			s.logger.Debugf("Response Dump***:\n %q\n", dump)
 		}
-		return status.Error(codes.Internal, "Error fetching log data")
+		return status.Errorf(httpStatusToGRPCCode(resp.StatusCode), "Error fetching log data (HTTP %d)", resp.StatusCode)
 	}
 
 	data, err := io.ReadAll(resp.Body)
@@ -581,7 +604,7 @@ func getSplunkLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record
 
 	if resp.StatusCode != http.StatusCreated {
 		s.logger.Errorf("Splunk Job Creation API request failed with HTTP status code: %d", resp.StatusCode)
-		return status.Error(codes.Internal, "Error fetching log data - search job creation failed")
+		return status.Errorf(httpStatusToGRPCCode(resp.StatusCode), "Error fetching log data - search job creation failed (HTTP %d)", resp.StatusCode)
 	}
 
 	data, err := io.ReadAll(resp.Body)
@@ -630,8 +653,8 @@ func getSplunkLogs(s *LogServer, writer io.Writer, parent string, rec *db.Record
 	}()
 
 	if lresp.StatusCode != http.StatusOK {
-		s.logger.Errorf("Splunk Fetch Log API request failed with HTTP status code: %d", resp.StatusCode)
-		return status.Error(codes.Internal, "Error fetching log data - fetch log api failed")
+		s.logger.Errorf("Fetch Log API request failed with HTTP status code: %d", lresp.StatusCode)
+		return status.Errorf(httpStatusToGRPCCode(lresp.StatusCode), "Error fetching log data - fetch log api failed (HTTP %d)", lresp.StatusCode)
 	}
 
 	data, err = io.ReadAll(lresp.Body)

pkg/api/server/v1alpha2/plugin/plugin_logs_test.go

@@ -28,6 +28,8 @@ import (
 	pb3 "github.com/tektoncd/results/proto/v1alpha3/results_go_proto"
 	"google.golang.org/genproto/googleapis/api/httpbody"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -625,3 +627,117 @@ func TestGetLokiLogs_FailsWhenLineFormatUsesUndefinedField(t *testing.T) {
 		t.Fatalf("did not expect undefined field to be present in json parsing map, got: %s", gotQuery)
 	}
 }
+
+func TestHttpStatusToGRPCCode(t *testing.T) {
+	tests := []struct {
+		httpStatus int
+		wantCode   codes.Code
+	}{
+		{http.StatusBadRequest, codes.InvalidArgument},
+		{http.StatusUnauthorized, codes.Unauthenticated},
+		{http.StatusForbidden, codes.PermissionDenied},
+		{http.StatusNotFound, codes.NotFound},
+		{http.StatusTooManyRequests, codes.ResourceExhausted},
+		{http.StatusInternalServerError, codes.Internal},
+		{http.StatusBadGateway, codes.Internal},
+		{http.StatusServiceUnavailable, codes.Internal},
+	}
+	for _, tt := range tests {
+		t.Run(strconv.Itoa(tt.httpStatus), func(t *testing.T) {
+			got := plugin.HTTPStatusToGRPCCode(tt.httpStatus)
+			if got != tt.wantCode {
+				t.Errorf("httpStatusToGRPCCode(%d) = %v, want %v", tt.httpStatus, got, tt.wantCode)
+			}
+		})
+	}
+}
+
+func TestGetLog_LokiForbidden_ReturnsPermissionDenied(t *testing.T) {
+	mockLoki := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+		json.NewEncoder(w).Encode(map[string]interface{}{
+			"error":     "You don't have permission to access this tenant",
+			"errorType": "observatorium-api",
+			"status":    "error",
+		})
+	}))
+	defer mockLoki.Close()
+
+	tokenDir := t.TempDir()
+	tokenPath := filepath.Join(tokenDir, "token")
+	if err := os.WriteFile(tokenPath, []byte("dummytoken"), 0600); err != nil {
+		t.Fatalf("Failed to create token file: %v", err)
+	}
+
+	srv, err := server.New(&config.Config{
+		LOGS_API:                                true,
+		LOGS_TYPE:                               "Loki",
+		DB_ENABLE_AUTO_MIGRATION:                true,
+		LOGGING_PLUGIN_TOKEN_PATH:               tokenPath,
+		LOGGING_PLUGIN_PROXY_PATH:               "/app",
+		LOGGING_PLUGIN_API_URL:                  mockLoki.URL,
+		LOGGING_PLUGIN_TLS_VERIFICATION_DISABLE: true,
+		LOGGING_PLUGIN_STATIC_LABELS:            "namespace=\"foo\"",
+		LOGGING_PLUGIN_NAMESPACE_KEY:            "namespace",
+		LOGGING_PLUGIN_CONTAINER_KEY:            "kubernetes.container_name",
+		LOGGING_PLUGIN_QUERY_LIMIT:              1500,
+		LOGGING_PLUGIN_QUERY_PARAMS:             "direction=forward",
+	}, logger.Get("info"), test.NewDB(t))
+	if err != nil {
+		t.Fatalf("failed to create server: %v", err)
+	}
+
+	ctx := context.Background()
+	mockServer := &mockGetLogServer{ctx: ctx}
+
+	res, err := srv.CreateResult(ctx, &pb.CreateResultRequest{
+		Parent: "deleted-namespace",
+		Result: &pb.Result{
+			Name: "deleted-namespace/results/bar",
+		},
+	})
+	if err != nil {
+		t.Fatalf("CreateResult: %v", err)
+	}
+
+	_, err = srv.CreateRecord(ctx, &pb.CreateRecordRequest{
+		Parent: res.GetName(),
+		Record: &pb.Record{
+			Name: record.FormatName(res.GetName(), "baz"),
+			Data: &pb.Any{
+				Type: "tekton.dev/v1.PipelineRun",
+				Value: jsonutil.AnyBytes(t, pipelinev1.PipelineRun{
+					Status: pipelinev1.PipelineRunStatus{
+						PipelineRunStatusFields: pipelinev1.PipelineRunStatusFields{
+							StartTime:      &metav1.Time{Time: time.Now()},
+							CompletionTime: &metav1.Time{Time: time.Now()},
+						},
+					},
+				}),
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("CreateRecord: %v", err)
+	}
+
+	req := &pb3.GetLogRequest{
+		Name: log.FormatName(res.GetName(), "baz"),
+	}
+
+	err = srv.LogPluginServer.GetLog(req, mockServer)
+	if err == nil {
+		t.Fatal("expected GetLog to return error for 403 Forbidden")
+	}
+
+	st, ok := status.FromError(err)
+	if !ok {
+		t.Fatalf("expected gRPC status error, got: %v", err)
+	}
+	if st.Code() != codes.PermissionDenied {
+		t.Errorf("expected PermissionDenied, got %v", st.Code())
+	}
+	if !strings.Contains(st.Message(), "403") {
+		t.Errorf("expected error message to contain HTTP status code 403, got: %s", st.Message())
+	}
+}

pkg/watcher/reconciler/namespace/controller.go

@@ -0,0 +1,56 @@
+// Copyright 2026 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 namespace provides a controller that cleans up Results API data when namespaces are deleted.
+package namespace
+
+import (
+	"context"
+
+	leaderelection "github.com/tektoncd/results/pkg/watcher/reconciler/leaderelection"
+	pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto"
+	factory "knative.dev/pkg/client/injection/kube/informers/factory"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+)
+
+// NewController creates a controller that watches for namespace deletions
+// and cleans up associated Results API data.
+func NewController(ctx context.Context, resultsClient pb.ResultsClient) *controller.Impl {
+	logger := logging.FromContext(ctx)
+
+	informerFactory := factory.Get(ctx)
+	nsInformer := informerFactory.Core().V1().Namespaces()
+
+	r := &Reconciler{
+		LeaderAwareFuncs: leaderelection.NewLeaderAwareFuncs(nsInformer.Lister().List),
+		resultsClient:    resultsClient,
+		namespaceLister:  nsInformer.Lister(),
+	}
+
+	impl := controller.NewContext(ctx, r, controller.ControllerOptions{
+		WorkQueueName: "NamespaceCleanup",
+		Logger:        logger.Desugar().Sugar(),
+	})
+
+	_, err := nsInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+	if err != nil {
+		logger.Panicf("Couldn't register Namespace informer event handler: %w", err)
+	}
+
+	informerFactory.Start(ctx.Done())
+	informerFactory.WaitForCacheSync(ctx.Done())
+
+	return impl
+}