Fix rule sync2

pkg/core/basic_cluster.go

@@ -309,3 +309,24 @@ func NewKeyRange(startKey, endKey string) KeyRange {
 		EndKey:   []byte(endKey),
 	}
 }
+
+// KeyRanges is a slice of KeyRange.
+type KeyRanges struct {
+	krs []*KeyRange
+}
+
+// Append appends a KeyRange.
+func (rs *KeyRanges) Append(startKey, endKey []byte) {
+	rs.krs = append(rs.krs, &KeyRange{
+		StartKey: startKey,
+		EndKey:   endKey,
+	})
+}
+
+// Ranges returns the slice of KeyRange.
+func (rs *KeyRanges) Ranges() []*KeyRange {
+	if rs == nil {
+		return nil
+	}
+	return rs.krs
+}

pkg/keyspace/tso_keyspace_group.go

@@ -245,9 +245,10 @@ func (m *GroupManager) initTSONodesWatcher(client *clientv3.Client, clusterID ui
 		client,
 		"tso-nodes-watcher",
 		tsoServiceKey,
+		func([]*clientv3.Event) error { return nil },
 		putFn,
 		deleteFn,
-		func() error { return nil },
+		func([]*clientv3.Event) error { return nil },
 		clientv3.WithRange(tsoServiceEndKey),
 	)
 }

pkg/mcs/scheduling/server/apis/v1/api.go

@@ -1330,5 +1330,5 @@ func checkRegionsReplicated(c *gin.Context) {
 		c.String(http.StatusBadRequest, err.Error())
 		return
 	}
-	c.String(http.StatusOK, state)
+	c.IndentedJSON(http.StatusOK, state)
 }

pkg/mcs/scheduling/server/cluster.go

@@ -65,7 +65,7 @@ func NewCluster(parentCtx context.Context, persistConfig *config.PersistConfig,
 		cancel()
 		return nil, err
 	}
