Use oplog to make mongodump/restore point in time SERVER-2025

2 files changed, 111 insertions, 5 deletions

diff --git a/tools/dump.cpp b/tools/dump.cpp
index e5910a26560..2c820e44379 100644
--- a/tools/dump.cpp
+++ b/tools/dump.cpp
@@ -32,6 +32,7 @@ public:
         add_options()
             ("out,o", po::value<string>()->default_value("dump"), "output directory or \"-\" for stdout")
             ("query,q", po::value<string>() , "json query" )
+            ("oplog", "Use oplog for point-in-time snapshotting" )
             ;
     }
 
@@ -59,6 +60,10 @@ public:
         else
             q = _query;
 
+        int queryOptions = QueryOption_SlaveOk | QueryOption_NoCursorTimeout;
+        if (startsWith(coll.c_str(), "local.oplog."))
+            queryOptions |= QueryOption_OplogReplay;
+
         DBClientBase& connBase = conn(true);
         Writer writer(out, m);
 
@@ -66,10 +71,10 @@ public:
         if (typeid(connBase) == typeid(DBClientConnection&)){
             DBClientConnection& conn = static_cast<DBClientConnection&>(connBase);
             boost::function<void(const BSONObj&)> castedWriter(writer); // needed for overload resolution
-            conn.query( castedWriter, coll.c_str() , q , NULL, QueryOption_SlaveOk | QueryOption_NoCursorTimeout | QueryOption_Exhaust);
+            conn.query( castedWriter, coll.c_str() , q , NULL, queryOptions | QueryOption_Exhaust);
         } else {
             //This branch should only be taken with DBDirectClient which doesn't support exhaust mode
-            scoped_ptr<DBClientCursor> cursor(connBase.query( coll.c_str() , q , 0 , 0 , 0 , QueryOption_SlaveOk | QueryOption_NoCursorTimeout ));
+            scoped_ptr<DBClientCursor> cursor(connBase.query( coll.c_str() , q , 0 , 0 , 0 , queryOptions ));
             while ( cursor->more() ) {
                 writer(cursor->next());
             }
@@ -129,6 +134,40 @@ public:
                 _query = fromjson( q );
         }
 
+        string opLogName = "";
+        unsigned long long opLogStart = 0;
+        if (hasParam("oplog")) {
+            if (hasParam("query") || hasParam("db") || hasParam("collection")){
+                cout << "oplog mode is only supported on full dumps" << endl;
+                return -1;
+            }
+
+            
+            BSONObj isMaster;
+            conn("true").simpleCommand("admin", &isMaster, "isMaster");
+
+            if (isMaster.hasField("hosts")) { // if connected to replica set member
+                opLogName = "local.oplog.rs";
+            } else {
+                opLogName = "local.oplog.$main";
+                if ( ! isMaster["ismaster"].trueValue() ){
+                    cout << "oplog mode is only supported on master or replica set member" << endl;
+                    return -1;
+                }
+            }
+
+            BSONObj op = conn(true).findOne(opLogName, Query().sort("$natural", -1), 0, QueryOption_SlaveOk);
+            if (op.isEmpty()) {
+                cout << "No operations in oplog. Please ensure you are connecting to a master." << endl;
+                return -1;
+            }
+            
+            assert(op["ts"].type() == Timestamp);
+            opLogStart = op["ts"]._numberLong();
+        }
+
+            
+
         // check if we're outputting to stdout
         string out = getParam("out");
         if ( out == "-" ) {
@@ -169,6 +208,16 @@ public:
             auth( db );
             go( db , root / db );
         }
+
+        if (!opLogName.empty()){
+            BSONObjBuilder b;
+            b.appendTimestamp("$gt", opLogStart);
+
+            _query = BSON("ts" << b.obj());
+
+            writeCollectionFile( opLogName , root / "oplog.bson" );
+        }
+
         return 0;
     }
 
