Add load-aware primary node placement webhook for volume creation

operator/cmd/main.go

@@ -42,6 +42,7 @@ import (
 	simplyblockv1alpha1 "github.com/simplyblock/simplyblock-operator/api/v1alpha1"
 	"github.com/simplyblock/simplyblock-operator/internal/controller"
 	"github.com/simplyblock/simplyblock-operator/internal/utils"
+	"github.com/simplyblock/simplyblock-operator/internal/volumemigration/autobalancing"
 	"github.com/simplyblock/simplyblock-operator/internal/webapi"
 	internalwebhook "github.com/simplyblock/simplyblock-operator/internal/webhook"
 	// +kubebuilder:scaffold:imports
@@ -418,6 +419,14 @@ func main() {
 		mgr.GetWebhookServer().Register("/mutate-v1-pod-simplyblock-rebalancer",
 			&webhook.Admission{Handler: &internalwebhook.SimplyblockRebalancerInjector{Client: mgr.GetClient()}})
 		setupLog.Info("registered simplyblock-rebalancer mutating webhook")
+
+		mgr.GetWebhookServer().Register("/mutate-v1-pvc-simplyblock-placement",
+			&webhook.Admission{Handler: &internalwebhook.SimplyblockVolumePlacementInjector{
+				Client:       mgr.GetClient(),
+				APIClient:    webapi.NewClient(),
+				NodeSelector: autobalancing.NewStorageNodeSelector(mgr.GetClient()),
+			}})
+		setupLog.Info("registered simplyblock-volume-placement mutating webhook")
 	}()
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {

operator/config/webhook/manifests.yaml

@@ -23,3 +23,22 @@ webhooks:
     resources:
     - pods
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /mutate-v1-pvc-simplyblock-placement
+  failurePolicy: Ignore
+  name: simplyblock-volume-placement-injector.simplyblock.io
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - persistentvolumeclaims
+  sideEffects: None

operator/dist/install.yaml

@@ -3446,3 +3446,22 @@ webhooks:
     resources:
     - pods
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: simplyblock-operator-webhook-service
+      namespace: simplyblock-operator-system
+      path: /mutate-v1-pvc-simplyblock-placement
+  failurePolicy: Ignore
+  name: simplyblock-volume-placement-injector.simplyblock.io
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - persistentvolumeclaims
+  sideEffects: None

operator/internal/volumemigration/autobalancing/storage_node_selector.go

@@ -114,7 +114,7 @@ func (sns *StorageNodeSelector) SelectStorageNodes(
 ) ([]NodeMigrationPair, error) {
 	// computeLatencyDeviations also emits the per-node / max deviation gauges as a side
 	// effect; the per-cluster stats it returns are not needed for pairing here.
-	deviations, _, err := sns.computeLatencyDeviations(ctx, cfg.PrometheusURL, cfg.LatencyPercentile, inputs...)
+	deviations, err := sns.computeLatencyDeviations(ctx, cfg.PrometheusURL, cfg.LatencyPercentile, inputs...)
 	if err != nil {
 		return nil, err
 	}
@@ -157,6 +157,41 @@ func (sns *StorageNodeSelector) SelectStorageNodes(
 	return pairs, nil
 }
 
+// SelectBestNode returns the least-loaded eligible node — the one with the lowest current
+// latency deviation — for primary volume placement at creation time. Unlike pickColdTarget
+// (migration-target selection), there is no source node to compare against and no
+// MinHotColdDifferencePct gate: placement always wants the single best candidate, however
+// small its lead over the second-best. A node absent from the current Prometheus reading
+// (e.g. freshly onboarded, not yet scraped) is treated as deviation 0 — the best possible
+// score — rather than excluded, so it isn't systematically avoided.
+func (sns *StorageNodeSelector) SelectBestNode(
+	ctx context.Context,
+	cfg RebalancingConfig,
+	eligible map[string]bool,
+	inputs ...StorageNodeSelectorInput,
+) (nodeUUID string, ok bool, err error) {
+	deviations, err := sns.computeLatencyDeviations(ctx, cfg.PrometheusURL, cfg.LatencyPercentile, inputs...)
+	if err != nil {
+		return "", false, err
+	}
+
+	var best string
+	var bestDev float64
+	found := false
+	for uuid, isEligible := range eligible {
+		if !isEligible {
+			continue
+		}
+		dev := deviations[uuid]
+		if !found || dev < bestDev {
+			best = uuid
+			bestDev = dev
+			found = true
+		}
+	}
+	return best, found, nil
+}
+
 // pickColdTarget returns the coolest eligible migration target for the hot source from
 // the flat node pool, or ok=false when none qualifies. A candidate qualifies when it is
 // eligible (isMigrationTargetEligible) and at least cfg.MinHotColdDifferencePct
@@ -191,29 +226,30 @@ func isMigrationTargetEligible(src, cand nodeRef, _ RebalancingConfig) bool {
 }
 
 // computeLatencyDeviations collects per-node latency from Prometheus and StorageNode CRs,
-// computes deviation from baseline, emits Prometheus gauges, and returns the per-node
-// deviation map and per-cluster aggregate statistics.
+// computes deviation from baseline, and emits Prometheus gauges (per-node deviation and
+// per-cluster max deviation) as a side effect. Only the per-node deviation map is returned —
+// no caller currently needs the per-cluster stats beyond the gauge they're emitted into.
 func (sns *StorageNodeSelector) computeLatencyDeviations(
 	ctx context.Context,
 	prometheusURL string,
 	percentile string,
 	inputs ...StorageNodeSelectorInput,
-) (deviations map[string]float64, statsByCluster map[string]ClusterDeviationStats, err error) {
+) (deviations map[string]float64, err error) {
 	latencyByNode, err := sns.collectLatencyState(ctx, prometheusURL, percentile, inputs...)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	deviations = make(map[string]float64, len(latencyByNode))
 	for nodeUUID, ld := range latencyByNode {
 		dev := volumemigration.ComputeLatencyDeviationPct(ld.baselineNS, ld.currentNS)
 		rebalancerNodeLatencyDeviationPct.WithLabelValues(ld.clusterUUID, nodeUUID).Set(dev)
 		deviations[nodeUUID] = dev
 	}
-	statsByCluster = deviationStats(latencyByNode, deviations)
+	statsByCluster := deviationStats(latencyByNode, deviations)
 	for clusterUUID, stats := range statsByCluster {
 		rebalancerMaxLatencyDeviationPct.WithLabelValues(clusterUUID).Set(stats.MaxDeviationPct)
 	}
-	return deviations, statsByCluster, nil
+	return deviations, nil
 }
 
 // collectLatencyState builds a nodeUUID → nodeLatencyData map by combining:

operator/internal/webapi/rebalancing.go

@@ -16,6 +16,14 @@ type StorageNodeInfo struct {
 	Status     string `json:"status"`
 	Healthy    bool   `json:"health_check"`
 	TotalBytes int64  `json:"total_capacity_bytes"`
+	// Lvols and LvolsMax are the current and maximum lvol-subsystem counts for the
+	// node, used to filter out at-capacity nodes during primary node placement.
+	Lvols    int `json:"lvols"`
+	LvolsMax int `json:"lvols_max"`
+	// IsSecondary is true when this node record is a secondary (HA failover) node
+	// rather than a primary-capable one. Secondary nodes are never eligible as a
+	// new volume's primary.
+	IsSecondary bool `json:"is_secondary"`
 }
 
 // CapacityStat holds the capacity sub-object present on VolumeDTO.

operator/internal/webhook/simplyblock_volume_placement_injector.go

@@ -0,0 +1,190 @@
+package webhook
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+
+	simplyblockv1alpha1 "github.com/simplyblock/simplyblock-operator/api/v1alpha1"
+	"github.com/simplyblock/simplyblock-operator/internal/utils"
+	"github.com/simplyblock/simplyblock-operator/internal/volumemigration"
+	"github.com/simplyblock/simplyblock-operator/internal/volumemigration/autobalancing"
+	"github.com/simplyblock/simplyblock-operator/internal/webapi"
+)
+
+const (
+	// annotationHostID and deprecatedAnnotationHostID must match the annotation names
+	// spdk-csi reads in prepareCreateVolumeReq (pkg/spdk/controllerserver.go) — this
+	// webhook and spdk-csi share the contract but not any code.
+	annotationHostID           = "simplyblock.io/host-id"
+	deprecatedAnnotationHostID = "simplybk/host-id"
+
+	// clusterIDParam is the StorageClass parameter key holding the target cluster's
+	// backend UUID (see upsertStorageClass in simplyblockpool_controller.go).
+	clusterIDParam = "cluster_id"
+)
+
+// +kubebuilder:webhook:path=/mutate-v1-pvc-simplyblock-placement,mutating=true,failurePolicy=ignore,sideEffects=None,groups="",resources=persistentvolumeclaims,verbs=create,versions=v1,name=simplyblock-volume-placement-injector.simplyblock.io,admissionReviewVersions=v1
+// +kubebuilder:rbac:groups=storage.k8s.io,resources=storageclasses,verbs=get;list;watch
+// +kubebuilder:rbac:groups=storage.simplyblock.io,resources=storageclusters,verbs=get;list;watch
+// +kubebuilder:rbac:groups=storage.simplyblock.io,resources=storagenodesets,verbs=get;list;watch
+
+// SimplyblockVolumePlacementInjector is a mutating admission webhook that computes the
+// least-loaded eligible storage node for a new PVC's primary volume — using the same
+// latency-deviation signal the auto-rebalancer (Issue #130) uses — and stamps it onto the
+// PVC as the simplyblock.io/host-id annotation, which spdk-csi already reads and forwards
+// as host_id on CreateVolume. failurePolicy=ignore, and every skip/error path below allows
+// the PVC unmodified, so this can never block volume provisioning: sbcli's own
+// weighted-random pick (_get_next_3_nodes) runs as the fallback exactly as it does today.
+type SimplyblockVolumePlacementInjector struct {
+	Client       client.Client
+	APIClient    *webapi.Client
+	NodeSelector *autobalancing.StorageNodeSelector
+}
+
+func (h *SimplyblockVolumePlacementInjector) Handle(
+	ctx context.Context,
+	req admission.Request,
+) admission.Response {
+	log := logf.FromContext(ctx).WithValues("pvc", req.Name, "namespace", req.Namespace)
+
+	pvc := &corev1.PersistentVolumeClaim{}
+	if err := json.Unmarshal(req.Object.Raw, pvc); err != nil {
+		return admission.Errored(http.StatusBadRequest, err)
+	}
+
+	if pvc.Annotations[annotationHostID] != "" || pvc.Annotations[deprecatedAnnotationHostID] != "" {
+		log.V(1).Info("Skipping: host-id already set")
+		return admission.Allowed("host-id already set")
+	}
+
+	clusterUUID, ok := h.resolveClusterID(ctx, pvc, log)
+	if !ok {
+		return admission.Allowed("not a simplyblock-provisioned PVC")
+	}
+
+	nodeUUID, ok := h.selectPrimaryNode(ctx, clusterUUID, log)
+	if !ok {
+		return admission.Allowed("no load-based placement decision available")
+	}
+
+	patched := pvc.DeepCopy()
+	if patched.Annotations == nil {
+		patched.Annotations = make(map[string]string)
+	}
+	patched.Annotations[annotationHostID] = nodeUUID
+
+	original, err := json.Marshal(pvc)
+	if err != nil {
+		return admission.Errored(http.StatusInternalServerError, err)
+	}
+	patchedRaw, err := json.Marshal(patched)
+	if err != nil {
+		return admission.Errored(http.StatusInternalServerError, err)
+	}
+	log.Info("Selected primary node for new volume", "nodeUUID", nodeUUID, "clusterUUID", clusterUUID)
+	return admission.PatchResponseFromRaw(original, patchedRaw)
+}
+
+// resolveClusterID resolves the PVC's StorageClass to a simplyblock cluster UUID. ok=false
+// means "not our PVC" (no StorageClass, not our provisioner, no cluster_id) — not an error.
+func (h *SimplyblockVolumePlacementInjector) resolveClusterID(
+	ctx context.Context,
+	pvc *corev1.PersistentVolumeClaim,
+	log logr.Logger,
+) (clusterUUID string, ok bool) {
+	if pvc.Spec.StorageClassName == nil || *pvc.Spec.StorageClassName == "" {
+		return "", false
+	}
+
+	sc := &storagev1.StorageClass{}
+	if err := h.Client.Get(ctx, client.ObjectKey{Name: *pvc.Spec.StorageClassName}, sc); err != nil {
+		log.V(1).Info("Skipping: cannot get StorageClass", "storageClass", *pvc.Spec.StorageClassName, "error", err.Error())
+		return "", false
+	}
+	if sc.Provisioner != utils.CSIProvisioner {
+		return "", false
+	}
+	clusterUUID = sc.Parameters[clusterIDParam]
+	return clusterUUID, clusterUUID != ""
+}
+
+// selectPrimaryNode resolves the StorageCluster owning clusterUUID, checks that it has
+// opted into latency-driven auto-rebalancing (Issue #130), fetches its storage nodes,
+// filters them for placement eligibility, and ranks the survivors by current latency
+// deviation. ok=false covers every "no signal available" case (feature not enabled,
+// backend/Prometheus unreachable, no eligible node) — all of them fall through to the
+// caller allowing the PVC unmodified.
+func (h *SimplyblockVolumePlacementInjector) selectPrimaryNode(
+	ctx context.Context,
+	clusterUUID string,
+	log logr.Logger,
+) (nodeUUID string, ok bool) {
+	var list simplyblockv1alpha1.StorageClusterList
+	if err := h.Client.List(ctx, &list); err != nil {
+		log.Error(err, "Failed to list StorageClusters")
+		return "", false
+	}
+	var cr *simplyblockv1alpha1.StorageCluster
+	for i := range list.Items {
+		if list.Items[i].Status.UUID == clusterUUID {
+			cr = &list.Items[i]
+			break
+		}
+	}
+	if cr == nil {
+		log.V(1).Info("Skipping: no matching StorageCluster found", "clusterUUID", clusterUUID)
+		return "", false
+	}
+
+	vms := cr.Spec.VolumeMigrationSettings
+	if vms == nil || vms.AutoRebalancing == nil {
+		log.V(1).Info("Skipping: auto-rebalancing not configured", "cluster", cr.Name)
+		return "", false
+	}
+	spec := vms.AutoRebalancing
+	if spec.Enabled != nil && !*spec.Enabled {
+		log.V(1).Info("Skipping: auto-rebalancing disabled", "cluster", cr.Name)
+		return "", false
+	}
+
+	cfg, err := autobalancing.ResolveRebalancingConfig(spec)
+	if err != nil {
+		log.V(1).Info("Skipping: invalid rebalancing configuration", "cluster", cr.Name, "error", err.Error())
+		return "", false
+	}
+
+	nodes, err := h.APIClient.GetStorageNodes(ctx, clusterUUID)
+	if err != nil {
+		log.Error(err, "Cannot list storage nodes; skipping smart placement", "cluster", cr.Name)
+		return "", false
+	}
+
+	eligible := make(map[string]bool, len(nodes))
+	storageNodes := make([]volumemigration.StorageNode, 0, len(nodes))
+	for _, n := range nodes {
+		storageNodes = append(storageNodes, volumemigration.StorageNode{UUID: n.UUID, ClusterUUID: clusterUUID})
+		eligible[n.UUID] = n.Status == "online" && n.Healthy && !n.IsSecondary && n.Lvols < n.LvolsMax
+	}
+
+	nodeUUID, ok, err = h.NodeSelector.SelectBestNode(ctx, cfg, eligible, autobalancing.StorageNodeSelectorInput{
+		Namespace:    cr.Namespace,
+		StorageNodes: storageNodes,
+	})
+	if err != nil {
+		log.Error(err, "Cannot select best node; skipping smart placement", "cluster", cr.Name)
+		return "", false
+	}
+	if !ok {
+		log.V(1).Info("Skipping: no eligible node found", "cluster", cr.Name)
+		return "", false
+	}
+	return nodeUUID, true
+}