Add custom StatusReader for Config Connector resources (#2626)

Author: mortent

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/xlab/treeprint v0.0.0-20181112141820-a009c3971eca
 	golang.org/x/mod v0.5.1
 	gotest.tools v2.2.0+incompatible
-	k8s.io/api v0.22.3 // indirect
+	k8s.io/api v0.22.3
 	k8s.io/apiextensions-apiserver v0.22.2
 	k8s.io/apimachinery v0.22.3
 	k8s.io/cli-runtime v0.22.2
@@ -23,6 +23,7 @@ require (
 	k8s.io/kube-openapi v0.0.0-20211109043139-026bd182f079 // indirect
 	k8s.io/kubectl v0.22.2
 	sigs.k8s.io/cli-utils v0.27.0
+	sigs.k8s.io/controller-runtime v0.10.1
 	sigs.k8s.io/kustomize/api v0.8.11
 	sigs.k8s.io/kustomize/kyaml v0.13.1-0.20211203194734-cd2c6a1ad117
 )

internal/cmdapply/cmdapply.go

@@ -25,6 +25,7 @@ import (
 	"github.com/GoogleContainerTools/kpt/internal/util/argutil"
 	"github.com/GoogleContainerTools/kpt/internal/util/strings"
 	"github.com/GoogleContainerTools/kpt/pkg/live"
+	"github.com/GoogleContainerTools/kpt/pkg/status"
 	"github.com/spf13/cobra"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -214,10 +215,20 @@ func runApply(r *Runner, invInfo inventory.InventoryInfo, objs []*unstructured.U
 	if err != nil {
 		return err
 	}
+
+	statusPoller, err := status.NewStatusPoller(r.factory)
+	if err != nil {
+		return err
+	}
+
 	applier, err := apply.NewApplier(r.factory, invClient)
 	if err != nil {
 		return err
 	}
+	// TODO(mortent): See if we can improve this. Having to change the Applier after it has been
+	// created feels a bit awkward.
+	applier.StatusPoller = statusPoller
+
 	ch := applier.Run(r.ctx, invInfo, objs, apply.Options{
 		ServerSideOptions:      r.serverSideOptions,
 		PollInterval:           r.period,

internal/cmddestroy/cmddestroy.go

@@ -23,6 +23,7 @@ import (
 	"github.com/GoogleContainerTools/kpt/internal/util/argutil"
 	"github.com/GoogleContainerTools/kpt/internal/util/strings"
 	"github.com/GoogleContainerTools/kpt/pkg/live"
+	"github.com/GoogleContainerTools/kpt/pkg/status"
 	"github.com/spf13/cobra"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/kubectl/pkg/cmd/util"
@@ -160,10 +161,18 @@ func runDestroy(r *Runner, inv inventory.InventoryInfo, dryRunStrategy common.Dr
 	if err != nil {
 		return err
 	}
+
+	statusPoller, err := status.NewStatusPoller(r.factory)
+	if err != nil {
+		return err
+	}
+
 	destroyer, err := apply.NewDestroyer(r.factory, invClient)
 	if err != nil {
 		return err
 	}
+	destroyer.StatusPoller = statusPoller
+
 	options := apply.DestroyerOptions{
 		InventoryPolicy:  r.inventoryPolicy,
 		DryRunStrategy:   dryRunStrategy,

pkg/live/inventoryrg.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/GoogleContainerTools/kpt/pkg/status"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -31,8 +32,6 @@ import (
 	"sigs.k8s.io/cli-utils/pkg/apply/taskrunner"
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
-	"sigs.k8s.io/cli-utils/pkg/kstatus/polling"
-	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/engine"
 	"sigs.k8s.io/cli-utils/pkg/object"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
@@ -280,7 +279,7 @@ func InstallResourceGroupCRD(factory cmdutil.Factory) error {
 		for _, t := range tasks {
 			taskQueue <- t
 		}
-		statusPoller, err := polling.NewStatusPollerFromFactory(factory, []engine.StatusReader{})
+		statusPoller, err := status.NewStatusPoller(factory)
 		if err != nil {
 			handleError(eventChannel, err)
 			return

pkg/status/configconnector.go

@@ -0,0 +1,160 @@
+// Copyright 2021 Google LLC
+//
+// 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 status
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/engine"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/event"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
+	"sigs.k8s.io/cli-utils/pkg/object"
+)
+
+// ConfigConnectorStatusReader can compute reconcile status for Config Connector
+// resources. It leverages information in the `Reason` field of the `Ready` condition.
+// TODO(mortent): Make more of the convencience functions and types from cli-utils
+// exported so we can simplify this.
+type ConfigConnectorStatusReader struct {
+	Mapper meta.RESTMapper
+}
+
+// Supports returns true for all Config Connector resources.
+func (c *ConfigConnectorStatusReader) Supports(gk schema.GroupKind) bool {
+	return strings.HasSuffix(gk.Group, "cnrm.cloud.google.com")
+}
+
+func (c *ConfigConnectorStatusReader) ReadStatus(ctx context.Context, reader engine.ClusterReader, id object.ObjMetadata) *event.ResourceStatus {
+	gvk, err := toGVK(id.GroupKind, c.Mapper)
+	if err != nil {
+		return newUnknownResourceStatus(id, nil, err)
+	}
+
+	key := types.NamespacedName{
+		Name:      id.Name,
+		Namespace: id.Namespace,
+	}
+
+	var u unstructured.Unstructured
+	u.SetGroupVersionKind(gvk)
+	err = reader.Get(ctx, key, &u)
+	if err != nil {
+		return newUnknownResourceStatus(id, nil, err)
+	}
+
+	return c.ReadStatusForObject(ctx, reader, &u)
+}
+
+func (c *ConfigConnectorStatusReader) ReadStatusForObject(_ context.Context, _ engine.ClusterReader, u *unstructured.Unstructured) *event.ResourceStatus {
+	id := object.UnstructuredToObjMetadata(u)
+
+	// First check if the resource is in the process of being deleted.
+	deletionTimestamp, found, err := unstructured.NestedString(u.Object, "metadata", "deletionTimestamp")
+	if err != nil {
+		return newUnknownResourceStatus(id, u, err)
+	}
+	if found && deletionTimestamp != "" {
+		return newResourceStatus(id, status.TerminatingStatus, u, "Resource scheduled for deletion")
+	}
+
+	// ensure that the meta generation is observed
+	generation, found, err := unstructured.NestedInt64(u.Object, "metadata", "generation")
+	if err != nil {
+		e := fmt.Errorf("looking up metadata.generation from resource: %w", err)
+		return newUnknownResourceStatus(id, u, e)
+	}
+	if !found {
+		e := fmt.Errorf("metadata.generation not found")
+		return newUnknownResourceStatus(id, u, e)
+	}
+
+	observedGeneration, found, err := unstructured.NestedInt64(u.Object, "status", "observedGeneration")
+	if err != nil {
+		e := fmt.Errorf("looking up status.observedGeneration from resource: %w", err)
+		return newUnknownResourceStatus(id, u, e)
+	}
+	if !found {
+		// We know that Config Connector resources uses the ObservedGeneration pattern, so consider it
+		// an error if it is not found.
+		e := fmt.Errorf("status.ObservedGeneration not found")
+		return newUnknownResourceStatus(id, u, e)
+	}
+	if generation != observedGeneration {
+		msg := fmt.Sprintf("%s generation is %d, but latest observed generation is %d", u.GetKind(), generation, observedGeneration)
+		return newResourceStatus(id, status.InProgressStatus, u, msg)
+	}
+
+	obj, err := status.GetObjectWithConditions(u.Object)
+	if err != nil {
+		return newUnknownResourceStatus(id, u, err)
+	}
+
+	var readyCond status.BasicCondition
+	foundCond := false
+	for i := range obj.Status.Conditions {
+		if obj.Status.Conditions[i].Type == "Ready" {
+			readyCond = obj.Status.Conditions[i]
+			foundCond = true
+		}
+	}
+
+	if !foundCond {
+		return newResourceStatus(id, status.InProgressStatus, u, "Ready condition not set")
+	}
+
+	if readyCond.Status == v1.ConditionTrue {
+		return newResourceStatus(id, status.CurrentStatus, u, "Resource is Current")
+	}
+
+	switch readyCond.Reason {
+	case "ManagementConflict", "UpdateFailed", "DeleteFailed", "DependencyInvalid":
+		return newResourceStatus(id, status.FailedStatus, u, readyCond.Message)
+	}
+
+	return newResourceStatus(id, status.InProgressStatus, u, readyCond.Message)
+}
+
+func toGVK(gk schema.GroupKind, mapper meta.RESTMapper) (schema.GroupVersionKind, error) {
+	mapping, err := mapper.RESTMapping(gk)
+	if err != nil {
+		return schema.GroupVersionKind{}, err
+	}
+	return mapping.GroupVersionKind, nil
+}
+
+func newResourceStatus(id object.ObjMetadata, s status.Status, u *unstructured.Unstructured, msg string) *event.ResourceStatus {
+	return &event.ResourceStatus{
+		Identifier: id,
+		Status:     s,
+		Resource:   u,
+		Message:    msg,
+	}
+}
+
+func newUnknownResourceStatus(id object.ObjMetadata, u *unstructured.Unstructured, err error) *event.ResourceStatus {
+	return &event.ResourceStatus{
+		Identifier: id,
+		Status:     status.UnknownStatus,
+		Error:      err,
+		Resource:   u,
+	}
+}

pkg/status/configconnector_test.go

@@ -0,0 +1,138 @@
+package status
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/testutil"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
+	"sigs.k8s.io/cli-utils/pkg/object"
+	fakemapper "sigs.k8s.io/cli-utils/pkg/testutil"
+)
+
+func TestSupports(t *testing.T) {
+	testCases := map[string]struct {
+		gk       schema.GroupKind
+		supports bool
+	}{
+		"matches config connector group": {
+			gk: schema.GroupKind{
+				Group: "sql.cnrm.cloud.google.com",
+				Kind:  "SQLDatabase",
+			},
+			supports: true,
+		},
+		"doesn't match other resources": {
+			gk: schema.GroupKind{
+				Group: "apps",
+				Kind:  "StatefulSet",
+			},
+			supports: false,
+		},
+	}
+
+	for tn, tc := range testCases {
+		t.Run(tn, func(t *testing.T) {
+			fakeMapper := fakemapper.NewFakeRESTMapper()
+
+			statusReader := &ConfigConnectorStatusReader{
+				Mapper: fakeMapper,
+			}
+
+			supports := statusReader.Supports(tc.gk)
+
+			assert.Equal(t, tc.supports, supports)
+		})
+	}
+}
+
+func TestReadStatus(t *testing.T) {
+	testCases := map[string]struct {
+		resource       string
+		gvk            schema.GroupVersionKind
+		expectedStatus status.Status
+	}{
+		"Resource with deletionTimestap is Terminating": {
+			resource: `
+apiVersion: serviceusage.cnrm.cloud.google.com/v1beta1
+kind: Service
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+  deletionTimestamp: "2020-01-09T20:56:25Z"
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "serviceusage.cnrm.cloud.google.com",
+				Version: "v1beta1",
+				Kind:    "Service",
+			},
+			expectedStatus: status.TerminatingStatus,
+		},
+		"Resource where observedGeneration doesn't match generation is InProgress": {
+			resource: `
+apiVersion: serviceusage.cnrm.cloud.google.com/v1beta1
+kind: Service
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+status:
+  observedGeneration: 41
+  conditions:
+  - type: Ready
+    status: "False"
+    reason: UpdateFailed
+    message: "Resource couldn't be updated"
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "serviceusage.cnrm.cloud.google.com",
+				Version: "v1beta1",
+				Kind:    "Service",
+			},
+			expectedStatus: status.InProgressStatus,
+		},
+		"Resource with reason UpdateFailed is Failed": {
+			resource: `
+apiVersion: serviceusage.cnrm.cloud.google.com/v1beta1
+kind: Service
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+status:
+  observedGeneration: 42
+  conditions:
+  - type: Ready
+    status: "False"
+    reason: UpdateFailed
+    message: "Resource couldn't be updated"
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "serviceusage.cnrm.cloud.google.com",
+				Version: "v1beta1",
+				Kind:    "Service",
+			},
+			expectedStatus: status.FailedStatus,
+		},
+	}
+
+	for tn, tc := range testCases {
+		t.Run(tn, func(t *testing.T) {
+			obj := testutil.YamlToUnstructured(t, tc.resource)
+
+			fakeClusterReader := &fakeClusterReader{
+				getResource: obj,
+			}
+			fakeMapper := fakemapper.NewFakeRESTMapper(tc.gvk)
+			statusReader := &ConfigConnectorStatusReader{
+				Mapper: fakeMapper,
+			}
+
+			res := statusReader.ReadStatus(context.Background(), fakeClusterReader, object.UnstructuredToObjMetadata(obj))
+			assert.Equal(t, tc.expectedStatus, res.Status)
+		})
+	}
+}