[#2273] Prevent a GracefulShutdown if view is != #replicas or any cache is not in a HEALTHY state

- Make org.infinispan.LOCKS cache as AVAILABLE automatically in upgrade
  testsuite to workaround known issues such as: https://issues.redhat.com/browse/ISPN-15191

Author: ryanemerson

pkg/http/curl/curl.go

@@ -120,6 +120,11 @@ func (c *Client) executeCurlWithAuth(httpURL, headers, args string) (*http.Respo
 		return nil, err
 	}
 
+	// Handle anonymous endpoints such as /health/status that will not return a 401 response
+	if rsp.StatusCode >= 200 && rsp.StatusCode < 300 {
+		return processResponse(rsp)
+	}
+
 	if rsp.StatusCode != http.StatusUnauthorized {
 		return rsp, fmt.Errorf("expected 401 DIGEST response before content. Received '%s'", rsp.Status)
 	}
@@ -172,9 +177,13 @@ func handleContent(reader *bufio.Reader) (*http.Response, error) {
 		}
 	}
 
+	return processResponse(rsp)
+}
+
+func processResponse(rsp *http.Response) (*http.Response, error) {
 	// Save response body
 	b := new(bytes.Buffer)
-	if _, err = io.Copy(b, rsp.Body); err != nil {
+	if _, err := io.Copy(b, rsp.Body); err != nil {
 		return nil, err
 	}
 	if err := rsp.Body.Close(); err != nil {

pkg/infinispan/client/api/infinispan.go

@@ -24,6 +24,7 @@ type Infinispan interface {
 type Container interface {
 	Info() (*ContainerInfo, error)
 	Backups() Backups
+	Health() (*Health, error)
 	HealthStatus() (HealthStatus, error)
 	Members() ([]string, error)
 	RebalanceDisable() error
@@ -101,12 +102,26 @@ type Xsite interface {
 	PushAllState() error
 }
 
+type Health struct {
+	ClusterHealth ClusterHealth `json:"cluster_health"`
+	CacheHealth   []CacheHealth `json:"cache_health"`
+}
+
+type CacheHealth struct {
+	Name   string       `json:"cache_name"`
+	Status HealthStatus `json:"status"`
+}
+
+type ClusterHealth struct {
+	Status HealthStatus `json:"health_status"`
+}
+
 // HealthStatus indicated the possible statuses of the Infinispan server
 type HealthStatus string
 
 const (
 	HealthStatusDegraded          HealthStatus = "DEGRADED"
-	HealthStatusHealth            HealthStatus = "HEALTHY"
+	HealthStatusHealthy           HealthStatus = "HEALTHY"
 	HealthStatusHealthRebalancing HealthStatus = "HEALTHY_REBALANCING"
 	HealthStatusFailed            HealthStatus = "FAILED"
 )
@@ -152,9 +167,11 @@ type BackupRestoreResources struct {
 }
 
 type ContainerInfo struct {
-	Coordinator bool           `json:"coordinator"`
-	SitesView   *[]interface{} `json:"sites_view,omitempty"`
-	Version     string         `json:"version"`
+	ClusterMembers []string       `json:"cluster_members"`
+	ClusterSize    int32          `json:"cluster_size"`
+	Coordinator    bool           `json:"coordinator"`
+	SitesView      *[]interface{} `json:"sites_view,omitempty"`
+	Version        string         `json:"version"`
 }
 
 type NotSupportedError struct {

pkg/infinispan/client/v14/container.go

@@ -31,6 +31,21 @@ func (c *Container) Info() (info *api.ContainerInfo, err error) {
 	return
 }
 
+func (c *Container) Health() (status *api.Health, err error) {
+	rsp, err := c.Get(c.CacheManager("/health"), nil)
+	defer func() {
+		err = httpClient.CloseBody(rsp, err)
+	}()
+
+	if err = httpClient.ValidateResponse(rsp, err, "getting cache manager health", http.StatusOK); err != nil {
+		return
+	}
+	if err = json.NewDecoder(rsp.Body).Decode(&status); err != nil {
+		return nil, fmt.Errorf("unable to decode: %w", err)
+	}
+	return
+}
+
 func (c *Container) HealthStatus() (status api.HealthStatus, err error) {
 	rsp, err := c.Get(c.CacheManager("/health/status"), nil)
 	defer func() {

pkg/reconcile/pipeline/infinispan/handler/manage/upgrades.go

@@ -7,8 +7,8 @@ import (
 
 	ispnv1 "github.com/infinispan/infinispan-operator/api/v1"
 	consts "github.com/infinispan/infinispan-operator/controllers/constants"
+	ispnApi "github.com/infinispan/infinispan-operator/pkg/infinispan/client/api"
 	"github.com/infinispan/infinispan-operator/pkg/infinispan/version"
-	kube "github.com/infinispan/infinispan-operator/pkg/kubernetes"
 	pipeline "github.com/infinispan/infinispan-operator/pkg/reconcile/pipeline/infinispan"
 	"github.com/infinispan/infinispan-operator/pkg/reconcile/pipeline/infinispan/handler/provision"
 	routev1 "github.com/openshift/api/route/v1"
@@ -130,7 +130,8 @@ func GracefulShutdown(i *ispnv1.Infinispan, ctx pipeline.Context) {
 	// Initiate the GracefulShutdown if it's not already in progress
 	if i.Spec.Replicas == 0 {
 		logger.Info(".Spec.Replicas==0")
-		if *statefulSet.Spec.Replicas != 0 {
+		replicas := *statefulSet.Spec.Replicas
+		if replicas != 0 {
 			logger.Info("StatefulSet.Spec.Replicas!=0")
 			// Only send a GracefulShutdown request to the server if it hasn't succeeded already
 			if !i.IsConditionTrue(ispnv1.ConditionStopping) {
@@ -141,18 +142,94 @@ func GracefulShutdown(i *ispnv1.Infinispan, ctx pipeline.Context) {
 					return
 				}
 
-				var rebalanceDisabled bool
+				type ServerMeta struct {
+					client ispnApi.Infinispan
+					skip   bool
+					state  string
+				}
+
+				// Loop through all pods to initiate the clients, determine that all pods have the expected
+				// number of cluster members and ensure that the cluster is in a HEALTHY state
+				var podMetaMap = make(map[string]*ServerMeta, len(podList.Items))
+				var shutdownAlreadyInitiated bool
 				for _, pod := range podList.Items {
-					ispnClient, err := ctx.InfinispanClientUnknownVersion(pod.Name)
+					podMeta := &ServerMeta{}
+					podMetaMap[pod.Name] = podMeta
+					podMeta.client, err = ctx.InfinispanClientUnknownVersion(pod.Name)
 					if err != nil {
 						if shutdown, state := containerAlreadyShutdown(err); shutdown {
-							logger.Info("Skipping pod whose cache-container has already been shutdown by the Operator", "pod", pod.Name, "state", state)
+							podMeta.skip, podMeta.state = true, state
+							logger.Info("At least one cache-container has already been shutdown by the Operator, resuming shutdown", "pod", pod.Name, "state", state)
+							shutdownAlreadyInitiated = true
+							// Continue processing other pods here as it's possible that one or more pods still haven't
+							// been shutdown and we need to initiate the client
 							continue
 						}
-						ctx.Requeue(fmt.Errorf("unable to create Infinispan client for cluster being upgraded: %w", err))
+						ctx.Requeue(fmt.Errorf("unable to create Infinispan client to determine if split-brain is present: %w", err))
+						return
+					}
+
+					// If one or more of the pods have already been shutdown then we must continue to shutdown the remaining
+					// members of the cluster
+					if shutdownAlreadyInitiated {
+						continue
+					}
+
+					info, err := podMeta.client.Container().Info()
+					if err != nil {
+						ctx.Requeue(fmt.Errorf("unable to retrieve cache-container info for pod '%s': %w", pod.Name, err))
+						return
+					}
+
+					updateCondition := func(msg string) {
+						err := ctx.UpdateInfinispan(func() {
+							i.SetCondition(ispnv1.ConditionStopping, metav1.ConditionFalse, msg)
+						})
+						if err != nil {
+							ctx.Log().Error(err, "unable to set condition to False", "condition", "stopping")
+						}
+					}
+
+					if info.ClusterSize != replicas {
+						msg := fmt.Sprintf(
+							"unable to proceed with GracefulShutdown as pod '%s' has '%d' cluster members, expected '%d'. Members: '%s'",
+							pod.Name,
+							info.ClusterSize,
+							i.Spec.Replicas,
+							strings.Join(info.ClusterMembers, ","),
+						)
+						updateCondition(msg)
+						ctx.EventRecorder().Event(i, corev1.EventTypeWarning, "GracefulShutdownBlocked", msg)
+						ctx.Requeue(errors.New(msg))
 						return
 					}
 
+					health, err := podMeta.client.Container().Health()
+					if err != nil {
+						ctx.Requeue(fmt.Errorf("unable to retrieve cluster health status for pod '%s': %w", pod.Name, err))
+						return
+					}
+
+					// If any of the caches are not marked as HEALTHY we must prevent a GracefulShutdown to prevent
+					// the cluster from entering an unexpected state
+					if health.ClusterHealth.Status != ispnApi.HealthStatusHealthy {
+						msg := fmt.Sprintf("unable to proceed with GracefulShutdown as the cluster health is '%s': %s", health.ClusterHealth.Status, health.CacheHealth)
+						updateCondition(msg)
+						ctx.EventRecorder().Event(i, corev1.EventTypeWarning, "GracefulShutdownBlocked", msg)
+						ctx.Requeue(errors.New(msg))
+						return
+					}
+				}
+
+				var rebalanceDisabled bool
+				for _, pod := range podList.Items {
+					podMeta := podMetaMap[pod.Name]
+					if podMeta.skip {
+						logger.Info("Skipping pod whose cache-container has already been shutdown by the Operator", "pod", pod, "state", podMeta.state)
+						continue
+					}
+					ispnClient := podMeta.client
+
 					// Disabling rebalancing is a cluster-wide operation so we only need to perform this on a single pod
 					// However, multiple calls to this endpoint should be safe, so it's ok if a subsequent reconciliation
 					// executes this again
@@ -164,13 +241,11 @@ func GracefulShutdown(i *ispnv1.Infinispan, ctx pipeline.Context) {
 						rebalanceDisabled = true
 					}
 
-					if kube.IsPodReady(pod) {
-						if err := ispnClient.Container().Shutdown(); err != nil {
-							ctx.Requeue(fmt.Errorf("error encountered on container shutdown: %w", err))
-							return
-						} else {
-							logger.Info("Executed Container Shutdown on pod: ", "Pod.Name", pod.Name)
-						}
+					if err := ispnClient.Container().Shutdown(); err != nil {
+						ctx.Requeue(fmt.Errorf("error encountered on container shutdown: %w", err))
+						return
+					} else {
+						logger.Info("Executed Container Shutdown on pod: ", "Pod.Name", pod.Name)
 					}
 				}
 
@@ -286,10 +361,14 @@ func EnableRebalanceAfterScaleUp(i *ispnv1.Infinispan, ctx pipeline.Context) {
 			return
 		}
 
-		if members, err := ispnClient.Container().Members(); err != nil {
-			ctx.Requeue(fmt.Errorf("unable to retrieve cluster members on scale up: %w", err))
+		// TODO why is this failing in TestOperandUpgrade for 14.0.x servers?
+		/*
+			2025-05-16T17:11:58.093+0100	ERROR	Reconciler error	{"controller": "infinispan", "controllerGroup": "infinispan.org", "controllerKind": "Infinispan", "infinispan": {"name":"test-operand-upgrade","namespace":"namespace-for-testing"}, "namespace": "namespace-for-testing", "name": "test-operand-upgrade", "reconcileID": "eda96c7e-fd83-43cc-b903-4a623e4e2785", "error": "unable to retrieve cluster information on scale up: unexpected error getting cache manager info: stderr: , err: the server does not allow this method on the requested resource"}
+		*/
+		if info, err := ispnClient.Container().Info(); err != nil {
+			ctx.Requeue(fmt.Errorf("unable to retrieve cluster information on scale up: %w", err))
 			return
-		} else if len(members) != int(i.Spec.Replicas) {
+		} else if info.ClusterSize != i.Spec.Replicas {
 			ctx.Log().Info("waiting for cluster to form", "replicas", i.Spec.Replicas)
 			ctx.RequeueAfter(consts.DefaultWaitClusterPodsNotReady, nil)
 			return

test/e2e/infinispan/upgrade_operand_test.go

@@ -2,6 +2,7 @@ package infinispan
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	ispnv1 "github.com/infinispan/infinispan-operator/api/v1"
@@ -11,6 +12,7 @@ import (
 	"github.com/infinispan/infinispan-operator/pkg/mime"
 	"github.com/infinispan/infinispan-operator/pkg/reconcile/pipeline/infinispan/handler/provision"
 	tutils "github.com/infinispan/infinispan-operator/test/e2e/utils"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 
 	"github.com/stretchr/testify/assert"
@@ -422,6 +424,35 @@ func TestPodAlreadyShutdownOnUpgrade(t *testing.T) {
 	assert.EqualValues(t, int32(0), ss.Status.UpdatedReplicas)
 }
 
+func TestScaleDownBlockedWithDegradedCache(t *testing.T) {
+	defer testKube.CleanNamespaceAndLogOnPanic(t, tutils.Namespace)
+
+	replicas := 1
+	ispn := tutils.DefaultSpec(t, testKube, func(i *ispnv1.Infinispan) {
+		i.Spec.Replicas = int32(replicas)
+	})
+	testKube.CreateInfinispan(ispn, tutils.Namespace)
+	testKube.WaitForInfinispanPods(replicas, tutils.SinglePodTimeout, ispn.Name, tutils.Namespace)
+	ispn = testKube.WaitForInfinispanCondition(ispn.Name, ispn.Namespace, ispnv1.ConditionWellFormed)
+
+	_client := tutils.HTTPClientForClusterWithVersionManager(ispn, testKube, tutils.VersionManager())
+	cacheName := "cache"
+	cacheConfig := "<distributed-cache><partition-handling when-split=\"DENY_READ_WRITES\" merge-policy=\"PREFERRED_ALWAYS\"/></distributed-cache>"
+	cacheHelper := tutils.NewCacheHelper(cacheName, _client)
+	cacheHelper.Create(cacheConfig, mime.ApplicationXml)
+	cacheHelper.Available(false)
+
+	tutils.ExpectNoError(
+		testKube.UpdateInfinispan(ispn, func() {
+			ispn.Spec.Replicas = 0
+		}),
+	)
+	testKube.WaitForInfinispanState(ispn.Name, ispn.Namespace, func(i *ispnv1.Infinispan) bool {
+		c := i.GetCondition(ispnv1.ConditionStopping)
+		return c.Status == metav1.ConditionFalse && strings.Contains(c.Message, "unable to proceed with GracefulShutdown as the cluster health is 'DEGRADED'")
+	})
+}
+
 // specImageOperands() returns two latest Operands with the matching major/minor version
 func specImageOperands() (*version.Operand, *version.Operand) {
 	operands := tutils.VersionManager().Operands

test/e2e/upgrade/common.go

@@ -259,7 +259,10 @@ func checkBatch(t *testing.T, infinispan *ispnv1.Infinispan) {
 	testKube.DeleteBatch(batch)
 }
 
-func assertNoDegradedCaches() {
+func locksCacheDegraded(operand version.Operand) bool {
+	if operand.UpstreamVersion.Major > 14 {
+		return false
+	}
 	podList := &corev1.PodList{}
 	tutils.ExpectNoError(
 		testKube.Kubernetes.ResourcesList(tutils.Namespace, map[string]string{"app": "infinispan-pod"}, podList, ctx),
@@ -269,7 +272,15 @@ func assertNoDegradedCaches() {
 		log, err := testKube.Kubernetes.Logs(provision.InfinispanContainer, pod.GetName(), pod.GetNamespace(), false, ctx)
 		tutils.ExpectNoError(err)
 		if strings.Contains(log, "DEGRADED_MODE") {
-			panic("Detected a cache in DEGRADED_MODE, unable to continue with test")
+			return true
 		}
 	}
+	return false
+}
+
+// healDegradedLocksCache forces the org.infinispan.LOCKS cache to AVAILABLE so that subsequent upgrades can proceed
+// can be removed once Infinispan 14.0.x servers are no longer supported
+func healDegradedLocksCache(operand version.Operand, client tutils.HTTPClient) {
+	tutils.Log().Infof("Setting org.infinispan.LOCKS cache to AVAILABLE for Operand %s", operand)
+	tutils.NewCacheHelperForOperand("org.infinispan.LOCKS", operand, client).Available(true)
 }

test/e2e/upgrade/upgrade_operands_test.go

@@ -41,11 +41,12 @@ func TestOperandUpgrades(t *testing.T) {
 
 	// Create the Infinispan CR
 	replicas := 2
+	startingOperand := versionManager.Operands[startingOperandIdx]
 	ispn := tutils.DefaultSpec(t, testKube, func(i *ispnv1.Infinispan) {
 		i.Spec.Replicas = int32(replicas)
 		i.Spec.Service.Container.EphemeralStorage = false
 		i.Spec.Logging.Categories["org.infinispan.topology"] = ispnv1.LoggingLevelTrace
-		i.Spec.Version = versionManager.Operands[startingOperandIdx].Ref()
+		i.Spec.Version = startingOperand.Ref()
 	})
 
 	log.Infof("Testing upgrades from Operand '%s' to '%s'", ispn.Spec.Version, versionManager.Latest().Ref())
@@ -56,15 +57,17 @@ func TestOperandUpgrades(t *testing.T) {
 	numEntries := 100
 	client := tutils.HTTPClientForClusterWithVersionManager(ispn, testKube, versionManager)
 
+	if op := *startingOperand; locksCacheDegraded(op) {
+		healDegradedLocksCache(op, client)
+	}
+
 	// Add a persistent cache with data to ensure contents can be read after upgrade(s)
 	createAndPopulatePersistentCache(persistentCacheName, numEntries, client)
 
 	// Add a volatile cache with data to ensure contents can be backed up and then restored after upgrade(s)
 	createAndPopulateVolatileCache(volatileCacheName, numEntries, client)
 
-	assertNoDegradedCaches()
 	backup := createBackupAndWaitToSucceed(ispn.Name, t)
-
 	skippedOperands := tutils.OperandSkipSet()
 	for _, operand := range versionManager.Operands[startingOperandIdx:] {
 		// Skip an Operand in the upgrade graph if there's a known issue
@@ -75,6 +78,10 @@ func TestOperandUpgrades(t *testing.T) {
 		log.Debugf("Next Operand %s", operand.Ref())
 
 		ispn = testKube.WaitForInfinispanConditionWithTimeout(ispn.Name, tutils.Namespace, ispnv1.ConditionWellFormed, conditionTimeout)
+		if op := *operand; locksCacheDegraded(op) {
+			healDegradedLocksCache(op, client)
+		}
+
 		tutils.ExpectNoError(
 			testKube.UpdateInfinispan(ispn, func() {
 				ispn.Spec.Version = operand.Ref()
@@ -89,8 +96,6 @@ func TestOperandUpgrades(t *testing.T) {
 				i.Status.Operand.Phase == ispnv1.OperandPhaseRunning
 		})
 		log.Info("Upgrade complete")
-		assertOperandImage(operand.Image, ispn)
-		assertNoDegradedCaches()
 
 		// Ensure that persistent cache entries have survived the upgrade(s)
 		// Refresh the hostAddr and client as the url will change if NodePort is used.