fix: invalid mariadb repo fixed to 11.2.4, remove old commands and fi…

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{})
 }

internal/nightwatch/watcher/user/event.go

@@ -0,0 +1,76 @@
+package user
+
+import (
+	"context"
+	"time"
+
+	"github.com/looplab/fsm"
+
+	"github.com/superproj/onex/internal/pkg/client/store"
+	known "github.com/superproj/onex/internal/pkg/known/usercenter"
+	"github.com/superproj/onex/internal/pkg/onexx"
+	"github.com/superproj/onex/pkg/log"
+)
+
+const (
+	UserEventAfterEvent = "after_event"
+)
+
+func NewActiveUserCallback(store store.Interface) fsm.Callback {
+	return func(ctx context.Context, event *fsm.Event) {
+		userM := onexx.FromUserM(ctx)
+		log.Infow("Now active user", "event", event.Event, "username", userM.Username)
+		// Fake active user operations.
+		time.Sleep(5 * time.Second)
+		log.Infow("Success to active user", "event", event.Event, "username", userM.Username)
+	}
+}
+
+func NewDisableUserCallback(store store.Interface) fsm.Callback {
+	return func(ctx context.Context, event *fsm.Event) {
+		userM := onexx.FromUserM(ctx)
+		log.Infow("Now disable user", "event", event.Event, "username", userM.Username)
+		// Fake disable user operations.
+		time.Sleep(5 * time.Second)
+		log.Infow("Success to disable user", "event", event.Event, "username", userM.Username)
+	}
+}
+
+func NewDeleteUserCallback(store store.Interface) fsm.Callback {
+	return func(ctx context.Context, event *fsm.Event) {
+		userM := onexx.FromUserM(ctx)
+		log.Infow("Now delete user", "event", event.Event, "username", userM.Username)
+		// Fake delete user operations.
+		time.Sleep(5 * time.Second)
+		log.Infow("Success to delete user", "event", event.Event, "username", userM.Username)
+	}
+}
+
+func NewUserEventAfterEvent(store store.Interface) fsm.Callback {
+	return func(ctx context.Context, event *fsm.Event) {
+		alarmStatus := "success"
+		userM := onexx.FromUserM(ctx)
+
+		defer func() {
+			log.Infow("This is a fake alarm message", "status", alarmStatus, "username", userM.Username)
+		}()
+
+		if event.Err != nil {
+			alarmStatus = "failed"
+			log.Errorw(event.Err, "Failed to handle event", "event", event.Event)
+			// We can add some alerts here in the future.
+			return
+		}
+
+		user := onexx.FromUserM(ctx)
+		user.Status = event.FSM.Current()
+		if err := store.UserCenter().Users().Update(ctx, user); err != nil {
+			log.Errorw(err, "Failed to update status into database", "event", event.Event)
+		}
+
+		if user.Status == known.UserStatusDeleted {
+			// We can add some lark card here in the future.
+			log.Infow("Finish to handle user", "event", event.Event, "username", user.Username)
+		}
+	}
+}

internal/nightwatch/watcher/user/fsm.go

@@ -0,0 +1,26 @@
+package user
+
+import (
+	"github.com/looplab/fsm"
+
+	known "github.com/superproj/onex/internal/pkg/known/usercenter"
+)
+
+func NewFSM(initial string, w *userWatcher) *fsm.FSM {
+	return fsm.NewFSM(
+		initial,
+		fsm.Events{
+			{Name: known.UserStatusRegistered, Src: []string{known.UserStatusRegistered}, Dst: known.UserStatusActive},
+			{Name: known.UserStatusBlacklisted, Src: []string{known.UserStatusBlacklisted}, Dst: known.UserStatusDisabled},
+			{Name: known.UserStatusDisabled, Src: []string{known.UserStatusDisabled}, Dst: known.UserStatusDeleted},
+		},
+		fsm.Callbacks{
+			known.UserStatusActive:   NewActiveUserCallback(w.store),
+			known.UserStatusDisabled: NewDisableUserCallback(w.store),
+			known.UserStatusDeleted:  NewDeleteUserCallback(w.store),
+			// log, alert, save to stoer, etc for all events.
+			// Alert the status of each step of the operation.
+			UserEventAfterEvent: NewUserEventAfterEvent(w.store),
+		},
+	)
+}

internal/nightwatch/watcher/user/watcher.go

@@ -0,0 +1,83 @@
+// Copyright 2022 Lingfei Kong <[email protected]>. All rights reserved.
+// Use of this source code is governed by a MIT style
+// license that can be found in the LICENSE file. The original repo for
+// this file is https://github.com/superproj/onex.
+//
+
+// Package user is a watcher implement.
+package user
+
+import (
+	"context"
+
+	"github.com/gammazero/workerpool"
+	"github.com/looplab/fsm"
+
+	"github.com/superproj/onex/internal/nightwatch/watcher"
+	"github.com/superproj/onex/internal/pkg/client/store"
+	known "github.com/superproj/onex/internal/pkg/known/usercenter"
+	"github.com/superproj/onex/internal/pkg/onexx"
+	"github.com/superproj/onex/internal/usercenter/model"
+	"github.com/superproj/onex/pkg/log"
+	stringsutil "github.com/superproj/onex/pkg/util/strings"
+)
+
+var _ watcher.Watcher = (*userWatcher)(nil)
+
+// watcher implement.
+type userWatcher struct {
+	store      store.Interface
+	maxWorkers int64
+}
+
+type User struct {
+	*model.UserM
+	*fsm.FSM
+}
+
+// Run runs the watcher.
+func (w *userWatcher) Run() {
+	_, users, err := w.store.UserCenter().Users().List(context.Background())
+	if err != nil {
+		log.Errorw(err, "Failed to list users")
+		return
+	}
+
+	allowOperations := []string{
+		known.UserStatusRegistered,
+		known.UserStatusBlacklisted,
+		known.UserStatusDisabled,
+	}
+
+	wp := workerpool.New(int(w.maxWorkers))
+	for _, user := range users {
+		if !stringsutil.StringIn(user.Status, allowOperations) {
+			continue
+		}
+
+		wp.Submit(func() {
+			ctx := onexx.NewUserM(context.Background(), user)
+
+			u := &User{UserM: user, FSM: NewFSM(user.Status, w)}
+			if err := u.Event(ctx, user.Status); err != nil {
+				log.Errorw(err, "Failed to event user", "username", user.Username, "status", user.Status)
+				return
+			}
+
+			return
+		})
+	}
+
+	wp.StopWait()
+}
+
+// Init initializes the watcher for later execution.
+func (w *userWatcher) Init(ctx context.Context, config *watcher.Config) error {
+	w.store = config.Store
+	w.maxWorkers = config.UserWatcherMaxWorkers
+	return nil
+}
+
+func init() {
+	watcher.Register("user", &userWatcher{})
+}