
After [setting up](/real-time-search-with-mongodb-and-elasticsearch/) ElasticSearch, you'll be faced with the task of optimizing your index configuration for speed and for size. There are millions of documents in our index, and, for performance, it’s important that all of that be kept in memory. As a result, index size is pretty important. We spent a while tweaking ElasticSearch to minimize its index size, and it turns out that there’s a decent number of low-hanging fruit.

## _source

For example, ElasticSearch [stores](http://www.elasticsearch.org/guide/reference/mapping/source-field.html) the original data along with the indexed data for each document and returns the full document with each search result. We just wanted ElasticSearch queries to assemble a set of document IDs to be fetched from the database, so this overhead is unnecessary. Cutting this data shrank our search index four-fold.

```rb
:_source => { :enabled => false }
```

## _all

In addition, ElasticSearch stores an aggregated [`_all`](http://www.elasticsearch.org/guide/reference/mapping/all-field.html) field on each document, which contains the analyzed output from all of the other fields in the document. This doesn’t add any new information, and its purpose is just to simplify the query interface.

We don’t need to be able to query all fields (for example, we only use user IDs to partition the index); setting a flag to exclude these from the `_all` field and preventing them from being analyzed saved us another 4-5GB for ~2 million documents.

```rb
'id' => { :type => 'string', :include_in_all => false }
```

## Multi-field types

Other problems were a little trickier. How do we minimize the size of the email address index given that we’d like to be able to perform both prefix and substring searches on them?

Email addresses may be very long and overflow our ngram tokenizer (which maxes out at fifteen characters), and so we decided to construct a [multi-field](http://www.elasticsearch.org/guide/reference/mapping/multi-field-type.html) tiered index for certain fields in to accommodate all of the search use cases. We don't know if this is a best practice, but it seems to work pretty well for us.


Elasticsearch Space-savers


Something I worked on a couple of weeks ago at Stripe was overhauling the entire search infrastructure. If you’ve ever used the search feature in manage, it may have appeared sluggish or may have even timed out on you - this is mostly due to the fact that our search consisted of running a query over our database (which you can imagine is very slow for full-text search).

We finally decided to make the switch to a dedicated full-text search engine and chose ElasticSearch, which is a distributed RESTful search engine built on top of Apache Lucene. I won’t go into the details, but you can read about ElasticSearch on the [official website](http://www.elasticsearch.org/).

## Setting up ElasticSearch

To set up ElasticSearch, I followed a [third-party tutorial](http://www.elasticsearch.org/tutorials/2010/07/01/setting-up-elasticsearch.html) - it’s pretty straight forward, although you should be sure to swap in the current verison number.

## Hooking up MongoDB

Assuming you’re already using MongoDB for the data that you want to index^1, you’re probably wondering how to hook it up to index documents in real-time - this is one of the major hurdles that we faced. ElasticSearch has a built in feature of Rivers, which are essentially plugins for specific services to constantly stream in new updates for indexing.

Unfortunately, there’s no MongoDB River (probably due to the lack of built-in database triggers), so I did some research and realized that I could use the MongoDB oplog to continually capture updates to our main databases.

The [oplog](http://www.mongodb.org/display/DOCS/Replica+Sets+-+Oplog) is a special collection held by MongoDB for replication purposes - I was able to make use of a tailable cursor (which performs similarly to the `tail -f` command) for shipping new updates from MongoDB to our search server (we’re using [Tire](https://github.com/karmi/tire) as our Elasticsearch Ruby DSL).

```rb
def initialize
  @host = ENV['MONGO_RUBY_DRIVER_HOST'] || 'localhost'
  @port = ENV['MONGO_RUBY_DRIVER_PORT'] || Connection::DEFAULT_PORT

  @db_whitelist = %w(payments customers invoices)

  @stream = Queue.new
  @max_bulk = 10000

  # Or, just set up a connection the normal way if you’re not
  # using a repl set
  @connection = ReplSetConnection.new(['database1', 27017],
                                      ['database2', 27017],
                                      ['database3', 27017],
                                      :read => :secondary)

  local = @connection.db('local')

  # Change to oplog.$main if running mongod as a master instance
  oplog = local.collection('oplog.rs')

  @tail = Cursor.new(oplog,
                      :tailable => true,
                      :order => [['$natural', 1]])

  while true
    tailer
    indexer

    # In production, we keep this constantly running in a busy
    # loop; we constantly have new things to index
    # sleep 5
  end
end
```

The code here basically tails the oplog and reads off new operations, pushing them onto `@stream`, until there are none left or we’ve reached the max number of entries that we want to batch at a time:

```rb
def tailer
  # While there are new entries
  while @tail.has_next?
    _next = @tail.next

    # We only want to index specific collections
    match = /some namespace regex/.match(_next["ns"])
    if match && @db_whitelist.include?(match[1])
      @stream << [match[1], _next]
    end

    # Upon hitting the maximum bulk size we want to send over
    break if @stream.length >= @max_bulk
  end
end
```

Afterwards, we pull rows out of `@stream`, transform them into hashes that we actually want to index into Elasticsearch, and bulk index them via Tire. Rinse and repeat.

```rb
def indexer
  bulk_index = []
  while !@stream.empty?
    _next = @stream.pop
    item = _next[1]
    type = _next[0]

    # Our TransformRow.transform_for method just returns a hash
    # to index. Ex. {:type => “customer”, :name => “Amber”}
    new_row = TransformRow.transform_for(type, item['o'])

    # Add keywords if they exist on the database object
    bulk_index << new_row
  end

  # Bulk index the data via Tire
  if !bulk_index.empty?
    Tire.index "search" do
      import bulk_index
    end
  end
end
```

Of course, this is a very basic synchronous example - you could imagine making this into a more flexible asynchronous producer-consumer model.

In addition, this sort of naive first pass doesn’t save any state, so if the driver crashes due to an timeout on the database side or if the search server is down, the restarted driver will start again from the very beginning. To combat this, I simply wrote a timestamp to `/tmp` to keep track of the last record indexed, and queried the collection to start at that specific timestamp the next time around.

(**Update**: I recently changed this to write to permanent storage instead; although not necessary, it's simple enough to do and is a better idea in general.)

```rb
@tail = oplog.find({
  'ts' => {'$gte'=> BSON::Timestamp.new(@last_ts, 0) }
})
@tail.add_option(Constants::OP_QUERY_TAILABLE)
```

But wait, there’s more! Scanning the entire collection in most restart scenarios can be pretty expensive (by the way, the oplog doesn’t allow any indexes), especially if your oplog is large like ours, at 2GB.

You can take advantage of a special flag for this kind of use only, called `OPLOG_REPLAY`. This flag is meant for replication purposes - it first starts from the bottom of the log and searches upward for around 100MB or so, assuming that you just restarted (which is usually the case if the tailer crashes and restarts immediately). If it can’t be found, it’ll continue by traversing extents in the most efficient manner possible.

```rb
@tail.add_option(Constants::OP_QUERY_OPLOG_REPLAY)
```

Running this tailer as a daemontools service, we were able to attain nearly real-time search latency. It takes around 3-5 seconds from the time a record is created or updated to when it gets indexed (and is searchable) in ElasticSearch - pretty cool!

####Footnotes
1. It’s possible to use ElasticSearch as a data store itself (it’ll store the document data as well as the index data), but I’m not convinced this is a good idea in practice.



Real-time Search with MongoDB and Elasticsearch


A [couple of posts ago](/eventual-consistency/), I talked about the distributed systems programming class I was taking here in my last semester at Berkeley. Our projects are really cool - we've done everything from a quorum KVS to a distributed lock manager, and for our final project, my group chose to implement Multi-paxos. I'll start by explaining general Paxos first. 

Basically, Paxos is an protocol (or technically, a "family" of protocols, says [Wikipedia](http://en.wikipedia.org/wiki/Paxos_(computer_science)) that solves the problem of getting a group of nodes to agree, or reach consensus, on a single value.

## Why is this important?

If there's a stable master, there's hardly any use for this, because the master can just decide and propose values itself and have the other nodes always follow it. However, what if leader election fails (and it sometimes does), and there are two nodes proposing values?

The Paxos algorithm ensures that only one value is chosen out of all of the proposed values. Where this algorithm comes in handy is when you'd like your distributed system to decide on a sequence of values, or in other words, a consistent ordering over a set of values. As [Ayende Rahien](http://ayende.com/blog/4496/paxos-enlightment) describes, if these values are events, then you know that all of the machines in the cluster have the same state or subset of the state.

## Algorithm

The Paxos algorithm is divided into two phases, the _prepare_ phase, and the _accept_ phase. 

**Disclaimer**: the text in this section is partially paraphrased from Leslie Lamport's excellent [Paxos Made Simple](http://research.microsoft.com/en-us/um/people/lamport/pubs/paxos- simple.pdf) paper.

###Prepare Phase

In the _prepare_ phase, the proposer selects a proposal number `n` (which is greater than any `n` it has previously sent, and distinct from any `n` that any other proposers' `n` values, I'll discuss this more later) and sends a `PREPARE` request to a majority of the acceptors.

On the acceptor side, if it ever receives a `PREPARE` request with a proposal number `n` greater than that of any other proposal number `n` that it has previously received, it responds with a "promise" to not accept any proposals with a lower numbered `n` in the future and the value of the highest-numbered proposal that it has already accepted (if any).

Starting to sound confusing? I had to read this over several times before I actually understood it, so maybe the pseudocode below will help.
    
```rb
# Proposer
for acceptor in acceptors
  send :prepare_req, next_n()

# Acceptor
if (req.n > highest_proposed_n)
  highest_proposed_n = req.n
  reply :prepare_resp, {
    :n => highest_acc.n,
    :value => highest_acc.value
  }
```

###Accept Phase

After a proposer has received a response to its `PREPARE` requests from a majority of the acceptors, it then sends an `ACCEPT` to those acceptors with a value `v`, where `v` is the value of the highest numbered proposal among the responses, or any value if the responses reported no other proposals.

By contrast, when an acceptor receives an `ACCEPT` request, it always accepts it unless it has already promised not to in the _prepare_ phase. 
    
```rb
# Proposer
for acceptor in acceptors
  send :accept_req, responses.argmax { |i| i.n }.value }

# Acceptor
if (req.n >= highest_proposed_n)
  highest_acc = {:n => req.n, :value => req.value}
  reply :accept_resp
```
    
## Multi-paxos

All this talk about Paxos. How does Multi-paxos fit in? 

I mentioned in the introduction that one of the main useful applications of the Paxos application is having the group of participants decide on a sequence of numbers. Since one round of Paxos results in a decision of one value, the naive way to go about finding a sequence of numbers would be to run Paxos many times, right?

One optimization that can be made in this case, assuming a single stable leader, is to skip the _prepare_ phase. If we assume that the leadership will remain "sticky", there is no need to continue sending out proposal numbers - the first proposal number sent out will never be "overridden" since there is only one leader.

Thus, we only need to do the _prepare_ phase once. In subsequent rounds of Paxos, we can just send the `ACCEPT` messages, with `n` as the proposal number used in the original `PREPARE` request and an additional parameter that indicates the sequence number (the current round we're in). We don't have to worry about the worst case where leadership isn't stable (or distinct), because the algorithm will degrade gracefully into the general Paxos algorithm (both _prepare_ and _accept_ phases for each round). Cool!

## Other Considerations

###Persistent Storage

Machines must keep certain things in persistent storage to be able to recover from failures. In particular, acceptors need to be able to remember which `PREPARE` requests they have promised to follow, and which `ACCEPT` requests they have responded to in order to make decisions about which proposals to promise to and to pass necessary information back to the proposes. 

###Distinct Proposal Numbers `n`
In order for the algorithm to work, each proposer must propose monotonically increasing proposal numbers `n` that are distinct from any other proposers' numbers `n`.

To achieve this, we can assign disjoint sets of numbers for each proposer and have them only choose numbers from their own set (ex. assign each node a unique prime number, they choose multiples of that prime number).

Or, if we're assuming static membership of participants, assign each node a unique number `i` between `0` and `k`, where `k` is the total number of participants, and `n = i + (k * round_number)`. 

## Conclusion

**tl;dr** Paxos is for getting a group of machines to agree on a value (or more usefully, provide a consistent ordering on a sequence of values). Multi-paxos is an optimization on using Paxos for consecutive rounds, where you can skip one of the phases if you assume a stable leader.

Hope this post was interesting and/or informative! Similar to the [eventual consistency](/eventual-consistency/) post, I don't claim to be an expert in this area, so I'd love to hear any comments or corrections. 



Paxos/Multi-paxos Algorithm


I mentioned in my previous post that I was taking a [Programming in the Cloud](http://db.cs.berkeley.edu/w/cs-194-17-programming-the-cloud/) class on distributed systems and programming this semester, and one of the perks of this class is that we occasionally have speakers from the industry come in to talk about various topics. 

Last week, [Doug Terry](http://research.microsoft.com/en-us/people/terry/) from Microsoft Research came in to speak about eventual consistency, which I found super, super interesting. I don't claim to be an expert on the topic now, but here are my notes (more or less from memory). First, the two opposite sides of the spectrum:

## Strong vs. Eventual

#### Strong Consistency -  guaranteed consistency across replicas
Strong consistency guarantees, well, strong consistency. Any read always returns up-to-date data, although this may result in lower availability since you're limiting the machines that you're able to read from to keep this guarantee. 

#### Eventual Consistency - all replicas will "eventually" become consistent.
Although this weaker form of consistency has better availability because you can read to/write from any machine, you can only hope that the data will be fresh (for some applications, this is enough).

## And everything in between

This isn't an exhaustive list of consistency models, but a couple of other models in the middle of the spectrum that Doug went over that day:

#### Read My Writes - guaranteed to be able to read updates you write
In applications where you're doing some sort of read then update operations (like `x+=1`), it's _very_ important to be able to your own writes. The example Doug gave during lecture was a baseball analogy, where a scorekeeper is keeping track of the current score of the game with operations like `homeScore +=1` for every home run scored for the home team. It's a different story if there are multiple scorekeepers, where you might need some causal consistency guarantees, but a single scorekeeper must at minimum be able to read his own writes. 

Afterwards in class, we had a discussion about the various ways that this consistency model could be implemented. There needs to be some sort of versioning system for the clients to be able to compare with the machines to determine if their writes are stored there or not, or as an optimization, the client could store the writes locally and always be able to "read their own writes" if they take the union of their write set and any other machines' write set. 

####Consistent Prefix - updates returned are some prefix of all the updates, with no gaps
In case the title didn't make sense, a "consistent prefix" means that some prefix of all of the writes (assuming the writes have a global order) are applied such that the prefix is consistent ("up to date" up until the last write in the prefix, with no gaps). 

For example, if the writes are A, B, C, D, E, F, you will be able to read some consecutive set of these writes, starting with A. Note that the empty set is also trivially a consistent prefix. 

Why is this useful? Following the baseball analogy from before, if you are recording score writes of `A: 1`, `B: 2`, `A: 3`, `B: 6`, `A: 5`, `B:7` and you don't have a consistent prefix guarantee, you could be reading a database that only has the third and last writes applied, giving you `A: 3` and `B: 7` when in reality, that combination of scores never existed. 

#### Monotonic Reads - for each client, guaranteed to read a more recent or at least the same set of updates
Monotonic reads guarantees that you'll never "go back in time" in terms of seeing updates - you'll always see a set of updates that is at least as new as the updates you saw the last time you read from the system. The implementation of this requires a client to keep some kind of version of when it performed its last read, and search for a machine that has at least those writes. However, like Read My Writes, we can apply the same optimization of keeping a read set on the client (and performing a union whenever reading from the system). 

Since we're only concerned with reading data that _at least_ as fresh as the last time we read it, we can also introduce a notion of "stickiness" or "affinity", where a client stays with the last machine it read (until it goes down, at least) instead of re-searching for another machine that is more up-to-date. Assuming a single machine will never lose any updates it has applied, staying with the same machine is trivially monotonic.

#### Bounded Staleness
Last one! Hopefully you find this fascinating (like I do), and not boring. The idea of staleness allows consistency to be defined in terms of how old a replica can be compared to the most up-to-date data. In some applications, it's okay to have the data be a little stale - for example, weather reports can be a couple hours old. Strong consistency is a special case of Bounded Staleness where the amount of staleness is none. 

## Conclusion
Wow, this blog post took longer than I thought it would to write. I'm really interested in these topics (and databases/distributed systems in general), so I'm pretty excited for what the class has in store for us for the rest of the semester! 

Again, I don't claim to be an expert in this field (and most of these notes are from memory), so feel free to point out any mistakes/incorrect assumptions or start a discussion in the comments. Thanks for reading! (:



←

systems

Elasticsearch Space-savers

Real-time Search with MongoDB and Elasticsearch

Paxos/Multi-paxos Algorithm

Eventual Consistency