Store PVC reference counts in NodeInfo cache

This map will be queried as part of enforcement of the ReadWriteOncePod
access mode for PVCs

pkg/scheduler/framework/types.go

@@ -387,6 +387,10 @@ type NodeInfo struct {
 	// state information.
 	ImageStates map[string]*ImageStateSummary
 
+	// PVCRefCounts contains a mapping of PVC names to the number of pods on the node using it.
+	// Keys are in the format "namespace/name".
+	PVCRefCounts map[string]int
+
 	// Whenever NodeInfo changes, generation is bumped.
 	// This is used to avoid cloning it if the object didn't change.
 	Generation int64
@@ -512,6 +516,7 @@ func NewNodeInfo(pods ...*v1.Pod) *NodeInfo {
 		Generation:       nextGeneration(),
 		UsedPorts:        make(HostPortInfo),
 		ImageStates:      make(map[string]*ImageStateSummary),
+		PVCRefCounts:     make(map[string]int),
 	}
 	for _, pod := range pods {
 		ni.AddPod(pod)
@@ -536,6 +541,7 @@ func (n *NodeInfo) Clone() *NodeInfo {
 		Allocatable:      n.Allocatable.Clone(),
 		UsedPorts:        make(HostPortInfo),
 		ImageStates:      n.ImageStates,
+		PVCRefCounts:     n.PVCRefCounts,
 		Generation:       n.Generation,
 	}
 	if len(n.Pods) > 0 {
@@ -595,6 +601,7 @@ func (n *NodeInfo) AddPodInfo(podInfo *PodInfo) {
 
 	// Consume ports when pods added.
 	n.updateUsedPorts(podInfo.Pod, true)
+	n.updatePVCRefCounts(podInfo.Pod, true)
 
 	n.Generation = nextGeneration()
 }
@@ -672,6 +679,7 @@ func (n *NodeInfo) RemovePod(pod *v1.Pod) error {
 
 			// Release ports when remove Pods.
 			n.updateUsedPorts(pod, false)
+			n.updatePVCRefCounts(pod, false)
 
 			n.Generation = nextGeneration()
 			n.resetSlicesIfEmpty()
@@ -749,6 +757,25 @@ func (n *NodeInfo) updateUsedPorts(pod *v1.Pod, add bool) {
 	}
 }
 
+// updatePVCRefCounts updates the PVCRefCounts of NodeInfo.
+func (n *NodeInfo) updatePVCRefCounts(pod *v1.Pod, add bool) {
+	for _, v := range pod.Spec.Volumes {
+		if v.PersistentVolumeClaim == nil {
+			continue
+		}
+
+		key := pod.Namespace + "/" + v.PersistentVolumeClaim.ClaimName
+		if add {
+			n.PVCRefCounts[key] += 1
+		} else {
+			n.PVCRefCounts[key] -= 1
+			if n.PVCRefCounts[key] <= 0 {
+				delete(n.PVCRefCounts, key)
+			}
+		}
+	}
+}
+
 // SetNode sets the overall node information.
 func (n *NodeInfo) SetNode(node *v1.Node) {
 	n.node = node

pkg/scheduler/framework/types_test.go

@@ -211,7 +211,7 @@ type testingMode interface {
 	Fatalf(format string, args ...interface{})
 }
 
-func makeBasePod(t testingMode, nodeName, objName, cpu, mem, extended string, ports []v1.ContainerPort) *v1.Pod {
+func makeBasePod(t testingMode, nodeName, objName, cpu, mem, extended string, ports []v1.ContainerPort, volumes []v1.Volume) *v1.Pod {
 	req := v1.ResourceList{}
 	if cpu != "" {
 		req = v1.ResourceList{
@@ -240,6 +240,7 @@ func makeBasePod(t testingMode, nodeName, objName, cpu, mem, extended string, po
 				Ports: ports,
 			}},
 			NodeName: nodeName,
+			Volumes:  volumes,
 		},
 	}
 }
@@ -247,8 +248,8 @@ func makeBasePod(t testingMode, nodeName, objName, cpu, mem, extended string, po
 func TestNewNodeInfo(t *testing.T) {
 	nodeName := "test-node"
 	pods := []*v1.Pod{
-		makeBasePod(t, nodeName, "test-1", "100m", "500", "", []v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 80, Protocol: "TCP"}}),
+		makeBasePod(t, nodeName, "test-1", "100m", "500", "", []v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 80, Protocol: "TCP"}}, nil),
-		makeBasePod(t, nodeName, "test-2", "200m", "1Ki", "", []v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 8080, Protocol: "TCP"}}),
+		makeBasePod(t, nodeName, "test-2", "200m", "1Ki", "", []v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 8080, Protocol: "TCP"}}, nil),
 	}
 
 	expected := &NodeInfo{
@@ -274,7 +275,8 @@ func TestNewNodeInfo(t *testing.T) {
 				{Protocol: "TCP", Port: 8080}: {},
 			},
 		},
