etcd_worker: batch etcd patch (#3277) (#3391)

pingcap · Dec 15, 2021 · ab3f501 · ab3f501
1 parent eccaa57
commit ab3f501
Show file tree

Hide file tree

Showing 19 changed files with 1,707 additions and 70 deletions.
diff --git a/cdc/capture.go b/cdc/capture.go
@@ -162,7 +162,7 @@ func (c *Capture) Run(ctx context.Context) (err error) {
 			return errors.Trace(err)
 		}
 		log.Info("start to listen processor task...")
-		if err := etcdWorker.Run(ctx, c.session, 200*time.Millisecond); err != nil {
+		if err := etcdWorker.Run(ctx, c.session, 200*time.Millisecond, c.info.AdvertiseAddr); err != nil {
 			// We check ttl of lease instead of check `session.Done`, because
 			// `session.Done` is only notified when etcd client establish a
 			// new keepalive request, there could be a time window as long as

diff --git a/cdc/capture/capture.go b/cdc/capture/capture.go
@@ -281,7 +281,8 @@ func (c *Capture) runEtcdWorker(ctx cdcContext.Context, reactor orchestrator.Rea
 	if err != nil {
 		return errors.Trace(err)
 	}
-	if err := etcdWorker.Run(ctx, c.session, timerInterval); err != nil {
+	captureAddr := c.info.AdvertiseAddr
+	if err := etcdWorker.Run(ctx, c.session, timerInterval, captureAddr); err != nil {
 		// We check ttl of lease instead of check `session.Done`, because
 		// `session.Done` is only notified when etcd client establish a
 		// new keepalive request, there could be a time window as long as

diff --git a/cdc/metrics.go b/cdc/metrics.go
@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/ticdc/cdc/puller/sorter"
 	"github.com/pingcap/ticdc/cdc/sink"
 	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/orchestrator"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -37,6 +38,7 @@ func init() {
 	sink.InitMetrics(registry)
 	entry.InitMetrics(registry)
 	sorter.InitMetrics(registry)
+	orchestrator.InitMetrics(registry)
 	if config.NewReplicaImpl {
 		processor.InitMetrics(registry)
 		tablepipeline.InitMetrics(registry)

diff --git a/cdc/task_test.go b/cdc/task_test.go
@@ -179,7 +179,9 @@ func (s *taskSuite) TestWatch(c *check.C) {
 	}
 
 	// Watch with a normal context
-	ch := s.w.Watch(context.Background())
+	ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+	ch := s.w.Watch(ctx)
 
 	// Trigger the ErrCompacted error
 	c.Assert(failpoint.Enable("github.com/pingcap/ticdc/cdc.restart_task_watch", "50%off"), check.IsNil)

diff --git a/errors.toml b/errors.toml
@@ -196,6 +196,16 @@ error = '''
 the etcd txn should be aborted and retried immediately
 '''
 
+["CDC:ErrEtcdTxnOpsExceed"]
+error = '''
+patch ops:%d of a single changefeed exceed etcd txn max ops:%d
+'''
+
+["CDC:ErrEtcdTxnSizeExceed"]
+error = '''
+patch size:%d of a single changefeed exceed etcd txn max size:%d
+'''
+
 ["CDC:ErrEventFeedAborted"]
 error = '''
 single event feed aborted

diff --git a/go.mod b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/Shopify/sarama v1.27.2
 	github.com/apache/pulsar-client-go v0.1.1
+	github.com/benbjohnson/clock v1.0.3
 	github.com/bradleyjkemp/grpc-tools v0.2.5
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e

diff --git a/go.sum b/go.sum
@@ -84,6 +84,8 @@ github.com/aws/aws-sdk-go v1.30.19/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZve
 github.com/aws/aws-sdk-go v1.35.3 h1:r0puXncSaAfRt7Btml2swUo74Kao+vKhO3VLjwDjK54=
 github.com/aws/aws-sdk-go v1.35.3/go.mod h1:H7NKnBqNVzoTJpGfLrQkkD+ytBA93eiDYi/+8rV9s48=
 github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6/go.mod h1:6YNgTHLutezwnBvyneBbwvB8C82y3dcoOj5EQJIdGXA=
+github.com/benbjohnson/clock v1.0.3 h1:vkLuvpK4fmtSCuo60+yC63p7y0BmQ8gm5ZXGuBCJyXg=
+github.com/benbjohnson/clock v1.0.3/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=