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

cdc/capture/capture.go

@@ -283,7 +283,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

cdc/metrics.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sorter/unified"
 	"github.com/pingcap/tiflow/pkg/actor"
 	"github.com/pingcap/tiflow/pkg/etcd"
+	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -44,6 +45,7 @@ func init() {
 	etcd.InitMetrics(registry)
 	initServerMetrics(registry)
 	actor.InitMetrics(registry)
+	orchestrator.InitMetrics(registry)
 	// Sorter metrics
 	memory.InitMetrics(registry)
 	unified.InitMetrics(registry)

errors.toml

@@ -241,6 +241,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

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751
 	github.com/apache/pulsar-client-go v0.6.0
 	github.com/aws/aws-sdk-go v1.35.3
+	github.com/benbjohnson/clock v1.1.0
 	github.com/bradleyjkemp/grpc-tools v0.2.5
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/chaos-mesh/go-sqlsmith v0.0.0-20211025024535-03ae33408684