k8s: Create negative test for upgrade procedure

When cluster is unhealthy the upgrade/restarting procedure should not be
executed.

src/go/k8s/controllers/redpanda/suite_test.go

@@ -102,6 +102,8 @@ var _ = BeforeSuite(func(done Done) {
 		}
 		return testAdminAPI, nil
 	}
+	testAdminAPI.SetClusterHealth(true)
+
 	testStore = consolepkg.NewStore(k8sManager.GetClient(), k8sManager.GetScheme())
 	testKafkaAdmin = &mockKafkaAdmin{}
 	testKafkaAdminFactory = func(context.Context, client.Client, *redpandav1alpha1.Cluster, *consolepkg.Store) (consolepkg.KafkaAdminClient, error) {

src/go/k8s/pkg/admin/mock_admin.go

@@ -32,6 +32,7 @@ type MockAdminAPI struct {
 	directValidation bool
 	brokers          []admin.Broker
 	monitor          sync.Mutex
+	clusterHealth    bool
 }
 
 var _ AdminAPIClient = &MockAdminAPI{}
@@ -47,6 +48,12 @@ func (*unavailableError) Error() string {
 	return "unavailable"
 }
 
+func (m *MockAdminAPI) SetClusterHealth(health bool) {
+	m.monitor.Lock()
+	defer m.monitor.Unlock()
+	m.clusterHealth = health
+}
+
 func (m *MockAdminAPI) Config(context.Context, bool) (admin.Config, error) {
 	m.monitor.Lock()
 	defer m.monitor.Unlock()
@@ -363,7 +370,7 @@ func (m *MockAdminAPI) DisableMaintenanceMode(_ context.Context, _ int) error {
 
 func (m *MockAdminAPI) GetHealthOverview(_ context.Context) (admin.ClusterHealthOverview, error) {
 	return admin.ClusterHealthOverview{
-		IsHealthy: true,
+		IsHealthy: m.clusterHealth,
 	}, nil
 }
 

src/go/k8s/pkg/resources/statefulset_test.go

@@ -11,6 +11,7 @@ package resources_test
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -65,17 +66,25 @@ func TestEnsure(t *testing.T) {
 	// Remove shadow-indexing-cache from the volume claim templates
 	stsWithoutSecondPersistentVolume.Spec.VolumeClaimTemplates = stsWithoutSecondPersistentVolume.Spec.VolumeClaimTemplates[:1]
 
+	unhealthyRedpandaCluster := cluster.DeepCopy()
+
 	tests := []struct {
 		name           string
 		existingObject client.Object
 		pandaCluster   *redpandav1alpha1.Cluster
 		expectedObject *v1.StatefulSet
+		clusterHealth  bool
+		expectedError  error
 	}{
-		{"none existing", nil, cluster, stsResource},
-		{"update resources", stsResource, resourcesUpdatedCluster, resourcesUpdatedSts},
-		{"update redpanda resources", stsResource, resourcesUpdatedRedpandaCluster, resourcesUpdatedSts},
-		{"disabled sidecar", nil, noSidecarCluster, noSidecarSts},
-		{"cluster without shadow index cache dir", stsResource, withoutShadowIndexCacheDirectory, stsWithoutSecondPersistentVolume},
+		{"none existing", nil, cluster, stsResource, true, nil},
+		{"update resources", stsResource, resourcesUpdatedCluster, resourcesUpdatedSts, true, nil},
+		{"update redpanda resources", stsResource, resourcesUpdatedRedpandaCluster, resourcesUpdatedSts, true, nil},
+		{"disabled sidecar", nil, noSidecarCluster, noSidecarSts, true, nil},
+		{"cluster without shadow index cache dir", stsResource, withoutShadowIndexCacheDirectory, stsWithoutSecondPersistentVolume, true, nil},
+		{"update none healthy cluster", stsResource, unhealthyRedpandaCluster, stsResource, false, &res.RequeueAfterError{
+			RequeueAfter: res.RequeueDuration,
+			Msg:          "wait for cluster to become healthy (cluster restarting)",
+		}},
 	}
 
 	for _, tt := range tests {
@@ -112,13 +121,20 @@ func TestEnsure(t *testing.T) {
 				},
 				func(ctx context.Context) (string, error) { return hash, nil },
 				func(ctx context.Context, k8sClient client.Reader, redpandaCluster *redpandav1alpha1.Cluster, fqdn string, adminTLSProvider resourcetypes.AdminTLSConfigProvider, ordinals ...int32) (adminutils.AdminAPIClient, error) {
-					return &adminutils.MockAdminAPI{}, nil
+					health := tt.clusterHealth
+					adminAPI := &adminutils.MockAdminAPI{}
+					adminAPI.SetClusterHealth(health)
+					return adminAPI, nil
 				},
 				time.Second,
 				ctrl.Log.WithName("test"))
 
 			err = sts.Ensure(context.Background())
-			assert.NoError(t, err, tt.name)
+			if tt.expectedError != nil && errors.Is(err, tt.expectedError) {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err, tt.name)
+			}
 
 			actual := &v1.StatefulSet{}
 			err = c.Get(context.Background(), sts.Key(), actual)

src/go/k8s/pkg/resources/statefulset_update.go

@@ -457,6 +457,10 @@ func (e *RequeueAfterError) Error() string {
 	return fmt.Sprintf("RequeueAfterError %s", e.Msg)
 }
 
+func (e *RequeueAfterError) Is(target error) bool {
+	return e.Error() == target.Error()
+}
+
 // RequeueError error to requeue using default retry backoff.
 type RequeueError struct {
 	Msg string