Merge pull request kubernetes#85703 from robscott/endpointslice-controller-race-fix

Fixing Potential Race Condition in EndpointSlice Controller.

Author: k8s-ci-robot

pkg/controller/endpointslice/BUILD

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "endpointset.go",
         "endpointslice_controller.go",
+        "endpointslice_tracker.go",
         "reconciler.go",
         "utils.go",
     ],
@@ -49,6 +50,7 @@ go_test(
     name = "go_default_test",
     srcs = [
         "endpointslice_controller_test.go",
+        "endpointslice_tracker_test.go",
         "reconciler_test.go",
         "utils_test.go",
     ],

pkg/controller/endpointslice/endpointslice_controller.go

@@ -63,7 +63,7 @@ const (
 func NewController(podInformer coreinformers.PodInformer,
 	serviceInformer coreinformers.ServiceInformer,
 	nodeInformer coreinformers.NodeInformer,
-	esInformer discoveryinformers.EndpointSliceInformer,
+	endpointSliceInformer discoveryinformers.EndpointSliceInformer,
 	maxEndpointsPerSlice int32,
 	client clientset.Interface,
 ) *Controller {
@@ -105,15 +105,23 @@ func NewController(podInformer coreinformers.PodInformer,
 	c.nodeLister = nodeInformer.Lister()
 	c.nodesSynced = nodeInformer.Informer().HasSynced
 
-	c.endpointSliceLister = esInformer.Lister()
-	c.endpointSlicesSynced = esInformer.Informer().HasSynced
+	endpointSliceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    c.onEndpointSliceAdd,
+		UpdateFunc: c.onEndpointSliceUpdate,
+		DeleteFunc: c.onEndpointSliceDelete,
+	})
+
+	c.endpointSliceLister = endpointSliceInformer.Lister()
+	c.endpointSlicesSynced = endpointSliceInformer.Informer().HasSynced
+	c.endpointSliceTracker = newEndpointSliceTracker()
 
 	c.maxEndpointsPerSlice = maxEndpointsPerSlice
 
 	c.reconciler = &reconciler{
 		client:               c.client,
 		nodeLister:           c.nodeLister,
 		maxEndpointsPerSlice: c.maxEndpointsPerSlice,
+		endpointSliceTracker: c.endpointSliceTracker,
 		metricsCache:         endpointslicemetrics.NewCache(maxEndpointsPerSlice),
 	}
 	c.triggerTimeTracker = endpointutil.NewTriggerTimeTracker()
@@ -152,6 +160,10 @@ type Controller struct {
 	// endpointSlicesSynced returns true if the endpoint slice shared informer has been synced at least once.
 	// Added as a member to the struct to allow injection for testing.
 	endpointSlicesSynced cache.InformerSynced
+	// endpointSliceTracker tracks the list of EndpointSlices and associated
+	// resource versions expected for each Service. It can help determine if a
+	// cached EndpointSlice is out of date.
+	endpointSliceTracker *endpointSliceTracker
 
 	// nodeLister is able to list/get nodes and is populated by the
 	// shared informer passed to NewController
@@ -343,6 +355,57 @@ func (c *Controller) onServiceDelete(obj interface{}) {
 	c.queue.Add(key)
 }
 
+// onEndpointSliceAdd queues a sync for the relevant Service for a sync if the
+// EndpointSlice resource version does not match the expected version in the
+// endpointSliceTracker.
+func (c *Controller) onEndpointSliceAdd(obj interface{}) {
+	endpointSlice := obj.(*discovery.EndpointSlice)
+	if endpointSlice == nil {
+		utilruntime.HandleError(fmt.Errorf("Invalid EndpointSlice provided to onEndpointSliceAdd()"))
+		return
+	}
+	if managedByController(endpointSlice) && c.endpointSliceTracker.Stale(endpointSlice) {
+		c.queueServiceForEndpointSlice(endpointSlice)
+	}
+}
+
+// onEndpointSliceUpdate queues a sync for the relevant Service for a sync if
+// the EndpointSlice resource version does not match the expected version in the
+// endpointSliceTracker or the managed-by value of the EndpointSlice has changed
+// from or to this controller.
+func (c *Controller) onEndpointSliceUpdate(prevObj, obj interface{}) {
+	prevEndpointSlice := obj.(*discovery.EndpointSlice)
+	endpointSlice := obj.(*discovery.EndpointSlice)
+	if endpointSlice == nil || prevEndpointSlice == nil {
+		utilruntime.HandleError(fmt.Errorf("Invalid EndpointSlice provided to onEndpointSliceUpdate()"))
+		return
+	}
+	if managedByChanged(prevEndpointSlice, endpointSlice) || (managedByController(endpointSlice) && c.endpointSliceTracker.Stale(endpointSlice)) {
+		c.queueServiceForEndpointSlice(endpointSlice)
+	}
+}
+
+// onEndpointSliceDelete queues a sync for the relevant Service for a sync if the
+// EndpointSlice resource version does not match the expected version in the
+// endpointSliceTracker.
+func (c *Controller) onEndpointSliceDelete(obj interface{}) {
+	endpointSlice := getEndpointSliceFromDeleteAction(obj)
+	if endpointSlice != nil && managedByController(endpointSlice) && c.endpointSliceTracker.Has(endpointSlice) {
+		c.queueServiceForEndpointSlice(endpointSlice)
+	}
+}
+
+// queueServiceForEndpointSlice attempts to queue the corresponding Service for
+// the provided EndpointSlice.
+func (c *Controller) queueServiceForEndpointSlice(endpointSlice *discovery.EndpointSlice) {
+	key, err := serviceControllerKey(endpointSlice)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("Couldn't get key for EndpointSlice %+v: %v", endpointSlice, err))
+		return
+	}
+	c.queue.Add(key)
+}
+
 func (c *Controller) addPod(obj interface{}) {
 	pod := obj.(*v1.Pod)
 	services, err := c.serviceSelectorCache.GetPodServiceMemberships(c.serviceLister, pod)

pkg/controller/endpointslice/endpointslice_tracker.go

@@ -0,0 +1,123 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package endpointslice
+
+import (
+	"sync"
+
+	discovery "k8s.io/api/discovery/v1beta1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// endpointSliceResourceVersions tracks expected EndpointSlice resource versions
+// by EndpointSlice name.
+type endpointSliceResourceVersions map[string]string
+
+// endpointSliceTracker tracks EndpointSlices and their associated resource
+// versions to help determine if a change to an EndpointSlice has been processed
+// by the EndpointSlice controller.
+type endpointSliceTracker struct {
+	// lock protects resourceVersionsByService.
+	lock sync.Mutex
+	// resourceVersionsByService tracks the list of EndpointSlices and
+	// associated resource versions expected for a given Service.
+	resourceVersionsByService map[types.NamespacedName]endpointSliceResourceVersions
+}
+
+// newEndpointSliceTracker creates and initializes a new endpointSliceTracker.
+func newEndpointSliceTracker() *endpointSliceTracker {
+	return &endpointSliceTracker{
+		resourceVersionsByService: map[types.NamespacedName]endpointSliceResourceVersions{},
+	}
+}
+
+// Has returns true if the endpointSliceTracker has a resource version for the
+// provided EndpointSlice.
+func (est *endpointSliceTracker) Has(endpointSlice *discovery.EndpointSlice) bool {
+	est.lock.Lock()
+	defer est.lock.Unlock()
+
+	rrv := est.relatedResourceVersions(endpointSlice)
+	_, ok := rrv[endpointSlice.Name]
+	return ok
+}
+
+// Stale returns true if this endpointSliceTracker does not have a resource
+// version for the provided EndpointSlice or it does not match the resource
+// version of the provided EndpointSlice.
+func (est *endpointSliceTracker) Stale(endpointSlice *discovery.EndpointSlice) bool {
+	est.lock.Lock()
+	defer est.lock.Unlock()
+
+	rrv := est.relatedResourceVersions(endpointSlice)
+	return rrv[endpointSlice.Name] != endpointSlice.ResourceVersion
+}
+
+// Update adds or updates the resource version in this endpointSliceTracker for
+// the provided EndpointSlice.
+func (est *endpointSliceTracker) Update(endpointSlice *discovery.EndpointSlice) {
+	est.lock.Lock()
+	defer est.lock.Unlock()
+
+	rrv := est.relatedResourceVersions(endpointSlice)
+	rrv[endpointSlice.Name] = endpointSlice.ResourceVersion
+}
+
+// Delete removes the resource version in this endpointSliceTracker for the
+// provided EndpointSlice.
+func (est *endpointSliceTracker) Delete(endpointSlice *discovery.EndpointSlice) {
+	est.lock.Lock()
+	defer est.lock.Unlock()
+
+	rrv := est.relatedResourceVersions(endpointSlice)
+	delete(rrv, endpointSlice.Name)
+}
+
+// relatedResourceVersions returns the set of resource versions tracked for the
+// Service corresponding to the provided EndpointSlice. If no resource versions
+// are currently tracked for this service, an empty set is initialized.
+func (est *endpointSliceTracker) relatedResourceVersions(endpointSlice *discovery.EndpointSlice) endpointSliceResourceVersions {
+	serviceNN := getServiceNN(endpointSlice)
+	vers, ok := est.resourceVersionsByService[serviceNN]
+
+	if !ok {
+		vers = endpointSliceResourceVersions{}
+		est.resourceVersionsByService[serviceNN] = vers
+	}
+
+	return vers
+}
+
+// getServiceNN returns a namespaced name for the Service corresponding to the
+// provided EndpointSlice.
+func getServiceNN(endpointSlice *discovery.EndpointSlice) types.NamespacedName {
+	serviceName, _ := endpointSlice.Labels[discovery.LabelServiceName]
+	return types.NamespacedName{Name: serviceName, Namespace: endpointSlice.Namespace}
+}
+
+// managedByChanged returns true if one of the provided EndpointSlices is
+// managed by the EndpointSlice controller while the other is not.
+func managedByChanged(endpointSlice1, endpointSlice2 *discovery.EndpointSlice) bool {
+	return managedByController(endpointSlice1) != managedByController(endpointSlice2)
+}
+
+// managedByController returns true if the controller of the provided
+// EndpointSlices is the EndpointSlice controller.
+func managedByController(endpointSlice *discovery.EndpointSlice) bool {
+	managedBy, _ := endpointSlice.Labels[discovery.LabelManagedBy]
+	return managedBy == controllerName
+}