The System of Record Approach to Multi-Master Database Applications

Multi-master database systems that span sites are an increasingly common requirement in business applications.  Yet the way such applications work in practice is not quite what you would think from accounts of NoSQL systems like Cassandra or SQL-based systems like Oracle RAC.  In this article I would like to introduce a versatile design pattern for multi-master SQL applications in which individual schemas are updated in a single location only but may have many copies elsewhere both locally as well as on other sites.  This pattern is known as a system of record architecture.  You can build it with off-the-shelf MySQL and master/slave replication.

Let's start by picking a representative software-as-a-service (SaaS) application:  call center automation.    Call center software integrates with a local PBX or VOIP to allow agents to answer and make phone calls for telemarketing campaigns in a systematic and automated way using standard procedures known as "agent scripts." Admins set up agent scripts and define lists of people to call as well as marketing campaigns.  Finally and perhaps most importantly, managers receive a wide variety of detailed reports that allow them to optimize current work, examine historical performance, and make predictions about the future for planning purposes.  Here is a typical application architecture.

Figure 1:   Call Center Application Architecture

Bear in mind that this is a greatly simplified view.  Like most business applications, call center automation systems may contain hundreds of database tables and many types of user services.  There are also practical complications that go beyond the application itself.    Call center automation is vital to the businesses that use it.  Customers want assurance they can continue processing on another site if a SaaS vendor site goes dark.  This means we have to think about maintaining applications and data on multiple sites.

The ideal solution for most SaaS vendors would be to have call center data and applications for all customers live on multiple sites at all times.  Multiple live sites mean that failover is instantaneous since both applications and database servers are already up and running.  Constant update means there is little or no data loss on failure. Customers could connect to the nearest site.  Here is a picture of that dream that includes two sites and two customers, Acme Inc. and Pinnacle, Ltd.

Figure 2:  Dream Architecture for Call Center Automation

This solution has only one problem.  It is impossible to build.  Readers may nod wisely and say this is because of CAP Theorem limitations, but that somewhat misses the point.  Let's say we use a NoSQL DBMS like Cassandra that permits updates in multiple locations and reconciles the data using eventual consistency.   However, there's a catch:  as we saw above, much of the value of call center automation is in operational queries and reports.  That drives you back to an RDBMS with cross-table joins, aggregation functions, referential integrity, and convenient SQL-based report writing tools.  For this reason alone, Cassandra is a non-starter for call center automation.

What about a SQL DBMS?  MySQL obviously has all the features you need for query-intensive solutions on smallish data sets (e.g. hundreds of millions of rows, not many billions or trillions).  The problem is multi-master replication.   Updating the same table from two or more places on a LAN is already quite difficult:  witness the complexity of Oracle RAC or MySQL Cluster.  The problem becomes intractable when you combine complex SQL transactions, referential integrity, and high-latency WAN connections.  If you want full SQL semantics you cannot have updates on multiple sites.   This is a serious dilemma and not just for call center automation.  The same problems or worse affect a multitude of valuable business applications including market automation, credit card processing, customer relationship management (CRM), time/expense tracking, accounting, and many others.

Fortunately we are not really stuck.  If we give up some requirements customers do not really want anyway, there is a perfectly good solution that will work for a wide range of problems.  Data warehousing architects long ago developed the notion of a system of record.  Bill Inmon's classic Building the Data Warehouse defined system of record as follows:

The definitive and singular source of operational data.  If data element abc has a value of 25 in a database record but a value of 45 in the system of record, by definition the first value is incorrect and must be reconciled.  

System of record applies to multi-master systems in the form of a simple rule.  We just assert that every customer has master data in one and only one location and copies everywhere else.  When particular customers update information they do so on their own master.  Customers can have masters on different hosts or sites, but the system of record rule says that no customer has one in two places.  This eliminates conflicts between masters, and multi-master replication now works without a lot of difficulty.

System of record thus meets the original requirement of having data on multiple sites, which was to handle a site failure.  We can store data economically using off-the-shelf MySQL.  We can update copies within and across sites using master/slave replication.  We can shard customer data into independent schemas. The result looks like the following.  Acme has a master in San Jose, whereas Pinnacle has its master in New York.

System of Record Architecture for Call Center Automation

Using the system of record approach simplifies other problems as well.  Standard backup and restore techniques still work.  If you mess up a customer copy, you re-provision from the master shard.  You can implement failover across sites and also fail over locally onto slave copies, which can be complete copies containing data for all customers.

Meanwhile, most users are fine with a single site.  Pinnacle is close to New York, which is why the SaaS vendor puts Pinnacle's data there and gives them the New York site DNS for login.   It is also possible to run reports on the cross-site copies as well.  You can even run full applications provided you forward writes to the system of record, as shown above for Acme.

The real issue in implementing system of record architectures is that existing replication and clustering tools are not quite up to the job of handling cross-site applications build on system of record.  We are extending Tungsten to handle some of the obvious problems in building these types of systems using MySQL.

1. Locating the customer master and connecting applications to it. 
2. Moving the customer master from one location to another.  This happens more often than you would think, for example to minimize multi-master replication which can introduce problems beyond conflicts. 
3. Detecting accidental updates to copies and preventing them from either reaching the DBMS and/or preventing them from propagating to other locations.  
4. Proving a clean failover model that works on both cross-site as well as local copies of data.  
5. Recovering corrupted copies of customer data from the master.  

