feat(api): add core marker to SubGroup

.changes/unreleased/Added-20261013-151922.yaml

@@ -0,0 +1,6 @@
+kind: Added
+body: Add optional core marker to SubGroup, with PodGroup webhook validation
+time: 2026-10-13T15:19:22.494278+03:00
+custom:
+    Author: SiorMeir
+    Issue: "25"

config/crd/scheduling.run.ai_podgroups.yaml

@@ -115,6 +115,12 @@ spec:
                   the PodGroup with individual scheduling constraints
                 items:
                   properties:
+                    core:
+                      description: |-
+                        Core marks this SubGroup as part of its parent's core (minimal satisfying set).
+                        Only meaningful when the PodGroup is semi-preemptible.
+                        The number of core SubGroups under one parent must not exceed that parent's effective minSubGroup.
+                      type: boolean
                     minMember:
                       description: |-
                         MinMember defines the minimal number of members to run this SubGroup;

scheduling/v2alpha2/podgroup_types.go

@@ -180,6 +180,12 @@ type SubGroup struct {
 
 	// TopologyConstraint defines the topology constraints for this SubGroup
 	TopologyConstraint *TopologyConstraint `json:"topologyConstraint,omitempty" protobuf:"bytes,4,opt,name=topologyConstraint"`
+
+	// Core marks this SubGroup as part of its parent's core (minimal satisfying set).
+	// Only meaningful when the PodGroup is semi-preemptible.
+	// The number of core SubGroups under one parent must not exceed that parent's effective minSubGroup.
+	// +kubebuilder:validation:Optional
+	Core bool `json:"core,omitempty"`
 }
 
 // PodGroupStatus defines the observed state of PodGroup

scheduling/v2alpha2/podgroup_webhook.go

@@ -32,12 +32,15 @@ func (_ *PodGroup) ValidateCreate(ctx context.Context, podGroup *PodGroup) (admi
 			"namespace", podGroup.Namespace, "name", podGroup.Name, "error", validationErrors.structuralError)
 		return nil, validationErrors.structuralError
 	}
+
+	warnings := ignoredCoreWarnings(&podGroup.Spec)
 	if len(validationErrors.minDefinitionErrors) > 0 {
-		return handleMinDefinitionErrors(ctx, validationErrors.minDefinitionErrors, podGroup,
+		minWarnings, err := handleMinDefinitionErrors(ctx, validationErrors.minDefinitionErrors, podGroup,
 			[]error{&minSubGroupExceedsChildCountError{}})
+		return append(warnings, minWarnings...), err
 	}
 
-	return nil, nil
+	return warnings, nil
 }
 
 // ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
@@ -61,12 +64,100 @@ func (_ *PodGroup) ValidateUpdate(ctx context.Context, oldPodGroup *PodGroup, po
 		}
 	}
 
