src/mongo/shell/replsettest.js
/**
     * Check oplogs on all nodes, by reading from the last time. Since the oplog is a capped
     * collection, each node may not contain the same number of entries and stop if the cursor
     * is exhausted on any node being checked.
     */
Check oplogs on all nodes, by reading from the last time. Since the oplog is a capped
collection, each node may not contain the same number of entries and stop if the cursor
is exhausted on any node being checked.
Check oplogs on all nodes by reading from the last time . Since the oplog is a capped collection each node may not contain the same number of entries and stop if the cursor is exhausted on any node being checked .
function checkOplogs(rst, secondaries, msgPrefix = 'checkOplogs') {
        secondaries = secondaries || rst._secondaries;
        const kCappedPositionLostSentinel = Object.create(null);
        const OplogReader = function(mongo) {
            this._safelyPerformCursorOperation = function(name, operation, onCappedPositionLost) {
                if (!this.cursor) {
                    throw new Error("OplogReader is not open!");
                }
                if (this._cursorExhausted) {
                    return onCappedPositionLost;
                }
                try {
                    return operation(this.cursor);
                } catch (err) {
                    print("Error: " + name + " threw '" + err.message + "' on " + this.mongo.host);
                    if (err.code === ErrorCodes.CappedPositionLost) {
                        this.cursor.close();
                        this._cursorExhausted = true;
                        return onCappedPositionLost;
                    }
                    throw err;
                }
            };
            this.next = function() {
                return this._safelyPerformCursorOperation('next', function(cursor) {
                    return cursor.next();
                }, kCappedPositionLostSentinel);
            };
            this.hasNext = function() {
                return this._safelyPerformCursorOperation('hasNext', function(cursor) {
                    return cursor.hasNext();
                }, false);
            };
            this.query = function(ts) {
                const coll = this.getOplogColl();
                const query = {ts: {$gte: ts ? ts : new Timestamp()}};
                this._cursorExhausted = false;
                this.cursor =
                    coll.find(query).sort({$natural: -1}).noCursorTimeout().readConcern("local");
            };
            this.getFirstDoc = function() {
                return this.getOplogColl()
                    .find()
                    .sort({$natural: 1})
                    .readConcern("local")
                    .limit(-1)
                    .next();
            };
            this.getOplogColl = function() {
                return this.mongo.getDB("local")[oplogName];
            };
            this.cursor = null;
            this._cursorExhausted = true;
            this.mongo = mongo;
        };
        function assertOplogEntriesEq(oplogEntry0, oplogEntry1, reader0, reader1, prevOplogEntry) {
            if (!bsonBinaryEqual(oplogEntry0, oplogEntry1)) {
                const query = prevOplogEntry ? {ts: {$lte: prevOplogEntry.ts}} : {};
                rst.nodes.forEach(node => this.dumpOplog(node, query, 100));
                const log = msgPrefix + ", non-matching oplog entries for the following nodes: \n" +
                    reader0.mongo.host + ": " + tojsononeline(oplogEntry0) + "\n" +
                    reader1.mongo.host + ": " + tojsononeline(oplogEntry1);
                assert(false, log);
            }
        }
        print("checkOplogs starting oplog checks.");
        print("checkOplogs waiting for secondaries to be ready.");
        this.awaitSecondaryNodes(self.kDefaultTimeoutMS, secondaries);
        if (secondaries.length >= 1) {
            let readers = [];
            let smallestTS = new Timestamp(Math.pow(2, 32) - 1, Math.pow(2, 32) - 1);
            const nodes = rst.nodes;
            let firstReaderIndex;
            for (let i = 0; i < nodes.length; i++) {
                const node = nodes[i];
                if (rst._primary !== node && !secondaries.includes(node)) {
                    print("checkOplogs skipping oplog of node: " + node.host);
                    continue;
                }
                if (isNodeArbiter(node)) {
                    jsTestLog("checkOplogs skipping oplog of arbiter: " + node.host);
                    continue;
                }
                print("checkOplogs going to check oplog of node: " + node.host);
                readers[i] = new OplogReader(node);
                const currTS = readers[i].getFirstDoc().ts;
                if (timestampCmp(currTS, smallestTS) < 0) {
                    smallestTS = currTS;
                    firstReaderIndex = i;
                }
                readers[i].query();
            }
            const firstReader = readers[firstReaderIndex];
            let prevOplogEntry;
            assert(firstReader.hasNext(), "oplog is empty while checkOplogs is called");
            let bytesSinceGC = 0;
            while (firstReader.hasNext()) {
                const oplogEntry = firstReader.next();
                bytesSinceGC += Object.bsonsize(oplogEntry);
                if (oplogEntry === kCappedPositionLostSentinel) {
                    break;
                }
                for (let i = 0; i < nodes.length; i++) {
                    if (i === firstReaderIndex || !(readers[i] && readers[i].hasNext())) {
                        continue;
                    }
                    const otherOplogEntry = readers[i].next();
                    bytesSinceGC += Object.bsonsize(otherOplogEntry);
                    if (otherOplogEntry && otherOplogEntry !== kCappedPositionLostSentinel) {
                        assertOplogEntriesEq.call(this,
                                                  oplogEntry,
                                                  otherOplogEntry,
                                                  firstReader,
                                                  readers[i],
                                                  prevOplogEntry);
                    }
                }
                if (bytesSinceGC > (10 * 1024 * 1024)) {
                    gc();
                    bytesSinceGC = 0;
                }
                prevOplogEntry = oplogEntry;
            }
        }
        print("checkOplogs oplog checks complete.");
    }
