Merge pull request #5556 from abhgupta/abhgupta-dev

Headless Services: Adding option to specify None for PortalIP

cluster/addons/dns/kube2sky/kube2sky.go

@@ -51,6 +51,12 @@ func removeDNS(record string, etcdClient *etcd.Client) error {
 }
 
 func addDNS(record string, service *kapi.Service, etcdClient *etcd.Client) error {
+	// if PortalIP is not set, a DNS entry should not be created
+	if !kapi.IsServiceIPSet(service) {
+		log.Printf("Skipping dns record for headless service: %s\n", service.Name)
+		return nil
+	}
+
 	svc := skymsg.Service{
 		Host:     service.Spec.PortalIP,
 		Port:     service.Spec.Port,

pkg/api/helpers.go

@@ -84,3 +84,14 @@ func IsStandardResourceName(str string) bool {
 func NewDeleteOptions(grace int64) *DeleteOptions {
 	return &DeleteOptions{GracePeriodSeconds: &grace}
 }
+
+// this function aims to check if the service portal IP is set or not
+// the objective is not to perform validation here
+func IsServiceIPSet(service *Service) bool {
+	return service.Spec.PortalIP != PortalIPNone && service.Spec.PortalIP != ""
+}
+
+// this function aims to check if the service portal IP is requested or not
+func IsServiceIPRequested(service *Service) bool {
+	return service.Spec.PortalIP == ""
+}

pkg/api/types.go

@@ -710,6 +710,12 @@ type ReplicationControllerList struct {
 	Items []ReplicationController `json:"items"`
 }
 
+const (
+	// PortalIPNone - do not assign a portal IP
+	// no proxying required and no environment variables should be created for pods
+	PortalIPNone = "None"
+)
+
 // ServiceList holds a list of services.
 type ServiceList struct {
 	TypeMeta `json:",inline"`
@@ -749,6 +755,8 @@ type ServiceSpec struct {
 	// PortalIP is usually assigned by the master.  If specified by the user
 	// we will try to respect it or else fail the request.  This field can
 	// not be changed by updates.
+	// Valid values are None, empty string (""), or a valid IP address
+	// None can be specified for headless services when proxying is not required
 	PortalIP string `json:"portalIP,omitempty"`
 
 	// CreateExternalLoadBalancer indicates whether a load balancer should be created for this service.

pkg/api/v1beta1/types.go

@@ -578,6 +578,12 @@ const (
 	AffinityTypeNone AffinityType = "None"
 )
 
+const (
+	// PortalIPNone - do not assign a portal IP
+	// no proxying required and no environment variables should be created for pods
+	PortalIPNone = "None"
+)
+
 // ServiceList holds a list of services.
 type ServiceList struct {
 	TypeMeta `json:",inline"`
@@ -615,7 +621,9 @@ type Service struct {
 	// PortalIP is usually assigned by the master.  If specified by the user
 	// we will try to respect it or else fail the request.  This field can
 	// not be changed by updates.
-	PortalIP string `json:"portalIP,omitempty" description:"IP address of the service; usually assigned by the system; if specified, it will be allocated to the service if unused, and creation of the service will fail otherwise; cannot be updated"`
+	// Valid values are None, empty string (""), or a valid IP address
+	// None can be specified for headless services when proxying is not required
+	PortalIP string `json:"portalIP,omitempty" description:"IP address of the service; usually assigned by the system; if specified, it will be allocated to the service if unused, and creation of the service will fail otherwise; cannot be updated; 'None' can be specified for a headless service when proxying is not required"`
 
 	// DEPRECATED: has no implementation.
 	ProxyPort int `json:"proxyPort,omitempty" description:"if non-zero, a pre-allocated host port used for this service by the proxy on each node; assigned by the master and ignored on input"`

pkg/api/v1beta2/types.go

@@ -581,6 +581,12 @@ const (
 	AffinityTypeNone AffinityType = "None"
 )
 
+const (
+	// PortalIPNone - do not assign a portal IP
+	// no proxying required and no environment variables should be created for pods
+	PortalIPNone = "None"
+)
+
 // ServiceList holds a list of services.
 type ServiceList struct {
 	TypeMeta `json:",inline"`
@@ -620,7 +626,9 @@ type Service struct {
 	// PortalIP is usually assigned by the master.  If specified by the user
 	// we will try to respect it or else fail the request.  This field can
 	// not be changed by updates.
-	PortalIP string `json:"portalIP,omitempty" description:"IP address of the service; usually assigned by the system; if specified, it will be allocated to the service if unused, and creation of the service will fail otherwise; cannot be updated"`
+	// Valid values are None, empty string (""), or a valid IP address
+	// None can be specified for headless services when proxying is not required
+	PortalIP string `json:"portalIP,omitempty" description:"IP address of the service; usually assigned by the system; if specified, it will be allocated to the service if unused, and creation of the service will fail otherwise; cannot be updated; 'None' can be specified for a headless service when proxying is not required"`
 
 	// DEPRECATED: has no implementation.
 	ProxyPort int `json:"proxyPort,omitempty" description:"if non-zero, a pre-allocated host port used for this service by the proxy on each node; assigned by the master and ignored on input"`

pkg/api/v1beta3/types.go

@@ -743,7 +743,9 @@ type ServiceSpec struct {
 	// PortalIP is usually assigned by the master.  If specified by the user
 	// we will try to respect it or else fail the request.  This field can
 	// not be changed by updates.
-	PortalIP string `json:"portalIP,omitempty description: IP address of the service; usually assigned by the system; if specified, it will be allocated to the service if unused, and creation of the service will fail otherwise; cannot be updated"`
+	// Valid values are None, empty string (""), or a valid IP address
+	// None can be specified for headless services when proxying is not required
+	PortalIP string `json:"portalIP,omitempty description: IP address of the service; usually assigned by the system; if specified, it will be allocated to the service if unused, and creation of the service will fail otherwise; cannot be updated; 'None' can be specified for a headless service when proxying is not required"`
 
 	// CreateExternalLoadBalancer indicates whether a load balancer should be created for this service.
 	CreateExternalLoadBalancer bool `json:"createExternalLoadBalancer,omitempty" description:"set up a cloud-provider-specific load balancer on an external IP"`
@@ -775,6 +777,12 @@ type Service struct {
 	Status ServiceStatus `json:"status,omitempty" description:"most recently observed status of the service; populated by the system, read-only; https://github.com/GoogleCloudPlatform/kubernetes/blob/master/docs/api-conventions.md#spec-and-status"`
 }
 
+const (
+	// PortalIPNone - do not assign a portal IP
+	// no proxying required and no environment variables should be created for pods
+	PortalIPNone = "None"
+)
+
 // ServiceList holds a list of services.
 type ServiceList struct {
 	TypeMeta `json:",inline"`

pkg/api/validation/validation.go

@@ -18,6 +18,7 @@ package validation
 
 import (
 	"fmt"
+	"net"
 	"path"
 	"strings"
 
@@ -751,6 +752,12 @@ func ValidateService(service *api.Service) errs.ValidationErrorList {
 		allErrs = append(allErrs, errs.NewFieldNotSupported("spec.sessionAffinity", service.Spec.SessionAffinity))
 	}
 
+	if api.IsServiceIPSet(service) {
+		if ip := net.ParseIP(service.Spec.PortalIP); ip == nil {
+			allErrs = append(allErrs, errs.NewFieldInvalid("spec.portalIP", service.Spec.PortalIP, "portalIP should be empty, 'None', or a valid IP address"))
+		}
+	}
+
 	return allErrs
 }
 
@@ -760,8 +767,8 @@ func ValidateServiceUpdate(oldService, service *api.Service) errs.ValidationErro
 	allErrs = append(allErrs, ValidateObjectMetaUpdate(&oldService.ObjectMeta, &service.ObjectMeta).Prefix("metadata")...)
 
 	// TODO: PortalIP should be a Status field, since the system can set a value != to the user's value
-	// PortalIP can only be set, not unset.
+	// once PortalIP is set, it cannot be unset.
-	if oldService.Spec.PortalIP != "" && service.Spec.PortalIP != oldService.Spec.PortalIP {
+	if api.IsServiceIPSet(oldService) && service.Spec.PortalIP != oldService.Spec.PortalIP {
 		allErrs = append(allErrs, errs.NewFieldInvalid("spec.portalIP", service.Spec.PortalIP, "field is immutable"))
 	}
 

pkg/api/validation/validation_test.go

@@ -1134,6 +1134,13 @@ func TestValidateService(t *testing.T) {
 			},
 			numErrs: 1,
 		},
+		{
+			name: "invalid portal ip",
+			makeSvc: func(s *api.Service) {
+				s.Spec.PortalIP = "invalid"
+			},
+			numErrs: 1,
+		},
 		{
 			name: "missing port",
 			makeSvc: func(s *api.Service) {
@@ -1191,6 +1198,20 @@ func TestValidateService(t *testing.T) {
 			},
 			numErrs: 0,
 		},
+		{
+			name: "valid portal ip - none ",
+			makeSvc: func(s *api.Service) {
+				s.Spec.PortalIP = "None"
+			},
+			numErrs: 0,
+		},
+		{
+			name: "valid portal ip - empty",
+			makeSvc: func(s *api.Service) {
+				s.Spec.PortalIP = ""
+			},
+			numErrs: 0,
+		},
 	}
 
 	for _, tc := range testCases {

pkg/kubelet/envvars/envvars.go

@@ -30,6 +30,12 @@ import (
 func FromServices(services *api.ServiceList) []api.EnvVar {
 	var result []api.EnvVar
 	for _, service := range services.Items {
+		// ignore services where PortalIP is "None" or empty
+		// the services passed to this method should be pre-filtered
+		// only services that have the portal IP set should be included here
+		if !api.IsServiceIPSet(&service) {
+			continue
+		}
 		// Host
 		name := makeEnvVariableName(service.Name) + "_SERVICE_HOST"
 		result = append(result, api.EnvVar{Name: name, Value: service.Spec.PortalIP})

pkg/kubelet/envvars/envvars_test.go

@@ -54,6 +54,24 @@ func TestFromServices(t *testing.T) {
 					PortalIP: "9.8.7.6",
 				},
 			},
+			{
+				ObjectMeta: api.ObjectMeta{Name: "svrc-portalip-none"},
+				Spec: api.ServiceSpec{
+					Port:     8082,
+					Selector: map[string]string{"bar": "baz"},
+					Protocol: "TCP",
+					PortalIP: "None",
+				},
+			},
+			{
+				ObjectMeta: api.ObjectMeta{Name: "svrc-portalip-empty"},
+				Spec: api.ServiceSpec{
+					Port:     8082,
+					Selector: map[string]string{"bar": "baz"},
+					Protocol: "TCP",
+					PortalIP: "",
+				},
+			},
 		},
 	}
 	vars := envvars.FromServices(&sl)

pkg/kubelet/kubelet.go

@@ -845,6 +845,10 @@ func (kl *Kubelet) getServiceEnvVarMap(ns string) (map[string]string, error) {
 
 	// project the services in namespace ns onto the master services
 	for _, service := range services.Items {
+		// ignore services where PortalIP is "None" or empty
+		if !api.IsServiceIPSet(&service) {
+			continue
+		}
 		serviceName := service.Name
 
 		switch service.Namespace {

pkg/kubelet/kubelet_test.go

@@ -1828,6 +1828,20 @@ func TestMakeEnvironmentVariables(t *testing.T) {
 				PortalIP: "1.2.3.2",
 			},
 		},
+		{
+			ObjectMeta: api.ObjectMeta{Name: "kubernetes-ro", Namespace: api.NamespaceDefault},
+			Spec: api.ServiceSpec{
+				Port:     8082,
+				PortalIP: "None",
+			},
+		},
+		{
+			ObjectMeta: api.ObjectMeta{Name: "kubernetes-ro", Namespace: api.NamespaceDefault},
+			Spec: api.ServiceSpec{
+				Port:     8082,
+				PortalIP: "",
+			},
+		},
 		{
 			ObjectMeta: api.ObjectMeta{Name: "test", Namespace: "test1"},
 			Spec: api.ServiceSpec{
@@ -1849,6 +1863,19 @@ func TestMakeEnvironmentVariables(t *testing.T) {
 				PortalIP: "1.2.3.5",
 			},
 		},
+		{
+			ObjectMeta: api.ObjectMeta{Name: "test", Namespace: "test2"},
+			Spec: api.ServiceSpec{
+				Port:     8085,
+				PortalIP: "None",
+			},
+		},
+		{
+			ObjectMeta: api.ObjectMeta{Name: "test", Namespace: "test2"},
+			Spec: api.ServiceSpec{
+				Port: 8085,
+			},
+		},
 		{
 			ObjectMeta: api.ObjectMeta{Name: "kubernetes", Namespace: "kubernetes"},
 			Spec: api.ServiceSpec{
@@ -1870,6 +1897,20 @@ func TestMakeEnvironmentVariables(t *testing.T) {
 				PortalIP: "1.2.3.8",
 			},
 		},
+		{
+			ObjectMeta: api.ObjectMeta{Name: "not-special", Namespace: "kubernetes"},
+			Spec: api.ServiceSpec{
+				Port:     8088,
+				PortalIP: "None",
+			},
+		},
+		{
+			ObjectMeta: api.ObjectMeta{Name: "not-special", Namespace: "kubernetes"},
+			Spec: api.ServiceSpec{
+				Port:     8088,
+				PortalIP: "",
+			},
+		},
 	}
 
 	testCases := []struct {

pkg/proxy/proxier.go

@@ -468,6 +468,10 @@ func (proxier *Proxier) OnUpdate(services []api.Service) {
 	glog.V(4).Infof("Received update notice: %+v", services)
 	activeServices := make(map[types.NamespacedName]bool) // use a map as a set
 	for _, service := range services {
+		// if PortalIP is "None" or empty, skip proxying
+		if !api.IsServiceIPSet(&service) {
+			continue
+		}
 		serviceName := types.NamespacedName{service.Namespace, service.Name}
 		activeServices[serviceName] = true
 		info, exists := proxier.getServiceInfo(serviceName)

pkg/proxy/proxier_test.go

@@ -400,7 +400,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
 	}
 	waitForNumProxyLoops(t, p, 0)
 	p.OnUpdate([]api.Service{
-		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "TCP"}, Status: api.ServiceStatus{}},
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "TCP", PortalIP: "1.2.3.4"}, Status: api.ServiceStatus{}},
 	})
 	svcInfo, exists := p.getServiceInfo(service)
 	if !exists {
@@ -440,7 +440,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
 	}
 	waitForNumProxyLoops(t, p, 0)
 	p.OnUpdate([]api.Service{
-		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "UDP"}, Status: api.ServiceStatus{}},
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "UDP", PortalIP: "1.2.3.4"}, Status: api.ServiceStatus{}},
 	})
 	svcInfo, exists := p.getServiceInfo(service)
 	if !exists {
@@ -471,7 +471,7 @@ func TestTCPProxyUpdatePort(t *testing.T) {
 	waitForNumProxyLoops(t, p, 1)
 
 	p.OnUpdate([]api.Service{
-		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: 99, Protocol: "TCP"}, Status: api.ServiceStatus{}},
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: 99, Protocol: "TCP", PortalIP: "1.2.3.4"}, Status: api.ServiceStatus{}},
 	})
 	// Wait for the socket to actually get free.
 	if err := waitForClosedPortTCP(p, svcInfo.proxyPort); err != nil {
@@ -507,7 +507,7 @@ func TestUDPProxyUpdatePort(t *testing.T) {
 	waitForNumProxyLoops(t, p, 1)
 
 	p.OnUpdate([]api.Service{
-		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: 99, Protocol: "UDP"}, Status: api.ServiceStatus{}},
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: 99, Protocol: "UDP", PortalIP: "1.2.3.4"}, Status: api.ServiceStatus{}},
 	})
 	// Wait for the socket to actually get free.
 	if err := waitForClosedPortUDP(p, svcInfo.proxyPort); err != nil {
@@ -521,4 +521,59 @@ func TestUDPProxyUpdatePort(t *testing.T) {
 	waitForNumProxyLoops(t, p, 1)
 }
 
+func TestProxyUpdatePortal(t *testing.T) {
+	lb := NewLoadBalancerRR()
+	service := types.NewNamespacedNameOrDie("testnamespace", "echo")
+	lb.OnUpdate([]api.Endpoints{
+		{
+			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
+			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: tcpServerPort}},
+		},
+	})
+
+	p := CreateProxier(lb, net.ParseIP("0.0.0.0"), &fakeIptables{}, net.ParseIP("127.0.0.1"))
+	waitForNumProxyLoops(t, p, 0)
+
+	svcInfo, err := p.addServiceOnPort(service, "TCP", 0, time.Second)
+	if err != nil {
+		t.Fatalf("error adding new service: %#v", err)
+	}
+	testEchoTCP(t, "127.0.0.1", svcInfo.proxyPort)
+	waitForNumProxyLoops(t, p, 1)
+
+	p.OnUpdate([]api.Service{
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "TCP"}, Status: api.ServiceStatus{}},
+	})
+	_, exists := p.getServiceInfo(service)
+	if exists {
+		t.Fatalf("service without portalIP should not be included in the proxy")
+	}
+
+	p.OnUpdate([]api.Service{
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "TCP", PortalIP: ""}, Status: api.ServiceStatus{}},
+	})
+	_, exists = p.getServiceInfo(service)
+	if exists {
+		t.Fatalf("service with empty portalIP should not be included in the proxy")
+	}
+
+	p.OnUpdate([]api.Service{
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "TCP", PortalIP: "None"}, Status: api.ServiceStatus{}},
+	})
+	_, exists = p.getServiceInfo(service)
+	if exists {
+		t.Fatalf("service with 'None' as portalIP should not be included in the proxy")
+	}
+
+	p.OnUpdate([]api.Service{
+		{ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Spec: api.ServiceSpec{Port: svcInfo.proxyPort, Protocol: "TCP", PortalIP: "1.2.3.4"}, Status: api.ServiceStatus{}},
+	})
+	svcInfo, exists = p.getServiceInfo(service)
+	if !exists {
+		t.Fatalf("service with portalIP set not found in the proxy")
+	}
+	testEchoTCP(t, "127.0.0.1", svcInfo.proxyPort)
+	waitForNumProxyLoops(t, p, 1)
+}
+
 // TODO: Test UDP timeouts.

