robocanic commented on code in PR #1314: URL: https://github.com/apache/dubbo-admin/pull/1314#discussion_r2296882900
########## pkg/core/controller/informer.go: ########## @@ -0,0 +1,230 @@ +/* + * 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() + + 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 + }() + + defer func() { Review Comment: This code is transferred from client-go, but it would be more appropriate to place it earlier，will fix it -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@dubbo.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: notifications-unsubscr...@dubbo.apache.org For additional commands, e-mail: notifications-h...@dubbo.apache.org
