livesync.js

/**
 * This script will automatically sync all updates from one database to another. It is meant to be run while
 * syncing the database using mongodump and mongorestore.
 *
 * Example:
 *  node livesync.js mongodb://<user>:<pass>@dbhost.com:10645/app-production \
 *		mongodb://<user>:<pass>@dbhost.com:10499/local?authSource=app-production \
 *		app-production \
 *		mongodb://<user>:<pass>@newdbhost.com/app-prod
 */

const mongojs = require('mongojs');
var MongoOplog = require('mongo-oplog');

var fromDbUrl = process.argv[2];
var fromDbOplogUrl = process.argv[3];
var fromDbName = process.argv[4];
var toDbUrl = process.argv[5];

var fromDb = mongojs(fromDbUrl);
fromDb.on('error', function(err) {
	console.log(`Can't connect to 'from' db`, err);
	process.exit(1);
});

var toDb = mongojs(toDbUrl);
toDb.on('error', function(err) {
	console.log(`Can't connect to 'to' db`, err);
	process.exit(1);
});

var oplog = MongoOplog(fromDbOplogUrl, {
	ns: fromDbName
}).tail();

oplog.on('insert', function(doc) {
	var id = doc.o._id;
	var ns = stripNamespace(doc.ns);
	syncObject(ns, id);
});

oplog.on('update', function(doc) {
	var id = doc.o2._id;
	var ns = stripNamespace(doc.ns);
	syncObject(ns, id);
});

oplog.on('error', function(error) {
	console.log(error);
});

oplog.on('end', function() {
	console.error('Stream ended');
});

oplog.stop(function() {
	console.error('server stopped');
});

// Strips the namespace from the oplog update.
function stripNamespace(namespace) {
	return namespace.replace(new RegExp(`^${fromDbName}\.`), '');
}

// Grabs an object from the 'from' db and moves it to the 'to' db.
function syncObject(ns, id) {
	fromDb.collection(ns).findOne({
		_id: id
	}, (err, res) => {
		if (err) {
			console.error(`ERROR: could not find document to insert with id ${id}`, err);
		} else if (!res) {
			console.error(`ERROR: got oplog message but couldn't find ${id}`);
		} else {
			console.log(`syncing ${ns}`);
			toDb.collection(ns).update({
				_id: id
			}, res, {
				upsert: true
			}, (err, res) => {
				if (err) console.error(`ERROR: upserting doc`, err);
			});
		}
	});
}

transfer.sh

#!/bin/bash

#
# This script will transfer data from one database to another host. See parameters below
#
# Example:
#	./migration_script.sh dbhost.com:10499 app-production migrate 1234 newdbhost:27017 app-prod migrate 1234
#

# Temporary directory where to keep indexes
TMP_INDEX_DIRECTORY=`mktemp -d`

# Number of cores
NUM_WORKERS=`nproc`

OUTBOUND_CONNECTION_STRING=$1
OUTBOUND_DATABASE_NAME=$2
OUTBOUND_USER=$3
OUTBOUND_PASS=$4

INBOUND_CONNECTION_STRING=$5
INBOUND_DATABASE_NAME=$6
INBOUND_USER=$7
INBOUND_PASS=$8

COLLECTIONS=$(mongo $OUTBOUND_CONNECTION_STRING/$OUTBOUND_DATABASE_NAME --username $OUTBOUND_USER --password $OUTBOUND_PASS --eval 'db.getCollectionNames()' | awk '{print $2}' FS='"' ORS=' ' | tr '\t' ' ')

for COLLECTION in ${COLLECTIONS[@]}; do
    echo "DUMPING INDEXES"
    mongodump --host $OUTBOUND_CONNECTION_STRING --db $OUTBOUND_DATABASE_NAME --username $OUTBOUND_USER --password $OUTBOUND_PASS --collection $COLLECTION --query '{ "_id": 0 }' --out $TMP_INDEX_DIRECTORY

    echo "RESTORING INDEXES"
    mongorestore --numInsertionWorkersPerCollection=$NUM_WORKERS --host $INBOUND_CONNECTION_STRING --db $INBOUND_DATABASE_NAME --username $INBOUND_USER --password $INBOUND_PASS --collection $COLLECTION $TMP_INDEX_DIRECTORY/$OUTBOUND_DATABASE_NAME/$COLLECTION.bson

    echo "DUMPING AND LOADING DATA FROM STDOUT/STDIN"
    mongodump --host $OUTBOUND_CONNECTION_STRING --db $OUTBOUND_DATABASE_NAME --username $OUTBOUND_USER --password $OUTBOUND_PASS  --collection $COLLECTION --out - | mongorestore --ssl --numInsertionWorkersPerCollection=$NUM_WORKERS --host $INBOUND_CONNECTION_STRING --db $INBOUND_DATABASE_NAME --username $INBOUND_USER --password $INBOUND_PASS  --collection $COLLECTION -

    echo "DONE"
done