Repository: istio/istio (go)

Issue:
Prioritized leader election: race condition at start leads to stealing lock unexpectedly
Everytime my non-default revision starts it is taking the leader election lock. I think the problem is:
https://github.com/istio/istio/blob/7d5070373f99613e59b3aa6f0a425a3fe115a21f/pilot/pkg/leaderelection/leaderelection.go#L161

We do not wait for the informer to sync. As a result, `l.defaultWatcher.GetDefault()` is `""` and KeyComparison returns `true` the first attempt.

I think we need to:
* Add a WaitForCacheSync.
* Possibly just use one default watcher so we don't need to sync N times

Pull request:
Wait for cache sync before proceeding with leader election
Makes it so that we wait for caches to sync before proceeding with leader election. This prevents a race condition where revision with name "" would steal the leader lock on startup

fixes #35765


Patch:
diff --git a/pilot/pkg/leaderelection/leaderelection.go b/pilot/pkg/leaderelection/leaderelection.go
index 97417c2ed9..386f74f3ac 100644
--- a/pilot/pkg/leaderelection/leaderelection.go
+++ b/pilot/pkg/leaderelection/leaderelection.go
@@ -75,6 +75,7 @@ type LeaderElection struct {
 
 // Run will start leader election, calling all runFns when we become the leader.
 func (l *LeaderElection) Run(stop <-chan struct{}) {
+	go l.defaultWatcher.Run(stop)
 	for {
 		le, err := l.create()
 		if err != nil {
@@ -97,10 +98,9 @@ func (l *LeaderElection) Run(stop <-chan struct{}) {
 			return
 		default:
 			cancel()
-			// Otherwise, we may have lost our lock. In practice, this is extremely rare; we need to have the lock, then lose it
-			// Typically this means something went wrong, such as API server downtime, etc
-			// If this does happen, we will start the cycle over again
-			log.Errorf("Leader election cycle %v lost. Trying again", l.cycle.Load())
+			// Otherwise, we may have lost our lock. This can happen when the default revision changes and steals
+			// the lock from us.
+			log.Infof("Leader election cycle %v lost. Trying again", l.cycle.Load())
 		}
 	}
 }
@@ -140,7 +140,10 @@ func (l *LeaderElection) create() (*k8sleaderelection.LeaderElector, error) {
 	if l.prioritized {
 		// Function to use to decide whether this revision should steal the existing lock.
 		config.KeyComparison = func(currentLeaderRevision string) bool {
-			return l.revision != currentLeaderRevision && l.defaultWatcher.GetDefault() == l.revision
+			defaultRevision := l.defaultWatcher.GetDefault()
+			return l.revision != currentLeaderRevision &&
+				// empty default revision indicates that there is no default set
+				defaultRevision != "" && defaultRevision == l.revision
 		}
 	}
 
diff --git a/pkg/revisions/default_watcher.go b/pkg/revisions/default_watcher.go
index f8bf60c498..bb054fe0d0 100644
--- a/pkg/revisions/default_watcher.go
+++ b/pkg/revisions/default_watcher.go
@@ -16,22 +16,31 @@ package revisions
 
 import (
 	"sync"
+	"time"
 
+	"go.uber.org/atomic"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
 
 	"istio.io/api/label"
 	"istio.io/istio/pkg/kube"
+	"istio.io/pkg/log"
 )
 
 const (
 	defaultTagWebhookName = "istio-revision-tag-default"
+	initSignal            = "INIT"
 )
 
 // DefaultWatcher keeps track of the current default revision and can notify watchers
 // when the default revision changes.
 type DefaultWatcher interface {
+	Run(stopCh <-chan struct{})
+	HasSynced() bool
 	GetDefault() string
 	AddHandler(handler DefaultHandler)
 }
@@ -43,14 +52,19 @@ type defaultWatcher struct {
 	revision        string
 	defaultRevision string
 	handlers        []DefaultHandler
+
+	queue           workqueue.RateLimitingInterface
 	webhookInformer cache.SharedInformer
-	mu              sync.Mutex
+	initialSync     *atomic.Bool
+	mu              sync.RWMutex
 }
 
 func NewDefaultWatcher(client kube.Client, revision string) DefaultWatcher {
 	p := &defaultWatcher{
-		revision: revision,
-		mu:       sync.Mutex{},
+		revision:    revision,
+		queue:       workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		initialSync: atomic.NewBool(false),
+		mu:          sync.RWMutex{},
 	}
 	p.webhookInformer = client.KubeInformer().Admissionregistration().V1().MutatingWebhookConfigurations().Informer()
 	p.webhookInformer.AddEventHandler(p.makeHandler())
@@ -58,6 +72,58 @@ func NewDefaultWatcher(client kube.Client, revision string) DefaultWatcher {
 	return p
 }
 
+func (p *defaultWatcher) Run(stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer p.queue.ShutDown()
+	if !kube.WaitForCacheSyncInterval(stopCh, time.Second, p.webhookInformer.HasSynced) {
+		log.Errorf("failed to sync default watcher")
+		return
+	}
+	p.queue.Add(initSignal)
+	go wait.Until(p.runQueue, time.Second, stopCh)
+	<-stopCh
+}
+
+// GetDefault returns the current default revision.
+func (p *defaultWatcher) GetDefault() string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return p.defaultRevision
+}
+
+// AddHandler registers a new handler for updates to default revision changes.
+func (p *defaultWatcher) AddHandler(handler DefaultHandler) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.handlers = append(p.handlers, handler)
+}
+
+func (p *defaultWatcher) HasSynced() bool {
+	return p.initialSync.Load()
+}
+
+func (p *defaultWatcher) runQueue() {
+	for p.processNextItem() {
+	}
+}
+
+func (p *defaultWatcher) processNextItem() bool {
+	item, quit := p.queue.Get()
+	if quit {
+		log.Debug("default watcher shutting down, returning")
+		return false
+	}
+	defer p.queue.Done(item)
+
+	if item.(string) == initSignal {
+		p.initialSync.Store(true)
+	} else {
+		p.setDefault(item.(string))
+	}
+
+	return true
+}
+
 func (p *defaultWatcher) makeHandler() *cache.ResourceEventHandlerFuncs {
 	return &cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
@@ -65,43 +131,26 @@ func (p *defaultWatcher) makeHandler() *cache.ResourceEventHandlerFuncs {
 			if filterUpdate(meta) {
 				return
 			}
-			p.setDefaultFromLabels(meta.GetLabels())
+			p.queue.Add(getDefault(meta))
 		},
 		UpdateFunc: func(oldObj interface{}, newObj interface{}) {
 			meta, _ := meta.Accessor(newObj)
 			if filterUpdate(meta) {
 				return
 			}
-			p.setDefaultFromLabels(meta.GetLabels())
+			p.queue.Add(getDefault(meta))
 		},
 		DeleteFunc: func(obj interface{}) {
 			meta, _ := meta.Accessor(obj)
 			if filterUpdate(meta) {
 				return
 			}
-			p.mu.Lock()
-			p.defaultRevision = ""
-			p.notifyHandlers()
-			p.mu.Unlock()
+			// treat "" to mean no default revision is set
+			p.queue.Add("")
 		},
 	}
 }
 
-func filterUpdate(obj metav1.Object) bool {
-	return obj.GetName() != defaultTagWebhookName
-}
-
-func (p *defaultWatcher) setDefaultFromLabels(labels map[string]string) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	if revision, ok := labels[label.IoIstioRev.Name]; ok {
-		if revision != p.defaultRevision {
-			p.defaultRevision = revision
-			p.notifyHandlers()
-		}
-	}
-}
-
 // notifyHandlers notifies all registered handlers on default revision change.
 // assumes externally locked.
 func (p *defaultWatcher) notifyHandlers() {
@@ -110,16 +159,20 @@ func (p *defaultWatcher) notifyHandlers() {
 	}
 }
 