diff --git a/tools/restore.cpp b/tools/restore.cpp
index 1258bd9f662..c81521c0ed5 100644
--- a/tools/restore.cpp
+++ b/tools/restore.cpp
@@ -19,6 +19,7 @@
 #include "../pch.h"
 #include "../client/dbclient.h"
 #include "../util/mmap.h"
+#include "../util/version.h"
 #include "tool.h"
 
 #include <boost/program_options.hpp>
@@ -29,6 +30,10 @@ using namespace mongo;
 
 namespace po = boost::program_options;
 
+namespace {
+    const char* OPLOG_SENTINEL = "$oplog";  // compare by ptr not strcmp
+}
+
 class Restore : public BSONTool {
 public:
     
@@ -38,6 +43,7 @@ public:
     Restore() : BSONTool( "restore" ) , _drop(false){
         add_options()
             ("drop" , "drop each collection before import" )
+            ("oplogReplay" , "replay oplog for point-in-time restore")
             ;
         add_hidden_options()
             ("dir", po::value<string>()->default_value("dump"), "directory to restore from")
@@ -61,6 +67,34 @@ public:
         
         _drop = hasParam( "drop" );
 
+        bool doOplog = hasParam( "oplogReplay" );
+        if (doOplog){
+            // fail early if errors
+
+            if (_db != ""){
+                cout << "Can only replay oplog on full restore" << endl;
+                return -1;
+            }
+
+            if ( ! exists(root / "oplog.bson") ){
+                cout << "No oplog file to replay. Make sure you run mongodump with --oplog." << endl;
+                return -1;
+            }
+
+
+            BSONObj out;
+            if (! conn().simpleCommand("admin", &out, "buildinfo")){
+                cout << "buildinfo command failed: " << out["errmsg"].String() << endl;
+                return -1;
+            }
+
+            StringData version = out["version"].valuestr();
+            if (versionCmp(version, "1.7.4-pre-") < 0){
+                cout << "Can only replay oplog to server version >= 1.7.4" << endl;
+                return -1;
+            }
+        }
+
         /* If _db is not "" then the user specified a db name to restore as.
          *
          * In that case we better be given either a root directory that
@@ -70,12 +104,19 @@ public:
          * given either a root directory that contains only a single
          * .bson file, or a single .bson file itself (a collection).
          */
-        drillDown(root, _db != "", _coll != "");
+        drillDown(root, _db != "", _coll != "", true);
         conn().getLastError();
+
+        if (doOplog){
+            out() << "\t Replaying oplog" << endl;
+            _curns = OPLOG_SENTINEL;
+            processFile( root / "oplog.bson" );
+        }
+
         return EXIT_CLEAN;
     }
 
-    void drillDown( path root, bool use_db = false, bool use_coll = false ) {
+    void drillDown( path root, bool use_db, bool use_coll, bool top_level=false ) {
         log(2) << "drillDown: " << root.string() << endl;
 
         // skip hidden files and directories
@@ -108,6 +149,10 @@ public:
                     }
                 }
 
+                // don't insert oplog
+                if (top_level && !use_db && p.leaf() == "oplog.bson")
+                    continue;
+
                 if ( p.leaf() == "system.indexes.bson" )
                     indexes = p;
                 else
@@ -170,7 +215,19 @@ public:
     }
 
     virtual void gotObject( const BSONObj& obj ){
-        conn().insert( _curns , obj );
+        if (_curns == OPLOG_SENTINEL) { // intentional ptr compare
+            if (obj["op"].valuestr()[0] == 'n') // skip no-ops
+                return;
+
+            string db = obj["ns"].valuestr();
+            db = db.substr(0, db.find('.'));
+
+            BSONObj cmd = BSON( "applyOps" << BSON_ARRAY( obj ) );
+            BSONObj out;
+            conn().runCommand(db, cmd, out);
+        } else {
+            conn().insert( _curns , obj );
+        }
     }