-		ImageStates: map[string]*ImageStateSummary{},
+		ImageStates:  map[string]*ImageStateSummary{},
+		PVCRefCounts: map[string]int{},
 		Pods: []*PodInfo{
 			{
 				Pod: &v1.Pod{
@@ -366,7 +368,8 @@ func TestNodeInfoClone(t *testing.T) {
 						{Protocol: "TCP", Port: 8080}: {},
 					},
 				},
-				ImageStates: map[string]*ImageStateSummary{},
+				ImageStates:  map[string]*ImageStateSummary{},
+				PVCRefCounts: map[string]int{},
 				Pods: []*PodInfo{
 					{
 						Pod: &v1.Pod{
@@ -439,7 +442,8 @@ func TestNodeInfoClone(t *testing.T) {
 						{Protocol: "TCP", Port: 8080}: {},
 					},
 				},
-				ImageStates: map[string]*ImageStateSummary{},
+				ImageStates:  map[string]*ImageStateSummary{},
+				PVCRefCounts: map[string]int{},
 				Pods: []*PodInfo{
 					{
 						Pod: &v1.Pod{
@@ -548,6 +552,15 @@ func TestNodeInfoAddPod(t *testing.T) {
 				Overhead: v1.ResourceList{
 					v1.ResourceCPU: resource.MustParse("500m"),
 				},
+				Volumes: []v1.Volume{
+					{
+						VolumeSource: v1.VolumeSource{
+							PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+								ClaimName: "pvc-1",
+							},
+						},
+					},
+				},
 			},
 		},
 		{
@@ -578,6 +591,15 @@ func TestNodeInfoAddPod(t *testing.T) {
 					v1.ResourceCPU:    resource.MustParse("500m"),
 					v1.ResourceMemory: resource.MustParse("500"),
 				},
+				Volumes: []v1.Volume{
+					{
+						VolumeSource: v1.VolumeSource{
+							PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+								ClaimName: "pvc-1",
+							},
+						},
+					},
+				},
 			},
 		},
 		{
@@ -618,6 +640,15 @@ func TestNodeInfoAddPod(t *testing.T) {
 					v1.ResourceCPU:    resource.MustParse("500m"),
 					v1.ResourceMemory: resource.MustParse("500"),
 				},
+				Volumes: []v1.Volume{
+					{
+						VolumeSource: v1.VolumeSource{
+							PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+								ClaimName: "pvc-2",
+							},
+						},
+					},
+				},
 			},
 		},
 	}
@@ -649,7 +680,8 @@ func TestNodeInfoAddPod(t *testing.T) {
 				{Protocol: "TCP", Port: 8080}: {},
 			},
 		},
-		ImageStates: map[string]*ImageStateSummary{},
+		ImageStates:  map[string]*ImageStateSummary{},
+		PVCRefCounts: map[string]int{"node_info_cache_test/pvc-1": 2, "node_info_cache_test/pvc-2": 1},
 		Pods: []*PodInfo{
 			{
 				Pod: &v1.Pod{
@@ -680,6 +712,15 @@ func TestNodeInfoAddPod(t *testing.T) {
 						Overhead: v1.ResourceList{
 							v1.ResourceCPU: resource.MustParse("500m"),
 						},
+						Volumes: []v1.Volume{
+							{
+								VolumeSource: v1.VolumeSource{
+									PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+										ClaimName: "pvc-1",
+									},
+								},
+							},
+						},
 					},
 				},
 			},