pkg/registry/service/rest.go

@@ -73,8 +73,7 @@ func reloadIPsFromStorage(ipa *ipAllocator, registry Registry) {
 	}
 	for i := range services.Items {
 		service := &services.Items[i]
-		if service.Spec.PortalIP == "" {
+		if !api.IsServiceIPSet(service) {
-			glog.Warningf("service %q has no PortalIP", service.Name)
 			continue
 		}
 		if err := ipa.Allocate(net.ParseIP(service.Spec.PortalIP)); err != nil {
@@ -91,14 +90,14 @@ func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, err
 		return nil, err
 	}
 
-	if len(service.Spec.PortalIP) == 0 {
+	if api.IsServiceIPRequested(service) {
 		// Allocate next available.
 		ip, err := rs.portalMgr.AllocateNext()
 		if err != nil {
 			return nil, err
 		}
 		service.Spec.PortalIP = ip.String()
-	} else {
+	} else if api.IsServiceIPSet(service) {
 		// Try to respect the requested IP.
 		if err := rs.portalMgr.Allocate(net.ParseIP(service.Spec.PortalIP)); err != nil {
 			el := errors.ValidationErrorList{errors.NewFieldInvalid("spec.portalIP", service.Spec.PortalIP, err.Error())}
@@ -111,14 +110,18 @@ func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, err
 	if service.Spec.CreateExternalLoadBalancer {
 		err := rs.createExternalLoadBalancer(ctx, service)
 		if err != nil {
+			if api.IsServiceIPSet(service) {
 				rs.portalMgr.Release(net.ParseIP(service.Spec.PortalIP))
+			}
 			return nil, err
 		}
 	}
 
 	out, err := rs.registry.CreateService(ctx, service)
 	if err != nil {
+		if api.IsServiceIPSet(service) {
 			rs.portalMgr.Release(net.ParseIP(service.Spec.PortalIP))
+		}
 		err = rest.CheckGeneratedNameError(rest.Services, err, service)
 	}
 	return out, err
@@ -137,7 +140,9 @@ func (rs *REST) Delete(ctx api.Context, id string) (runtime.Object, error) {
 	if err != nil {
 		return nil, err
 	}
+	if api.IsServiceIPSet(service) {
 		rs.portalMgr.Release(net.ParseIP(service.Spec.PortalIP))
+	}
 	if service.Spec.CreateExternalLoadBalancer {
 		rs.deleteExternalLoadBalancer(ctx, service)
 	}

pkg/registry/service/rest_test.go

@@ -735,6 +735,7 @@ func TestCreate(t *testing.T) {
 		&api.Service{
 			Spec: api.ServiceSpec{
 				Selector:        map[string]string{"bar": "baz"},
+				PortalIP:        "None",
 				Port:            6502,
 				Protocol:        "TCP",
 				SessionAffinity: "None",
@@ -744,5 +745,15 @@ func TestCreate(t *testing.T) {
 		&api.Service{
 			Spec: api.ServiceSpec{},
 		},
+		// invalid
+		&api.Service{
+			Spec: api.ServiceSpec{
+				Selector:        map[string]string{"bar": "baz"},
+				Port:            6502,
+				Protocol:        "TCP",
+				PortalIP:        "invalid",
+				SessionAffinity: "None",
+			},
+		},
 	)
 }