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 CRs 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 given 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" } gvr := schema.GroupVersionResource{ Group: "gateway.riotpiao.com", Version: "v1", Resource: "serviceadapters", } factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory( dynamicClient, 30*time.Second, namespace, nil, ) informer := factory.ForResource(gvr).Informer() _, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { u := obj.(*unstructured.Unstructured) if a, err := parseServiceAdapter(u); err == nil { if registry.Add(a) == nil { log.Printf("serviceadapter added: %s", a.ServiceName) } } else { log.Printf("serviceadapter parse error on add: %v", err) } }, UpdateFunc: func(_, newObj interface{}) { u := newObj.(*unstructured.Unstructured) if a, err := parseServiceAdapter(u); err == nil { registry.Update(a) } }, DeleteFunc: func(obj interface{}) { u, ok := obj.(*unstructured.Unstructured) if !ok { tombstone, ok := obj.(cache.DeletedFinalStateUnknown) if !ok { return } u, ok = tombstone.Obj.(*unstructured.Unstructured) if !ok { return } } if name, ok := u.Object["metadata"].(map[string]interface{})["name"]; ok { registry.Delete(name.(string)) log.Printf("serviceadapter deleted: %s", name) } }, }) if err != nil { return nil, fmt.Errorf("adding event handler: %w", err) } return &InformerManager{ registry: registry, informer: informer, stopChan: make(chan struct{}), factory: factory, }, nil } // Start begins watching ServiceAdapter resources. Blocks until cache syncs or timeout. func (im *InformerManager) Start(ctx context.Context) error { im.factory.Start(im.stopChan) done := make(chan bool, 1) go func() { done <- cache.WaitForCacheSync(im.stopChan, im.informer.HasSynced) }() select { case synced := <-done: if !synced { return fmt.Errorf("cache sync failed") } case <-time.After(30 * time.Second): return fmt.Errorf("cache sync timed out") } return nil } // Stop stops the informer. func (im *InformerManager) Stop() { close(im.stopChan) } // parseServiceAdapter converts an unstructured object to ServiceAdapter. func parseServiceAdapter(u *unstructured.Unstructured) (*ServiceAdapter, error) { metadata, ok := u.Object["metadata"].(map[string]interface{}) if !ok { return nil, fmt.Errorf("invalid metadata") } name, _ := metadata["name"].(string) namespace, _ := metadata["namespace"].(string) if namespace == "" { namespace = "default" } spec, ok := u.Object["spec"].(map[string]interface{}) if !ok { return nil, fmt.Errorf("missing spec") } serviceName, ok := spec["serviceName"].(string) if !ok { return nil, fmt.Errorf("missing spec.serviceName") } adapter := &ServiceAdapter{ Name: name, Namespace: namespace, ServiceName: serviceName, CreatedAt: time.Now(), } 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) } } 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 }