I will discuss two of the upcoming Tungsten features in follow-up articles.  The first is assigning a shard master in the Tungsten Replicator using the new shard API.  The shard API enables multi-master but enforces system of record constraints to avoid messing up data should you accidentally update in the wrong location.  The second feature is cross-site management and connectivity using Tungsten Enterprise.  This handles failovers within and between sites and automatically connects applications to the active master regardless of which site or DBMS it lives on.

The need for availability is pushing an increasing number of SaaS vendors and other application providers to operate systems across multiple sites.  Applications like call center automation depend on the features of SQL and cannot be implemented using NoSQL DBMS's like Cassandra.  The system of record architecture eliminates replication conflicts and enables multi-master updates to work on ordinary SQL databases between sites.  If you are building complex SQL applications and thinking about going multi-site, this design pattern should be in your toolbox.  

So Where's the Fall MySQL Community Conference?

Last week Percona announced plans to sponsor the Percona MySQL Conference in Santa Clara in April 2012.  It is meant to replace the O'Reilly conferences of previous years.  The announcement led to some reasonable questions, for example from Giuseppe Maxia.  These and other online posts initiated a thoughtful exchange of views about the pros and cons of Percona's conference announcement by various members of the MySQL community.

Not everyone agrees with what Percona is doing.  However, you have to give Peter, Baron, and others at Percona credit for taking the risk to organize a replacement conference.  It's a big financial commitment to rent space in Santa Clara.  The conference will be a huge amount of work, much of it quite thankless.  Even so I imagine it might be tempting for some people to try to organize an anti-conference just to spite Percona, much as Percona did at the MySQL Conference in 2009.  This would be a mistake.

Here is a better suggestion.  Why not put the energy into organizing a Fall 2012 MySQL Community Conference in Europe?   Make it somewhere pleasant like Barcelona or Verona and everybody will want to go.  Two solid conferences in different locations at opposite ends of the calendar would benefit the MySQL community at all levels and make it possible for more people to attend.  For those of you who want to grow the community, it's time to stand up and be counted.  Let's get another conference off the ground.

Adding Parallel Replication to MySQL in a Hurry

A previous article on this blog described Tungsten parallel replication using on-disk queues.  On-disk queues are now more or less finished, and I just closed the covering issue for the feature.  The work is bug fixing and performance testing from here on out.  Speaking of performance, that looks fairly good.   A recent on-site test using production workloads showed 3.3X improvement over native MySQL replication while holding resources like memory down to much more reasonable levels than in-memory queues.  We have further optimizations on the way, so this should improve.

Now that parallel replication is working a lot better, what is it good for?  Here is a good start:  assuming your workload is suitable for shard-based replication, Tungsten offers a nice replacement for MySQL slave replication that will give you immediate parallel replication for any MySQL server starting with version 5.0 and up.  We call this native slave takeover.  It looks like the following diagram.

You can migrate MySQL slaves to Tungsten very easily thanks to some great work on the Tungsten installer by Jeff Mace that Giuseppe Maxia also praised in a recent article.  Seeing that Giuseppe is also the Continuent QA director, I guess this means it passed.  :)   Here is how to set up.

First, configure MySQL master/slave replication.  Here are typical commands after you synchronize the slave with a backup from the master.  For these examples, we assume the slave is running on host mercury, and the master is on host saturn.

mysql -utungsten -psecret -hmercury -e "CHANGE MASTER TO \
  MASTER_HOST='saturn', \
  MASTER_USER='repl', \
  MASTER_PASSWORD='s3cr3t', \
  MASTER_LOG_FILE='mysql-bin.000338', \
  MASTER_LOG_POS=4534"
mysql -utungsten -psecret -hmercury -e "START SLAVE"
mysql -utungsten -psecret -hmercury -e "SHOW SLAVE STATUS\G"

Next, download and unpack Tungsten from the latest dev build.  Here are sample commands for build 231.  

wget --no-check-certificate https://s3.amazonaws.com/files.continuent.com/builds/nightly/tungsten-2.0-snapshots/tungsten-replicator-2.0.4-231.tar.gz
tar -xvzf tungsten-replicator-2.0.4-231.tar.gz
cd tungsten-replicator-2.0.4-231

Finally, run the installation using the following handy command.  This will copy the download into its proper location (here /opt/tungsten) and start Tungsten.  Tungsten will stop MySQL replication if it is running and start Tungsten replication from the exact point where native replication left off.
tools/tungsten-installer \

  --direct  \
  --native-slave-takeover \

  --master-host=saturn  \

  --master-user=repl  \

  --master-password=s3cr3t  \

  --slave-host=mercury \

  --slave-user=tungsten  \

  --slave-password=secret  \

  --service-name=takeover \

  --home-directory=/opt/continuent \

  --svc-parallelization-type=disk \

  --channels=10 \

  --start-and-report

