manager/allocator: lift portAllocator out of CNM

The port allocation logic does not depend on the network allocator
implementation in any meaningful way. It has no knowledge of the CNM
network allocator's state, and it does not need to change if the
network allocator changes. Allocating node ports is fundamentally a
seaparate concern from allocating network resources for services and
tasks. Therefore the low-level network allocator should not be
responsible for allocating both. Lift the port allocator into the
Allocator's network context as a sibling of the low-level network
allocator.

Signed-off-by: Cory Snider <[email protected]>

Author: corhere

manager/allocator/allocator_test.go

@@ -21,19 +21,6 @@ func init() {
 	retryInterval = 5 * time.Millisecond
 }
 
-// Temporary copy of constants from cnmallocator/portallocator.go
-// to allow tests to build before portallocator.go is moved into
-// this package.
-const (
-	// Start of the dynamic port range from which node ports will
-	// be allocated when the user did not specify a port.
-	dynamicPortStart = 30000
-
-	// End of the dynamic port range from which node ports will be
-	// allocated when the user did not specify a port.
-	dynamicPortEnd = 32767
-)
-
 func TestAllocator(t *testing.T) {
 	s := store.NewMemoryStore(nil)
 	assert.NotNil(t, s)

manager/allocator/cnmallocator/networkallocator.go

@@ -40,9 +40,6 @@ type cnmNetworkAllocator struct {
 	// The driver registry for all internal and external network drivers.
 	networkRegistry drvregistry.Networks
 
-	// The port allocator instance for allocating node ports
-	portAllocator *portAllocator
-
 	// Local network state used by cnmNetworkAllocator to do network management.
 	networks map[string]*network
 
@@ -108,8 +105,6 @@ func New(pg plugingetter.PluginGetter, netConfig *NetworkConfig) (networkallocat
 		tasks:    make(map[string]struct{}),
 		nodes:    make(map[string]map[string]struct{}),
 		pg:       pg,
-
-		portAllocator: newPortAllocator(),
 	}
 
 	for ntype, i := range initializers {
@@ -205,11 +200,8 @@ func (na *cnmNetworkAllocator) Deallocate(n *api.Network) error {
 }
 
 // AllocateService allocates all the network resources such as virtual
-// IP and ports needed by the service.
+// IP needed by the service.
 func (na *cnmNetworkAllocator) AllocateService(s *api.Service) (err error) {
-	if err = na.portAllocator.serviceAllocatePorts(s); err != nil {
-		return err
-	}
 	defer func() {
 		if err != nil {
 			na.DeallocateService(s)
@@ -296,7 +288,7 @@ networkLoop:
 }
 
 // DeallocateService de-allocates all the network resources such as
-// virtual IP and ports associated with the service.
+// virtual IP associated with the service.
 func (na *cnmNetworkAllocator) DeallocateService(s *api.Service) error {
 	if s.Endpoint == nil {
 		return nil
@@ -312,7 +304,6 @@ func (na *cnmNetworkAllocator) DeallocateService(s *api.Service) error {
 	}
 	s.Endpoint.VirtualIPs = nil
 
-	na.portAllocator.serviceDeallocatePorts(s)
 	delete(na.services, s.ID)
 
 	return nil
@@ -369,19 +360,8 @@ func (na *cnmNetworkAllocator) IsTaskAllocated(t *api.Task) bool {
 	return true
 }
 
-// HostPublishPortsNeedUpdate returns true if the passed service needs
-// allocations for its published ports in host (non ingress) mode
-func (na *cnmNetworkAllocator) HostPublishPortsNeedUpdate(s *api.Service) bool {
-	return na.portAllocator.hostPublishPortsNeedUpdate(s)
-}
-
 // IsServiceAllocated returns false if the passed service needs to have network resources allocated/updated.
 func (na *cnmNetworkAllocator) IsServiceAllocated(s *api.Service, flags ...func(*networkallocator.ServiceAllocationOpts)) bool {
-	var options networkallocator.ServiceAllocationOpts
-	for _, flag := range flags {
-		flag(&options)
-	}
-
 	specNetworks := serviceNetworks(s)
 
 	// If endpoint mode is VIP and allocator does not have the
@@ -443,10 +423,6 @@ func (na *cnmNetworkAllocator) IsServiceAllocated(s *api.Service, flags ...func(
 		}
 	}
 
-	if (s.Spec.Endpoint != nil && len(s.Spec.Endpoint.Ports) != 0) ||
-		(s.Endpoint != nil && len(s.Endpoint.Ports) != 0) {
-		return na.portAllocator.isPortsAllocatedOnInit(s, options.OnInit)
-	}
 	return true
 }
 

manager/allocator/cnmallocator/networkallocator_test.go

@@ -563,11 +563,7 @@ func TestAllocateService(t *testing.T) {
 
 	err = na.AllocateService(s)
 	assert.NoError(t, err)
-	assert.Equal(t, 2, len(s.Endpoint.Ports))
-	assert.True(t, s.Endpoint.Ports[0].PublishedPort >= dynamicPortStart &&
-		s.Endpoint.Ports[0].PublishedPort <= dynamicPortEnd)
-	assert.True(t, s.Endpoint.Ports[1].PublishedPort >= dynamicPortStart &&
-		s.Endpoint.Ports[1].PublishedPort <= dynamicPortEnd)
+	assert.Len(t, s.Endpoint.Ports, 0) // Network allocator is not responsible for allocating ports.
 
 	assert.Equal(t, 1, len(s.Endpoint.VirtualIPs))
 
@@ -579,94 +575,6 @@ func TestAllocateService(t *testing.T) {
 	assert.Equal(t, true, subnet.Contains(ip))
 }
 
-func TestAllocateServiceUserDefinedPorts(t *testing.T) {
-	na := newNetworkAllocator(t)
-	s := &api.Service{
-		ID: "testID1",
-		Spec: api.ServiceSpec{
-			Endpoint: &api.EndpointSpec{
-				Ports: []*api.PortConfig{
-					{
-						Name:          "some_tcp",
-						TargetPort:    1234,
-						PublishedPort: 1234,
-					},
-					{
-						Name:          "some_udp",
-						TargetPort:    1234,
-						PublishedPort: 1234,
-						Protocol:      api.ProtocolUDP,
-					},
-				},
-			},
-		},
-	}
-
-	err := na.AllocateService(s)
-	assert.NoError(t, err)
-	assert.Equal(t, 2, len(s.Endpoint.Ports))
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[1].PublishedPort)
-}
-
-func TestAllocateServiceConflictingUserDefinedPorts(t *testing.T) {
-	na := newNetworkAllocator(t)
-	s := &api.Service{
-		ID: "testID1",
-		Spec: api.ServiceSpec{
-			Endpoint: &api.EndpointSpec{
-				Ports: []*api.PortConfig{
-					{
-						Name:          "some_tcp",
-						TargetPort:    1234,
-						PublishedPort: 1234,
-					},
-					{
-						Name:          "some_other_tcp",
-						TargetPort:    1234,
-						PublishedPort: 1234,
-					},
-				},
-			},
-		},
-	}
-
-	err := na.AllocateService(s)
-	assert.Error(t, err)
-}
-
-func TestDeallocateServiceAllocate(t *testing.T) {
-	na := newNetworkAllocator(t)
-	s := &api.Service{
-		ID: "testID1",
-		Spec: api.ServiceSpec{
-			Endpoint: &api.EndpointSpec{
-				Ports: []*api.PortConfig{
-					{
-						Name:          "some_tcp",
-						TargetPort:    1234,
-						PublishedPort: 1234,
-					},
-				},
-			},
-		},
-	}
-
-	err := na.AllocateService(s)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(s.Endpoint.Ports))
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-
-	err = na.DeallocateService(s)
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(s.Endpoint.Ports))
-	// Allocate again.
-	err = na.AllocateService(s)
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(s.Endpoint.Ports))
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-}
-
 func TestDeallocateServiceAllocateIngressMode(t *testing.T) {
 	na := newNetworkAllocator(t)
 
@@ -705,8 +613,6 @@ func TestDeallocateServiceAllocateIngressMode(t *testing.T) {
 
 	err = na.AllocateService(s)
 	assert.NoError(t, err)
-	assert.Len(t, s.Endpoint.Ports, 1)
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
 	assert.Len(t, s.Endpoint.VirtualIPs, 1)
 
 	err = na.DeallocateService(s)
@@ -719,129 +625,7 @@ func TestDeallocateServiceAllocateIngressMode(t *testing.T) {
 
 	err = na.AllocateService(s)
 	assert.NoError(t, err)
-	assert.Len(t, s.Endpoint.Ports, 1)
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-	assert.Len(t, s.Endpoint.VirtualIPs, 1)
-}
-
-func TestServiceAddRemovePortsIngressMode(t *testing.T) {
-	na := newNetworkAllocator(t)
-
-	n := &api.Network{
-		ID: "testNetID1",
-		Spec: api.NetworkSpec{
-			Annotations: api.Annotations{
-				Name: "test",
-			},
-			Ingress: true,
-		},
-	}
-
-	err := na.Allocate(n)
-	assert.NoError(t, err)
-
-	s := &api.Service{
-		ID: "testID1",
-		Spec: api.ServiceSpec{
-			Endpoint: &api.EndpointSpec{
-				Ports: []*api.PortConfig{
-					{
-						Name:          "some_tcp",
-						TargetPort:    1234,
-						PublishedPort: 1234,
-						PublishMode:   api.PublishModeIngress,
-					},
-				},
-			},
-		},
-		Endpoint: &api.Endpoint{},
-	}
-
-	s.Endpoint.VirtualIPs = append(s.Endpoint.VirtualIPs,
-		&api.Endpoint_VirtualIP{NetworkID: n.ID})
-
-	err = na.AllocateService(s)
-	assert.NoError(t, err)
-	assert.Len(t, s.Endpoint.Ports, 1)
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
 	assert.Len(t, s.Endpoint.VirtualIPs, 1)
-	allocatedVIP := s.Endpoint.VirtualIPs[0].Addr
-
-	// Unpublish port
-	s.Spec.Endpoint.Ports = s.Spec.Endpoint.Ports[:0]
-	err = na.AllocateService(s)
-	assert.NoError(t, err)
-	assert.Len(t, s.Endpoint.Ports, 0)
-	assert.Len(t, s.Endpoint.VirtualIPs, 0)
-
-	// Publish port again and ensure VIP is not the same that was deallocated.
-	// Since IP allocation is serial we should  receive the next available IP.
-	s.Spec.Endpoint.Ports = append(s.Spec.Endpoint.Ports, &api.PortConfig{Name: "some_tcp",
-		TargetPort:    1234,
-		PublishedPort: 1234,
-		PublishMode:   api.PublishModeIngress,
-	})
-	s.Endpoint.VirtualIPs = append(s.Endpoint.VirtualIPs,
-		&api.Endpoint_VirtualIP{NetworkID: n.ID})
-	err = na.AllocateService(s)
-	assert.NoError(t, err)
-	assert.Len(t, s.Endpoint.Ports, 1)
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-	assert.Len(t, s.Endpoint.VirtualIPs, 1)
-	assert.NotEqual(t, allocatedVIP, s.Endpoint.VirtualIPs[0].Addr)
-}
-
-func TestServiceUpdate(t *testing.T) {
-	na1 := newNetworkAllocator(t)
-	na2 := newNetworkAllocator(t)
-	s := &api.Service{
-		ID: "testID1",
-		Spec: api.ServiceSpec{
-			Endpoint: &api.EndpointSpec{
-				Ports: []*api.PortConfig{
-					{
-						Name:          "some_tcp",
-						TargetPort:    1234,
-						PublishedPort: 1234,
-					},
-					{
-						Name:          "some_other_tcp",
-						TargetPort:    1235,
-						PublishedPort: 0,
-					},
-				},
-			},
-		},
-	}
-
-	err := na1.AllocateService(s)
-	assert.NoError(t, err)
-	assert.True(t, na1.IsServiceAllocated(s))
-	assert.Equal(t, 2, len(s.Endpoint.Ports))
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-	assert.NotEqual(t, 0, s.Endpoint.Ports[1].PublishedPort)
-
-	// Cache the secode node port
-	allocatedPort := s.Endpoint.Ports[1].PublishedPort
-
-	// Now allocate the same service in another allocator instance
-	err = na2.AllocateService(s)
-	assert.NoError(t, err)
-	assert.True(t, na2.IsServiceAllocated(s))
-	assert.Equal(t, 2, len(s.Endpoint.Ports))
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-	// Make sure we got the same port
-	assert.Equal(t, allocatedPort, s.Endpoint.Ports[1].PublishedPort)
-
-	s.Spec.Endpoint.Ports[1].PublishedPort = 1235
-	assert.False(t, na1.IsServiceAllocated(s))
-
-	err = na1.AllocateService(s)
-	assert.NoError(t, err)
-	assert.True(t, na1.IsServiceAllocated(s))
-	assert.Equal(t, 2, len(s.Endpoint.Ports))
-	assert.Equal(t, uint32(1234), s.Endpoint.Ports[0].PublishedPort)
-	assert.Equal(t, uint32(1235), s.Endpoint.Ports[1].PublishedPort)
 }
 
 func TestServiceNetworkUpdate(t *testing.T) {