chore: ci & review bugfix

--story=120702860

Author: wcy00000000000000

src/source_controller/cacheservice/app/server.go

@@ -103,6 +103,7 @@ func Run(ctx context.Context, cancel context.CancelFunc, op *options.ServerOptio
 	}
 	select {
 	case <-ctx.Done():
+		cacheService.Scheduler().Stop()
 	}
 	return nil
 }

src/source_controller/cacheservice/cache/biz-topo/topo.go

@@ -37,7 +37,6 @@ import (
 	"configcenter/src/source_controller/cacheservice/cache/custom/cache"
 	watchcli "configcenter/src/source_controller/cacheservice/event/watch"
 	"configcenter/src/storage/driver/mongodb"
-	"configcenter/src/storage/stream/task"
 )
 
 // Topo defines the business topology caching logics
@@ -47,14 +46,14 @@ type Topo struct {
 }
 
 // New Topo
-func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task, cacheSet *cache.CacheSet,
-	watchCli *watchcli.Client) (*Topo, error) {
+func New(isMaster discovery.ServiceManageInterface, cacheSet *cache.CacheSet, watchCli *watchcli.Client) (*Topo,
+	error) {
 
 	t := &Topo{
 		isMaster: isMaster,
 	}
 
-	watcher, err := watch.New(isMaster, watchTask, cacheSet, watchCli)
+	watcher, err := watch.New(isMaster, cacheSet, watchCli)
 	if err != nil {
 		return nil, fmt.Errorf("new watcher failed, err: %v", err)
 	}
@@ -67,6 +66,11 @@ func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task, cacheS
 	return t, nil
 }
 
+// Watcher returns the business topology event watcher
+func (t *Topo) Watcher() *watch.Watcher {
+	return t.watcher
+}
+
 // loopBizTopoCache launch the task to loop business's brief topology every interval minutes.
 func (t *Topo) loopBizTopoCache(topoKey key.Key) {
 	for {

src/source_controller/cacheservice/cache/biz-topo/watch/kube.go

@@ -36,6 +36,7 @@ import (
 	tokenhandler "configcenter/src/source_controller/cacheservice/cache/token-handler"
 	dbtypes "configcenter/src/storage/dal/types"
 	"configcenter/src/storage/driver/mongodb"
+	"configcenter/src/storage/stream/task"
 	streamtypes "configcenter/src/storage/stream/types"
 
 	"go.mongodb.org/mongo-driver/bson/primitive"
@@ -113,11 +114,12 @@ func (w *kubeWatcher) watchTopo(obj string, doBatch func(*streamtypes.DBInfo, []
 			opts.WatchTaskOptions.CollOpts.EventStruct = new(kubetypes.Pod)
 		}
 
-		err = w.watcher.task.AddLoopBatchTask(opts)
+		watchTask, err := task.NewLoopBatchTask(opts)
 		if err != nil {
-			blog.Errorf("watch kube biz topo collection %s failed, err: %v", collection, err)
+			blog.Errorf("generate kube biz topo collection %s watch task failed, err: %v", collection, err)
 			return err
 		}
+		w.watcher.tasks = append(w.watcher.tasks, watchTask)
 	}
 
 	return nil

src/source_controller/cacheservice/cache/biz-topo/watch/watch.go

@@ -28,20 +28,20 @@ import (
 // Watcher defines mongodb event watcher for biz topology
 type Watcher struct {
 	isMaster discovery.ServiceManageInterface
-	task     *task.Task
 	cacheSet *cache.CacheSet
 	watchCli *watchcli.Client
+	tasks    []*task.Task
 }
 
 // New  biz topology mongodb event watcher
-func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task, cacheSet *cache.CacheSet,
-	watchCli *watchcli.Client) (*Watcher, error) {
+func New(isMaster discovery.ServiceManageInterface, cacheSet *cache.CacheSet, watchCli *watchcli.Client) (*Watcher,
+	error) {
 
 	watcher := &Watcher{
 		isMaster: isMaster,
-		task:     watchTask,
 		cacheSet: cacheSet,
 		watchCli: watchCli,
+		tasks:    make([]*task.Task, 0),
 	}
 
 	if err := watcher.watchKube(); err != nil {
@@ -54,3 +54,8 @@ func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task, cacheS
 
 	return watcher, nil
 }
+
+// GetWatchTasks returns the event watch tasks
+func (w *Watcher) GetWatchTasks() []*task.Task {
+	return w.tasks
+}

src/source_controller/cacheservice/cache/cache.go

@@ -29,24 +29,24 @@ import (
 )
 
 // NewCache new cache service
-func NewCache(watchTask *task.Task, isMaster discovery.ServiceManageInterface) (*ClientSet, error) {
+func NewCache(isMaster discovery.ServiceManageInterface) (*ClientSet, error) {
 	if err := mainline.NewMainlineCache(isMaster); err != nil {
 		return nil, fmt.Errorf("new mainline cache failed, err: %v", err)
 	}
 
-	customCache, err := custom.New(isMaster, watchTask)
+	customCache, err := custom.New(isMaster)
 	if err != nil {
 		return nil, fmt.Errorf("new custom resource cache failed, err: %v", err)
 	}
 
 	watchCli := watch.NewClient(mongodb.Dal("watch"), mongodb.Dal(), redis.Client())
 
-	generalCache, err := general.New(isMaster, watchTask, watchCli)
+	generalCache, err := general.New(isMaster, watchCli)
 	if err != nil {
 		return nil, fmt.Errorf("new general resource cache failed, err: %v", err)
 	}
 
-	topoTreeClient, err := biztopo.New(isMaster, watchTask, customCache.CacheSet(), watchCli)
+	topoTreeClient, err := biztopo.New(isMaster, customCache.CacheSet(), watchCli)
 	if err != nil {
 		return nil, fmt.Errorf("new common topo cache failed, err: %v", err)
 	}
@@ -73,3 +73,11 @@ type ClientSet struct {
 	Custom   *custom.Cache
 	General  *general.Cache
 }
+
+// GetWatchTasks returns the event watch tasks
+func (c *ClientSet) GetWatchTasks() []*task.Task {
+	tasks := c.Topo.Watcher().GetWatchTasks()
+	tasks = append(tasks, c.Custom.Watcher().GetWatchTasks()...)
+	tasks = append(tasks, c.General.FullSyncCond().GetWatchTasks()...)
+	return tasks
+}

src/source_controller/cacheservice/cache/custom/cache.go

@@ -24,23 +24,25 @@ import (
 	"configcenter/src/apimachinery/discovery"
 	"configcenter/src/source_controller/cacheservice/cache/custom/cache"
 	"configcenter/src/source_controller/cacheservice/cache/custom/watch"
-	"configcenter/src/storage/stream/task"
 )
 
 // Cache defines the custom resource caching logics
 type Cache struct {
 	cacheSet *cache.CacheSet
+	watcher  *watch.Watcher
 }
 
 // New Cache
-func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task) (*Cache, error) {
+func New(isMaster discovery.ServiceManageInterface) (*Cache, error) {
 	t := &Cache{
 		cacheSet: cache.New(isMaster),
 	}
 
-	if err := watch.Init(watchTask, t.cacheSet); err != nil {
+	watcher, err := watch.Init(t.cacheSet)
+	if err != nil {
 		return nil, fmt.Errorf("initialize custom resource watcher failed, err: %v", err)
 	}
+	t.watcher = watcher
 
 	t.cacheSet.LoopRefreshCache()
 	return t, nil
@@ -50,3 +52,8 @@ func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task) (*Cach
 func (c *Cache) CacheSet() *cache.CacheSet {
 	return c.cacheSet
 }
+
+// Watcher returns custom resource event watcher
+func (c *Cache) Watcher() *watch.Watcher {
+	return c.watcher
+}

src/source_controller/cacheservice/cache/custom/watch/watch.go

@@ -35,26 +35,31 @@ import (
 
 // Watcher defines mongodb event watcher for custom resource
 type Watcher struct {
-	task     *task.Task
 	cacheSet *cache.CacheSet
+	tasks    []*task.Task
 }
 
 // Init custom resource mongodb event watcher
-func Init(watchTask *task.Task, cacheSet *cache.CacheSet) error {
+func Init(cacheSet *cache.CacheSet) (*Watcher, error) {
 	watcher := &Watcher{
-		task:     watchTask,
 		cacheSet: cacheSet,
+		tasks:    make([]*task.Task, 0),
 	}
 
 	if err := watcher.watchPodLabel(); err != nil {
-		return err
+		return nil, err
 	}
 
 	if err := watcher.watchSharedNsRel(); err != nil {
-		return err
+		return nil, err
 	}
 
-	return nil
+	return watcher, nil
+}
+
+// GetWatchTasks returns the event watch tasks
+func (w *Watcher) GetWatchTasks() []*task.Task {
+	return w.tasks
 }
 
 type watchOptions struct {
@@ -102,11 +107,12 @@ func (w *Watcher) watchCustomResource(opt *watchOptions) (bool, error) {
 		BatchSize: 200,
 	}
 
-	err = w.task.AddLoopBatchTask(opts)
+	watchTask, err := task.NewLoopBatchTask(opts)
 	if err != nil {
-		blog.Errorf("watch custom resource %s, but add loop batch task failed, err: %v", name, err)
+		blog.Errorf("watch custom resource %s, but generate loop batch task failed, err: %v", name, err)
 		return false, err
 	}
+	w.tasks = append(w.tasks, watchTask)
 
 	return exists, nil
 }

src/source_controller/cacheservice/cache/general/cache.go

@@ -28,7 +28,6 @@ import (
 	"configcenter/src/source_controller/cacheservice/cache/general/types"
 	"configcenter/src/source_controller/cacheservice/cache/general/watch"
 	watchcli "configcenter/src/source_controller/cacheservice/event/watch"
-	"configcenter/src/storage/stream/task"
 )
 
 // Cache defines the general resource caching logics
@@ -38,7 +37,7 @@ type Cache struct {
 }
 
 // New Cache
-func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task, watchCli *watchcli.Client) (*Cache, error) {
+func New(isMaster discovery.ServiceManageInterface, watchCli *watchcli.Client) (*Cache, error) {
 	cacheSet := cache.GetAllCache()
 
 	fullSyncCondChMap := make(map[general.ResType]chan<- types.FullSyncCondEvent)
@@ -49,7 +48,7 @@ func New(isMaster discovery.ServiceManageInterface, watchTask *task.Task, watchC
 		fullSyncCondChMap[resType] = cacheInst.FullSyncCondCh()
 	}
 
-	fullSyncCondCli, err := fullsynccond.New(watchTask, fullSyncCondChMap)
+	fullSyncCondCli, err := fullsynccond.New(fullSyncCondChMap)
 	if err != nil {
 		return nil, fmt.Errorf("init full sync cond failed, err: %v", err)
 	}

src/source_controller/cacheservice/cache/general/full-sync-cond/full_sync_cond.go

@@ -28,14 +28,13 @@ import (
 
 // FullSyncCond defines the full sync cond related logics
 type FullSyncCond struct {
-	task  *task.Task
 	chMap map[general.ResType]chan<- types.FullSyncCondEvent
+	tasks []*task.Task
 }
 
 // New FullSyncCond
-func New(watchTask *task.Task, chMap map[general.ResType]chan<- types.FullSyncCondEvent) (*FullSyncCond, error) {
+func New(chMap map[general.ResType]chan<- types.FullSyncCondEvent) (*FullSyncCond, error) {
 	f := &FullSyncCond{
-		task:  watchTask,
 		chMap: chMap,
 	}
 
@@ -45,3 +44,8 @@ func New(watchTask *task.Task, chMap map[general.ResType]chan<- types.FullSyncCo
 
 	return f, nil
 }
+
+// GetWatchTasks returns the event watch tasks
+func (f *FullSyncCond) GetWatchTasks() []*task.Task {
+	return f.tasks
+}

src/source_controller/cacheservice/cache/general/full-sync-cond/watch.go

@@ -33,6 +33,7 @@ import (
 	tokenhandler "configcenter/src/source_controller/cacheservice/cache/token-handler"
 	"configcenter/src/storage/dal/mongo/local"
 	"configcenter/src/storage/driver/mongodb"
+	"configcenter/src/storage/stream/task"
 	"configcenter/src/storage/stream/types"
 )
 
@@ -67,11 +68,12 @@ func (f *FullSyncCond) Watch() error {
 		BatchSize: 200,
 	}
 
-	err := f.task.AddLoopBatchTask(opts)
+	watchTask, err := task.NewLoopBatchTask(opts)
 	if err != nil {
-		blog.Errorf("add watch full sync cond task failed, err: %v", err)
+		blog.Errorf("generate watch full sync cond task failed, err: %v", err)
 		return err
 	}
+	f.tasks = append(f.tasks, watchTask)
 
 	return nil
 }

src/source_controller/cacheservice/event/bsrelation/bsrelation.go

@@ -14,7 +14,6 @@
 package bsrelation
 
 import (
-	"context"
 	"time"
 
 	"configcenter/src/common"
@@ -29,11 +28,22 @@ const (
 	bizSetRelationLockTTL = 1 * time.Minute
 )
 
+// BizSetRelation is the biz set relation event flow struct
+type BizSetRelation struct {
+	tasks []*task.Task
+}
+
+// GetWatchTasks returns the event watch tasks
+func (b *BizSetRelation) GetWatchTasks() []*task.Task {
+	return b.tasks
+}
+
 // NewBizSetRelation init and run biz set relation event watch
-func NewBizSetRelation(task *task.Task) error {
+func NewBizSetRelation() (*BizSetRelation, error) {
+	bizSetRel := &BizSetRelation{tasks: make([]*task.Task, 0)}
+
 	base := mixevent.MixEventFlowOptions{
 		MixKey:       event.BizSetRelationKey,
-		Task:         task,
 		EventLockKey: bizSetRelationLockKey,
 		EventLockTTL: bizSetRelationLockTTL,
 	}
@@ -42,20 +52,20 @@ func NewBizSetRelation(task *task.Task) error {
 	bizSet := base
 	bizSet.Key = event.BizSetKey
 	bizSet.WatchFields = []string{common.BKBizSetIDField, common.BKBizSetScopeField}
-	if err := newBizSetRelation(context.Background(), bizSet); err != nil {
+	if err := bizSetRel.addWatchTask(bizSet); err != nil {
 		blog.Errorf("watch biz set event for biz set relation failed, err: %v", err)
-		return err
+		return nil, err
 	}
 	blog.Info("watch biz set relation events, watch biz set success")
 
 	// watch biz event
 	biz := base
 	biz.Key = event.BizKey
-	if err := newBizSetRelation(context.Background(), biz); err != nil {
+	if err := bizSetRel.addWatchTask(biz); err != nil {
 		blog.Errorf("watch biz event for biz set relation failed, err: %v", err)
-		return err
+		return nil, err
 	}
 	blog.Info("watch biz set relation events, watch biz success")
 
-	return nil
+	return bizSetRel, nil
 }

src/source_controller/cacheservice/event/bsrelation/event.go

@@ -33,8 +33,8 @@ import (
 	"configcenter/src/storage/stream/types"
 )
 
-// newBizSetRelation init and run biz set relation event watch with sub event key
-func newBizSetRelation(ctx context.Context, opts mixevent.MixEventFlowOptions) error {
+// addWatchTask add biz set relation event watch task with sub event key
+func (b *BizSetRelation) addWatchTask(opts mixevent.MixEventFlowOptions) error {
 	relation := bizSetRelation{
 		mixKey: opts.MixKey,
 		key:    opts.Key,
@@ -66,7 +66,13 @@ func newBizSetRelation(ctx context.Context, opts mixevent.MixEventFlowOptions) e
 		return err
 	}
 
-	return flow.RunFlow(ctx)
+	flowTask, err := flow.GenWatchTask()
+	if err != nil {
+		return err
+	}
+
+	b.tasks = append(b.tasks, flowTask)
+	return nil
 }
 
 // bizSetRelation biz set relation event watch logic struct