If there is a problem with your pre-requisites the installer will print out a message and stop.  Most of these messages should be pretty self-explanatory.   (In some builds there's a problem with OpenJDK failing the install check.  Use --no-validation to get around that.) 

Here is another interesting tip.  If you try Tungsten and want to go back to MySQL replication, you can flip back to MySQL native replication with two commands.  Just take Tungsten offline cleanly, which shuts down all the replicator channels at the same point, and start MySQL replication again: 

trepctl offline

mysql -urepl -ps3cr3t -e 'START SLAVE'

We are currently doing a lot of performance profiling and testing of parallel replication.  I hope to post more about performance results in a future article.  Meanwhile, there are 36 new features (including on-disk queues) and bug fixes that will roll into a final Tungsten 2.0.4 in the next week or so.   Try out the latest builds and see what you think. 

Mobile Internet Access in Germany for Open Source Road Warriors

Reliable Internet access is a long-standing problem for road warriors visiting foreign countries.  Open source developers in particular have problems reconciling travel with addiction to high-bandwidth network access from laptop computers.  Wi-Fi hotspots are scarce, costly, often slow, and in some cases complicated by inconvenient local laws like Italy's Pisanu Decree.  International mobile network access plans are ridiculously expensive or like DROAM have download limits that make them useless for serious programming.

The best solution in many cases is to look for a local pre-paid mobile access plan in each country you visit.   Mobile networks are widely available and fast in developed regions, and there are cheap plans that limit the amount you pay while providing solid connectivity.  On a recent trip to Germany I found a great solution for local Internet access from FONIC, which offers a prepaid data plan using their FONIC Surf-Stick.  The Surf-Stick is a USB modem that plugs into a USB port on your laptop and looks like the following. 

After you buy the Surf Stick, you can add money to it using credits purchased in local stores.  It's a relatively simple solution provided you understand a little about networking and can work your way through a bit of German.

The rest of this article is a description of how to use FONIC in Germany, as well as a review of the performance and a couple of downsides.  I do not have any connection with FONIC other than using their products.  You may have different experiences or find something better.  If so, please write an article about it.

Getting Started

I bought got my Surf Stick for 49 Euro at Saturn, a big German consumer electronics chain.  You get the modem itself plus a SIM card and one day of free surfing.  You can get also the stick from other FONIC partners or order it off the FONIC website.  Surprisingly it does not appear to be available in places like Frankfurt or Munich Airport.

The first step once you have the modem is to initialize the SIM on FONIC's site by registering yourself as a new user.  To do this you'll need to have your FONIC phone number and activation code ("Freischaltungsnummer"), which are written on the side of the envelope that contains the SIM and that you should try to avoid losing.  You'll also need a German address that you enter as part of the sign-up.  A hotel address is fine.  It takes a couple of hours after registration to complete provisioning.  

Once provisioned, connecting is easy.  Put the SIM in the surf stick and plug the stick in a USB port on your computer.  On Mac OS X, you'll see an application called "Mobile Partner."  Here are the steps to activate a connection: 

1. Click the Mobile Partner app.  
2. Enter your PIN (also on the SIM envelope) and press Return. 
3. The application will stall for a few seconds while it looks for the modem.  Once complete it will show a screen entitled "Verbinden" (connect).   
4. Click Verbinden.  A couple of pop-ups will appear and you are online. 
5. When you want to drop the connection click "Trennen" (disconnect).  

You will now need to load some money into your account.  There's no way to top up online with a credit card--you either have to register a German bank account or purchase credits.  For foreigners the only practical thing is therefore to buy a credit, which is called an "Aufladebon."  You can get them at stores like Lidl, Rossman, and Jet as well as Esso gas stations.  Just ask for the following in German:  FONIC Aufladebon für 20 Euro, which is a 20 Euro credit and gives you 8 days of surfing.   I recommend buying this at the same store where you get your modem, which means the start-up cost is about 70 Euros.  

The Aufladebon itself a bit disappointing:  a printed receipt with a 12 digit number and some instructions in German.  Start your Mobile Partner app, connect to the Internet, and click on "Guthaben verwalten" (manage credit).  Select "Guthabenkarte aufladen" (load credit), enter your 12 digit number, and press enter.   Your credit is now topped up for a while.  You can check the level using the "Guthabenabfrage" (check credit) button.  

Performance

I used the FONIC modem for about a week.  It functioned well in every location that had at all reasonable connectivity.  FONIC handles transitions across cells really well--provided there's connectivity it seems to use the fastest available protocol, such as HSPA, and switches seamlessly down to lower capacity protocols like Edge if that is all that is available.  You can use it in a car or train without experiencing application problems--I used it on the Autobahn between Hamburg and Lübeck and it worked surprisingly well.   (Just be clear, somebody else was driving at the time.) 

With good connectivity, the rated speed is 7.2Mbps down and 5.76 upload, which is comparable to average household connections.  PDFs and similar files downloaded at 150K/sec.  Upload was similarly snappy.  

Applications like ssh and even remote debugging/profiling worked very well.  In my case this included running Yourkit Java profiler connecting into an application running on servers elsewhere in Germany.  I also hosted GotoMeeting web conferences.  The latency seems higher than DSL--typical latencies run about 100 to 200ms, which is about the same latency you get when connecting from the US West Coast to sites in Europe. 

Daily transfer limits are very competitive.  The current limit is 500Mb per day at full speed access, though in practice it seemed to cut off around 600Mb for me.  This is better than offerings from the big guys like T-Mobile and Vodafone.  FONIC implements the limit quite elegantly.  First you get a nice SMS that you are near the limit, followed by another that you are over the limit.  However, after that you do not get cut off.  The connection just switches to 64kbps until the next day.  This is still faster than the Wi-Fi connectivity in many hotel rooms.

If you don't want to wait for the friendly SMS message about limits, you can check your limits easily on using the Mobile Partner app.

Speaking of the app, Mobile Partner seems really solid.  There were no bugs, crashes or weird hangs the way you get with some of the US providers like Verizon's prepaid Internet app. I only ran it on Mac OS X but I would guess it is just as good on Windows.  

Caveats

In general FONIC is outstanding.   Life is not perfect, so here are a few of the limitations I found.   

1. German.  FONIC is 100% German, including the website and the app.  You'll need to know some German or make friends fast.  Also, the data plan apparently only works in Germany.  
2. Phone connectivity.  If your cell phone does not work, FONIC won't either.  The nice thing is that you can have connectivity drop out for a bit while crossing cells in a train or car without losing TCP/IP sessions.  
3. Occasional routing problems. On one evening in one particular location the FONIC modem did not work at all well, reporting ping latencies of 50+ seconds or dropping out entirely.  It was not apparent whether this was due to poor cell coverage or another issue.  It worked fine everywhere else.

There appear to be some competitors to FONIC but I have no idea how well they work.  One of the advantages of FONIC is that they have many partners in Germany that sell both modems as well as the credits.  That would also be a consideration in evaluating competing products.

Summary

Some day it may be possible to get decent mobile data plans that work for international travel.  Given the complexity of telecom regulation and the motivations of local carriers to prevent competition, you probably don't want to hold your breath until they arrive.

In the meantime, FONIC looks like a good Internet access solution for anybody traveling to Germany for more than a few days who needs constant, high-capacity access.  I was stuck in a couple of locations that had no network access, so it was a life-saver in those cases.  However, FONIC is also cheaper than hotel Wi-Fi and has better performance than many Wi-Fi hotspots.   If you move about or are uncertain about the quality of the Internet connectivity where you are staying, I highly recommend it.

If you know of similar plans for other countries, please share your experiences.  I would love to find something like FONIC for Italy, Spain, and the US.  

Introducing Tungsten On-Disk Queues for Parallel Replication

Tungsten Replicator has offered shard-based parallel replication to slaves since late 2010.  The initial implementation uses in-memory queues.  Working purely in memory keeps latency low and throughput high.   On the other hand, working in memory consumes valuable RAM.  It also forces us to buffer all in-flight transactions and therefore greatly limits the span of time permissible between the slowest and fastest shard.

Hence our newest improvement:  on-disk parallel queues.  In this article I will cover how parallel replication works in general, how on-disk queues help with parallel replication, and then show how to set up from the latest builds.

First, let's review the basic mechanics.  Parallel replication is really "parallel apply," which means taking a stream of serialized transactions, splitting them into separate streams, and applying them in parallel to a slave DBMS.  In the Tungsten pipeline architecture, we implement this kind of flow using a combination of stages and stores.  One stage reads transactions from the persistent transaction history log (aka THL) to a "parallel store."  The parallel store splits the stream into a set of queues.  The next stage extracts from those queues and applies to the slave.   It looks like the following picture:

From a conceptual point of view the incoming thl-to-q task thread performs an indexing function.  It  guides construction of the queues read by task threads in the q-to-dbms stage.  Within this framework there are many ways to feed events into the parallel queues.    In the case of on-disk queues there are two obvious design options.  

1. Read data out of the THL and split them into separate transaction logs per parallel queue.  This is very similar to the in-memory approach, except that the queues are now on disk (or SSD or whatever storage you pick).  It can be implemented without adding any extra threads to the parallel store. 
2. Leave all data in THL.  Add a cursor for each parallel queue that scans the THL and picks only the transactions that belong in that parallel queue.   This requires extra threads to do the scans, hence is more complex to implement.  

Both approaches achieve the primary goal, which is to keep the transactions in storage until we actually need them and thereby minimize memory usage.  This in turn solves a major problem, namely that individual shards can now be many thousands or even millions of transactions apart in the serial history.  Beyond that, it is not completely obvious which approach is better.

For example, option 1 isolates the reads to individual files.  This minimizes overall I/O at the cost of making it more random, since reads and writes are spread over many files.  Option 2 avoids extra writes and keeps I/O sequential, but introduces a bunch of threads doing the equivalent of table scans across the same patch of storage.  Up to a point we can assume that pages are coming out of the OS page cache rather than storage but this assumption will not hold for all operating environments and workloads.  The only way to prove the trade-offs is to implement and test.  (We may end up implementing both.)

After some discussion internally at Continuent as well as with Harrison Fisk from Facebook, we picked option 2 for the initial implementation.  Here is a diagram that shows how it works.

Here is a quick tour of the implementation in Java class THLParallelQueue.  This class maintains an in-memory blocking queue for each channel.  Each queue has a corresponding read thread that scans the THL and places matching events into the queue.  The THLParallelQueue class synchronizes read threads and handles issues like serialization and clean shutdown.  Some memory is therefore consumed, for queues, but they are quite small and amount to far less than keeping all transactions in memory.

So much for the theoretical description. If you would like to test on-disk queues yourself, you can get started in three steps.

1. Download the latest nightly build of Tungsten Replicator to /tmp. 
2. Untar and cd into the resulting release directory.  
3. Install using the new tungsten-installer as described by Guiseppe Maxia.  

Here is an example of set-up commands.  My test system uses logos1 as the master and logos2 as the slave.  The MySQL DBMS build is Percona Server 5.1.54.  Tungsten is installed in /opt/tungsten.

# Download and unpack build.
cd /tmp
wget --no-check-certificate https://s3.amazonaws.com/files.continuent.com/builds/nightly/tungsten-2.0-snapshots/tungsten-replicator-2.0.4-154.tar.gz
# Untar.
tar -xvzf tungsten-replicator-2.0.4-154.tar.gz
cd tungsten-replicator-2.0.4-154
# Set up and start replicators. 
export TUNGSTEN_HOME=/opt/tungsten
/tmp/tungsten-replicator-2.0.4-154/tools/tungsten-installer \
  --master-slave  \
  --master-host=logos1  \
  --datasource-user=tungsten  \
  --datasource-password=secret  \
  --service-name=percona \
  --home-directory=${TUNGSTEN_HOME} \
  --cluster-hosts=logos1,logos2 \
  --relay-directory=${TUNGSTEN_HOME}/relay-logs \
  --datasource-log-directory=/usr/local/percona-5.1.54/data \
  --thl-directory=${TUNGSTEN_HOME}/thl-logs \
  --channels=10 \
  --svc-parallelization-type=disk \
  --start-and-report                                  

Note the bold options to select disk queues--"memory" is the other option--and the number of channels.  You can confirm you have the right queue installed by running the following command against any slave replicator.  You should see the storage class THLParallelQueue in the status output.

$ trepctl -host logos2 status -name stores
Processing status command (stores)...
NAME                VALUE
----                -----
doChecksum        : false
logDir            : /opt/rhodges4/thl-logs/percona
logFileRetention  : 7d
logFileSize       : 100000000
maximumStoredSeqNo: 0
minimumStoredSeqNo: 0
name              : thl
storeClass        : com.continuent.tungsten.replicator.thl.THL
NAME                VALUE
----                -----
criticalPartition : -1
discardCount      : 0
eventCount        : 1
headSeqno         : 0
maxSize           : 10
name              : parallel-queue
queues            : 10
serializationCount: 0
serialized        : false
stopRequested     : false
store.0           : THLParallelReadTask task_id=0 thread_name=store-thl-0 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.1           : THLParallelReadTask task_id=1 thread_name=store-thl-1 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.2           : THLParallelReadTask task_id=2 thread_name=store-thl-2 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.3           : THLParallelReadTask task_id=3 thread_name=store-thl-3 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.4           : THLParallelReadTask task_id=4 thread_name=store-thl-4 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.5           : THLParallelReadTask task_id=5 thread_name=store-thl-5 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.6           : THLParallelReadTask task_id=6 thread_name=store-thl-6 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.7           : THLParallelReadTask task_id=7 thread_name=store-thl-7 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.8           : THLParallelReadTask task_id=8 thread_name=store-thl-8 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.9           : THLParallelReadTask task_id=9 thread_name=store-thl-9 hi_seqno=0 lo_seqno=0 read=1 discarded=0 events=0
storeClass        : com.continuent.tungsten.replicator.thl.THLParallelQueue
syncEnabled       : true
syncInterval      : 2000
Finished status command (stores)...

Now for some fine print.  On-disk queues are implemented but are still undergoing QA.  There are bugs.  The most important problem is performance--the latency is a lot higher than expected on some of our systems, which I suspect is due to an as-yet undiagnosed bug.  If you try them out now you can expect to hit a few problems.  On the other hand, we take any and all feedback quite seriously, so this is your chance provide input and help guide the final implementation.  Please log issues on the Tungsten Replicator issue tracker or bring up questions on the tungsten-discuss mailing list.

Finally, if you would like to learn more about the parallel queue implementation, check out the design documentation on our wiki as well as the source code.  They are both pretty readable.

Introducing MySQL to MongoDB Replication

The last article on this blog described our planned MySQL to MongoDB replication hackathon at the recent Open DB Camp in Sardinia.  Well, it worked, and the code is now checked into the Tungsten Replicator project.   This article describes exactly what we did to write the code and set up replication.  You can view it as a kind of cookbook both for implementing new database types in Tungsten as well as setting up replication to MongoDB.

The Team

MySQL to MongoDB replication was a group effort with three people:  Flavio Percoco, Stephane Giron, and me.  Flavio has worked on MongoDB for a couple of years and is extremely well-informed both on database setup as well as application design.  Stephane Giron is a replication engineer at Continuent and has done a substantial amount of the work on data extraction from MySQL, especially row replication.  I work on the core execution framework as well as performance.

Getting Started with MongoDB

There were a couple of talks on MongoDB during the first morning of Open DB camp (Saturday May 7th), which Stephane and I dutifully attended to raise our consciousness.  We got cracking on implementation around 2pm that afternoon.   The first step was to bring up MongoDB 1.8.1 and study its habits with help from Flavio.

MongoDB is definitely easy to set up.  You get binary builds from the MongoDB download page.  Here is a minimal set of commands to unpack MongoDB 1.8.1 and start the mongod using directory data to hold tables.  

$ tar -xvzf mongodb-osx-x86_64-1.8.1.tgz
$ cd mongodb-osx-x86_64-1.8.1
$ mkdir data
$ bin/mongo --dbpath data
(... messages ...)

You connect to mongod using the mongo client.  Here's an example of connecting and creating a table with a single row.

$ bin/mongo localhost:27017
MongoDB shell version: 1.8.1
connecting to: localhost:27017/test
> use mydb
switched to db mydb
> db.test.insert({"test": "test value", "anumber" : 5 })                
> db.test.find()
{ "_id" : ObjectId("4dce9a4f3d6e186ffccdd4bb"), "test" : "test value", "anumber" : 5 }
> exit

This is schema-less programming in action.  You just insert BSON documents (BSON =  Binary JSON) into collections, which is Mongolese for tables.  MongoDB creates the collection for you as soon as you put something in it. The automatic materialization is quite addictive once you get used to it, which takes about 5 minutes.

The MongoDB client language is really handy.  It is based on JavaScript.  There are what seem to be some non-Javascript commands like "show dbs" to show databases or "show collections" to list the tables.  Everything else is object-oriented and easy to understand.  For example, to find all the records in collection test, as we saw above, you just connect to the database and issue a command on the local db object.  Collections appear as properties of db, and operations on the collection are methods.

It helps that the MongoDB folks provide very accessible documentation, for example a SQL to MongoDB translation chart.   I put together a little practice program using the MongoDB Java driver to insert, referring to the Javadoc for the class library when in doubt about API calls.  There are also a couple of very helpful examples, like this one, included with the driver.

All told, setup and orientation took us about 45 minutes.  It helped enormously that Flavio is a MongoDB expert, which minimized flail considerably.

Implementing Basic Replication from MySQL to MongoDB

After setup we proceeded to implement replication.  Here is an overview of the replicator pipeline to move data from MySQL to MongoDB.  Pipelines are message processing flows within the replicator.

Direct pipelines move data from DBMS to another within a single replicator.  They are already a standard part of Tungsten Replicator and most of the code shown above already exists, except for the parts shown in red.   Before we started, we therefore needed to set up a replicator with a direct pipeline.

We first built the code according to the instructions on the Tungsten project wiki, uploaded the binary to our test host, and configured the replicator.   First, we ran the Tungsten configure script to set defaults for the MySQL server (user name, extract method, etc.).   Next we ran the configure-service command to set up the direct pipeline configuration file.   Both commands together look like the following:

./configure
./configure-service -C --role=direct mongodb

The second command created a file called tungsten-replicator/conf/static-mongodb.properties with all the information about the direct pipeline implementation but of course nothing yet about MongoDB.

Now we could start the implementation.   To move data to MongoDB, we needed two new components:

1. A Tungsten RawApplier to apply row updates to MongoDB.  RawApplier is the basic interface you implement to create an applier to a database.  
2. A Tungsten Filter to stick column names on row updates after extracting from MySQL.  MySQL row replication does not do this automatically, which makes it difficult to construct JSON at the other end because you do not have the right property names. 

To get started on the applier I implemented a very simple class named MongoApplier that could take an insert from MySQL, turn it into a BSON document, and add it to an equivalently named database and collection in MongoDB.  I added this to the replicator code tree, then built and uploaded tungsten-replicator.jar.  (Use 'ant dist' in the replicator directory to build the JAR.)

To start using the new MongoDB applier, we needed to edit the service properties file to use this component instead of the standard MySQL applier that configuration adds by default.  To do this, you can open up static-mongodb.properties with your favorite editor.  Add the following properties at the bottom of the APPLIERS section.

# MongoDB applier.  You must specify a connection string for the server.
# This currently supports only a single server.
replicator.applier.mongodb=com.continuent.tungsten.replicator.applier.MongoApplier
replicator.applier.mongodb.connectString=localhost:27017

Next, you need to fix up the direct pipeline so that the last stage uses the new applier.  We located the direct pipeline definition (around line 208 in the properties file) and set the applier to mongodb as shown in the following example.

# Write from parallel queue to database.
replicator.stage.d-pq-to-dbms=com.continuent.tungsten.replicator.pipeline.SingleThreadStageTask
replicator.stage.d-pq-to-dbms.extractor=parallel-q-extractor
replicator.stage.d-pq-to-dbms.applier=mongodb
replicator.stage.d-pq-to-dbms.filters=mysqlsessions
replicator.stage.d-pq-to-dbms.taskCount=${replicator.global.apply.channels}
replicator.stage.d-pq-to-dbms.blockCommitRowCount=${replicator.global.buffer.size}

We then started the replicator using 'replicator start.'  At that point we could do the following on MySQL:

mysql> create table foo(id int primary key, msg varchar(35));
Query OK, 0 rows affected (0.05 sec)
mysql> insert into foo values(1, 'hello from MySQL!');
Query OK, 1 row affected (0.00 sec)

...And within a second we could see the following over in MongoDB:

> show collections
foo
system.indexes
> db.foo.find();
{ "_id" : ObjectId("4dc55e45ad90a25b9b57909d"), "1" : "1", "2" : "hello from MySQL!" }

This kind of progress was very encouraging.  It took roughly 2 hours to get to move the first inserts across.  Compared to replicating to a new SQL database like Oracle that's lightning fast.  However, there were still no property names because we were not adding column names to row updates.

Meanwhile, Stephane had finished the column name filter (ColumnNameFilter) and checked it in.  I rebuilt and refreshed the replicator code, then edited static-mongodb.properties as follows to add the filter.  First put in the filter definition in the FILTERS section:

# Column name filter.  Adds column name metadata to row updates.  This is
# required for MySQL row replication if you have logic that requires column
# names.
replicator.filter.colnames=com.continuent.tungsten.replicator.filter.ColumnNameFilter

Next, make the first stage of the direct pipeline use the filter:

# Extract from binlog into queue.
replicator.stage.d-binlog-to-q=com.continuent.tungsten.replicator.pipeline.SingleThreadStageTask
replicator.stage.d-binlog-to-q.extractor=mysql
replicator.stage.d-binlog-to-q.filters=colnames,pkey
replicator.stage.d-binlog-to-q.applier=queue

We then restarted the replicator.  Thereupon, we started to see inserts like the following, complete with property names:

> db.foo.find()
{ "_id" : ObjectId("4dc77bacad9092bd1aef046d"), "id" : "25", "data" : "data value" }

That was better, much better!   To this point we had put in exactly 2 hours and 45 minutes wall clock time.  It was enough to prove the point and more than enough for a demo the next day.   The hackathon was a rousing success.

Further Development

Over the next couple of days I rounded out the MongoApplier to add support for UPDATE and DELETE operations, as well as to implement restart.  The full implementation is now checked in on code.google.com, so you can repeat our experiences by downloading code and building yourself or by grabbing one of the Tungsten nightly builds.

Restart is an interesting topic.  Tungsten uses a table to store the sequence number of the last transaction it applied.  We do this by creating an equivalent collection in MongoDB, which is updated after each commit.  There is a problem in that MongoDB does not have transactions.  Each update is effectively auto-commit, much like MyISAM table type on MySQL.  This means that while Tungsten can restart properly after a clean shutdown, slave replication is not crash safe.  Lack of atomic transactions is a bigger issue with MongoDB and other NoSQL databases that goes far beyond replication.  For now, this is just how Tungsten's MongoDB support works.

Speaking of things that don't work, the current implementation is a prototype only.  We have not tested it with more than a few data types.  It only works with a single MongoDB daemon.  It does not set keys properly or specify indexes on tables.  There are no guarantees about performance, except to say that if you had more than a small amount of data it would be quite slow.  (OK, that's a guarantee after all.)

