Merge pull request #472 from dciabrin/seqno-push

Let pods report their state to the operator

Author: openshift-merge-bot[bot]

go.mod

@@ -11,7 +11,6 @@ require (
 	github.com/openstack-k8s-operators/lib-common/modules/common v0.6.1-0.20260515134210-2e2a0d06648c
 	github.com/openstack-k8s-operators/mariadb-operator/api v0.0.0-00010101000000-000000000000
 	go.uber.org/zap v1.28.0
-	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67
 	k8s.io/api v0.31.14
 	k8s.io/apimachinery v0.31.14
 	k8s.io/client-go v0.31.14
@@ -82,6 +81,7 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
+	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67 // indirect
 	golang.org/x/mod v0.32.0 // indirect
 	golang.org/x/net v0.49.0 // indirect
 	golang.org/x/oauth2 v0.30.0 // indirect

internal/controller/galera_controller.go

@@ -23,6 +23,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -51,8 +52,6 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/kubectl/pkg/util/podutils"
 
-	"golang.org/x/exp/maps"
-
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -110,12 +109,21 @@ func GetLog(ctx context.Context, controller string) logr.Logger {
 //
 
 // findBestCandidate returns the node with the lowest seqno
-func findBestCandidate(g *mariadbv1.Galera) (node string, found bool) {
-	sortednodes := maps.Keys(g.Status.Attributes)
-	sort.Strings(sortednodes)
+func findBestCandidate(g *mariadbv1.Galera, pods []corev1.Pod, log logr.Logger) (node string, found bool) {
+	log.Info("Known attributes for bootstrapping the cluster", "attributes", g.Status.Attributes)
+	var sortedNodes []string
+	for _, pod := range pods {
+		if cidFound, cid := getGaleraContainerID(&pod); cidFound {
+			attr, attrFound := g.Status.Attributes[pod.Name]
+			if attrFound && attr.ContainerID == cid {
+				sortedNodes = append(sortedNodes, pod.Name)
+			}
+		}
+	}
+	sort.Strings(sortedNodes)
 	bestnode := ""
 	bestseqno := -1
-	for _, node := range sortednodes {
+	for _, node := range sortedNodes {
 		// On clean shutdown, galera sets the last
 		// stopped node as 'safe to bootstrap', so use
 		// this hint when we can
@@ -131,7 +139,7 @@ func findBestCandidate(g *mariadbv1.Galera) (node string, found bool) {
 	}
 	// if we pass here, a candidate is only valid if we
 	// inspected all the expected replicas (e.g. typically 3)
-	if len(g.Status.Attributes) != int(*g.Spec.Replicas) {
+	if len(sortedNodes) != int(*g.Spec.Replicas) {
 		return "", false
 	}
 	return bestnode, true //"galera-0"
@@ -186,15 +194,42 @@ func getReadyPods(pods []corev1.Pod) (ret []corev1.Pod) {
 	return
 }
 
-// getRunningPodsMissingAttributes returns all the pods for which the operator
+// getWaitingPodsNotUsingNewReportScript filters out pods which are not yet
+// using the new galera state report script (pre-push behaviour ONLY)
+func getPodsNotUsingNewReportScript(pods []corev1.Pod) (ret []corev1.Pod) {
+	for _, pod := range pods {
+		cmIdx := slices.IndexFunc(pod.Spec.Volumes, func(v corev1.Volume) bool {
+			return v.Name == "operator-scripts"
+		})
+		cm := *pod.Spec.Volumes[cmIdx].ConfigMap
+		scriptIdx := slices.IndexFunc(cm.Items, func(k corev1.KeyToPath) bool {
+			return k.Key == "report_local_galera_state.py"
+		})
+		if scriptIdx == -1 {
+			ret = append(ret, pod)
+		}
+	}
+	return
+}
+
+// getWaitingPodsMissingAttributes returns all the pods for which the operator
 // has no seqno information, and which are ready for being inspected.
 // Note: a pod is considered 'ready for inspection' when its main container is
 // started and its inner process is currently waiting for a gcomm URI
 // (i.e. it is not running mysqld)
-func getRunningPodsMissingAttributes(ctx context.Context, pods []corev1.Pod, instance *mariadbv1.Galera, h *helper.Helper, config *rest.Config) (ret []corev1.Pod) {
+func getWaitingPodsMissingAttributes(ctx context.Context, pods []corev1.Pod, instance *mariadbv1.Galera, h *helper.Helper, config *rest.Config) (ret []corev1.Pod) {
 	for _, pod := range pods {
 		if pod.Status.Phase == corev1.PodRunning && !podutils.IsPodReady(&pod) {
 			_, attrFound := instance.Status.Attributes[pod.Name]
+			if attrFound {
+				cidFound, cid := getGaleraContainerID(&pod)
+				if !cidFound || cid != instance.Status.Attributes[pod.Name].ContainerID {
+					// The info we have for this pod came from an old container,
+					// so drop it and mark this pod as missing attributes
+					clearPodAttributes(ctx, instance, pod.Name)
+					attrFound = false
+				}
+			}
 			if !attrFound && isGaleraContainerStartedAndWaiting(ctx, &pod, instance, h, config) {
 				ret = append(ret, pod)
 			}
@@ -203,12 +238,28 @@ func getRunningPodsMissingAttributes(ctx context.Context, pods []corev1.Pod, ins
 	return
 }
 
-// getRunningPodsMissingGcomm returns all the pods which are not running galera
+// retrieveAttributesForPods fetches galera database status from pods which are not
+// not yet using the new galera state report script (pre-push behaviour ONLY)
+func retrieveAttributesForPods(ctx context.Context, pods []corev1.Pod, instance *mariadbv1.Galera, h *helper.Helper, config *rest.Config) {
+	for _, pod := range pods {
+		name := pod.Name
+		util.LogForObject(h, fmt.Sprintf("Pod %s running, retrieve seqno", name), instance)
+		warn, err := retrieveSequenceNumber(ctx, h, config, instance, &pod)
+		if len(warn) > 0 {
+			util.LogForObject(h, fmt.Sprintf("Warning: %q", warn), instance)
+		}
+		if err != nil {
+			util.LogForObject(h, fmt.Sprintf("Failed to retrieve seqno: %q", err), instance)
+		}
+	}
+}
+
+// getPodsWaitingForGcomm returns all the pods which are not running galera
 // yet but are ready to join the cluster.
-// Note: a pod is considered 'ready to join' when its main container is
+// NOTE: a pod is considered 'ready to join' when its main container is
 // started and its inner process is currently waiting for a gcomm URI
 // (i.e. it is not running mysqld)
-func getRunningPodsMissingGcomm(ctx context.Context, pods []corev1.Pod, instance *mariadbv1.Galera, h *helper.Helper, config *rest.Config) (ret []corev1.Pod) {
+func getPodsWaitingForGcomm(ctx context.Context, pods []corev1.Pod, instance *mariadbv1.Galera, h *helper.Helper, config *rest.Config) (ret []corev1.Pod) {
 	for _, pod := range pods {
 		if pod.Status.Phase == corev1.PodRunning && !podutils.IsPodReady(&pod) &&
 			isGaleraContainerStartedAndWaiting(ctx, &pod, instance, h, config) {
@@ -259,10 +310,18 @@ func injectGcommURI(ctx context.Context, h *helper.Helper, config *rest.Config,
 		func(_ *bytes.Buffer, _ *bytes.Buffer) error {
 			attr := instance.Status.Attributes[pod.Name]
 			attr.Gcomm = uri
-			attr.ContainerID = pod.Status.ContainerStatuses[0].ContainerID
 			instance.Status.Attributes[pod.Name] = attr
 			return nil
 		})
+	// If we could not push the file in the pod, this might be due
+	// to a transient connection error. Return an error, so that
+	// this reconcile event gets reprocessed, to give us a chance
+	// to re-inject the URI into the pod.
+	//
+	// NOTE: if the error was due to a pod being restarted, the next
+	// processing of this reconcile event will automatically clean up
+	// the outdated attributes and wait for the newly restarted pod
+	// to publish up-to-date state before injecting any URI.
 	return err
 }
 
@@ -279,15 +338,19 @@ func retrieveSequenceNumber(ctx context.Context, helper *helper.Helper, config *
 			if stderr.Len() > 0 {
 				errStr = strings.Split(strings.TrimSuffix(stderr.String(), "\n"), "\n")
 			}
+			// Remember the container from which the probe was extracted
+			attr.ContainerID = pod.Status.ContainerStatuses[0].ContainerID
 			instance.Status.Attributes[pod.Name] = attr
 			return nil
 		})
 	return
 }
 
 // clearPodAttributes clears information known by the operator about a pod
-func clearPodAttributes(instance *mariadbv1.Galera, podName string) {
+func clearPodAttributes(ctx context.Context, instance *mariadbv1.Galera, podName string) {
 	delete(instance.Status.Attributes, podName)
+	log := GetLog(ctx, "galera")
+	log.Info("Clear tracked attributes for pod", "pod", podName, "instance", instance)
 	// If the pod was deemed safeToBootstrap, this state has to be reassessed
 	if instance.Status.SafeToBootstrap == podName {
 		instance.Status.SafeToBootstrap = ""
@@ -297,40 +360,14 @@ func clearPodAttributes(instance *mariadbv1.Galera, podName string) {
 // clearOldPodsAttributesOnScaleDown removes known information from old pods
 // that no longer exist after a scale down of the galera CR
 func clearOldPodsAttributesOnScaleDown(ctx context.Context, instance *mariadbv1.Galera) {
-	log := GetLog(ctx, "galera")
 	replicas := int(*instance.Spec.Replicas)
 
 	// a pod's name is built as 'statefulsetname-n'
 	for node := range instance.Status.Attributes {
 		parts := strings.Split(node, "-")
 		index, _ := strconv.Atoi(parts[len(parts)-1])
 		if index >= replicas {
-			clearPodAttributes(instance, node)
-			log.Info("Remove old pod from status after scale-down", "instance", instance, "pod", node)
-		}
-	}
-}
-
-// assertPodsAttributesValidity compares the current state of the pods that are starting galera
-// against their known state in the CR's attributes. If a pod's attributes don't match its actual
-// state (i.e. it failed to start galera), the attributes are cleared from the CR's status
-func assertPodsAttributesValidity(helper *helper.Helper, instance *mariadbv1.Galera, pods []corev1.Pod) {
-	for _, pod := range pods {
-		_, found := instance.Status.Attributes[pod.Name]
-		if !found {
-			continue
-		}
-		// A node can have various attributes depending on its known state.
-		// A ContainerID attribute is only present if the node is being started.
-		attrCID := instance.Status.Attributes[pod.Name].ContainerID
-		containerFound, podCID := getGaleraContainerID(&pod)
-		if !containerFound || (attrCID != "" && attrCID != podCID) {
-			// This gcomm URI was pushed in a pod which was restarted
-			// before the attribute got cleared, which means the pod
-			// failed to start galera. Clear the attribute here, and
-			// reprobe the pod's state in the next reconcile loop
-			clearPodAttributes(instance, pod.Name)
-			util.LogForObject(helper, "Pod restarted while galera was starting", instance, "pod", pod.Name, "recorded ID", attrCID)
+			clearPodAttributes(ctx, instance, node)
 		}
 	}
 }
@@ -502,6 +539,11 @@ func (r *GaleraReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res
 			Resources: []string{"galeras"},
 			Verbs:     []string{"get", "list"},
 		},
+		{
+			APIGroups: []string{"mariadb.openstack.org"},
+			Resources: []string{"galeras/status"},
+			Verbs:     []string{"get", "list", "update", "patch"},
+		},
 		{
 			APIGroups: []string{"mariadb.openstack.org"},
 			Resources: []string{"mariadbaccounts"},
@@ -867,8 +909,6 @@ func (r *GaleraReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res
 		return sfres, sferr
 	}
 
-	// util.LogForObject(helper, fmt.Sprintf("DAM BEFORE %v - AFTER %v", helper.GetBefore(), helper.GetAfter()), instance)
-
 	statefulset := commonstatefulset.GetStatefulSet()
 
 	// If a full cluster restart was requested,
@@ -892,6 +932,12 @@ func (r *GaleraReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res
 		return ctrl.Result{}, err
 	}
 
+	// NOTE: Pre-push behaviour ONLY
+	// If we're doing a rolling update to the new way of probing galera,
+	// there might still be old pods whose probing must be managed by the operator.
+	// keep track of those pods for later
+	oldPods := getPodsNotUsingNewReportScript(podList.Items)
+	oldWaitingPods := getWaitingPodsMissingAttributes(ctx, oldPods, instance, helper, r.config)
 	//
 	// Reconstruct the state of the galera resource based on the replicaset and its pods
 	//
@@ -901,9 +947,6 @@ func (r *GaleraReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res
 		clearOldPodsAttributesOnScaleDown(ctx, instance)
 	}
 
-	// Ensure that all the ongoing galera start actions are still running
-	assertPodsAttributesValidity(helper, instance, podList.Items)
-
 	// Note:
 	//   . A pod is available in the statefulset if the pod's readiness
 	//     probe returns true (i.e. galera is running in the pod and clustered)
@@ -920,20 +963,14 @@ func (r *GaleraReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res
 			name := pod.Name
 			if _, found := instance.Status.Attributes[name]; found {
 				log.Info("Galera started", "pod", name)
-				clearPodAttributes(instance, name)
+				clearPodAttributes(ctx, instance, name)
 			}
 		}
 
-		runningPods := getRunningPodsMissingGcomm(ctx, podList.Items, instance, helper, r.config)
-		// Special case for 1-node deployment: if the statefulset reports 1 node is available
-		// but the pod shows up in runningPods (i.e. NotReady), do not consider it a joiner.
-		// Wait for the two statuses to re-sync after another k8s probe is run.
-		if *instance.Spec.Replicas == 1 && len(runningPods) == 1 {
-			log.Info("Galera node no longer running. Requeuing")
-			return ctrl.Result{RequeueAfter: time.Duration(3) * time.Second}, nil
-		}
-
-		// The other 'Running' pods can join the existing cluster.
+		// The remaining pods will become joiner nodes. Wait until they have
+		// reported their status to the operator, and then configure them
+		// as joiners.
+		runningPods := getPodsWaitingForGcomm(ctx, podList.Items, instance, helper, r.config)
 		for _, pod := range runningPods {
 			name := pod.Name
 			joinerURI := buildGcommURI(instance)
@@ -942,51 +979,33 @@ func (r *GaleraReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res
 			err := injectGcommURI(ctx, helper, r.config, instance, &pod, joinerURI)
 			if err != nil {
 				log.Error(err, "Failed to push gcomm URI", "pod", name)
-				// A failed injection likely means the pod's status has changed.
-				// drop it from status and reprobe it in another reconcile loop
-				clearPodAttributes(instance, name)
+				// Force an error here to retry this reconcile, and try
+				// another injection (see details in injectGcommURI())
 				return ctrl.Result{}, err
 			}
 		}
 	}
 
-	// If the cluster is not running, probe the available pods for seqno
-	// to determine the bootstrap node.
+	// If the cluster is not running, wait until pods advertise their state
+	// in the Galera CR, and determine which node to bootstrap from
 	// Note:
 	//   . a pod whose phase is Running but who is not Ready hasn't started
 	//     galera, it is waiting for the operator's instructions.
-	//     We can record its galera's seqno in our status.
-	//   . any other status means the the pod is starting/restarting. We can't
-	//     exec into the pod yet, so we will probe it in another reconcile loop.
 	if !instance.Status.Bootstrapped && !isBootstrapInProgress(instance) {
 		var node string
 		found := false
-		for _, pod := range getRunningPodsMissingAttributes(ctx, podList.Items, instance, helper, r.config) {
-			name := pod.Name
-			util.LogForObject(helper, fmt.Sprintf("Pod %s running, retrieve seqno", name), instance)
-			warn, err := retrieveSequenceNumber(ctx, helper, r.config, instance, &pod)
-			if len(warn) > 0 {
-				util.LogForObject(helper, fmt.Sprintf("Warning: %q", warn), instance)
-			}
-			if err != nil {
-				log.Error(err, fmt.Sprintf("Failed to retrieve seqno for %s", name))
-				return ctrl.Result{}, err
-			}
-			log.Info(fmt.Sprintf("Attributes retrieved for %s", name),
-				"UUID", instance.Status.Attributes[name].UUID,
-				"Seqno", instance.Status.Attributes[name].Seqno,
-				"SafeToBootstrap", instance.Status.Attributes[name].SafeToBootstrap,
-			)
-			if instance.Status.Attributes[name].SafeToBootstrap {
-				node = name
-				found = true
-				break
-			}
+
+		// NOTE: Pre-push behaviour ONLY
+		// If old pods are still running, the operator must probe their status.
+		if len(oldWaitingPods) > 0 {
+			retrieveAttributesForPods(ctx, oldWaitingPods, instance, helper, r.config)
 		}
 
 		// Check if we have enough info to bootstrap the cluster now
-		if !found && int(*instance.Spec.Replicas) > 0 {
-			node, found = findBestCandidate(instance)
+		// (either one of the node reported itself as `safe_to_bootstrap`
+		// or all the nodes have reported their state and we'll determine one candidate)
+		if int(*instance.Spec.Replicas) > 0 {
+			node, found = findBestCandidate(instance, podList.Items, log)
 		}
 		if found {
 			pod := getPodFromName(podList.Items, node)
@@ -995,22 +1014,25 @@ func (r *GaleraReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res
 			err := injectGcommURI(ctx, helper, r.config, instance, pod, "gcomm://")
 			if err != nil {
 				log.Error(err, "Failed to push gcomm URI", "pod", node)
-				// A failed injection likely means the pod's status has changed.
-				// drop it from status and reprobe it in another reconcile loop
-				clearPodAttributes(instance, node)
+				// Force an error here to retry this reconcile, and try
+				// another injection (see details in injectGcommURI())
 				return ctrl.Result{}, err
 			}
 		}
 	}
 
-	// The statefulset usually instantiates the pods instantly, and the galera
-	// operator doesn't receive individual events for pod's phase transition or
-	// readiness, as it is not controlling the pods (the statefulset is).
-	// So until all pods become available, we have to requeue this event to get
-	// a chance to react to all pod's transitions.
+	// Stop here if we don't have all galera pods running yet. Another
+	// reconciliation will be triggered if either:
+	//   - a pod (re)starts and push its local state to the galera CR
+	//   - A pod becomes ready, which bumps the AvailableReplicas in the StatefulSet
 	if statefulset.Status.AvailableReplicas != statefulset.Status.Replicas {
-		log.Info("Requeuing until all replicas are available")
-		return ctrl.Result{RequeueAfter: time.Duration(3) * time.Second}, nil
+		// NOTE: Pre-push behaviour ONLY
+		// If old pods are still running, requeue until all pods are available
+		if len(oldPods) > 0 {
+			log.Info("Requeuing until all replicas are available")
+			return ctrl.Result{RequeueAfter: time.Duration(3) * time.Second}, nil
+		}
+		return ctrl.Result{}, nil
 	}
 
 	// We reached the end of the Reconcile, update the Ready condition based on