feat: implement serviceadapter informer to watch CRDs in cluster

internal/serviceadapter/informer.go

@@ -0,0 +1,177 @@
+package serviceadapter
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/tools/cache"
+)
+
+// InformerManager watches ServiceAdapter resources and populates the registry.
+type InformerManager struct {
+	registry   *Registry
+	informer   cache.SharedIndexInformer
+	stopChan   chan struct{}
+	factory    dynamicinformer.DynamicSharedInformerFactory
+}
+
+// NewInformerManager creates a new informer that watches ServiceAdapters in the api namespace.
+func NewInformerManager(dynamicClient dynamic.Interface, registry *Registry, namespace string) (*InformerManager, error) {
+	if registry == nil {
+		return nil, fmt.Errorf("registry cannot be nil")
+	}
+
+	if namespace == "" {
+		namespace = "api"
+	}
+
+	// ServiceAdapter GVR
+	gvr := schema.GroupVersionResource{
+		Group:    "gateway.riotpiao.com",
+		Version:  "v1",
+		Resource: "serviceadapters",
+	}
+
+	// Create informer factory scoped to namespace
+	factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(
+		dynamicClient,
+		30*time.Second,
+		namespace,
+		nil,
+	)
+
+	// Get the informer for ServiceAdapters
+	informer := factory.ForResource(gvr).Informer()
+
+	// Add event handlers
+	_, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			unstructObj := obj.(*unstructured.Unstructured)
+			if adapter, err := parseServiceAdapter(unstructObj); err == nil {
+				if registry.Add(adapter) == nil {
+					log.Printf("[INFO] ServiceAdapter added: %s/%s", adapter.Namespace, adapter.ServiceName)
+				}
+			}
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			unstructObj := newObj.(*unstructured.Unstructured)
+			if adapter, err := parseServiceAdapter(unstructObj); err == nil {
+				if registry.Update(adapter) == nil {
+					log.Printf("[INFO] ServiceAdapter updated: %s/%s", adapter.Namespace, adapter.ServiceName)
+				}
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			unstructObj, ok := obj.(*unstructured.Unstructured)
+			if !ok {
+				// Handle tombstone (for objects that were deleted)
+				tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					return
+				}
+				unstructObj, ok = tombstone.Obj.(*unstructured.Unstructured)
+				if !ok {
+					return
+				}
+			}
+
+			if name, ok := unstructObj.Object["metadata"].(map[string]interface{})["name"]; ok {
+				registry.Delete(name.(string))
+				log.Printf("[INFO] ServiceAdapter deleted: %s", name)
+			}
+		},
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &InformerManager{
+		registry: registry,
+		informer: informer,
+		stopChan: make(chan struct{}),
+		factory:  factory,
+	}, nil
+}
+
+// Start begins watching ServiceAdapter resources.
+func (im *InformerManager) Start(ctx context.Context) error {
+	im.factory.Start(im.stopChan)
+
+	// Wait for informer cache to sync
+	if !cache.WaitForCacheSync(im.stopChan, im.informer.HasSynced) {
+		return fmt.Errorf("failed to sync ServiceAdapter informer cache")
+	}
+
+	log.Printf("[INFO] ServiceAdapter informer started, synced from cluster")
+	return nil
+}
+
+// Stop stops watching ServiceAdapter resources.
+func (im *InformerManager) Stop() {
+	close(im.stopChan)
+}
+
+// parseServiceAdapter converts an unstructured object to ServiceAdapter.
+func parseServiceAdapter(unstructured *unstructured.Unstructured) (*ServiceAdapter, error) {
+	metadata, ok := unstructured.Object["metadata"].(map[string]interface{})
+	if !ok {
+		return nil, fmt.Errorf("invalid metadata")
+	}
+
+	name, ok := metadata["name"].(string)
+	if !ok {
+		return nil, fmt.Errorf("missing metadata.name")
+	}
+
+	namespace, ok := metadata["namespace"].(string)
+	if !ok {
+		namespace = "default"
+	}
+
+	spec, ok := unstructured.Object["spec"].(map[string]interface{})
+	if !ok {
+		return nil, fmt.Errorf("missing spec")
+	}
+
+	// Parse spec fields (simplified - in real implementation would use JSON unmarshaling)
+	serviceName, ok := spec["serviceName"].(string)
+	if !ok {
+		return nil, fmt.Errorf("missing serviceName")
+	}
+
+	adapter := &ServiceAdapter{
+		Name:        name,
+		Namespace:   namespace,
+		ServiceName: serviceName,
+		CreatedAt:   time.Now(),
+	}
+
+	// Parse upstream if present
+	if upstreamMap, ok := spec["upstream"].(map[string]interface{}); ok {
+		if url, ok := upstreamMap["url"].(string); ok {
+			adapter.Spec.Upstream.URL = url
+		}
+		if timeout, ok := upstreamMap["timeoutSeconds"].(float64); ok {
+			adapter.Spec.Upstream.TimeoutSeconds = int32(timeout)
+		}
+	}
+
+	// Parse auth if present
+	if authMap, ok := spec["auth"].(map[string]interface{}); ok {
+		if required, ok := authMap["required"].(bool); ok {
+			adapter.Spec.Auth.Required = required
+		}
+		if capability, ok := authMap["capability"].(string); ok {
+			adapter.Spec.Auth.Capability = capability
+		}
+	}
+
+	return adapter, nil
+}