feat[proxier server]: support /pods interface for http server

Author: FirmlyReality

cmd/kubeocean-proxier/main.go

@@ -44,6 +44,7 @@ import (
 
 	cloudv1beta1 "github.com/gocrane/kubeocean/api/v1beta1"
 	"github.com/gocrane/kubeocean/pkg/proxier"
+	"github.com/gocrane/kubeocean/pkg/proxier/podcache"
 	"github.com/gocrane/kubeocean/pkg/version"
 )
 
@@ -112,7 +113,7 @@ func main() {
 	}
 
 	// Setup proxier services
-	kubeletProxy, httpServer, vnodeProxierAgent, err := setupProxierServices(ctx, config, tlsConfig, virtualClient, physicalClient, physicalConfig, clusterBinding, tokenManager)
+	kubeletProxy, httpServer, vnodeProxierAgent, err := setupProxierServices(ctx, config, tlsConfig, virtualClient, physicalClient, physicalConfig, clusterBinding, tokenManager, virtualManager)
 	if err != nil {
 		setupLog.Error(err, "failed to setup proxier services")
 		os.Exit(1)
@@ -550,7 +551,7 @@ func setupTLSConfiguration(ctx context.Context, config *ProxierConfig, clusterBi
 }
 
 // setupProxierServices sets up kubelet proxy, HTTP server, and optionally VNode proxier agent
-func setupProxierServices(ctx context.Context, config *ProxierConfig, tlsConfig *TLSConfiguration, virtualClient client.Client, physicalClient kubernetes.Interface, physicalConfig *rest.Config, clusterBinding *cloudv1beta1.ClusterBinding, tokenManager *proxier.TokenManager) (proxier.KubeletProxy, proxier.HTTPServer, *proxier.VNodeProxierAgent, error) {
+func setupProxierServices(ctx context.Context, config *ProxierConfig, tlsConfig *TLSConfiguration, virtualClient client.Client, physicalClient kubernetes.Interface, physicalConfig *rest.Config, clusterBinding *cloudv1beta1.ClusterBinding, tokenManager *proxier.TokenManager, virtualManager ctrl.Manager) (proxier.KubeletProxy, proxier.HTTPServer, *proxier.VNodeProxierAgent, error) {
 	virtualClientset, err := kubernetes.NewForConfig(ctrl.GetConfigOrDie())
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("unable to create virtual cluster clientset: %w", err)
@@ -590,6 +591,13 @@ func setupProxierServices(ctx context.Context, config *ProxierConfig, tlsConfig
 		ctrl.Log.WithName("http-server"),
 	)
 
+	// Create pod cache manager using virtualManager
+	podCacheManager := podcache.NewPodCacheManager(virtualManager)
+	if err := podCacheManager.Setup(ctx); err != nil {
+		return nil, nil, nil, fmt.Errorf("failed to setup pod cache manager: %w", err)
+	}
+	setupLog.Info("Pod cache manager setup completed")
+
 	// Create VNode proxier agent if metrics enabled
 	var vnodeProxierAgent *proxier.VNodeProxierAgent
 	if config.MetricsEnabled {
@@ -614,6 +622,7 @@ func setupProxierServices(ctx context.Context, config *ProxierConfig, tlsConfig
 			virtualClientset,
 			clusterBinding.Spec.ClusterID,
 			ctrl.Log.WithName("vnode-proxier-agent"),
+			podCacheManager,
 		)
 
 		if err := vnodeProxierAgent.Start(ctx); err != nil {

pkg/proxier/metrics_collector.go

@@ -32,11 +32,15 @@ import (
 	"github.com/gorilla/mux"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apiserver/pkg/util/flushwriter"
 	"k8s.io/client-go/kubernetes"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	clientremotecommand "k8s.io/client-go/tools/remotecommand"
 
+	"github.com/gocrane/kubeocean/pkg/proxier/podcache"
 	localremotecommand "github.com/gocrane/kubeocean/pkg/proxier/remotecommand"
 )
 
@@ -60,12 +64,13 @@ type NodeEventHandler interface {
 
 // NodeInfo node information (consistent with NodeController)
 type NodeInfo struct {
+	NodeName    string
 	InternalIP  string
 	ProxierPort string
 }
 
 func (n NodeInfo) String() string {
-	return fmt.Sprintf("%s %s", n.InternalIP, n.ProxierPort)
+	return fmt.Sprintf("%s %s %s", n.NodeName, n.InternalIP, n.ProxierPort)
 }
 
 // ServerEntry HTTP server entry
@@ -120,6 +125,9 @@ type VNodeProxierAgent struct {
 	// Cluster identification for VNode name generation
 	clusterID string // ClusterBinding.Spec.ClusterID for generating VNode names
 
+	// Pod cache manager
+	podCacheManager podcache.Manager
+
 	// Node state management
 	nodeStates  map[string]NodeInfo     // key: nodeName, value: NodeInfo
 	httpServers map[string]*ServerEntry // key: port
@@ -135,26 +143,27 @@ type VNodeProxierAgent struct {
 }
 
 // NewVNodeProxierAgent creates a new VNode proxier agent
-func NewVNodeProxierAgent(config *MetricsConfig, tokenManager *TokenManager, kubeletProxy KubeletProxy, kubeClient kubernetes.Interface, clusterID string, log logr.Logger) *VNodeProxierAgent {
+func NewVNodeProxierAgent(config *MetricsConfig, tokenManager *TokenManager, kubeletProxy KubeletProxy, kubeClient kubernetes.Interface, clusterID string, log logr.Logger, podCacheManager podcache.Manager) *VNodeProxierAgent {
 	// Initialize VictoriaMetrics unmarshal workers (referring to vnode_metrics)
 	log.Info("Starting VictoriaMetrics unmarshal workers")
 	common.StartUnmarshalWorkers()
 
 	mc := &VNodeProxierAgent{
-		config:        config,
-		tokenManager:  tokenManager,
-		kubeletClient: NewKubeletClient(log.WithName("kubelet-client"), tokenManager),
-		metricsParser: NewMetricsParser(),
-		kubeletProxy:  kubeletProxy,
-		log:           log,
-		kubeClient:    kubeClient,
-		clusterID:     clusterID,
-		nodeStates:    make(map[string]NodeInfo),
-		httpServers:   make(map[string]*ServerEntry),
-		metricsCache:  make(map[string][]byte),
-		summaryCache:  make(map[string]*Summary),
-		lastUpdate:    make(map[string]time.Time),
-		stopChan:      make(chan struct{}),
+		config:          config,
+		tokenManager:    tokenManager,
+		kubeletClient:   NewKubeletClient(log.WithName("kubelet-client"), tokenManager),
+		metricsParser:   NewMetricsParser(),
+		kubeletProxy:    kubeletProxy,
+		log:             log,
+		kubeClient:      kubeClient,
+		clusterID:       clusterID,
+		podCacheManager: podCacheManager,
+		nodeStates:      make(map[string]NodeInfo),
+		httpServers:     make(map[string]*ServerEntry),
+		metricsCache:    make(map[string][]byte),
+		summaryCache:    make(map[string]*Summary),
+		lastUpdate:      make(map[string]time.Time),
+		stopChan:        make(chan struct{}),
 	}
 
 	// Load TLS configuration if provided
@@ -646,6 +655,9 @@ func (va *VNodeProxierAgent) setupRoutes(port string, nodeInfo NodeInfo) *mux.Ro
 	// Container exec endpoint (same as main server)
 	router.HandleFunc("/exec/{namespace}/{pod}/{container}", va.handleContainerExec).Methods("POST", "GET")
 
+	// Pods endpoint
+	router.HandleFunc("/pods", va.handlePods(nodeInfo.NodeName)).Methods("GET")
+
 	// Health check endpoint
 	router.HandleFunc("/healthz", va.handleHealthz).Methods("GET")
 
@@ -703,6 +715,57 @@ func (va *VNodeProxierAgent) handleSummary(port string) http.HandlerFunc {
 	}
 }
 
+// handlePods handles /pods endpoint to return pods by node name
+func (va *VNodeProxierAgent) handlePods(nodeName string) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if va.podCacheManager == nil {
+			va.log.Error(nil, "Pod cache manager is not available")
+			http.Error(w, "Pod cache manager is not available", http.StatusServiceUnavailable)
+			return
+		}
+
+		va.log.V(1).Info("Processing pods request",
+			"nodeName", nodeName,
+			"remoteAddr", r.RemoteAddr,
+			"userAgent", r.UserAgent(),
+		)
+
+		// Get pods by node name
+		ctx := r.Context()
+		podList, err := va.podCacheManager.GetPodsByNode(ctx, nodeName)
+		if err != nil {
+			va.log.Error(err, "Failed to get pods by node name", "nodeName", nodeName)
+			http.Error(w, fmt.Sprintf("Failed to get pods: %v", err), http.StatusInternalServerError)
+			return
+		}
+
+		// Encode PodList to JSON using k8s client-go codec (following kubelet's encodePods pattern)
+		// Use LegacyCodec like kubelet does for v1 API version
+		codec := clientgoscheme.Codecs.LegacyCodec(schema.GroupVersion{Group: corev1.GroupName, Version: "v1"})
+		jsonBytes, err := runtime.Encode(codec, podList)
+		if err != nil {
+			va.log.Error(err, "Failed to encode PodList to JSON", "nodeName", nodeName)
+			http.Error(w, fmt.Sprintf("Failed to encode pods: %v", err), http.StatusInternalServerError)
+			return
+		}
+
+		// Set response headers
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+
+		// Write JSON response
+		if _, err := w.Write(jsonBytes); err != nil {
+			va.log.Error(err, "Failed to write response", "nodeName", nodeName)
+			return
+		}
+
+		va.log.V(2).Info("Successfully served pods data",
+			"nodeName", nodeName,
+			"podCount", len(podList.Items),
+			"dataSize", len(jsonBytes))
+	}
+}
+
 // handleContainerLogs handles container logs requests (same as main server)
 func (va *VNodeProxierAgent) handleContainerLogs(w http.ResponseWriter, r *http.Request) {
 	if va.kubeletProxy == nil {

pkg/proxier/node_controller.go

@@ -139,6 +139,7 @@ func (r *NodeController) extractNodeInfo(node *corev1.Node) (*NodeInfo, error) {
 	}
 
 	return &NodeInfo{
+		NodeName:    node.Name,
 		InternalIP:  internalIP,
 		ProxierPort: proxierPort,
 	}, nil

pkg/proxier/podcache/pod_cache_manager.go

@@ -0,0 +1,146 @@
+// Copyright 2025 The Kubeocean Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package podcache
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+const (
+	// IndexNameNodeName is the index name for pods by node name
+	IndexNameNodeName = "spec.nodeName"
+)
+
+// Manager defines the interface for pod cache management
+type Manager interface {
+	// GetPodsByNode returns all pods running on the specified node
+	GetPodsByNode(ctx context.Context, nodeName string) (*corev1.PodList, error)
+}
+
+// PodCacheManager manages a cache of pods using controller-runtime manager
+// PodCacheManager implements the Manager interface
+type PodCacheManager struct {
+	manager ctrl.Manager
+	indexer cache.Indexer
+	logger  logr.Logger
+	mu      sync.RWMutex
+}
+
+// NewPodCacheManager creates a new PodCacheManager
+func NewPodCacheManager(manager ctrl.Manager) *PodCacheManager {
+	return &PodCacheManager{
+		manager: manager,
+		logger:  ctrl.Log.WithName("pod-cache"),
+	}
+}
+
+// Setup sets up the index for pods by node name
+// Note: This should be called before starting the manager
+func (pcm *PodCacheManager) Setup(ctx context.Context) error {
+	pcm.mu.Lock()
+	defer pcm.mu.Unlock()
+
+	// Get pod informer from controller-runtime cache
+	podInformer, err := pcm.manager.GetCache().GetInformer(ctx, &corev1.Pod{})
+	if err != nil {
+		return fmt.Errorf("failed to get pod informer: %w", err)
+	}
+
+	// Type assert to get the underlying SharedIndexInformer
+	// controller-runtime's cache wraps client-go's SharedIndexInformer
+	sharedIndexInformer, ok := podInformer.(cache.SharedIndexInformer)
+	if !ok {
+		return fmt.Errorf("failed to get indexer from informer: informer is not a SharedIndexInformer")
+	}
+
+	// Add indexer using client-go's AddIndexers method
+	// This must be called before the informer starts
+	indexers := cache.Indexers{
+		IndexNameNodeName: func(obj interface{}) ([]string, error) {
+			pod, ok := obj.(*corev1.Pod)
+			if !ok {
+				return nil, fmt.Errorf("object is not a Pod")
+			}
+			if pod.Spec.NodeName == "" {
+				return nil, nil
+			}
+			return []string{pod.Spec.NodeName}, nil
+		},
+	}
+
+	// Add the indexers to the informer's indexer
+	// Note: AddIndexers will return an error if the index already exists
+	// This can happen if the informer has already started or if the index was added elsewhere
+	if err := sharedIndexInformer.GetIndexer().AddIndexers(indexers); err != nil {
+		// Check if it's an "indexer conflict" error, which means the index already exists
+		// In that case, we can safely proceed as the index is already set up
+		if err.Error() != "indexer conflict: "+fmt.Sprintf("map[%s:{}]", IndexNameNodeName) {
+			return fmt.Errorf("failed to add indexers: %w", err)
+		}
+		pcm.logger.V(1).Info("Index already exists, reusing existing index", "indexName", IndexNameNodeName)
+	}
+
+	// Store the indexer for later use
+	pcm.indexer = sharedIndexInformer.GetIndexer()
+
+	pcm.logger.Info("Pod cache manager index setup completed")
+
+	return nil
+}
+
+// GetPodsByNode returns all pods running on the specified node
+func (pcm *PodCacheManager) GetPodsByNode(ctx context.Context, nodeName string) (*corev1.PodList, error) {
+	pcm.mu.RLock()
+	defer pcm.mu.RUnlock()
+
+	if pcm.indexer == nil {
+		return nil, fmt.Errorf("pod indexer is not initialized")
+	}
+
+	// Directly use indexer.ByIndex for optimal performance
+	// This bypasses the cache.List overhead and directly uses the index
+	pods, err := pcm.indexer.ByIndex(IndexNameNodeName, nodeName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get pods by node name %s: %w", nodeName, err)
+	}
+
+	// Convert to PodList
+	podList := &corev1.PodList{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "PodList",
+			APIVersion: "v1",
+		},
+		Items: make([]corev1.Pod, 0, len(pods)),
+	}
+
+	for _, obj := range pods {
+		pod, ok := obj.(*corev1.Pod)
+		if !ok {
+			pcm.logger.V(1).Info("Object in indexer is not a pod, skipping", "nodeName", nodeName)
+			continue
+		}
+		podList.Items = append(podList.Items, *pod)
+	}
+
+	return podList, nil
+}