Merge pull request #52421 from WIZARD-CXY/fixpredicate

Automatic merge from submit-queue (batch tested with PRs 53780, 55663, 55321, 52421, 55659). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

add hostip and protocol to the hostport predicates

**What this PR does / why we need it**:
This PR adds "hostIP and protocol" to scheduler hostport predicate procedure
**Which issue this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close that issue when PR gets merged)*: fixes #
fix #51950 
**Special notes for your reviewer**:
- [x] basic implementation, need review
- [x] e2e test
- [x] update doc (will be done in seperate PR)

**Release note**:

```release-note
add hostIP and protocol to the original hostport predicates procedure in scheduler.
```

plugin/pkg/scheduler/algorithm/predicates/metadata.go

@@ -46,7 +46,7 @@ type predicateMetadata struct {
 	pod           *v1.Pod
 	podBestEffort bool
 	podRequest    *schedulercache.Resource
-	podPorts      map[int]bool
+	podPorts      map[string]bool
 	//key is a pod full name with the anti-affinity rules.
 	matchingAntiAffinityTerms          map[string][]matchingPodAntiAffinityTerm
 	serviceAffinityInUse               bool
@@ -172,7 +172,7 @@ func (meta *predicateMetadata) ShallowCopy() algorithm.PredicateMetadata {
 		podRequest:           meta.podRequest,
 		serviceAffinityInUse: meta.serviceAffinityInUse,
 	}
-	newPredMeta.podPorts = map[int]bool{}
+	newPredMeta.podPorts = map[string]bool{}
 	for k, v := range meta.podPorts {
 		newPredMeta.podPorts[k] = v
 	}

plugin/pkg/scheduler/algorithm/predicates/metadata_test.go

@@ -373,7 +373,7 @@ func TestPredicateMetadata_ShallowCopy(t *testing.T) {
 			Memory:           300,
 			AllowedPodNumber: 4,
 		},
-		podPorts: map[int]bool{1234: true, 456: false},
+		podPorts: map[string]bool{"1234": true, "456": false},
 		matchingAntiAffinityTerms: map[string][]matchingPodAntiAffinityTerm{
 			"term1": {
 				{

plugin/pkg/scheduler/algorithm/predicates/predicates.go

@@ -890,7 +890,7 @@ func (s *ServiceAffinity) checkServiceAffinity(pod *v1.Pod, meta algorithm.Predi
 
 // PodFitsHostPorts checks if a node has free ports for the requested pod ports.
 func PodFitsHostPorts(pod *v1.Pod, meta algorithm.PredicateMetadata, nodeInfo *schedulercache.NodeInfo) (bool, []algorithm.PredicateFailureReason, error) {
-	var wantPorts map[int]bool
+	var wantPorts map[string]bool
 	if predicateMeta, ok := meta.(*predicateMetadata); ok {
 		wantPorts = predicateMeta.podPorts
 	} else {
@@ -902,11 +902,12 @@ func PodFitsHostPorts(pod *v1.Pod, meta algorithm.PredicateMetadata, nodeInfo *s
 	}
 
 	existingPorts := nodeInfo.UsedPorts()
-	for wport := range wantPorts {
-		if wport != 0 && existingPorts[wport] {
-			return false, []algorithm.PredicateFailureReason{ErrPodNotFitsHostPorts}, nil
-		}
+
+	// try to see whether existingPorts and  wantPorts will conflict or not
+	if portsConflict(existingPorts, wantPorts) {
+		return false, []algorithm.PredicateFailureReason{ErrPodNotFitsHostPorts}, nil
 	}
+
 	return true, nil, nil
 }
 

plugin/pkg/scheduler/algorithm/predicates/predicates_test.go

@@ -554,10 +554,17 @@ func TestPodFitsHost(t *testing.T) {
 	}
 }
 
-func newPod(host string, hostPorts ...int) *v1.Pod {
+func newPod(host string, hostPortInfos ...string) *v1.Pod {
 	networkPorts := []v1.ContainerPort{}
-	for _, port := range hostPorts {
-		networkPorts = append(networkPorts, v1.ContainerPort{HostPort: int32(port)})
+	for _, portInfo := range hostPortInfos {
+		hostPortInfo := decode(portInfo)
+		hostPort, _ := strconv.Atoi(hostPortInfo.hostPort)
+
+		networkPorts = append(networkPorts, v1.ContainerPort{
+			HostIP:   hostPortInfo.hostIP,
+			HostPort: int32(hostPort),
+			Protocol: v1.Protocol(hostPortInfo.protocol),
+		})
 	}
 	return &v1.Pod{
 		Spec: v1.PodSpec{
@@ -585,32 +592,88 @@ func TestPodFitsHostPorts(t *testing.T) {
 			test:     "nothing running",
 		},
 		{
-			pod: newPod("m1", 8080),
+			pod: newPod("m1", "UDP/127.0.0.1/8080"),
 			nodeInfo: schedulercache.NewNodeInfo(
-				newPod("m1", 9090)),
+				newPod("m1", "UDP/127.0.0.1/9090")),
 			fits: true,
 			test: "other port",
 		},
 		{
-			pod: newPod("m1", 8080),
+			pod: newPod("m1", "UDP/127.0.0.1/8080"),
 			nodeInfo: schedulercache.NewNodeInfo(
-				newPod("m1", 8080)),
+				newPod("m1", "UDP/127.0.0.1/8080")),
 			fits: false,
-			test: "same port",
+			test: "same udp port",
 		},
 		{
-			pod: newPod("m1", 8000, 8080),
+			pod: newPod("m1", "TCP/127.0.0.1/8080"),
 			nodeInfo: schedulercache.NewNodeInfo(
-				newPod("m1", 8080)),
+				newPod("m1", "TCP/127.0.0.1/8080")),
 			fits: false,
-			test: "second port",
+			test: "same tcp port",
 		},
 		{
-			pod: newPod("m1", 8000, 8080),
+			pod: newPod("m1", "TCP/127.0.0.1/8080"),
 			nodeInfo: schedulercache.NewNodeInfo(
-				newPod("m1", 8001, 8080)),
+				newPod("m1", "TCP/127.0.0.2/8080")),
+			fits: true,
+			test: "different host ip",
+		},
+		{
+			pod: newPod("m1", "UDP/127.0.0.1/8080"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "TCP/127.0.0.1/8080")),
+			fits: true,
+			test: "different protocol",
+		},
+		{
+			pod: newPod("m1", "UDP/127.0.0.1/8000", "UDP/127.0.0.1/8080"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "UDP/127.0.0.1/8080")),
 			fits: false,
-			test: "second port",
+			test: "second udp port conflict",
+		},
+		{
+			pod: newPod("m1", "TCP/127.0.0.1/8001", "UDP/127.0.0.1/8080"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "TCP/127.0.0.1/8001", "UDP/127.0.0.1/8081")),
+			fits: false,
+			test: "first tcp port conflict",
+		},
+		{
+			pod: newPod("m1", "TCP/0.0.0.0/8001"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "TCP/127.0.0.1/8001")),
+			fits: false,
+			test: "first tcp port conflict due to 0.0.0.0 hostIP",
+		},
+		{
+			pod: newPod("m1", "TCP/10.0.10.10/8001", "TCP/0.0.0.0/8001"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "TCP/127.0.0.1/8001")),
+			fits: false,
+			test: "TCP hostPort conflict due to 0.0.0.0 hostIP",
+		},
+		{
+			pod: newPod("m1", "TCP/127.0.0.1/8001"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "TCP/0.0.0.0/8001")),
+			fits: false,
+			test: "second tcp port conflict to 0.0.0.0 hostIP",
+		},
+		{
+			pod: newPod("m1", "UDP/127.0.0.1/8001"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "TCP/0.0.0.0/8001")),
+			fits: true,
+			test: "second different protocol",
+		},
+		{
+			pod: newPod("m1", "UDP/127.0.0.1/8001"),
+			nodeInfo: schedulercache.NewNodeInfo(
+				newPod("m1", "TCP/0.0.0.0/8001", "UDP/0.0.0.0/8001")),
+			fits: false,
+			test: "UDP hostPort conflict due to 0.0.0.0 hostIP",
 		},
 	}
 	expectedFailureReasons := []algorithm.PredicateFailureReason{ErrPodNotFitsHostPorts}
