Repository: GoogleContainerTools/kpt
Pull request #3477 (2022-08-20 00:32:05)

Add status reader for the rollout object
This change adds a customized status reader for the Rollout object.
The motivation is to integrate Config Sync with Argo Rollouts and track the status of the Rollout object with kpt, so the status reader is added in kpt instead of Config Sync.


Patch:
diff --git a/pkg/status/poller.go b/pkg/status/poller.go
index 529d805b..2d5ae290 100644
--- a/pkg/status/poller.go
+++ b/pkg/status/poller.go
@@ -36,6 +36,9 @@ func NewStatusPoller(f util.Factory) (*polling.StatusPoller, error) {
 			&ConfigConnectorStatusReader{
 				Mapper: mapper,
 			},
+			&RolloutStatusReader{
+				Mapper: mapper,
+			},
 		},
 	})
 }
@@ -55,7 +58,10 @@ func NewStatusWatcher(f util.Factory) (watcher.StatusWatcher, error) {
 		DynamicClient: dynamicClient,
 		Mapper:        mapper,
 		ResyncPeriod:  1 * time.Hour,
-		StatusReader:  statusreaders.NewStatusReader(mapper, NewConfigConnectorStatusReader(mapper)),
+		StatusReader: statusreaders.NewStatusReader(
+			mapper,
+			NewConfigConnectorStatusReader(mapper),
+			NewRolloutStatusReader(mapper)),
 		ClusterReader: &clusterreader.DynamicClusterReader{
 			DynamicClient: dynamicClient,
 			Mapper:        mapper,
diff --git a/pkg/status/rollout.go b/pkg/status/rollout.go
new file mode 100644
index 00000000..a791709c
--- /dev/null
+++ b/pkg/status/rollout.go
@@ -0,0 +1,216 @@
+package status
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"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"
+	"sigs.k8s.io/yaml"
+)
+
+const (
+	ArgoGroup   = "argoproj.io"
+	Rollout     = "Rollout"
+	Degraded    = "Degraded"
+	Failed      = "Failed"
+	Healthy     = "Healthy"
+	Paused      = "Paused"
+	Progressing = "Progressing"
+)
+
+type RolloutStatusReader struct {
+	Mapper meta.RESTMapper
+}
+
+func NewRolloutStatusReader(mapper meta.RESTMapper) engine.StatusReader {
+	return &RolloutStatusReader{
+		Mapper: mapper,
+	}
+}
+
+var _ engine.StatusReader = &RolloutStatusReader{}
+
+// Supports returns true for all rollout resources.
+func (r *RolloutStatusReader) Supports(gk schema.GroupKind) bool {
+	return gk.Group == ArgoGroup && gk.Kind == Rollout
+}
+
+func (r *RolloutStatusReader) Compute(u *unstructured.Unstructured) (*status.Result, error) {
+	result := status.Result{
+		Status:     status.UnknownStatus,
+		Message:    status.GetStringField(u.Object, ".status.message", ""),
+		Conditions: make([]status.Condition, 0),
+	}
+	// ensure that the meta generation is observed
+	generation, found, err := unstructured.NestedInt64(u.Object, "metadata", "generation")
+	if err != nil {
+		return &result, fmt.Errorf("looking up metadata.generation from resource: %w", err)
+	}
+	if !found {
+		return &result, fmt.Errorf("metadata.generation not found")
+	}
+
+	// Argo Rollouts defines the observedGeneration field in the Rollout object as a string
+	// so read it as a string here
+	observedGenerationString, found, err := unstructured.NestedString(u.Object, "status", "observedGeneration")
+	if err != nil {
+		return &result, fmt.Errorf("looking up status.observedGeneration from resource: %w", err)
+	}
+	if !found {
+		// We know that Rollout resources uses the ObservedGeneration pattern, so consider it
+		// an error if it is not found.
+		return &result, fmt.Errorf("status.ObservedGeneration not found")
+	}
+	// If no errors detected and the field is found
+	// Parse it to become an integer
+	observedGeneration, err := strconv.ParseInt(observedGenerationString, 10, 64)
+	if err != nil {
+		return &result, fmt.Errorf("looking up status.observedGeneration from resource: %w", err)
+	}
+
+	if generation != observedGeneration {
+		msg := fmt.Sprintf("%s generation is %d, but latest observed generation is %d", u.GetKind(), generation, observedGeneration)
+		result.Status = status.InProgressStatus
+		result.Message = msg
+		return &result, nil
+	}
+
+	phase, phaseFound, err := unstructured.NestedString(u.Object, "status", "phase")
+	if err != nil {
+		return &result, fmt.Errorf("looking up status.phase from resource: %w", err)
+	}
+	if !phaseFound {
+		// We know that Rollout resources uses the phase pattern, so consider it
+		// an error if it is not found.
+		return &result, fmt.Errorf("status.phase not found")
+	}
+
+	conditions, condFound, err := unstructured.NestedSlice(u.Object, "status", "conditions")
+	if err != nil {
+		return &result, fmt.Errorf("looking up status.conditions from resource: %w", err)
+	}
+	if condFound {
+		data, err := yaml.Marshal(conditions)
+		if err != nil {
+			return &result, fmt.Errorf("failed to marshal conditions for %s/%s", u.GetNamespace(), u.GetName())
+		}
+		err = yaml.Unmarshal(data, &result.Conditions)
+		if err != nil {
+			return &result, fmt.Errorf("failed to unmarshal conditions for %s/%s", u.GetNamespace(), u.GetName())
+		}
+	}
+
+	specReplicas := status.GetIntField(u.Object, ".spec.replicas", 1) // Controller uses 1 as default if not specified.
+	statusReplicas := status.GetIntField(u.Object, ".status.replicas", 0)
+	updatedReplicas := status.GetIntField(u.Object, ".status.updatedReplicas", 0)
+	readyReplicas := status.GetIntField(u.Object, ".status.readyReplicas", 0)
+	availableReplicas := status.GetIntField(u.Object, ".status.availableReplicas", 0)
+
+	if specReplicas > statusReplicas {
+		message := fmt.Sprintf("replicas: %d/%d", statusReplicas, specReplicas)
+		result.Status = status.InProgressStatus
+		result.Message = message
+
+		return &result, nil
+	}
+
+	if statusReplicas > specReplicas {
+		message := fmt.Sprintf("Pending termination: %d", statusReplicas-specReplicas)
+		result.Status = status.InProgressStatus
+		result.Message = message
+		return &result, nil
+	}
+
+	if updatedReplicas > availableReplicas {
+		message := fmt.Sprintf("Available: %d/%d", availableReplicas, updatedReplicas)
+		result.Status = status.InProgressStatus
+		result.Message = message
+		return &result, nil
+	}
+
+	if specReplicas > readyReplicas {
+		message := fmt.Sprintf("Ready: %d/%d", readyReplicas, specReplicas)
+		result.Status = status.InProgressStatus
+		result.Message = message
+		return &result, nil
+	}
+
+	message := status.GetStringField(u.Object, ".status.message", "")
+	if message != "" {
+		message += " "
+	}
+	message += fmt.Sprintf("Ready Replicas: %d, Updated Replicas: %d", readyReplicas, updatedReplicas)
+	result.Message = message
+
+	switch phase {
+	case Degraded, Failed:
+		result.Status = status.FailedStatus
+	case Healthy:
+		result.Status = status.CurrentStatus
+	case Paused, Progressing:
+		result.Status = status.InProgressStatus
+	default:
+		// Undefined status
+		result.Status = status.UnknownStatus
+	}
+	return &result, nil
+}
+
+func (r *RolloutStatusReader) ReadStatus(ctx context.Context, reader engine.ClusterReader, id object.ObjMetadata) (
+	*event.ResourceStatus, error) {
+	gvk, err := toGVK(id.GroupKind, r.Mapper)
+	if err != nil {
+		return newUnknownResourceStatus(id, nil, err), nil
+	}
+
+	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 {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			return nil, err
+		}
+		if apierrors.IsNotFound(err) {
+			return newResourceStatus(id, status.NotFoundStatus, &u, "Resource not found"), nil
+		}
+		return newUnknownResourceStatus(id, nil, err), nil
+	}
+
+	return r.ReadStatusForObject(ctx, reader, &u)
+}
+
+func (r *RolloutStatusReader) ReadStatusForObject(_ context.Context, _ engine.ClusterReader, u *unstructured.Unstructured) (
+	*event.ResourceStatus, error) {
+	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), nil
+	}
+	if found && deletionTimestamp != "" {
+		return newResourceStatus(id, status.TerminatingStatus, u, "Resource scheduled for deletion"), nil
+	}
+
+	res, err := r.Compute(u)
+	if err != nil {
+		return newUnknownResourceStatus(id, u, err), nil
+	}
+
+	return newResourceStatus(id, res.Status, u, res.Message), nil
+}


