feat(api): add PodGroup safeToDisrupt to opt out of consolidation

.changes/unreleased/added-20261012-120909.yaml

@@ -0,0 +1,6 @@
+kind: Added
+body: |-
+  PodGroup safeToDisrupt field and kai.scheduler/safe-to-disrupt annotation key to opt out of consolidation
+custom:
+  Author: "sam-huang1223"
+  Issue: "22"

config/crd/scheduling.run.ai_podgroups.yaml

@@ -98,6 +98,14 @@ spec:
                   Queue defines the queue to allocate resource for PodGroup; if queue does not exist,
                   the PodGroup will not be scheduled.
                 type: string
+              safeToDisrupt:
+                description: |-
+                  SafeToDisrupt is whether the scheduler may evict this PodGroup's pods for reasons other than
+                  fairness or priority. Today that is consolidation, which moves pods to other nodes to make room
+                  for a pending workload. A workload that does not survive a restart (for example replicas with
+                  restartPolicy Never, or bare pods) can set it to false. It does not affect preempt or reclaim.
+                  Defaults to true.
+                type: boolean
               schedulingBackoff:
                 description: The number of scheduling cycles to try before marking
                   the pod group as UnschedulableOnNodePool. Currently only supporting

podgrouper/constants/constants.go

@@ -12,6 +12,7 @@ const (
 
 	PreemptionDelayAnnotationKey      = "kai.scheduler/preemption-delay"
 	StalenessGracePeriodAnnotationKey = "kai.scheduler/staleness-grace-period"
+	SafeToDisruptAnnotationKey        = "kai.scheduler/safe-to-disrupt"
 
 	BuildPriorityClass     = "build"
 	TrainPriorityClass     = "train"

scheduling/v2alpha2/podgroup_types.go

@@ -21,6 +21,7 @@ package v2alpha2
 
 import (
 	"fmt"
+	"strconv"
 	"strings"
 	"time"
 
@@ -90,6 +91,14 @@ type PodGroupSpec struct {
 	// eviction for this PodGroup. Defaults to the scheduler's global staleness grace period.
 	// +optional
 	StalenessGracePeriod *metav1.Duration `json:"stalenessGracePeriod,omitempty" protobuf:"bytes,10,opt,name=stalenessGracePeriod"`
+
+	// SafeToDisrupt is whether the scheduler may evict this PodGroup's pods for reasons other than
+	// fairness or priority. Today that is consolidation, which moves pods to other nodes to make room
+	// for a pending workload. A workload that does not survive a restart (for example replicas with
+	// restartPolicy Never, or bare pods) can set it to false. It does not affect preempt or reclaim.
+	// Defaults to true.
+	// +optional
+	SafeToDisrupt *bool `json:"safeToDisrupt,omitempty" protobuf:"varint,11,opt,name=safeToDisrupt"`
 }
 
 // Preemptibility defines whether this PodGroup can be preempted
@@ -151,6 +160,16 @@ func ParseStalenessGracePeriod(value string) (*metav1.Duration, error) {
 	return &metav1.Duration{Duration: stale}, nil
 }
 
+// ParseSafeToDisrupt parses a safe-to-disrupt value ("true" or "false").
+// Returns an error for anything strconv.ParseBool does not accept.
+func ParseSafeToDisrupt(value string) (*bool, error) {
+	safe, err := strconv.ParseBool(value)
+	if err != nil {
+		return nil, err
+	}
+	return &safe, nil
+}
+
 type SubGroup struct {
 	// Name uniquely identifies the SubGroup within the PodGroup.
 	// Must be a valid DNS label (RFC 1123).

scheduling/v2alpha2/podgroup_types_test.go

@@ -0,0 +1,40 @@
+// Copyright 2025 NVIDIA CORPORATION
+// SPDX-License-Identifier: Apache-2.0
+
+package v2alpha2
+
+import (
+	"testing"
+)
+
+func TestParseSafeToDisrupt(t *testing.T) {
+	tests := []struct {
+		value   string
+		want    bool
+		wantErr bool
+	}{
+		{value: "true", want: true},
+		{value: "false", want: false},
+		{value: "False", want: false},
+		{value: "0", want: false},
+		{value: "", wantErr: true},
+		{value: "no", wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.value, func(t *testing.T) {
+			got, err := ParseSafeToDisrupt(tt.value)
+			if tt.wantErr {
+				if err == nil {
+					t.Fatalf("ParseSafeToDisrupt(%q) = %v, want an error", tt.value, *got)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("ParseSafeToDisrupt(%q) returned error %v", tt.value, err)
+			}
+			if got == nil || *got != tt.want {
+				t.Fatalf("ParseSafeToDisrupt(%q) = %v, want %v", tt.value, got, tt.want)
+			}
+		})
+	}
+}