func (r *ReconcileKubeDirectorCluster) syncCluster(
	reqLogger logr.Logger,
	cr *kdv1.KubeDirectorCluster,
) error {
	hadFinalizer := shared.HasFinalizer(cr)
	oldStatus := cr.Status.DeepCopy()
	if cr.Status == nil {
		cr.Status = &kdv1.KubeDirectorClusterStatus{}
		cr.Status.Roles = make([]kdv1.RoleStatus, 0)
	}
	if cr.Status.SpecGenerationToProcess == nil {
		initSpecGen := int64(0)
		cr.Status.SpecGenerationToProcess = &initSpecGen
	}
	annotations := cr.Annotations
	if annotations == nil {
		annotations = make(map[string]string)
		cr.Annotations = annotations
	}
	var statusBackup *kdv1.KubeDirectorStatusBackup
	var statusBackupErr error
	var statusBackupExists bool
	var statusBackupShouldExist bool
	fetchBackup := func() {
		statusBackup, statusBackupErr = observer.GetStatusBackup(cr.Namespace, cr.Name)
		statusBackupExists = (statusBackupErr == nil)
		if !statusBackupExists {
			statusBackup = nil
		}
		statusBackupShouldExist = shared.GetBackupClusterStatus()
	}
	defer func() {
		fetchBackup()
		executor.UpdateClusterStatusBackupOwner(reqLogger, cr, statusBackup)
		syncMemberNotifies(reqLogger, cr)
		updateStateRollup(cr)
		nowHasFinalizer := shared.HasFinalizer(cr)
		statusChanged := false
		backupAnnotationNeedsReconcile := false
		if (cr.DeletionTimestamp == nil) || nowHasFinalizer {
			statusChanged = !equality.Semantic.DeepEqual(cr.Status, oldStatus)
			if !statusChanged {
				statusChanged = (statusBackupShouldExist != statusBackupExists)
			}
			backupAnnotationNeedsReconcile = executor.BackupAnnotationNeedsReconcile(
				reqLogger,
				cr,
				statusBackupShouldExist,
			)
		}
		finalizersChanged := (hadFinalizer != nowHasFinalizer)
		if !(statusChanged || backupAnnotationNeedsReconcile || finalizersChanged) {
			return
		}
		wait := time.Second
		maxWait := 4096 * time.Second
		for {
			var updateErr error
			if backupAnnotationNeedsReconcile {
				updateErr = executor.SetBackupAnnotation(
					reqLogger,
					cr,
					statusBackupShouldExist,
				)
				if updateErr == nil {
					backupAnnotationNeedsReconcile = false
				}
			}
			if (updateErr == nil) && statusChanged {
				cr.Status.GenerationUID = uuid.New().String()
				ClusterStatusGens.WriteStatusGen(cr.UID, cr.Status.GenerationUID)
				updateErr = executor.UpdateClusterStatus(
					cr,
					statusBackupShouldExist,
					statusBackup,
				)
				if updateErr == nil {
					statusChanged = false
				}
			}
			if (updateErr == nil) && finalizersChanged {
				crWithoutStatus := cr.DeepCopy()
				crWithoutStatus.Status = nil
				updateErr = shared.Update(context.TODO(), crWithoutStatus)
			}
			if updateErr == nil {
				return
			}
			currentCluster, currentClusterErr := observer.GetCluster(
				cr.Namespace,
				cr.Name,
			)
			if currentClusterErr != nil {
				if errors.IsNotFound(currentClusterErr) {
					return
				}
			} else {
				if currentCluster.DeletionTimestamp != nil {
					statusChanged = false
					backupAnnotationNeedsReconcile = false
				}
				if errors.IsConflict(updateErr) {
					fetchBackup()
					currentCluster.Status = cr.Status
					currentHasFinalizer := shared.HasFinalizer(currentCluster)
					if currentHasFinalizer {
						if !nowHasFinalizer {
							shared.RemoveFinalizer(currentCluster)
						}
					} else {
						if nowHasFinalizer {
							shared.EnsureFinalizer(currentCluster)
						}
					}
					*cr = *currentCluster
					if currentCluster.DeletionTimestamp != nil {
						backupAnnotationNeedsReconcile = executor.BackupAnnotationNeedsReconcile(
							reqLogger,
							cr,
							statusBackupShouldExist,
						)
					}
					continue
				}
			}
			if wait < maxWait {
				wait = wait * 2
			}
			shared.LogErrorf(
				reqLogger,
				updateErr,
				cr,
				shared.EventReasonCluster,
				"trying status update again in %v; failed",
				wait,
			)
			time.Sleep(wait)
		}
	}()
	cr.Status.RestoreProgress = nil
	currentHash := calcConnectionsHash(&cr.Spec.Connections, cr.Namespace)
	doExit, finalizerErr := r.handleFinalizers(reqLogger, cr)
	if finalizerErr != nil {
		return finalizerErr
	}
	if doExit {
		return nil
	}
	shouldProcessCR, processErr := r.handleNewCluster(reqLogger, cr)
	if processErr != nil {
		return processErr
	}
	if !shouldProcessCR {
		return nil
	}
	errLog := func(domain string, err error) {
		shared.LogErrorf(
			reqLogger,
			err,
			cr,
			shared.EventReasonCluster,
			"failed to sync %s: %v",
			domain,
			err,
		)
	}
	checkContainerStates(reqLogger, cr)
	clusterServiceErr := syncClusterService(reqLogger, cr)
	if clusterServiceErr != nil {
		errLog("cluster service", clusterServiceErr)
		return clusterServiceErr
	}
	roles, state, rolesErr := syncClusterRoles(reqLogger, cr)
	if rolesErr != nil {
		errLog("roles", rolesErr)
		return rolesErr
	}
	if state == clusterMembersChangedUnready || (currentHash != cr.Status.LastConnectionHash) {
		if currentHash != cr.Status.LastConnectionHash {
			annotations := cr.Annotations
			if hashVersion, ok := annotations[shared.HashChangeIncrementor]; ok {
				newV, _ := strconv.ParseInt(hashVersion, 10, 64)
				annotations[shared.HashChangeIncrementor] = strconv.FormatInt(newV+int64(1), 10)
			} else {
				annotations[shared.HashChangeIncrementor] = "1"
				shared.LogInfo(
					reqLogger,
					cr,
					shared.EventReasonCluster,
					"Annotation initialized to 1",
				)
			}
			cr.Annotations = annotations
			if shared.Update(context.TODO(), cr) == nil {
				shared.LogInfo(
					reqLogger,
					cr,
					shared.EventReasonCluster,
					"Updated context",
				)
			}
		}
		incremented := *cr.Status.SpecGenerationToProcess + int64(1)
		cr.Status.SpecGenerationToProcess = &incremented
		cr.Status.LastConnectionHash = currentHash
	}
	memberServicesErr := syncMemberServices(reqLogger, cr, roles)
	if memberServicesErr != nil {
		errLog("member services", memberServicesErr)
		return memberServicesErr
	}
	if state == clusterMembersStableReady {
		if cr.Status.State != string(clusterReady) {
			shared.LogInfo(
				reqLogger,
				cr,
				shared.EventReasonCluster,
				"stable",
			)
			amIBeingConnectedToThis := func(otherCluster kdv1.KubeDirectorCluster) bool {
				for _, connectedName := range otherCluster.Spec.Connections.Clusters {
					if cr.Name == connectedName {
						return true
					}
				}
				return false
			}
			allClusters := &kdv1.KubeDirectorClusterList{}
			shared.List(context.TODO(), allClusters)
			for _, kubecluster := range allClusters.Items {
				if amIBeingConnectedToThis(kubecluster) {
					shared.LogInfof(
						reqLogger,
						cr,
						shared.EventReasonCluster,
						"connected to cluster {%s}; updating it",
						kubecluster.Name,
					)
					shared.LogInfof(
						reqLogger,
						&kubecluster,
						shared.EventReasonCluster,
						"connected cluster {%s} has changed",
						cr.Name,
					)
					wait := time.Second
					maxWait := 4096 * time.Second
					for {
						updateMetaGenerator := &kubecluster
						annotations := updateMetaGenerator.Annotations
						if annotations == nil {
							annotations = make(map[string]string)
							updateMetaGenerator.Annotations = annotations
						}
						if v, ok := annotations[shared.ConnectionsIncrementor]; ok {
							newV, _ := strconv.Atoi(v)
							annotations[shared.ConnectionsIncrementor] = strconv.Itoa(newV + 1)
						} else {
							annotations[shared.ConnectionsIncrementor] = "1"
						}
						updateMetaGenerator.Annotations = annotations
						if shared.Update(context.TODO(), updateMetaGenerator) == nil {
							break
						}
						updateMetaGenerator, fetchErr := observer.GetCluster(kubecluster.Namespace, kubecluster.Name)
						if fetchErr != nil {
							if errors.IsNotFound(fetchErr) {
								break
							}
						}
						if wait > maxWait {
							return fmt.Errorf(
								"Unable to notify cluster {%s} of configmeta change",
								updateMetaGenerator.Name)
						}
						time.Sleep(wait)
						wait = wait * 2
					}
				}
			}
			cr.Status.State = string(clusterReady)
		}
		if currentHash == cr.Status.LastConnectionHash {
			return nil
		}
	}
	if cr.Status.State != string(clusterCreating) {
		cr.Status.State = string(clusterUpdating)
	}
	configmetaGen, configMetaErr := catalog.ConfigmetaGenerator(
		cr,
		calcMembersForRoles(roles),
	)
	if configMetaErr != nil {
		shared.LogError(
			reqLogger,
			configMetaErr,
			cr,
			shared.EventReasonCluster,
			"failed to generate cluster config",
		)
		return configMetaErr
	}
	membersErr := syncMembers(reqLogger, cr, roles, configmetaGen)
	if membersErr != nil {
		errLog("members", membersErr)
		return membersErr
	}
	return nil
}
syncCluster runs the reconciliation logic. It is invoked because of a
change in or addition of a KubeDirectorCluster instance, or a periodic
polling to check on such a resource.
package kubedirectorcluster
import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"strconv"
	"time"
	kdv1 "github.com/bluek8s/kubedirector/pkg/apis/kubedirector/v1beta1"
	"github.com/bluek8s/kubedirector/pkg/catalog"
	"github.com/bluek8s/kubedirector/pkg/executor"
	"github.com/bluek8s/kubedirector/pkg/observer"
	"github.com/bluek8s/kubedirector/pkg/shared"
	"github.com/go-logr/logr"
	"github.com/google/uuid"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/errors"
)
var (
	ClusterStatusGens = shared.NewStatusGens()
)