fix(watcher): add debouncing to kubeconfig watcher for Windows compatibility (containers#545)

The kubeconfig watcher was triggering multiple callbacks on Windows due
to fsnotify generating multiple events for a single file write operation.
This caused race conditions during provider reset, leading to flaky tests.

Changes:
- Add debouncing to Kubeconfig watcher (default 100ms, configurable via
  KUBECONFIG_DEBOUNCE_WINDOW_MS environment variable)
- Add proper close synchronization with stopCh/stoppedCh channels
- Align Kubeconfig watcher implementation with ClusterState watcher pattern

Signed-off-by: Marc Nuri <[email protected]>

Author: manusa

pkg/kubernetes/provider_watch_test.go

@@ -29,6 +29,7 @@ func (s *ProviderWatchTargetsTestSuite) SetupTest() {
 
 	s.T().Setenv("CLUSTER_STATE_POLL_INTERVAL_MS", "100")
 	s.T().Setenv("CLUSTER_STATE_DEBOUNCE_WINDOW_MS", "50")
+	s.T().Setenv("KUBECONFIG_DEBOUNCE_WINDOW_MS", "50")
 
 	// Add multiple fake contexts to allow testing of context changes
 	s.kubeconfig = s.mockServer.Kubeconfig()
@@ -103,16 +104,11 @@ func (s *ProviderWatchTargetsTestSuite) TestKubeConfigClusterProvider() {
 		})
 
 		s.Run("Keeps watching for further changes", func() {
-			kubeconfig := *s.kubeconfig
-			kubeconfig.CurrentContext = "context-2"
-			s.Require().NoError(clientcmd.WriteToFile(kubeconfig, s.staticConfig.KubeConfig))
+			s.kubeconfig.CurrentContext = "context-2"
+			s.Require().NoError(clientcmd.WriteToFile(*s.kubeconfig, s.staticConfig.KubeConfig))
 			s.Require().NoError(waitForCallback(5 * time.Second))
 
 			s.Run("Replaces default target with new context", func() {
-				// In Windows systems, fsnotify may trigger multiple events for a single file change,
-				s.Require().NoError(test.WaitForCondition(5*time.Second, func() bool {
-					return provider.GetDefaultTarget() == "context-2"
-				}))
 				s.Equal("context-2", provider.GetDefaultTarget(), "Expected default target context to be updated")
 			})
 		})

pkg/kubernetes/watcher/kubeconfig.go

@@ -1,24 +1,64 @@
 package watcher
 
 import (
+	"os"
+	"strconv"
+	"sync"
+	"time"
+
 	"github.com/fsnotify/fsnotify"
 	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/klog/v2"
+)
+
+const (
+	// DefaultKubeconfigDebounceWindow is the default debounce window for kubeconfig file changes
+	DefaultKubeconfigDebounceWindow = 100 * time.Millisecond
 )
 
 type Kubeconfig struct {
 	clientcmd.ClientConfig
-	close func()
+	debounceWindow time.Duration
+	debounceTimer  *time.Timer
+	mu             sync.Mutex
+	stopCh         chan struct{}
+	stoppedCh      chan struct{}
+	started        bool
 }
 
 var _ Watcher = (*Kubeconfig)(nil)
 
 func NewKubeconfig(clientConfig clientcmd.ClientConfig) *Kubeconfig {
+	debounceWindow := DefaultKubeconfigDebounceWindow
+
+	// Allow override via environment variable for testing
+	if envDebounce := os.Getenv("KUBECONFIG_DEBOUNCE_WINDOW_MS"); envDebounce != "" {
+		if ms, err := strconv.Atoi(envDebounce); err == nil && ms > 0 {
+			debounceWindow = time.Duration(ms) * time.Millisecond
+			klog.V(2).Infof("Using custom kubeconfig debounce window: %v", debounceWindow)
+		}
+	}
+
 	return &Kubeconfig{
-		ClientConfig: clientConfig,
+		ClientConfig:   clientConfig,
+		debounceWindow: debounceWindow,
+		stopCh:         make(chan struct{}),
+		stoppedCh:      make(chan struct{}),
 	}
 }
 
+// Watch starts a background watcher that monitors kubeconfig file changes
+// and triggers a debounced reload when changes are detected.
+// It can only be called once per Kubeconfig instance.
 func (w *Kubeconfig) Watch(onChange func() error) {
+	w.mu.Lock()
+	if w.started {
+		w.mu.Unlock()
+		return
+	}
+	w.started = true
+	w.mu.Unlock()
+
 	kubeConfigFiles := w.ConfigAccess().GetLoadingPrecedence()
 	if len(kubeConfigFiles) == 0 {
 		return
@@ -30,29 +70,72 @@ func (w *Kubeconfig) Watch(onChange func() error) {
 	for _, file := range kubeConfigFiles {
 		_ = watcher.Add(file)
 	}
+
 	go func() {
+		defer close(w.stoppedCh)
+		defer func() { _ = watcher.Close() }()
+
+		klog.V(2).Infof("Started kubeconfig watcher (debounce: %v)", w.debounceWindow)
+
 		for {
 			select {
+			case <-w.stopCh:
+				klog.V(2).Info("Stopping kubeconfig watcher")
+				return
 			case _, ok := <-watcher.Events:
 				if !ok {
 					return
 				}
-				_ = onChange()
+				w.mu.Lock()
+				klog.V(3).Info("Kubeconfig file change detected, scheduling debounced reload")
+				if w.debounceTimer != nil {
+					w.debounceTimer.Stop()
+				}
+				w.debounceTimer = time.AfterFunc(w.debounceWindow, func() {
+					klog.V(2).Info("Kubeconfig debounce window expired, triggering reload")
+					if err := onChange(); err != nil {
+						klog.Errorf("Failed to reload after kubeconfig change: %v", err)
+					}
+				})
+				w.mu.Unlock()
 			case _, ok := <-watcher.Errors:
 				if !ok {
 					return
 				}
 			}
 		}
 	}()
-	if w.close != nil {
-		w.close()
-	}
-	w.close = func() { _ = watcher.Close() }
 }
 
+// Close stops the kubeconfig watcher
 func (w *Kubeconfig) Close() {
-	if w.close != nil {
-		w.close()
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	if w.debounceTimer != nil {
+		w.debounceTimer.Stop()
+	}
+
+	if w.stopCh == nil || w.stoppedCh == nil {
+		return // Already closed
+	}
+
+	if !w.started {
+		return
+	}
+
+	select {
+	case <-w.stopCh:
+		// Already closed or stopped
+		return
+	default:
+		close(w.stopCh)
+		w.mu.Unlock()
+		<-w.stoppedCh
+		w.mu.Lock()
+		w.started = false
+		// Recreate channels for potential restart
+		w.stopCh = make(chan struct{})
+		w.stoppedCh = make(chan struct{})
 	}
 }