From 42558d229caea0399d6acd11b38330fc2af939a8 Mon Sep 17 00:00:00 2001 From: csuzhangxc Date: Fri, 27 Nov 2020 18:06:24 +0800 Subject: [PATCH 1/5] *: check binlog purge before send binlog --- drainer/pump.go | 22 +++++++++++++++++++++- pump/server.go | 5 ++++- pump/storage/storage.go | 29 +++++++++++++++++++---------- pump/storage/storage_test.go | 6 +++--- pump/storage/ts.go | 36 ++++++++++++++++++++++++++++++++++++ 5 files changed, 83 insertions(+), 15 deletions(-) create mode 100644 pump/storage/ts.go diff --git a/drainer/pump.go b/drainer/pump.go index c3c0b41e5..12f08d7ee 100644 --- a/drainer/pump.go +++ b/drainer/pump.go @@ -24,6 +24,7 @@ import ( "github.com/pingcap/log" "github.com/pingcap/tidb-binlog/pkg/util" "github.com/pingcap/tidb-binlog/pump" + "github.com/pingcap/tidb-binlog/pump/storage" "github.com/pingcap/tidb/store/tikv/oracle" pb "github.com/pingcap/tipb/go-binlog" "go.uber.org/zap" @@ -101,9 +102,11 @@ func (p *Pump) PullBinlog(pctx context.Context, last int64) chan MergeItem { labelReceive := "receive binlog" labelCreateConn := "create conn" labelPaused := "pump paused" + labelBinlogGCed := "binlog purged" pLog.Add(labelReceive, 10*time.Second) pLog.Add(labelCreateConn, 10*time.Second) pLog.Add(labelPaused, 30*time.Second) + pLog.Add(labelBinlogGCed, 30*time.Second) ret := make(chan MergeItem, binlogChanSize) @@ -119,6 +122,7 @@ func (p *Pump) PullBinlog(pctx context.Context, last int64) chan MergeItem { }() needReCreateConn := false + isBinlogPurged := false for { if atomic.LoadInt32(&p.isClosed) == 1 { return @@ -134,6 +138,16 @@ func (p *Pump) PullBinlog(pctx context.Context, last int64) chan MergeItem { continue } + if isBinlogPurged { + // some binlogs have been purged in pump, just print the log and wait to exit by user. + pLog.Print(labelBinlogGCed, func() { + p.logger.Error("some binlogs have been purged in pump") + }) + + time.Sleep(time.Second) + continue + } + if p.grpcConn == nil || needReCreateConn { p.logger.Info("pump create pull binlogs client") if err := p.createPullBinlogsClient(pctx, last); err != nil { @@ -153,7 +167,13 @@ func (p *Pump) PullBinlog(pctx context.Context, last int64) chan MergeItem { }) } - needReCreateConn = true + // Pump return binlog GCed error via gRPC response error. + if strings.Contains(err.Error(), storage.ErrRequestGCedBinlog.Error()) { + needReCreateConn = false // re-create connection will have no effect. + isBinlogPurged = true + } else { + needReCreateConn = true + } time.Sleep(time.Second) // TODO: add metric here diff --git a/pump/server.go b/pump/server.go index cb35bf89f..37a4a89c2 100644 --- a/pump/server.go +++ b/pump/server.go @@ -289,8 +289,11 @@ func (s *Server) PullBinlogs(in *binlog.PullBinlogReq, stream binlog.Pump_PullBi last := in.StartFrom.Offset gcTS := s.storage.GetGCTS() - if last <= gcTS { + if last != 0 && last <= gcTS { + // if requested with 0, then send binlog from oldest but not GCed TS. + // simple check TS before read binlog, but more checks are still needed in storage. log.Error("drainer request a purged binlog TS, some binlog events may be loss", zap.Int64("gc TS", gcTS), zap.Reflect("request", in)) + return errors.Annotatef(storage.ErrRequestGCedBinlog, "requested TS %d, GC TS %d", last, gcTS) } ctx, cancel := context.WithCancel(context.Background()) diff --git a/pump/storage/storage.go b/pump/storage/storage.go index 6053e2b5a..c053e8a40 100644 --- a/pump/storage/storage.go +++ b/pump/storage/storage.go @@ -50,6 +50,9 @@ const ( ) var ( + // ErrRequestGCedBinlog indicates a Drainer is requesting some purged binlogs. + ErrRequestGCedBinlog = errors.New("request a purged binlog") + // save gcTS, the max TS we have gc, for binlog not greater than gcTS, we can delete it from storage gcTSKey = []byte("!binlog!gcts") // save maxCommitTS, we can get binlog in range [gcTS, maxCommitTS] from PullCommitBinlog @@ -104,7 +107,7 @@ type Append struct { latestTS int64 gcWorking int32 - gcTS int64 + gcTS GCTS maxCommitTS int64 headPointer valuePointer handlePointer valuePointer @@ -166,11 +169,12 @@ func NewAppendWithResolver(dir string, options *Options, tiStore kv.Storage, tiL sortItems: make(chan sortItem, 1024), } - append.gcTS, err = append.readGCTSFromDB() + gcTS, err := append.readGCTSFromDB() if err != nil { return nil, errors.Trace(err) } - gcTSGauge.Set(float64(oracle.ExtractPhysical(uint64(append.gcTS)))) + append.gcTS.Store(gcTS) + gcTSGauge.Set(float64(oracle.ExtractPhysical(uint64(gcTS)))) append.maxCommitTS, err = append.readInt64(maxCommitTSKey) if err != nil { @@ -214,7 +218,7 @@ func NewAppendWithResolver(dir string, options *Options, tiStore kv.Storage, tiL minPointer = append.handlePointer } - log.Info("Append info", zap.Int64("gcTS", append.gcTS), + log.Info("Append info", zap.Int64("gcTS", gcTS), zap.Int64("maxCommitTS", append.maxCommitTS), zap.Reflect("headPointer", append.headPointer), zap.Reflect("handlePointer", append.handlePointer)) @@ -646,12 +650,12 @@ func (a *Append) Close() error { // GetGCTS implement Storage.GetGCTS func (a *Append) GetGCTS() int64 { - return atomic.LoadInt64(&a.gcTS) + return a.gcTS.Load() } // GC implement Storage.GC func (a *Append) GC(ts int64) { - lastTS := atomic.LoadInt64(&a.gcTS) + lastTS := a.gcTS.Load() if ts <= lastTS { log.Info("ignore gc request", zap.Int64("ts", ts), zap.Int64("lastTS", lastTS)) return @@ -665,7 +669,7 @@ func (a *Append) GC(ts int64) { return } - atomic.StoreInt64(&a.gcTS, ts) + a.gcTS.Store(ts) // once `Store` returned, no guarantee for metadata or vlog. if err := a.saveGCTSToDB(ts); err != nil { log.Error("Failed to save GCTS", zap.Int64("ts", ts), zap.Error(err)) } @@ -1113,10 +1117,15 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte } }() - gcTS := atomic.LoadInt64(&a.gcTS) + gcTS := a.gcTS.Load() if last < gcTS { - log.Warn("last ts less than gcTS", zap.Int64("last ts", last), zap.Int64("gcTS", gcTS)) - last = gcTS + if last == 0 { + log.Warn("last TS is 0, will send binlog from gcTS", zap.Int64("gcTS", gcTS)) + last = gcTS + } else { + log.Error("last TS less than gcTS", zap.Int64("lastTS", last), zap.Int64("gcTS", gcTS)) + // TODO: report error + } } values := make(chan []byte, 5) diff --git a/pump/storage/storage_test.go b/pump/storage/storage_test.go index 9bbc96a67..6566a709d 100644 --- a/pump/storage/storage_test.go +++ b/pump/storage/storage_test.go @@ -200,7 +200,7 @@ func (as *AppendSuit) TestCloseAndOpenAgain(c *check.C) { populateBinlog(c, append, 128, 1) time.Sleep(time.Millisecond * 100) - gcTS := append.gcTS + gcTS := append.gcTS.Load() maxCommitTS := append.maxCommitTS headPointer := append.headPointer handlePointer := append.handlePointer @@ -213,7 +213,7 @@ func (as *AppendSuit) TestCloseAndOpenAgain(c *check.C) { append, err = NewAppend(append.dir, append.options) c.Assert(err, check.IsNil) - c.Assert(gcTS, check.Equals, append.gcTS) + c.Assert(gcTS, check.Equals, append.gcTS.Load()) c.Assert(maxCommitTS, check.Equals, append.maxCommitTS) c.Assert(headPointer, check.Equals, append.headPointer) c.Assert(handlePointer, check.Equals, append.handlePointer) @@ -350,7 +350,7 @@ func (as *AppendSuit) TestReadWriteGCTS(c *check.C) { append, err = NewAppend(append.dir, append.options) c.Assert(err, check.IsNil) - c.Assert(append.gcTS, check.Equals, int64(100)) + c.Assert(append.gcTS.Load(), check.Equals, int64(100)) append.Close() } diff --git a/pump/storage/ts.go b/pump/storage/ts.go new file mode 100644 index 000000000..40d0b8632 --- /dev/null +++ b/pump/storage/ts.go @@ -0,0 +1,36 @@ +// Copyright 2019 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 storage + +import "sync" + +// GCTS wraps a int64 TS value with a mutex to protect/detect binlog purging when fetching binlogs. +type GCTS struct { + mux sync.RWMutex + ts int64 +} + +// Store stores or updates the current TS value. +func (g *GCTS) Store(ts int64) { + g.mux.Lock() + defer g.mux.Unlock() + g.ts = ts +} + +// Load loads or reads the current TS value. +func (g *GCTS) Load() int64 { + g.mux.RLock() + defer g.mux.RUnlock() + return g.ts +} From b92f923f44473bdcac268ff37096ac0723ac3a4d Mon Sep 17 00:00:00 2001 From: csuzhangxc Date: Mon, 30 Nov 2020 11:16:11 +0800 Subject: [PATCH 2/5] *: check whether binlog GCed in every read round --- pump/server.go | 5 +++- pump/server_test.go | 13 ++++----- pump/storage/bench_test.go | 5 +++- pump/storage/storage.go | 51 ++++++++++++++++++++++++++++-------- pump/storage/storage_test.go | 3 ++- pump/storage/ts.go | 11 ++++++++ 6 files changed, 68 insertions(+), 20 deletions(-) diff --git a/pump/server.go b/pump/server.go index 37a4a89c2..0df56bbef 100644 --- a/pump/server.go +++ b/pump/server.go @@ -298,12 +298,15 @@ func (s *Server) PullBinlogs(in *binlog.PullBinlogReq, stream binlog.Pump_PullBi ctx, cancel := context.WithCancel(context.Background()) defer cancel() - binlogs := s.storage.PullCommitBinlog(ctx, last) + binlogs, errs := s.storage.PullCommitBinlog(ctx, last) for { select { case <-s.pullClose: return nil + case err2 := <-errs: + log.Error("pull binlog failed", zap.Error(err2)) + return err2 case data, ok := <-binlogs: if !ok { return nil diff --git a/pump/server_test.go b/pump/server_test.go index 646103ada..e28afac7b 100644 --- a/pump/server_test.go +++ b/pump/server_test.go @@ -136,22 +136,23 @@ func (s *noOpStorage) GetGCTS() int64 { return 0 } func (s *noOpStorage) GC(ts int64) {} func (s *noOpStorage) MaxCommitTS() int64 { return 0 } func (s *noOpStorage) GetBinlog(ts int64) (*binlog.Binlog, error) { return nil, nil } -func (s *noOpStorage) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte { - return make(chan []byte) +func (s *noOpStorage) PullCommitBinlog(ctx context.Context, last int64) (<-chan []byte, <-chan error) { + return make(chan []byte), make(chan error) } func (s *noOpStorage) Close() error { return nil } type fakePullable struct{ noOpStorage } -func (s *fakePullable) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte { +func (s *fakePullable) PullCommitBinlog(ctx context.Context, last int64) (<-chan []byte, <-chan error) { chl := make(chan []byte) + errs := make(chan error) go func() { for i := 0; i < 3; i++ { chl <- []byte(fmt.Sprintf("payload_%d", i)) } close(chl) }() - return chl + return chl, errs } func (s *pullBinlogsSuite) TestPullBinlogFromStorage(c *C) { @@ -664,8 +665,8 @@ func (s *startStorage) MaxCommitTS() int64 { return 0 } func (s *startStorage) GetBinlog(ts int64) (*binlog.Binlog, error) { return nil, errors.New("server_test") } -func (s *startStorage) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte { - return make(chan []byte) +func (s *startStorage) PullCommitBinlog(ctx context.Context, last int64) (<-chan []byte, <-chan error) { + return make(chan []byte), make(<-chan error) } func (s *startStorage) Close() error { <-s.sig diff --git a/pump/storage/bench_test.go b/pump/storage/bench_test.go index 2e5b344a3..0ddbc9c4a 100644 --- a/pump/storage/bench_test.go +++ b/pump/storage/bench_test.go @@ -80,7 +80,7 @@ func benchmarkPull(b *testing.B, prewriteValueSize int, binlogNum int) { runtime.GC() b.ResetTimer() - pulller := append.PullCommitBinlog(context.Background(), 0) + pulller, errs := append.PullCommitBinlog(context.Background(), 0) cnt := 0 for b := range pulller { @@ -101,6 +101,9 @@ func benchmarkPull(b *testing.B, prewriteValueSize int, binlogNum int) { // just count the prewriteValueSize b.SetBytes(int64(prewriteValueSize)) b.ReportAllocs() + if len(errs) > 0 { + b.Fatalf("pull binlog got some errors") + } } func benchmarkWrite(b *testing.B, prewriteValueSize int, parallelism int, sync bool) { diff --git a/pump/storage/storage.go b/pump/storage/storage.go index c053e8a40..9ba0c21ae 100644 --- a/pump/storage/storage.go +++ b/pump/storage/storage.go @@ -43,6 +43,7 @@ import ( const ( maxTxnTimeoutSecond int64 = 600 + gcMaxBlockTime = 30 * time.Minute // we run GC at every 1 hour, but may block GC when reading and sending binlog at most in this duration. chanCapacity = 1 << 20 // if pump takes a long time to write binlog, pump will display the binlog meta information (unit: Second) slowWriteThreshold = 1.0 @@ -86,7 +87,7 @@ type Storage interface { GetBinlog(ts int64) (binlog *pb.Binlog, err error) // PullCommitBinlog return the chan to consume the binlog - PullCommitBinlog(ctx context.Context, last int64) <-chan []byte + PullCommitBinlog(ctx context.Context, last int64) (<-chan []byte, <-chan error) Close() error } @@ -1105,7 +1106,7 @@ func (a *Append) feedPreWriteValue(cbinlog *pb.Binlog) error { } // PullCommitBinlog return commit binlog > last -func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte { +func (a *Append) PullCommitBinlog(ctx context.Context, last int64) (<-chan []byte, <-chan error) { log.Debug("new PullCommitBinlog", zap.Int64("last ts", last)) ctx, cancel := context.WithCancel(ctx) @@ -1117,19 +1118,21 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte } }() + values := make(chan []byte, 5) + errs := make(chan error, 5) // we `return` after sending an error now, so it should never block on this chan. + gcTS := a.gcTS.Load() if last < gcTS { if last == 0 { log.Warn("last TS is 0, will send binlog from gcTS", zap.Int64("gcTS", gcTS)) last = gcTS } else { - log.Error("last TS less than gcTS", zap.Int64("lastTS", last), zap.Int64("gcTS", gcTS)) - // TODO: report error + log.Error("last TS less than gcTS, some binlog events may be loss", zap.Int64("lastTS", last), zap.Int64("gcTS", gcTS)) + errs <- errors.Annotatef(ErrRequestGCedBinlog, "requested TS %d, GC TS %d", last, gcTS) + return values, errs } } - values := make(chan []byte, 5) - irange := &util.Range{ Start: encodeTSKey(0), Limit: encodeTSKey(math.MaxInt64), @@ -1142,6 +1145,7 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte go func() { defer close(values) + outerForLoop: for { startTS := last + 1 limitTS := atomic.LoadInt64(&a.maxCommitTS) + 1 @@ -1154,6 +1158,16 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte continue } + // acquire the lock to block GC. + // NOTE: do not forget to release the lock carefully. + gcTS = a.gcTS.LoadAndLock() + if last < gcTS { + a.gcTS.ReleaseLoadLock() + log.Error("last TS less than gcTS, some binlog events may be loss", zap.Int64("lastTS", last), zap.Int64("gcTS", gcTS)) + errs <- errors.Annotatef(ErrRequestGCedBinlog, "requested TS %d, GC TS %d", last, gcTS) + return + } + irange.Start = encodeTSKey(startTS) irange.Limit = encodeTSKey(limitTS) iter := a.metadata.NewIterator(irange, nil) @@ -1172,17 +1186,21 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte value, err := a.vlog.readValue(vp) if err != nil { - log.Error("read value failed", zap.Error(err)) iter.Release() errorCount.WithLabelValues("read_value").Add(1.0) + a.gcTS.ReleaseLoadLock() + log.Error("read value failed", zap.Int64("TS", decodeTSKey(iter.Key())), zap.Error(err)) + errs <- errors.Errorf("read value failed, TS %d", decodeTSKey(iter.Key())) return } binlog := new(pb.Binlog) err = binlog.Unmarshal(value) if err != nil { - log.Error("Unmarshal Binlog failed", zap.Error(err)) iter.Release() + a.gcTS.ReleaseLoadLock() + log.Error("Unmarshal Binlog failed", zap.Int64("TS", decodeTSKey(iter.Key())), zap.Error(err)) + errs <- errors.Errorf("Unmarshal Binlog failed, TS %d", decodeTSKey(iter.Key())) return } @@ -1206,30 +1224,41 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte } errorCount.WithLabelValues("feed_pre_write_value").Add(1.0) - log.Error("feed pre write value failed", zap.Error(err)) iter.Release() + a.gcTS.ReleaseLoadLock() + log.Error("feed pre write value failed", zap.Int64("TS", decodeTSKey(iter.Key())), zap.Error(err)) + errs <- errors.Errorf("feed pre write value failed, TS %d", decodeTSKey(iter.Key())) return } } value, err = binlog.Marshal() if err != nil { - log.Error("marshal failed", zap.Error(err)) iter.Release() + a.gcTS.ReleaseLoadLock() + log.Error("marshal failed", zap.Int64("TS", decodeTSKey(iter.Key())), zap.Error(err)) + errs <- errors.Errorf("marshal failed, TS %d", decodeTSKey(iter.Key())) return } select { case values <- value: log.Debug("send value success") + case <-time.After(gcMaxBlockTime): + iter.Release() + a.gcTS.ReleaseLoadLock() + log.Warn("can not send the binlog for a long time, will try to read again", zap.Duration("duration", gcMaxBlockTime), zap.Int64("current TS", decodeTSKey(iter.Key()))) + break outerForLoop case <-ctx.Done(): iter.Release() + a.gcTS.ReleaseLoadLock() return } last = decodeTSKey(iter.Key()) } iter.Release() + a.gcTS.ReleaseLoadLock() err := iter.Error() if err != nil { log.Error("encounter iterator error", zap.Error(err)) @@ -1244,7 +1273,7 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) <-chan []byte } }() - return values + return values, errs } type storageSize struct { diff --git a/pump/storage/storage_test.go b/pump/storage/storage_test.go index 6566a709d..ed3138751 100644 --- a/pump/storage/storage_test.go +++ b/pump/storage/storage_test.go @@ -241,7 +241,7 @@ func (as *AppendSuit) testWriteBinlogAndPullBack(c *check.C, prewriteValueSize i } ctx, cancel := context.WithCancel(context.Background()) - values := appendStorage.PullCommitBinlog(ctx, 0) + values, errs := appendStorage.PullCommitBinlog(ctx, 0) // pull the binlogs back and check sorted var binlogs []*pb.Binlog @@ -265,6 +265,7 @@ func (as *AppendSuit) testWriteBinlogAndPullBack(c *check.C, prewriteValueSize i for i := 1; i < len(binlogs); i++ { c.Assert(binlogs[i].CommitTs, check.Greater, binlogs[i-1].CommitTs) } + c.Assert(errs, check.HasLen, 0) cancel() } diff --git a/pump/storage/ts.go b/pump/storage/ts.go index 40d0b8632..deadb8cfc 100644 --- a/pump/storage/ts.go +++ b/pump/storage/ts.go @@ -34,3 +34,14 @@ func (g *GCTS) Load() int64 { defer g.mux.RUnlock() return g.ts } + +// LoadAndLock loads the current TS value and hold the read-lock. +func (g *GCTS) LoadAndLock() int64 { + g.mux.RLock() + return g.ts +} + +// ReleaseLoadLock releases the read-lock acquired by LoadAndLock. +func (g *GCTS) ReleaseLoadLock() { + g.mux.RUnlock() +} From 2a82ef8d9e7868b26dc3a00679b44c83e0b9de81 Mon Sep 17 00:00:00 2001 From: csuzhangxc Date: Mon, 30 Nov 2020 11:31:41 +0800 Subject: [PATCH 3/5] *: fix break loop --- pump/storage/storage.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/pump/storage/storage.go b/pump/storage/storage.go index 9ba0c21ae..94a74bf75 100644 --- a/pump/storage/storage.go +++ b/pump/storage/storage.go @@ -1145,7 +1145,6 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) (<-chan []byt go func() { defer close(values) - outerForLoop: for { startTS := last + 1 limitTS := atomic.LoadInt64(&a.maxCommitTS) + 1 @@ -1174,6 +1173,7 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) (<-chan []byt // log.Debugf("try to get range [%d,%d)", startTS, atomic.LoadInt64(&a.maxCommitTS)+1) + readForLoop: for ok := iter.Seek(encodeTSKey(startTS)); ok; ok = iter.Next() { var vp valuePointer err := vp.UnmarshalBinary(iter.Value()) @@ -1245,10 +1245,9 @@ func (a *Append) PullCommitBinlog(ctx context.Context, last int64) (<-chan []byt case values <- value: log.Debug("send value success") case <-time.After(gcMaxBlockTime): - iter.Release() - a.gcTS.ReleaseLoadLock() + // do not update `last` anymore. log.Warn("can not send the binlog for a long time, will try to read again", zap.Duration("duration", gcMaxBlockTime), zap.Int64("current TS", decodeTSKey(iter.Key()))) - break outerForLoop + break readForLoop case <-ctx.Done(): iter.Release() a.gcTS.ReleaseLoadLock() From 1e59d8e397a166311f81d9440bfaf508a08c66b7 Mon Sep 17 00:00:00 2001 From: csuzhangxc Date: Mon, 30 Nov 2020 11:35:46 +0800 Subject: [PATCH 4/5] *: fix log for alert GC --- pump/server.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pump/server.go b/pump/server.go index 0df56bbef..5046f0d2c 100644 --- a/pump/server.go +++ b/pump/server.go @@ -642,12 +642,14 @@ func (s *Server) detectDrainerCheckPoints(ctx context.Context, gcTS int64) { } if drainer.MaxCommitTS < gcTS { - log.Error("drainer's checkpoint is older than pump gc ts, some binlogs are purged", + log.Error("drainer's checkpoint is older than pump alert gc ts, some binlogs may be purged after alert time", zap.String("drainer", drainer.NodeID), - zap.Int64("gc ts", gcTS), + zap.Int64("alert gc ts", gcTS), zap.Int64("drainer checkpoint", drainer.MaxCommitTS), + zap.Duration("alert time", earlyAlertGC) ) // will add test when binlog have failpoint + // NOTE: this metrics do not mean the binlogs have been purge, but mean some binlogs will be purge after alert time. detectedDrainerBinlogPurged.WithLabelValues(drainer.NodeID).Inc() } } From c2d13f918ea5b0edf12c34c46a7830a92d4095ee Mon Sep 17 00:00:00 2001 From: csuzhangxc Date: Mon, 30 Nov 2020 11:37:40 +0800 Subject: [PATCH 5/5] *: fix syntax --- pump/server.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pump/server.go b/pump/server.go index 5046f0d2c..6450dccf8 100644 --- a/pump/server.go +++ b/pump/server.go @@ -646,7 +646,7 @@ func (s *Server) detectDrainerCheckPoints(ctx context.Context, gcTS int64) { zap.String("drainer", drainer.NodeID), zap.Int64("alert gc ts", gcTS), zap.Int64("drainer checkpoint", drainer.MaxCommitTS), - zap.Duration("alert time", earlyAlertGC) + zap.Duration("alert time", earlyAlertGC), ) // will add test when binlog have failpoint // NOTE: this metrics do not mean the binlogs have been purge, but mean some binlogs will be purge after alert time.