fix(scheduler): namespace PodGroup identities

.changes/unreleased/fixed-20260822-184759.yaml

@@ -0,0 +1,3 @@
+kind: Fixed
+body: |-
+  Isolate same-named PodGroups across Kubernetes namespaces

pkg/scheduler/api/common_info/common.go

@@ -9,6 +9,10 @@ type PodID types.UID
 
 type PodGroupID types.UID
 
+func NewPodGroupID(namespace, name string) PodGroupID {
+	return PodGroupID(NewObjectKey(namespace, name))
+}
+
 type QueueID types.UID
 
 type StorageClassID types.UID

pkg/scheduler/cache/cluster_info/cluster_info.go

@@ -441,7 +441,7 @@ func (c *ClusterInfo) snapshotPodGroups(
 
 	result := map[common_info.PodGroupID]*podgroup_info.PodGroupInfo{}
 	for _, podGroup := range podGroups {
-		podGroupID := common_info.PodGroupID(podGroup.Name)
+		podGroupID := common_info.NewPodGroupID(podGroup.Namespace, podGroup.Name)
 		podGroupInfo := podgroup_info.NewPodGroupInfoWithVectorMap(podGroupID, vectorMap)
 
 		if err := validatePodgroupQueue(existingQueues, podGroup); err != nil {
@@ -453,7 +453,7 @@ func (c *ClusterInfo) snapshotPodGroups(
 		}
 
 		c.setPodGroupWithIndex(podGroup, podGroupInfo)
-		rawPods, err := c.dataLister.ListPodByIndex(podByPodGroupIndexerName, podGroup.Name)
+		rawPods, err := c.dataLister.ListPodByIndex(podByPodGroupIndexerName, string(podGroupID))
 		if err != nil {
 			log.InfraLogger.Errorf("failed to get indexed pods: %s", err)
 			return nil, err
@@ -464,10 +464,11 @@ func (c *ClusterInfo) snapshotPodGroups(
 				log.InfraLogger.Errorf("Snapshot podGroups: Error getting pod from rawPod: %v", rawPod)
 			}
 			podInfo := c.getPodInfo(pod, existingPods, vectorMap)
+			podInfo.Job = podGroupID
 			podGroupInfo.AddTaskInfo(podInfo)
 		}
 
-		result[common_info.PodGroupID(podGroup.Name)] = podGroupInfo
+		result[podGroupID] = podGroupInfo
 	}
 
 	return result, nil

pkg/scheduler/cache/cluster_info/cluster_info_test.go

@@ -1151,8 +1151,9 @@ func TestSnapshotPodGroups(t *testing.T) {
 			objs: []runtime.Object{
 				&enginev2alpha2.PodGroup{
 					ObjectMeta: metav1.ObjectMeta{
-						Name: "podGroup-0",
-						UID:  "ABC",
+						Namespace: testNamespace,
+						Name:      "podGroup-0",
+						UID:       "ABC",
 					},
 					Spec: enginev2alpha2.PodGroupSpec{
 						Queue:     "queue-0",
@@ -1226,6 +1227,7 @@ func TestSnapshotPodGroups(t *testing.T) {
 					subGroupSet.AddPodSet(subGroup1)
 
 					return &podgroup_info.PodGroupInfo{
+						Namespace:       testNamespace,
 						Name:            "podGroup-0",
 						Queue:           "queue-0",
 						RootSubGroupSet: subGroupSet,
@@ -1241,8 +1243,9 @@ func TestSnapshotPodGroups(t *testing.T) {
 			objs: []runtime.Object{
 				&enginev2alpha2.PodGroup{
 					ObjectMeta: metav1.ObjectMeta{
-						Name: "podGroup-0",
-						UID:  "ABC",
+						Namespace: testNamespace,
+						Name:      "podGroup-0",
+						UID:       "ABC",
 					},
 					Spec: enginev2alpha2.PodGroupSpec{
 						Queue: "queue-0",
@@ -1292,6 +1295,7 @@ func TestSnapshotPodGroups(t *testing.T) {
 					subGroupSet.AddPodSet(subGroup0)
 
 					return &podgroup_info.PodGroupInfo{
+						Namespace:       testNamespace,
 						Name:            "podGroup-0",
 						Queue:           "queue-0",
 						RootSubGroupSet: subGroupSet,
@@ -1302,7 +1306,7 @@ func TestSnapshotPodGroups(t *testing.T) {
 				}(),
 			},
 			invalidSubGroupTasks: map[common_info.PodGroupID][]common_info.PodID{
-				"podGroup-0": {common_info.PodID(fmt.Sprintf("%s/pod-invalid", testNamespace))},
+				common_info.NewPodGroupID(testNamespace, "podGroup-0"): {common_info.PodID(fmt.Sprintf("%s/pod-invalid", testNamespace))},
 			},
 		},
 	}
@@ -1325,7 +1329,8 @@ func TestSnapshotPodGroups(t *testing.T) {
 
 		assert.Equal(t, len(test.results), len(podGroups))
 		for _, expected := range test.results {
-			pg, found := podGroups[common_info.PodGroupID(expected.Name)]
+			podGroupID := common_info.NewPodGroupID(expected.Namespace, expected.Name)
+			pg, found := podGroups[podGroupID]
 			assert.True(t, found, "PodGroup not found", expected.Name)
 
 			assert.Equal(t, expected.Name, pg.Name)
@@ -1348,7 +1353,7 @@ func TestSnapshotPodGroups(t *testing.T) {
 				}
 			}
 
-			expectedInvalidTasks := test.invalidSubGroupTasks[common_info.PodGroupID(expected.Name)]
+			expectedInvalidTasks := test.invalidSubGroupTasks[podGroupID]
 			assert.Len(t, pg.GetInvalidSubGroupTasks(), len(expectedInvalidTasks))
 			for _, taskID := range expectedInvalidTasks {
 				assert.Contains(t, pg.GetInvalidSubGroupTasks(), taskID)
@@ -1358,6 +1363,60 @@ func TestSnapshotPodGroups(t *testing.T) {
 	}
 }
 
+func TestSnapshotPodGroupsWithSameNameInDifferentNamespaces(t *testing.T) {
+	const podGroupName = "shared-name"
+	namespaces := []string{"live", "shadow"}
+
+	podGroups := make([]runtime.Object, 0, len(namespaces))
+	pods := make([]runtime.Object, 0, len(namespaces))
+	for _, namespace := range namespaces {
+		podGroups = append(podGroups, &enginev2alpha2.PodGroup{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: namespace,
+				Name:      podGroupName,
+				UID:       types.UID(namespace + "-podgroup"),
+			},
+			Spec: enginev2alpha2.PodGroupSpec{Queue: "queue-0"},
+		})
+		pods = append(pods, &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: namespace,
+				Name:      namespace + "-pod",
+				UID:       types.UID(namespace + "-pod"),
+				Annotations: map[string]string{
+					commonconstants.PodGroupAnnotationForPod: podGroupName,
+				},
+			},
+		})
+	}
+
+	clusterInfo := newClusterInfoTests(t, clusterInfoTestParams{
+		kubeObjects:         pods,
+		kaiSchedulerObjects: podGroups,
+	})
+	result, err := clusterInfo.snapshotPodGroups(
+		map[common_info.QueueID]*queue_info.QueueInfo{"queue-0": {Name: "queue-0"}},
+		map[common_info.PodID]*pod_info.PodInfo{},
+		resource_info.NewResourceVectorMap(),
+	)
+	assert.NoError(t, err)
+	assert.Len(t, result, len(namespaces))
+
+	for _, namespace := range namespaces {
+		podGroupID := common_info.NewPodGroupID(namespace, podGroupName)
+		podGroup, found := result[podGroupID]
+		if !assert.True(t, found, "PodGroup not found: %s", podGroupID) {
+			continue
+		}
+		assert.Equal(t, namespace, podGroup.Namespace)
+		assert.Len(t, podGroup.GetAllPodsMap(), 1)
+		for _, pod := range podGroup.GetAllPodsMap() {
+			assert.Equal(t, namespace, pod.Namespace)
+			assert.Equal(t, podGroupID, pod.Job)
+		}
+	}
+}
+
 func TestSnapshotPodGroups_QueueDoesNotExist_AddsJobFitError(t *testing.T) {
 	clusterInfo := newClusterInfoTests(t,
 		clusterInfoTestParams{
@@ -1397,7 +1456,7 @@ func TestSnapshotPodGroups_QueueDoesNotExist_AddsJobFitError(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, 1, len(podGroups), "Expected 1 podgroup even with missing queue")
 
-	pg, found := podGroups[common_info.PodGroupID("podGroup-missing-queue")]
+	pg, found := podGroups[common_info.NewPodGroupID(testNamespace, "podGroup-missing-queue")]
 	assert.True(t, found, "PodGroup not found")
 	assert.Equal(t, "nonexistent-queue", string(pg.Queue))
 
@@ -2110,8 +2169,9 @@ func TestNotSchedulingPodWithTerminatingPVC(t *testing.T) {
 		},
 		&enginev2alpha2.PodGroup{
 			ObjectMeta: metav1.ObjectMeta{
-				Name: "podGroup-0",
-				UID:  "ABC",
+				Namespace: "test",
+				Name:      "podGroup-0",
+				UID:       "ABC",
 			},
 			Spec: enginev2alpha2.PodGroupSpec{
 				Queue: "queue-0",
@@ -2128,7 +2188,8 @@ func TestNotSchedulingPodWithTerminatingPVC(t *testing.T) {
 	snapshot, err := clusterInfo.Snapshot()
 	assert.Equal(t, nil, err)
 	node := snapshot.Nodes["node-1"]
-	task := snapshot.PodGroupInfos["podGroup-0"].GetAllPodsMap()["pod-1"]
+	podGroupID := common_info.NewPodGroupID("test", "podGroup-0")
+	task := snapshot.PodGroupInfos[podGroupID].GetAllPodsMap()["pod-1"]
 	assert.Equal(t, node.IsTaskAllocatable(task), false)
 
 	pvc.OwnerReferences = nil
@@ -2142,7 +2203,7 @@ func TestNotSchedulingPodWithTerminatingPVC(t *testing.T) {
 	snapshot, err = clusterInfo.Snapshot()
 	assert.Equal(t, nil, err)
 	node = snapshot.Nodes["node-1"]
-	task = snapshot.PodGroupInfos["podGroup-0"].GetAllPodsMap()["pod-1"]
+	task = snapshot.PodGroupInfos[podGroupID].GetAllPodsMap()["pod-1"]
 	assert.Equal(t, node.IsTaskAllocatable(task), true, "Expected task to be allocatable, but got %v", node.IsTaskAllocatable(task))
 }
 

pkg/scheduler/cache/cluster_info/indexers.go

@@ -7,6 +7,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 
 	commonconstants "github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
+	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/common_info"
 	"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/log"
 )
 
@@ -22,5 +23,8 @@ func podByPodGroupIndexer(obj interface{}) ([]string, error) {
 	}
 
 	podGroup := pod.Annotations[commonconstants.PodGroupAnnotationForPod]
-	return []string{podGroup}, nil
+	if podGroup == "" {
+		return []string{""}, nil
+	}
+	return []string{string(common_info.NewPodGroupID(pod.Namespace, podGroup))}, nil
 }

pkg/scheduler/cache/cluster_info/indexers_test.go

@@ -16,6 +16,7 @@ import (
 func TestPodByPodGroupIndexerValidPod(t *testing.T) {
 	pod := &v1.Pod{
 		ObjectMeta: v12.ObjectMeta{
+			Namespace: "my-namespace",
 			Annotations: map[string]string{
 				commonconstants.PodGroupAnnotationForPod: "my-pod-group",
 			},
@@ -26,7 +27,7 @@ func TestPodByPodGroupIndexerValidPod(t *testing.T) {
 
 	assert.Equal(t, nil, err)
 	assert.Equal(t, 1, len(podGroups))
-	assert.Equal(t, "my-pod-group", podGroups[0])
+	assert.Equal(t, "my-namespace/my-pod-group", podGroups[0])
 }
 
 func TestPodByPodGroupIndexerNotPod(t *testing.T) {
@@ -38,3 +39,12 @@ func TestPodByPodGroupIndexerNotPod(t *testing.T) {
 	assert.Equal(t, 1, len(podGroups))
 	assert.Equal(t, "", podGroups[0])
 }
+
+func TestPodByPodGroupIndexerPodWithoutPodGroup(t *testing.T) {
+	pod := &v1.Pod{ObjectMeta: v12.ObjectMeta{Namespace: "my-namespace"}}
+
+	podGroups, err := podByPodGroupIndexer(pod)
+
+	assert.NoError(t, err)
+	assert.Equal(t, []string{""}, podGroups)
+}