apiextensions: avoid two HA API server to fight for NonStructural condition message

staging/src/k8s.io/apiextensions-apiserver/pkg/controller/nonstructuralschema/nonstructuralschema_controller.go

@@ -18,6 +18,7 @@ package nonstructuralschema
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -47,6 +48,11 @@ type ConditionController struct {
 	syncFn func(key string) error
 
 	queue workqueue.RateLimitingInterface
+
+	// last generation this controller updated the condition per CRD name (to avoid two
+	// different version of the apiextensions-apiservers in HA to fight for the right message)
+	lastSeenGenerationLock sync.Mutex
+	lastSeenGeneration     map[string]int64
 }
 
 // NewConditionController constructs a non-structural schema condition controller.
@@ -55,16 +61,17 @@ func NewConditionController(
 	crdClient client.CustomResourceDefinitionsGetter,
 ) *ConditionController {
 	c := &ConditionController{
-		crdClient: crdClient,
-		crdLister: crdInformer.Lister(),
-		crdSynced: crdInformer.Informer().HasSynced,
-		queue:     workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "non_structural_schema_condition_controller"),
+		crdClient:          crdClient,
+		crdLister:          crdInformer.Lister(),
+		crdSynced:          crdInformer.Informer().HasSynced,
+		queue:              workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "non_structural_schema_condition_controller"),
+		lastSeenGeneration: map[string]int64{},
 	}
 
 	crdInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    c.addCustomResourceDefinition,
 		UpdateFunc: c.updateCustomResourceDefinition,
-		DeleteFunc: nil,
+		DeleteFunc: c.deleteCustomResourceDefinition,
 	})
 
 	c.syncFn = c.sync
@@ -130,6 +137,14 @@ func (c *ConditionController) sync(key string) error {
 		return err
 	}
 
+	// avoid repeated calculation for the same generation
+	c.lastSeenGenerationLock.Lock()
+	lastSeen, seenBefore := c.lastSeenGeneration[inCustomResourceDefinition.Name]
+	c.lastSeenGenerationLock.Unlock()
+	if seenBefore && inCustomResourceDefinition.Generation <= lastSeen {
+		return nil
+	}
+
 	// check old condition
 	cond := calculateCondition(inCustomResourceDefinition)
 	old := apiextensions.FindCRDCondition(inCustomResourceDefinition, apiextensions.NonStructuralSchema)
@@ -159,6 +174,12 @@ func (c *ConditionController) sync(key string) error {
 		return err
 	}
 
+	// store generation in order to avoid repeated updates for the same generation (and potential
+	// fights of API server in HA environments).
+	c.lastSeenGenerationLock.Lock()
+	defer c.lastSeenGenerationLock.Unlock()
+	c.lastSeenGeneration[crd.Name] = crd.Generation
+
 	return nil
 }
 
@@ -227,3 +248,23 @@ func (c *ConditionController) updateCustomResourceDefinition(obj, _ interface{})
 	klog.V(4).Infof("Updating %s", castObj.Name)
 	c.enqueue(castObj)
 }
+
+func (c *ConditionController) deleteCustomResourceDefinition(obj interface{}) {
+	castObj, ok := obj.(*apiextensions.CustomResourceDefinition)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			klog.Errorf("Couldn't get object from tombstone %#v", obj)
+			return
+		}
+		castObj, ok = tombstone.Obj.(*apiextensions.CustomResourceDefinition)
+		if !ok {
+			klog.Errorf("Tombstone contained object that is not expected %#v", obj)
+			return
+		}
+	}
+
+	c.lastSeenGenerationLock.Lock()
+	defer c.lastSeenGenerationLock.Unlock()
+	delete(c.lastSeenGeneration, castObj.Name)
+}