Create an Oplog Process File

my_doc_manager.py

from mongo_connector.compat import u
from mongo_connector.doc_managers.doc_manager_base import DocManagerBase


class DocManager(DocManagerBase):

    def __init__(self, url=None, **kwargs):
        pass

    def stop(self):
        """Stop the auto-commit thread."""
        self.auto_commit_interval = None

    def update(self, document_id, update_spec, namespace, timestamp):
        dispatch(namespace, document_id, 'update', timestamp)

    def upsert(self, doc, namespace, timestamp):
        doc_id = doc.pop("_id")
        dispatch(namespace, doc_id, 'add', timestamp)

    def remove(self, document_id, namespace, timestamp):
        dispatch(namespace, document_id, 'delete', timestamp)


def dispatch(namespace, document_id, action, timestamp):
    """
    Dispatcher oplog
    :param namespace:
    :param document_id:
    :param action:
    :param timestamp:
    :return:
    """
    print(namespace, document_id, action, timestamp)

scripts.md

Create Oplog Process File

• mongo-connector script

mongo-connector -m mongodb://localhost:27017 -t http://localhost:9200 -n oplog.events -d my_doc_manager --batch-size 100

• insert some data into oplog.events

mongo
use oplog
db.events.insert([{},{},{}])

• mongo-connector script output

oplog.events 56e904c01fc64deebbdefb20 add 6262541979515617281
oplog.events 56e904c01fc64deebbdefb21 add 6262541979515617282
oplog.events 56e904c01fc64deebbdefb22 add 6262541979515617283

• cat oplog.timestamp

cat oplog.timestamp
["Collection(Database(MongoClient('localhost', 27017), 'local'), 'oplog.rs')", 6262541979515617283]

• stop mongo-connector script
• modify oplog.timestamp to specify operation point: 6262541979515617283 ==> 6262541979515617281

cat oplog.timestamp
["Collection(Database(MongoClient('localhost', 27017), 'local'), 'oplog.rs')", 6262541979515617281]

• start mongo-connector script, and do nothing with oplog.events
• mongo-connector output, since it continues after 6262541979515617281

oplog.events 56e904c01fc64deebbdefb21 add 6262541979515617282
oplog.events 56e904c01fc64deebbdefb22 add 6262541979515617283

• cat oplog.timestamp

cat oplog.timestamp
["Collection(Database(MongoClient('localhost', 27017), 'local'), 'oplog.rs')", 6262541979515617283]

OplogThread: Last entry no longer in oplog cannot recover!

• stop mongo-connector script
• modify oplog.timestamp to specify operation point: 6262541979515617283 ==> 5262541979515617281 (a ts shorter than 6262541979515617281)

cat oplog.timestamp
["Collection(Database(MongoClient('localhost', 27017), 'local'), 'oplog.rs')", 5262541979515617281]

• start mongo-connector script, and do nothing with oplog.events
• mongo-connector will exit, since 5262541979515617281 could not found.

cat mongo-connector.log
2016-03-16 14:59:06,358 [ERROR] mongo_connector.util:87 - Fatal Exception
2016-03-16 16:07:29,860 [ERROR] mongo_connector.oplog_manager:162 - OplogThread: Last entry no longer in oplog cannot recover! Collection(Database(MongoClient('localhost', 27017), 'local'), 'oplog.rs')
2016-03-16 16:07:30,227 [ERROR] mongo_connector.connector:302 - MongoConnector: OplogThread <OplogThread(Thread-2, stopped 123145312813056)> unexpectedly stopped! Shutting down

Reference

• Oplog Progress File · mongodb-labs/mongo-connector Wiki