如何实现K8S代理Watcher API-Server？技术方案咨询

Great question! The core issue here is that SharedInformer doesn't implement the watch.Interface interface natively, but there are two solid approaches to bridge this gap or use a direct alternative that fits your WatchServer needs:

方案一：实现适配器（Adapter）将SharedInformer事件适配为watch.Interface #

Since SharedInformer exposes events via callback handlers (Add/Update/Delete) while watch.Interface relies on a channel of watch.Event objects, you can build a simple adapter that translates between the two. This adapter will implement watch.Interface by funneling SharedInformer events into a channel that your WatchServer can consume.

Here's a concrete code example:

import (
    "k8s.io/apimachinery/pkg/watch"
    "k8s.io/client-go/tools/cache"
)

// SharedInformerWatchAdapter adapts a SharedInformer to the watch.Interface interface
type SharedInformerWatchAdapter struct {
    resultChan chan watch.Event
    stopChan   chan struct{}
    informer   cache.SharedIndexInformer
}

func NewSharedInformerWatchAdapter(informer cache.SharedIndexInformer) *SharedInformerWatchAdapter {
    adapter := &SharedInformerWatchAdapter{
        resultChan: make(chan watch.Event, 100), // Buffer to handle burst events
        stopChan:   make(chan struct{}),
        informer:   informer,
    }

    // Hook up SharedInformer callbacks to feed events into our channel
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            select {
            case adapter.resultChan <- watch.Event{Type: watch.Added, Object: obj}:
            case <-adapter.stopChan:
            }
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            select {
            case adapter.resultChan <- watch.Event{Type: watch.Modified, Object: newObj}:
            case <-adapter.stopChan:
            }
        },
        DeleteFunc: func(obj interface{}) {
            // Handle tombstone objects (common in informers for deleted resources)
            if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
                obj = tombstone.Obj
            }
            select {
            case adapter.resultChan <- watch.Event{Type: watch.Deleted, Object: obj}:
            case <-adapter.stopChan:
            }
        },
    })

    return adapter
}

// ResultChan returns the event channel required by watch.Interface
func (a *SharedInformerWatchAdapter) ResultChan() <-chan watch.Event {
    return a.resultChan
}

// Stop cleans up resources and stops event forwarding
func (a *SharedInformerWatchAdapter) Stop() {
    close(a.stopChan)
    close(a.resultChan)
    // Note: Only stop the informer if it's exclusively used by this adapter
    // a.informer.Stop()
}

How this works: #

1. The adapter registers handlers with your SharedInformer to capture all add/update/delete events.
2. Each event is converted to a watch.Event and sent to the adapter's internal channel.
3. The adapter implements both methods required by watch.Interface (ResultChan() and Stop()), making it compatible with your WatchServer's Watching field.

方案二：直接创建原生watch.Interface实例（更适合代理场景） #

If you don't need the local caching capabilities of SharedInformer, you can skip the adapter entirely and create a direct watch connection to the upstream Kubernetes API Server using client-go. This returns a native watch.Interface implementation that works seamlessly with your WatchServer.

Example code using the dynamic client:

import (
    "context"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime/schema"
    "k8s.io/client-go/dynamic"
    "k8s.io/client-go/rest"
    "k8s.io/apimachinery/pkg/watch"
)

func CreateDirectWatch(config *rest.Config, gvr schema.GroupVersionResource, namespace string) (watch.Interface, error) {
    dynamicClient, err := dynamic.NewForConfig(config)
    if err != nil {
        return nil, err
    }

    // Establish a direct watch connection to the upstream API Server
    watcher, err := dynamicClient.Resource(gvr).Namespace(namespace).Watch(context.TODO(), metav1.ListOptions{
        Watch: true,
        // Add any additional filters (e.g., LabelSelector: "app=my-service")
    })
    if err != nil {
        return nil, err
    }

    return watcher, nil
}

Why this is better for proxy scenarios: #

• No local cache overhead: You're just forwarding the raw watch stream from the upstream API Server to your client.
• Real-time event delivery: No delay from cache synchronization.
• No adapter complexity: The returned watcher is a native watch.Interface implementation, so you can pass it directly to your WatchServer.

Key Notes: #

• If using the adapter approach, be careful with SharedInformer lifecycle management—don't stop the informer unless it's exclusively used by your adapter (SharedInformers are often shared across multiple components).
• For proxy use cases, the direct watch approach is usually the simpler and more efficient choice.
• Always ensure proper cleanup (call Stop() on your watcher/adapter) to avoid resource leaks.

内容的提问来源于stack exchange，提问作者Yicheng Liu