feat(kubernetes): cluster state and kubeconfig watchers

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

Author: manusa

pkg/kubernetes/kubernetes_derived_test.go

@@ -50,7 +50,6 @@ users:
 		s.Run("without authorization header returns original clientset", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			derived, err := testManager.Derived(s.T().Context())
 			s.Require().NoErrorf(err, "failed to create derived kubernetes: %v", err)
@@ -61,7 +60,6 @@ users:
 		s.Run("with invalid authorization header returns original clientset", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			ctx := context.WithValue(s.T().Context(), HeaderKey("Authorization"), "invalid-token")
 			derived, err := testManager.Derived(ctx)
@@ -73,7 +71,6 @@ users:
 		s.Run("with valid bearer token creates derived kubernetes with correct configuration", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			ctx := context.WithValue(s.T().Context(), HeaderKey("Authorization"), "Bearer aiTana-julIA")
 			derived, err := testManager.Derived(ctx)
@@ -150,7 +147,6 @@ users:
 		s.Run("with bearer token but RawConfig fails returns original clientset", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			// Corrupt the clientCmdConfig by setting it to a config that will fail on RawConfig()
 			// We'll do this by creating a config with an invalid file path
@@ -191,7 +187,6 @@ users:
 			`)))
 			testManager, err := NewKubeconfigManager(workingConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			// Now create a bad manager with RequireOAuth=true
 			badManager, _ := NewManager(testStaticConfig, testManager.accessControlClientset.cfg, testManager.accessControlClientset.clientCmdConfig)
@@ -219,7 +214,6 @@ users:
 		s.Run("with bearer token but invalid rest config returns original clientset", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			// Corrupt the rest config to make NewAccessControlClientset fail
 			// Setting an invalid Host URL should cause client creation to fail
@@ -241,7 +235,6 @@ users:
 		s.Run("with bearer token but invalid rest config returns error", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			// Corrupt the rest config to make NewAccessControlClientset fail
 			testManager.accessControlClientset.cfg.Host = "://invalid-url"
@@ -263,7 +256,6 @@ users:
 		s.Run("with no authorization header returns oauth token required error", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			derived, err := testManager.Derived(s.T().Context())
 			s.Require().Error(err, "expected error for missing oauth token, got nil")
@@ -274,7 +266,6 @@ users:
 		s.Run("with invalid authorization header returns oauth token required error", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			ctx := context.WithValue(s.T().Context(), HeaderKey("Authorization"), "invalid-token")
 			derived, err := testManager.Derived(ctx)
@@ -286,7 +277,6 @@ users:
 		s.Run("with valid bearer token creates derived kubernetes", func() {
 			testManager, err := NewKubeconfigManager(testStaticConfig, "")
 			s.Require().NoErrorf(err, "failed to create test manager: %v", err)
-			s.T().Cleanup(testManager.Close)
 
 			ctx := context.WithValue(s.T().Context(), HeaderKey("Authorization"), "Bearer aiTana-julIA")
 			derived, err := testManager.Derived(ctx)

pkg/kubernetes/manager.go

@@ -5,14 +5,10 @@ import (
 	"errors"
 	"fmt"
 	"os"
-	"sort"
 	"strconv"
 	"strings"
-	"sync"
-	"time"
 
 	"github.com/containers/kubernetes-mcp-server/pkg/config"
-	"github.com/fsnotify/fsnotify"
 	authenticationv1api "k8s.io/api/authentication/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/rest"
@@ -24,40 +20,11 @@ import (
 type Manager struct {
 	accessControlClientset *AccessControlClientset
 
-	staticConfig         *config.StaticConfig
-	CloseWatchKubeConfig CloseWatchKubeConfig
-
-	clusterWatcher *clusterStateWatcher
-}
-
-// clusterState represents the cached state of the cluster
-type clusterState struct {
-	apiGroups   []string
-	isOpenShift bool
-}
-
-// clusterStateWatcher monitors cluster state changes and triggers debounced reloads
-type clusterStateWatcher struct {
-	manager        *Manager
-	pollInterval   time.Duration
-	debounceWindow time.Duration
-	lastKnownState clusterState
-	reloadCallback func() error
-	debounceTimer  *time.Timer
-	mu             sync.Mutex
-	stopCh         chan struct{}
-	stoppedCh      chan struct{}
+	staticConfig *config.StaticConfig
 }
 
 var _ Openshift = (*Manager)(nil)
 
-const (
-	// DefaultClusterStatePollInterval is the default interval for polling cluster state changes
-	DefaultClusterStatePollInterval = 30 * time.Second
-	// DefaultClusterStateDebounceWindow is the default debounce window for cluster state changes
-	DefaultClusterStateDebounceWindow = 5 * time.Second
-)
-
 var (
 	ErrorKubeconfigInClusterNotAllowed = errors.New("kubeconfig manager cannot be used in in-cluster deployments")
 	ErrorInClusterNotInCluster         = errors.New("in-cluster manager cannot be used outside of a cluster")
@@ -148,48 +115,6 @@ func NewManager(config *config.StaticConfig, restConfig *rest.Config, clientCmdC
 	return k8s, nil
 }
 
-func (m *Manager) WatchKubeConfig(onKubeConfigChange func() error) {
-	kubeConfigFiles := m.accessControlClientset.ToRawKubeConfigLoader().ConfigAccess().GetLoadingPrecedence()
-	if len(kubeConfigFiles) == 0 {
-		return
-	}
-	watcher, err := fsnotify.NewWatcher()
-	if err != nil {
-		return
-	}
-	for _, file := range kubeConfigFiles {
-		_ = watcher.Add(file)
-	}
-	go func() {
-		for {
-			select {
-			case _, ok := <-watcher.Events:
-				if !ok {
-					return
-				}
-				_ = onKubeConfigChange()
-			case _, ok := <-watcher.Errors:
-				if !ok {
-					return
-				}
-			}
-		}
-	}()
-	if m.CloseWatchKubeConfig != nil {
-		_ = m.CloseWatchKubeConfig()
-	}
-	m.CloseWatchKubeConfig = watcher.Close
-}
-
-func (m *Manager) Close() {
-	if m.CloseWatchKubeConfig != nil {
-		_ = m.CloseWatchKubeConfig()
-	}
-	if m.clusterWatcher != nil {
-		m.clusterWatcher.stop()
-	}
-}
-
 func (m *Manager) VerifyToken(ctx context.Context, token, audience string) (*authenticationv1api.UserInfo, []string, error) {
 	tokenReviewClient := m.accessControlClientset.AuthenticationV1().TokenReviews()
 	tokenReview := &authenticationv1api.TokenReview{
@@ -266,6 +191,11 @@ func (m *Manager) Derived(ctx context.Context) (*Kubernetes, error) {
 	return &Kubernetes{derived}, nil
 }
 
+// Invalidate invalidates the cached discovery information.
+func (m *Manager) Invalidate() {
+	m.accessControlClientset.DiscoveryClient().Invalidate()
+}
+
 // applyRateLimitFromEnv applies QPS and Burst rate limits from environment variables if set.
 // This is primarily useful for tests to avoid client-side rate limiting.
 // Environment variables:
@@ -283,117 +213,3 @@ func applyRateLimitFromEnv(cfg *rest.Config) {
 		}
 	}
 }
-
-// WatchClusterState starts a background watcher that periodically polls for cluster state changes
-// and triggers a debounced reload when changes are detected.
-func (m *Manager) WatchClusterState(pollInterval, debounceWindow time.Duration, onClusterStateChange func() error) {
-	if m.clusterWatcher != nil {
-		m.clusterWatcher.stop()
-	}
-
-	watcher := &clusterStateWatcher{
-		manager:        m,
-		pollInterval:   pollInterval,
-		debounceWindow: debounceWindow,
-		reloadCallback: onClusterStateChange,
-		stopCh:         make(chan struct{}),
-		stoppedCh:      make(chan struct{}),
-	}
-
-	captureState := func() clusterState {
-		state := clusterState{apiGroups: []string{}}
-		if groups, err := m.accessControlClientset.DiscoveryClient().ServerGroups(); err == nil {
-			for _, group := range groups.Groups {
-				state.apiGroups = append(state.apiGroups, group.Name)
-			}
-			sort.Strings(state.apiGroups)
-		}
-		state.isOpenShift = m.IsOpenShift(context.Background())
-		return state
-	}
-	watcher.lastKnownState = captureState()
-
-	m.clusterWatcher = watcher
-
-	// Start background monitoring
-	go func() {
-		defer close(watcher.stoppedCh)
-		ticker := time.NewTicker(pollInterval)
-		defer ticker.Stop()
-
-		klog.V(2).Infof("Started cluster state watcher (poll interval: %v, debounce: %v)", pollInterval, debounceWindow)
-
-		for {
-			select {
-			case <-watcher.stopCh:
-				klog.V(2).Info("Stopping cluster state watcher")
-				return
-			case <-ticker.C:
-				// Invalidate discovery cache to get fresh API groups
-				m.accessControlClientset.DiscoveryClient().Invalidate()
-
-				watcher.mu.Lock()
-				current := captureState()
-				klog.V(3).Infof("Polled cluster state: %d API groups, OpenShift=%v", len(current.apiGroups), current.isOpenShift)
-
-				changed := current.isOpenShift != watcher.lastKnownState.isOpenShift ||
-					len(current.apiGroups) != len(watcher.lastKnownState.apiGroups)
-
-				if !changed {
-					for i := range current.apiGroups {
-						if current.apiGroups[i] != watcher.lastKnownState.apiGroups[i] {
-							changed = true
-							break
-						}
-					}
-				}
-
-				if changed {
-					klog.V(2).Info("Cluster state changed, scheduling debounced reload")
-					if watcher.debounceTimer != nil {
-						watcher.debounceTimer.Stop()
-					}
-					watcher.debounceTimer = time.AfterFunc(debounceWindow, func() {
-						klog.V(2).Info("Debounce window expired, triggering reload")
-						if err := onClusterStateChange(); err != nil {
-							klog.Errorf("Failed to reload: %v", err)
-						} else {
-							watcher.mu.Lock()
-							watcher.lastKnownState = captureState()
-							watcher.mu.Unlock()
-							klog.V(2).Info("Reload completed")
-						}
-					})
-				}
-				watcher.mu.Unlock()
-			}
-		}
-	}()
-}
-
-// stop stops the cluster state watcher
-func (w *clusterStateWatcher) stop() {
-	if w == nil {
-		return
-	}
-
-	w.mu.Lock()
-	defer w.mu.Unlock()
-
-	if w.debounceTimer != nil {
-		w.debounceTimer.Stop()
-	}
-
-	if w.stopCh == nil || w.stoppedCh == nil {
-		return
-	}
-
-	select {
-	case <-w.stopCh:
-		// Already closed or stopped
-		return
-	default:
-		close(w.stopCh)
-		<-w.stoppedCh
-	}
-}

pkg/kubernetes/manager_test.go

@@ -228,49 +228,6 @@ func (s *ManagerTestSuite) TestNewManager() {
 	})
 }
 
-func (s *ManagerTestSuite) TestClusterStateWatcherStop() {
-	s.Run("stop() on nil watcher", func() {
-		var watcher *clusterStateWatcher
-		// Should not panic
-		watcher.stop()
-	})
-
-	s.Run("stop() on uninitialized watcher (nil channels)", func() {
-		watcher := &clusterStateWatcher{}
-		// Should not panic even with nil channels
-		watcher.stop()
-	})
-
-	s.Run("stop() on initialized watcher", func() {
-		watcher := &clusterStateWatcher{
-			stopCh:    make(chan struct{}),
-			stoppedCh: make(chan struct{}),
-		}
-		// Close the stoppedCh to simulate a running goroutine
-		go func() {
-			<-watcher.stopCh
-			close(watcher.stoppedCh)
-		}()
-		// Should not panic and should stop cleanly
-		watcher.stop()
-	})
-
-	s.Run("stop() called multiple times", func() {
-		watcher := &clusterStateWatcher{
-			stopCh:    make(chan struct{}),
-			stoppedCh: make(chan struct{}),
-		}
-		go func() {
-			<-watcher.stopCh
-			close(watcher.stoppedCh)
-		}()
-		// First stop
-		watcher.stop()
-		// Second stop should not panic
-		watcher.stop()
-	})
-}
-
 func TestManager(t *testing.T) {
 	suite.Run(t, new(ManagerTestSuite))
 }

pkg/kubernetes/openshift.go

@@ -3,7 +3,7 @@ package kubernetes
 import (
 	"context"
 
-	"k8s.io/apimachinery/pkg/runtime/schema"
+	"github.com/containers/kubernetes-mcp-server/pkg/openshift"
 )
 
 type Openshift interface {
@@ -16,9 +16,5 @@ func (m *Manager) IsOpenShift(ctx context.Context) bool {
 	if err != nil {
 		return false
 	}
-	_, err = k.AccessControlClientset().DiscoveryClient().ServerResourcesForGroupVersion(schema.GroupVersion{
-		Group:   "project.openshift.io",
-		Version: "v1",
-	}.String())
-	return err == nil
+	return openshift.IsOpenshift(k.AccessControlClientset().DiscoveryClient())
 }