@@ -450,6 +450,11 @@ type MasterNetworkConfig struct {

 	// CIDR will be rejected. Rejections will be applied first, then the IP checked against one of the allowed CIDRs. You

 	// should ensure this range does not overlap with your nodes, pods, or service CIDRs for security reasons.

 	ExternalIPNetworkCIDRs []string

+	// IngressIPNetworkCIDR controls the range to assign ingress ips from for services of type LoadBalancer on bare

+	// metal. If empty, ingress ips will not be assigned. It may contain a single CIDR that will be allocated from.

+	// For security reasons, you should ensure that this range does not overlap with the CIDRs reserved for external ips,

+	// nodes, pods, or services.

+	IngressIPNetworkCIDR string

 }

 type ImageConfig struct {

@@ -482,6 +482,7 @@ var map_MasterNetworkConfig = map[string]string{

 	"hostSubnetLength":       "HostSubnetLength is the number of bits to allocate to each host's subnet e.g. 8 would mean a /24 network on the host",

 	"serviceNetworkCIDR":     "ServiceNetwork is the CIDR string to specify the service networks",

 	"externalIPNetworkCIDRs": "ExternalIPNetworkCIDRs controls what values are acceptable for the service external IP field. If empty, no externalIP may be set. It may contain a list of CIDRs which are checked for access. If a CIDR is prefixed with !, IPs in that CIDR will be rejected. Rejections will be applied first, then the IP checked against one of the allowed CIDRs. You should ensure this range does not overlap with your nodes, pods, or service CIDRs for security reasons.",

+	"ingressIPNetworkCIDR":   "IngressIPNetworkCIDR controls the range to assign ingress ips from for services of type LoadBalancer on bare metal. If empty, ingress ips will not be assigned. It may contain a single CIDR that will be allocated from. For security reasons, you should ensure that this range does not overlap with the CIDRs reserved for external ips, nodes, pods, or services.",

 }

 func (MasterNetworkConfig) SwaggerDoc() map[string]string {

@@ -401,6 +401,11 @@ type MasterNetworkConfig struct {

 	// CIDR will be rejected. Rejections will be applied first, then the IP checked against one of the allowed CIDRs. You

 	// should ensure this range does not overlap with your nodes, pods, or service CIDRs for security reasons.

 	ExternalIPNetworkCIDRs []string `json:"externalIPNetworkCIDRs"`

+	// IngressIPNetworkCIDR controls the range to assign ingress ips from for services of type LoadBalancer on bare

+	// metal. If empty, ingress ips will not be assigned. It may contain a single CIDR that will be allocated from.

+	// For security reasons, you should ensure that this range does not overlap with the CIDRs reserved for external ips,

+	// nodes, pods, or services.

+	IngressIPNetworkCIDR string `json:"ingressIPNetworkCIDR"`

 }

 // ImageConfig holds the necessary configuration options for building image names for system components

@@ -199,6 +199,7 @@ networkConfig:

   clusterNetworkCIDR: ""

   externalIPNetworkCIDRs: null

   hostSubnetLength: 0

+  ingressIPNetworkCIDR: ""

   networkPluginName: ""

   serviceNetworkCIDR: ""

 oauthConfig:

@@ -157,6 +157,12 @@ func ValidateMasterConfig(config *api.MasterConfig, fldPath *field.Path) Validat

 			}

 		}

 	}

+	if len(config.NetworkConfig.IngressIPNetworkCIDR) > 0 {

+		cidr := config.NetworkConfig.IngressIPNetworkCIDR

+		if _, ipNet, err := net.ParseCIDR(cidr); err != nil || ipNet.IP.IsUnspecified() {

+			validationResults.AddErrors(field.Invalid(fldPath.Child("networkConfig", "ingressIPNetworkCIDR").Index(0), cidr, "must be a valid CIDR notation IP range (e.g. 172.30.0.0/16)"))

+		}

+	}

 	validationResults.AddErrors(ValidateKubeConfig(config.MasterClients.OpenShiftLoopbackKubeConfig, fldPath.Child("masterClients", "openShiftLoopbackKubeConfig"))...)