@@ -712,6 +753,15 @@ func TestNodeInfoAddPod(t *testing.T) {
 							v1.ResourceCPU:    resource.MustParse("500m"),
 							v1.ResourceMemory: resource.MustParse("500"),
 						},
+						Volumes: []v1.Volume{
+							{
+								VolumeSource: v1.VolumeSource{
+									PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+										ClaimName: "pvc-1",
+									},
+								},
+							},
+						},
 					},
 				},
 			},
@@ -754,6 +804,15 @@ func TestNodeInfoAddPod(t *testing.T) {
 							v1.ResourceCPU:    resource.MustParse("500m"),
 							v1.ResourceMemory: resource.MustParse("500"),
 						},
+						Volumes: []v1.Volume{
+							{
+								VolumeSource: v1.VolumeSource{
+									PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+										ClaimName: "pvc-2",
+									},
+								},
+							},
+						},
 					},
 				},
 			},
@@ -779,9 +838,10 @@ func TestNodeInfoAddPod(t *testing.T) {
 func TestNodeInfoRemovePod(t *testing.T) {
 	nodeName := "test-node"
 	pods := []*v1.Pod{
-		makeBasePod(t, nodeName, "test-1", "100m", "500", "", []v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 80, Protocol: "TCP"}}),
+		makeBasePod(t, nodeName, "test-1", "100m", "500", "",
-
+			[]v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 80, Protocol: "TCP"}},
-		makeBasePod(t, nodeName, "test-2", "200m", "1Ki", "", []v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 8080, Protocol: "TCP"}}),
+			[]v1.Volume{{VolumeSource: v1.VolumeSource{PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{ClaimName: "pvc-1"}}}}),
+		makeBasePod(t, nodeName, "test-2", "200m", "1Ki", "", []v1.ContainerPort{{HostIP: "127.0.0.1", HostPort: 8080, Protocol: "TCP"}}, nil),
 	}
 
 	// add pod Overhead
@@ -798,7 +858,7 @@ func TestNodeInfoRemovePod(t *testing.T) {
 		expectedNodeInfo *NodeInfo
 	}{
 		{
-			pod:         makeBasePod(t, nodeName, "non-exist", "0", "0", "", []v1.ContainerPort{{}}),
+			pod:         makeBasePod(t, nodeName, "non-exist", "0", "0", "", []v1.ContainerPort{{}}, []v1.Volume{}),
 			errExpected: true,
 			expectedNodeInfo: &NodeInfo{
 				node: &v1.Node{
@@ -828,7 +888,8 @@ func TestNodeInfoRemovePod(t *testing.T) {
 						{Protocol: "TCP", Port: 8080}: {},
 					},
 				},
-				ImageStates: map[string]*ImageStateSummary{},
+				ImageStates:  map[string]*ImageStateSummary{},
+				PVCRefCounts: map[string]int{"node_info_cache_test/pvc-1": 1},
 				Pods: []*PodInfo{
 					{
 						Pod: &v1.Pod{
@@ -860,6 +921,15 @@ func TestNodeInfoRemovePod(t *testing.T) {
 									v1.ResourceCPU:    resource.MustParse("500m"),
 									v1.ResourceMemory: resource.MustParse("500"),
 								},
+								Volumes: []v1.Volume{
+									{
+										VolumeSource: v1.VolumeSource{
+											PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+												ClaimName: "pvc-1",
+											},
+										},
+									},
+								},
 							},
 						},
 					},
@@ -929,6 +999,15 @@ func TestNodeInfoRemovePod(t *testing.T) {
 						v1.ResourceCPU:    resource.MustParse("500m"),
 						v1.ResourceMemory: resource.MustParse("500"),
 					},
+					Volumes: []v1.Volume{
+						{
+							VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: "pvc-1",
+								},
+							},
+						},
+					},
 				},
 			},
 			errExpected: false,
@@ -959,7 +1038,8 @@ func TestNodeInfoRemovePod(t *testing.T) {
 						{Protocol: "TCP", Port: 8080}: {},
 					},
 				},
-				ImageStates: map[string]*ImageStateSummary{},
+				ImageStates:  map[string]*ImageStateSummary{},
+				PVCRefCounts: map[string]int{},
 				Pods: []*PodInfo{
 					{
 						Pod: &v1.Pod{