@@ -631,29 +694,28 @@ func TestPodFitsHostPorts(t *testing.T) {
 
 func TestGetUsedPorts(t *testing.T) {
 	tests := []struct {
-		pods []*v1.Pod
-
-		ports map[int]bool
+		pods  []*v1.Pod
+		ports map[string]bool
 	}{
 		{
 			[]*v1.Pod{
-				newPod("m1", 9090),
+				newPod("m1", "UDP/127.0.0.1/9090"),
 			},
-			map[int]bool{9090: true},
+			map[string]bool{"UDP/127.0.0.1/9090": true},
 		},
 		{
 			[]*v1.Pod{
-				newPod("m1", 9090),
-				newPod("m1", 9091),
+				newPod("m1", "UDP/127.0.0.1/9090"),
+				newPod("m1", "UDP/127.0.0.1/9091"),
 			},
-			map[int]bool{9090: true, 9091: true},
+			map[string]bool{"UDP/127.0.0.1/9090": true, "UDP/127.0.0.1/9091": true},
 		},
 		{
 			[]*v1.Pod{
-				newPod("m1", 9090),
-				newPod("m2", 9091),
+				newPod("m1", "TCP/0.0.0.0/9090"),
+				newPod("m2", "UDP/127.0.0.1/9091"),
 			},
-			map[int]bool{9090: true, 9091: true},
+			map[string]bool{"TCP/0.0.0.0/9090": true, "UDP/127.0.0.1/9091": true},
 		},
 	}
 

plugin/pkg/scheduler/algorithm/predicates/utils.go

@@ -17,9 +17,12 @@ limitations under the License.
 package predicates
 
 import (
+	"strings"
+
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	schedutil "k8s.io/kubernetes/plugin/pkg/scheduler/util"
 )
 
 // FindLabelsInSet gets as many key/value pairs as possible out of a label set.
@@ -89,3 +92,69 @@ func GetEquivalencePod(pod *v1.Pod) interface{} {
 type EquivalencePod struct {
 	ControllerRef metav1.OwnerReference
 }
+
+type hostPortInfo struct {
+	protocol string
+	hostIP   string
+	hostPort string
+}
+
+// decode decodes string ("protocol/hostIP/hostPort") to *hostPortInfo object.
+func decode(info string) *hostPortInfo {
+	hostPortInfoSlice := strings.Split(info, "/")
+
+	protocol := hostPortInfoSlice[0]
+	hostIP := hostPortInfoSlice[1]
+	hostPort := hostPortInfoSlice[2]
+
+	return &hostPortInfo{
+		protocol: protocol,
+		hostIP:   hostIP,
+		hostPort: hostPort,
+	}
+}
+
+// specialPortConflictCheck detects whether specailHostPort(whose hostIP is 0.0.0.0) is conflict with otherHostPorts.
+// return true if we have a conflict.
+func specialPortConflictCheck(specialHostPort string, otherHostPorts map[string]bool) bool {
+	specialHostPortInfo := decode(specialHostPort)
+
+	if specialHostPortInfo.hostIP == schedutil.DefaultBindAllHostIP {
+		// loop through all the otherHostPorts to see if there exists a conflict
+		for hostPortItem := range otherHostPorts {
+			hostPortInfo := decode(hostPortItem)
+
+			// if there exists one hostPortItem which has the same hostPort and protocol with the specialHostPort, that will cause a conflict
+			if specialHostPortInfo.hostPort == hostPortInfo.hostPort && specialHostPortInfo.protocol == hostPortInfo.protocol {
+				return true
+			}
+		}
+
+	}
+
+	return false
+}
+
+// portsConflict check whether existingPorts and wantPorts conflict with each other
+// return true if we have a conflict
+func portsConflict(existingPorts, wantPorts map[string]bool) bool {
+
+	for existingPort := range existingPorts {
+		if specialPortConflictCheck(existingPort, wantPorts) {
+			return true
+		}
+	}
+
+	for wantPort := range wantPorts {
+		if specialPortConflictCheck(wantPort, existingPorts) {
+			return true
+		}
+
+		// general check hostPort conflict procedure for hostIP is not 0.0.0.0
+		if existingPorts[wantPort] {
+			return true
+		}
+	}
+
+	return false
+}

plugin/pkg/scheduler/algorithm/predicates/utils_test.go

@@ -18,6 +18,8 @@ package predicates
 
 import (
 	"fmt"
+	"reflect"
+	"testing"
 
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -68,3 +70,194 @@ func ExampleFindLabelsInSet() {
 	// label1=value1,label2=value2,label3=will_see_this
 	// pod1,pod2,
 }
+
+func Test_decode(t *testing.T) {
+	tests := []struct {
+		name string
+		args string
+		want *hostPortInfo
+	}{
+		{
+			name: "test1",
+			args: "UDP/127.0.0.1/80",
+			want: &hostPortInfo{
+				protocol: "UDP",
+				hostIP:   "127.0.0.1",
+				hostPort: "80",
+			},
+		},
+		{
+			name: "test2",
+			args: "TCP/127.0.0.1/80",
+			want: &hostPortInfo{
+				protocol: "TCP",
+				hostIP:   "127.0.0.1",
+				hostPort: "80",
+			},
+		},
+		{
+			name: "test3",
+			args: "TCP/0.0.0.0/80",
+			want: &hostPortInfo{
+				protocol: "TCP",
+				hostIP:   "0.0.0.0",
+				hostPort: "80",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		if got := decode(tt.args); !reflect.DeepEqual(got, tt.want) {
+			t.Errorf("test name = %v, decode() = %v, want %v", tt.name, got, tt.want)
+		}
+
+	}
+}
+
+func Test_specialPortConflictCheck(t *testing.T) {
+	type args struct {
+		specialHostPort string
+		otherHostPorts  map[string]bool
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "test-1",
+			args: args{
+				specialHostPort: "TCP/0.0.0.0/80",
+				otherHostPorts: map[string]bool{
+					"TCP/127.0.0.2/8080": true,
+					"TCP/127.0.0.1/80":   true,
+					"UDP/127.0.0.2/8080": true,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "test-2",
+			args: args{
+				specialHostPort: "TCP/0.0.0.0/80",
+				otherHostPorts: map[string]bool{
+					"TCP/127.0.0.2/8080": true,
+					"UDP/127.0.0.1/80":   true,
+					"UDP/127.0.0.2/8080": true,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "test-3",
+			args: args{
+				specialHostPort: "TCP/0.0.0.0/80",
+				otherHostPorts: map[string]bool{
+					"TCP/127.0.0.2/8080": true,
+					"TCP/127.0.0.1/8090": true,
+					"UDP/127.0.0.2/8080": true,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "test-4",
+			args: args{
+				specialHostPort: "TCP/0.0.0.0/80",
+				otherHostPorts: map[string]bool{
+					"UDP/127.0.0.2/8080": true,
+					"UDP/127.0.0.1/8090": true,
+					"TCP/127.0.0.2/8080": true,
+				},
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := specialPortConflictCheck(tt.args.specialHostPort, tt.args.otherHostPorts); got != tt.want {
+				t.Errorf("specialPortConflictCheck() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_portsConflict(t *testing.T) {
+	type args struct {
+		existingPorts map[string]bool
+		wantPorts     map[string]bool
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "test1",
+			args: args{
+				existingPorts: map[string]bool{
+					"UDP/127.0.0.1/8080": true,
+				},
+				wantPorts: map[string]bool{
+					"UDP/127.0.0.1/8080": true,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "test2",
+			args: args{
+				existingPorts: map[string]bool{
+					"UDP/127.0.0.2/8080": true,
+				},
+				wantPorts: map[string]bool{
+					"UDP/127.0.0.1/8080": true,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "test3",
+			args: args{
+				existingPorts: map[string]bool{
+					"TCP/127.0.0.1/8080": true,
+				},
+				wantPorts: map[string]bool{
+					"UDP/127.0.0.1/8080": true,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "test4",
+			args: args{
+				existingPorts: map[string]bool{
+					"TCP/0.0.0.0/8080": true,
+				},
+				wantPorts: map[string]bool{
+					"TCP/127.0.0.1/8080": true,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "test5",
+			args: args{
+				existingPorts: map[string]bool{
+					"TCP/127.0.0.1/8080": true,
+				},
+				wantPorts: map[string]bool{
+					"TCP/0.0.0.0/8080": true,
+				},
+			},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := portsConflict(tt.args.existingPorts, tt.args.wantPorts); got != tt.want {
+				t.Errorf("portsConflict() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}