func (rc *ReconciliationContext) CheckDecommissioningNodes(epData httphelper.CassMetadataEndpoints) result.ReconcileResult {
	if rc.Datacenter.GetConditionStatus(api.DatacenterScalingDown) != corev1.ConditionTrue {
		return result.Continue()
	}
	for _, pod := range rc.dcPods {
		if pod.Labels[api.CassNodeState] == stateDecommissioning {
			if !IsDoneDecommissioning(pod, epData) {
				if !HasStartedDecommissioning(pod, epData) {
					rc.ReqLogger.Info("Decommission has not started trying again")
					if err := rc.NodeMgmtClient.CallDecommissionNodeEndpoint(pod); err != nil {
						rc.ReqLogger.Info(fmt.Sprintf("Error from decomimssion attempt. This is only an attempt and can fail. Error: %v", err))
					}
				} else {
					rc.ReqLogger.Info("Node decommissioning, reconciling again soon")
				}
			} else {
				rc.ReqLogger.Info("Node finished decommissioning")
				if res := rc.cleanUpAfterDecommissionedPod(pod); res != nil {
					return res
				}
			}
			return result.RequeueSoon(5)
		}
	}
	dcPatch := client.MergeFrom(rc.Datacenter.DeepCopy())
	updated := false
	updated = rc.setCondition(
		api.NewDatacenterCondition(
			api.DatacenterScalingDown, corev1.ConditionFalse)) || updated
	if updated {
		err := rc.Client.Status().Patch(rc.Ctx, rc.Datacenter, dcPatch)
		if err != nil {
			rc.ReqLogger.Error(err, "error patching datacenter status for scaling down finished")
			return result.Error(err)
		}
	}
	return result.Continue()
}
Wait for decommissioning nodes to finish before continuing to reconcile
package reconciliation
import (
	"fmt"
	"strconv"
	"strings"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	v1 "k8s.io/api/core/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	api "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
	"github.com/k8ssandra/cass-operator/pkg/events"
	"github.com/k8ssandra/cass-operator/pkg/httphelper"
	"github.com/k8ssandra/cass-operator/pkg/internal/result"
	"k8s.io/apimachinery/pkg/types"
)
func (rc *ReconciliationContext) CalculateRackInfoForDecomm(currentSize int) ([]*RackInformation, error) {
	racks := rc.Datacenter.GetRacks()
	rackCount := len(racks)
	desiredSize := currentSize - 1
	if desiredSize < rackCount {
		return nil, fmt.Errorf("the number of nodes cannot be smaller than the number of racks")
	}
	var decommRackInfo []*RackInformation
	rackNodeCounts := api.SplitRacks(desiredSize, rackCount)
	for rackIndex, currentRack := range racks {
		nextRack := &RackInformation{}
		nextRack.RackName = currentRack.Name
		nextRack.NodeCount = rackNodeCounts[rackIndex]
		decommRackInfo = append(decommRackInfo, nextRack)
	}
	return decommRackInfo, nil
}
func (rc *ReconciliationContext) DecommissionNodes(epData httphelper.CassMetadataEndpoints) result.ReconcileResult {
	logger := rc.ReqLogger
	logger.Info("reconcile_racks::DecommissionNodes")
	dc := rc.Datacenter
	var currentSize int32
	for _, sts := range rc.statefulSets {
		if sts != nil {
			currentSize += *sts.Spec.Replicas
		}
	}
	if currentSize <= dc.Spec.Size {
		return result.Continue()
	}
	decommRackInfo, err := rc.CalculateRackInfoForDecomm(int(currentSize))
	if err != nil {
		logger.Error(err, "error calculating rack info for decommissioning nodes")
		return result.Error(err)
	}
	for idx := range decommRackInfo {
		rackInfo := decommRackInfo[idx]
		statefulSet := rc.statefulSets[idx]
		desiredNodeCount := int32(rackInfo.NodeCount)
		maxReplicas := *statefulSet.Spec.Replicas
		lastPodSuffix := stsLastPodSuffix(maxReplicas)
		if maxReplicas > desiredNodeCount {
			dcPatch := client.MergeFrom(dc.DeepCopy())
			updated := false
			updated = rc.setCondition(
				api.NewDatacenterCondition(
					api.DatacenterScalingDown, corev1.ConditionTrue)) || updated
			if updated {
				err := rc.Client.Status().Patch(rc.Ctx, dc, dcPatch)
				if err != nil {
					logger.Error(err, "error patching datacenter status for scaling down rack started")
					return result.Error(err)
				}
			}
			rc.ReqLogger.Info(
				"Need to update the rack's node count",
				"Rack", rackInfo.RackName,
				"maxReplicas", maxReplicas,
				"desiredSize", desiredNodeCount,
			)
			rc.Recorder.Eventf(rc.Datacenter, corev1.EventTypeNormal, events.ScalingDownRack,
				"Scaling down rack %s", rackInfo.RackName)
			if err := setOperatorProgressStatus(rc, api.ProgressUpdating); err != nil {
				return result.Error(err)
			}
			err := rc.DecommissionNodeOnRack(rackInfo.RackName, epData, lastPodSuffix)
			if err != nil {
				return result.Error(err)
			}
			return result.RequeueSoon(10)
		}
	}
	return result.Continue()
}
func (rc *ReconciliationContext) DecommissionNodeOnRack(rackName string, epData httphelper.CassMetadataEndpoints, lastPodSuffix string) error {
	for _, pod := range rc.dcPods {
		podRack := pod.Labels[api.RackLabel]
		if podRack == rackName && strings.HasSuffix(pod.Name, lastPodSuffix) {
			mgmtApiUp := isMgmtApiRunning(pod)
			if !mgmtApiUp {
				return fmt.Errorf("Management API is not up on node that we are trying to decommission")
			}
			if err := rc.EnsurePodsCanAbsorbDecommData(pod, epData); err != nil {
				return err
			}
			if err := rc.NodeMgmtClient.CallDecommissionNodeEndpoint(pod); err != nil {
				rc.ReqLogger.Info(fmt.Sprintf("Error from decommission attempt. This is only an attempt and can"+
					" fail it will be retried later if decomission has not started. Error: %v", err))
			}
			rc.ReqLogger.Info("Marking node as decommissioning")
			patch := client.MergeFrom(pod.DeepCopy())
			pod.Labels[api.CassNodeState] = stateDecommissioning
			if err := rc.Client.Patch(rc.Ctx, pod, patch); err != nil {
				return err
			}
			rc.Recorder.Eventf(rc.Datacenter, corev1.EventTypeNormal, events.LabeledPodAsDecommissioning,
				"Labeled node as decommissioning %s", pod.Name)
			return nil
		}
	}
	return fmt.Errorf("Could not find pod to decommission on rack %s", rackName)
}