@@ -60,6 +60,9 @@ const (

 	InfraEndpointControllerServiceAccountName = "endpoint-controller"

 	EndpointControllerRoleName                = "system:endpoint-controller"

+

+	InfraServiceIngressIPControllerServiceAccountName = "service-ingress-ip-controller"

+	ServiceIngressIPControllerRoleName                = "system:service-ingress-ip-controller"

 )

 type InfraServiceAccounts struct {

@@ -766,4 +769,41 @@ func init() {

 		panic(err)

 	}

+	err = InfraSAs.addServiceAccount(

+		InfraServiceIngressIPControllerServiceAccountName,

+		authorizationapi.ClusterRole{

+			ObjectMeta: kapi.ObjectMeta{

+				Name: ServiceIngressIPControllerRoleName,

+			},

+			Rules: []authorizationapi.PolicyRule{

+				// Listing and watching services

+				{

+					APIGroups: []string{kapi.GroupName},

+					Verbs:     sets.NewString("list", "watch"),

+					Resources: sets.NewString("services"),

+				},

+				// IngressIPController.persistSpec changes the spec of the service

+				{

+					APIGroups: []string{kapi.GroupName},

+					Verbs:     sets.NewString("update"),

+					Resources: sets.NewString("services"),

+				},

+				// IngressIPController.persistStatus changes the status of the service

+				{

+					APIGroups: []string{kapi.GroupName},

+					Verbs:     sets.NewString("update"),

+					Resources: sets.NewString("services/status"),

+				},

+				// IngressIPController.recorder

+				{

+					Verbs:     sets.NewString("create", "update", "patch"),

+					Resources: sets.NewString("events"),

+				},

+			},

+		},

+	)

+	if err != nil {

+		panic(err)

+	}

+

 }

@@ -464,7 +464,9 @@ func newAdmissionChain(pluginNames []string, admissionConfigFilename string, plu

 				// should have been caught with validation

 				return nil, err

 			}

-			plugins = append(plugins, serviceadmit.NewExternalIPRanger(reject, admit))

+			// TODO need to disallow if a cloud provider is configured

+			allowIngressIP := len(options.NetworkConfig.IngressIPNetworkCIDR) > 0

+			plugins = append(plugins, serviceadmit.NewExternalIPRanger(reject, admit, allowIngressIP))

 		case serviceadmit.RestrictedEndpointsPluginName:

 			// we need to set some customer parameters, so create by hand

@@ -44,6 +44,7 @@ import (

 	"github.com/openshift/origin/pkg/security/mcs"

 	"github.com/openshift/origin/pkg/security/uid"

 	"github.com/openshift/origin/pkg/security/uidallocator"

+	"github.com/openshift/origin/pkg/service/controller/ingressip"

 	servingcertcontroller "github.com/openshift/origin/pkg/service/controller/servingcert"

 	configapi "github.com/openshift/origin/pkg/cmd/server/api"

@@ -61,6 +62,8 @@ const (

 	// from CMServer MinResyncPeriod

 	defaultReplenishmentSyncPeriod time.Duration = 12 * time.Hour

+

+	defaultIngressIPSyncPeriod time.Duration = 10 * time.Minute

 )

 // RunProjectAuthorizationCache starts the project authorization cache

@@ -516,3 +519,19 @@ func (c *MasterConfig) RunClusterQuotaReconciliationController() {

 	c.ClusterQuotaMappingController.GetClusterQuotaMapper().AddListener(controller)

 	go controller.Run(5, utilwait.NeverStop)

 }

+

+// RunIngressIPController starts the ingress ip controller if IngressIPNetworkCIDR is configured.

+func (c *MasterConfig) RunIngressIPController(client *kclient.Client) {

+	// TODO need to disallow if a cloud provider is configured

+	if len(c.Options.NetworkConfig.IngressIPNetworkCIDR) == 0 {

+		return

+	}

+

+	_, ipNet, err := net.ParseCIDR(c.Options.NetworkConfig.IngressIPNetworkCIDR)

+	if err != nil {

+		// should have been caught with validation

+		glog.Fatalf("Unable to start ingress ip controller: %v", err)

+	}

+	ingressIPController := ingressip.NewIngressIPController(client, ipNet, defaultIngressIPSyncPeriod)

+	go ingressIPController.Run(utilwait.NeverStop)

+}

@@ -669,6 +669,12 @@ func startControllers(oc *origin.MasterConfig, kc *kubernetes.MasterConfig) erro

 	}

 	oc.RunServiceServingCertController(serviceServingCertClient)

+	_, _, ingressIPClient, err := oc.GetServiceAccountClients(bootstrappolicy.InfraServiceIngressIPControllerServiceAccountName)

+	if err != nil {

+		glog.Fatalf("Could not get client: %v", err)

+	}

+	oc.RunIngressIPController(ingressIPClient)

+

 	glog.Infof("Started Origin Controllers")

 	return nil

