Add periodic kubeconfig refresh functionality

This commit adds the ability to periodically re-read the kubeconfig file to reflect changes like token updates. The implementation includes:

1. Thread-safe client refresh mechanism
2. Periodic refresh with configurable interval
3. Command-line flag to enable and configure refresh interval
4. Comprehensive test coverage

Author: JAORMX

cmd/server/main.go

@@ -19,6 +19,7 @@ func main() {
 	addr := flag.String("addr", ":8080", "Address to listen on")
 	serveResources := flag.Bool("serve-resources", true, "Whether to serve cluster resources as MCP resources. Setting to false can reduce context size for LLMs when working with large clusters")
 	readWrite := flag.Bool("read-write", false, "Whether to allow write operations on the cluster. When false, the server operates in read-only mode")
+	kubeconfigRefreshInterval := flag.Duration("kubeconfig-refresh-interval", 0, "Interval to periodically re-read the kubeconfig (e.g., 5m for 5 minutes). If 0, no refresh will be performed")
 	flag.Parse()
 
 	// Create a context that can be cancelled
@@ -39,6 +40,20 @@ func main() {
 	if err != nil {
 		log.Fatalf("Failed to create Kubernetes client: %v", err)
 	}
+	
+	// Start periodic refresh if interval is set
+	if *kubeconfigRefreshInterval > 0 {
+		log.Printf("Starting periodic kubeconfig refresh every %v", *kubeconfigRefreshInterval)
+		if err := k8sClient.StartPeriodicRefresh(*kubeconfigRefreshInterval); err != nil {
+			log.Fatalf("Failed to start periodic kubeconfig refresh: %v", err)
+		}
+		// Ensure we stop the refresh when shutting down
+		defer func() {
+			if err := k8sClient.StopPeriodicRefresh(); err != nil {
+				log.Printf("Error stopping periodic kubeconfig refresh: %v", err)
+			}
+		}()
+	}
 
 	// Create MCP server config
 	config := &mcp.Config{
@@ -51,10 +66,10 @@ func main() {
 
 	// Create SSE server
 	sseServer := mcp.CreateSSEServer(mcpServer)
-
+	
 	// Channel to receive server errors
 	serverErrCh := make(chan error, 1)
-
+	
 	// Start the server in a goroutine
 	go func() {
 		log.Printf("Starting MCP server on %s", *addr)
@@ -63,19 +78,19 @@ func main() {
 			serverErrCh <- err
 		}
 	}()
-
+	
 	// Wait for either a server error or a shutdown signal
 	select {
 	case err := <-serverErrCh:
 		log.Fatalf("Server failed to start: %v", err)
 	case <-ctx.Done():
 		log.Println("Shutting down server...")
 	}
-
+	
 	// Create a context with timeout for shutdown
 	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer shutdownCancel()
-
+	
 	// Attempt to shut down the server gracefully
 	shutdownCh := make(chan error, 1)
 	go func() {
@@ -87,7 +102,7 @@ func main() {
 		shutdownCh <- err
 		close(shutdownCh)
 	}()
-
+	
 	// Wait for shutdown to complete or timeout
 	select {
 	case err, ok := <-shutdownCh:
@@ -103,7 +118,7 @@ func main() {
 		// Force exit after timeout
 		os.Exit(1)
 	}
-
+	
 	log.Println("Server shutdown complete, exiting...")
 	// Ensure we exit the program
 	os.Exit(0)

pkg/k8s/client.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"path/filepath"
+	"sync"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -25,6 +27,15 @@ type Client struct {
 	dynamicClient   dynamic.Interface
 	clientset       kubernetes.Interface
 	getPodLogs      PodLogsFunc
+	kubeconfigPath  string
+	mu              sync.RWMutex // Protects access to client components
+	
+	// For periodic refresh
+	refreshCtx       context.Context
+	refreshCancel    context.CancelFunc
+	refreshInterval  time.Duration
+	refreshing       bool
+	refreshMu        sync.Mutex // Protects refreshing state
 }
 
 // NewClient creates a new Kubernetes client
@@ -56,6 +67,7 @@ func NewClient(kubeconfigPath string) (*Client, error) {
 		discoveryClient: discoveryClient,
 		dynamicClient:   dynamicClient,
 		clientset:       clientset,
+		kubeconfigPath:  kubeconfigPath,
 	}
 
 	// Set the default implementation for getPodLogs
@@ -64,8 +76,11 @@ func NewClient(kubeconfigPath string) (*Client, error) {
 	return client, nil
 }
 
-// getConfig returns a Kubernetes client configuration
-func getConfig(kubeconfigPath string) (*rest.Config, error) {
+// ConfigGetter is a function type for getting Kubernetes client configuration
+type ConfigGetter func(kubeconfigPath string) (*rest.Config, error)
+
+// defaultConfigGetter is the default implementation of ConfigGetter
+func defaultConfigGetter(kubeconfigPath string) (*rest.Config, error) {
 	if kubeconfigPath == "" {
 		// Try in-cluster config first
 		config, err := rest.InClusterConfig()
@@ -83,8 +98,14 @@ func getConfig(kubeconfigPath string) (*rest.Config, error) {
 	return clientcmd.BuildConfigFromFlags("", kubeconfigPath)
 }
 
+// getConfig is the current ConfigGetter implementation
+var getConfig ConfigGetter = defaultConfigGetter
+
 // ListAPIResources returns all API resources supported by the Kubernetes API server
 func (c *Client) ListAPIResources(ctx context.Context) ([]*metav1.APIResourceList, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	_, resourcesList, err := c.discoveryClient.ServerGroupsAndResources()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get server resources: %w", err)
@@ -94,16 +115,25 @@ func (c *Client) ListAPIResources(ctx context.Context) ([]*metav1.APIResourceLis
 
 // ListClusteredResources returns all clustered resources of the specified group/version/kind
 func (c *Client) ListClusteredResources(ctx context.Context, gvr schema.GroupVersionResource) (*unstructured.UnstructuredList, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	return c.dynamicClient.Resource(gvr).List(ctx, metav1.ListOptions{})
 }
 
 // ListNamespacedResources returns all namespaced resources of the specified group/version/kind in the given namespace
 func (c *Client) ListNamespacedResources(ctx context.Context, gvr schema.GroupVersionResource, namespace string) (*unstructured.UnstructuredList, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	return c.dynamicClient.Resource(gvr).Namespace(namespace).List(ctx, metav1.ListOptions{})
 }
 
 // ApplyClusteredResource creates or updates a clustered resource
 func (c *Client) ApplyClusteredResource(ctx context.Context, gvr schema.GroupVersionResource, obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	name := obj.GetName()
 
 	// Check if resource exists
@@ -122,16 +152,25 @@ func (c *Client) ApplyClusteredResource(ctx context.Context, gvr schema.GroupVer
 
 // GetClusteredResource gets a clustered resource
 func (c *Client) GetClusteredResource(ctx context.Context, gvr schema.GroupVersionResource, name string) (interface{}, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	return c.dynamicClient.Resource(gvr).Get(ctx, name, metav1.GetOptions{})
 }
 
 // GetNamespacedResource gets a namespaced resource
 func (c *Client) GetNamespacedResource(ctx context.Context, gvr schema.GroupVersionResource, namespace, name string) (interface{}, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	return c.dynamicClient.Resource(gvr).Namespace(namespace).Get(ctx, name, metav1.GetOptions{})
 }
 
 // ApplyNamespacedResource creates or updates a namespaced resource
 func (c *Client) ApplyNamespacedResource(ctx context.Context, gvr schema.GroupVersionResource, namespace string, obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	name := obj.GetName()
 
 	// Check if resource exists
@@ -149,31 +188,165 @@ func (c *Client) ApplyNamespacedResource(ctx context.Context, gvr schema.GroupVe
 
 // SetDynamicClient sets the dynamic client (for testing purposes)
 func (c *Client) SetDynamicClient(dynamicClient dynamic.Interface) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	
 	c.dynamicClient = dynamicClient
 }
 
 // SetDiscoveryClient sets the discovery client (for testing purposes)
 func (c *Client) SetDiscoveryClient(discoveryClient discovery.DiscoveryInterface) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	
 	c.discoveryClient = discoveryClient
 }
 
 // SetClientset sets the clientset (for testing purposes)
 func (c *Client) SetClientset(clientset kubernetes.Interface) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	
 	// Store the interface directly, we'll use it through the interface methods
 	c.clientset = clientset
 }
 
 // SetPodLogsFunc sets the function used to get pod logs (for testing purposes)
 func (c *Client) SetPodLogsFunc(getPodLogs PodLogsFunc) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	
 	c.getPodLogs = getPodLogs
 }
 
 // GetPodLogs returns the current pod logs function
 func (c *Client) GetPodLogs() PodLogsFunc {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	return c.getPodLogs
 }
 
 // IsReady returns true if the client is ready to use
 func (c *Client) IsReady() bool {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	
 	return c.discoveryClient != nil && c.dynamicClient != nil && c.clientset != nil
+}
+
+// RefreshClient re-reads the kubeconfig and updates the client's components
+func (c *Client) RefreshClient() error {
+	// Get the updated config
+	config, err := getConfig(c.kubeconfigPath)
+	if err != nil {
+		return fmt.Errorf("failed to get updated Kubernetes config: %w", err)
+	}
+
+	// Create new discovery client
+	discoveryClient, err := discovery.NewDiscoveryClientForConfig(config)
+	if err != nil {
+		return fmt.Errorf("failed to create discovery client: %w", err)
+	}
+
+	// Create new dynamic client
+	dynamicClient, err := dynamic.NewForConfig(config)
+	if err != nil {
+		return fmt.Errorf("failed to create dynamic client: %w", err)
+	}
+
+	// Create new clientset for typed API access
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return fmt.Errorf("failed to create clientset: %w", err)
+	}
+
+	// Update the client's components with proper locking
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	
+	c.discoveryClient = discoveryClient
+	c.dynamicClient = dynamicClient
+	c.clientset = clientset
+
+	return nil
+}
+
+// StartPeriodicRefresh starts a goroutine that periodically refreshes the client's configuration
+// The interval specifies how often to refresh the configuration
+// Returns an error if refresh is already running
+func (c *Client) StartPeriodicRefresh(interval time.Duration) error {
+	c.refreshMu.Lock()
+	defer c.refreshMu.Unlock()
+	
+	if c.refreshing {
+		return fmt.Errorf("periodic refresh is already running")
+	}
+	
+	// Create a cancellable context for the refresh goroutine
+	ctx, cancel := context.WithCancel(context.Background())
+	c.refreshCtx = ctx
+	c.refreshCancel = cancel
+	c.refreshInterval = interval
+	c.refreshing = true
+	
+	// Start the refresh goroutine
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		
+		for {
+			select {
+			case <-ticker.C:
+				// Refresh the client
+				if err := c.RefreshClient(); err != nil {
+					// Log the error but continue refreshing
+					fmt.Printf("Error refreshing Kubernetes client: %v\n", err)
+				}
+			case <-ctx.Done():
+				// Context cancelled, stop refreshing
+				return
+			}
+		}
+	}()
+	
+	return nil
+}
+
+// StopPeriodicRefresh stops the periodic refresh goroutine
+// Returns an error if refresh is not running
+func (c *Client) StopPeriodicRefresh() error {
+	c.refreshMu.Lock()
+	defer c.refreshMu.Unlock()
+	
+	if !c.refreshing {
+		return fmt.Errorf("periodic refresh is not running")
+	}
+	
+	// Cancel the refresh context to stop the goroutine
+	c.refreshCancel()
+	c.refreshing = false
+	
+	return nil
+}
+
+// IsRefreshing returns true if the client is periodically refreshing
+func (c *Client) IsRefreshing() bool {
+	c.refreshMu.Lock()
+	defer c.refreshMu.Unlock()
+	
+	return c.refreshing
+}
+
+// GetRefreshInterval returns the current refresh interval
+// Returns 0 if refresh is not running
+func (c *Client) GetRefreshInterval() time.Duration {
+	c.refreshMu.Lock()
+	defer c.refreshMu.Unlock()
+	
+	if !c.refreshing {
+		return 0
+	}
+	
+	return c.refreshInterval
 }