feat: support finite state machine (fsm)

Author: colin404

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

@@ -8,6 +8,8 @@
 package options
 
 import (
+	"math"
+
 	"github.com/spf13/viper"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/tools/clientcmd"
@@ -31,10 +33,12 @@ var _ app.CliOptions = (*Options)(nil)
 
 // Options contains everything necessary to create and run a nightwatch server.
 type Options struct {
-	HealthOptions *genericoptions.HealthOptions  `json:"health" mapstructure:"health"`
-	MySQLOptions  *genericoptions.MySQLOptions   `json:"mysql" mapstructure:"mysql"`
-	RedisOptions  *genericoptions.RedisOptions   `json:"redis" mapstructure:"redis"`
-	Metrics       *genericoptions.MetricsOptions `json:"metrics" mapstructure:"metrics"`
+	HealthOptions         *genericoptions.HealthOptions  `json:"health" mapstructure:"health"`
+	MySQLOptions          *genericoptions.MySQLOptions   `json:"db" mapstructure:"db"`
+	RedisOptions          *genericoptions.RedisOptions   `json:"redis" mapstructure:"redis"`
+	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"`
 	FeatureGates map[string]bool `json:"feature-gates"`
@@ -44,11 +48,13 @@ type Options struct {
 // NewOptions returns initialized Options.
 func NewOptions() *Options {
 	o := &Options{
-		HealthOptions: genericoptions.NewHealthOptions(),
-		MySQLOptions:  genericoptions.NewMySQLOptions(),
-		RedisOptions:  genericoptions.NewRedisOptions(),
-		Metrics:       genericoptions.NewMetricsOptions(),
-		Log:           log.NewOptions(),
+		HealthOptions:         genericoptions.NewHealthOptions(),
+		MySQLOptions:          genericoptions.NewMySQLOptions(),
+		RedisOptions:          genericoptions.NewRedisOptions(),
+		UserWatcherMaxWorkers: math.MaxInt64,
+		DisableWatchers:       []string{},
+		Metrics:               genericoptions.NewMetricsOptions(),
+		Log:                   log.NewOptions(),
 	}
 
 	return o
@@ -57,7 +63,7 @@ func NewOptions() *Options {
 // Flags returns flags for a specific server by section name.
 func (o *Options) Flags() (fss cliflag.NamedFlagSets) {
 	o.HealthOptions.AddFlags(fss.FlagSet("health"))
-	o.MySQLOptions.AddFlags(fss.FlagSet("mysql"))
+	o.MySQLOptions.AddFlags(fss.FlagSet("db"))
 	o.RedisOptions.AddFlags(fss.FlagSet("redis"))
 	o.Metrics.AddFlags(fss.FlagSet("metrics"))
 	o.Log.AddFlags(fss.FlagSet("log"))
@@ -66,6 +72,8 @@ func (o *Options) Flags() (fss cliflag.NamedFlagSets) {
 	// arrange these text blocks sensibly. Grrr.
 	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
@@ -77,6 +85,10 @@ func (o *Options) Complete() error {
 		return err
 	}
 
+	if o.UserWatcherMaxWorkers < 1 {
+		o.UserWatcherMaxWorkers = math.MaxInt64
+	}
+
 	_ = feature.DefaultMutableFeatureGate.SetFromMap(o.FeatureGates)
 	return nil
 }
@@ -98,6 +110,8 @@ func (o *Options) Validate() error {
 func (o *Options) ApplyTo(c *nightwatch.Config) error {
 	c.MySQLOptions = o.MySQLOptions
 	c.RedisOptions = o.RedisOptions
+	c.UserWatcherMaxWorkers = o.UserWatcherMaxWorkers
+	c.DisableWatchers = o.DisableWatchers
 	return nil
 }
 

configs/onex.sql

@@ -235,6 +235,7 @@ CREATE TABLE `uc_user` (
   `user_id` varchar(253) NOT NULL DEFAULT '' COMMENT '用户 ID',
   `username` varchar(253) NOT NULL DEFAULT '' COMMENT '用户名称',
   `status` tinyint(3) unsigned NOT NULL DEFAULT 1 COMMENT '用户状态，0-禁用；1-启用',
+  `status` varchar(64) NOT NULL DEFAULT '' COMMENT '用户状态：registered,active,disabled,blacklisted,locked,deleted',
   `nickname` varchar(253) NOT NULL DEFAULT '' COMMENT '用户昵称',
   `password` varchar(64) NOT NULL DEFAULT '' COMMENT '用户加密后的密码',
   `email` varchar(253) NOT NULL DEFAULT '' COMMENT '用户电子邮箱',

go.mod

@@ -46,6 +46,7 @@ require (
 	github.com/jinzhu/copier v0.3.5
 	github.com/kisielk/errcheck v1.5.0
 	github.com/likexian/host-stat-go v0.0.0-20190516151207-c9cf36dd6ce9
+	github.com/looplab/fsm v1.0.2
 	github.com/mitchellh/go-wordwrap v1.0.1
 	github.com/nicksnyder/go-i18n/v2 v2.2.1
 	github.com/olekukonko/tablewriter v0.0.5

go.sum

@@ -1228,6 +1228,8 @@ github.com/likexian/host-stat-go v0.0.0-20190516151207-c9cf36dd6ce9/go.mod h1:Hn
 github.com/likexian/simplejson-go v0.0.0-20190409170913-40473a74d76d/go.mod h1:Typ1BfnATYtZ/+/shXfFYLrovhFyuKvzwrdOnIDHlmg=
 github.com/likexian/simplejson-go v0.0.0-20190419151922-c1f9f0b4f084/go.mod h1:U4O1vIJvIKwbMZKUJ62lppfdvkCdVd2nfMimHK81eec=
 github.com/likexian/simplejson-go v0.0.0-20190502021454-d8787b4bfa0b/go.mod h1:3BWwtmKP9cXWwYCr5bkoVDEfLywacOv0s06OBEDpyt8=
+github.com/looplab/fsm v1.0.2 h1:f0kdMzr4CRpXtaKKRUxwLYJ7PirTdwrtNumeLN+mDx8=
+github.com/looplab/fsm v1.0.2/go.mod h1:PmD3fFvQEIsjMEfvZdrCDZ6y8VwKTwWNjlpEr6IKPO4=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a h1:N9zuLhTvBSRt0gWSiJswwQ2HqDmtX/ZCDJURnKUt1Ik=
 github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a/go.mod h1:JKx41uQRwqlTZabZc+kILPrO/3jlKnQ2Z8b7YiVw5cE=

internal/controller/miner/controller.go

@@ -26,7 +26,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
-
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/superproj/onex/internal/controller/miner/apis/config"

internal/nightwatch/nightwatch.go

@@ -25,6 +25,7 @@ import (
 	// 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 (
@@ -37,15 +38,20 @@ var (
 type nightWatch struct {
 	runner *cron.Cron
 	// distributed lock
-	locker *redsync.Mutex
-	config *watcher.Config
+	locker          *redsync.Mutex
+	config          *watcher.Config
+	disableWatchers []string
 }
 
 // Config is the configuration for the nightwatch server.
 type Config struct {
 	MySQLOptions *genericoptions.MySQLOptions
 	RedisOptions *genericoptions.RedisOptions
-	Client       clientset.Interface
+	// The maximum concurrency event of user watcher.
+	UserWatcherMaxWorkers int64
+	// The list of watchers that should be disabled.
+	DisableWatchers []string
+	Client          clientset.Interface
 }
 
 // CompletedConfig same as Config, just to swap private object.
@@ -68,7 +74,11 @@ func (c *Config) CreateWatcherConfig() (*watcher.Config, error) {
 		return nil, err
 	}
 
-	return &watcher.Config{Store: storeClient, Client: c.Client}, nil
+	return &watcher.Config{
+		Store:                 storeClient,
+		Client:                c.Client,
+		UserWatcherMaxWorkers: c.UserWatcherMaxWorkers,
+	}, nil
 }
 
 // New creates an asynchronous task instance.
@@ -104,7 +114,7 @@ func (c *Config) New() (*nightWatch, error) {
 		return nil, err
 	}
 
-	nw := &nightWatch{runner: runner, locker: locker, config: cfg}
+	nw := &nightWatch{runner: runner, locker: locker, config: cfg, disableWatchers: c.DisableWatchers}
 	if err := nw.addWatchers(); err != nil {
 		return nil, err
 	}
@@ -115,6 +125,10 @@ func (c *Config) New() (*nightWatch, error) {
 // 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.Init(context.Background(), nw.config); err != nil {
 			log.Errorw(err, "Failed to construct watcher", "watcher", n)
 			return err

internal/nightwatch/watcher/all/all.go

@@ -10,4 +10,5 @@ package all
 import (
 	_ "github.com/superproj/onex/internal/nightwatch/watcher/clean"
 	_ "github.com/superproj/onex/internal/nightwatch/watcher/secretsclean"
+	_ "github.com/superproj/onex/internal/nightwatch/watcher/user"
 )

internal/nightwatch/watcher/clean/watcher.go

@@ -42,5 +42,5 @@ func (w *cleanWatcher) Init(ctx context.Context, config *watcher.Config) error {
 }
 
 func init() {
-	watcher.Register(&cleanWatcher{})
+	watcher.Register("clean", &cleanWatcher{})
 }

internal/nightwatch/watcher/config.go

@@ -20,4 +20,7 @@ type Config struct {
 
 	// Client is the client for onex-apiserver.
 	Client clientset.Interface
+
+	// Then maximum concurrency event of user watcher.
+	UserWatcherMaxWorkers int64
 }

internal/nightwatch/watcher/secretsclean/watcher.go

@@ -50,5 +50,5 @@ func (w *secretsCleanWatcher) Init(ctx context.Context, config *watcher.Config)
 }
 
 func init() {
-	watcher.Register(&secretsCleanWatcher{})
+	watcher.Register("secretsclean", &secretsCleanWatcher{})
 }