feat: add watch code chassis

Author: colin404

cmd/onex-nightwatch/app/options/options.go

@@ -22,6 +22,7 @@ import (
 	clientset "github.com/superproj/onex/pkg/generated/clientset/versioned"
 	"github.com/superproj/onex/pkg/log"
 	genericoptions "github.com/superproj/onex/pkg/options"
+	"github.com/superproj/onex/pkg/watch"
 )
 
 const (
@@ -36,8 +37,8 @@ type Options struct {
 	HealthOptions         *genericoptions.HealthOptions  `json:"health" mapstructure:"health"`
 	MySQLOptions          *genericoptions.MySQLOptions   `json:"mysql" mapstructure:"mysql"`
 	RedisOptions          *genericoptions.RedisOptions   `json:"redis" mapstructure:"redis"`
+	WatchOptions          *watch.Options                 `json:"nightwatch" mapstructure:"nightwatch"`
 	UserWatcherMaxWorkers int64                          `json:"user-watcher-max-workers" mapstructure:"user-watcher-max-workers"`
-	DisableWatchers       []string                       `json:"disable-watchers" mapstructure:"disable-watchers"`
 	Metrics               *genericoptions.MetricsOptions `json:"metrics" mapstructure:"metrics"`
 	// Path to kubeconfig file with authorization and master location information.
 	Kubeconfig   string          `json:"kubeconfig" mapstructure:"kubeconfig"`
@@ -52,7 +53,7 @@ func NewOptions() *Options {
 		MySQLOptions:          genericoptions.NewMySQLOptions(),
 		RedisOptions:          genericoptions.NewRedisOptions(),
 		UserWatcherMaxWorkers: math.MaxInt64,
-		DisableWatchers:       []string{},
+		WatchOptions:          watch.NewOptions(),
 		Metrics:               genericoptions.NewMetricsOptions(),
 		Log:                   log.NewOptions(),
 	}
@@ -65,6 +66,7 @@ func (o *Options) Flags() (fss cliflag.NamedFlagSets) {
 	o.HealthOptions.AddFlags(fss.FlagSet("health"))
 	o.MySQLOptions.AddFlags(fss.FlagSet("mysql"))
 	o.RedisOptions.AddFlags(fss.FlagSet("redis"))
+	o.WatchOptions.AddFlags(fss.FlagSet("watch"))
 	o.Metrics.AddFlags(fss.FlagSet("metrics"))
 	o.Log.AddFlags(fss.FlagSet("log"))
 
@@ -73,7 +75,6 @@ func (o *Options) Flags() (fss cliflag.NamedFlagSets) {
 	fs := fss.FlagSet("misc")
 	fs.StringVar(&o.Kubeconfig, "kubeconfig", o.Kubeconfig, "Path to kubeconfig file with authorization and master location information.")
 	fs.Int64Var(&o.UserWatcherMaxWorkers, "user-watcher-max-workers", o.UserWatcherMaxWorkers, "Specify the maximum concurrency event of user watcher.")
-	fs.StringSliceVar(&o.DisableWatchers, "disable-watchers", o.DisableWatchers, "The list of watchers that should be disabled.")
 	feature.DefaultMutableFeatureGate.AddFlag(fs)
 
 	return fss
@@ -100,6 +101,7 @@ func (o *Options) Validate() error {
 	errs = append(errs, o.HealthOptions.Validate()...)
 	errs = append(errs, o.MySQLOptions.Validate()...)
 	errs = append(errs, o.RedisOptions.Validate()...)
+	errs = append(errs, o.WatchOptions.Validate()...)
 	errs = append(errs, o.Metrics.Validate()...)
 	errs = append(errs, o.Log.Validate()...)
 
@@ -110,8 +112,8 @@ func (o *Options) Validate() error {
 func (o *Options) ApplyTo(c *nightwatch.Config) error {
 	c.MySQLOptions = o.MySQLOptions
 	c.RedisOptions = o.RedisOptions
+	c.WatchOptions = o.WatchOptions
 	c.UserWatcherMaxWorkers = o.UserWatcherMaxWorkers
-	c.DisableWatchers = o.DisableWatchers
 	return nil
 }
 

internal/nightwatch/nightwatch.go

@@ -7,51 +7,34 @@
 package nightwatch
 
 import (
-	"context"
-	"errors"
-	"time"
-
-	"github.com/go-redsync/redsync/v4"
-	"github.com/go-redsync/redsync/v4/redis/goredis/v9"
 	"github.com/jinzhu/copier"
-	"github.com/robfig/cron/v3"
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	"github.com/superproj/onex/internal/nightwatch/watcher"
 	"github.com/superproj/onex/pkg/db"
 	clientset "github.com/superproj/onex/pkg/generated/clientset/versioned"
 	"github.com/superproj/onex/pkg/log"
+	"github.com/superproj/onex/pkg/watch"
+	onexlogger "github.com/superproj/onex/pkg/watch/logger/onex"
 
 	// trigger init functions in `internal/nightwatch/watcher/all`.
 	_ "github.com/superproj/onex/internal/nightwatch/watcher/all"
 	genericoptions "github.com/superproj/onex/pkg/options"
-	stringsutil "github.com/superproj/onex/pkg/util/strings"
-)
-
-var (
-	lockName          = "onex-nightwatch-lock2"
-	jobStopTimeout    = 3 * time.Minute
-	extendExpiration  = 5 * time.Second
-	defaultExpiration = 10 * extendExpiration
 )
 
 type nightWatch struct {
-	runner *cron.Cron
-	// distributed lock
-	locker          *redsync.Mutex
-	config          *watcher.Config
-	disableWatchers []string
+	*watch.Watch
 }
 
 // Config is the configuration for the nightwatch server.
 type Config struct {
 	MySQLOptions *genericoptions.MySQLOptions
 	RedisOptions *genericoptions.RedisOptions
+	WatchOptions *watch.Options
 	// The maximum concurrency event of user watcher.
 	UserWatcherMaxWorkers int64
 	// The list of watchers that should be disabled.
-	DisableWatchers []string
-	Client          clientset.Interface
+	Client clientset.Interface
 }
 
 // CompletedConfig same as Config, just to swap private object.
@@ -65,7 +48,7 @@ func (c *Config) Complete() *CompletedConfig {
 }
 
 // CreateWatcherConfig used to create configuration used by all watcher.
-func (c *Config) CreateWatcherConfig() (*watcher.Config, error) {
+func (c *Config) CreateWatcherConfig() (*watcher.AggregateConfig, error) {
 	var mysqlOptions db.MySQLOptions
 	_ = copier.Copy(&mysqlOptions, c.MySQLOptions)
 	storeClient, err := wireStoreClient(&mysqlOptions)
@@ -74,7 +57,7 @@ func (c *Config) CreateWatcherConfig() (*watcher.Config, error) {
 		return nil, err
 	}
 
-	return &watcher.Config{
+	return &watcher.AggregateConfig{
 		Store:                 storeClient,
 		Client:                c.Client,
 		UserWatcherMaxWorkers: c.UserWatcherMaxWorkers,
@@ -83,118 +66,36 @@ func (c *Config) CreateWatcherConfig() (*watcher.Config, error) {
 
 // New creates an asynchronous task instance.
 func (c *Config) New() (*nightWatch, error) {
-	// Create a pool with go-redis which is the pool redisync will
-	// use while communicating with Redis. This can also be any pool that
-	// implements the `redis.Pool` interface.
 	client, err := c.RedisOptions.NewClient()
 	if err != nil {
 		log.Errorw(err, "Failed to create Redis client")
 		return nil, err
 	}
 
-	logger := newCronLogger()
-	runner := cron.New(
-		cron.WithSeconds(),
-		cron.WithLogger(logger),
-		cron.WithChain(cron.SkipIfStillRunning(logger), cron.Recover(logger)),
-	)
-
-	pool := goredis.NewPool(client)
-	lockOpts := []redsync.Option{
-		redsync.WithRetryDelay(50 * time.Microsecond),
-		redsync.WithTries(3),
-		redsync.WithExpiry(defaultExpiration),
-	}
-	// Create an instance of redisync and obtain a new mutex by using the same name
-	// for all instances wanting the same lock.
-	locker := redsync.New(pool).NewMutex(lockName, lockOpts...)
-
 	cfg, err := c.CreateWatcherConfig()
 	if err != nil {
 		return nil, err
 	}
 
-	nw := &nightWatch{runner: runner, locker: locker, config: cfg, disableWatchers: c.DisableWatchers}
-	if err := nw.addWatchers(); err != nil {
-		return nil, err
+	initialize := watcher.NewWatcherInitializer(cfg.Store, cfg.Client, cfg.UserWatcherMaxWorkers)
+	opts := []watch.Option{
+		watch.WithInitialize(initialize),
+		watch.WithLogger(onexlogger.NewLogger()),
+		watch.WithLockName("onex-nightwatch-lock"),
 	}
 
-	return nw, nil
-}
-
-// addWatchers used to initialize all registered watchers and add the watchers as a Cron job.
-func (nw *nightWatch) addWatchers() error {
-	for n, w := range watcher.ListWatchers() {
-		if stringsutil.StringIn(n, nw.disableWatchers) {
-			continue
-		}
-
-		if err := w.Initialize(context.Background(), nw.config); err != nil {
-			log.Errorw(err, "Failed to construct watcher", "watcher", n)
-			return err
-		}
-
-		spec := watcher.Every3Seconds
-		if obj, ok := w.(watcher.ISpec); ok {
-			spec = obj.Spec()
-		}
-
-		if _, err := nw.runner.AddJob(spec, w); err != nil {
-			log.Errorw(err, "Failed to add job to the cron", "watcher", n)
-			return err
-		}
+	nw, err := watch.NewWatch(c.WatchOptions, client, opts...)
+	if err != nil {
+		return nil, err
 	}
 
-	return nil
+	return &nightWatch{nw}, nil
 }
 
 // Run keep retrying to acquire lock and then start the Cron job.
 func (nw *nightWatch) Run(stopCh <-chan struct{}) {
-	ctx := wait.ContextForChannel(stopCh)
-
-	ticker := time.NewTicker(defaultExpiration + (5 * time.Second))
-	for {
-		// Obtain a lock for our given mutex. After this is successful, no one else
-		// can obtain the same lock (the same mutex name) until we unlock it.
-		err := nw.locker.LockContext(ctx)
-		if err == nil {
-			log.Debugw("Successfully acquired lock", "lockName", lockName)
-			break
-		}
-		log.Debugw("Failed to acquire lock.", "lockName", lockName, "err", err)
-		<-ticker.C
-	}
-
-	go func() {
-		ticker := time.NewTicker(extendExpiration)
-		for {
-			<-ticker.C
-			if ok, err := nw.locker.ExtendContext(ctx); !ok || err != nil {
-				log.Debugw("Failed to extend mutex", "err", err, "status", ok)
-			}
-		}
-	}()
-
-	nw.runner.Start()
-	log.Infof("Successfully started nightwatch server")
-
+	nw.Start(wait.ContextForChannel(stopCh))
+	// graceful shutdown
 	<-stopCh
-
-	nw.stop()
-}
-
-// stop used to blocking waits for the job to complete and releases the lock.
-func (nw *nightWatch) stop() {
-	ctx := nw.runner.Stop()
-	select {
-	case <-ctx.Done():
-	case <-time.After(jobStopTimeout):
-		log.Errorw(errors.New("context was not done immediately"), "timeout", jobStopTimeout.String())
-	}
-
-	if ok, err := nw.locker.Unlock(); !ok || err != nil {
-		log.Debugw("Failed to unlock", "err", err, "status", ok)
-	}
-
-	log.Infof("Successfully stopped nightwatch server")
+	nw.Stop()
 }

internal/nightwatch/watcher/clean/watcher.go

@@ -13,9 +13,10 @@ import (
 	"github.com/superproj/onex/internal/nightwatch/watcher"
 	"github.com/superproj/onex/internal/pkg/client/store"
 	"github.com/superproj/onex/pkg/log"
+	"github.com/superproj/onex/pkg/watch"
 )
 
-var _ watcher.Watcher = (*cleanWatcher)(nil)
+var _ watch.Watcher = (*cleanWatcher)(nil)
 
 // watcher implement.
 type cleanWatcher struct {
@@ -35,12 +36,11 @@ func (w *cleanWatcher) Run() {
 	}
 }
 
-// Initialize initializes the watcher for later execution.
-func (w *cleanWatcher) Initialize(ctx context.Context, config *watcher.Config) error {
+// SetAggregateConfig initializes the watcher for later execution.
+func (w *cleanWatcher) SetAggregateConfig(config *watcher.AggregateConfig) {
 	w.store = config.Store
-	return nil
 }
 
 func init() {
-	watcher.Register("clean", &cleanWatcher{})
+	watch.Register("clean", &cleanWatcher{})
 }

internal/nightwatch/watcher/config.go

@@ -12,8 +12,8 @@ import (
 	clientset "github.com/superproj/onex/pkg/generated/clientset/versioned"
 )
 
-// Config aggregates the configurations of all watchers and serves as a configuration aggregator.
-type Config struct {
+// AggregateConfig aggregates the configurations of all watchers and serves as a configuration aggregator.
+type AggregateConfig struct {
 	// The purpose of nightwatch is to handle asynchronous tasks on the onex platform
 	// in a unified manner, so a store aggregation type is needed here.
 	Store store.Interface

internal/nightwatch/watcher/initializer.go

@@ -0,0 +1,47 @@
+package watcher
+
+import (
+	"github.com/superproj/onex/internal/pkg/client/store"
+	clientset "github.com/superproj/onex/pkg/generated/clientset/versioned"
+	"github.com/superproj/onex/pkg/watch"
+)
+
+// WatcherInitializer is used for initialization of the onex specific watcher plugins.
+type WatcherInitializer struct {
+	// The purpose of nightwatch is to handle asynchronous tasks on the onex platform
+	// in a unified manner, so a store aggregation type is needed here.
+	store store.Interface
+
+	// Client is the client for onex-apiserver.
+	client clientset.Interface
+
+	// Then maximum concurrency event of user watcher.
+	userWatcherMaxWorkers int64
+}
+
+var _ watch.WatcherInitializer = &WatcherInitializer{}
+
+func NewWatcherInitializer(store store.Interface, client clientset.Interface, maxWorkers int64) *WatcherInitializer {
+	return &WatcherInitializer{
+		store:                 store,
+		client:                client,
+		userWatcherMaxWorkers: maxWorkers,
+	}
+}
+
+func (w *WatcherInitializer) Initialize(wc watch.Watcher) {
+	// We can set a specific configuration as needed, as shown in the example below.
+	// However, for convenience, I directly assign all configurations to each watcher,
+	// allowing the watcher to choose which ones to use.
+	if wants, ok := wc.(WantsStore); ok {
+		wants.SetStore(w.store)
+	}
+
+	if wants, ok := wc.(WantsAggregateConfig); ok {
+		wants.SetAggregateConfig(&AggregateConfig{
+			Store:                 w.store,
+			Client:                w.client,
+			UserWatcherMaxWorkers: w.userWatcherMaxWorkers,
+		})
+	}
+}

internal/nightwatch/watcher/interface.go

@@ -0,0 +1,18 @@
+package watcher
+
+import (
+	"github.com/superproj/onex/internal/pkg/client/store"
+	"github.com/superproj/onex/pkg/watch"
+)
+
+// WantsAggregateConfig defines a function which sets AggregateConfig for watcher plugins that need it.
+type WantsAggregateConfig interface {
+	watch.Watcher
+	SetAggregateConfig(config *AggregateConfig)
+}
+
+// WantsStore defines a function which sets store for watcher plugins that need it.
+type WantsStore interface {
+	watch.Watcher
+	SetStore(store store.Interface)
+}