Bseto/DB-49 migrate multiple slots

Dockerfile

@@ -37,7 +37,7 @@ COPY --from=build /kvctl/_build/kvctl ./bin/
 VOLUME /var/lib/kvctl
 
 COPY ./LICENSE ./NOTICE ./licenses ./
-COPY ./config/config.yaml /var/lib/kvctl/
+COPY ./config/* /var/lib/kvctl/
 
 EXPOSE 9379:9379
 ENTRYPOINT ["./bin/kvctl-server", "-c", "/var/lib/kvctl/config.yaml"]

cmd/client/command/consts.go

@@ -26,3 +26,8 @@ const (
 	ResourceShard     = "shard"
 	ResourceNode      = "node"
 )
+
+const (
+	MigrateSlot  = "slot"
+	MigrateClear = "clear"
+)

cmd/client/command/migrate.go

@@ -46,15 +46,20 @@ var MigrateCommand = &cobra.Command{
 	Example: `
 # Migrate slot between cluster shards 
 kvctl migrate slot <slot> --target <target_shard_index> -n <namespace> -c <cluster>
+
+# Clear the queue for migration - does not stop the current migration
+kvctl migrate clear -n <namespace> -c <cluster>
 `,
 	PreRunE: migrationPreRun,
 	RunE: func(cmd *cobra.Command, args []string) error {
 		host, _ := cmd.Flags().GetString("host")
 		client := newClient(host)
 		resource := strings.ToLower(args[0])
 		switch resource {
-		case "slot":
+		case MigrateSlot:
 			return migrateSlot(client, &migrateOptions)
+		case MigrateClear:
+			return migrateClear(client, &migrateOptions)
 		default:
 			return fmt.Errorf("unsupported resource type: %s", resource)
 		}
@@ -63,10 +68,41 @@ kvctl migrate slot <slot> --target <target_shard_index> -n <namespace> -c <clust
 	SilenceErrors: true,
 }
 
+func migrateClear(client *client, options *MigrationOptions) error {
+	rsp, err := client.restyCli.R().
+		SetPathParam("namespace", options.namespace).
+		SetPathParam("cluster", options.cluster).
+		SetBody(map[string]interface{}{
+			"slot":     options.slot,
+			"target":   options.target,
+			"slotOnly": strconv.FormatBool(options.slotOnly),
+		}).
+		Delete("/namespaces/{namespace}/clusters/{cluster}/migrate")
+	if err != nil {
+		return err
+	}
+	if rsp.IsError() {
+		return errors.New(rsp.String())
+	}
+	printLine("migrate slot[%s] task is submitted successfully.", options.slot)
+	return nil
+}
+
 func migrationPreRun(_ *cobra.Command, args []string) error {
 	if len(args) < 1 {
 		return fmt.Errorf("resource type should be specified")
 	}
+	resource := strings.ToLower(args[0])
+	if migrateOptions.namespace == "" {
+		return fmt.Errorf("namespace is required, please specify with -n or --namespace")
+	}
+	if migrateOptions.cluster == "" {
+		return fmt.Errorf("cluster is required, please specify with -c or --cluster")
+	}
+	// for migrate clear, we only need namespace and cluster
+	if resource == MigrateClear {
+		return nil
+	}
 	if len(args) < 2 {
 		return fmt.Errorf("the slot number should be specified")
 	}
@@ -76,12 +112,6 @@ func migrationPreRun(_ *cobra.Command, args []string) error {
 	}
 	migrateOptions.slot = args[1]
 
-	if migrateOptions.namespace == "" {
-		return fmt.Errorf("namespace is required, please specify with -n or --namespace")
-	}
-	if migrateOptions.cluster == "" {
-		return fmt.Errorf("cluster is required, please specify with -c or --cluster")
-	}
 	if migrateOptions.target < 0 {
 		return fmt.Errorf("target is required, please specify with --target")
 	}

config/config-local-dev-consul.yaml

@@ -0,0 +1,49 @@
+# 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.
+#
+
+addr: "0.0.0.0:9379"
+
+# Which store engine should be used by controller
+# options: etcd, zookeeper, raft, consul
+# Note: the raft engine is an experimental feature and is not recommended for production use.
+#
+# default: etcd
+storage_type: consul
+
+consul:
+  addrs:
+    - "consul0:8500"
+  elect_path:
+  tls:
+    enable: false
+    cert_file:
+    key_file:
+    ca_file:
+
+controller:
+  failover:
+    ping_interval_seconds: 3
+    min_alive_size: 5
+# Uncomment this part to save logs to filename instead of stdout
+#log:
+#  level: info
+#  filename: /data/logs/kvctl.log
+#  max_backups: 10
+#  max_age: 7
+#  max_size: 100
+#  compress: false

consts/errors.go

@@ -39,4 +39,5 @@ var (
 	ErrShardSlotIsMigrating             = errors.New("shard slot is migrating")
 	ErrShardNoMatchNewMaster            = errors.New("no match new master in shard")
 	ErrSlotStartAndStopEqual            = errors.New("start and stop of a range cannot be equal")
+	ErrNoMigrationsAvailable            = errors.New("no more migrations available in queue")
 )

controller/cluster.go

@@ -22,6 +22,7 @@ package controller
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"sync"
 	"time"
@@ -58,9 +59,23 @@ type ClusterChecker struct {
 	ctx      context.Context
 	cancelFn context.CancelFunc
 
+	startHooks []Hook
+
 	wg sync.WaitGroup
 }
 
+// HookParams, any writes/updates to the `cluster` variable will automatically
+// be saved to the ClusterChecker after all the hooks are run
+type HookParams struct {
+	options      ClusterCheckOptions
+	clusterStore store.Store
+	cluster      *store.Cluster
+	namespace    string
+	clusterName  string
+}
+
+type Hook func(context.Context, *HookParams) error
+
 func NewClusterChecker(s store.Store, ns, cluster string) *ClusterChecker {
 	ctx, cancel := context.WithCancel(context.Background())
 	c := &ClusterChecker{
@@ -78,16 +93,72 @@ func NewClusterChecker(s store.Store, ns, cluster string) *ClusterChecker {
 		ctx:      ctx,
 		cancelFn: cancel,
 	}
+	c.AddStartHook(InitializeClusterInfo()) // needs to be first to initialize cluster info
+	c.AddStartHook(MigrateAvailableSlots())
 	return c
 }
 
 func (c *ClusterChecker) Start() {
+	c.triggerStartHooks()
+
 	c.wg.Add(1)
 	go c.probeLoop()
 	c.wg.Add(1)
 	go c.migrationLoop()
 }
 
+func InitializeClusterInfo() Hook {
+	return func(ctx context.Context, params *HookParams) error {
+		clusterInfo, err := params.clusterStore.GetCluster(ctx, params.namespace, params.clusterName)
+		if err != nil {
+			return fmt.Errorf("failed to get the clusterName info from the clusterStore: %w", err)
+		}
+		params.cluster = clusterInfo
+		return nil
+	}
+}
+
+func MigrateAvailableSlots() Hook {
+	return func(ctx context.Context, params *HookParams) error {
+		if params.cluster.MigrationQueue.Available() {
+			err := params.cluster.MigrateAvailableSlots(ctx)
+			if err != nil {
+				return err
+			}
+			if err := params.clusterStore.SetCluster(ctx, params.namespace, params.cluster); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
+func (c *ClusterChecker) AddStartHook(hooks ...Hook) {
+	c.startHooks = append(c.startHooks, hooks...)
+}
+
+func (c *ClusterChecker) triggerStartHooks() {
+	log := logger.Get().With(
+		zap.String("namespace", c.namespace),
+		zap.String("cluster", c.clusterName))
+	c.clusterMu.Lock()
+	defer c.clusterMu.Unlock()
+
+	params := &HookParams{
+		options:      c.options,
+		clusterStore: c.clusterStore,
+		cluster:      c.cluster, // we have the clusterMu lock, so it's ok that hooks can access this cluster var directly
+		namespace:    c.namespace,
+		clusterName:  c.clusterName,
+	}
+	for _, hook := range c.startHooks {
+		err := hook(c.ctx, params)
+		if err != nil {
+			log.Error("start hook", zap.Error(err))
+		}
+	}
+}
+
 func (c *ClusterChecker) WithPingInterval(interval time.Duration) *ClusterChecker {
 	c.options.pingInterval = interval
 	if c.options.pingInterval < 200*time.Millisecond {
@@ -183,9 +254,12 @@ func (c *ClusterChecker) syncClusterToNodes(ctx context.Context) error {
 		return err
 	}
 	version := clusterInfo.Version.Load()
+	operationCount := 0
+	staggerDelay := 5000 * time.Millisecond
 	for _, shard := range clusterInfo.Shards {
 		for _, node := range shard.Nodes {
-			go func(n store.Node) {
+			go func(n store.Node, delay time.Duration) {
+				time.Sleep(delay)
 				log := logger.Get().With(
 					zap.String("namespace", c.namespace),
 					zap.String("cluster", c.clusterName),
@@ -198,7 +272,8 @@ func (c *ClusterChecker) syncClusterToNodes(ctx context.Context) error {
 				} else {
 					log.Info("Succeed to sync the cluster topology to the node")
 				}
-			}(node)
+			}(node, time.Duration(operationCount)*staggerDelay)
+			operationCount++
 		}
 	}
 	return nil
@@ -210,10 +285,13 @@ func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.
 	var latestClusterNodesStr string
 	var wg sync.WaitGroup
 
+	operationCount := 0
+	staggerDelay := 5000 * time.Millisecond
 	for i, shard := range cluster.Shards {
 		for _, node := range shard.Nodes {
 			wg.Add(1)
-			go func(shardIdx int, n store.Node) {
+			go func(shardIdx int, n store.Node, delay time.Duration) {
+				time.Sleep(delay)
 				defer wg.Done()
 				log := logger.Get().With(
 					zap.String("id", n.ID()),
@@ -261,7 +339,8 @@ func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.
 					mu.Unlock()
 				}
 				c.resetFailureCount(n.ID())
-			}(i, node)
+			}(i, node, time.Duration(operationCount)*staggerDelay)
+			operationCount++
 		}
 	}
 
@@ -302,8 +381,8 @@ func (c *ClusterChecker) probeLoop() {
 			}
 			c.clusterMu.Lock()
 			c.cluster = clusterInfo
-			c.clusterMu.Unlock()
 			c.parallelProbeNodes(c.ctx, clusterInfo)
+			c.clusterMu.Unlock()
 		case <-c.syncCh:
 			if err := c.syncClusterToNodes(c.ctx); err != nil {
 				log.Error("Failed to sync the clusterName to the nodes", zap.Error(err))
@@ -376,6 +455,19 @@ func (c *ClusterChecker) tryUpdateMigrationStatus(ctx context.Context, clonedClu
 				log.Info("Migrate the slot successfully", zap.String("slot", migratedSlot.String()))
 			}
 			c.updateCluster(clonedCluster)
+			if clonedCluster.MigrationQueue.Available() {
+				log.Info("Migration queue is not empty, migrating queue'd requests")
+				err = clonedCluster.MigrateAvailableSlots(ctx)
+				if err != nil {
+					log.Error("Unable to trigger", zap.Error(err))
+					continue
+				}
+				if err := c.clusterStore.SetCluster(ctx, c.namespace, clonedCluster); err != nil {
+					log.Error("Failed to update the cluster", zap.Error(err))
+					return
+				}
+				c.updateCluster(clonedCluster)
+			}
 		default:
 			clonedCluster.Shards[i].ClearMigrateState()
 			if err := c.clusterStore.SetCluster(ctx, c.namespace, clonedCluster); err != nil {