-	ruleManager := placement.NewRuleManager(storage, basicCluster, persistConfig)
+	ruleManager := placement.NewRuleManager(ctx, storage, basicCluster, persistConfig)
 	c := &Cluster{
 		ctx:               ctx,
 		cancel:            cancel,

pkg/mcs/scheduling/server/config/watcher.go

@@ -139,22 +139,21 @@ func (cw *Watcher) initializeConfigWatcher() error {
 	deleteFn := func(kv *mvccpb.KeyValue) error {
 		return nil
 	}
-	postEventFn := func() error {
-		return nil
-	}
 	cw.configWatcher = etcdutil.NewLoopWatcher(
 		cw.ctx, &cw.wg,
 		cw.etcdClient,
 		"scheduling-config-watcher", cw.configPath,
-		putFn, deleteFn, postEventFn,
+		func([]*clientv3.Event) error { return nil },
+		putFn, deleteFn,
+		func([]*clientv3.Event) error { return nil },
 	)
 	cw.configWatcher.StartWatchLoop()
 	return cw.configWatcher.WaitLoad()
 }
 
 func (cw *Watcher) initializeTTLConfigWatcher() error {
 	putFn := func(kv *mvccpb.KeyValue) error {
-		key := string(kv.Key)[len(sc.TTLConfigPrefix)+1:]
+		key := strings.TrimPrefix(string(kv.Key), sc.TTLConfigPrefix+"/")
 		value := string(kv.Value)
 		leaseID := kv.Lease
 		resp, err := cw.etcdClient.TimeToLive(cw.ctx, clientv3.LeaseID(leaseID))
@@ -166,18 +165,18 @@ func (cw *Watcher) initializeTTLConfigWatcher() error {
 		return nil
 	}
 	deleteFn := func(kv *mvccpb.KeyValue) error {
-		key := string(kv.Key)[len(sc.TTLConfigPrefix)+1:]
+		key := strings.TrimPrefix(string(kv.Key), sc.TTLConfigPrefix+"/")
 		cw.ttl.PutWithTTL(key, nil, 0)
 		return nil
 	}
-	postEventFn := func() error {
-		return nil
-	}
 	cw.ttlConfigWatcher = etcdutil.NewLoopWatcher(
 		cw.ctx, &cw.wg,
 		cw.etcdClient,
 		"scheduling-ttl-config-watcher", cw.ttlConfigPrefix,
-		putFn, deleteFn, postEventFn, clientv3.WithPrefix(),
+		func([]*clientv3.Event) error { return nil },
+		putFn, deleteFn,
+		func([]*clientv3.Event) error { return nil },
+		clientv3.WithPrefix(),
 	)
 	cw.ttlConfigWatcher.StartWatchLoop()
 	return cw.ttlConfigWatcher.WaitLoad()
@@ -186,13 +185,14 @@ func (cw *Watcher) initializeTTLConfigWatcher() error {
 func (cw *Watcher) initializeSchedulerConfigWatcher() error {
 	prefixToTrim := cw.schedulerConfigPathPrefix + "/"
 	putFn := func(kv *mvccpb.KeyValue) error {
-		name := strings.TrimPrefix(string(kv.Key), prefixToTrim)
+		key := string(kv.Key)
+		name := strings.TrimPrefix(key, prefixToTrim)
 		log.Info("update scheduler config", zap.String("name", name),
 			zap.String("value", string(kv.Value)))
 		err := cw.storage.SaveSchedulerConfig(name, kv.Value)
 		if err != nil {
 			log.Warn("failed to save scheduler config",
-				zap.String("event-kv-key", string(kv.Key)),
+				zap.String("event-kv-key", key),
 				zap.String("trimmed-key", name),
 				zap.Error(err))
 			return err
@@ -204,19 +204,19 @@ func (cw *Watcher) initializeSchedulerConfigWatcher() error {
 		return nil
 	}
 	deleteFn := func(kv *mvccpb.KeyValue) error {
-		log.Info("remove scheduler config", zap.String("key", string(kv.Key)))
+		key := string(kv.Key)
+		log.Info("remove scheduler config", zap.String("key", key))
 		return cw.storage.RemoveSchedulerConfig(
-			strings.TrimPrefix(string(kv.Key), prefixToTrim),
+			strings.TrimPrefix(key, prefixToTrim),
 		)
 	}
-	postEventFn := func() error {
-		return nil
-	}
 	cw.schedulerConfigWatcher = etcdutil.NewLoopWatcher(
 		cw.ctx, &cw.wg,
 		cw.etcdClient,
 		"scheduling-scheduler-config-watcher", cw.schedulerConfigPathPrefix,
-		putFn, deleteFn, postEventFn,
+		func([]*clientv3.Event) error { return nil },
+		putFn, deleteFn,
+		func([]*clientv3.Event) error { return nil },
 		clientv3.WithPrefix(),
 	)
 	cw.schedulerConfigWatcher.StartWatchLoop()

pkg/mcs/scheduling/server/meta/watcher.go

@@ -73,9 +73,10 @@ func NewWatcher(
 func (w *Watcher) initializeStoreWatcher() error {
 	putFn := func(kv *mvccpb.KeyValue) error {
 		store := &metapb.Store{}
+		key := string(kv.Key)
 		if err := proto.Unmarshal(kv.Value, store); err != nil {
 			log.Warn("failed to unmarshal store entry",
-				zap.String("event-kv-key", string(kv.Key)), zap.Error(err))
+				zap.String("event-kv-key", key), zap.Error(err))
 			return err
 		}
 		origin := w.basicCluster.GetStore(store.GetId())
@@ -104,14 +105,13 @@ func (w *Watcher) initializeStoreWatcher() error {
 		}
 		return nil
 	}
-	postEventFn := func() error {
-		return nil
-	}
 	w.storeWatcher = etcdutil.NewLoopWatcher(
 		w.ctx, &w.wg,
 		w.etcdClient,
 		"scheduling-store-watcher", w.storePathPrefix,
-		putFn, deleteFn, postEventFn,
+		func([]*clientv3.Event) error { return nil },
+		putFn, deleteFn,
+		func([]*clientv3.Event) error { return nil },
 		clientv3.WithPrefix(),
 	)
 	w.storeWatcher.StartWatchLoop()