refactor: share node-eligibility, pod-status and drive-ordering helpers (OP-345)

internal/config/env.go

@@ -461,15 +461,7 @@ func init() {
 	Consts.NewContainersLimit = 1000 // virtually no limit for now
 	Consts.PeriodicDrivesCheckInterval = 1 * time.Minute
 	Consts.CheckDriversInterval = 7 * time.Minute
-	// Default minimum drive/compute containers required to form a cluster. The 5-container default
-	// suits production 3+2+1 (minFdNum=6); a single-parity 2+1 cluster (minFdNum=3) legitimately
-	// forms with as few as 3, so AllowSingleParity lowers the default. Both remain env-overridable.
-	formClusterMinDefault := 5
-	if getBoolEnvOrDefault("ALLOW_SINGLE_PARITY", false) {
-		formClusterMinDefault = 3
-	}
-	Consts.FormClusterMinComputeContainers = getIntEnvOrDefault("FORM_CLUSTER_MIN_COMPUTE_CONTAINERS", formClusterMinDefault)
-	Consts.FormClusterMinDriveContainers = getIntEnvOrDefault("FORM_CLUSTER_MIN_DRIVE_CONTAINERS", formClusterMinDefault)
+	loadFormClusterMinContainers()
 	Consts.FormClusterMaxComputeContainers = 10
 	Consts.FormClusterMaxDriveContainers = 10
 	Consts.FormS3ClusterMaxContainerCount = 3
@@ -487,6 +479,18 @@ func init() {
 	Consts.SsdProxyDpdkMemoryMiB = 2048
 }
 
+func loadFormClusterMinContainers() {
+	// The 5-container default suits production 3+2+1 (minFdNum=6); a single-parity 2+1 cluster
+	// (minFdNum=3) legitimately forms with as few as 3, so AllowSingleParity lowers the default.
+	// Both remain env-overridable.
+	formClusterMinDefault := 5
+	if getBoolEnvOrDefault("ALLOW_SINGLE_PARITY", false) {
+		formClusterMinDefault = 3
+	}
+	Consts.FormClusterMinComputeContainers = getIntEnvOrDefault("FORM_CLUSTER_MIN_COMPUTE_CONTAINERS", formClusterMinDefault)
+	Consts.FormClusterMinDriveContainers = getIntEnvOrDefault("FORM_CLUSTER_MIN_DRIVE_CONTAINERS", formClusterMinDefault)
+}
+
 // LoadCapacityEnv populates the drive-sharing, cluster-capacity and compute-hugepages configuration
 // from environment variables, with the built-in defaults. It is the single source of these defaults,
 // shared by ConfigureEnv (the operator) and standalone callers such as the weka-capacity dry-run CLI,
@@ -519,6 +523,15 @@ func LoadCapacityEnv() {
 
 	// Compute hugepages cap
 	Config.ComputeMaxHugepagesMiB = getIntEnvOrDefault("COMPUTE_MAX_HUGEPAGES_MIB", 360000)
+
+	// Physical-CPU accounting: read here (not just ConfigureEnv) so standalone callers like the
+	// weka-capacity CLI, which never call ConfigureEnv, still get HT-aware core counting.
+	Config.FullPcpusOnly = getBoolEnvOrDefault("FULL_PCPUS_ONLY", false)
+
+	// Re-derive from whatever ALLOW_SINGLE_PARITY/FORM_CLUSTER_MIN_* the caller has set by now: standalone
+	// callers such as the weka-capacity CLI overlay the operator's env via os.Setenv and call only
+	// LoadCapacityEnv, long after this package's init() already ran against the CLI's own environment.
+	loadFormClusterMinContainers()
 }
 
 func ConfigureEnv(ctx context.Context) {
@@ -599,7 +612,6 @@ func ConfigureEnv(ctx context.Context) {
 	Config.DNSPolicy.HostNetwork = env.GetString("DNS_POLICY_HOST_NETWORK", "")
 	Config.SignDrivesImage = env.GetString("SIGN_DRIVES_IMAGE", "")
 	Config.TaskmonDefaultImage = env.GetString("TASKMON_DEFAULT_IMAGE", "")
-	Config.FullPcpusOnly = getBoolEnvOrDefault("FULL_PCPUS_ONLY", false)
 	Config.SkipUnhealthyToleration = getBoolEnvOrDefault("SKIP_UNHEALTHY_TOLERATION", false)
 	Config.SkipClientNoScheduleToleration = getBoolEnvOrDefault("SKIP_CLIENT_NO_SCHEDULE_TOLERATION", false)
 	Config.SkipAuxNoScheduleToleration = getBoolEnvOrDefault("SKIP_AUX_NO_SCHEDULE_TOLERATION", false)

internal/config/env_test.go

@@ -8,14 +8,14 @@ import (
 
 func TestEffectiveProtection(t *testing.T) {
 	tests := []struct {
-		name     string
-		cfg      config.DriveSharingConfig
-		specSW   int
-		specRL   int
-		specHS   int
-		wantSW   int
-		wantRL   int
-		wantHS   int
+		name   string
+		cfg    config.DriveSharingConfig
+		specSW int
+		specRL int
+		specHS int
+		wantSW int
+		wantRL int
+		wantHS int
 	}{
 		{
 			name:   "all-zero spec and all-zero defaults yields zero",
@@ -53,3 +53,27 @@ func TestEffectiveProtection(t *testing.T) {
 		})
 	}
 }
+
+// TestLoadCapacityEnv_RederivesFormClusterMinimumsAndFullPcpus reproduces the weka-capacity CLI's
+// startup order: it scrapes the operator's env and applies it via os.Setenv AFTER this package's
+// init() already ran against the CLI's own process environment, then calls only LoadCapacityEnv
+// (never ConfigureEnv/init() again). Both the ALLOW_SINGLE_PARITY-lowered form-cluster minimums and
+// FullPcpusOnly must therefore be re-derived by LoadCapacityEnv itself, not only by init()/ConfigureEnv.
+func TestLoadCapacityEnv_RederivesFormClusterMinimumsAndFullPcpus(t *testing.T) {
+	t.Setenv("ALLOW_SINGLE_PARITY", "true")
+	t.Setenv("FORM_CLUSTER_MIN_COMPUTE_CONTAINERS", "")
+	t.Setenv("FORM_CLUSTER_MIN_DRIVE_CONTAINERS", "")
+	t.Setenv("FULL_PCPUS_ONLY", "true")
+
+	config.LoadCapacityEnv()
+
+	if config.Consts.FormClusterMinComputeContainers != 3 {
+		t.Errorf("FormClusterMinComputeContainers = %d, want 3 (ALLOW_SINGLE_PARITY-lowered default)", config.Consts.FormClusterMinComputeContainers)
+	}
+	if config.Consts.FormClusterMinDriveContainers != 3 {
+		t.Errorf("FormClusterMinDriveContainers = %d, want 3 (ALLOW_SINGLE_PARITY-lowered default)", config.Consts.FormClusterMinDriveContainers)
+	}
+	if !config.Config.FullPcpusOnly {
+		t.Error("Config.FullPcpusOnly = false, want true: LoadCapacityEnv must read FULL_PCPUS_ONLY for CLI callers")
+	}
+}

internal/consts/consts.go

@@ -23,18 +23,17 @@ const WekaContainerName = "weka-container"
 
 // Node annotation keys for drive management
 const (
-	// AnnotationWekaDrives stores drive serial IDs for non-proxy mode.
-	// Format: ["SERIAL1", "SERIAL2", ...]
-	// Deprecated for writing: use AnnotationWekaFullDrives instead. Kept for backward compatibility reading.
+	// AnnotationWekaDrives stores drive serial IDs for non-proxy mode: ["SERIAL1", "SERIAL2", ...].
+	// Deprecated for writing: use AnnotationWekaFullDrives; kept for backward-compat reading.
 	AnnotationWekaDrives = "weka.io/weka-drives"
 
-	// AnnotationWekaFullDrives stores drive entries with full metadata (serial + capacity_gib) for non-proxy mode.
-	// Format: [{"serial":"SERIAL1","capacity_gib":14307},...]
-	// This supersedes AnnotationWekaDrives which is deprecated for writing but still supported for reading (fallback).
+	// AnnotationWekaFullDrives stores drive entries with full metadata for non-proxy mode:
+	// [{"serial":"SERIAL1","capacity_gib":14307},...]. Supersedes AnnotationWekaDrives (still read
+	// as fallback). TLC drives only: full-drives mode has no QLC accounting, so discovery excludes
+	// QLC drives here and every consumer charges these entries as TLC.
 	AnnotationWekaFullDrives = "weka.io/weka-full-drives"
 
-	// AnnotationBlockedDrives stores blocked drive serial IDs (non-proxy mode)
-	// Format: ["SERIAL1", "SERIAL2", ...]
+	// AnnotationBlockedDrives stores blocked drive serial IDs (non-proxy mode): ["SERIAL1", ...].
 	AnnotationBlockedDrives = "weka.io/blocked-drives"
 
 	// AnnotationSharedDrives stores shared drive information for proxy mode
@@ -62,8 +61,7 @@ const (
 	// Format: ["uuid1", "uuid2", ...]
 	AnnotationBlockedDrivesVirtualUuids = "weka.io/blocked-drives-virtual-uuids"
 
-	// AnnotationSignDrivesHash stores hash of signed drives to track changes
-	// Used to determine if drives need to be re-signed
+	// AnnotationSignDrivesHash stores a hash of signed drives, used to detect when re-signing is needed.
 	AnnotationSignDrivesHash = "weka.io/sign-drives-hash"
 )
 
@@ -77,13 +75,14 @@ const PodConfigCodeVersion = "1"
 
 // Kubernetes extended resource names
 const (
-	// ResourceDrives is the extended resource name for tracking available drives (non-proxy mode)
+	// ResourceDrives tracks available drives (non-proxy mode). TLC only: it counts the non-blocked
+	// entries of AnnotationWekaFullDrives, which excludes QLC.
 	ResourceDrives = "weka.io/drives"
 
-	// ResourceSharedDrivesCapacity is the extended resource name for tracking shared drive capacity (proxy mode)
+	// ResourceSharedDrivesCapacity tracks shared drive capacity (proxy mode).
 	ResourceSharedDrivesCapacity = "weka.io/shared-drives-capacity"
 
-	// ResourceSharedDrivesCapacityTLC is the extended resource name for tracking shared drive capacity of QLC drives (proxy mode)
+	// ResourcesSharedDrivesCapacityQLC tracks shared drive capacity of QLC drives (proxy mode).
 	ResourcesSharedDrivesCapacityQLC = "weka.io/shared-drives-capacity-qlc"
 
 	// WekaNumaRegionResourcePrefix is the extended resource name prefix for NUMA region confinement

internal/controllers/operations/enable_local_drivers_distribution.go

@@ -26,6 +26,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
+	"github.com/weka/weka-operator/internal/controllers/resources"
 	"github.com/weka/weka-operator/internal/drivers"
 	"github.com/weka/weka-operator/internal/services/discovery"
 	"github.com/weka/weka-operator/internal/services/kubernetes"
@@ -518,7 +519,7 @@ func (o *EnsureDistServiceOperation) DeleteIfNodeNotReady(ctx context.Context, c
 		return fmt.Errorf("failed to get node %s: %w", nodeName, err)
 	}
 
-	if NodeNotReady(node) {
+	if !resources.NodeIsReady(node) {
 		logger.Info("Node is not ready, deleting dist container", "container", wc.Name, "node", nodeName)
 		deleteErr := o.client.Delete(ctx, container)
 
@@ -906,15 +907,3 @@ func (o *EnsureDistServiceOperation) AsStep() lifecycle.Step {
 		Run:  AsRunFunc(o), // Assuming AsRunFunc helper exists
 	}
 }
-
-func NodeNotReady(node *corev1.Node) bool {
-	if node == nil {
-		return true // If node is nil, consider it not ready
-	}
-	for _, condition := range node.Status.Conditions {
-		if condition.Type == corev1.NodeReady && condition.Status != corev1.ConditionTrue {
-			return true // Node is not ready
-		}
-	}
-	return false // Node is ready
-}

internal/controllers/operations/prepull_utils.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/weka/weka-operator/internal/config"
 	"github.com/weka/weka-operator/internal/controllers/resources"
-	"github.com/weka/weka-operator/pkg/util"
 )
 
 const (
@@ -154,8 +153,8 @@ type PrePullStatusResult struct {
 	AllReady     bool
 }
 
-// GetTargetNodes returns nodes that match the given selectors and tolerations
-// Nodes must be Ready and not Unschedulable
+// GetTargetNodes returns nodes matching nodeSelector that can currently host a new weka pod: not
+// cordoned, Ready, and carrying no taint outside tolerations (resources.NodeIneligibleReason).
 func GetTargetNodes(ctx context.Context, c client.Client, nodeSelector map[string]string, tolerations []corev1.Toleration) ([]corev1.Node, error) {
 	nodeList := &corev1.NodeList{}
 	listOpts := []client.ListOption{}
@@ -169,19 +168,9 @@ func GetTargetNodes(ctx context.Context, c client.Client, nodeSelector map[strin
 
 	var targetNodes []corev1.Node
 	for i := range nodeList.Items {
-		// Skip unschedulable nodes
-		if nodeList.Items[i].Spec.Unschedulable {
+		if resources.NodeIneligibleReason(&nodeList.Items[i], tolerations) != "" {
 			continue
 		}
-		// Skip nodes that are not Ready
-		if NodeNotReady(&nodeList.Items[i]) {
-			continue
-		}
-		// Check if tolerations match node taints
-		if !util.CheckTolerations(nodeList.Items[i].Spec.Taints, tolerations, nil) {
-			continue
-		}
-
 		targetNodes = append(targetNodes, nodeList.Items[i])
 	}
 

internal/controllers/operations/prepull_utils_test.go

@@ -0,0 +1,54 @@
+package operations
+
+import (
+	"context"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func readyNode(name string) *corev1.Node {
+	return &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: name},
+		Status:     corev1.NodeStatus{Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}},
+	}
+}
+
+// TestGetTargetNodes_NodeEligibility exercises GetTargetNodes' node filtering, which now runs entirely
+// through resources.NodeIneligibleReason: cordoned and untolerated-taint nodes are excluded like before,
+// and a node reporting no NodeReady condition at all (not just NodeReady=False) is excluded too.
+func TestGetTargetNodes_NodeEligibility(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		t.Fatalf("AddToScheme: %v", err)
+	}
+
+	cordoned := readyNode("cordoned")
+	cordoned.Spec.Unschedulable = true
+
+	noReadyCondition := readyNode("no-ready-condition")
+	noReadyCondition.Status.Conditions = nil
+
+	notReady := readyNode("not-ready")
+	notReady.Status.Conditions = []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionFalse}}
+
+	eligible := readyNode("eligible")
+
+	fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cordoned, noReadyCondition, notReady, eligible).Build()
+
+	got, err := GetTargetNodes(context.Background(), fakeClient, nil, nil)
+	if err != nil {
+		t.Fatalf("GetTargetNodes: %v", err)
+	}
+
+	var names []string
+	for _, n := range got {
+		names = append(names, n.Name)
+	}
+	if len(names) != 1 || names[0] != "eligible" {
+		t.Errorf("GetTargetNodes returned %v, want only [eligible]", names)
+	}
+}

internal/controllers/resources/node.go

@@ -1,12 +1,18 @@
 package resources
 
-import v1 "k8s.io/api/core/v1"
+import (
+	v1 "k8s.io/api/core/v1"
 
+	"github.com/weka/weka-operator/pkg/util"
+)
+
+// NodeIsReady reports whether node carries a NodeReady=True condition. A node that has not reported a
+// NodeReady condition at all (nil node, or the condition simply absent) reads as NOT ready — it has not
+// yet told us it can run pods, so it is not a safe placement target.
 func NodeIsReady(node *v1.Node) bool {
 	if node == nil {
 		return false
 	}
-	// check if the node has a NodeReady condition set to True
 	isNodeReady := false
 	for _, condition := range node.Status.Conditions {
 		if condition.Type == v1.NodeReady && condition.Status == v1.ConditionTrue {
@@ -16,3 +22,23 @@ func NodeIsReady(node *v1.Node) bool {
 	}
 	return isNodeReady
 }
+
+// NodeIneligibleReason reports why a node cannot host a new weka pod right now — cordoned, not ready, or
+// carrying a taint outside tolerations — or "" when the node is a valid placement candidate. This is the
+// single predicate for "can this node receive a new pod": every caller across the operator and CLI that
+// needs this check goes through it, so the classifications can never quietly diverge between call sites.
+func NodeIneligibleReason(node *v1.Node, tolerations []v1.Toleration) string {
+	if node == nil {
+		return "not ready"
+	}
+	if node.Spec.Unschedulable {
+		return "cordoned"
+	}
+	if !NodeIsReady(node) {
+		return "not ready"
+	}
+	if !util.CheckTolerations(node.Spec.Taints, tolerations, nil) {
+		return "untolerated taint"
+	}
+	return ""
+}