Feat/add probs

config/crd/bases/keldon.io_databaseclusters.yaml

@@ -249,11 +249,13 @@ spec:
               recovery:
                 properties:
                   autoRebalance:
+                    default: true
                     description: |-
                       AutoRebalance enables automatic rebalancing after recovery
                       completes. Tri-state.
                     type: boolean
                   autoRecover:
+                    default: true
                     description: |-
                       AutoRecover enables automatic recovery of down segments via
                       gprecoverseg. Tri-state — see StandbySpec.Enabled for details.

config/rbac/role.yaml

@@ -4,6 +4,14 @@ kind: ClusterRole
 metadata:
   name: manager-role
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - nodes
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resources:

internal/controller/databasecluster_pods_starting.go

@@ -51,7 +51,7 @@ func (r *DatabaseClusterReconciler) reconcilePodsStarting(rctx *ReconcileContext
 	}
 
 	for _, role := range roles {
-		if err := r.ensurePodsisRunning(rctx.Ctx, cluster, role); err != nil {
+		if err := r.ensurePodsAreUp(rctx.Ctx, cluster, role); err != nil {
 			log.Info("pods not ready yet, requeuing",
 				"name", cluster.Name, "role", role)
 			return ctrl.Result{RequeueAfter: 10 * time.Second}, nil

internal/controller/databasecluster_probes.go

@@ -0,0 +1,96 @@
+/*
+Copyright 2026.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+
+	keldoniov1alpha1 "github.com/keldonio/keldon-operator/api/v1alpha1"
+)
+
+// addDatabaseProbes attaches startup, readiness, and liveness probes
+// to the database container.
+//
+// The probe is a single pg_isready check against the role's local port.
+// Port discovery:
+//
+//	coordinator, standby → fixed 5432
+//	segment              → 6000 + ordinal (derived from $HOSTNAME)
+//	mirror               → 7000 + ordinal (derived from $HOSTNAME)
+//
+// 3-probe layout:
+//
+//	StartupProbe   — generous boot window (~10 min) to absorb gpinitsystem
+//	                 on coord-0 and the wait-for-coordinator phase on
+//	                 segments/mirrors. Once it succeeds, readiness + liveness
+//	                 take over.
+//	ReadinessProbe — tight. Pod only appears in Service endpoints when
+//	                 postgres truly accepts connections.
+//	LivenessProbe  — moderate threshold. Hung postgres → pod restart, but
+//	                 not so aggressive that load spikes cause false restarts.
+//
+// All probes pass cleanly through StatefulSet rollouts and operator restarts
+// because they read state from the pod itself, not from operator status.
+func addDatabaseProbes(c *corev1.Container, suffix string, image *keldoniov1alpha1.DatabaseImage) {
+	var portExpr string
+	switch suffix {
+	case "coordinator", "standby":
+		portExpr = "5432"
+	case "segment", "segments":
+		portExpr = "6000"
+	case "mirror", "mirrors":
+		portExpr = "7000"
+	default:
+		portExpr = "5432"
+	}
+
+	cmd := []string{
+		"bash", "-c",
+		fmt.Sprintf("source %s && pg_isready -h localhost -p %s -U %s",
+			image.Spec.Paths.EnvScript, portExpr, image.Spec.AdminUser),
+	}
+
+	c.StartupProbe = &corev1.Probe{
+		ProbeHandler: corev1.ProbeHandler{
+			Exec: &corev1.ExecAction{Command: cmd},
+		},
+		InitialDelaySeconds: 30,
+		PeriodSeconds:       10,
+		TimeoutSeconds:      5,
+		FailureThreshold:    60, // 60 × 10s = 10-min boot window
+	}
+
+	c.ReadinessProbe = &corev1.Probe{
+		ProbeHandler: corev1.ProbeHandler{
+			Exec: &corev1.ExecAction{Command: cmd},
+		},
+		PeriodSeconds:    5,
+		TimeoutSeconds:   3,
+		FailureThreshold: 3,
+	}
+
+	c.LivenessProbe = &corev1.Probe{
+		ProbeHandler: corev1.ProbeHandler{
+			Exec: &corev1.ExecAction{Command: cmd},
+		},
+		PeriodSeconds:    10,
+		TimeoutSeconds:   5,
+		FailureThreshold: 6, // 60s before kill
+	}
+}

internal/controller/databasecluster_scaling.go

@@ -200,7 +200,7 @@
 			fmt.Sprintf("export COORDINATOR_DATA_DIRECTORY=%s && %s",
 				paths.CoordinatorDataDir, expandRollbackCmd),
 		)
 		r.execInPod(ctx, cluster, []string{"bash", "-c", rollbackCmd}, activeCoordinator)
 		return fmt.Errorf("gpexpand init failed: %w", err)
 	}
 
@@ -274,12 +274,12 @@
 	adminUser := rctx.Resolved.Image.Spec.AdminUser
 
 	// 1. Wait for new segment pods to be Ready.
-	if err := r.ensurePodsisRunning(ctx, cluster, "segment"); err != nil {
+	if err := r.ensurePodsAreUp(ctx, cluster, "segment"); err != nil {
 		log.Info("New segment pods not ready yet, requeuing", "name", cluster.Name)
 		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
 	}
 	if mirroringEnabled(spec) {
-		if err := r.ensurePodsisRunning(ctx, cluster, "mirror"); err != nil {
+		if err := r.ensurePodsAreUp(ctx, cluster, "mirror"); err != nil {
 			log.Info("New mirror pods not ready yet, requeuing", "name", cluster.Name)
 			return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
 		}

internal/controller/databasecluster_service.go

@@ -90,7 +90,7 @@
 				},
 			}
 			Service.Spec.Selector = desiredSelector
 			ctrl.SetControllerReference(cluster, &Service, r.Scheme)
 			return r.Create(ctx, &Service)
 		}
 		return err
@@ -133,6 +133,7 @@
 			headlessService.Name = headlessServiceNamespacedName.Name
 			headlessService.Namespace = headlessServiceNamespacedName.Namespace
 			headlessService.Spec.ClusterIP = "None"
+			headlessService.Spec.PublishNotReadyAddresses = true
 			headlessService.Spec.Selector = map[string]string{
 				"app": cluster.Name + "-" + suffix,
 			}
@@ -142,7 +143,7 @@
 					Name: "postgres",
 				},
 			}
 			ctrl.SetControllerReference(cluster, &headlessService, r.Scheme)
 			err = r.Create(ctx, &headlessService)
 			if err != nil {
 				return err

internal/controller/databasecluster_ssh_distribute.go

@@ -136,7 +136,7 @@ chmod 600 %s
 // name + role, and the class-supplied segment count and mirror/standby
 // flags are honored.
 //
-// Called after ensurePodsisRunning checks have all passed, so the pods
+// Called after ensurePodsAreUp checks have all passed, so the pods
 // definitely exist and are ready to accept exec calls.
 //
 // adminUser is read from rctx.Resolved.Image and threaded down to the

internal/controller/databasecluster_statefulset.go

@@ -30,6 +30,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
@@ -54,25 +55,49 @@ func (r *DatabaseClusterReconciler) updateStatefulsetReplicas(ctx context.Contex
 	return nil
 }
 
-// ensurePodsisRunning checks that all replicas of the given StatefulSet are ready.
-// Returns an error if any pods are not yet running — caller should requeue.
-func (r *DatabaseClusterReconciler) ensurePodsisRunning(ctx context.Context, cluster *keldonv1alpha1.DatabaseCluster, suffix string) error {
-	var statefulset appsv1.StatefulSet
-
-	err := r.Get(ctx, types.NamespacedName{
+// ensurePodsAreUp returns nil only when every pod in the named StatefulSet
+// has its containers running (Pod.Status.Phase == PodRunning).
+//
+// It does NOT require readiness probes to pass. Use this for bootstrap and
+// scaling — moments when we need "pod is alive and reachable over SSH", not
+// "postgres is accepting connections."
+//
+// For the latter (failover detection in PhaseRunning), use coordinatorPodReady.
+func (r *DatabaseClusterReconciler) ensurePodsAreUp(ctx context.Context, cluster *keldonv1alpha1.DatabaseCluster, suffix string) error {
+	var sts appsv1.StatefulSet
+	if err := r.Get(ctx, types.NamespacedName{
 		Name:      cluster.Name + "-" + suffix,
 		Namespace: cluster.Namespace,
-	}, &statefulset)
-	if err != nil {
+	}, &sts); err != nil {
 		return err
 	}
 
-	if statefulset.Spec.Replicas == nil || statefulset.Status.ReadyReplicas != *statefulset.Spec.Replicas {
-		return fmt.Errorf("statefulset %s is not ready yet", statefulset.Name)
+	expected := int32(1)
+	if sts.Spec.Replicas != nil {
+		expected = *sts.Spec.Replicas
 	}
 
-	return nil
+	var pods corev1.PodList
+	if err := r.List(ctx, &pods,
+		client.InNamespace(cluster.Namespace),
+		client.MatchingLabels{"app": sts.Name},
+	); err != nil {
+		return fmt.Errorf("listing pods for statefulset %s: %w", sts.Name, err)
+	}
 
+	if int32(len(pods.Items)) != expected {
+		return fmt.Errorf("statefulset %s has %d pods, expected %d",
+			sts.Name, len(pods.Items), expected)
+	}
+
+	for _, pod := range pods.Items {
+		if pod.Status.Phase != corev1.PodRunning {
+			return fmt.Errorf("pod %s is %s, expected Running",
+				pod.Name, pod.Status.Phase)
+		}
+	}
+
+	return nil
 }
 
 // ensureStatefulset creates or updates the StatefulSet for the given role.
@@ -153,6 +178,9 @@ func (r *DatabaseClusterReconciler) buildStatefulset(
 
 	statefulset.Name = cluster.Name + "-" + suffix
 	statefulset.Namespace = cluster.Namespace
+
+	statefulset.Spec.PodManagementPolicy = appsv1.ParallelPodManagement
+
 	statefulset.Spec.Selector = &metav1.LabelSelector{
 		MatchLabels: map[string]string{
 			"app": cluster.Name + "-" + suffix,
@@ -201,6 +229,8 @@ func (r *DatabaseClusterReconciler) buildStatefulset(
 		},
 	}
 
+	addDatabaseProbes(&statefulset.Spec.Template.Spec.Containers[0], suffix, image)
+
 	var storage resource.Quantity
 	var storageClassName *string