+	warnings := ignoredCoreWarnings(&podGroup.Spec)
 	if len(validationErrors.minDefinitionErrors) > 0 {
-		return handleMinDefinitionErrors(ctx, validationErrors.minDefinitionErrors, podGroup,
+		minWarnings, err := handleMinDefinitionErrors(ctx, validationErrors.minDefinitionErrors, podGroup,
 			[]error{&parentMinMemberError{}, &minSubGroupExceedsChildCountError{}})
+		return append(warnings, minWarnings...), err
 	}
 
-	return nil, nil
+	return warnings, nil
+}
+
+// ignoredCoreWarnings warns that core marks have no effect unless the PodGroup is semi-preemptible.
+func ignoredCoreWarnings(spec *PodGroupSpec) admission.Warnings {
+	if spec.Preemptibility == SemiPreemptible {
+		return nil
+	}
+	for i := range spec.SubGroups {
+		if spec.SubGroups[i].Core {
+			return admission.Warnings{fmt.Sprintf(
+				"subgroup %q: core is ignored unless preemptibility is %q", spec.SubGroups[i].Name, SemiPreemptible)}
+		}
+	}
+	return nil
+}
+
+// validateCoreMarks rejects a parent with more core children than its effective minSubGroup
+// (minSubGroup, or the number of children when unset). Marks do not apply to non-semi-preemptible PodGroups.
+func validateCoreMarks(spec *PodGroupSpec) []error {
+	if spec.Preemptibility != SemiPreemptible {
+		return nil
+	}
+
+	childCounts := directChildCounts(spec.SubGroups)
+	subGroups := map[string]*SubGroup{}
+	for i := range spec.SubGroups {
+		subGroups[spec.SubGroups[i].Name] = &spec.SubGroups[i]
+	}
+
+	coreCounts := map[string]int{}
+	var parents []string // first-seen order keeps error output deterministic
+	for _, sg := range spec.SubGroups {
+		if !sg.Core {
+			continue
+		}
+		parent := ptr.Deref(sg.Parent, rootParentKey)
+		if coreCounts[parent] == 0 {
+			parents = append(parents, parent)
+		}
+		coreCounts[parent]++
+	}
+
+	var errs []error
+	for _, parent := range parents {
+		minSubGroup, scope := spec.MinSubGroup, "the PodGroup"
+		if parent != rootParentKey {
+			minSubGroup, scope = subGroups[parent].MinSubGroup, fmt.Sprintf("subgroup %q", parent)
+		}
+		effective := int(ptr.Deref(minSubGroup, int32(childCounts[parent])))
+		if coreCounts[parent] > effective {
+			errs = append(errs, &coreExceedsMinSubGroupError{msg: fmt.Sprintf(
+				"%s has %d core child SubGroups, exceeding its effective minSubGroup (%d)",
+				scope, coreCounts[parent], effective)})
+		}
+	}
+	return errs
+}
+
+// validateCoreImmutability rejects adding, removing or moving a core mark: it changes the protected set.
+func validateCoreImmutability(old, updated *PodGroupSpec) error {
+	oldParents := map[string]string{}
+	for _, sg := range old.SubGroups {
+		if sg.Core {
+			oldParents[sg.Name] = ptr.Deref(sg.Parent, rootParentKey)
+		}
+	}
+	newParents := map[string]string{}
+	for _, sg := range updated.SubGroups {
+		if !sg.Core {
+			continue
+		}
+		newParents[sg.Name] = ptr.Deref(sg.Parent, rootParentKey)
+		oldParent, wasCore := oldParents[sg.Name]
+		if !wasCore {
+			return fmt.Errorf("cannot add core to subgroup %q on a semi-preemptible PodGroup: it would change the protected set of running pods", sg.Name)
+		}
+		if oldParent != newParents[sg.Name] {
+			return fmt.Errorf("cannot move core subgroup %q to another parent on a semi-preemptible PodGroup", sg.Name)
+		}
+	}
+	for _, sg := range old.SubGroups {
+		if _, stillCore := newParents[sg.Name]; sg.Core && !stillCore {
+			return fmt.Errorf("cannot remove core from subgroup %q on a semi-preemptible PodGroup: it would change the protected set of running pods", sg.Name)
+		}
+	}
+	return nil
 }
 
 // validateSemiPreemptibleImmutability rejects increases to minMember or minSubGroup at the root spec and
@@ -117,6 +208,10 @@ func validateSemiPreemptibleImmutability(old, updated *PodGroupSpec) error {
 		}
 	}
 
+	// Switching away from semi-preemptible ignores marks, so only enforce while it stays semi-preemptible.
+	if updated.Preemptibility == SemiPreemptible {
+		return validateCoreImmutability(old, updated)
+	}
 	return nil
 }
 