Epilog

Overall all the hackathon was a great success, not to mention lots of fun.  It went especially well because we had a relatively small problem and three people (Stephane, Flavio, and Robert) with complementary skills that we could combine easily for a quick solution.  That seems to be a recipe for succeeding on future hackathons.

From a technical point of view, it helped that MongoDB is schema-less.  Unlike SQL databases, just adding a document materializes the table in MongoDB.  This made our applier implementation almost trivially easy, because processing row updates takes only a few dozen lines of Java code in total.  It also explains why a lot of people are quite attached to the NoSQL programming model.

I am looking forward to learning a lot more about MongoDB and other NoSQL databases.  It would take two or three weeks of work to get our prototype to work with real applications.  Also, it looks as if we can implement replication going from MongoDB to MySQL.  According to Flavio there is a way to search the transaction log of MongoDB as a regular collection.  By appropriately transforming BSON objects back to SQL tuples, we can offer replication back to MySQL.

There are many other lessons about MongoDB and NoSQL in general but it seems best to leave them for a future article when I have more experience and actually know what I'm talking about.  Meanwhile, you are welcome to try out our newest Tungsten replication feature.

Time for a SQL/NoSQL Group Hug

European Open Database Camp 2011 is this weekend in the hills above Cagliari, Sardinia.  In honor of the increasing number sites that use both NoSQL and SQL databases, I am going to be running a MySQL to NoSQL Hackathon to prototype Tungsten Replicator support for transferring data from MySQL to MongoDB.  The conference will have at least one well-informed MongoDB expert, so we should have enough critical mass to get this done.  It helps that I'll be completely jet-lagged after flying in from the US and unable to sleep anyway.  

