add split checker basic logic

maintainer/checker/balance_checker.go

@@ -0,0 +1,49 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+import (
+	"github.com/flowbehappy/tigate/maintainer/operator"
+	"github.com/flowbehappy/tigate/maintainer/replica"
+	"github.com/flowbehappy/tigate/server/watcher"
+)
+
+// BalanceChecker is used to check the balance status of all spans among all nodes
+type BalanceChecker struct {
+	changefeedID       string
+	operatorController *operator.Controller
+	replicationDB      *replica.ReplicationDB
+	nodeManager        *watcher.NodeManager
+}
+
+func NewBalanceChecker(
+	changefeedID string,
+	oc *operator.Controller,
+	db *replica.ReplicationDB,
+	nodeManager *watcher.NodeManager) *BalanceChecker {
+	return &BalanceChecker{
+		changefeedID:       changefeedID,
+		operatorController: oc,
+		replicationDB:      db,
+		nodeManager:        nodeManager,
+	}
+}
+
+func (b *BalanceChecker) Check() {
+
+}
+
+func (b *BalanceChecker) Name() string {
+	return "balance-checker"
+}

maintainer/checker/check_controller.go

@@ -0,0 +1,75 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+import (
+	"time"
+
+	"github.com/flowbehappy/tigate/maintainer/operator"
+	"github.com/flowbehappy/tigate/maintainer/replica"
+	"github.com/flowbehappy/tigate/maintainer/split"
+	"github.com/flowbehappy/tigate/server/watcher"
+)
+
+// Controller is the controller of all checkers, it will periodically execute all checkers
+type Controller struct {
+	batchSize          int
+	changefeedID       string
+	operatorController *operator.Controller
+	replicationDB      *replica.ReplicationDB
+	nodeManager        *watcher.NodeManager
+
+	maxTimePerRound time.Duration
+	checkers        []Checker
+	checkedIndex    int
+}
+
+func NewController(changefeedID string,
+	splitter *split.Splitter,
+	oc *operator.Controller,
+	db *replica.ReplicationDB,
+	nodeManager *watcher.NodeManager) *Controller {
+	c := &Controller{
+		changefeedID:       changefeedID,
+		operatorController: oc,
+		replicationDB:      db,
+		nodeManager:        nodeManager,
+		maxTimePerRound:    time.Second * 5,
+	}
+	c.checkers = []Checker{
+		NewSplitChecker(changefeedID, splitter, oc, db, nodeManager),
+		NewBalanceChecker(changefeedID, oc, db, nodeManager),
+	}
+	return c
+}
+
+// Execute periodically execute the operator
+func (ctl *Controller) Execute() time.Time {
+	now := time.Now()
+	for {
+		if ctl.checkedIndex >= len(ctl.checkers) {
+			ctl.checkedIndex = 0
+			break
+		}
+		checker := ctl.checkers[ctl.checkedIndex]
+		checker.Check()
+		ctl.checkedIndex++
+
+		if time.Since(now) > ctl.maxTimePerRound {
+			ctl.checkedIndex = ctl.checkedIndex % len(ctl.checkers)
+			break
+		}
+	}
+	return time.Now().Add(time.Second * 5)
+}

maintainer/checker/check_controller_test.go

@@ -0,0 +1,35 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestControllerExecute(t *testing.T) {
+	ctl := NewController("test", nil, nil, nil, nil)
+	require.Equal(t, 2, len(ctl.checkers))
+	ctl.maxTimePerRound = time.Hour
+	ctl.Execute()
+	require.Equal(t, 0, ctl.checkedIndex)
+	// only check the first checker
+	ctl.maxTimePerRound = -1
+	ctl.Execute()
+	require.Equal(t, 1, ctl.checkedIndex)
+	ctl.Execute()
+	require.Equal(t, 0, ctl.checkedIndex)
+}

maintainer/checker/checker.go

@@ -0,0 +1,25 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+// Checker is the interface for all checkers,
+// checker are used to check the status of the system, like balance, split, merge etc.
+// checker will be scheduled by the controller periodically and submit the result to the operator
+// checker is not an urgent task, don't have to be executed timely, it's ok to be delayed
+type Checker interface {
+	// Name returns the name of the checker
+	Name() string
+	// Check the main function of the checker, it will be called by the controller periodically
+	Check()
+}

maintainer/checker/split_checker.go

