Reset the current lag when closing the replication lag reader. (#12683)…

… (#12744)

* Reset the current lag when closing the replication lag reader.



* Fix tests.



---------

Signed-off-by: Johan Stenberg <[email protected]>
Signed-off-by: Florent Poinsard <[email protected]>
Co-authored-by: Johan Stenberg <[email protected]>

go/vt/vttablet/tabletserver/repltracker/reader.go

@@ -122,6 +122,9 @@ func (r *heartbeatReader) Close() {
 	}
 	r.ticks.Stop()
 	r.pool.Close()
+
+	currentLagNs.Set(0)
+
 	r.isOpen = false
 	log.Info("Heartbeat Reader: closed")
 }

go/vt/vttablet/tabletserver/repltracker/reader_test.go

@@ -39,9 +39,13 @@ import (
 func TestReaderReadHeartbeat(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
-	tr := newReader(db, mockNowFunc)
+
+	now := time.Now()
+	tr := newReader(db, &now)
 	defer tr.Close()
 
+	tr.pool.Open(tr.env.Config().DB.AppWithDB(), tr.env.Config().DB.DbaWithDB(), tr.env.Config().DB.AppDebugWithDB())
+
 	db.AddQuery(fmt.Sprintf("SELECT ts FROM %s.heartbeat WHERE keyspaceShard='%s'", "_vt", tr.keyspaceShard), &sqltypes.Result{
 		Fields: []*querypb.Field{
 			{Name: "ts", Type: sqltypes.Int64},
@@ -79,14 +83,46 @@ func TestReaderReadHeartbeat(t *testing.T) {
 	utils.MustMatch(t, expectedHisto, heartbeatLagNsHistogram.Counts(), "wrong counts in histogram")
 }
 
+// TestReaderCloseSetsCurrentLagToZero tests that when closing the heartbeat reader, the current lag is
+// set to zero.
+func TestReaderCloseSetsCurrentLagToZero(t *testing.T) {
+	db := fakesqldb.New(t)
+	defer db.Close()
+	tr := newReader(db, nil)
+
+	db.AddQuery(fmt.Sprintf("SELECT ts FROM %s.heartbeat WHERE keyspaceShard='%s'", "_vt", tr.keyspaceShard), &sqltypes.Result{
+		Fields: []*querypb.Field{
+			{Name: "ts", Type: sqltypes.Int64},
+		},
+		Rows: [][]sqltypes.Value{{
+			sqltypes.NewInt64(time.Now().Add(-10 * time.Second).UnixNano()),
+		}},
+	})
+
+	currentLagNs.Reset()
+
+	tr.Open()
+	time.Sleep(2 * time.Second)
+
+	assert.Greater(t, currentLagNs.Get(), int64(0), "lag should be greater than zero")
+
+	tr.Close()
+
+	assert.Equal(t, int64(0), currentLagNs.Get(), "lag should be be zero after closing the reader.")
+}
+
 // TestReaderReadHeartbeatError tests that we properly account for errors
 // encountered in the reading of heartbeat.
 func TestReaderReadHeartbeatError(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
-	tr := newReader(db, mockNowFunc)
+
+	now := time.Now()
+	tr := newReader(db, &now)
 	defer tr.Close()
 
+	tr.pool.Open(tr.env.Config().DB.AppWithDB(), tr.env.Config().DB.DbaWithDB(), tr.env.Config().DB.AppDebugWithDB())
+
 	cumulativeLagNs.Reset()
 	readErrors.Reset()
 
@@ -100,18 +136,23 @@ func TestReaderReadHeartbeatError(t *testing.T) {
 	assert.Equal(t, int64(1), readErrors.Get(), "wrong read error count")
 }
 
-func newReader(db *fakesqldb.DB, nowFunc func() time.Time) *heartbeatReader {
+func newReader(db *fakesqldb.DB, frozenTime *time.Time) *heartbeatReader {
 	config := tabletenv.NewDefaultConfig()
 	config.ReplicationTracker.Mode = tabletenv.Heartbeat
 	config.ReplicationTracker.HeartbeatIntervalSeconds = 1
 	params, _ := db.ConnParams().MysqlParams()
 	cp := *params
 	dbc := dbconfigs.NewTestDBConfigs(cp, cp, "")
+	config.DB = dbc
 
 	tr := newHeartbeatReader(tabletenv.NewEnv(config, "ReaderTest"))
 	tr.keyspaceShard = "test:0"
-	tr.now = nowFunc
-	tr.pool.Open(dbc.AppWithDB(), dbc.DbaWithDB(), dbc.AppDebugWithDB())
+
+	if frozenTime != nil {
+		tr.now = func() time.Time {
+			return *frozenTime
+		}
+	}
 
 	return tr
 }

go/vt/vttablet/tabletserver/repltracker/writer_test.go

@@ -32,17 +32,11 @@ import (
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 )
 
-var (
-	now         = time.Now()
-	mockNowFunc = func() time.Time {
-		return now
-	}
-)
-
 func TestCreateSchema(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
-	tw := newTestWriter(db, mockNowFunc)
+	now := time.Now()
+	tw := newTestWriter(db, &now)
 	defer tw.Close()
 	writes.Reset()
 
@@ -66,7 +60,8 @@ func TestWriteHeartbeat(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
 
-	tw := newTestWriter(db, mockNowFunc)
+	now := time.Now()
+	tw := newTestWriter(db, &now)
 	upsert := fmt.Sprintf("INSERT INTO %s.heartbeat (ts, tabletUid, keyspaceShard) VALUES (%d, %d, '%s') ON DUPLICATE KEY UPDATE ts=VALUES(ts), tabletUid=VALUES(tabletUid)",
 		"_vt", now.UnixNano(), tw.tabletAlias.Uid, tw.keyspaceShard)
 	db.AddQuery(upsert, &sqltypes.Result{})
@@ -83,7 +78,8 @@ func TestWriteHeartbeatError(t *testing.T) {
 	db := fakesqldb.New(t)
 	defer db.Close()
 
-	tw := newTestWriter(db, mockNowFunc)
+	now := time.Now()
+	tw := newTestWriter(db, &now)
 
 	writes.Reset()
 	writeErrors.Reset()
@@ -93,7 +89,7 @@ func TestWriteHeartbeatError(t *testing.T) {
 	assert.Equal(t, int64(1), writeErrors.Get())
 }
 
-func newTestWriter(db *fakesqldb.DB, nowFunc func() time.Time) *heartbeatWriter {
+func newTestWriter(db *fakesqldb.DB, frozenTime *time.Time) *heartbeatWriter {
 	config := tabletenv.NewDefaultConfig()
 	config.ReplicationTracker.Mode = tabletenv.Heartbeat
 	config.ReplicationTracker.HeartbeatIntervalSeconds = 1
@@ -104,7 +100,13 @@ func newTestWriter(db *fakesqldb.DB, nowFunc func() time.Time) *heartbeatWriter
 
 	tw := newHeartbeatWriter(tabletenv.NewEnv(config, "WriterTest"), &topodatapb.TabletAlias{Cell: "test", Uid: 1111})
 	tw.keyspaceShard = "test:0"
-	tw.now = nowFunc
+
+	if frozenTime != nil {
+		tw.now = func() time.Time {
+			return *frozenTime
+		}
+	}
+
 	tw.appPool.Open(dbc.AppWithDB())
 	tw.allPrivsPool.Open(dbc.AllPrivsWithDB())