Skip to content

Store last-tailed oplog timestamp so the tailer can resume properly #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Jul 10, 2018
Merged

Store last-tailed oplog timestamp so the tailer can resume properly #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
69c0dc3
Store last-tailed oplog entry to the tailer can resume properly
timvaillancourt Jul 9, 2018
05dbf3e
Store a pointer to the last oplog 'ts' instead of the entire oplog doc
timvaillancourt Jul 9, 2018
f08bb0e
Use a timestamp-only struct to reduce unmarshalling
timvaillancourt Jul 9, 2018
a937fae
Add struct for unmarshalling the 'ts' only
timvaillancourt Jul 9, 2018
6f25cd9
Retry iterable cursor on bson.Raw unmarshal error
timvaillancourt Jul 9, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions internal/oplog/oplog_tail.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,9 @@ import (
type chanDataTye []byte

type OplogTail struct {
session *mgo.Session
oplogCollection string
lastOplogEntry *mdbstructs.Oplog
session *mgo.Session
oplogCollection string
lastOplogTimestamp *bson.MongoTimestamp

totalSize int64
docsCount int64
Expand Down Expand Up @@ -114,8 +114,14 @@ func (ot *OplogTail) tail() {
}
result := bson.Raw{}
if iter.Next(&result) {
ot.dataChan <- result.Data
continue
oplog := mdbstructs.OplogTimestampOnly{}
err := result.Unmarshal(&oplog)
if err == nil {
ot.dataChan <- result.Data
ot.lastOplogTimestamp = &oplog.Timestamp
continue
}
iter.Close()
}
if iter.Timeout() {
continue
Expand All @@ -128,8 +134,8 @@ func (ot *OplogTail) tail() {
}

func (ot *OplogTail) getOplogTailTimestamp(col *mgo.Collection) bson.MongoTimestamp {
oplog := &mdbstructs.Oplog{}
err := col.Find(nil).Sort("$natural").Limit(1).One(oplog)
oplog := mdbstructs.OplogTimestampOnly{}
err := col.Find(nil).Sort("$natural").Limit(1).One(&oplog)
if err != nil {
return bson.MongoTimestamp(0)
}
Expand All @@ -138,8 +144,8 @@ func (ot *OplogTail) getOplogTailTimestamp(col *mgo.Collection) bson.MongoTimest

func (ot *OplogTail) tailQuery(col *mgo.Collection) bson.M {
query := bson.M{"op": bson.M{"$ne": mdbstructs.OperationNoop}}
if ot.lastOplogEntry != nil {
query["ts"] = bson.M{"$gt": ot.lastOplogEntry.Timestamp}
if ot.lastOplogTimestamp != nil {
query["ts"] = bson.M{"$gt": *ot.lastOplogTimestamp}
} else {
query["ts"] = bson.M{"$gte": ot.getOplogTailTimestamp(col)}
}
Expand Down
4 changes: 4 additions & 0 deletions mdbstructs/oplog.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,10 @@ type Oplog struct {
WallTime time.Time `bson:"wall,omitempty"`
}

type OplogTimestampOnly struct {
Timestamp bson.MongoTimestamp `bson:"ts"`
}

type OplogPosition struct {
Timestamp bson.MongoTimestamp
HistoryID int64
Expand Down