stripe-archive / mongoriver

A library for writing MongoDB oplog tailers.

This project is deprecated and is no longer being actively maintained.

Mongoriver

mongoriver is a library to monitor updates to your Mongo databases in near-realtime. It provides a simple interface for you to take actions when records are inserted, removed, or updated.

How it works

MongoDB has an oplog, a log of all write operations. mongoriver monitors updates to this oplog. See the Mongo documentation for its oplog for more.

How to use it

Step 1: Create an outlet

You'll need to write a class subclassing Mongoriver::AbstractOutlet.

You can override any of these methods:

• update_optime(timestamp)
• insert(db_name, collection_name, document)
• remove(db_name, collection_name, document)
• update(db_name, collection_name, selector, update)
• create_index(db_name, collection_name, index_key, options)
• drop_index(db_name, collection_name, index_name)
• create_collection(db_name, collection_name, options)
• drop_collection(db_name, collection_name)
• rename_collection(db_name, old_collection_name, new_collection_name)
• drop_database(db_name)

You should think of these methods like callbacks -- if you want to do something every time a document is inserted into the Mongo database, override the insert method. You don't need to override all the methods -- if you only want to take action on insert and update, just override insert and update.

Step 2: Create a stream and start the logger

Once you've written your class, you can start tailing the Mongo oplog! Here's the code you'll need to use:

mongo = Mongo::MongoClient.from_uri(mongo_uri)
tailer = Mongoriver::Tailer.new([mongo], :existing)
outlet = YourOutlet.new(your_params) # Your subclass of Mongoriver::AbstractOutlet here
stream = Mongoriver::Stream.new(tailer, outlet)
stream.run_forever(starting_timestamp)

starting_timestamp here is the time you want the tailer to start at. We use this to resume interrupted tailers so that no information is lost.

Version history

0.4

Add support for tokumx. Backwards incompatible changes to persistent tailers to accomodate that. See UPGRADING.md.

mongoriver's Issues

Update README with Usage

Hi, this seems like a really cool project however it would be great if you explain what it is, what it does and how to get it running/integrated in readme file or a separate wiki page.

Support for Mongo driver >= 2.0

Please correct me if I am wrong, but does the current gem support Mongo ruby driver v2 and above ? I had to downgrade it to 1.9.2 for it to work.

stream throws unhandled exception for sparse indexes

With a standard setup of Mongo and a stream that uses a Mongoriver::Tailer and empty sub-class of AbstractOutlet, the creation of certain indexes crashes the stream.

Implemented code:

mongo = Mongo::MongoClient.new(host,port)
mongo.db("admin").authenticate(username,password) # mongo has auth enabled
tailer = Mongoriver::Tailer.new([mongo], :existing)
output = Mongoriver::AbstractOutlet.new()
stream = Mongoriver::Stream.new(tailer,outlet)
stream.run_forever()

Here is the entry in the oplog causing the crash (it's the creation of a sparse index)

my_rs:PRIMARY> use local
my_rs:PRIMARY> db.oplog.rs.find({"ns","my_db.system.indexes"})
{ "ts" : Timestamp(1404818603000, 25), "h" : NumberLong("-6393196251659512871"), "v" : 2, "op" : "i", "ns" : "my_db.system.indexes", "o" : { "name" : "pid.myGroups.primary_1_sparse_", "ns" : "my_db.product", "key" : { "pid.myGroups.primary" : 1, "sparse" : true } } }

And it crashes with the following:

lib/mongoriver/stream.rb:121:in `block in handle_create_index': undefined method `round' for true:TrueClass (NoMethodError)
        from lib/mongoriver/stream.rb:118:in `each'
        from lib/mongoriver/stream.rb:118:in `map'
        from lib/mongoriver/stream.rb:118:in `handle_create_index'
        from lib/mongoriver/stream.rb:109:in `handle_insert'
        from lib/mongoriver/stream.rb:87:in `handle_op'
        from lib/mongoriver/stream.rb:33:in `block in run_forever'
        from lib/mongoriver/tailer.rb:101:in `stream'
        from lib/mongoriver/stream.rb:32:in `run_forever'

Mongo stores the keys from the creation of an index in the oplog using the sytax of "key":1 for fields generally, but uses a syntax of "sparse": true for sparse indexes. L117 of stream.rb assumes a numeric value, and always does a dir.round, which crashes for true.round.

Is this repo still maintained?

@den-stripe @brandur-stripe

persistent tailer

Is there an example of how to use the persistent tailer?

I'm trying to use and I've got it creating the collection to store the tail, however, it only updates the tail when I start tailing the ops log. It does not appear to be writing to the trail collection while tailing.

Here's how I've written it so far... Thanks!

client = Mongo::MongoClient.from_uri([DATABASE STUFF])
outlet = Mongoriver::OplogWatcher.new
config = client['admin'].command(:ismaster => 1)
persistent_tailer = Mongoriver::PersistentTailer.new([client], :existing, config['setName'], {:db => database, :save_frequency => 15})

stream = Mongoriver::Stream.new(persistent_tailer, outlet)

stream.run_forever()

Handle `applyOps` somehow?

Mongo's applyOps command generates weird oplog entries. We should consider transparently supporting those somehow, or, at a minimum, Streamer should support them:

nelhage:PRIMARY> db.getSiblingDB('admin').runCommand({applyOps: [{
... "ts" : {
... "t" : 1409704895,
... "i" : 2
... },
... "h" : NumberLong("1547919913461096899"),
... "v" : 2,
... "op" : "i",
... "ns" : "nelhage.nelhage",
... "o" : {
... "_id" : ObjectId("540663be6f3242b06b9c3101"),
... "i" : 2
... }
... }]})
{ "applied" : 1, "results" : [ true ], "ok" : 1 }
nelhage:PRIMARY> db.oplog.rs.find().sort({$natural:-1}).limit(1).next()
{
    "ts" : {
        "t" : 1409704955,
        "i" : 1
    },
    "h" : NumberLong("8371608664109460151"),
    "v" : 2,
    "op" : "c",
    "ns" : "admin.$cmd",
    "o" : {
        "applyOps" : [
            {
                "ts" : {
                    "t" : 1409704895,
                    "i" : 2
                },
                "h" : NumberLong("1547919913461096899"),
                "v" : 2,
                "op" : "i",
                "ns" : "nelhage.nelhage",
                "o" : {
                    "_id" : ObjectId("540663be6f3242b06b9c3101"),
                    "i" : 2
                }
            }
        ]
    }
}