add more log in reader to debug

alibaba · Jul 24, 2020 · c2d2452 · c2d2452
1 parent e452ad6
commit c2d2452
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 3 deletions.
diff --git a/src/mongoshake/collector/docsyncer/doc_reader.go b/src/mongoshake/collector/docsyncer/doc_reader.go
@@ -125,6 +125,7 @@ func NewDocumentSplitter(src string, ns utils.NS) *DocumentSplitter {
 		return nil
 	}
 
+
 	// get total count
 	count, err := ds.conn.Session.DB(ds.ns.Database).C(ds.ns.Collection).Count()
 	if err != nil {

diff --git a/src/mongoshake/collector/reader/event_reader.go b/src/mongoshake/collector/reader/event_reader.go
@@ -12,6 +12,7 @@ import (
 	"github.com/vinllen/mgo/bson"
 	LOG "github.com/vinllen/log4go"
 	"github.com/vinllen/go-diskqueue"
+	"fmt"
 )
 
 const (
@@ -57,6 +58,10 @@ func NewEventReader(src string, replset string) *EventReader {
 	}
 }
 
+func (er *EventReader) String() string {
+	return fmt.Sprintf("EventReader[src:%s replset:%s]", er.src, er.replset)
+}
+
 func (er *EventReader) Name() string {
 	return utils.VarIncrSyncMongoFetchMethodChangeStream
 }
@@ -136,6 +141,8 @@ func (er *EventReader) EnsureNetwork() error {
 		return nil
 	}
 
+	LOG.Info("%s ensure network", er.String())
+
 	if er.client != nil {
 		er.client.Close() // close old client
 	}

diff --git a/src/mongoshake/collector/reader/oplog_reader.go b/src/mongoshake/collector/reader/oplog_reader.go
@@ -37,6 +37,7 @@ type OplogReader struct {
 	// source mongo address url
 	src     string
 	replset string
+
 	// mongo oplog reader
 	conn           *utils.MongoConn
 	oplogsIterator *mgo.Iter
@@ -64,6 +65,10 @@ func NewOplogReader(src string, replset string) *OplogReader {
 	}
 }
 
+func (or *OplogReader) String() string {
+	return fmt.Sprintf("oplogReader[src:%s replset:%s]", or.src, or.replset)
+}
+
 func (or *OplogReader) Name() string {
 	return utils.VarIncrSyncMongoFetchMethodOplog
 }
@@ -148,7 +153,7 @@ func (or *OplogReader) fetcher() {
 					LOG.Error("oplog collection capped may happen: %v", err)
 					or.oplogChan <- &retOplog{nil, CollectionCappedError}
 				} else {
-					or.oplogChan <- &retOplog{nil, fmt.Errorf("get next oplog failed. release oplogsIterator, %s", err.Error())}
+					or.oplogChan <- &retOplog{nil, fmt.Errorf("get next oplog failed, release oplogsIterator, %s", err.Error())}
 				}
 				// wait a moment
 				time.Sleep(1 * time.Second)
@@ -169,6 +174,8 @@ func (or *OplogReader) EnsureNetwork() (err error) {
 	if or.oplogsIterator != nil {
 		return nil
 	}
+	LOG.Info("%s ensure network", or.String())
+
 	if or.conn == nil || (or.conn != nil && !or.conn.IsGood()) {
 		if or.conn != nil {
 			or.conn.Close()
@@ -178,6 +185,9 @@ func (or *OplogReader) EnsureNetwork() (err error) {
 			err = fmt.Errorf("oplog_reader reconnect mongo instance [%s] error. %s", or.src, err.Error())
 			return err
 		}
+
+		or.conn.Session.SetBatch(BatchSize)
+		or.conn.Session.SetPrefetch(PrefetchPercent)
 	}
 
 	var queryTs bson.MongoTimestamp
@@ -210,8 +220,6 @@ func (or *OplogReader) EnsureNetwork() (err error) {
 	or.firstRead = false
 
 	// rebuild syncerGroup condition statement with current checkpoint timestamp
-	or.conn.Session.SetBatch(BatchSize)
-	or.conn.Session.SetPrefetch(PrefetchPercent)
 	or.oplogsIterator = or.conn.Session.DB(localDB).C(utils.OplogNS).
 		Find(or.query).LogReplay().Tail(time.Second * tailTimeout) // this timeout is useless
 	return