diff --git a/AGENTS.md b/AGENTS.md index d6a2102fff..82df25c3d5 100644 --- a/AGENTS.md +++ b/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//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//results.proto` - **Database migrations**: See `tools/tkn-results-migrator/` - **Integration tests**: Follow examples in `test/e2e/` diff --git a/cmd/watcher/main.go b/cmd/watcher/main.go index 04e8274834..6e4ba26aee 100644 --- a/cmd/watcher/main.go +++ b/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) }, } diff --git a/config/base/100-watcher-serviceaccount.yaml b/config/base/100-watcher-serviceaccount.yaml index 1b5db6cf2e..aa500d6b37 100644 --- a/config/base/100-watcher-serviceaccount.yaml +++ b/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"] diff --git a/docs/README.md b/docs/README.md index 1adce12b45..d798dd3ff5 100644 --- a/docs/README.md +++ b/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 diff --git a/docs/logging-support.md b/docs/logging-support.md index 63790c3252..99e705103f 100644 --- a/docs/logging-support.md +++ b/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. \ No newline at end of file diff --git a/docs/watcher/README.md b/docs/watcher/README.md index 7e4b31b871..af93138c4d 100644 --- a/docs/watcher/README.md +++ b/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). diff --git a/pkg/api/server/v1alpha2/plugin/export_test.go b/pkg/api/server/v1alpha2/plugin/export_test.go index 5d60db37b6..d4cdd88b38 100644 --- a/pkg/api/server/v1alpha2/plugin/export_test.go +++ b/pkg/api/server/v1alpha2/plugin/export_test.go @@ -5,3 +5,4 @@ package plugin var MergeLogParts = mergeLogParts var GetLokiLogs = getLokiLogs +var HTTPStatusToGRPCCode = httpStatusToGRPCCode diff --git a/pkg/api/server/v1alpha2/plugin/plugin_logs.go b/pkg/api/server/v1alpha2/plugin/plugin_logs.go index 6ce32d0d60..117b52a861 100644 --- a/pkg/api/server/v1alpha2/plugin/plugin_logs.go +++ b/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) diff --git a/pkg/api/server/v1alpha2/plugin/plugin_logs_test.go b/pkg/api/server/v1alpha2/plugin/plugin_logs_test.go index 921ddf35c4..16208f3c79 100644 --- a/pkg/api/server/v1alpha2/plugin/plugin_logs_test.go +++ b/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()) + } +} diff --git a/pkg/watcher/reconciler/namespace/controller.go b/pkg/watcher/reconciler/namespace/controller.go new file mode 100644 index 0000000000..9f43aad4b4 --- /dev/null +++ b/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 +} diff --git a/pkg/watcher/reconciler/namespace/reconciler.go b/pkg/watcher/reconciler/namespace/reconciler.go new file mode 100644 index 0000000000..3ba8992b86 --- /dev/null +++ b/pkg/watcher/reconciler/namespace/reconciler.go @@ -0,0 +1,92 @@ +// 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 + +import ( + "context" + "fmt" + + pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto" + apierrors "k8s.io/apimachinery/pkg/api/errors" + corev1listers "k8s.io/client-go/listers/core/v1" + "knative.dev/pkg/logging" + "knative.dev/pkg/reconciler" +) + +const maxPageSize = 10000 + +// Reconciler cleans up Results API data when a namespace is deleted. +type Reconciler struct { + reconciler.LeaderAwareFuncs + resultsClient pb.ResultsClient + namespaceLister corev1listers.NamespaceLister +} + +// Reconcile handles namespace events, deleting associated Results when a namespace is deleted. +func (r *Reconciler) Reconcile(ctx context.Context, key string) error { + logger := logging.FromContext(ctx) + namespaceName := key + + ns, err := r.namespaceLister.Get(namespaceName) + if err != nil { + if !apierrors.IsNotFound(err) { + return fmt.Errorf("error getting namespace %s: %w", namespaceName, err) + } + } + + if ns != nil && ns.DeletionTimestamp == nil { + return nil + } + + logger.Infof("Namespace %s deleted, cleaning up Results", namespaceName) + return r.deleteResultsForNamespace(ctx, namespaceName) +} + +func (r *Reconciler) deleteResultsForNamespace(ctx context.Context, namespace string) error { + logger := logging.FromContext(ctx) + + var totalDeleted int + pageToken := "" + + for { + resp, err := r.resultsClient.ListResults(ctx, &pb.ListResultsRequest{ + Parent: namespace, + PageSize: maxPageSize, + PageToken: pageToken, + }) + if err != nil { + return fmt.Errorf("error listing results for namespace %s: %w", namespace, err) + } + + for _, result := range resp.GetResults() { + _, err := r.resultsClient.DeleteResult(ctx, &pb.DeleteResultRequest{ + Name: result.GetName(), + }) + if err != nil { + logger.Warnf("Failed to delete result %s: %v", result.GetName(), err) + continue + } + totalDeleted++ + } + + pageToken = resp.GetNextPageToken() + if pageToken == "" { + break + } + } + + logger.Infof("Cleaned up %d results for deleted namespace %s", totalDeleted, namespace) + return nil +} diff --git a/pkg/watcher/reconciler/namespace/reconciler_test.go b/pkg/watcher/reconciler/namespace/reconciler_test.go new file mode 100644 index 0000000000..5fd0ec7f14 --- /dev/null +++ b/pkg/watcher/reconciler/namespace/reconciler_test.go @@ -0,0 +1,221 @@ +// 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 + +import ( + "context" + "fmt" + "testing" + "time" + + pb "github.com/tektoncd/results/proto/v1alpha2/results_go_proto" + "google.golang.org/grpc" + "google.golang.org/protobuf/types/known/emptypb" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + corev1listers "k8s.io/client-go/listers/core/v1" +) + +type mockResultsClient struct { + pb.ResultsClient + results map[string]*pb.Result + deletedNames []string + listErr error + deleteErr error + deleteErrName string +} + +func (m *mockResultsClient) ListResults(_ context.Context, req *pb.ListResultsRequest, _ ...grpc.CallOption) (*pb.ListResultsResponse, error) { + if m.listErr != nil { + return nil, m.listErr + } + var results []*pb.Result + for _, r := range m.results { + if req.GetParent() == "" || getParent(r.GetName()) == req.GetParent() { + results = append(results, r) + } + } + return &pb.ListResultsResponse{Results: results}, nil +} + +func (m *mockResultsClient) DeleteResult(_ context.Context, req *pb.DeleteResultRequest, _ ...grpc.CallOption) (*emptypb.Empty, error) { + if m.deleteErr != nil && req.GetName() == m.deleteErrName { + return nil, m.deleteErr + } + m.deletedNames = append(m.deletedNames, req.GetName()) + delete(m.results, req.GetName()) + return &emptypb.Empty{}, nil +} + +func getParent(name string) string { + for i, c := range name { + if c == '/' { + return name[:i] + } + } + return name +} + +type mockNamespaceLister struct { + namespaces map[string]*corev1.Namespace +} + +func (m *mockNamespaceLister) List(_ labels.Selector) ([]*corev1.Namespace, error) { + var result []*corev1.Namespace + for _, ns := range m.namespaces { + result = append(result, ns) + } + return result, nil +} + +func (m *mockNamespaceLister) Get(name string) (*corev1.Namespace, error) { + ns, ok := m.namespaces[name] + if !ok { + return nil, fmt.Errorf("namespace %q not found: %w", name, ¬FoundError{}) + } + return ns, nil +} + +type notFoundError struct{} + +func (e *notFoundError) Error() string { return "not found" } +func (e *notFoundError) Status() metav1.Status { + return metav1.Status{Reason: metav1.StatusReasonNotFound} +} + +var _ corev1listers.NamespaceLister = &mockNamespaceLister{} + +func TestReconcile_ActiveNamespace_NoOp(t *testing.T) { + client := &mockResultsClient{ + results: map[string]*pb.Result{ + "active-ns/results/abc": {Name: "active-ns/results/abc"}, + }, + } + lister := &mockNamespaceLister{ + namespaces: map[string]*corev1.Namespace{ + "active-ns": {ObjectMeta: metav1.ObjectMeta{Name: "active-ns"}}, + }, + } + r := &Reconciler{resultsClient: client, namespaceLister: lister} + + err := r.Reconcile(context.Background(), "active-ns") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(client.deletedNames) != 0 { + t.Errorf("expected no deletions, got %v", client.deletedNames) + } +} + +func TestReconcile_DeletedNamespace_CleansUpResults(t *testing.T) { + client := &mockResultsClient{ + results: map[string]*pb.Result{ + "deleted-ns/results/aaa": {Name: "deleted-ns/results/aaa"}, + "deleted-ns/results/bbb": {Name: "deleted-ns/results/bbb"}, + "other-ns/results/ccc": {Name: "other-ns/results/ccc"}, + }, + } + lister := &mockNamespaceLister{namespaces: map[string]*corev1.Namespace{}} + r := &Reconciler{resultsClient: client, namespaceLister: lister} + + err := r.Reconcile(context.Background(), "deleted-ns") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(client.deletedNames) != 2 { + t.Errorf("expected 2 deletions, got %d: %v", len(client.deletedNames), client.deletedNames) + } + if _, exists := client.results["other-ns/results/ccc"]; !exists { + t.Error("result from other namespace should not be deleted") + } +} + +func TestReconcile_TerminatingNamespace_CleansUpResults(t *testing.T) { + now := metav1.NewTime(time.Now()) + client := &mockResultsClient{ + results: map[string]*pb.Result{ + "terminating-ns/results/aaa": {Name: "terminating-ns/results/aaa"}, + }, + } + lister := &mockNamespaceLister{ + namespaces: map[string]*corev1.Namespace{ + "terminating-ns": { + ObjectMeta: metav1.ObjectMeta{ + Name: "terminating-ns", + DeletionTimestamp: &now, + }, + }, + }, + } + r := &Reconciler{resultsClient: client, namespaceLister: lister} + + err := r.Reconcile(context.Background(), "terminating-ns") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(client.deletedNames) != 1 { + t.Errorf("expected 1 deletion, got %d", len(client.deletedNames)) + } +} + +func TestReconcile_ListResultsError_ReturnsError(t *testing.T) { + client := &mockResultsClient{ + results: map[string]*pb.Result{}, + listErr: fmt.Errorf("connection refused"), + } + lister := &mockNamespaceLister{namespaces: map[string]*corev1.Namespace{}} + r := &Reconciler{resultsClient: client, namespaceLister: lister} + + err := r.Reconcile(context.Background(), "deleted-ns") + if err == nil { + t.Fatal("expected error, got nil") + } +} + +func TestReconcile_DeleteResultPartialFailure_Continues(t *testing.T) { + client := &mockResultsClient{ + results: map[string]*pb.Result{ + "deleted-ns/results/aaa": {Name: "deleted-ns/results/aaa"}, + "deleted-ns/results/bbb": {Name: "deleted-ns/results/bbb"}, + }, + deleteErr: fmt.Errorf("permission denied"), + deleteErrName: "deleted-ns/results/aaa", + } + lister := &mockNamespaceLister{namespaces: map[string]*corev1.Namespace{}} + r := &Reconciler{resultsClient: client, namespaceLister: lister} + + err := r.Reconcile(context.Background(), "deleted-ns") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(client.deletedNames) != 1 { + t.Errorf("expected 1 successful deletion, got %d", len(client.deletedNames)) + } +} + +func TestReconcile_EmptyResults_CompletesSuccessfully(t *testing.T) { + client := &mockResultsClient{results: map[string]*pb.Result{}} + lister := &mockNamespaceLister{namespaces: map[string]*corev1.Namespace{}} + r := &Reconciler{resultsClient: client, namespaceLister: lister} + + err := r.Reconcile(context.Background(), "deleted-ns") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(client.deletedNames) != 0 { + t.Errorf("expected no deletions, got %v", client.deletedNames) + } +}