feat: observability into cluster scheduling state

api/v1alpha1/valkeycluster_types.go

@@ -417,8 +417,28 @@ const (
 	// considers risky, for example a terminationGracePeriodSeconds too short for
 	// graceful failover.
 	ConditionConfigurationWarning = "ConfigurationWarning"
+	// ConditionSchedulingSatisfied reports whether every in-scope pod in the
+	// cluster is scheduled. It is False while any pod is Pending with the
+	// scheduler's Unschedulable reason. The operator reports the instantaneous
+	// state only; the "stuck for how long" threshold for alerting is left to
+	// the consumer's Prometheus rule (via its `for:` clause).
+	ConditionSchedulingSatisfied = "SchedulingSatisfied"
 )
 
+// ClusterConditionTypes lists the ValkeyCluster condition types exported by the
+// valkey_operator_cluster_condition metric. Keep in sync with the constants
+// above; a type missing from this list is still exported once present in
+// status.conditions, but its series are not pre-created at zero.
+var ClusterConditionTypes = []string{
+	ConditionReady,
+	ConditionProgressing,
+	ConditionDegraded,
+	ConditionClusterFormed,
+	ConditionSlotsAssigned,
+	ConditionConfigurationWarning,
+	ConditionSchedulingSatisfied,
+}
+
 const (
 	// Common reasons for conditions
 	ReasonInitializing             = "Initializing"
@@ -446,6 +466,8 @@ const (
 	ReasonSystemUsersAclError      = "SystemUsersACLError"
 	ReasonPodDisruptionBudgetError = "PodDisruptionBudgetError"
 	ReasonPodUnschedulable         = "PodUnschedulable"
+	ReasonAllPodsScheduled         = "AllPodsScheduled"
+	ReasonPodsPendingScheduling    = "PodsPendingScheduling"
 )
 
 // +kubebuilder:object:root=true

docs/status-conditions.md

@@ -94,6 +94,22 @@ Common reasons:
 - `RebalanceFailed` – slot rebalancing failed (scale-out or scale-in)
 - `PodUnschedulable` – Kubernetes scheduler cannot place one or more Valkey pods, for example because strict topology spread constraints cannot be satisfied
 
+#### `SchedulingSatisfied`
+Indicates whether every in-scope pod in the cluster is currently schedulable.
+
+| Status | Meaning |
+|---|---|
+| `True` | All in-scope pods are scheduled. |
+| `False` | At least one pod is Pending because the Kubernetes scheduler reported it unschedulable. The message names the pod and the scheduler's detail. |
+
+Reasons:
+- `AllPodsScheduled` – every in-scope pod is scheduled (status `True`)
+- `PodsPendingScheduling` – at least one pod is unschedulable (status `False`)
+
+The operator reports the instantaneous state only — it does not wait out a threshold. Decide how long "stuck" is too long in your Prometheus alert's `for:` clause against the [`valkey_operator_cluster_condition`](valkeycluster.md#operator-metrics) metric (`type="SchedulingSatisfied"`).
+
+This overlaps with `Degraded`'s and `Ready`'s `PodUnschedulable` reason (all surface the same scheduler signal); `SchedulingSatisfied` is the dedicated, alertable surface.
+
 ---
 
 ### Valkey-specific conditions

docs/valkeycluster.md

@@ -82,6 +82,24 @@ exporter:
   enabled: false
 ```
 
+#### Operator metrics
+
+Separate from the per-pod exporter above, the operator exposes its own controller metrics on its `/metrics` endpoint. Among them:
+
+- `valkey_operator_cluster_condition{valkey_cluster, target_namespace, type, status}` — a projection of the cluster's [status conditions](status-conditions.md), in the same shape as kube-state-metrics' `kube_pod_status_condition`. For each condition `type` there are three series (`status="true"`, `"false"`, `"unknown"`); the condition's current status reads `1` and the other two `0`. A condition the operator has not reported (for example on a brand-new cluster, or one whose reconcile fails before the condition's check runs) reads `0` on all three series.
+
+To alert on a condition, match its `status="false"` (or `"true"`, for abnormal-true conditions) series — this way an unreported condition raises no alert. The operator reports instantaneous state; choose your own "stuck for how long" threshold in the alert's `for:` clause. For example, for [`SchedulingSatisfied`](status-conditions.md#schedulingsatisfied):
+
+```yaml
+- alert: ValkeyClusterPodsUnschedulable
+  expr: valkey_operator_cluster_condition{type="SchedulingSatisfied", status="false"} == 1
+  for: 10m
+  labels:
+    severity: warning
+  annotations:
+    summary: "ValkeyCluster {{ $labels.valkey_cluster }} has pods that cannot be scheduled"
+```
+
 ### Persistence
 
 ```yaml

go.mod

@@ -46,6 +46,7 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect

internal/controller/metrics.go

@@ -17,8 +17,11 @@ limitations under the License.
 package controller
 
 import (
+	"strings"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/metrics"
 
 	valkeyiov1alpha1 "github.com/valkey-io/valkey-operator/api/v1alpha1"
@@ -71,8 +74,34 @@ var (
 		},
 		[]string{labelValkeyCluster, labelTargetNamespace},
 	)
+
+	clusterCondition = factory.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "valkey_operator_cluster_condition",
+			Help: "Status of a ValkeyCluster condition. 1 for the condition's current status (true/false/unknown), 0 for the others; all zeros when the condition is not reported.",
+		},
+		[]string{labelValkeyCluster, labelTargetNamespace, "type", "status"},
+	)
 )
 
