Repository: pingcap/tidb-binlog
Pull request #1182 (2022-07-29 09:12:44)

reparo: support pprof and metrics
<!--
Thank you for contributing to TiDB! Please read TiDB's [CONTRIBUTING](https://github.com/pingcap/tidb/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: close #xxx

### What is changed and how it works?
Support pprof and http metrics serving for reparo.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Manual test
 
Code changes

- None

Side effects

- None

Related changes

- None

### Release note

<!-- bugfix or new feature needs a release note -->

- No release note


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tidb-binlog/pulls/1182/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tidb-binlog/pulls/1182/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found [here](https://prow.tidb.io/command-help?repo=pingcap%2Ftidb-binlog).

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
ti-chi-bot: @lichunzhu: This cherry pick PR is for a release branch and has not yet been approved by release team.
Adding the `do-not-merge/cherry-pick-not-approved` label.

To merge this cherry pick, it must first be approved by the collaborators.

**AFTER** it has been approved by collaborators, please ping the release team in a comment to request a cherry pick review.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

Patch:
diff --git a/reparo/config.go b/reparo/config.go
index 32a78a41..f3f57c6e 100644
--- a/reparo/config.go
+++ b/reparo/config.go
@@ -46,6 +46,7 @@ type Config struct {
 	StopTSO       int64  `toml:"stop-tso" json:"stop-tso"`
 	TxnBatch      int    `toml:"txn-batch" json:"txn-batch"`
 	WorkerCount   int    `toml:"worker-count" json:"worker-count"`
+	StatusAddr    string `toml:"status-addr" json:"status-addr"`
 
 	DestType string           `toml:"dest-type" json:"dest-type"`
 	DestDB   *syncer.DBConfig `toml:"dest-db" json:"dest-db"`
@@ -77,6 +78,7 @@ func NewConfig() *Config {
 	fs.StringVar(&c.Dir, "data-dir", "", "drainer data directory path")
 	fs.StringVar(&c.StartDatetime, "start-datetime", "", "recovery from start-datetime, empty string means starting from the beginning of the first file")
 	fs.StringVar(&c.StopDatetime, "stop-datetime", "", "recovery end in stop-datetime, empty string means never end.")
+	fs.StringVar(&c.StatusAddr, "status-addr", ":8381", "reparo API server and pprof addr")
 	fs.Int64Var(&c.StartTSO, "start-tso", 0, "similar to start-datetime but in pd-server tso format")
 	fs.Int64Var(&c.StopTSO, "stop-tso", 0, "similar to stop-datetime, but in pd-server tso format")
 	fs.IntVar(&c.TxnBatch, "txn-batch", 20, "number of binlog events in a transaction batch")
diff --git a/reparo/decode.go b/reparo/decode.go
index 9763bd4d..3fcb2b55 100644
--- a/reparo/decode.go
+++ b/reparo/decode.go
@@ -28,6 +28,7 @@ func Decode(r io.Reader) (*pb.Binlog, int64, error) {
 	if err != nil {
 		return nil, 0, errors.Trace(err)
 	}
+	readBinlogSizeHistogram.Observe(float64(length))
 
 	binlog := &pb.Binlog{}
 	err = binlog.Unmarshal(payload)
diff --git a/reparo/http_handler.go b/reparo/http_handler.go
new file mode 100644
index 00000000..9c435ea0
--- /dev/null
+++ b/reparo/http_handler.go
@@ -0,0 +1,69 @@
+// Copyright 2020 PingCAP, Inc. Licensed under Apache-2.0.
+
+package reparo
+
+import (
+	"net"
+	"net/http"
+	"net/http/pprof"
+	"strings"
+	"time"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/soheilhy/cmux"
+	"go.uber.org/zap"
+)
+
+var cmuxReadTimeout = 10 * time.Second
+
+func startHTTPServer(lis net.Listener) {
+	router := http.NewServeMux()
+	router.Handle("/metrics", promhttp.Handler())
+
+	router.HandleFunc("/debug/pprof/", pprof.Index)
+	router.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+	router.HandleFunc("/debug/pprof/profile", pprof.Profile)
+	router.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+	router.HandleFunc("/debug/pprof/trace", pprof.Trace)
+
+	httpServer := &http.Server{
+		Handler: router,
+	}
+	err := httpServer.Serve(lis)
+	err = errors.Cause(err)
+	if err != nil && !isErrNetClosing(err) && err != http.ErrServerClosed {
+		log.Info("reparo http handler return with error", zap.String("error", err.Error()))
+	}
+}
+
+func startReparoService(addr string) error {
+	rootLis, err := net.Listen("tcp", addr)
+	if err != nil {
+		return errors.Annotate(err, "start listening")
+	}
+
+	// create a cmux
+	m := cmux.New(rootLis)
+	m.SetReadTimeout(cmuxReadTimeout) // set a timeout, ref: https://github.com/pingcap/tidb-binlog/pull/352
+
+	httpL := m.Match(cmux.HTTP1Fast())
+	go startHTTPServer(httpL)
+
+	err = m.Serve() // start serving, block
+	if err != nil && isErrNetClosing(err) {
+		err = nil
+	}
+	return err
+}
+
+var useOfClosedErrMsg = "use of closed network connection"
+
+// isErrNetClosing checks whether is an ErrNetClosing error
+func isErrNetClosing(err error) bool {
+	if err == nil {
+		return false
+	}
+	return strings.Contains(err.Error(), useOfClosedErrMsg)
+}
diff --git a/reparo/metrics.go b/reparo/metrics.go
new file mode 100644
index 00000000..08abbbe6
--- /dev/null
+++ b/reparo/metrics.go
@@ -0,0 +1,73 @@
+package reparo
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	eventCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "binlog",
+			Subsystem: "drainer",
+			Name:      "event",
+			Help:      "the count of sql event(dml, ddl).",
+		}, []string{"type"})
+
+	checkpointTSOGauge = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: "binlog",
+			Subsystem: "drainer",
+			Name:      "checkpoint_tso",
+			Help:      "save checkpoint tso of drainer.",
+		})
+	executeHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "binlog",
+			Subsystem: "drainer",
+			Name:      "execute_duration_time",
+			Help:      "Bucketed histogram of processing time (s) of a execute to sync data to downstream.",
+			Buckets:   prometheus.ExponentialBuckets(0.00005, 2, 18),
+		})
+
+	queryHistogramVec = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "binlog",
+			Subsystem: "drainer",
+			Name:      "query_duration_time",
+			Help:      "Bucketed histogram of processing time (s) of a query to sync data to downstream.",
+			Buckets:   prometheus.ExponentialBuckets(0.00005, 2, 18),
+		}, []string{"type"})
+
+	readBinlogSizeHistogram = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "binlog",
+			Subsystem: "drainer",
+			Name:      "read_binlog_size",
+			Help:      "Bucketed histogram of size of a binlog.",
+			Buckets:   prometheus.ExponentialBuckets(16, 2, 25),
+		})
+
+	queueSizeGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "binlog",
+			Subsystem: "drainer",
+			Name:      "queue_size",
+			Help:      "the size of queue",
+		}, []string{"name"})
+)
+
+func init() {
+	registry := prometheus.DefaultRegisterer
+	registry.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
+	registry.MustRegister(prometheus.NewGoCollector())
+	registry.MustRegister(checkpointTSOGauge)
+	registry.MustRegister(eventCounter)
+	registry.MustRegister(executeHistogram)
+	registry.MustRegister(readBinlogSizeHistogram)
+	registry.MustRegister(queryHistogramVec)
+	registry.MustRegister(queueSizeGauge)
+
+	if gatherer, ok := registry.(prometheus.Gatherer); ok {
+		prometheus.DefaultGatherer = gatherer
+	}
+}
diff --git a/reparo/reparo.go b/reparo/reparo.go
index 5600b166..fbe0fba3 100644
--- a/reparo/reparo.go
+++ b/reparo/reparo.go
@@ -15,10 +15,12 @@ package reparo
 
 import (
 	"io"
+	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb-binlog/pkg/filter"
+	"github.com/pingcap/tidb-binlog/pkg/loader"
 	pb "github.com/pingcap/tidb-binlog/proto/binlog"
 	"github.com/pingcap/tidb-binlog/reparo/syncer"
 	"github.com/pingcap/tidb/store/tikv/oracle"
@@ -37,7 +39,11 @@ type Reparo struct {
 func New(cfg *Config) (*Reparo, error) {
 	log.Info("New Reparo", zap.Stringer("config", cfg))
 
-	syncer, err := syncer.New(cfg.DestType, cfg.DestDB, cfg.WorkerCount, cfg.TxnBatch, cfg.SafeMode)
+	syncer, err := syncer.New(cfg.DestType, cfg.DestDB, cfg.WorkerCount, cfg.TxnBatch, cfg.SafeMode, &loader.MetricsGroup{
+		EventCounterVec:   eventCounter,
+		QueryHistogramVec: queryHistogramVec,
+		QueueSizeGauge:    queueSizeGauge,
+	})
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -53,6 +59,14 @@ func New(cfg *Config) (*Reparo, error) {
 
 // Process runs the main procedure.
 func (r *Reparo) Process() error {
+	if r.cfg.StatusAddr != "" {
+		go func() {
+			err := startReparoService(r.cfg.StatusAddr)
+			if err != nil {
+				log.Info("meet error when stopping reparo http service", zap.String("error", err.Error()))
+			}
+		}()
+	}
 	pbReader, err := newDirPbReader(r.cfg.Dir, r.cfg.StartTSO, r.cfg.StopTSO)
 	if err != nil {
 		return errors.Annotatef(err, "new reader failed dir: %s", r.cfg.Dir)
@@ -77,10 +91,12 @@ func (r *Reparo) Process() error {
 		if ignore {
 			continue
 		}
-
+		beginTime := time.Now()
 		err = r.syncer.Sync(binlog, func(binlog *pb.Binlog) {
 			dt := oracle.GetTimeFromTS(uint64(binlog.CommitTs))
 			log.Info("sync binlog success", zap.Int64("ts", binlog.CommitTs), zap.Time("datetime", dt))
+			checkpointTSOGauge.Set(float64(oracle.ExtractPhysical(uint64(binlog.CommitTs))))
+			executeHistogram.Observe(time.Since(beginTime).Seconds())
 		})
 
 		if err != nil {
diff --git a/reparo/syncer/mysql.go b/reparo/syncer/mysql.go
index f318eef9..c2c04866 100644
--- a/reparo/syncer/mysql.go
+++ b/reparo/syncer/mysql.go
@@ -49,17 +49,21 @@ var (
 // should be only used for unit test to create mock db
 var createDB = loader.CreateDB
 
-func newMysqlSyncer(cfg *DBConfig, worker int, batchSize int, safemode bool) (*mysqlSyncer, error) {
+func newMysqlSyncer(cfg *DBConfig, worker int, batchSize int, safemode bool, metricsGroup *loader.MetricsGroup) (*mysqlSyncer, error) {
 	db, err := createDB(cfg.User, cfg.Password, cfg.Host, cfg.Port, nil)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
-	return newMysqlSyncerFromSQLDB(db, worker, batchSize, safemode)
+	return newMysqlSyncerFromSQLDB(db, worker, batchSize, safemode, metricsGroup)
 }
 
-func newMysqlSyncerFromSQLDB(db *sql.DB, worker int, batchSize int, safemode bool) (*mysqlSyncer, error) {
-	loader, err := loader.NewLoader(db, loader.WorkerCount(worker), loader.BatchSize(batchSize))
+func newMysqlSyncerFromSQLDB(db *sql.DB, worker int, batchSize int, safemode bool, metricsGroup *loader.MetricsGroup) (*mysqlSyncer, error) {
+	opts := []loader.Option{loader.WorkerCount(worker), loader.BatchSize(batchSize)}
+	if metricsGroup != nil {
+		opts = append(opts, loader.Metrics(metricsGroup))
+	}
+	loader, err := loader.NewLoader(db, opts...)
 	if err != nil {
 		return nil, errors.Annotate(err, "new loader failed")
 	}
diff --git a/reparo/syncer/syncer.go b/reparo/syncer/syncer.go
index 8e1a1810..e679c1ce 100644
--- a/reparo/syncer/syncer.go
+++ b/reparo/syncer/syncer.go
@@ -16,6 +16,7 @@ package syncer
 import (
 	"fmt"
 
+	"github.com/pingcap/tidb-binlog/pkg/loader"
 	pb "github.com/pingcap/tidb-binlog/proto/binlog"
 )
 
@@ -29,10 +30,10 @@ type Syncer interface {
 }
 
 // New creates a new executor based on the name.
-func New(name string, cfg *DBConfig, worker int, batchSize int, safemode bool) (Syncer, error) {
+func New(name string, cfg *DBConfig, worker int, batchSize int, safemode bool, metricsGroup *loader.MetricsGroup) (Syncer, error) {
 	switch name {
 	case "mysql":
-		return newMysqlSyncer(cfg, worker, batchSize, safemode)
+		return newMysqlSyncer(cfg, worker, batchSize, safemode, metricsGroup)
 	case "print":
 		return newPrintSyncer()
 	case "memory":


Test patch:
diff --git a/reparo/syncer/mysql_test.go b/reparo/syncer/mysql_test.go
index 3c93d87e..a49641a0 100644
--- a/reparo/syncer/mysql_test.go
+++ b/reparo/syncer/mysql_test.go
@@ -5,7 +5,7 @@ import (
 	"database/sql"
 	"time"
 
-	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/pingcap/check"
 	pb "github.com/pingcap/tidb-binlog/proto/binlog"
 )
@@ -33,7 +33,7 @@ func (s *testMysqlSuite) testMysqlSyncer(c *check.C, safemode bool) {
 		createDB = oldCreateDB
 	}()
 
-	syncer, err := newMysqlSyncer(&DBConfig{}, 1, 20, safemode)
+	syncer, err := newMysqlSyncer(&DBConfig{}, 1, 20, safemode, nil)
 	c.Assert(err, check.IsNil)
 
 	mock.ExpectBegin()
diff --git a/reparo/syncer/syncer_test.go b/reparo/syncer/syncer_test.go
index 4ebd8c83..395a30c1 100644
--- a/reparo/syncer/syncer_test.go
+++ b/reparo/syncer/syncer_test.go
@@ -34,7 +34,7 @@ func (s *testSyncerSuite) TestNewSyncer(c *check.C) {
 	}
 
 	for _, testCase := range testCases {
-		syncer, err := New(testCase.typeStr, cfg, 16, 20, false)
+		syncer, err := New(testCase.typeStr, cfg, 16, 20, false, nil)
 		c.Assert(err, check.IsNil)
 		c.Assert(reflect.TypeOf(syncer), testCase.checker, testCase.tp)
 	}