feat: informer general framework and engine/discovery interface definition

app/dubbo-admin/dubbo-admin.yaml

@@ -48,5 +48,9 @@ discovery:
       metadataReport: nacos://47.76.94.134:8848?username=nacos&password=nacos
   - type: istio
 engine:
-  type: k8s
+  name: k8s1.28.6
+  type: kubernetes
+  properties:
+    apiServerAddress: https://192.168.1.1:6443
+    kubeConfig: /etc/kubernetes/admin.conf
 controlPlane:

pkg/config/app/admin.go

@@ -41,7 +41,7 @@ type AdminConfig struct {
 	// Store configuration
 	Store *store.Config `json:"store"`
 	// Discovery configuration
-	Discovery *discovery.Config `json:"discovery"`
+	Discovery []*discovery.Config `json:"discovery"`
 	// Engine configuration
 	Engine *engine.Config `json:"engine"`
 }
@@ -50,16 +50,26 @@ var _ = &AdminConfig{}
 
 func (c *AdminConfig) Sanitize() {
 	c.Engine.Sanitize()
-	c.Discovery.Sanitize()
+	for _, d := range c.Discovery {
+		d.Sanitize()
+	}
 	c.Store.Sanitize()
 	c.Console.Sanitize()
 	c.Diagnostics.Sanitize()
 }
 
 func (c *AdminConfig) PostProcess() error {
+	discoveryPostProcess := func() error {
+		for _, d := range c.Discovery {
+			if err := d.PostProcess(); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
 	return multierr.Combine(
 		c.Engine.PostProcess(),
-		c.Discovery.PostProcess(),
+		discoveryPostProcess(),
 		c.Store.PostProcess(),
 		c.Console.PostProcess(),
 		c.Diagnostics.PostProcess(),

pkg/config/discovery/config.go

@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 an "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 discovery
 
 import "github.com/apache/dubbo-admin/pkg/config"
@@ -9,7 +26,7 @@ const (
 	nacos     Type = "nacos"
 )
 
-// Config defines Discovery Engine configuration
+// Config defines Discovery configuration
 type Config struct {
 	config.BaseConfig
 	Name    string `json:"name"`

pkg/config/engine/config.go

@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 an "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 engine
 
 import "github.com/apache/dubbo-admin/pkg/config"
@@ -11,20 +28,15 @@ const (
 
 type Config struct {
 	config.BaseConfig
-	Name string `json:"name"`
-	Type Type   `json:"type"`
-}
-
-// AddressConfig defines Discovery Engine address
-type AddressConfig struct {
-	Registry       string `json:"registry"`
-	ConfigCenter   string `json:"configCenter"`
-	MetadataReport string `json:"metadataReport"`
+	Name       string            `json:"name"`
+	Type       Type              `json:"type"`
+	Properties map[string]string `json:"properties"`
 }
 
 func DefaultResourceEngineConfig() *Config {
 	return &Config{
-		Name: "default",
-		Type: VM,
+		Name:       "default",
+		Type:       VM,
+		Properties: map[string]string{},
 	}
 }

pkg/console/component.go

@@ -52,10 +52,6 @@ func (c *consoleWebServer) Type() runtime.ComponentType {
 	return runtime.Console
 }
 
-func (c *consoleWebServer) SubType() runtime.ComponentSubType {
-	return runtime.DefaultComponentSubType
-}
-
 func (c *consoleWebServer) Order() int {
 	return math.MaxInt
 }

pkg/core/bootstrap/bootstrap.go

@@ -33,6 +33,10 @@ func Bootstrap(appCtx context.Context, cfg app.AdminConfig) (runtime.Runtime, er
 	if err != nil {
 		return nil, err
 	}
+	// 0. initialize event bus
+	if err := initEventBus(builder); err != nil {
+		return nil, err
+	}
 	// 1. initialize resource store
 	if err := initResourceStore(cfg, builder); err != nil {
 		return nil, err
@@ -64,6 +68,14 @@ func Bootstrap(appCtx context.Context, cfg app.AdminConfig) (runtime.Runtime, er
 	return rt, nil
 }
 
+func initEventBus(builder *runtime.Builder) error {
+	comp, err := runtime.ComponentRegistry().EventBus()
+	if err != nil {
+		return err
+	}
+	return initAndActivateComponent(builder, comp)
+}
+
 func initResourceStore(cfg app.AdminConfig, builder *runtime.Builder) error {
 	comp, err := runtime.ComponentRegistry().ResourceStore()
 	if err != nil {

pkg/core/controller/informer.go

@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 an "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 controller
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/apache/dubbo-admin/pkg/core/events"
+	"github.com/apache/dubbo-admin/pkg/core/logger"
+	"github.com/apache/dubbo-admin/pkg/core/resource/model"
+	"github.com/apache/dubbo-admin/pkg/core/store"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+)
+
+// Informer is transferred from cache.SharedInformer, and modified to support event distribution in events.EventBus
+type Informer interface {
+	// Run starts and runs the shared informer, returning after it stops.
+	// The informer will be stopped when stopCh is closed.
+	Run(stopCh <-chan struct{})
+	// IsStopped reports whether the informer has already been stopped.
+	// Adding event handlers to already stopped informers is not possible.
+	// An informer already stopped will never be started again.
+	IsStopped() bool
+}
+
+// Options configures an informer.
+type Options struct {
+	// ResyncPeriod is the default event handler resync period and resync check
+	// period. If unset/unspecified, these are defaulted to 0 (do not resync).
+	ResyncPeriod time.Duration
+}
+
+// informer implements Informer and has three
+// main components.  One is the cache.Indexer which provides curd operations for objects.
+// The second main component is a cache.Controller that pulls
+// objects/notifications using the ListerWatcher and pushes them into
+// a cache.DeltaFIFO --- whose knownObjects is the informer's indexer
+// --- while concurrently Popping Deltas values from that fifo and
+// processing them with informer.HandleDeltas.  Each
+// invocation of HandleDeltas, which is done with the fifo's lock
+// held, processes each Delta in turn.  For each cache.Delta this both
+// updates the store and emit the event to the events.EventBus
+// The third main component is emitter, which is responsible for
+// event distribution
+type informer struct {
+	// see store.ResourceStore
+	indexer cache.Indexer
+	// controller is the underlying cache.Controller that pop cache.Delta from the fifo queue
+	controller cache.Controller
+	// listerWatcher is where we got our initial list of objects and where we perform a watch from.
+	listerWatcher cache.ListerWatcher
+	// emitter is used to emit events to events.EventBus
+	emitter events.Emitter
+	// objectType is an example object of the type this informer is expected to handle. If set, an event
+	// with an object with a mismatching type is dropped instead of being delivered to listeners.
+	objectType runtime.Object
+	// resyncCheckPeriod is how often we want the reflector's resync timer to fire so it can call
+	// ShouldResync to check if any of our listeners need a resync.
+	resyncCheckPeriod time.Duration
+
+	started, stopped bool
+	startedLock      sync.Mutex
+	// blockDeltas gives a way to stop all event distribution so that a late event handler
+	// can safely join the shared informer.
+	blockDeltas sync.Mutex
+	// Called whenever the ListAndWatch drops the connection with an error.
+	watchErrorHandler cache.WatchErrorHandler
+	// transform is an optional function that is called on each object before it is pushed into the queue.
+	transform cache.TransformFunc
+}
+
+func NewInformerWithOptions(lw cache.ListerWatcher, emitter events.Emitter, store store.ResourceStore,
+	exampleObject runtime.Object, options Options) Informer {
+	return &informer{
+		indexer:           store,
+		listerWatcher:     lw,
+		emitter:           emitter,
+		objectType:        exampleObject,
+		resyncCheckPeriod: options.ResyncPeriod,
+	}
+}
+
+func (s *informer) SetWatchErrorHandler(handler cache.WatchErrorHandler) error {
+	s.startedLock.Lock()
+	defer s.startedLock.Unlock()
+
+	if s.started {
+		return fmt.Errorf("informer has already started")
+	}
+
+	s.watchErrorHandler = handler
+	return nil
+}
+
+func (s *informer) SetTransform(handler cache.TransformFunc) error {
+	s.startedLock.Lock()
+	defer s.startedLock.Unlock()
+
+	if s.started {
+		return fmt.Errorf("informer has already started")
+	}
+
+	s.transform = handler
+	return nil
+}
+
+func (s *informer) Run(stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer func() {
+		s.startedLock.Lock()
+		defer s.startedLock.Unlock()
+		s.stopped = true // Don't want any new listeners
+	}()
+
+	if s.HasStarted() {
+		klog.Warningf("The informer has started, run more than once is not allowed")
+		return
+	}
+
+	func() {
+		s.startedLock.Lock()
+		defer s.startedLock.Unlock()
+
+		fifo := cache.NewDeltaFIFOWithOptions(cache.DeltaFIFOOptions{
+			KnownObjects:          s.indexer,
+			EmitDeltaTypeReplaced: true,
+			Transformer:           s.transform,
+		})
+
+		// We turn off the resync mechanism because we don't want to re-list all objects.
+		cfg := &cache.Config{
+			Queue:             fifo,
+			ListerWatcher:     s.listerWatcher,
+			ObjectType:        s.objectType,
+			FullResyncPeriod:  s.resyncCheckPeriod,
+			ShouldResync:      s.ShouldResync,
+			Process:           s.HandleDeltas,
+			WatchErrorHandler: s.watchErrorHandler,
+		}
+
+		s.controller = cache.New(cfg)
+		s.started = true
+	}()
+
+	s.controller.Run(stopCh)
+}
+
+func (s *informer) HasStarted() bool {
+	s.startedLock.Lock()
+	defer s.startedLock.Unlock()
+	return s.started
+}
+
+// ShouldResync if the informer's resyncPeriod is non-zero, resync will be periodically triggered.
+func (s *informer) ShouldResync() bool {
+	return s.resyncCheckPeriod != 0
+}
+
+// HandleDeltas is called for each delta when pop out from queue.
+func (s *informer) HandleDeltas(obj interface{}, _ bool) error {
+	s.blockDeltas.Lock()
+	defer s.blockDeltas.Unlock()
+
+	deltas, ok := obj.(cache.Deltas)
+	if !ok {
+		return errors.New("object given as Process argument is not Deltas")
+	}
+	// from oldest to newest
+	for _, d := range deltas {
+		obj := d.Object
+		resource, ok := obj.(model.Resource)
+		if !ok {
+			logger.Errorf("object from ListWatcher is not conformed to Resource, obj: %v", obj)
+			return errors.New("object from ListWatcher is not conformed to Resource")
+		}
+		switch d.Type {
+		case cache.Sync, cache.Replaced, cache.Added, cache.Updated:
+			if old, exists, err := s.indexer.Get(resource); err == nil && exists {
+				if err := s.indexer.Update(resource); err != nil {
+					return err
+				}
+				s.EmitEvent(d.Type, old.(model.Resource), resource)
+			} else {
+				if err := s.indexer.Add(obj); err != nil {
+					return err
+				}
+				s.EmitEvent(d.Type, nil, resource)
+			}
+		case cache.Deleted:
+			if err := s.indexer.Delete(obj); err != nil {
+				return err
+			}
+			s.EmitEvent(d.Type, resource, nil)
+		}
+	}
+	return nil
+}
+
+// EmitEvent emits an event to the event bus.
+func (s *informer) EmitEvent(typ cache.DeltaType, oldObj model.Resource, newObj model.Resource) {
+	event := events.NewResourceChangedEvent(typ, oldObj, newObj)
+	s.emitter.Send(event)
+}
+
+// IsStopped reports whether the informer has already been stopped.
+func (s *informer) IsStopped() bool {
+	s.startedLock.Lock()
+	defer s.startedLock.Unlock()
+	return s.stopped
+}