+var conditionStatuses = []metav1.ConditionStatus{
+	metav1.ConditionTrue,
+	metav1.ConditionFalse,
+	metav1.ConditionUnknown,
+}
+
+// setConditionSeries sets the three status series for one condition type.
+// A nil cond (not reported) leaves all three at 0.
+func setConditionSeries(name, namespace, condType string, cond *metav1.Condition) {
+	for _, s := range conditionStatuses {
+		val := float64(0)
+		if cond != nil && cond.Status == s {
+			val = 1
+		}
+		clusterCondition.WithLabelValues(name, namespace, condType, strings.ToLower(string(s))).Set(val)
+	}
+}
+
 // initClusterMetrics creates empty metrics for a valkey cluster
 func initClusterMetrics(name, namespace string) {
 	for _, s := range valkeyiov1alpha1.ClusterStates {
@@ -85,6 +114,10 @@ func initClusterMetrics(name, namespace string) {
 	clusterShards.WithLabelValues(name, namespace)
 	clusterShardsReady.WithLabelValues(name, namespace)
 	slotMigrationBatchesTotal.WithLabelValues(name, namespace)
+
+	for _, condType := range valkeyiov1alpha1.ClusterConditionTypes {
+		setConditionSeries(name, namespace, condType, nil)
+	}
 }
 
 // updateClusterMetrics sets the Prometheus gauges for a ValkeyCluster.
@@ -103,6 +136,21 @@ func updateClusterMetrics(cluster *valkeyiov1alpha1.ValkeyCluster) {
 
 	clusterShards.WithLabelValues(name, ns).Set(float64(cluster.Status.Shards))
 	clusterShardsReady.WithLabelValues(name, ns).Set(float64(cluster.Status.ReadyShards))
+
+	// Export every condition, registered or not; a condition absent from
+	// status.conditions reads 0 on all three status series.
+	reported := make(map[string]*metav1.Condition, len(cluster.Status.Conditions))
+	for i := range cluster.Status.Conditions {
+		cond := &cluster.Status.Conditions[i]
+		reported[cond.Type] = cond
+	}
+	for _, condType := range valkeyiov1alpha1.ClusterConditionTypes {
+		setConditionSeries(name, ns, condType, reported[condType])
+		delete(reported, condType)
+	}
+	for condType, cond := range reported {
+		setConditionSeries(name, ns, condType, cond)
+	}
 }
 
 // deleteClusterMetrics removes all metrics for a deleted ValkeyCluster.
@@ -114,4 +162,5 @@ func deleteClusterMetrics(name, namespace string) {
 	clusterShardsReady.DeleteLabelValues(name, namespace)
 	failoversTotal.DeletePartialMatch(prometheus.Labels{labelValkeyCluster: name, labelTargetNamespace: namespace})
 	slotMigrationBatchesTotal.DeleteLabelValues(name, namespace)
+	clusterCondition.DeletePartialMatch(prometheus.Labels{labelValkeyCluster: name, labelTargetNamespace: namespace})
 }

internal/controller/metrics_test.go

@@ -0,0 +1,111 @@
+/*
+Copyright 2025 Valkey Contributors.
+
+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 (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	valkeyiov1alpha1 "github.com/valkey-io/valkey-operator/api/v1alpha1"
+)
+
+// testNamespace is the namespace used by all metrics tests.
+const testNamespace = "default"
+
+func conditionGaugeValue(t *testing.T, name, condType, status string) float64 {
+	t.Helper()
+	return testutil.ToFloat64(clusterCondition.WithLabelValues(name, testNamespace, condType, status))
+}
+
+// expectConditionSeries asserts the three status series for a condition type.
+func expectConditionSeries(t *testing.T, name, condType string, wantTrue, wantFalse, wantUnknown float64) {
+	t.Helper()
+	if got := conditionGaugeValue(t, name, condType, "true"); got != wantTrue {
+		t.Errorf("%s{status=true} = %v, want %v", condType, got, wantTrue)
+	}
+	if got := conditionGaugeValue(t, name, condType, "false"); got != wantFalse {
+		t.Errorf("%s{status=false} = %v, want %v", condType, got, wantFalse)
+	}
+	if got := conditionGaugeValue(t, name, condType, "unknown"); got != wantUnknown {
+		t.Errorf("%s{status=unknown} = %v, want %v", condType, got, wantUnknown)
+	}
+}
+
+func TestInitClusterMetrics_PreCreatesConditionSeries(t *testing.T) {
+	const name, ns = "metrics-cond-init-test", "default"
+	before := testutil.CollectAndCount(clusterCondition)
+	initClusterMetrics(name, ns)
+
+	want := before + len(valkeyiov1alpha1.ClusterConditionTypes)*3
+	if got := testutil.CollectAndCount(clusterCondition); got != want {
+		t.Fatalf("expected %d condition series after init, got %d", want, got)
+	}
+
+	deleteClusterMetrics(name, ns)
+	if got := testutil.CollectAndCount(clusterCondition); got != before {
+		t.Fatalf("expected %d condition series after delete, got %d", before, got)
+	}
+}
+
+func TestUpdateClusterMetrics_ClusterCondition(t *testing.T) {
+	const name, ns = "metrics-cond-test", "default"
+	const condType = valkeyiov1alpha1.ConditionSchedulingSatisfied
+	initClusterMetrics(name, ns)
+	defer deleteClusterMetrics(name, ns)
+
+	cluster := &valkeyiov1alpha1.ValkeyCluster{}
+	cluster.Name = name
+	cluster.Namespace = ns
+
+	// Condition absent -> all three status series read 0.
+	updateClusterMetrics(cluster)
+	expectConditionSeries(t, name, condType, 0, 0, 0)
+
+	setCondition(cluster, condType, valkeyiov1alpha1.ReasonAllPodsScheduled, "ok", metav1.ConditionTrue)
+	updateClusterMetrics(cluster)
+	expectConditionSeries(t, name, condType, 1, 0, 0)
+
+	setCondition(cluster, condType, valkeyiov1alpha1.ReasonPodsPendingScheduling, "pending", metav1.ConditionFalse)
+	updateClusterMetrics(cluster)
+	expectConditionSeries(t, name, condType, 0, 1, 0)
+
+	setCondition(cluster, condType, valkeyiov1alpha1.ReasonPodsPendingScheduling, "unknown", metav1.ConditionUnknown)
+	updateClusterMetrics(cluster)
+	expectConditionSeries(t, name, condType, 0, 0, 1)
+
+	// Condition removed again -> back to all zeros.
+	cluster.Status.Conditions = nil
+	updateClusterMetrics(cluster)
+	expectConditionSeries(t, name, condType, 0, 0, 0)
+}
+
+func TestUpdateClusterMetrics_UnregisteredConditionType(t *testing.T) {
+	const name, ns = "metrics-cond-unregistered-test", "default"
+	const condType = "NotInClusterConditionTypes"
+	initClusterMetrics(name, ns)
+	defer deleteClusterMetrics(name, ns)
+
+	cluster := &valkeyiov1alpha1.ValkeyCluster{}
+	cluster.Name = name
+	cluster.Namespace = ns
+	setCondition(cluster, condType, "SomeReason", "some message", metav1.ConditionTrue)
+
+	updateClusterMetrics(cluster)
+	expectConditionSeries(t, name, condType, 1, 0, 0)
+}

internal/controller/valkeycluster_controller.go

@@ -415,6 +415,7 @@ func (r *ValkeyClusterReconciler) handlePodSchedulingIssues(ctx context.Context,
 	if issue == nil {
 		removeConditionIfReason(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonPodUnschedulable)
 		removeConditionIfReason(&cluster.Status.Conditions, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonPodUnschedulable)
+		setCondition(cluster, valkeyiov1alpha1.ConditionSchedulingSatisfied, valkeyiov1alpha1.ReasonAllPodsScheduled, "All pods are scheduled", metav1.ConditionTrue)
 		return ctrl.Result{}, false, nil
 	}
 
@@ -426,6 +427,7 @@ func (r *ValkeyClusterReconciler) handlePodSchedulingIssues(ctx context.Context,
 	setCondition(cluster, valkeyiov1alpha1.ConditionDegraded, valkeyiov1alpha1.ReasonPodUnschedulable, message, metav1.ConditionTrue)
 	setCondition(cluster, valkeyiov1alpha1.ConditionReady, valkeyiov1alpha1.ReasonPodUnschedulable, message, metav1.ConditionFalse)
 	setCondition(cluster, valkeyiov1alpha1.ConditionProgressing, valkeyiov1alpha1.ReasonReconciling, "Waiting for unschedulable pods to be scheduled", metav1.ConditionTrue)
+	setCondition(cluster, valkeyiov1alpha1.ConditionSchedulingSatisfied, valkeyiov1alpha1.ReasonPodsPendingScheduling, message, metav1.ConditionFalse)
 	if err := r.updateStatus(ctx, cluster, nil); err != nil {
 		return ctrl.Result{}, false, err
 	}

internal/controller/valkeycluster_controller_test.go

@@ -361,6 +361,13 @@ var _ = Describe("pod scheduling issue handling", func() {
 		Expect(degraded.Status).To(Equal(metav1.ConditionTrue))
 		Expect(degraded.Reason).To(Equal(valkeyiov1alpha1.ReasonPodUnschedulable))
 
+		satisfied := testutils.FindCondition(updated.Status.Conditions, valkeyiov1alpha1.ConditionSchedulingSatisfied)
+		Expect(satisfied).NotTo(BeNil())
+		Expect(satisfied.Status).To(Equal(metav1.ConditionFalse))
+		Expect(satisfied.Reason).To(Equal(valkeyiov1alpha1.ReasonPodsPendingScheduling))
+		Expect(satisfied.Message).To(ContainSubstring("pod topology spread constraints not satisfied"))
+		Expect(satisfied.ObservedGeneration).To(Equal(updated.Generation))
+
 		events := collectEvents(fakeRecorder)
 		Expect(events).To(ContainElement(ContainSubstring("Warning")))
 		Expect(events).To(ContainElement(ContainSubstring(valkeyiov1alpha1.ReasonPodUnschedulable)))
@@ -394,6 +401,11 @@ var _ = Describe("pod scheduling issue handling", func() {
 		Expect(result).To(Equal(reconcile.Result{}))
 		Expect(testutils.FindCondition(cluster.Status.Conditions, valkeyiov1alpha1.ConditionReady)).To(BeNil())
 		Expect(testutils.FindCondition(cluster.Status.Conditions, valkeyiov1alpha1.ConditionDegraded)).To(BeNil())
+
+		satisfied := testutils.FindCondition(cluster.Status.Conditions, valkeyiov1alpha1.ConditionSchedulingSatisfied)
+		Expect(satisfied).NotTo(BeNil())
+		Expect(satisfied.Status).To(Equal(metav1.ConditionTrue))
+		Expect(satisfied.Reason).To(Equal(valkeyiov1alpha1.ReasonAllPodsScheduled))
 	})
 
 	It("ignores pods that are already scheduled", func() {

test/e2e/valkeycluster_test.go

@@ -1483,8 +1483,49 @@ spec:
 				g.Expect(degradedCond).NotTo(BeNil(), "Degraded condition not found")
 				g.Expect(degradedCond.Status).To(Equal(metav1.ConditionTrue))
 				g.Expect(degradedCond.Reason).To(Equal(valkeyiov1alpha1.ReasonPodUnschedulable))
+
+				satisfiedCond := utils.FindCondition(cr.Status.Conditions, valkeyiov1alpha1.ConditionSchedulingSatisfied)
+				g.Expect(satisfiedCond).NotTo(BeNil(), "SchedulingSatisfied condition not found")
+				g.Expect(satisfiedCond.Status).To(Equal(metav1.ConditionFalse))
+				g.Expect(satisfiedCond.Reason).To(Equal(valkeyiov1alpha1.ReasonPodsPendingScheduling))
 			}
 			Eventually(verifyUnschedulableStatus, 5*time.Minute, 2*time.Second).Should(Succeed())
+
+			By("labeling a second worker node so the spread constraint can be satisfied")
+			cmd = exec.Command("kubectl", "get", "nodes",
+				"--selector=!node-role.kubernetes.io/control-plane",
+				"-o", "jsonpath={.items[*].metadata.name}")
+			workerNodes, err := utils.Run(cmd)
+			Expect(err).NotTo(HaveOccurred(), fmt.Sprintf("Failed to list worker nodes: %s", workerNodes))
+			secondNode := ""
+			for _, node := range strings.Fields(workerNodes) {
+				if node != eligibleNode {
+					secondNode = node
+					break
+				}
+			}
+			Expect(secondNode).NotTo(BeEmpty(), "expected a second worker node")
+
+			cmd = exec.Command("kubectl", "label", "node", secondNode,
+				fmt.Sprintf("%s=%s", eligibleNodeLabelKey, eligibleNodeLabelValue), "--overwrite=true")
+			output, err = utils.Run(cmd)
+			Expect(err).NotTo(HaveOccurred(), fmt.Sprintf("Failed to label second worker node: %s", output))
+			defer func() {
+				cmd := exec.Command("kubectl", "label", "node", secondNode, eligibleNodeLabelKey+"-", "--overwrite=true")
+				_, _ = utils.Run(cmd)
+			}()
+
+			By("waiting for SchedulingSatisfied to recover to True")
+			verifyRecoveredStatus := func(g Gomega) {
+				cr, err := utils.GetValkeyClusterStatus(unschedulableClusterName)
+				g.Expect(err).NotTo(HaveOccurred())
+
+				satisfiedCond := utils.FindCondition(cr.Status.Conditions, valkeyiov1alpha1.ConditionSchedulingSatisfied)
+				g.Expect(satisfiedCond).NotTo(BeNil(), "SchedulingSatisfied condition not found")
+				g.Expect(satisfiedCond.Status).To(Equal(metav1.ConditionTrue))
+				g.Expect(satisfiedCond.Reason).To(Equal(valkeyiov1alpha1.ReasonAllPodsScheduled))
+			}
+			Eventually(verifyRecoveredStatus, 5*time.Minute, 2*time.Second).Should(Succeed())
 		})
 
 		It("spreads shard primaries across different nodes", func() {