@@ -16,14 +16,15 @@ const ExternalIPPluginName = "ExternalIPRanger"

 func init() {

 	kadmission.RegisterPlugin("ExternalIPRanger", func(client clientset.Interface, config io.Reader) (kadmission.Interface, error) {

-		return NewExternalIPRanger(nil, nil), nil

+		return NewExternalIPRanger(nil, nil, false), nil

 	})

 }

 type externalIPRanger struct {

 	*kadmission.Handler

-	reject []*net.IPNet

-	admit  []*net.IPNet

+	reject         []*net.IPNet

+	admit          []*net.IPNet

+	allowIngressIP bool

 }

 var _ kadmission.Interface = &externalIPRanger{}

@@ -51,11 +52,12 @@ func ParseRejectAdmitCIDRRules(rules []string) (reject, admit []*net.IPNet, err

 }

 // NewConstraint creates a new SCC constraint admission plugin.

-func NewExternalIPRanger(reject, admit []*net.IPNet) *externalIPRanger {

+func NewExternalIPRanger(reject, admit []*net.IPNet, allowIngressIP bool) *externalIPRanger {

 	return &externalIPRanger{

-		Handler: kadmission.NewHandler(kadmission.Create, kadmission.Update),

-		reject:  reject,

-		admit:   admit,

+		Handler:        kadmission.NewHandler(kadmission.Create, kadmission.Update),

+		reject:         reject,

+		admit:          admit,

+		allowIngressIP: allowIngressIP,

 	}

 }

@@ -84,11 +86,30 @@ func (r *externalIPRanger) Admit(a kadmission.Attributes) error {

 		return nil

 	}

+	// Determine if an ingress ip address should be allowed as an

+	// external ip by checking the loadbalancer status of the previous

+	// object state. Only updates need to be validated against the

+	// ingress ip since the loadbalancer status cannot be set on

+	// create.

+	ingressIP := ""

+	retrieveIngressIP := a.GetOperation() == kadmission.Update &&

+		r.allowIngressIP && svc.Spec.Type == kapi.ServiceTypeLoadBalancer

+	if retrieveIngressIP {

+		old, ok := a.GetOldObject().(*kapi.Service)

+		ipPresent := ok && old != nil && len(old.Status.LoadBalancer.Ingress) > 0

+		if ipPresent {

+			ingressIP = old.Status.LoadBalancer.Ingress[0].IP

+		}

+	}

+

 	var errs field.ErrorList

 	switch {

 	// administrator disabled externalIPs

 	case len(svc.Spec.ExternalIPs) > 0 && len(r.admit) == 0:

-		errs = append(errs, field.Forbidden(field.NewPath("spec", "externalIPs"), "externalIPs have been disabled"))

+		onlyIngressIP := len(svc.Spec.ExternalIPs) == 1 && svc.Spec.ExternalIPs[0] == ingressIP

+		if !onlyIngressIP {

+			errs = append(errs, field.Forbidden(field.NewPath("spec", "externalIPs"), "externalIPs have been disabled"))

+		}

 	// administrator has limited the range

 	case len(svc.Spec.ExternalIPs) > 0 && len(r.admit) > 0:

 		for i, s := range svc.Spec.ExternalIPs {

@@ -97,7 +118,8 @@ func (r *externalIPRanger) Admit(a kadmission.Attributes) error {

 				errs = append(errs, field.Forbidden(field.NewPath("spec", "externalIPs").Index(i), "externalIPs must be a valid address"))

 				continue

 			}

-			if NetworkSlice(r.reject).Contains(ip) || !NetworkSlice(r.admit).Contains(ip) {

+			notIngressIP := s != ingressIP

+			if (NetworkSlice(r.reject).Contains(ip) || !NetworkSlice(r.admit).Contains(ip)) && notIngressIP {

 				errs = append(errs, field.Forbidden(field.NewPath("spec", "externalIPs").Index(i), "externalIP is not allowed"))

 				continue

 			}

@@ -14,6 +14,7 @@ func TestAdmission(t *testing.T) {

 	svc := &kapi.Service{

 		ObjectMeta: kapi.ObjectMeta{Name: "test"},

 	}

+	var oldSvc *kapi.Service

 	_, ipv4, err := net.ParseCIDR("172.0.0.0/16")

 	if err != nil {

@@ -43,6 +44,8 @@ func TestAdmission(t *testing.T) {

 		externalIPs     []string

 		admit           bool

 		errFn           func(err error) bool

+		loadBalancer    bool

+		ingressIP       string

 	}{

 		{

 			admit:    true,

@@ -128,7 +131,6 @@ func TestAdmission(t *testing.T) {

 			op:          admission.Update,

 			testName:    "IP in range on update",

 		},

-

 		// other checks

 		{

 			admit:       false,

@@ -156,12 +158,65 @@ func TestAdmission(t *testing.T) {

 			op:          admission.Create,

 			testName:    "rejections can cover the entire range",

 		},

+		// Ingress IP checks

+		{

+			admit:        true,

+			externalIPs:  []string{"1.2.3.4"},

+			op:           admission.Update,

+			testName:     "Ingress ip allowed when external ips are disabled",

+			loadBalancer: true,

+			ingressIP:    "1.2.3.4",

+		},

+		{

+			admit:        true,

+			admits:       []*net.IPNet{ipv4},

+			externalIPs:  []string{"1.2.3.4", "172.0.0.1"},

+			op:           admission.Update,

+			testName:     "Ingress ip allowed when external ips are enabled",

+			loadBalancer: true,

+			ingressIP:    "1.2.3.4",

+		},

+		{

+			admit:        false,

+			admits:       []*net.IPNet{ipv4},

+			externalIPs:  []string{"1.2.3.4", "172.0.0.1"},

+			op:           admission.Update,

+			testName:     "Ingress ip not allowed for non-lb service",

+			loadBalancer: false,

+			ingressIP:    "1.2.3.4",

+		},

 	}

 	for _, test := range tests {

 		svc.Spec.ExternalIPs = test.externalIPs

-		handler := NewExternalIPRanger(test.rejects, test.admits)

+		allowIngressIP := len(test.ingressIP) > 0 || test.loadBalancer

+		handler := NewExternalIPRanger(test.rejects, test.admits, allowIngressIP)

+

+		if test.loadBalancer {

+			svc.Spec.Type = kapi.ServiceTypeLoadBalancer

+		} else {

+			svc.Spec.Type = kapi.ServiceTypeClusterIP

+		}

+

+		if len(test.ingressIP) > 0 {

+			// Provide an ingress ip via the previous object state

+			oldSvc = &kapi.Service{

+				ObjectMeta: kapi.ObjectMeta{Name: "test"},

+				Status: kapi.ServiceStatus{

+					LoadBalancer: kapi.LoadBalancerStatus{

+						Ingress: []kapi.LoadBalancerIngress{

+							{

+								IP: test.ingressIP,

+							},

+						},

+					},

+				},

+			}

+

+		} else {

+			oldSvc = nil

+		}

-		err := handler.Admit(admission.NewAttributesRecord(svc, nil, kapi.Kind("Service").WithVersion("version"), "namespace", svc.ObjectMeta.Name, kapi.Resource("services").WithVersion("version"), "", test.op, nil))

+		err := handler.Admit(admission.NewAttributesRecord(svc, oldSvc, kapi.Kind("Service").WithVersion("version"), "namespace", svc.ObjectMeta.Name, kapi.Resource("services").WithVersion("version"), "", test.op, nil))

 		if test.admit && err != nil {

 			t.Errorf("%s: expected no error but got: %s", test.testName, err)

 		} else if !test.admit && err == nil {

@@ -180,7 +235,7 @@ func TestHandles(t *testing.T) {

 		admission.Connect: false,

 		admission.Delete:  false,

 	} {

-		ranger := NewExternalIPRanger(nil, nil)

+		ranger := NewExternalIPRanger(nil, nil, false)

 		if e, a := shouldHandle, ranger.Handles(op); e != a {

 			t.Errorf("%v: shouldHandle=%t, handles=%t", op, e, a)

 		}

new file mode 100644

@@ -0,0 +1,698 @@

+package ingressip

+

+import (

+	"fmt"

+	"net"

+	"sort"

+	"sync"

+	"time"

+

+	"github.com/golang/glog"

+	kapi "k8s.io/kubernetes/pkg/api"

+	kerrors "k8s.io/kubernetes/pkg/api/errors"

+	"k8s.io/kubernetes/pkg/client/cache"

+	"k8s.io/kubernetes/pkg/client/record"

+	kclient "k8s.io/kubernetes/pkg/client/unversioned"

+	"k8s.io/kubernetes/pkg/controller"

+	"k8s.io/kubernetes/pkg/controller/framework"

+	"k8s.io/kubernetes/pkg/registry/service/allocator"

+	"k8s.io/kubernetes/pkg/registry/service/ipallocator"

+	"k8s.io/kubernetes/pkg/runtime"

+	utilruntime "k8s.io/kubernetes/pkg/util/runtime"

+	"k8s.io/kubernetes/pkg/util/sets"

+	"k8s.io/kubernetes/pkg/util/wait"

+	"k8s.io/kubernetes/pkg/util/workqueue"

+	"k8s.io/kubernetes/pkg/watch"

+)

+

+const (

+	// It's necessary to allocate for the initial sync in consistent

+	// order rather than in the order received.  This requires waiting

+	// until the initial sync has been processed, and to avoid a hot

+	// loop, we'll wait this long between checks.

+	SyncProcessedPollPeriod = 100 * time.Millisecond

+

+	clientRetryCount    = 5

+	clientRetryInterval = 5 * time.Second

+)

+

+// IngressIPController is responsible for allocating ingress ip

+// addresses to Service objects of type LoadBalancer.

+type IngressIPController struct {

+	client kclient.ServicesNamespacer

+

+	controller *framework.Controller

+

+	maxRetries int

+

+	// Tracks ip allocation for the configured range

+	ipAllocator *ipallocator.Range

+	// Tracks ip -> service key to allow detection of duplicate ip

+	// allocations.

+	allocationMap map[string]string

+

+	// Tracks services requeued for allocation when the range is full.

+	requeuedAllocations sets.String

+

+	// Protects the transition between initial sync and regular processing

+	lock  sync.Mutex

+	cache cache.Store

+	queue workqueue.RateLimitingInterface

+

+	// recorder is used to record events.

+	recorder record.EventRecorder

+

+	// changeHandler does the work. It can be factored out for unit testing.

+	changeHandler func(change *serviceChange) error

+	// persistenceHandler persists service changes.  It can be factored out for unit testing

+	persistenceHandler func(client kclient.ServicesNamespacer, service *kapi.Service, targetStatus bool) error

+}

+

+type serviceChange struct {

+	key                string

+	oldService         *kapi.Service

+	requeuedAllocation bool

+}

+

+// NewIngressIPController creates a new IngressIPController.

+// TODO this should accept a shared informer

+func NewIngressIPController(kc kclient.Interface, ipNet *net.IPNet, resyncInterval time.Duration) *IngressIPController {

+	eventBroadcaster := record.NewBroadcaster()

+	eventBroadcaster.StartRecordingToSink(kc.Events(""))

+	recorder := eventBroadcaster.NewRecorder(kapi.EventSource{Component: "ingressip-controller"})

+

+	ic := &IngressIPController{

+		client:     kc,

+		queue:      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),

+		maxRetries: 10,

+		recorder:   recorder,

+	}

+

+	ic.cache, ic.controller = framework.NewInformer(

+		&cache.ListWatch{

+			ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+				return ic.client.Services(kapi.NamespaceAll).List(options)

+			},

+			WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+				return ic.client.Services(kapi.NamespaceAll).Watch(options)

+			},

+		},

+		&kapi.Service{},

+		resyncInterval,

+		framework.ResourceEventHandlerFuncs{

+			AddFunc: func(obj interface{}) {

+				service := obj.(*kapi.Service)

+				glog.V(5).Infof("Adding service %s/%s", service.Namespace, service.Name)

+				ic.enqueueChange(obj, nil)

+			},

+			UpdateFunc: func(old, cur interface{}) {

+				service := cur.(*kapi.Service)

+				glog.V(5).Infof("Updating service %s/%s", service.Namespace, service.Name)

+				ic.enqueueChange(cur, old)

+			},

+			DeleteFunc: func(obj interface{}) {

+				service := obj.(*kapi.Service)

+				glog.V(5).Infof("Deleting service %s/%s", service.Namespace, service.Name)

+				ic.enqueueChange(nil, obj)

+			},

+		},

+	)

+

+	ic.changeHandler = ic.processChange

+	ic.persistenceHandler = persistService

+

+	ic.ipAllocator = ipallocator.NewAllocatorCIDRRange(ipNet, func(max int, rangeSpec string) allocator.Interface {

+		return allocator.NewAllocationMap(max, rangeSpec)

+	})

+

+	ic.allocationMap = make(map[string]string)

+	ic.requeuedAllocations = sets.NewString()

+

+	return ic

+}

+

+// enqueueChange transforms the old and new objects into a change

+// object and queues it.  A lock is shared with processInitialSync to

+// avoid enqueueing while the changes from the initial sync are being

+// processed.

+func (ic *IngressIPController) enqueueChange(new interface{}, old interface{}) {

+	ic.lock.Lock()

+	defer ic.lock.Unlock()

+

+	change := &serviceChange{}

+

+	if new != nil {

+		// Queue the key needed to retrieve the lastest state from the

+		// cache when the change is processed.

+		key, err := controller.KeyFunc(new)

+		if err != nil {

+			utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %+v: %v", new, err))

+			return

+		}

+		change.key = key

+	}

+

+	if old != nil {

+		change.oldService = old.(*kapi.Service)

+	}

+

+	ic.queue.Add(change)

+}

+

+// Run begins watching and syncing.

+func (ic *IngressIPController) Run(stopCh <-chan struct{}) {

+	defer utilruntime.HandleCrash()

+	go ic.controller.Run(stopCh)

+

+	glog.V(5).Infof("Waiting for the initial sync to be completed")

+	for !ic.controller.HasSynced() {

+		select {

+		case <-time.After(SyncProcessedPollPeriod):

+		case <-stopCh:

+			return

+		}

+	}

+

+	if !ic.processInitialSync() {

+		return

+	}

+

+	glog.V(5).Infof("Starting normal worker")

+	for {

+		if !ic.work() {

+			break

+		}

+	}

+

+	glog.V(5).Infof("Shutting down ingress ip controller")

+	ic.queue.ShutDown()

+}

+

+type serviceAge []*kapi.Service

+

+func (s serviceAge) Len() int      { return len(s) }

+func (s serviceAge) Swap(i, j int) { s[i], s[j] = s[j], s[i] }

+func (s serviceAge) Less(i, j int) bool {

+	if s[i].CreationTimestamp.Before(s[j].CreationTimestamp) {

+		return true

+	}

+	return (s[i].CreationTimestamp == s[j].CreationTimestamp && s[i].UID < s[j].UID)

+}

+

+// processInitialSync processes the items queued by informer's initial sync.

+// A lock is shared between this method and enqueueService to ensure

+// that queue additions are blocked while the sync is being processed.

+// Returns a boolean indication of whether processing should continue.

+func (ic *IngressIPController) processInitialSync() bool {

+	ic.lock.Lock()

+	defer ic.lock.Unlock()

+

+	glog.V(5).Infof("Processing initial sync")

+

+	// Track services that need to be processed after existing

+	// allocations are recorded.

+	var pendingServices []*kapi.Service

+

+	// Track post-sync changes that need to be added back the queue

+	// after allocations are recorded.  These changes may end up in

+	// the queue if watch events were queued before completion of the

+	// initial sync was detected.

+	var pendingChanges []*serviceChange

+

+	// Drain the queue.  Len() should be safe because enqueueService

+	// requires the same lock held by this method.

+	for ic.queue.Len() > 0 {

+		item, quit := ic.queue.Get()

+		if quit {

+			return false

+		}

+		ic.queue.Done(item)

+		ic.queue.Forget(item)

+		change := item.(*serviceChange)

+

+		// The initial sync only includes additions, so if an update

+		// or delete is seen (indicated by the presence of oldService),

+		// it and all subsequent changes are post-sync watch events that

+		// should be queued without processing.

+		postSyncChange := change.oldService != nil || len(pendingChanges) > 0

+		if postSyncChange {

+			pendingChanges = append(pendingChanges, change)

+			continue

+		}

+

+		service := ic.getCachedService(change.key)

+		if service == nil {

+			// Service was deleted

+			continue

+		}

+

+		if service.Spec.Type == kapi.ServiceTypeLoadBalancer {

+			// Save for subsequent addition back to the queue to

+			// ensure persistent state is updated during regular

+			// processing.

+			pendingServices = append(pendingServices, service)

+

+			if len(service.Status.LoadBalancer.Ingress) > 0 {

+				// The service has an existing allocation

+				ipString := service.Status.LoadBalancer.Ingress[0].IP

+				// Return values indicating that reallocation is

+				// necessary or that an error occurred can be ignored

+				// since the service will be processed again.

+				ic.recordLocalAllocation(change.key, ipString)

+			}

+		}

+	}

+

+	// Add pending service additions back to the queue in consistent order.

+	sort.Sort(serviceAge(pendingServices))

+	for _, service := range pendingServices {

+		if key, err := controller.KeyFunc(service); err == nil {

+			glog.V(5).Infof("Adding service back to queue: %v ", key)

+			change := &serviceChange{key: key}

+			ic.queue.Add(change)

+		} else {

+			// This error should have been caught by enqueueService

+			utilruntime.HandleError(fmt.Errorf("Couldn't get key for service %+v: %v", service, err))

+			continue

+		}

+	}

+

+	// Add watch events back to the queue

+	for _, change := range pendingChanges {

+		ic.queue.Add(change)

+	}

+

+	glog.V(5).Infof("Completed processing initial sync")

+

+	return true

+}

+

+// getCachedService logs if unable to retrieve a service for the given key.

+func (ic *IngressIPController) getCachedService(key string) *kapi.Service {

+	if len(key) == 0 {

+		return nil

+	}

+	if obj, exists, err := ic.cache.GetByKey(key); err != nil {

+		glog.V(5).Infof("Unable to retrieve service %v from store: %v", key, err)

+	} else if !exists {

+		glog.V(6).Infof("Service %v has been deleted", key)

+	} else {

+		return obj.(*kapi.Service)

+	}

+	return nil

+}

+

+// recordLocalAllocation attempts to update local state for the given

+// service key and ingress ip.  Returns a boolean indication of

+// whether reallocation is necessary and an error indicating the

+// reason for reallocation.  If reallocation is not indicated, a

+// non-nil error indicates an exceptional condition.

+func (ic *IngressIPController) recordLocalAllocation(key, ipString string) (reallocate bool, err error) {

+	ip := net.ParseIP(ipString)

+	if ip == nil {

+		return true, fmt.Errorf("Service %v has an invalid ingress ip %v.  A new ip will be allocated.", key, ipString)

+	}

+

+	ipKey, ok := ic.allocationMap[ipString]

+	switch {

+	case ok && ipKey == key:

+		// Allocation exists for this service

+		return false, nil

+	case ok && ipKey != key:

+		// TODO prefer removing the allocation from a service that does not have a matching LoadBalancerIP

+		return true, fmt.Errorf("Another service is using ingress ip %v.  A new ip will be allocated for %v.", ipString, key)

+	}

+

+	err = ic.ipAllocator.Allocate(ip)

+	switch {

+	case err == ipallocator.ErrNotInRange:

+		return true, fmt.Errorf("The ingress ip %v for service %v is not in the ingress range.  A new ip will be allocated.", ipString, key)

+	case err != nil:

+		// The only other error that Allocate() can throw is ErrAllocated, but that

+		// should not happen after the check against the allocation map.

+		return false, fmt.Errorf("Unexpected error from ip allocator for service %v: %v", key, err)

+	}

+	ic.allocationMap[ipString] = key

+	glog.V(5).Infof("Recorded allocation of ip %v for service %v", ipString, key)

+	return false, nil

+}

+

+// work dispatches the next item in the queue to the change handler.

+// If the change handler returns an error, the change will be added to

+// the end of the queue to be processed again.  Returns a boolean

+// indication of whether processing should continue.

+func (ic *IngressIPController) work() bool {

+	item, quit := ic.queue.Get()

+	if quit {

+		return false

+	}

+	change := item.(*serviceChange)

+	defer ic.queue.Done(change)

+

+	if change.requeuedAllocation {

+		// Reset the allocation state so that the change can be

+		// requeued if necessary.  Only additions/updates are requeued

+		// for allocation so change.key should be non-empty.

+		change.requeuedAllocation = false

+		ic.requeuedAllocations.Delete(change.key)

+	}

+

+	if err := ic.changeHandler(change); err == nil {

+		// No further processing required

+		ic.queue.Forget(change)

+	} else {

+		if err == ipallocator.ErrFull {

+			// When the range is full, avoid requeueing more than a

+			// single change requiring allocation per service.

+			// Otherwise the queue could grow without bounds as every

+			// service update would add another change that would be

+			// endlessly requeued.

+			if ic.requeuedAllocations.Has(change.key) {

+				return true

+			}

+			change.requeuedAllocation = true

+			ic.requeuedAllocations.Insert(change.key)

+			service := ic.getCachedService(change.key)

+			if service != nil {

+				ic.recorder.Eventf(service, kapi.EventTypeWarning, "IngressIPRangeFull", "No available ingress ip to allocate to service %s", change.key)

+			}

+		}

+		// Failed but can be retried

+		utilruntime.HandleError(fmt.Errorf("error syncing service, it will be retried: %v", err))

+		ic.queue.AddRateLimited(change)

+	}

+

+	return true

+}

+

+// processChange responds to a service change by synchronizing the

+// local and persisted ingress ip allocation state of the service.

+func (ic *IngressIPController) processChange(change *serviceChange) error {

+	service := ic.getCachedService(change.key)

+

+	ic.clearOldAllocation(service, change.oldService)

+

+	if service == nil {

+		// Service was deleted - no further processing required

+		return nil

+	}

+

+	typeLoadBalancer := service.Spec.Type == kapi.ServiceTypeLoadBalancer

+	hasAllocation := len(service.Status.LoadBalancer.Ingress) > 0

+	switch {

+	case typeLoadBalancer && hasAllocation:

+		return ic.recordAllocation(service, change.key)

+	case typeLoadBalancer && !hasAllocation:

+		return ic.allocate(service, change.key)

+	case !typeLoadBalancer && hasAllocation:

+		return ic.deallocate(service, change.key)

+	default:

+		return nil

+	}

+}

+

+// clearOldAllocation clears the old allocation for a service if it

+// differs from a new allocation.  Returns a boolean indication of

+// whether the old allocation was cleared.

+func (ic *IngressIPController) clearOldAllocation(new, old *kapi.Service) bool {

+	oldIP := ""

+	if old != nil && old.Spec.Type == kapi.ServiceTypeLoadBalancer && len(old.Status.LoadBalancer.Ingress) > 0 {

+		oldIP = old.Status.LoadBalancer.Ingress[0].IP

+	}

+	noOldAllocation := len(oldIP) == 0

+	if noOldAllocation {

+		return false

+	}

+

+	newIP := ""

+	if new != nil && new.Spec.Type == kapi.ServiceTypeLoadBalancer && len(new.Status.LoadBalancer.Ingress) > 0 {

+		newIP = new.Status.LoadBalancer.Ingress[0].IP

+	}

+	allocationUnchanged := newIP == oldIP

+	if allocationUnchanged {

+		return false

+	}

+

+	// New allocation differs from old due to update or deletion

+

+	// Get the key from the old service since the new service may be nil

+	if key, err := controller.KeyFunc(old); err == nil {

+		ic.clearLocalAllocation(key, oldIP)

+		return true

+	} else {

+		// Recovery/retry not possible for this error

+		utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %+v: %v", old, err))

+		return false

+	}

+}

+

+// recordAllocation updates local state with the ingress ip indicated

+// in a service's status and ensures that the ingress ip appears in

+// the service's list of external ips.  If the service's ingress ip is

+// invalid for any reason, a new ip will be allocated.

+func (ic *IngressIPController) recordAllocation(service *kapi.Service, key string) error {

+	// If more than one ingress ip is present, it will be ignored

+	ipString := service.Status.LoadBalancer.Ingress[0].IP

+

+	reallocate, err := ic.recordLocalAllocation(key, ipString)

+	if !reallocate && err != nil {

+		return err

+	}

+	reallocateMessage := ""

+	if err != nil {

+		reallocateMessage = err.Error()

+	}

+

+	// Make a copy to modify to avoid mutating cache state

+	t, err := kapi.Scheme.DeepCopy(service)

+	if err != nil {

+		return err

+	}

+	service = t.(*kapi.Service)

+

+	if reallocate {

+		// TODO update the external ips but not the status since

+		// allocate() will overwrite any existing allocation.

+		if err = ic.clearPersistedAllocation(service, key, reallocateMessage); err != nil {

+			return err

+		}

+		ic.recorder.Eventf(service, kapi.EventTypeWarning, "IngressIPReallocated", reallocateMessage)

+		return ic.allocate(service, key)

+	} else {

+		// Ensure that the ingress ip is present in the service's spec.

+		return ic.ensureExternalIP(service, key, ipString)

+	}

+}

+

+// allocate assigns an unallocated ip to a service and updates the

+// service's persisted state.

+func (ic *IngressIPController) allocate(service *kapi.Service, key string) error {

+	// Make a copy to avoid mutating cache state

+	t, err := kapi.Scheme.DeepCopy(service)

+	if err != nil {

+		return err

+	}

+	service = t.(*kapi.Service)

+

+	ip, err := ic.allocateIP(service.Spec.LoadBalancerIP)

+	if err != nil {

+		return err

+	}

+	ipString := ip.String()

+

+	glog.V(5).Infof("Allocating ip %v to service %v", ipString, key)

+	service.Status = kapi.ServiceStatus{

+		LoadBalancer: kapi.LoadBalancerStatus{

+			Ingress: []kapi.LoadBalancerIngress{

+				{

+					IP: ipString,

+				},

+			},

+		},

+	}

+	if err = ic.persistServiceStatus(service); err != nil {

+		if releaseErr := ic.ipAllocator.Release(ip); releaseErr != nil {

+			// Release from contiguous allocator should never return an error, but just in case...

+			utilruntime.HandleError(fmt.Errorf("Error releasing ip %v for service %v: %v", ipString, key, releaseErr))

+		}

+		return err

+	}

+	ic.allocationMap[ipString] = key

+