@@ -172,6 +267,7 @@ func validatePodGroupSpec(spec *PodGroupSpec) *validationErrors {
 	if validationErrors.structuralError != nil {
 		return validationErrors
 	}
+	validationErrors.minDefinitionErrors = append(validationErrors.minDefinitionErrors, validateCoreMarks(spec)...)
 
 	if spec.MinSubGroup != nil {
 		rootCount := countRootSubGroups(spec.SubGroups)

scheduling/v2alpha2/podgroup_webhook_core_test.go

@@ -0,0 +1,218 @@
+// Copyright 2025 NVIDIA CORPORATION
+// SPDX-License-Identifier: Apache-2.0
+
+package v2alpha2
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"k8s.io/utils/ptr"
+)
+
+func leaf(name string, core bool) SubGroup {
+	return SubGroup{Name: name, MinMember: ptr.To(int32(1)), Core: core}
+}
+
+func child(name, parent string, core bool) SubGroup {
+	sg := leaf(name, core)
+	sg.Parent = ptr.To(parent)
+	return sg
+}
+
+func mid(name string, minSubGroup *int32, core bool) SubGroup {
+	return SubGroup{Name: name, MinSubGroup: minSubGroup, Core: core}
+}
+
+func TestCoreMarksOnCreate(t *testing.T) {
+	tests := []struct {
+		name         string
+		spec         PodGroupSpec
+		wantErr      string
+		wantWarnings int
+	}{
+		{
+			name: "marks within root minSubGroup accepted",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible, MinSubGroup: ptr.To(int32(1)),
+				SubGroups: []SubGroup{leaf("a", true), leaf("b", false)}},
+		},
+		{
+			name: "marks equal to root minSubGroup accepted",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible, MinSubGroup: ptr.To(int32(2)),
+				SubGroups: []SubGroup{leaf("a", true), leaf("b", true), leaf("c", false)}},
+		},
+		{
+			name: "marks exceeding root minSubGroup rejected",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible, MinSubGroup: ptr.To(int32(1)),
+				SubGroups: []SubGroup{leaf("a", true), leaf("b", true)}},
+			wantErr: "the PodGroup has 2 core child SubGroups, exceeding its effective minSubGroup (1)",
+		},
+		{
+			name: "redundant marks with unset minSubGroup accepted",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible,
+				SubGroups: []SubGroup{leaf("a", true), leaf("b", true)}},
+		},
+		{
+			name: "marks within nested minSubGroup accepted",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible, MinSubGroup: ptr.To(int32(1)),
+				SubGroups: []SubGroup{
+					mid("p", ptr.To(int32(1)), true), child("x", "p", true), child("y", "p", false),
+					leaf("z", false),
+				}},
+		},
+		{
+			name: "marks exceeding nested minSubGroup rejected",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible, MinSubGroup: ptr.To(int32(1)),
+				SubGroups: []SubGroup{
+					mid("p", ptr.To(int32(1)), true), child("x", "p", true), child("y", "p", true),
+				}},
+			wantErr: `subgroup "p" has 2 core child SubGroups, exceeding its effective minSubGroup (1)`,
+		},
+		{
+			name: "unset nested minSubGroup counts every child",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible,
+				SubGroups: []SubGroup{
+					mid("p", nil, false), child("x", "p", true), child("y", "p", true),
+				}},
+		},
+		{
+			name: "core on preemptible warns and ignores over-quota marks",
+			spec: PodGroupSpec{Preemptibility: Preemptible, MinSubGroup: ptr.To(int32(1)),
+				SubGroups: []SubGroup{leaf("a", true), leaf("b", true)}},
+			wantWarnings: 1,
+		},
+		{
+			name: "core on non-preemptible warns",
+			spec: PodGroupSpec{Preemptibility: NonPreemptible,
+				SubGroups: []SubGroup{leaf("a", true)}},
+			wantWarnings: 1,
+		},
+		{
+			name: "no warning without marks on preemptible",
+			spec: PodGroupSpec{Preemptibility: Preemptible,
+				SubGroups: []SubGroup{leaf("a", false)}},
+		},
+		{
+			name: "no warning on semi-preemptible",
+			spec: PodGroupSpec{Preemptibility: SemiPreemptible,
+				SubGroups: []SubGroup{leaf("a", true)}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			warnings, err := (&PodGroup{}).ValidateCreate(context.Background(), &PodGroup{Spec: tt.spec})
+			if tt.wantErr == "" && err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if tt.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tt.wantErr)) {
+				t.Fatalf("got error %v, want it to contain %q", err, tt.wantErr)
+			}
+			if len(warnings) != tt.wantWarnings {
+				t.Fatalf("got warnings %v, want %d", warnings, tt.wantWarnings)
+			}
+		})
+	}
+}
+
+func TestCoreMarksOnUpdate(t *testing.T) {
+	semi := func(subGroups ...SubGroup) PodGroupSpec {
+		return PodGroupSpec{Preemptibility: SemiPreemptible, MinSubGroup: ptr.To(int32(1)), SubGroups: subGroups}
+	}
+	nested := func(coreX, coreY bool, yParent string) PodGroupSpec {
+		return semi(mid("p", ptr.To(int32(1)), false), mid("q", ptr.To(int32(1)), false),
+			child("x", "p", coreX), child("y", yParent, coreY), child("pad", "q", false))
+	}
+
+	tests := []struct {
+		name    string
+		old     PodGroupSpec
+		updated PodGroupSpec
+		wantErr string
+	}{
+		{
+			name:    "unchanged marks accepted",
+			old:     semi(leaf("a", true), leaf("b", false)),
+			updated: semi(leaf("a", true), leaf("b", false)),
+		},
+		{
+			name:    "adding a mark rejected",
+			old:     semi(leaf("a", false), leaf("b", false)),
+			updated: semi(leaf("a", true), leaf("b", false)),
+			wantErr: `cannot add core to subgroup "a"`,
+		},
+		{
+			name:    "removing a mark rejected",
+			old:     semi(leaf("a", true), leaf("b", false)),
+			updated: semi(leaf("a", false), leaf("b", false)),
+			wantErr: `cannot remove core from subgroup "a"`,
+		},
+		{
+			name:    "moving a mark to a sibling rejected",
+			old:     semi(leaf("a", true), leaf("b", false)),
+			updated: semi(leaf("a", false), leaf("b", true)),
+			wantErr: "cannot ",
+		},
+		{
+			name:    "deleting a marked subgroup rejected",
+			old:     semi(leaf("a", true), leaf("b", false)),
+			updated: semi(leaf("b", false)),
+			wantErr: `cannot remove core from subgroup "a"`,
+		},
+		{
+			name:    "adding a new marked subgroup rejected",
+			old:     semi(leaf("a", true), leaf("b", false)),
+			updated: semi(leaf("a", true), leaf("b", false), leaf("c", true)),
+			wantErr: `cannot add core to subgroup "c"`,
+		},
+		{
+			name:    "adding a new unmarked subgroup accepted",
+			old:     semi(leaf("a", true), leaf("b", false)),
+			updated: semi(leaf("a", true), leaf("b", false), leaf("c", false)),
+		},
+		{
+			name:    "nested marks unchanged accepted",
+			old:     nested(true, false, "p"),
+			updated: nested(true, false, "p"),
+		},
+		{
+			name:    "nested mark added rejected",
+			old:     nested(true, false, "p"),
+			updated: nested(true, true, "p"),
+			wantErr: `cannot add core to subgroup "y"`,
+		},
+		{
+			name:    "nested mark reparented rejected",
+			old:     nested(false, true, "p"),
+			updated: nested(false, true, "q"),
+			wantErr: `cannot move core subgroup "y"`,
+		},
+		{
+			name: "marking preemptible group allowed (ignored)",
+			old: PodGroupSpec{Preemptibility: Preemptible,
+				SubGroups: []SubGroup{leaf("a", false)}},
+			updated: PodGroupSpec{Preemptibility: Preemptible,
+				SubGroups: []SubGroup{leaf("a", true)}},
+		},
+		{
+			name: "unmarking while leaving semi-preemptible allowed",
+			old:  semi(leaf("a", true), leaf("b", false)),
+			updated: PodGroupSpec{Preemptibility: Preemptible, MinSubGroup: ptr.To(int32(1)),
+				SubGroups: []SubGroup{leaf("a", false), leaf("b", false)}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := (&PodGroup{}).ValidateUpdate(context.Background(),
+				&PodGroup{Spec: tt.old}, &PodGroup{Spec: tt.updated})
+			if tt.wantErr == "" && err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if tt.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tt.wantErr)) {
+				t.Fatalf("got error %v, want it to contain %q", err, tt.wantErr)
+			}
+		})
+	}
+}