function checkOplogs ( rst , secondaries , msgPrefix = 'checkOplogs' ) { secondaries = secondaries || rst . _secondaries ; const kCappedPositionLostSentinel = Object . create ( null ) ; const OplogReader = function ( mongo ) { this . _safelyPerformCursorOperation = function ( name , operation , onCappedPositionLost ) { if ( ! this . cursor ) { throw new Error ( "OplogReader is not open!" ) ; } if ( this . _cursorExhausted ) { return onCappedPositionLost ; } try { return operation ( this . cursor ) ; } catch ( err ) { print ( "Error: " + name + " threw '" + err . message + "' on " + this . mongo . host ) ; if ( err . code === ErrorCodes . CappedPositionLost ) { this . cursor . close ( ) ; this . _cursorExhausted = true ; return onCappedPositionLost ; } throw err ; } } ; this . next = function ( ) { return this . _safelyPerformCursorOperation ( 'next' , function ( cursor ) { return cursor . next ( ) ; } , kCappedPositionLostSentinel ) ; } ; this . hasNext = function ( ) { return this . _safelyPerformCursorOperation ( 'hasNext' , function ( cursor ) { return cursor . hasNext ( ) ; } , false ) ; } ; this . query = function ( ts ) { const coll = this . getOplogColl ( ) ; const query = { ts : { $gte : ts ? ts : new Timestamp ( ) } } ; this . _cursorExhausted = false ; this . cursor = coll . find ( query ) . sort ( { $natural : - 1 } ) . noCursorTimeout ( ) . readConcern ( "local" ) ; } ; this . getFirstDoc = function ( ) { return this . getOplogColl ( ) . find ( ) . sort ( { $natural : 1 } ) . readConcern ( "local" ) . limit ( - 1 ) . next ( ) ; } ; this . getOplogColl = function ( ) { return this . mongo . getDB ( "local" ) [ oplogName ] ; } ; this . cursor = null ; this . _cursorExhausted = true ; this . mongo = mongo ; } ; function assertOplogEntriesEq ( oplogEntry0 , oplogEntry1 , reader0 , reader1 , prevOplogEntry ) { if ( ! bsonBinaryEqual ( oplogEntry0 , oplogEntry1 ) ) { const query = prevOplogEntry ? { ts : { $lte : prevOplogEntry . ts } } : { } ; rst . nodes . forEach ( node => this . dumpOplog ( node , query , 100 ) ) ; const log = msgPrefix + ", non-matching oplog entries for the following nodes: \n" + reader0 . mongo . host + ": " + tojsononeline ( oplogEntry0 ) + "\n" + reader1 . mongo . host + ": " + tojsononeline ( oplogEntry1 ) ; assert ( false , log ) ; } } print ( "checkOplogs starting oplog checks." ) ; print ( "checkOplogs waiting for secondaries to be ready." ) ; this . awaitSecondaryNodes ( self . kDefaultTimeoutMS , secondaries ) ; if ( secondaries . length >= 1 ) { let readers = [ ] ; let smallestTS = new Timestamp ( Math . pow ( 2 , 32 ) - 1 , Math . pow ( 2 , 32 ) - 1 ) ; const nodes = rst . nodes ; let firstReaderIndex ; for ( let i = 0 ; i < nodes . length ; i ++ ) { const node = nodes [ i ] ; if ( rst . _primary !== node && ! secondaries . includes ( node ) ) { print ( "checkOplogs skipping oplog of node: " + node . host ) ; continue ; } if ( isNodeArbiter ( node ) ) { jsTestLog ( "checkOplogs skipping oplog of arbiter: " + node . host ) ; continue ; } print ( "checkOplogs going to check oplog of node: " + node . host ) ; readers [ i ] = new OplogReader ( node ) ; const currTS = readers [ i ] . getFirstDoc ( ) . ts ; if ( timestampCmp ( currTS , smallestTS ) < 0 ) { smallestTS = currTS ; firstReaderIndex = i ; } readers [ i ] . query ( ) ; } const firstReader = readers [ firstReaderIndex ] ; let prevOplogEntry ; assert ( firstReader . hasNext ( ) , "oplog is empty while checkOplogs is called" ) ; let bytesSinceGC = 0 ; while ( firstReader . hasNext ( ) ) { const oplogEntry = firstReader . next ( ) ; bytesSinceGC += Object . bsonsize ( oplogEntry ) ; if ( oplogEntry === kCappedPositionLostSentinel ) { break ; } for ( let i = 0 ; i < nodes . length ; i ++ ) { if ( i === firstReaderIndex || ! ( readers [ i ] && readers [ i ] . hasNext ( ) ) ) { continue ; } const otherOplogEntry = readers [ i ] . next ( ) ; bytesSinceGC += Object . bsonsize ( otherOplogEntry ) ; if ( otherOplogEntry && otherOplogEntry !== kCappedPositionLostSentinel ) { assertOplogEntriesEq . call ( this , oplogEntry , otherOplogEntry , firstReader , readers [ i ] , prevOplogEntry ) ; } } if ( bytesSinceGC > ( 10 * 1024 * 1024 ) ) { gc ( ) ; bytesSinceGC = 0 ; } prevOplogEntry = oplogEntry ; } } print ( "checkOplogs oplog checks complete." ) ; }
Check oplogs on all nodes, by reading from the last time.
Check oplogs on all nodes by reading from the last time .
// Occasionally, the capped collection will get truncated while we are iterating // over it. Since we are iterating over the collection in reverse, getting a // truncated item means we've reached the end of the list, so return false. // Set the cursor to read backwards, from last to first. We also set the cursor not // to time out since it may take a while to process each batch and a test may have // changed "cursorTimeoutMillis" to a short time period. // Arbiters have no documents in the oplog. // Find the reader which has the smallestTS. This reader should have the most // number of documents in the oplog. // Start all oplogReaders at their last document. // Read from the reader which has the most oplog entries. // Note, we read the oplog backwards from last to first. // Track the number of bytes we are reading as we check the oplog. We use this to avoid // out-of-memory issues by calling to garbage collect whenever the memory footprint is // large. // When using legacy OP_QUERY/OP_GET_MORE reads against mongos, it is // possible for hasNext() to return true but for next() to throw an exception. // Skip reading from this reader if the index is the same as firstReader or // the cursor is exhausted. // Garbage collect every 10MB.