-// GetDefault returns the current default revision.
-func (p *defaultWatcher) GetDefault() string {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	return p.defaultRevision
+func getDefault(meta metav1.Object) string {
+	if revision, ok := meta.GetLabels()[label.IoIstioRev.Name]; ok {
+		return revision
+	}
+	return ""
 }
 
-// AddHandler registers a new handler for updates to default revision changes.
-func (p *defaultWatcher) AddHandler(handler DefaultHandler) {
+func (p *defaultWatcher) setDefault(revision string) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	p.handlers = append(p.handlers, handler)
+	p.defaultRevision = revision
+	p.notifyHandlers()
+}
+
+func filterUpdate(obj metav1.Object) bool {
+	return obj.GetName() != defaultTagWebhookName
 }


Test patch:
diff --git a/pilot/pkg/leaderelection/leaderelection_test.go b/pilot/pkg/leaderelection/leaderelection_test.go
index 662589358c..cf372e370d 100644
--- a/pilot/pkg/leaderelection/leaderelection_test.go
+++ b/pilot/pkg/leaderelection/leaderelection_test.go
@@ -74,6 +74,13 @@ func (w *fakeDefaultWatcher) setDefaultRevision(r string) {
 	w.defaultRevision = r
 }
 
+func (w *fakeDefaultWatcher) Run(stop <-chan struct{}) {
+}
+
+func (w *fakeDefaultWatcher) HasSynced() bool {
+	return true
+}
+
 func (w *fakeDefaultWatcher) GetDefault() string {
 	return w.defaultRevision
 }
diff --git a/pkg/revisions/default_watcher_test.go b/pkg/revisions/default_watcher_test.go
index 72260b3e08..6e0e67c9eb 100644
--- a/pkg/revisions/default_watcher_test.go
+++ b/pkg/revisions/default_watcher_test.go
@@ -93,6 +93,7 @@ func TestNoDefaultRevision(t *testing.T) {
 	client := kube.NewFakeClient()
 	w := newDefaultWatcher(client, "default")
 	client.RunAndWait(stop)
+	go w.Run(stop)
 	// if have no default tag for some reason, should return ""
 	expectRevision(t, w, "")
 	close(stop)
@@ -103,6 +104,7 @@ func TestDefaultRevisionChanges(t *testing.T) {
 	client := kube.NewFakeClient()
 	w := newDefaultWatcher(client, "default")
 	client.RunAndWait(stop)
+	go w.Run(stop)
 	expectRevision(t, w, "")
 	// change default to "red"
 	createDefaultWebhook(t, client, "red")
@@ -123,6 +125,7 @@ func TestHandlers(t *testing.T) {
 	client := kube.NewFakeClient()
 	w := newDefaultWatcher(client, "default")
 	client.RunAndWait(stop)
+	go w.Run(stop)
 	expectRevision(t, w, "")
 
 	// add a handler to watch default revision changes, ensure it's triggered