Added Hampel+MAD baseline with a configurable (default 6h) window and outlier ejection. The new baselining supports defer and partial window support for the initial 6h measurement window. The existing baseline is still available as a selectable strategy.

Author: noctarius

operator/api/v1alpha1/storagecluster_types.go

@@ -111,6 +111,40 @@ const (
 	MetricsBackendUniform MetricsBackend = "uniform"
 )
 
+// BaselineStrategy selects how the per-node latency baseline (the denominator of the
+// rebalancing deviation signal) is derived.
+// +kubebuilder:validation:Enum=benchmark;rollingWindow
+type BaselineStrategy string
+
+const (
+	// BaselineStrategyBenchmark uses the one-shot fio measurement taken by the baseline
+	// Job on a fresh cluster and frozen on the StorageNode CR status. Simple but tends to
+	// read too low, because an idle cluster is far faster than a loaded one — every loaded
+	// node then shows a large deviation.
+	BaselineStrategyBenchmark BaselineStrategy = "benchmark"
+	// BaselineStrategyRollingWindow derives the baseline from a rolling window of the
+	// probe-sidecar latency series in Prometheus, using a robust outlier-rejecting
+	// estimator. Reflects each node's actual recent operating latency rather than an idle
+	// measurement. This is the default.
+	BaselineStrategyRollingWindow BaselineStrategy = "rollingWindow"
+)
+
+// BaselineColdStartPolicy selects what happens for a node that has fewer than
+// BaselineMinSamples samples in the rolling window (e.g. a freshly onboarded node, or
+// shortly after the probe sidecar starts).
+// +kubebuilder:validation:Enum=defer;partialWindow
+type BaselineColdStartPolicy string
+
+const (
+	// BaselineColdStartDefer omits an under-sampled node from the evaluation cycle: it is
+	// neither a migration source nor a target until it has accumulated BaselineMinSamples
+	// samples. Avoids acting on a noisy baseline.
+	BaselineColdStartDefer BaselineColdStartPolicy = "defer"
+	// BaselineColdStartPartialWindow computes the baseline from whatever samples exist,
+	// accepting a noisier baseline early on so rebalancing engages sooner. This is the default.
+	BaselineColdStartPartialWindow BaselineColdStartPolicy = "partialWindow"
+)
+
 // VolumeAutoPlacementSettings controls the automatic, latency-driven volume rebalancing
 // behaviour. It is configured under StorageClusterSpec.VolumeAutoPlacement.
 type VolumeAutoPlacementSettings struct {
@@ -157,9 +191,32 @@ type VolumeAutoPlacementSettings struct {
 	// +optional
 	LatencyBenchmarkEnabled *bool `json:"latencyBenchmarkEnabled,omitempty"`
 	// LatencyBenchmarkInterval is how often fio benchmark Jobs run against each storage node.
-	// Defaults to 5m.
+	// It also sets the step of the rolling-window baseline query (the cadence at which the
+	// probe sidecar publishes latency samples). Defaults to 5m.
 	// +optional
 	LatencyBenchmarkInterval *metav1.Duration `json:"latencyBenchmarkInterval,omitempty"`
+	// BaselineStrategy selects how the per-node latency baseline is derived. Defaults to
+	// "rollingWindow" (robust estimate over BaselineWindow of the Prometheus latency series);
+	// "benchmark" uses the frozen one-shot fio measurement instead.
+	// +optional
+	BaselineStrategy *BaselineStrategy `json:"baselineStrategy,omitempty"`
+	// BaselineWindow is the look-back window used by the "rollingWindow" strategy. Defaults to 6h.
+	// +optional
+	BaselineWindow *metav1.Duration `json:"baselineWindow,omitempty"`
+	// BaselineColdStart selects what happens for a node with fewer than BaselineMinSamples
+	// samples in the window. Defaults to "partialWindow" (compute from available samples);
+	// "defer" skips the node until enough samples exist.
+	// +optional
+	BaselineColdStart *BaselineColdStartPolicy `json:"baselineColdStart,omitempty"`
+	// BaselineMinSamples is the number of samples below which a node is considered
+	// under-sampled (see BaselineColdStart). Defaults to 6.
+	// +optional
+	BaselineMinSamples *int32 `json:"baselineMinSamples,omitempty"`
+	// BaselineOutlierK is the Hampel-identifier threshold: a sample is rejected as an outlier
+	// when it lies more than k·1.4826·MAD from the window median. Lower is more aggressive.
+	// Defaults to 3.0.
+	// +optional
+	BaselineOutlierK *float64 `json:"baselineOutlierK,omitempty"`
 	// IOPSWeight is the weight applied to per-volume IOPS in the volume IO score. Defaults to 1.0.
 	// +optional
 	IOPSWeight *float64 `json:"iopsWeight,omitempty"`

operator/api/v1alpha1/zz_generated.deepcopy.go

@@ -220,6 +220,40 @@ spec:
                   VolumeAutoPlacement configures automatic, latency-driven volume rebalancing. When
                   nil/disabled the operator performs only manually-triggered VolumeMigrations.
                 properties:
+                  baselineColdStart:
+                    description: |-
+                      BaselineColdStart selects what happens for a node with fewer than BaselineMinSamples
+                      samples in the window. Defaults to "partialWindow" (compute from available samples);
+                      "defer" skips the node until enough samples exist.
+                    enum:
+                    - defer
+                    - partialWindow
+                    type: string
+                  baselineMinSamples:
+                    description: |-
+                      BaselineMinSamples is the number of samples below which a node is considered
+                      under-sampled (see BaselineColdStart). Defaults to 6.
+                    format: int32
+                    type: integer
+                  baselineOutlierK:
+                    description: |-
+                      BaselineOutlierK is the Hampel-identifier threshold: a sample is rejected as an outlier
+                      when it lies more than k·1.4826·MAD from the window median. Lower is more aggressive.
+                      Defaults to 3.0.
+                    type: number
+                  baselineStrategy:
+                    description: |-
+                      BaselineStrategy selects how the per-node latency baseline is derived. Defaults to
+                      "rollingWindow" (robust estimate over BaselineWindow of the Prometheus latency series);
+                      "benchmark" uses the frozen one-shot fio measurement instead.
+                    enum:
+                    - benchmark
+                    - rollingWindow
+                    type: string
+                  baselineWindow:
+                    description: BaselineWindow is the look-back window used by the
+                      "rollingWindow" strategy. Defaults to 6h.
+                    type: string
                   defaultCoolDownSeconds:
                     description: |-
                       DefaultCoolDownSeconds is the cool-down period (seconds) applied to a volume after
@@ -252,7 +286,8 @@ spec:
                   latencyBenchmarkInterval:
                     description: |-
                       LatencyBenchmarkInterval is how often fio benchmark Jobs run against each storage node.
-                      Defaults to 5m.
+                      It also sets the step of the rolling-window baseline query (the cadence at which the
+                      probe sidecar publishes latency samples). Defaults to 5m.
                     type: string
                   maxVolumeMigrationsPerCycle:
                     description: MaxVolumeMigrationsPerCycle is the maximum number

operator/config/crd/bases/storage.simplyblock.io_storageclusters.yaml

@@ -0,0 +1,100 @@
+package autoplacement
+
+import (
+	"math"
+	"sort"
+)
+
+const (
+	// madToSigma scales the median absolute deviation (MAD) to an estimate of the standard
+	// deviation for normally distributed data; it makes the Hampel k threshold comparable to
+	// a number of standard deviations.
+	madToSigma = 1.4826
+	// meanAbsDevToSigma is the equivalent scale factor for the mean absolute deviation, used
+	// only in the degenerate MAD==0 fallback.
+	meanAbsDevToSigma = 1.2533
+)
+
+// robustBaselineNS reduces a set of latency samples (ns) to a single baseline using the
+// Hampel identifier: samples further than k·1.4826·MAD from the window median are rejected
+// as outliers, and the median of the survivors is returned. The Hampel identifier has the
+// highest possible breakdown point (50%) — both its centre (median) and its scale (MAD) are
+// themselves robust, so the extreme journal/EC/HA spikes it is meant to reject cannot inflate
+// the threshold and hide themselves.
+//
+// Degrade rules:
+//   - fewer than 3 samples: rejection is not meaningful, so the plain median is returned;
+//   - MAD == 0 (a majority of identical samples): fall back to the mean absolute deviation,
+//     and if that is also 0 (all samples identical) skip rejection entirely.
+//
+// kept and rejected count survivors and rejects (kept+rejected == len(samples)). ok is false
+// only when samples is empty.
+func robustBaselineNS(samples []float64, k float64) (baselineNS int64, kept, rejected int, ok bool) {
+	if len(samples) == 0 {
+		return 0, 0, 0, false
+	}
+	if len(samples) < 3 {
+		return int64(math.Round(median(samples))), len(samples), 0, true
+	}
+
+	m := median(samples)
+	scale := madToSigma * medianAbsDev(samples, m)
+	if scale == 0 {
+		// MAD degenerate (majority identical): fall back to the mean absolute deviation.
+		scale = meanAbsDevToSigma * meanAbsDev(samples, m)
+	}
+	if scale == 0 {
+		// All samples identical: nothing to reject.
+		return int64(math.Round(m)), len(samples), 0, true
+	}
+
+	threshold := k * scale
+	survivors := make([]float64, 0, len(samples))
+	for _, x := range samples {
+		if math.Abs(x-m) <= threshold {
+			survivors = append(survivors, x)
+		}
+	}
+	// The median always survives, so survivors is never empty; guard defensively anyway.
+	if len(survivors) == 0 {
+		survivors = samples
+	}
+	return int64(math.Round(median(survivors))), len(survivors), len(samples) - len(survivors), true
+}
+
+// median returns the median of xs without mutating it. Returns 0 for an empty slice.
+func median(xs []float64) float64 {
+	n := len(xs)
+	if n == 0 {
+		return 0
+	}
+	sorted := make([]float64, n)
+	copy(sorted, xs)
+	sort.Float64s(sorted)
+	mid := n / 2
+	if n%2 == 1 {
+		return sorted[mid]
+	}
+	return (sorted[mid-1] + sorted[mid]) / 2
+}
+
+// medianAbsDev returns the median of the absolute deviations of xs from center.
+func medianAbsDev(xs []float64, center float64) float64 {
+	devs := make([]float64, len(xs))
+	for i, x := range xs {
+		devs[i] = math.Abs(x - center)
+	}
+	return median(devs)
+}
+
+// meanAbsDev returns the mean of the absolute deviations of xs from center.
+func meanAbsDev(xs []float64, center float64) float64 {
+	if len(xs) == 0 {
+		return 0
+	}
+	var sum float64
+	for _, x := range xs {
+		sum += math.Abs(x - center)
+	}
+	return sum / float64(len(xs))
+}

operator/internal/autoplacement/baseline.go

@@ -0,0 +1,140 @@
+package autoplacement
+
+import (
+	"context"
+	"fmt"
+
+	simplyblockv1alpha1 "github.com/simplyblock/simplyblock-operator/api/v1alpha1"
+	promlatency "github.com/simplyblock/simplyblock-operator/internal/metrics/prometheus"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// BaselineProvider resolves the per-node latency baseline (ns at the configured percentile)
+// that forms the denominator of the rebalancing deviation signal. Nodes without a usable
+// baseline are omitted from the returned map — ComputeLatencyDeviationPct then reads 0 for
+// them, so they are neither flagged hot nor preferred as targets.
+type BaselineProvider interface {
+	BaselineNS(ctx context.Context, inputs ...StorageNodeSelectorInput) (map[string]int64, error)
+}
+
+// newBaselineProvider selects the BaselineProvider implementation for cfg.BaselineStrategy.
+// "benchmark" reads the frozen one-shot fio measurement from the StorageNodeSet CRs;
+// rollingWindow (the default, and the fallback for any unrecognised value) derives a robust
+// estimate from a rolling window of the probe latency series in Prometheus.
+func newBaselineProvider(k8sClient client.Client, cfg RebalancingConfig) (BaselineProvider, error) {
+	if cfg.BaselineStrategy == string(simplyblockv1alpha1.BaselineStrategyBenchmark) {
+		return &benchmarkBaselineProvider{client: k8sClient, percentile: cfg.LatencyPercentile}, nil
+	}
+	provider, err := promlatency.New(cfg.PrometheusURL)
+	if err != nil {
+		return nil, fmt.Errorf("create prometheus baseline provider: %w", err)
+	}
+	return &rollingWindowBaselineProvider{prom: provider, cfg: cfg}, nil
+}
+
+// benchmarkBaselineProvider reads the one-shot fio baseline recorded once per node by the
+// baseline Job and stored on StorageNodeSet.status.latencyMetrics.
+type benchmarkBaselineProvider struct {
+	client     client.Client
+	percentile string
+}
+
+func (b *benchmarkBaselineProvider) BaselineNS(
+	ctx context.Context,
+	inputs ...StorageNodeSelectorInput,
+) (map[string]int64, error) {
+	result := make(map[string]int64)
+	for _, input := range inputs {
+		var snodeList simplyblockv1alpha1.StorageNodeSetList
+		if err := b.client.List(ctx, &snodeList, client.InNamespace(input.Namespace)); err != nil {
+			// Stay resilient to a transient list error: skip this namespace rather than
+			// failing the whole evaluation cycle (matches the previous CR-read behaviour).
+			continue
+		}
+		for _, snode := range snodeList.Items {
+			for _, lm := range snode.Status.LatencyMetrics {
+				baseline := lm.BaselineP50NS
+				if b.percentile == promlatency.PercentileP99 {
+					baseline = lm.BaselineP99NS
+				}
+				if baseline > 0 {
+					result[lm.NodeUUID] = baseline
+				}
+			}
+		}
+	}
+	return result, nil
+}
+
+// rollingWindowBaselineProvider derives each node's baseline from a rolling window of the
+// probe-sidecar latency series in Prometheus, reduced to a single value by robustBaselineNS
+// (Hampel outlier rejection + median of survivors). It emits the per-node baseline and
+// sample-count gauges as a side effect.
+type rollingWindowBaselineProvider struct {
+	prom *promlatency.Provider
+	cfg  RebalancingConfig
+}
+
+func (r *rollingWindowBaselineProvider) BaselineNS(
+	ctx context.Context,
+	inputs ...StorageNodeSelectorInput,
+) (map[string]int64, error) {
+	clusterIDs := distinctClusterUUIDs(inputs)
+	windowed, err := r.prom.GetClustersWindowedLatency(
+		ctx, clusterIDs, r.cfg.LatencyPercentile, r.cfg.BaselineWindow, r.cfg.BaselineStep,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	reduced := reduceWindowedBaselines(windowed, r.cfg)
+	result := make(map[string]int64, len(reduced))
+	for _, nb := range reduced {
+		result[nb.nodeUUID] = nb.baselineNS
+		setBaselineGauges(nb.clusterUUID, nb.nodeUUID, nb.baselineNS, nb.samplesTotal, nb.samplesRejected)
+	}
+	return result, nil
+}
+
+// nodeBaseline is one node's reduced rolling-window baseline plus the sample diagnostics.
+type nodeBaseline struct {
+	clusterUUID     string
+	nodeUUID        string
+	baselineNS      int64
+	samplesTotal    int
+	samplesRejected int
+}
+
+// reduceWindowedBaselines reduces per-node windowed samples to a single robust baseline each,
+// applying the cold-start policy. It is pure (no Prometheus, no metrics) so the cold-start
+// and estimator behaviour can be tested directly. A node is dropped when it is under-sampled
+// under the "defer" policy, or when no positive baseline can be computed from its samples.
+func reduceWindowedBaselines(
+	windowed map[string]map[string][]float64,
+	cfg RebalancingConfig,
+) []nodeBaseline {
+	deferUnderSampled := cfg.BaselineColdStart == string(simplyblockv1alpha1.BaselineColdStartDefer)
+
+	var out []nodeBaseline
+	for clusterUUID, byNode := range windowed {
+		for nodeUUID, samples := range byNode {
+			// Cold start: an under-sampled node is either skipped ("defer") or computed
+			// from whatever samples exist ("partialWindow").
+			if len(samples) < cfg.BaselineMinSamples && deferUnderSampled {
+				continue
+			}
+			baselineNS, kept, rejected, ok := robustBaselineNS(samples, cfg.BaselineOutlierK)
+			if !ok || baselineNS <= 0 {
+				continue
+			}
+			out = append(out, nodeBaseline{
+				clusterUUID:     clusterUUID,
+				nodeUUID:        nodeUUID,
+				baselineNS:      baselineNS,
+				samplesTotal:    kept + rejected,
+				samplesRejected: rejected,
+			})
+		}
+	}
+	return out
+}