Merge pull request #48418 from xiangpengzhao/refactor-create-svc

Automatic merge from submit-queue (batch tested with PRs 49409, 49352, 49266, 48418)

Use helper to init ClusterIP and NodePort in Create of service

**What this PR does / why we need it**:
Make service `Create` more readable and testable.

- use `initClusterIP` introduced in #46197 to init ClusterIP allocation in service `Create`
- add a new helper `initNodePort` to init NodePort allocation in service `Create`
- TBD: add test case for `initNodePort`. This will cover the NodePort allocation process in `Create`. If this PR makes sense, I will write a test case later.

**Which issue this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close that issue when PR gets merged)*: fixes #35354 (not directly. #35354 was fixed by #46197. The idea of this PR is from https://github.com/kubernetes/kubernetes/pull/46197#discussion_r120910077)

**Special notes for your reviewer**:
/cc @thockin @freehan 

**Release note**:

```release-note
NONE
```

pkg/registry/core/service/BUILD

@@ -59,6 +59,7 @@ go_test(
     tags = ["automanaged"],
     deps = [
         "//pkg/api:go_default_library",
+        "//pkg/api/helper:go_default_library",
         "//pkg/api/service:go_default_library",
         "//pkg/api/testing:go_default_library",
         "//pkg/features:go_default_library",

pkg/registry/core/service/rest.go

@@ -105,71 +105,19 @@ func (rs *REST) Create(ctx genericapirequest.Context, obj runtime.Object, includ
 		}
 	}()
 
-	if helper.IsServiceIPRequested(service) {
+	var err error
-		// Allocate next available.
+	if service.Spec.Type != api.ServiceTypeExternalName {
-		ip, err := rs.serviceIPs.AllocateNext()
+		if releaseServiceIP, err = rs.initClusterIP(service); err != nil {
-		if err != nil {
+			return nil, err
-			// TODO: what error should be returned here?  It's not a
-			// field-level validation failure (the field is valid), and it's
-			// not really an internal error.
-			return nil, errors.NewInternalError(fmt.Errorf("failed to allocate a serviceIP: %v", err))
 		}
-		service.Spec.ClusterIP = ip.String()
-		releaseServiceIP = true
-	} else if helper.IsServiceIPSet(service) {
-		// Try to respect the requested IP.
-		if err := rs.serviceIPs.Allocate(net.ParseIP(service.Spec.ClusterIP)); err != nil {
-			// TODO: when validation becomes versioned, this gets more complicated.
-			el := field.ErrorList{field.Invalid(field.NewPath("spec", "clusterIP"), service.Spec.ClusterIP, err.Error())}
-			return nil, errors.NewInvalid(api.Kind("Service"), service.Name, el)
-		}
-		releaseServiceIP = true
 	}
 
 	nodePortOp := portallocator.StartOperation(rs.serviceNodePorts)
 	defer nodePortOp.Finish()
 
-	assignNodePorts := shouldAssignNodePorts(service)
+	if service.Spec.Type == api.ServiceTypeNodePort || service.Spec.Type == api.ServiceTypeLoadBalancer {
-	svcPortToNodePort := map[int]int{}
+		if err := rs.initNodePorts(service, nodePortOp); err != nil {
-	for i := range service.Spec.Ports {
+			return nil, err
-		servicePort := &service.Spec.Ports[i]
-		allocatedNodePort := svcPortToNodePort[int(servicePort.Port)]
-		if allocatedNodePort == 0 {
-			// This will only scan forward in the service.Spec.Ports list because any matches
-			// before the current port would have been found in svcPortToNodePort. This is really
-			// looking for any user provided values.
-			np := findRequestedNodePort(int(servicePort.Port), service.Spec.Ports)
-			if np != 0 {
-				err := nodePortOp.Allocate(np)
-				if err != nil {
-					// TODO: when validation becomes versioned, this gets more complicated.
-					el := field.ErrorList{field.Invalid(field.NewPath("spec", "ports").Index(i).Child("nodePort"), np, err.Error())}
-					return nil, errors.NewInvalid(api.Kind("Service"), service.Name, el)
-				}
-				servicePort.NodePort = int32(np)
-				svcPortToNodePort[int(servicePort.Port)] = np
-			} else if assignNodePorts {
-				nodePort, err := nodePortOp.AllocateNext()
-				if err != nil {
-					// TODO: what error should be returned here?  It's not a
-					// field-level validation failure (the field is valid), and it's
-					// not really an internal error.
-					return nil, errors.NewInternalError(fmt.Errorf("failed to allocate a nodePort: %v", err))
-				}
-				servicePort.NodePort = int32(nodePort)
-				svcPortToNodePort[int(servicePort.Port)] = nodePort
-			}
-		} else if int(servicePort.NodePort) != allocatedNodePort {
-			if servicePort.NodePort == 0 {
-				servicePort.NodePort = int32(allocatedNodePort)
-			} else {
-				err := nodePortOp.Allocate(int(servicePort.NodePort))
-				if err != nil {
-					// TODO: when validation becomes versioned, this gets more complicated.
-					el := field.ErrorList{field.Invalid(field.NewPath("spec", "ports").Index(i).Child("nodePort"), servicePort.NodePort, err.Error())}
-					return nil, errors.NewInvalid(api.Kind("Service"), service.Name, el)
-				}
-			}
 		}
 	}
 
@@ -395,11 +343,11 @@ func (rs *REST) Update(ctx genericapirequest.Context, name string, objInfo rest.
 	// Update service from NodePort or LoadBalancer to ExternalName or ClusterIP, should release NodePort if exists.
 	if (oldService.Spec.Type == api.ServiceTypeNodePort || oldService.Spec.Type == api.ServiceTypeLoadBalancer) &&
 		(service.Spec.Type == api.ServiceTypeExternalName || service.Spec.Type == api.ServiceTypeClusterIP) {
-		rs.releaseNodePort(oldService, nodePortOp)
+		rs.releaseNodePorts(oldService, nodePortOp)
 	}
 	// Update service from any type to NodePort or LoadBalancer, should update NodePort.
 	if service.Spec.Type == api.ServiceTypeNodePort || service.Spec.Type == api.ServiceTypeLoadBalancer {
-		if err := rs.updateNodePort(oldService, service, nodePortOp); err != nil {
+		if err := rs.updateNodePorts(oldService, service, nodePortOp); err != nil {
 			return nil, false, err
 		}
 	}
@@ -518,22 +466,6 @@ func CollectServiceNodePorts(service *api.Service) []int {
 	return servicePorts
 }
 
-func shouldAssignNodePorts(service *api.Service) bool {
-	switch service.Spec.Type {
-	case api.ServiceTypeLoadBalancer:
-		return true
-	case api.ServiceTypeNodePort:
-		return true
-	case api.ServiceTypeClusterIP:
-		return false
-	case api.ServiceTypeExternalName:
-		return false
-	default:
-		glog.Errorf("Unknown service type: %v", service.Spec.Type)
-		return false
-	}
-}
-
 // Loop through the service ports list, find one with the same port number and
 // NodePort specified, return this NodePort otherwise return 0.
 func findRequestedNodePort(port int, servicePorts []api.ServicePort) int {
@@ -596,7 +528,56 @@ func (rs *REST) initClusterIP(service *api.Service) (bool, error) {
 	return false, nil
 }
 
-func (rs *REST) updateNodePort(oldService, newService *api.Service, nodePortOp *portallocator.PortAllocationOperation) error {
+func (rs *REST) initNodePorts(service *api.Service, nodePortOp *portallocator.PortAllocationOperation) error {
+	svcPortToNodePort := map[int]int{}
+	for i := range service.Spec.Ports {
+		servicePort := &service.Spec.Ports[i]
+		allocatedNodePort := svcPortToNodePort[int(servicePort.Port)]
+		if allocatedNodePort == 0 {
+			// This will only scan forward in the service.Spec.Ports list because any matches
+			// before the current port would have been found in svcPortToNodePort. This is really
+			// looking for any user provided values.
+			np := findRequestedNodePort(int(servicePort.Port), service.Spec.Ports)
+			if np != 0 {
+				err := nodePortOp.Allocate(np)
+				if err != nil {
+					// TODO: when validation becomes versioned, this gets more complicated.
+					el := field.ErrorList{field.Invalid(field.NewPath("spec", "ports").Index(i).Child("nodePort"), np, err.Error())}
+					return errors.NewInvalid(api.Kind("Service"), service.Name, el)
+				}
+				servicePort.NodePort = int32(np)
+				svcPortToNodePort[int(servicePort.Port)] = np
+			} else {
+				nodePort, err := nodePortOp.AllocateNext()
+				if err != nil {
+					// TODO: what error should be returned here?  It's not a
+					// field-level validation failure (the field is valid), and it's
+					// not really an internal error.
+					return errors.NewInternalError(fmt.Errorf("failed to allocate a nodePort: %v", err))
+				}
+				servicePort.NodePort = int32(nodePort)
+				svcPortToNodePort[int(servicePort.Port)] = nodePort
+			}
+		} else if int(servicePort.NodePort) != allocatedNodePort {
+			// TODO(xiangpengzhao): do we need to allocate a new NodePort in this case?
+			// Note: the current implementation is better, because it saves a NodePort.
+			if servicePort.NodePort == 0 {
+				servicePort.NodePort = int32(allocatedNodePort)
+			} else {
+				err := nodePortOp.Allocate(int(servicePort.NodePort))
+				if err != nil {
+					// TODO: when validation becomes versioned, this gets more complicated.
+					el := field.ErrorList{field.Invalid(field.NewPath("spec", "ports").Index(i).Child("nodePort"), servicePort.NodePort, err.Error())}
+					return errors.NewInvalid(api.Kind("Service"), service.Name, el)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func (rs *REST) updateNodePorts(oldService, newService *api.Service, nodePortOp *portallocator.PortAllocationOperation) error {
 	oldNodePorts := CollectServiceNodePorts(oldService)
 
 	newNodePorts := []int{}
@@ -640,7 +621,7 @@ func (rs *REST) updateNodePort(oldService, newService *api.Service, nodePortOp *
 	return nil
 }
 
-func (rs *REST) releaseNodePort(service *api.Service, nodePortOp *portallocator.PortAllocationOperation) {
+func (rs *REST) releaseNodePorts(service *api.Service, nodePortOp *portallocator.PortAllocationOperation) {
 	nodePorts := CollectServiceNodePorts(service)
 
 	for _, nodePort := range nodePorts {

pkg/registry/core/service/rest_test.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/apiserver/pkg/registry/rest"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/helper"
 	"k8s.io/kubernetes/pkg/api/service"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/registry/core/service/ipallocator"
@@ -1379,10 +1380,194 @@ func TestInitClusterIP(t *testing.T) {
 		if test.name == "Allocate specified ClusterIP" && test.svc.Spec.ClusterIP != "1.2.3.4" {
 			t.Errorf("%q: expected ClusterIP %q, but got %q", test.name, "1.2.3.4", test.svc.Spec.ClusterIP)
 		}
+
+		if hasAllocatedIP {
+			if helper.IsServiceIPSet(test.svc) {
+				storage.serviceIPs.Release(net.ParseIP(test.svc.Spec.ClusterIP))
+			}
+		}
 	}
 }
 
-func TestUpdateNodePort(t *testing.T) {
+func TestInitNodePorts(t *testing.T) {
+	storage, _ := NewTestREST(t, nil)
+	nodePortOp := portallocator.StartOperation(storage.serviceNodePorts)
+	defer nodePortOp.Finish()
+
+	testCases := []struct {
+		name                     string
+		service                  *api.Service
+		expectSpecifiedNodePorts []int
+	}{
+		{
+			name: "Service doesn't have specified NodePort",
+			service: &api.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: api.ServiceSpec{
+					Selector: map[string]string{"bar": "baz"},
+					Type:     api.ServiceTypeNodePort,
+					Ports: []api.ServicePort{
+						{
+							Name:       "port-tcp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolTCP,
+						},
+					},
+				},
+			},
+			expectSpecifiedNodePorts: []int{},
+		},
+		{
+			name: "Service has one specified NodePort",
+			service: &api.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: api.ServiceSpec{
+					Selector: map[string]string{"bar": "baz"},
+					Type:     api.ServiceTypeNodePort,
+					Ports: []api.ServicePort{{
+						Name:       "port-tcp",
+						Port:       53,
+						TargetPort: intstr.FromInt(6502),
+						Protocol:   api.ProtocolTCP,
+						NodePort:   30053,
+					}},
+				},
+			},
+			expectSpecifiedNodePorts: []int{30053},
+		},
+		{
+			name: "Service has two same ports with different protocols and specifies same NodePorts",
+			service: &api.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: api.ServiceSpec{
+					Selector: map[string]string{"bar": "baz"},
+					Type:     api.ServiceTypeNodePort,
+					Ports: []api.ServicePort{
+						{
+							Name:       "port-tcp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolTCP,
+							NodePort:   30054,
+						},
+						{
+							Name:       "port-udp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolUDP,
+							NodePort:   30054,
+						},
+					},
+				},
+			},
+			expectSpecifiedNodePorts: []int{30054, 30054},
+		},
+		{
+			name: "Service has two same ports with different protocols and specifies different NodePorts",
+			service: &api.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: api.ServiceSpec{
+					Selector: map[string]string{"bar": "baz"},
+					Type:     api.ServiceTypeNodePort,
+					Ports: []api.ServicePort{
+						{
+							Name:       "port-tcp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolTCP,
+							NodePort:   30055,
+						},
+						{
+							Name:       "port-udp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolUDP,
+							NodePort:   30056,
+						},
+					},
+				},
+			},
+			expectSpecifiedNodePorts: []int{30055, 30056},
+		},
+		{
+			name: "Service has two different ports with different protocols and specifies different NodePorts",
+			service: &api.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: api.ServiceSpec{
+					Selector: map[string]string{"bar": "baz"},
+					Type:     api.ServiceTypeNodePort,
+					Ports: []api.ServicePort{
+						{
+							Name:       "port-tcp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolTCP,
+							NodePort:   30057,
+						},
+						{
+							Name:       "port-udp",
+							Port:       54,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolUDP,
+							NodePort:   30058,
+						},
+					},
+				},
+			},
+			expectSpecifiedNodePorts: []int{30057, 30058},
+		},
+		{
+			name: "Service has two same ports with different protocols but only specifies one NodePort",
+			service: &api.Service{
+				ObjectMeta: metav1.ObjectMeta{Name: "foo"},
+				Spec: api.ServiceSpec{
+					Selector: map[string]string{"bar": "baz"},
+					Type:     api.ServiceTypeNodePort,
+					Ports: []api.ServicePort{
+						{
+							Name:       "port-tcp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolTCP,
+							NodePort:   30059,
+						},
+						{
+							Name:       "port-udp",
+							Port:       53,
+							TargetPort: intstr.FromInt(6502),
+							Protocol:   api.ProtocolUDP,
+						},
+					},
+				},
+			},
+			expectSpecifiedNodePorts: []int{30059, 30059},
+		},
+	}
+
+	for _, test := range testCases {
+		err := storage.initNodePorts(test.service, nodePortOp)
+		if err != nil {
+			t.Errorf("%q: unexpected error: %v", test.name, err)
+			continue
+		}
+
+		serviceNodePorts := CollectServiceNodePorts(test.service)
+
+		if len(test.expectSpecifiedNodePorts) == 0 {
+			for _, nodePort := range serviceNodePorts {
+				if !storage.serviceNodePorts.Has(nodePort) {
+					t.Errorf("%q: unexpected NodePort %d, out of range", test.name, nodePort)
+				}
+			}
+		} else if !reflect.DeepEqual(serviceNodePorts, test.expectSpecifiedNodePorts) {
+			t.Errorf("%q: expected NodePorts %v, but got %v", test.name, test.expectSpecifiedNodePorts, serviceNodePorts)
+		}
+
+	}
+}
+
+func TestUpdateNodePorts(t *testing.T) {
 	storage, _ := NewTestREST(t, nil)
 	nodePortOp := portallocator.StartOperation(storage.serviceNodePorts)
 	defer nodePortOp.Finish()
@@ -1541,12 +1726,11 @@ func TestUpdateNodePort(t *testing.T) {
 	}
 
 	for _, test := range testCases {
-		err := storage.updateNodePort(test.oldService, test.newService, nodePortOp)
+		err := storage.updateNodePorts(test.oldService, test.newService, nodePortOp)
 		if err != nil {
 			t.Errorf("%q: unexpected error: %v", test.name, err)
 			continue
 		}
-		_ = nodePortOp.Commit()
 
 		serviceNodePorts := CollectServiceNodePorts(test.newService)