VReplication Logger Stream

go.sum

@@ -456,6 +456,7 @@ github.com/mitchellh/copystructure v0.0.0-20160804032330-cdac8253d00f/go.mod h1:
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
+github.com/mitchellh/go-ps v0.0.0-20190716172923-621e5597135b h1:9+ke9YJ9KGWw5ANXK6ozjoK47uI3uNbXv4YVINBnGm8=
 github.com/mitchellh/go-ps v0.0.0-20190716172923-621e5597135b/go.mod h1:r1VsdOzOPt1ZSrGZWFoNhsAedKnEd6r9Np1+5blZCWk=
 github.com/mitchellh/go-testing-interface v1.0.0 h1:fzU/JVNcaqHQEcVFAKeR41fkiLdIPrefOvVG1VZ96U0=
 github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=

go/vt/vttablet/tabletmanager/vreplication/framework_test.go

@@ -55,6 +55,11 @@ var (
 	globalDBQueries = make(chan string, 1000)
 )
 
+type LogExpectation struct {
+	Type   string
+	Detail string
+}
+
 func init() {
 	tabletconn.RegisterDialer("test", func(tablet *topodatapb.Tablet, failFast grpcclient.FailFast) (queryservice.QueryService, error) {
 		return &fakeTabletConn{
@@ -391,6 +396,45 @@ func expectDeleteQueries(t *testing.T) {
 	})
 }
 
+func expectLogsAndUnsubscribe(t *testing.T, logs []LogExpectation, logCh chan interface{}) {
+	t.Helper()
+	defer vrLogStatsLogger.Unsubscribe(logCh)
+	failed := false
+	for i, log := range logs {
+		if failed {
+			t.Errorf("no logs received")
+			continue
+		}
+		select {
+		case data := <-logCh:
+			got, ok := data.(*VrLogStats)
+			if !ok {
+				t.Errorf("got not ok casting to VrLogStats: %v", data)
+			}
+			var match bool
+			match = (log.Type == got.Type)
+			if match {
+				if log.Detail[0] == '/' {
+					result, err := regexp.MatchString(log.Detail[1:], got.Detail)
+					if err != nil {
+						panic(err)
+					}
+					match = result
+				} else {
+					match = (got.Detail == log.Detail)
+				}
+			}
+
+			if !match {
+				t.Errorf("log:\n%q, does not match log %d:\n%q", got, i, log)
+			}
+		case <-time.After(5 * time.Second):
+			t.Errorf("no logs received, expecting %s", log)
+			failed = true
+		}
+	}
+}
+
 func expectDBClientQueries(t *testing.T, queries []string) {
 	t.Helper()
 	failed := false

go/vt/vttablet/tabletmanager/vreplication/http_stream_mock.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2019 The Vitess Authors.
+
+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,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+ * To be used instead of HttpTest in cases where a test needs to do synchronized sequential reads are required from the
+ * http stream. Note that this handles only one write at a time: the test should read the written data before the next
+ * write. Data is sent to the channel only on a Flush.
+ */
+
+package vreplication
+
+import (
+	"net/http"
+)
+
+// HTTPStreamWriterMock implements http.ResponseWriter and adds a channel to sync writes and reads
+type HTTPStreamWriterMock struct {
+	ch   chan interface{}
+	data []byte
+}
+
+// NewHTTPStreamWriterMock returns a new HTTPStreamWriterMock
+func NewHTTPStreamWriterMock() *HTTPStreamWriterMock {
+	return &HTTPStreamWriterMock{ch: make(chan interface{}, 1), data: make([]byte, 0)}
+}
+
+// Header is a stub
+func (w *HTTPStreamWriterMock) Header() http.Header {
+	return nil
+}
+
+// WriteHeader is a stub
+func (w *HTTPStreamWriterMock) WriteHeader(statuscode int) {
+}
+
+// Write buffers sent data
+func (w *HTTPStreamWriterMock) Write(data []byte) (int, error) {
+	w.data = append(w.data, data...)
+	return 0, nil
+}
+
+// Flush sends buffered data to the channel
+func (w *HTTPStreamWriterMock) Flush() {
+	w.ch <- w.data
+	w.data = w.data[:0]
+}

go/vt/vttablet/tabletmanager/vreplication/vplayer.go

@@ -192,7 +192,10 @@ func (vp *vplayer) applyRowEvent(ctx context.Context, rowEvent *binlogdatapb.Row
 	}
 	for _, change := range rowEvent.RowChanges {
 		_, err := tplan.applyChange(change, func(sql string) (*sqltypes.Result, error) {
-			return vp.vr.dbClient.ExecuteWithRetry(ctx, sql)
+			stats := NewVrLogStats("ROWCHANGE")
+			result, err := vp.vr.dbClient.ExecuteWithRetry(ctx, sql)
+			stats.Send(sql)
+			return result, err
 		})
 		if err != nil {
 			return err
@@ -307,6 +310,7 @@ func hasAnotherCommit(items [][]*binlogdatapb.VEvent, i, j int) bool {
 }
 
 func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, mustSave bool) error {
+	stats := NewVrLogStats(event.Type.String())
 	switch event.Type {
 	case binlogdatapb.VEventType_GTID:
 		pos, err := mysql.DecodePosition(event.Gtid)
@@ -352,6 +356,8 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			return err
 		}
 		vp.tablePlans[event.FieldEvent.TableName] = tplan
+		stats.Send(fmt.Sprintf("%v", event.FieldEvent))
+
 	case binlogdatapb.VEventType_INSERT, binlogdatapb.VEventType_DELETE, binlogdatapb.VEventType_UPDATE, binlogdatapb.VEventType_REPLACE:
 		// This is a player using stament based replication
 		if err := vp.vr.dbClient.Begin(); err != nil {
@@ -361,6 +367,7 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 		if err := vp.applyStmtEvent(ctx, event); err != nil {
 			return err
 		}
+		stats.Send(fmt.Sprintf(event.Dml))
 	case binlogdatapb.VEventType_ROW:
 		// This player is configured for row based replication
 		if err := vp.vr.dbClient.Begin(); err != nil {
@@ -369,6 +376,8 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 		if err := vp.applyRowEvent(ctx, event.RowEvent); err != nil {
 			return err
 		}
+		//Row event is logged AFTER RowChanges are applied so as to calculate the total elapsed time for the Row event
+		stats.Send(fmt.Sprintf("%v", event.RowEvent))
 	case binlogdatapb.VEventType_OTHER:
 		if vp.vr.dbClient.InTransaction {
 			// Unreachable
@@ -417,6 +426,7 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			if _, err := vp.vr.dbClient.ExecuteWithRetry(ctx, event.Ddl); err != nil {
 				return err
 			}
+			stats.Send(fmt.Sprintf("%v", event.Ddl))
 			posReached, err := vp.updatePos(event.Timestamp)
 			if err != nil {
 				return err
@@ -428,6 +438,7 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			if _, err := vp.vr.dbClient.ExecuteWithRetry(ctx, event.Ddl); err != nil {
 				log.Infof("Ignoring error: %v for DDL: %s", err, event.Ddl)
 			}
+			stats.Send(fmt.Sprintf("%v", event.Ddl))
 			posReached, err := vp.updatePos(event.Timestamp)
 			if err != nil {
 				return err
@@ -481,6 +492,7 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			}
 			return io.EOF
 		}
+		stats.Send(fmt.Sprintf("%v", event.Journal))
 		return io.EOF
 	case binlogdatapb.VEventType_HEARTBEAT:
 		// No-op: heartbeat timings are calculated in outer loop.

go/vt/vttablet/tabletmanager/vreplication/vplayer_test.go

@@ -173,6 +173,7 @@ func TestPlayerFilters(t *testing.T) {
 		Filter:   filter,
 		OnDdl:    binlogdatapb.OnDDLAction_IGNORE,
 	}
+
 	cancel, _ := startVReplication(t, bls, "")
 	defer cancel()
 
@@ -181,6 +182,7 @@ func TestPlayerFilters(t *testing.T) {
 		output []string
 		table  string
 		data   [][]string
+		logs   []LogExpectation // logs are defined for a few testcases since they are enough to test all log events
 	}{{
 		// insert with insertNormal
 		input: "insert into src1 values(1, 'aaa')",
@@ -194,6 +196,11 @@ func TestPlayerFilters(t *testing.T) {
 		data: [][]string{
 			{"1", "aaa"},
 		},
+		logs: []LogExpectation{
+			{"FIELD", "/src1.*id.*INT32.*val.*VARBINARY.*"},
+			{"ROWCHANGE", "insert into dst1(id,val) values (1,'aaa')"},
+			{"ROW", "/src1.*3.*1aaa.*"},
+		},
 	}, {
 		// update with insertNormal
 		input: "update src1 set val='bbb'",
@@ -207,6 +214,10 @@ func TestPlayerFilters(t *testing.T) {
 		data: [][]string{
 			{"1", "bbb"},
 		},
+		logs: []LogExpectation{
+			{"ROWCHANGE", "update dst1 set val='bbb' where id=1"},
+			{"ROW", "/src1.*3.*1aaa.*"},
+		},
 	}, {
 		// delete with insertNormal
 		input: "delete from src1 where id=1",
@@ -218,6 +229,10 @@ func TestPlayerFilters(t *testing.T) {
 		},
 		table: "dst1",
 		data:  [][]string{},
+		logs: []LogExpectation{
+			{"ROWCHANGE", "delete from dst1 where id=1"},
+			{"ROW", "/src1.*3.*1bbb.*"},
+		},
 	}, {
 		// insert with insertOnDup
 		input: "insert into src2 values(1, 2, 3)",
@@ -231,6 +246,10 @@ func TestPlayerFilters(t *testing.T) {
 		data: [][]string{
 			{"1", "2", "3", "1"},
 		},
+		logs: []LogExpectation{
+			{"FIELD", "/src2.*id.*val1.*val2.*"},
+			{"ROWCHANGE", "insert into dst2(id,val1,sval2,rcount) values (1,2,ifnull(3, 0),1) on duplicate key update val1=values(val1), sval2=sval2+ifnull(values(sval2), 0), rcount=rcount+1"},
+		},
 	}, {
 		// update with insertOnDup
 		input: "update src2 set val1=5, val2=1 where id=1",
@@ -244,6 +263,10 @@ func TestPlayerFilters(t *testing.T) {
 		data: [][]string{
 			{"1", "5", "1", "1"},
 		},
+		logs: []LogExpectation{
+			{"ROWCHANGE", "update dst2 set val1=5, sval2=sval2-ifnull(3, 0)+ifnull(1, 0), rcount=rcount where id=1"},
+			{"ROW", "/src2.*123.*"},
+		},
 	}, {
 		// delete with insertOnDup
 		input: "delete from src2 where id=1",
@@ -365,11 +388,15 @@ func TestPlayerFilters(t *testing.T) {
 		data:  [][]string{},
 	}}
 
-	for _, tcases := range testcases {
-		execStatements(t, []string{tcases.input})
-		expectDBClientQueries(t, tcases.output)
-		if tcases.table != "" {
-			expectData(t, tcases.table, tcases.data)
+	for _, tcase := range testcases {
+		if tcase.logs != nil {
+			logch := vrLogStatsLogger.Subscribe("vrlogstats")
+			defer expectLogsAndUnsubscribe(t, tcase.logs, logch)
+		}
+		execStatements(t, []string{tcase.input})
+		expectDBClientQueries(t, tcase.output)
+		if tcase.table != "" {
+			expectData(t, tcase.table, tcase.data)
 		}
 	}
 }