Test patch:
diff --git a/pkg/status/rollout_test.go b/pkg/status/rollout_test.go
new file mode 100644
index 00000000..de66b959
--- /dev/null
+++ b/pkg/status/rollout_test.go
@@ -0,0 +1,339 @@
+// 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"
+	"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 TestRolloutSupports(t *testing.T) {
+	testCases := map[string]struct {
+		gk       schema.GroupKind
+		supports bool
+	}{
+		"matches rollout group": {
+			gk: schema.GroupKind{
+				Group: "argoproj.io",
+				Kind:  "Rollout",
+			},
+			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 := &RolloutStatusReader{
+				Mapper: fakeMapper,
+			}
+
+			supports := statusReader.Supports(tc.gk)
+
+			assert.Equal(t, tc.supports, supports)
+		})
+	}
+}
+
+func TestRolloutReadStatus(t *testing.T) {
+	testCases := map[string]struct {
+		resource       string
+		gvk            schema.GroupVersionKind
+		expectedStatus status.Status
+		expectedMsg    string
+	}{
+		"Resource where observedGeneration doesn't match generation is InProgress": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+status:
+  observedGeneration: "41"
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Version: "v1alpha1",
+				Kind:    "Rollout",
+			},
+			expectedStatus: status.InProgressStatus,
+			expectedMsg:    "Rollout generation is 42, but latest observed generation is 41",
+		},
+		"Resource when spec.replicas more than status.replicas is InProgress": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 1
+  phase: Progressing
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.InProgressStatus,
+			expectedMsg:    "replicas: 1/2",
+		},
+		"Resource when status.replicas more than spec.replicas is InProgress": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 1
+status:
+  observedGeneration: "42"
+  replicas: 2
+  phase: Progressing
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.InProgressStatus,
+			expectedMsg:    "Pending termination: 1",
+		},
+		"Resource when status.updatedReplicas more than spec.availableReplicas is InProgress": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 2
+  updatedReplicas: 2
+  availableReplicas: 1
+  phase: Progressing
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.InProgressStatus,
+			expectedMsg:    "Available: 1/2",
+		},
+		"Resource when spec.replicas more than spec.readyReplicas is InProgress": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 2
+  updatedReplicas: 2
+  availableReplicas: 2
+  readyReplicas: 1
+  phase: Progressing
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.InProgressStatus,
+			expectedMsg:    "Ready: 1/2",
+		},
+		"Resource when status.phase is Degraded is Failed": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 2
+  updatedReplicas: 2
+  availableReplicas: 2
+  readyReplicas: 2
+  phase: Degraded
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.FailedStatus,
+			expectedMsg:    "Ready Replicas: 2, Updated Replicas: 2",
+		},
+		"Resource when status.phase is Failed is Failed": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 2
+  updatedReplicas: 2
+  availableReplicas: 2
+  readyReplicas: 2
+  phase: Failed
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.FailedStatus,
+			expectedMsg:    "Ready Replicas: 2, Updated Replicas: 2",
+		},
+		"Resource when status.phase is Healthy is Current": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 2
+  updatedReplicas: 2
+  availableReplicas: 2
+  readyReplicas: 2
+  phase: Healthy
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.CurrentStatus,
+			expectedMsg:    "Ready Replicas: 2, Updated Replicas: 2",
+		},
+		"Resource when status.phase is Paused is InProgress": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 2
+  updatedReplicas: 2
+  availableReplicas: 2
+  readyReplicas: 2
+  phase: Paused
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.InProgressStatus,
+			expectedMsg:    "Ready Replicas: 2, Updated Replicas: 2",
+		},
+		"Resource when status.phase is Progressing is InProgress": {
+			resource: `
+apiVersion: argoproj.io/v1alpha1
+kind: Rollout
+metadata:
+  name: pubsub.googleapis.com
+  namespace: cnrm
+  generation: 42
+spec:
+  replicas: 2
+status:
+  observedGeneration: "42"
+  replicas: 2
+  updatedReplicas: 2
+  availableReplicas: 2
+  readyReplicas: 2
+  phase: Progressing
+`,
+			gvk: schema.GroupVersionKind{
+				Group:   "argoproj.io",
+				Kind:    "Rollout",
+				Version: "v1alpha1",
+			},
+			expectedStatus: status.InProgressStatus,
+			expectedMsg:    "Ready Replicas: 2, Updated Replicas: 2",
+		},
+	}
+
+	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 := &RolloutStatusReader{
+				Mapper: fakeMapper,
+			}
+
+			res, err := statusReader.ReadStatus(context.Background(), fakeClusterReader, object.UnstructuredToObjMetadata(obj))
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expectedStatus, res.Status)
+			assert.Equal(t, tc.expectedMsg, res.Message)
+		})
+	}
+}