Over the past year SQL vs. NoSQL rants have started to abate as people get down to the practical work to make both types of systems work most effectively, often within the same site if not the same application.   I have spoken to a small but growing number of users who want to move data between MySQL and Cassandra, HBase, MongoDB, and others.  (Here's a typical example from StackOverflow.)  Some of them are even willing to pay money for Tungsten Replicator implementations.  You know there is a real need when that happens.  I expect such requests will grow as more applications implement NoSQL stores and have to set up feeds to or from SQL.   Heterogeneous replication in this sense is a good proxy for solution maturity, and at least a couple of the NoSQL stores seem to be getting there.

The MySQL to MongoDB hackathon is our first crack at this problem.  There is no guarantee the results will be especially usable, because Tungsten will not have a built-in solution for SQL to JSON mapping.  (I'm still noodling about that problem--suggestions welcome.)  Still, we will start to understand the problem space better and pave the way for more robust solutions over the course of the next year or so.  I'm looking forward to learning more a lot more about non-SQL stores along the way. 

If you are attending the conference you can hear about our results in person.  You can also tune into the #tungsten IRC channel at irc.freenode.net starting Friday May 5th around 10pm GMT.  I'll post regular updates over the weekend.  Feel free to drop by on IRC to see how we are doing or even to help out.  We will not single-handedly end the SQL/NoSQL wars, but at least we can help them learn how to share. 

Tungsten Supports Logical Replication on PostgreSQL

Thanks to my colleague Linas Virbalas, Tungsten Replicator has just taken the next step to support full logical replication for PostgreSQL.  Linas posted an article today on his new blog describing PostgreSQL logical replication using SLONY triggers.  I saw a demo of his implementation and was really impressed.   For more information you should read the article, which provides an excellent description of how Tungsten replicates from SLONY logs.

It is pretty exciting whenever Tungsten replicates data to or from a new DBMS type, but PostgreSQL logical replication is really special.  Tungsten Replicator has been able to manage native warm standby and log streaming replication more than a year using a script-based plugin.  This is fine for offering copies for failover but rather limited for problems beyond simple availability.  Logical replication opens up a whole new set of solutions that include multi-master replication, heterogeneous data transfer, and large-scale read scaling on replicas.  It is also a key building block for advanced clustering capabilities like zero-downtime upgrade as well as for building multi-tenant systems for SaaS applications.  After years of working on these problems for MySQL we are glad to finally attack them head on for PostgreSQL as well.

There is still a lot of work to achieve fully functional, easy-to-use PostgreSQL logical replication, but the work that Linas described gives Tungsten a clear path forward.  I expect we will make rapid progress, because so many of the other parts of Tungsten Replicator are already in place.  Meanwhile, Linas has put together a very readable blog that should make interesting reading for years to come.

Belated Thanks to MySQL Community

Tungsten Replicator won O'Reilly Application of the Year at the 2011 O'Reilly MySQL Conference, together with Percona's XtraBackup.  Giuseppe Maxia also received an award for Community Contributor of the Year. Having now worked with Giuseppe for almost half a year I know from personal experience his reward is truly deserved.  All in all we had a very good week, especially since the replicator award was a complete surprise.

Things were so busy during and after the MySQL conference it was difficult to write a timely thank-you note. I hope it's not too late to thank the committee now for both awards.

More importantly, I would like to thank the MySQL community as a whole.  Replicated data is the lifeblood of MySQL applications.  There has been a long history of innovation both within the MySQL engineering team as well as the community as a whole.  Working on replication for MySQL is a bit like building cars to operate on the German Autobahn.  If you can compete here you can compete anywhere.

Settling in at code.google.com

Tungsten Replicator code is now fully open source and published on code.google.com.  Here is our new home in case you do not yet know it:  http://code.google.com/p/tungsten-replicator.  I hope you will visit our new digs and admire the furniture.

The fact that the replicator is now fully open source under GPL V2 is kind of old news, so I would instead like to talk about something else:  our initial experience setting up the replicator project at code.google.com.  In a nutshell, it has been excellent.   There are several things that stand out.

1. The site is incredibly easy to use.   You can customize the home page, add members, add external links, etc. quickly and without having to resort to help. 
2. It has everything we need.  The front page is excellent--clean but also all the information users need to get started.  Useful features like issue trackers and Google Groups are cleanly integrated.   
3. It is very fast.  
4. So far it seems to have just the right mix of open and closed for our project.  Anybody can post to the groups or log issues, but only committers on the project have write access to code and ability to move bugs through issue status. 

The only problems I have run into personally involve SVN code access.  For example, say you check out using the http rather than https URL as in:
svn co http://tungsten-replicator.googlecode.com/svn/trunk/builder

If you edit something and try to check in you get a message like the following: 

$ svn commit -m "This does not work"
svn: Commit failed (details follow):
svn: Server sent unexpected return value (405 Method Not Allowed) in response to MKACTIVITY request for '/svn/!svn/act/8d94e398-83ba-46f3-aae2-bd10cb707c4b'
svn: Your commit message was left in a temporary file:
svn:    '/home/rhodges/google/tungsten-replicator/builder/svn-commit.tmp'

This message is definitely in the "not helpful" category.  Perhaps it is some sort of defense against evildoers.  However, this might be subversion behavior and nothing to do with Google.  If you receive such a message, run svn info to check the SVN URL.  If you see http instead of https you have found the cause.   Unfortunately the cure seems to be to check out again properly in another location, copy in your changes, and then commit.

Site credentials are a more insidious problem.  Android phone users need to have a Google GMail account to access updates and download apps.  (At least that's true for my provider.)  Browsers like Firefox do not keep accounts separated properly, so you may run into account confusion when you first get started.  On Mac OS X you can get the wrong account in your keychain, which in turn leads to more confusing error messages.  This is a Google problem.  There is a creeping form of web single-signon using Google, Facebook, and other accounts as identifiers with unintended side-effects for work and personal interactions.  It makes you wonder what other problems are out there.

But I digress. As far as the project is concerned the issues look pretty minor.  At this point I would recommend code.google.com wholeheartedly for open source projects.

p.s., Tungsten Replicator 2.0.2 is on the way.  More on that in another post.