Fixing memory leak in KDD watchers

When we create a cluster without our CRDs or if we remove one from a
running cluster it will start retrying to watch that non-existent
resource. During this loop we "resync" and then destroy the old
watchers. This process kicks off the leak which stems from somewhere in
client-go, which could be from fragmenting memory by quickly creating
and destroying the watches and underlying channels.

We now only close out watchers that have needed resync, this prevents us
from retrying watches on things that don't need to be stopped.

Author: heschlie

lib/backend/k8s/syncer.go

@@ -308,8 +308,19 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 
 	log.Info("Starting Kubernetes API read loop")
 	for {
+		needSync := false
+
+		// Find out if we need to resync.
+		for _, resync := range syn.needsResync {
+			// We found something that needs resync, we can stop and move on.
+			if resync {
+				needSync = true
+				break
+			}
+		}
+
 		// If we need to resync, do so.
-		if len(syn.needsResync) != 0 {
+		if needSync {
 			// Set status to ResyncInProgress.
 			log.Debugf("Resync required - latest versions: %+v", latestVersions)
 			syn.callbacks.OnStatusUpdated(api.ResyncInProgress)
@@ -336,9 +347,12 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 				return
 			}
 
-			// Close the previous crop of watchers to avoid leaking resources when we
-			// recreate them below.
-			syn.closeAllWatchers()
+			// Close out any watches that needed resync.
+			for k, resync := range syn.needsResync {
+				if _, exists := syn.openWatchers[k]; exists && resync {
+					syn.closeWatcher(k)
+				}
+			}
 		}
 
 		// Create the Kubernetes API watchers.
@@ -353,6 +367,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			}
 			syn.openWatchers[KEY_NS] = nsWatch
 			nsChan = nsWatch.ResultChan()
+			syn.needsResync[KEY_NS] = false
 		}
 
 		if _, exists := syn.openWatchers[KEY_PO]; !exists {
@@ -366,6 +381,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			}
 			syn.openWatchers[KEY_PO] = poWatch
 			poChan = poWatch.ResultChan()
+			syn.needsResync[KEY_PO] = false
 		}
 
 		if _, exists := syn.openWatchers[KEY_NP]; !exists {
@@ -380,6 +396,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			}
 			syn.openWatchers[KEY_NP] = npWatch
 			npChan = npWatch.ResultChan()
+			syn.needsResync[KEY_NP] = false
 		}
 
 		if _, exists := syn.openWatchers[KEY_GNP]; !exists {
@@ -394,6 +411,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			}
 			syn.openWatchers[KEY_GNP] = gnpWatch
 			gnpChan = gnpWatch.ResultChan()
+			syn.needsResync[KEY_GNP] = false
 		}
 
 		if _, exists := syn.openWatchers[KEY_GC]; !exists {
@@ -408,6 +426,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			}
 			syn.openWatchers[KEY_GC] = globalFelixConfigWatch
 			gcChan = globalFelixConfigWatch.ResultChan()
+			syn.needsResync[KEY_GC] = false
 		}
 
 		if _, exists := syn.openWatchers[KEY_IP]; !exists {
@@ -422,6 +441,7 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			}
 			syn.openWatchers[KEY_IP] = ipPoolWatch
 			poolChan = ipPoolWatch.ResultChan()
+			syn.needsResync[KEY_IP] = false
 		}
 
 		if _, exists := syn.openWatchers[KEY_NO]; !exists && !syn.disableNodePoll {
@@ -436,12 +456,10 @@ func (syn *kubeSyncer) readFromKubernetesAPI() {
 			}
 			syn.openWatchers[KEY_NO] = nodeWatch
 			noChan = nodeWatch.ResultChan()
+			syn.needsResync[KEY_NO] = false
+			syn.needsResync[KEY_HC] = false
 		}
 
-		// We resynced if we needed to, and have a complete set of watchers, so reset the
-		// needsResync flag.
-		syn.needsResync = map[string]bool{}
-
 		// Select on the various watch channels.
 		select {
 		case <-syn.stopChan:

lib/backend/k8s/syncer_test.go

@@ -323,19 +323,19 @@ var _ = Describe("Test Syncer", func() {
 			// Simulate error on pod watch.
 			tc.podC <- watch.Event{Type: watch.Error, Object: nil}
 			// Expect a single new list call, but that each watcher is restarted.
-			Eventually(tc.getNumWatchCalls).Should(BeNumerically("==", WATCH_CALLS+7))
+			Eventually(tc.getNumWatchCalls).Should(BeNumerically("==", WATCH_CALLS+1))
 			Expect(tc.getNumListCalls()).To(BeNumerically("==", LIST_CALLS+1))
 
 			// Simulate error on IP Pool watch.
 			tc.poolC <- watch.Event{Type: watch.Error, Object: nil}
 			// Expect a single new list call, but that each watcher is restarted.
-			Eventually(tc.getNumWatchCalls).Should(BeNumerically("==", WATCH_CALLS+14))
+			Eventually(tc.getNumWatchCalls).Should(BeNumerically("==", WATCH_CALLS+2))
 			Expect(tc.getNumListCalls()).To(BeNumerically("==", LIST_CALLS+2))
 
 			// Simulate empty event on IP Pool watch (resourceVersion too old for TPRs)
 			tc.poolC <- watch.Event{Object: nil}
 			// Expect a single new list call, but that each watcher is restarted.
-			Eventually(tc.getNumWatchCalls).Should(BeNumerically("==", WATCH_CALLS+21))
+			Eventually(tc.getNumWatchCalls).Should(BeNumerically("==", WATCH_CALLS+3))
 			Expect(tc.getNumListCalls()).To(BeNumerically("==", LIST_CALLS+3))
 		})
 
@@ -383,15 +383,24 @@ var _ = Describe("Test Syncer", func() {
 			// Check that, after the resync, the old watchers are stopped.
 			tc.stateMutex.Lock()
 			defer tc.stateMutex.Unlock()
-			// We expect 7 old watchers and 7 new. If that changes, we'll assert here
+			// We expect 7 old watchers and 1 new. If that changes, we'll assert here
 			// so the maintainer can re-check the test still matches the logic.
-			Expect(tc.openWatchers).To(HaveLen(14))
-			for _, w := range tc.openWatchers[:len(tc.openWatchers)/2] {
-				w.stopMutex.Lock()
-				stopped := w.stopped
-				w.stopMutex.Unlock()
-				Expect(stopped).To(BeTrue())
+			Expect(tc.openWatchers).To(HaveLen(8))
+
+			// Check and verify the old pod watcher was closed, make sure we ignore the
+			// newest pod watch that was added by only iterating over the old watchers.
+			closed := false
+			for _, w := range tc.openWatchers[:len(tc.openWatchers)-1] {
+				if w.name == "pod" {
+					w.stopMutex.Lock()
+					stopped := w.stopped
+					w.stopMutex.Unlock()
+					Expect(stopped).To(BeTrue())
+					closed = true
+				}
 			}
+			// If for some reason we never found a pod watch we should fail.
+			Expect(closed).To(BeTrue())
 		})
 	})
 })