func checkNodeEMM(provider EMMService) result.ReconcileResult {
	logger := provider.getLogger()
	didUpdate, err := provider.cleanupEMMAnnotations()
	if err != nil {
		logger.Error(err, "Encountered an error while cleaning up defunct EMM failure annotations")
		return result.Error(err)
	}
	if didUpdate {
		logger.Info("Cleaned up defunct EMM failure annotations")
		return result.RequeueSoon(2)
	}
	stillProcessing, err := provider.emmFailureStillProcessing()
	if err != nil {
		logger.Error(err, "Failed to check if EMM failures are still processing")
		return result.Error(err)
	}
	if stillProcessing {
		return result.RequeueSoon(2)
	}
	if !provider.IsInitialized() {
		logger.Info("Skipping EMM check as the cluster is not yet initialized")
		return result.Continue()
	}
	plannedDownNodeNameSet, err := provider.getPlannedDownTimeNodeNameSet()
	if err != nil {
		return result.Error(err)
	}
	evacuateDataNodeNameSet, err := provider.getEvacuateAllDataNodeNameSet()
	if err != nil {
		return result.Error(err)
	}
	allNodes, err := provider.getNodeNameSet()
	if err != nil {
		logger.Error(err, "Failed to get node name set")
		return result.Error(err)
	}
	unavailableNodes := utils.UnionStringSet(plannedDownNodeNameSet, evacuateDataNodeNameSet)
	availableNodes := utils.SubtractStringSet(allNodes, unavailableNodes)
	if len(provider.getPodNameSet()) > len(availableNodes) {
		logger.Info(fmt.Sprintf("Not enough space to do EMM. Total number of nodes: %v, unavailable: %v, available: %v, pods: %v",
			len(allNodes), len(unavailableNodes), len(availableNodes), len(provider.getPodNameSet())))
		anyUpdated := false
		updated := false
		for node := range unavailableNodes {
			if updated, err = provider.failEMM(node, NotEnoughResources); err != nil {
				logger.Error(err, "Failed to add "+EMMFailureAnnotation, "Node", node)
				return result.Error(err)
			}
			anyUpdated = anyUpdated || updated
		}
		if anyUpdated {
			return result.RequeueSoon(10)
		}
	}
	if provider.IsStopped() {
		didUpdate := false
		for nodeName, _ := range evacuateDataNodeNameSet {
			podsUpdated, err := provider.failEMM(nodeName, GenericFailure)
			if err != nil {
				return result.Error(err)
			}
			logger.Info("Failed EMM operation for evacuate all data as the datacenter is currently stopped", "node", nodeName)
			didUpdate = didUpdate || podsUpdated
		}
		if didUpdate {
			return result.RequeueSoon(2)
		}
	}
	racksWithDownPods := provider.getRacksWithNotReadyPodsBootstrapped()
	if len(racksWithDownPods) > 1 {
		allTaintedNameSet := utils.UnionStringSet(plannedDownNodeNameSet, evacuateDataNodeNameSet)
		didUpdate := false
		for nodeName, _ := range allTaintedNameSet {
			didUpdatePods, err := provider.failEMM(nodeName, TooManyExistingFailures)
			if err != nil {
				return result.Error(err)
			}
			logger.Info("Failed EMM operation as availability is already compromised due to multiple racks having not ready pods", "node", nodeName)
			didUpdate = didUpdate || didUpdatePods
		}
		if didUpdate {
			return result.RequeueSoon(2)
		}
	}
	downRack := ""
	for _, rackName := range racksWithDownPods {
		downRack = rackName
		break
	}
	if downRack != "" {
		nodeNameSetForDownRack := provider.getRackNodeNameSet(downRack)
		nodeNameSetForNoPodsInRack := utils.SubtractStringSet(
			utils.UnionStringSet(
				plannedDownNodeNameSet,
				evacuateDataNodeNameSet),
			nodeNameSetForDownRack)
		if len(nodeNameSetForNoPodsInRack) > 0 {
			didUpdate := false
			for nodeName, _ := range nodeNameSetForNoPodsInRack {
				podsUpdated, err := provider.failEMM(nodeName, TooManyExistingFailures)
				if err != nil {
					return result.Error(err)
				}
				logger.Info("Failed EMM operation as it did not contain any pods for the currently down rack", "node", nodeName)
				didUpdate = didUpdate || podsUpdated
			}
			if didUpdate {
				return result.RequeueSoon(2)
			}
		}
	}
	didUpdate, err = provider.removeAllNotReadyPodsOnEMMNodes()
	if err != nil {
		return result.Error(err)
	}
	if didUpdate {
		return result.RequeueSoon(2)
	}
	if downRack != "" {
		didUpdate, err = provider.performEvacuateDataPodReplace()
		if err != nil {
			return result.Error(err)
		}
		if didUpdate {
			return result.RequeueSoon(2)
		}
		return result.Continue()
	}
	didUpdate, err = provider.removeNextPodFromEvacuateDataNode()
	if err != nil {
		return result.Error(err)
	}
	if didUpdate {
		return result.RequeueSoon(2)
	}
	didUpdate, err = provider.removeAllPodsFromOnePlannedDowntimeNode()
	if err != nil {
		return result.Error(err)
	}
	if didUpdate {
		return result.RequeueSoon(2)
	}
	return result.Continue()
}
Check nodes for vmware PSP draining taints. This function embodies the
business logic around when EMM operations are executed.
package psp
import (
	"fmt"
	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	"github.com/k8ssandra/cass-operator/pkg/utils"
	api "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
	"github.com/k8ssandra/cass-operator/pkg/internal/result"
)
const (
	EMMFailureAnnotation   = "appplatform.vmware.com/emm-failure"
	VolumeHealthAnnotation = "volumehealth.storage.kubernetes.io/health"
	EMMTaintKey            = "node.vmware.com/drain"
)
type EMMTaintValue string
const (
	EvacuateAllData EMMTaintValue = "drain"
	PlannedDowntime               = "planned-downtime"
)
type VolumeHealth string
const (
	VolumeHealthInaccessible VolumeHealth = "inaccessible"
)
type EMMFailure string
const (
	GenericFailure          EMMFailure = "GenericFailure"
	NotEnoughResources      EMMFailure = "NotEnoughResources"
	TooManyExistingFailures EMMFailure = "TooManyExistingFailures"
)
type EMMSPI interface {
	GetAllNodesInDC() ([]*corev1.Node, error)
	GetDCPods() []*corev1.Pod
	GetNotReadyPodsBootstrappedInDC() []*corev1.Pod
	GetAllPodsNotReadyInDC() []*corev1.Pod
	GetPodPVCs(pod *corev1.Pod) ([]*corev1.PersistentVolumeClaim, error)
	StartNodeReplace(podName string) error
	GetInProgressNodeReplacements() []string
	RemovePod(pod *corev1.Pod) error
	UpdatePod(pod *corev1.Pod) error
	IsStopped() bool
	IsInitialized() bool
	GetLogger() logr.Logger
	GetAllNodes() ([]*corev1.Node, error)
}
type EMMChecks interface {
	getPodPVCSelectedNodeName(podName string) (string, error)
	getPodNameSetWithVolumeHealthInaccessiblePVC(rackName string) (utils.StringSet, error)
	getRacksWithNotReadyPodsBootstrapped() []string
	getRackNodeNameSet(rackName string) utils.StringSet
	getInProgressNodeReplacements() []string
	IsStopped() bool
	IsInitialized() bool
	getNodeNameSet() (utils.StringSet, error)
	getPodNameSet() utils.StringSet
}
type EMMOperations interface {
	cleanupEMMAnnotations() (bool, error)
	getPlannedDownTimeNodeNameSet() (utils.StringSet, error)
	getEvacuateAllDataNodeNameSet() (utils.StringSet, error)
	removeAllNotReadyPodsOnEMMNodes() (bool, error)
	failEMM(nodeName string, failure EMMFailure) (bool, error)
	performEvacuateDataPodReplace() (bool, error)
	removeNextPodFromEvacuateDataNode() (bool, error)
	removeAllPodsFromOnePlannedDowntimeNode() (bool, error)
	startNodeReplace(podName string) error
	emmFailureStillProcessing() (bool, error)
}
type EMMService interface {
	EMMOperations
	EMMChecks
	getLogger() logr.Logger
}
func getPodsRackNameSet(pods []*corev1.Pod) utils.StringSet {
	names := utils.StringSet{}
	for _, pod := range pods {
		names[pod.Labels[api.RackLabel]] = true
	}
	return names
}
func filterPodsByRackName(pods []*corev1.Pod, rackName string) []*corev1.Pod {
	return utils.FilterPodsWithLabel(pods, api.RackLabel, rackName)
}
type EMMServiceImpl struct {
	EMMSPI
}
func (impl *EMMServiceImpl) getPodPVCSelectedNodeName(podName string) (string, error) {
	matchingPods := utils.FilterPodsWithFn(impl.GetDCPods(), func(pod *corev1.Pod) bool {
		return pod != nil && pod.Name == podName
	})
	if len(matchingPods) == 0 {
		return "", nil
	}
	pod := matchingPods[0]
	pvcs, err := impl.GetPodPVCs(pod)
	if err != nil {
		return "", err
	}
	for _, pvc := range pvcs {
		nodeName := utils.GetPVCSelectedNodeName(pvc)
		if nodeName != "" {
			return nodeName, nil
		}
	}
	return "", nil
}
func (impl *EMMServiceImpl) cleanupEMMAnnotations() (bool, error) {
	nodes, err := impl.getEvacuateAllDataNodeNameSet()
	if err != nil {
		return false, err
	}
	nodes2, err := impl.getPlannedDownTimeNodeNameSet()
	if err != nil {
		return false, err
	}
	nodes = utils.UnionStringSet(nodes, nodes2)
	podsFailedEmm := impl.getPodsWithAnnotationKey(EMMFailureAnnotation)
	nodesWithPodsFailedEMM := utils.GetPodNodeNameSet(podsFailedEmm)
	nodesNoLongerEMM := utils.SubtractStringSet(
		nodesWithPodsFailedEMM,
		nodes)
	podsNoLongerFailed := utils.FilterPodsWithNodeInNameSet(podsFailedEmm, nodesNoLongerEMM)
	didUpdate := false
	for _, pod := range podsNoLongerFailed {
		err := impl.removePodAnnotation(pod, EMMFailureAnnotation)
		if err != nil {
			return false, err
		}
		didUpdate = true
	}
	return didUpdate, nil
}
func (impl *EMMServiceImpl) emmFailureStillProcessing() (bool, error) {
	nodes, err := impl.getEvacuateAllDataNodeNameSet()
	if err != nil {
		return false, err
	}
	nodes2, err := impl.getPlannedDownTimeNodeNameSet()
	if err != nil {
		return false, err
	}
	nodes = utils.UnionStringSet(nodes, nodes2)
	podsFailedEmm := impl.getPodsWithAnnotationKey(EMMFailureAnnotation)
	nodesWithPodsFailedEMM := utils.GetPodNodeNameSet(podsFailedEmm)
	return len(utils.IntersectionStringSet(nodes, nodesWithPodsFailedEMM)) > 0, nil
}
func (impl *EMMServiceImpl) getPlannedDownTimeNodeNameSet() (utils.StringSet, error) {
	nodes, err := impl.getNodesWithTaintKeyValueEffect(EMMTaintKey, string(PlannedDowntime), corev1.TaintEffectNoSchedule)
	if err != nil {
		return nil, err
	}
	return utils.GetNodeNameSet(nodes), nil
}
func (impl *EMMServiceImpl) getEvacuateAllDataNodeNameSet() (utils.StringSet, error) {
	nodes, err := impl.getNodesWithTaintKeyValueEffect(EMMTaintKey, string(EvacuateAllData), corev1.TaintEffectNoSchedule)
	if err != nil {
		return nil, err
	}
	return utils.GetNodeNameSet(nodes), nil
}
func (impl *EMMServiceImpl) removeAllNotReadyPodsOnEMMNodes() (bool, error) {
	podsNotReady := impl.GetAllPodsNotReadyInDC()
	plannedDownNameSet, err := impl.getPlannedDownTimeNodeNameSet()
	if err != nil {
		return false, err
	}
	evacuateDataNameSet, err := impl.getEvacuateAllDataNodeNameSet()
	if err != nil {
		return false, err
	}
	taintedNodesNameSet := utils.UnionStringSet(plannedDownNameSet, evacuateDataNameSet)
	podsNotReadyOnTaintedNodes := utils.FilterPodsWithNodeInNameSet(podsNotReady, taintedNodesNameSet)
	if len(podsNotReadyOnTaintedNodes) > 0 {
		for _, pod := range podsNotReadyOnTaintedNodes {
			err := impl.RemovePod(pod)
			if err != nil {
				return false, err
			}
		}
		return true, nil
	}
	return false, nil
}
func (impl *EMMServiceImpl) failEMM(nodeName string, failure EMMFailure) (bool, error) {
	pods := impl.getPodsForNodeName(nodeName)
	didUpdate := false
	for _, pod := range pods {
		added, err := impl.addPodAnnotation(pod, EMMFailureAnnotation, string(failure))
		if err != nil {
			return false, err
		}
		didUpdate = didUpdate || added
	}
	return didUpdate, nil
}
func (impl *EMMServiceImpl) getNodeNameSet() (utils.StringSet, error) {
	nodes, err := impl.EMMSPI.GetAllNodes()
	if err != nil {
		return nil, err
	}
	totalNodes := len(nodes)
	agentNodesIndex := []int{}
	for i := 0; i < totalNodes; i++ {
		if nodes[i].Labels["kubernetes.io/role"] == "agent" {
			agentNodesIndex = append(agentNodesIndex, i)
		}
	}
	agents := make([]*corev1.Node, len(agentNodesIndex))
	for i, agentIndex := range agentNodesIndex {
		agents[i] = nodes[agentIndex]
	}
	nameSet := utils.GetNodeNameSet(agents)
	return nameSet, nil
}
func (impl *EMMServiceImpl) getPodNameSet() utils.StringSet {
	return utils.GetPodNameSet(impl.EMMSPI.GetDCPods())
}
func (impl *EMMServiceImpl) performEvacuateDataPodReplace() (bool, error) {
	downPods := impl.GetNotReadyPodsBootstrappedInDC()
	if len(downPods) > 0 {
		evacuateAllDataNameSet, err := impl.getEvacuateAllDataNodeNameSet()
		if err != nil {
			return false, err
		}
		deletedPodsOrPVCs := false
		for _, pod := range downPods {
			if utils.IsPodUnschedulable(pod) {
				pvcNode, err := impl.getPodPVCSelectedNodeName(pod.Name)
				if err != nil {
					return false, err
				}
				if pvcNode != "" && pod.Spec.NodeName != pvcNode {
					if evacuateAllDataNameSet[pvcNode] {
						deletedPodsOrPVCs = true
						err := impl.StartNodeReplace(pod.Name)
						if err != nil {
							return false, err
						}
					}
				}
			}
		}
		return deletedPodsOrPVCs, nil
	}
	return false, nil
}
func (impl *EMMServiceImpl) removeNextPodFromEvacuateDataNode() (bool, error) {
	nodeNameSet, err := impl.getEvacuateAllDataNodeNameSet()
	if err != nil {
		return false, err
	}
	for name, _ := range nodeNameSet {
		for _, pod := range impl.getPodsForNodeName(name) {
			err := impl.RemovePod(pod)
			if err != nil {
				return false, err
			}
			return true, nil
		}
	}
	return false, nil
}
func (impl *EMMServiceImpl) removeAllPodsFromOnePlannedDowntimeNode() (bool, error) {
	nodeNameSet, err := impl.getPlannedDownTimeNodeNameSet()
	for nodeName, _ := range nodeNameSet {
		pods := impl.getPodsForNodeName(nodeName)
		if len(pods) > 0 {
			for _, pod := range pods {
				err := impl.RemovePod(pod)
				if err != nil {
					return false, err
				}
			}
			return true, err
		}
	}
	return false, nil
}
func (impl *EMMServiceImpl) startNodeReplace(podName string) error {
	return impl.StartNodeReplace(podName)
}
func (impl *EMMServiceImpl) getLogger() logr.Logger {
	return impl.GetLogger()
}
func (impl *EMMServiceImpl) getPodNameSetWithVolumeHealthInaccessiblePVC(rackName string) (utils.StringSet, error) {
	pods := impl.GetDCPods()
	result := []*corev1.Pod{}
	for _, pod := range pods {
		pvcs, err := impl.GetPodPVCs(pod)
		if err != nil {
			return nil, err
		}
		pvcs = utils.FilterPVCsWithFn(pvcs, func(pvc *corev1.PersistentVolumeClaim) bool {
			return pvc.Annotations != nil && pvc.Annotations[VolumeHealthAnnotation] == string(VolumeHealthInaccessible)
		})
		if len(pvcs) > 0 {
			result = append(result, pod)
		}
	}
	if rackName != "" {
		return utils.GetPodNameSet(filterPodsByRackName(result, rackName)), nil
	} else {
		return utils.GetPodNameSet(result), nil
	}
}
func (impl *EMMServiceImpl) getRacksWithNotReadyPodsBootstrapped() []string {
	pods := impl.GetNotReadyPodsBootstrappedInDC()
	rackNameSet := getPodsRackNameSet(pods)
	rackNames := []string{}
	for rackName, _ := range rackNameSet {
		rackNames = append(rackNames, rackName)
	}
	return rackNames
}
func (impl *EMMServiceImpl) getRackNodeNameSet(rackName string) utils.StringSet {
	podsForDownRack := utils.FilterPodsWithLabel(impl.GetDCPods(), api.RackLabel, rackName)
	nodeNameSetForRack := utils.GetPodNodeNameSet(podsForDownRack)
	return nodeNameSetForRack
}
func (impl *EMMServiceImpl) getInProgressNodeReplacements() []string {
	return impl.GetInProgressNodeReplacements()
}
func (impl *EMMServiceImpl) getNodesWithTaintKeyValueEffect(taintKey, value string, effect corev1.TaintEffect) ([]*corev1.Node, error) {
	nodes, err := impl.GetAllNodesInDC()
	if err != nil {
		return nil, err
	}
	return utils.FilterNodesWithTaintKeyValueEffect(nodes, taintKey, value, effect), nil
}
func (impl *EMMServiceImpl) getPodsForNodeName(nodeName string) []*corev1.Pod {
	return utils.FilterPodsWithNodeInNameSet(impl.GetDCPods(), utils.StringSet{nodeName: true})
}
func (impl *EMMServiceImpl) getPodsWithAnnotationKey(key string) []*corev1.Pod {
	pods := impl.GetDCPods()
	return utils.FilterPodsWithAnnotationKey(pods, key)
}
func (impl *EMMServiceImpl) addPodAnnotation(pod *corev1.Pod, key, value string) (bool, error) {
	if pod.ObjectMeta.Annotations == nil {
		pod.ObjectMeta.Annotations = map[string]string{}
	} else if currentValue, found := pod.Annotations[key]; found {
		return currentValue == value, nil
	}
	pod.Annotations[key] = value
	return true, impl.UpdatePod(pod)
}
func (impl *EMMServiceImpl) removePodAnnotation(pod *corev1.Pod, key string) error {
	if pod.ObjectMeta.Annotations != nil {
		delete(pod.ObjectMeta.Annotations, key)
		return impl.UpdatePod(pod)
	}
	return nil
}