By default block service proxy to external IP addresses.

Service proxy uses redirects to Pods instead of direct access.

pkg/features/kube_features.go

@@ -168,6 +168,13 @@ const (
 	// Enable nodes to exclude themselves from service load balancers
 	ServiceNodeExclusion utilfeature.Feature = "ServiceNodeExclusion"
 
+	// owner @brendandburns
+	// stable: v1.10
+	//
+	// Enable the service proxy to contact external IP addresses. Note this feature is present
+	// only for backward compatability, it will be removed in the 1.11 release.
+	ServiceProxyAllowExternalIPs utilfeature.Feature = "ServiceProxyAllowExternalIPs"
+
 	// owner: @jsafrane
 	// alpha: v1.9
 	//
@@ -268,4 +275,7 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	// inherited features from apiextensions-apiserver, relisted here to get a conflict if it is changed
 	// unintentionally on either side:
 	apiextensionsfeatures.CustomResourceValidation: {Default: true, PreRelease: utilfeature.Beta},
+
+	// backward compatability features that enable backwards compatability but should be removed soon.
+	ServiceProxyAllowExternalIPs: {Default: false, PreRelease: utilfeature.Beta},
 }

pkg/registry/core/rest/storage_core.go

@@ -172,7 +172,7 @@ func (c LegacyRESTStorageProvider) NewLegacyRESTStorage(restOptionsGetter generi
 
 	controllerStorage := controllerstore.NewStorage(restOptionsGetter)
 
-	serviceRest := service.NewStorage(serviceRegistry, endpointRegistry, ServiceClusterIPAllocator, ServiceNodePortAllocator, c.ProxyTransport)
+	serviceRest := service.NewStorage(serviceRegistry, endpointRegistry, podStorage.Pod, ServiceClusterIPAllocator, ServiceNodePortAllocator, c.ProxyTransport)
 
 	restStorageMap := map[string]rest.Storage{
 		"pods":             podStorage.Pod,

pkg/registry/core/service/BUILD

@@ -23,7 +23,9 @@ go_library(
         "//pkg/apis/core/helper:go_default_library",
         "//pkg/apis/core/validation:go_default_library",
         "//pkg/capabilities:go_default_library",
+        "//pkg/features:go_default_library",
         "//pkg/registry/core/endpoint:go_default_library",
+        "//pkg/registry/core/pod/storage:go_default_library",
         "//pkg/registry/core/service/ipallocator:go_default_library",
         "//pkg/registry/core/service/portallocator:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
@@ -39,6 +41,7 @@ go_library(
         "//vendor/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/registry/rest:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/names:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
     ],
 )
 
@@ -54,6 +57,7 @@ go_test(
         "//pkg/api/service:go_default_library",
         "//pkg/apis/core:go_default_library",
         "//pkg/apis/core/helper:go_default_library",
+        "//pkg/registry/core/pod/storage:go_default_library",
         "//pkg/registry/core/service/ipallocator:go_default_library",
         "//pkg/registry/core/service/portallocator:go_default_library",
         "//pkg/registry/registrytest:go_default_library",
@@ -65,7 +69,9 @@ go_test(
         "//vendor/k8s.io/apimachinery/pkg/util/net:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/rand:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/registry/generic:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/registry/rest:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/storage/etcd/testing:go_default_library",
     ],
 )
 

pkg/registry/core/service/rest.go

@@ -35,11 +35,14 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	apiservice "k8s.io/kubernetes/pkg/api/service"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
 	"k8s.io/kubernetes/pkg/apis/core/validation"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/registry/core/endpoint"
+	podstore "k8s.io/kubernetes/pkg/registry/core/pod/storage"
 	"k8s.io/kubernetes/pkg/registry/core/service/ipallocator"
 	"k8s.io/kubernetes/pkg/registry/core/service/portallocator"
 )
@@ -57,6 +60,7 @@ type REST struct {
 	serviceIPs       ipallocator.Interface
 	serviceNodePorts portallocator.Interface
 	proxyTransport   http.RoundTripper
+	pods             *podstore.REST
 }
 
 // ServiceNodePort includes protocol and port number of a service NodePort.
@@ -70,7 +74,7 @@ type ServiceNodePort struct {
 }
 
 // NewStorage returns a new REST.
-func NewStorage(registry Registry, endpoints endpoint.Registry, serviceIPs ipallocator.Interface,
+func NewStorage(registry Registry, endpoints endpoint.Registry, pods *podstore.REST, serviceIPs ipallocator.Interface,
 	serviceNodePorts portallocator.Interface, proxyTransport http.RoundTripper) *ServiceRest {
 	rest := &REST{
 		registry:         registry,
@@ -78,6 +82,7 @@ func NewStorage(registry Registry, endpoints endpoint.Registry, serviceIPs ipall
 		serviceIPs:       serviceIPs,
 		serviceNodePorts: serviceNodePorts,
 		proxyTransport:   proxyTransport,
+		pods:             pods,
 	}
 	return &ServiceRest{
 		Service: rest,
@@ -425,19 +430,57 @@ func (rs *REST) ResourceLocation(ctx genericapirequest.Context, id string) (*url
 		}
 		for i := range ss.Ports {
 			if ss.Ports[i].Name == portStr {
-				// Pick a random address.
-				ip := ss.Addresses[rand.Intn(len(ss.Addresses))].IP
-				port := int(ss.Ports[i].Port)
-				return &url.URL{
-					Scheme: svcScheme,
-					Host:   net.JoinHostPort(ip, strconv.Itoa(port)),
-				}, rs.proxyTransport, nil
+				addrSeed := rand.Intn(len(ss.Addresses))
+				// This is a little wonky, but it's expensive to test for the presence of a Pod
+				// So we repeatedly try at random and validate it, this means that for an invalid
+				// service with a lot of endpoints we're going to potentially make a lot of calls,
+				// but in the expected case we'll only make one.
+				for try := 0; try < len(ss.Addresses); try++ {
+					addr := ss.Addresses[(addrSeed+try)%len(ss.Addresses)]
+					if !utilfeature.DefaultFeatureGate.Enabled(features.ServiceProxyAllowExternalIPs) {
+						if err := isValidAddress(ctx, &addr, rs.pods); err != nil {
+							glog.Errorf("Address %v isn't valid (%v)", addr, err)
+							continue
+						}
+					}
+					ip := addr.IP
+					port := int(ss.Ports[i].Port)
+					return &url.URL{
+						Scheme: svcScheme,
+						Host:   net.JoinHostPort(ip, strconv.Itoa(port)),
+					}, rs.proxyTransport, nil
+				}
+				glog.Errorf("Failed to find a valid address, skipping subset: %v", ss)
 			}
 		}
 	}
 	return nil, nil, errors.NewServiceUnavailable(fmt.Sprintf("no endpoints available for service %q", id))
 }
 
+func isValidAddress(ctx genericapirequest.Context, addr *api.EndpointAddress, pods *podstore.REST) error {
+	if addr.TargetRef == nil {
+		return fmt.Errorf("Address has no target ref, skipping: %v", addr)
+	}
+	if genericapirequest.NamespaceValue(ctx) != addr.TargetRef.Namespace {
+		return fmt.Errorf("Address namespace doesn't match context namespace")
+	}
+	obj, err := pods.Get(ctx, addr.TargetRef.Name, &metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+	pod, ok := obj.(*api.Pod)
+	if !ok {
+		return fmt.Errorf("failed to cast to pod: %v", obj)
+	}
+	if pod == nil {
+		return fmt.Errorf("pod is missing, skipping (%s/%s)", addr.TargetRef.Namespace, addr.TargetRef.Name)
+	}
+	if pod.Status.PodIP != addr.IP {
+		return fmt.Errorf("pod ip doesn't match endpoint ip, skipping: %s vs %s (%s/%s)", pod.Status.PodIP, addr.IP, addr.TargetRef.Namespace, addr.TargetRef.Name)
+	}
+	return nil
+}
+
 // This is O(N), but we expect haystack to be small;
 // so small that we expect a linear search to be faster
 func containsNumber(haystack []int, needle int) bool {

pkg/registry/core/service/rest_test.go

@@ -30,10 +30,13 @@ import (
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/rand"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/registry/generic"
 	"k8s.io/apiserver/pkg/registry/rest"
+	etcdtesting "k8s.io/apiserver/pkg/storage/etcd/testing"
 	"k8s.io/kubernetes/pkg/api/service"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/helper"
+	podstore "k8s.io/kubernetes/pkg/registry/core/pod/storage"
 	"k8s.io/kubernetes/pkg/registry/core/service/ipallocator"
 	"k8s.io/kubernetes/pkg/registry/core/service/portallocator"
 	"k8s.io/kubernetes/pkg/registry/registrytest"
@@ -47,19 +50,40 @@ func generateRandomNodePort() int32 {
 	return int32(rand.IntnRange(30001, 30999))
 }
 
-func NewTestREST(t *testing.T, endpoints *api.EndpointsList) (*REST, *registrytest.ServiceRegistry) {
+func NewTestREST(t *testing.T, endpoints *api.EndpointsList) (*REST, *registrytest.ServiceRegistry, *etcdtesting.EtcdTestServer) {
+	return NewTestRESTWithPods(t, endpoints, nil)
+}
+
+func NewTestRESTWithPods(t *testing.T, endpoints *api.EndpointsList, pods *api.PodList) (*REST, *registrytest.ServiceRegistry, *etcdtesting.EtcdTestServer) {
 	registry := registrytest.NewServiceRegistry()
 	endpointRegistry := &registrytest.EndpointRegistry{
 		Endpoints: endpoints,
 	}
+	etcdStorage, server := registrytest.NewEtcdStorage(t, "")
+	restOptions := generic.RESTOptions{
+		StorageConfig:           etcdStorage,
+		Decorator:               generic.UndecoratedStorage,
+		DeleteCollectionWorkers: 3,
+		ResourcePrefix:          "pods",
+	}
+	podStorage := podstore.NewStorage(restOptions, nil, nil, nil)
+	if pods != nil && pods.Items != nil {
+		ctx := genericapirequest.NewDefaultContext()
+		for ix := range pods.Items {
+			key, _ := podStorage.Pod.KeyFunc(ctx, pods.Items[ix].Name)
+			if err := podStorage.Pod.Storage.Create(ctx, key, &pods.Items[ix], nil, 0); err != nil {
+				t.Fatalf("Couldn't create pod: %v", err)
+			}
+		}
+	}
 	r := ipallocator.NewCIDRRange(makeIPNet(t))
 
 	portRange := utilnet.PortRange{Base: 30000, Size: 1000}
 	portAllocator := portallocator.NewPortAllocator(portRange)
 
-	storage := NewStorage(registry, endpointRegistry, r, portAllocator, nil)
+	storage := NewStorage(registry, endpointRegistry, podStorage.Pod, r, portAllocator, nil)
 
-	return storage.Service, registry
+	return storage.Service, registry, server
 }
 
 func makeIPNet(t *testing.T) *net.IPNet {
@@ -89,7 +113,8 @@ func releaseServiceNodePorts(t *testing.T, ctx genericapirequest.Context, svcNam
 }
 
 func TestServiceRegistryCreate(t *testing.T) {
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
@@ -136,7 +161,9 @@ func TestServiceRegistryCreate(t *testing.T) {
 }
 
 func TestServiceRegistryCreateMultiNodePortsService(t *testing.T) {
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
+
 	testCases := []struct {
 		svc             *api.Service
 		name            string
@@ -264,7 +291,8 @@ func TestServiceRegistryCreateMultiNodePortsService(t *testing.T) {
 }
 
 func TestServiceStorageValidatesCreate(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	failureCases := map[string]api.Service{
 		"empty ID": {
 			ObjectMeta: metav1.ObjectMeta{Name: ""},
@@ -317,7 +345,9 @@ func TestServiceStorageValidatesCreate(t *testing.T) {
 
 func TestServiceRegistryUpdate(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
+
 	svc, err := registry.CreateService(ctx, &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: "1", Namespace: metav1.NamespaceDefault},
 		Spec: api.ServiceSpec{
@@ -368,7 +398,8 @@ func TestServiceRegistryUpdate(t *testing.T) {
 
 func TestServiceStorageValidatesUpdate(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	registry.CreateService(ctx, &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
 		Spec: api.ServiceSpec{
@@ -420,7 +451,8 @@ func TestServiceStorageValidatesUpdate(t *testing.T) {
 
 func TestServiceRegistryExternalService(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
 		Spec: api.ServiceSpec{
@@ -458,7 +490,8 @@ func TestServiceRegistryExternalService(t *testing.T) {
 
 func TestServiceRegistryDelete(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
 		Spec: api.ServiceSpec{
@@ -480,7 +513,8 @@ func TestServiceRegistryDelete(t *testing.T) {
 
 func TestServiceRegistryDeleteExternal(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
 		Spec: api.ServiceSpec{
@@ -502,7 +536,8 @@ func TestServiceRegistryDeleteExternal(t *testing.T) {
 
 func TestServiceRegistryUpdateExternalService(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	// Create non-external load balancer.
 	svc1 := &api.Service{
@@ -540,7 +575,8 @@ func TestServiceRegistryUpdateExternalService(t *testing.T) {
 
 func TestServiceRegistryUpdateMultiPortExternalService(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	// Create external load balancer.
 	svc1 := &api.Service{
@@ -577,7 +613,8 @@ func TestServiceRegistryUpdateMultiPortExternalService(t *testing.T) {
 
 func TestServiceRegistryGet(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	registry.CreateService(ctx, &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
 		Spec: api.ServiceSpec{
@@ -594,13 +631,27 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
 	endpoints := &api.EndpointsList{
 		Items: []api.Endpoints{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bad",
+					Namespace: metav1.NamespaceDefault,
+				},
+				Subsets: []api.EndpointSubset{{
+					Addresses: []api.EndpointAddress{
+						{IP: "1.2.3.4", TargetRef: &api.ObjectReference{Name: "foo", Namespace: "doesn't exist"}},
+						{IP: "1.2.3.4", TargetRef: &api.ObjectReference{Name: "doesn't exist", Namespace: metav1.NamespaceDefault}},
+						{IP: "23.2.3.4", TargetRef: &api.ObjectReference{Name: "foo", Namespace: metav1.NamespaceDefault}},
+					},
+					Ports: []api.EndpointPort{{Name: "", Port: 80}, {Name: "p", Port: 93}},
+				}},
+			},
 			{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "foo",
 					Namespace: metav1.NamespaceDefault,
 				},
 				Subsets: []api.EndpointSubset{{
-					Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+					Addresses: []api.EndpointAddress{{IP: "1.2.3.4", TargetRef: &api.ObjectReference{Name: "foo", Namespace: metav1.NamespaceDefault}}},
 					Ports:     []api.EndpointPort{{Name: "", Port: 80}, {Name: "p", Port: 93}},
 				}},
 			},
@@ -613,30 +664,65 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
 					Addresses: []api.EndpointAddress{},
 					Ports:     []api.EndpointPort{{Name: "", Port: 80}, {Name: "p", Port: 93}},
 				}, {
-					Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+					Addresses: []api.EndpointAddress{{IP: "1.2.3.4", TargetRef: &api.ObjectReference{Name: "foo", Namespace: metav1.NamespaceDefault}}},
 					Ports:     []api.EndpointPort{{Name: "", Port: 80}, {Name: "p", Port: 93}},
 				}, {
-					Addresses: []api.EndpointAddress{{IP: "1.2.3.5"}},
+					Addresses: []api.EndpointAddress{{IP: "1.2.3.5", TargetRef: &api.ObjectReference{Name: "bar", Namespace: metav1.NamespaceDefault}}},
 					Ports:     []api.EndpointPort{},
 				}},
 			},
 		},
 	}
-	storage, registry := NewTestREST(t, endpoints)
-	registry.CreateService(ctx, &api.Service{
-		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
-		Spec: api.ServiceSpec{
-			Selector: map[string]string{"bar": "baz"},
-			Ports: []api.ServicePort{
-				// Service port 9393 should route to endpoint port "p", which is port 93
-				{Name: "p", Port: 9393, TargetPort: intstr.FromString("p")},
-
-				// Service port 93 should route to unnamed endpoint port, which is port 80
-				// This is to test that the service port definition is used when determining resource location
-				{Name: "", Port: 93, TargetPort: intstr.FromInt(80)},
+	pods := &api.PodList{
+		Items: []api.Pod{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "foo",
+					Namespace: metav1.NamespaceDefault,
+				},
+				Spec: api.PodSpec{
+					RestartPolicy: "Always",
+					DNSPolicy:     "Default",
+					Containers:    []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent, TerminationMessagePolicy: api.TerminationMessageReadFile}},
+				},
+				Status: api.PodStatus{
+					PodIP: "1.2.3.4",
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bar",
+					Namespace: metav1.NamespaceDefault,
+				},
+				Spec: api.PodSpec{
+					RestartPolicy: "Always",
+					DNSPolicy:     "Default",
+					Containers:    []api.Container{{Name: "bar", Image: "test", ImagePullPolicy: api.PullIfNotPresent, TerminationMessagePolicy: api.TerminationMessageReadFile}},
+				},
+				Status: api.PodStatus{
+					PodIP: "1.2.3.5",
+				},
 			},
 		},
-	}, rest.ValidateAllObjectFunc)
+	}
+	storage, registry, server := NewTestRESTWithPods(t, endpoints, pods)
+	defer server.Terminate(t)
+	for _, name := range []string{"foo", "bad"} {
+		registry.CreateService(ctx, &api.Service{
+			ObjectMeta: metav1.ObjectMeta{Name: name},
+			Spec: api.ServiceSpec{
+				Selector: map[string]string{"bar": "baz"},
+				Ports: []api.ServicePort{
+					// Service port 9393 should route to endpoint port "p", which is port 93
+					{Name: "p", Port: 9393, TargetPort: intstr.FromString("p")},
+
+					// Service port 93 should route to unnamed endpoint port, which is port 80
+					// This is to test that the service port definition is used when determining resource location
+					{Name: "", Port: 93, TargetPort: intstr.FromInt(80)},
+				},
+			},
+		}, rest.ValidateAllObjectFunc)
+	}
 	redirector := rest.Redirector(storage)
 
 	// Test a simple id.
@@ -709,11 +795,18 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
 	if _, _, err = redirector.ResourceLocation(ctx, "bar"); err == nil {
 		t.Errorf("unexpected nil error")
 	}
+
+	// Test a simple id.
+	_, _, err = redirector.ResourceLocation(ctx, "bad")
+	if err == nil {
+		t.Errorf("Unexpected nil error")
+	}
 }
 
 func TestServiceRegistryList(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	registry.CreateService(ctx, &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: metav1.NamespaceDefault},
 		Spec: api.ServiceSpec{
@@ -744,7 +837,8 @@ func TestServiceRegistryList(t *testing.T) {
 }
 
 func TestServiceRegistryIPAllocation(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	svc1 := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
@@ -826,7 +920,8 @@ func TestServiceRegistryIPAllocation(t *testing.T) {
 }
 
 func TestServiceRegistryIPReallocation(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	svc1 := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo"},
@@ -881,7 +976,8 @@ func TestServiceRegistryIPReallocation(t *testing.T) {
 }
 
 func TestServiceRegistryIPUpdate(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: "1"},
@@ -935,7 +1031,8 @@ func TestServiceRegistryIPUpdate(t *testing.T) {
 }
 
 func TestServiceRegistryIPLoadBalancer(t *testing.T) {
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: "1"},
@@ -974,7 +1071,8 @@ func TestServiceRegistryIPLoadBalancer(t *testing.T) {
 }
 
 func TestUpdateServiceWithConflictingNamespace(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	service := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "test", Namespace: "not-default"},
 	}
@@ -995,7 +1093,8 @@ func TestUpdateServiceWithConflictingNamespace(t *testing.T) {
 // and type is LoadBalancer.
 func TestServiceRegistryExternalTrafficHealthCheckNodePortAllocation(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "external-lb-esipp"},
 		Spec: api.ServiceSpec{
@@ -1034,7 +1133,8 @@ func TestServiceRegistryExternalTrafficHealthCheckNodePortAllocation(t *testing.
 func TestServiceRegistryExternalTrafficHealthCheckNodePortUserAllocation(t *testing.T) {
 	randomNodePort := generateRandomNodePort()
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "external-lb-esipp"},
 		Spec: api.ServiceSpec{
@@ -1076,7 +1176,8 @@ func TestServiceRegistryExternalTrafficHealthCheckNodePortUserAllocation(t *test
 // Validate that the service creation fails when the requested port number is -1.
 func TestServiceRegistryExternalTrafficHealthCheckNodePortNegative(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "external-lb-esipp"},
 		Spec: api.ServiceSpec{
@@ -1102,7 +1203,8 @@ func TestServiceRegistryExternalTrafficHealthCheckNodePortNegative(t *testing.T)
 // Validate that the health check nodePort is not allocated when ExternalTrafficPolicy is set to Global.
 func TestServiceRegistryExternalTrafficGlobal(t *testing.T) {
 	ctx := genericapirequest.NewDefaultContext()
-	storage, registry := NewTestREST(t, nil)
+	storage, registry, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	svc := &api.Service{
 		ObjectMeta: metav1.ObjectMeta{Name: "external-lb-esipp"},
 		Spec: api.ServiceSpec{
@@ -1137,7 +1239,8 @@ func TestServiceRegistryExternalTrafficGlobal(t *testing.T) {
 }
 
 func TestInitClusterIP(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 
 	testCases := []struct {
 		name            string
@@ -1228,7 +1331,8 @@ func TestInitClusterIP(t *testing.T) {
 }
 
 func TestInitNodePorts(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	nodePortOp := portallocator.StartOperation(storage.serviceNodePorts)
 	defer nodePortOp.Finish()
 
@@ -1409,7 +1513,8 @@ func TestInitNodePorts(t *testing.T) {
 }
 
 func TestUpdateNodePorts(t *testing.T) {
-	storage, _ := NewTestREST(t, nil)
+	storage, _, server := NewTestREST(t, nil)
+	defer server.Terminate(t)
 	nodePortOp := portallocator.StartOperation(storage.serviceNodePorts)
 	defer nodePortOp.Finish()