@@ -0,0 +1,100 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+import (
+	"context"
+	"time"
+
+	"github.com/flowbehappy/tigate/maintainer/operator"
+	"github.com/flowbehappy/tigate/maintainer/replica"
+	"github.com/flowbehappy/tigate/maintainer/split"
+	"github.com/flowbehappy/tigate/server/watcher"
+	"github.com/pingcap/log"
+	"go.uber.org/zap"
+)
+
+// SplitChecker is used to check the split status of all spans
+type SplitChecker struct {
+	changefeedID string
+	splitter     *split.Splitter
+	opController *operator.Controller
+	db           *replica.ReplicationDB
+	nodeManager  *watcher.NodeManager
+
+	maxCheckTime  time.Duration
+	checkInterval time.Duration
+	lastCheckTime time.Time
+
+	checkedIndex int
+	cachedSpans  []*replica.SpanReplication
+}
+
+func NewSplitChecker(
+	changefeedID string,
+	splitter *split.Splitter,
+	opController *operator.Controller,
+	db *replica.ReplicationDB,
+	nodeManager *watcher.NodeManager) *SplitChecker {
+	return &SplitChecker{
+		changefeedID: changefeedID,
+		splitter:     splitter,
+		opController: opController,
+		db:           db,
+		nodeManager:  nodeManager,
+
+		maxCheckTime:  time.Second * 5,
+		checkInterval: time.Second * 120,
+	}
+}
+
+func (s *SplitChecker) Name() string {
+	return "split-checker"
+}
+
+func (s *SplitChecker) Check() {
+	if s.splitter == nil {
+		return
+	}
+	if time.Since(s.lastCheckTime) < s.checkInterval {
+		return
+	}
+	if s.cachedSpans == nil {
+		s.cachedSpans = s.db.GetReplicating()
+		s.checkedIndex = 0
+	}
+	start := time.Now()
+	for ; s.checkedIndex < len(s.cachedSpans); s.checkedIndex++ {
+		span := s.cachedSpans[s.checkedIndex]
+		if s.db.GetTaskByID(span.ID) == nil {
+			continue
+		}
+		spans := s.splitter.SplitSpans(context.Background(), span.Span, len(s.nodeManager.GetAliveNodes()))
+		if len(spans) > 1 {
+			log.Info("split span",
+				zap.String("changefeed", s.changefeedID),
+				zap.String("span", span.ID.String()),
+				zap.Int("span szie", len(spans)))
+			s.opController.AddOperator(operator.NewSplitDispatcherOperator(s.db, span, span.GetNodeID(), spans))
+		}
+		if time.Since(start) > s.maxCheckTime {
+			break
+		}
+	}
+	if s.checkedIndex >= len(s.cachedSpans) {
+		s.cachedSpans = nil
+		s.checkedIndex = 0
+		s.lastCheckTime = time.Now()
+	}
+}

maintainer/maintainer_controller.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/flowbehappy/tigate/heartbeatpb"
+	"github.com/flowbehappy/tigate/maintainer/checker"
 	"github.com/flowbehappy/tigate/maintainer/operator"
 	"github.com/flowbehappy/tigate/maintainer/replica"
 	"github.com/flowbehappy/tigate/maintainer/scheduler"
@@ -47,6 +48,7 @@ type Controller struct {
 
 	spanScheduler      *scheduler.Scheduler
 	operatorController *operator.Controller
+	checkController    *checker.Controller
 	replicationDB      *replica.ReplicationDB
 	messageCenter      messaging.MessageCenter
 	nodeManager        *watcher.NodeManager
@@ -62,6 +64,7 @@ type Controller struct {
 	taskScheduler            threadpool.ThreadPool
 	operatorControllerHandle *threadpool.TaskHandle
 	schedulerHandle          *threadpool.TaskHandle
+	checkerHandle            *threadpool.TaskHandle
 }
 
 func NewController(changefeedID string,
@@ -91,6 +94,7 @@ func NewController(changefeedID string,
 		s.splitter = split.NewSplitter(changefeedID, pdapi, regionCache, config)
 		s.spanReplicationEnabled = true
 	}
+	s.checkController = checker.NewController(changefeedID, s.splitter, oc, replicaSetDB, nodeManager)
 	return s
 }
 
@@ -221,6 +225,7 @@ func (c *Controller) FinishBootstrap(workingMap map[int64]utils.Map[*heartbeatpb
 	// start operator and scheduler
 	c.operatorControllerHandle = c.taskScheduler.Submit(c.spanScheduler, time.Now())
 	c.schedulerHandle = c.taskScheduler.Submit(c.operatorController, time.Now())
+	c.checkerHandle = c.taskScheduler.Submit(c.checkController, time.Now().Add(time.Second*120))
 	c.bootstrapped = true
 	c.initialTables = nil
 }
@@ -232,6 +237,9 @@ func (c *Controller) Stop() {
 	if c.schedulerHandle != nil {
 		c.schedulerHandle.Cancel()
 	}
+	if c.checkerHandle != nil {
+		c.checkerHandle.Cancel()
+	}
 }
 
 // GetTask queries a task by dispatcherID, return nil if not found