Today we have an action story based on real events. We will rebuild the plane while flying it: changing the architecture of a high-load system on the fly.
The setting is a very large big-data-class track & trace system. A move to a sharded storage architecture had long been postponed there. So the main hero has to deal with the evil awakened in the system under load: performance degradation, a lock plateau, and overload alerts.
At the end, as usual, there is a happy ending. Our hero fearlessly changes the architecture of the solution on the fly with no downtime (DT) and keeps the system running as normal. Evil is defeated, and the brave engineer basks in applause!
The article is based on a talk at the Saint Highload++ 2022 conference. If you would rather not read, you can watch the video version of the talk.
Track & Trace — this class of systems is meant to provide end-to-end tracking of a business process: the status of movement of various groups of goods through supply chains. It covers every stage of the lifecycle, from production to sale or another final state (write-off, destruction, etc.).
Original solution
A few words about the architecture of the data processing solution. The simplified pipeline looks like this:
- there is a source, and its data flows into an incoming processing queue built on Kafka;
- the application implements a Kafka consumer that reads data out of the queue;
- then comes the processing itself and serialization of its results into the operational store.
The primary database is the NoSQL database MongoDB.
And where could problems possibly arise? The vendor thought of everything: this database was originally designed with broad horizontal scaling capabilities and holds up well under high load, right?
The problem is that very often — especially at the early stage of a system or project — the chase for features leaves no time to think about the long term. What happens to the viability of the built solution under load? By the time problems appear, the accumulated data volume can be measured in tens or even hundreds of terabytes.
Our case is no exception. We started with a MongoDB deployment in a Single ReplicaSet topology, not a full-fledged sharded cluster, let alone geo-replication, which remains only a dream.
A replica set is a cluster of MongoDB servers that implements master-slave replication and automatic failover between them.
A few dry specs of our solution:
- Write RPS: ~10K ops/s
- Read RPS: ~15-20K ops/s
- Uncompressed dataset: ~100TB
Disclaimer: what I will not cover in this article
- Recipes for building high-load system architectures. Everyone has their own approach here.
- Horizontal and vertical scaling of applications.
- Scaling of Kafka.
- Delivery semantics in Kafka (at least once, at most once, exactly once).
- Architecture, fault tolerance, and scaling of the MongoDB database.
Now on to the main problem
Nothing foreshadowed trouble; the system was chewing through a fairly intense incoming load. But one fine day all the dashboards of the monitoring system literally went mad, and the operations team was buried in alerts. An emergency degradation of the processing system's performance was recorded, along with a sharp growth of the incoming task queues! Total alert!
We started digging into it. We enabled online profiling in the application and found an unpleasant surprise: we had hit the database's performance ceiling. Anticipating questions: at that point we had around 7 billion records.
At moments like this you remember a quote from a famous character: "There was never any such thing before, and here it is again."
Digging further. In the monitoring system we find data from the Mongo exporter, and there it is — a plateau of locks! We are talking about Read and Write tickets, which MongoDB uses to organize concurrent access.
"How delightful!" you say — and you would be absolutely right.
So there we were: 100 TB of data in the database and exactly zero chances of simply re-sharding that volume without DT.
The panickers are already shouting: "All is lost! What do we do?"
Solving the problem
To hell with emotions — let's engage the cold mind.
Recall how the WiredTiger engine works; it is the default and primary engine in MongoDB. In file storage terms it is simple: an entire collection on a given database node lives in one file (<collection>.wt), plus one file per index (<index>.wt). We open the folder with the database's data files and — bingo — files measured in terabytes are waiting for us there.
Logic suggests the problem is exactly the size of the data file.
The first solution that comes to mind is to cut the data into pieces (do a split). An attentive reader will say: "But, allow me, that is exactly what a sharded database does!" True, except we have no ability to shard with the database's own means right now — not at all.
We will split the data at the application level, live. As they say, audacity and courage are our thing, and there are no other options anyway.
How is sharding implemented in practically any database? Right, hashing is our friend. All that remains is to decide which hash function to take. In this case we take Adler32. Choosing a hash function in general is a big deal, but that is a topic for a separate article.
We split our dataset (a super-collection) into N pieces (N small collections) and add a small wrapper in the application's data access layer. The catch is that the new construction only applies to fresh data, and we start with an empty dataset.
What do we do with the existing huge file? Let's tear it into pieces as data in it is accessed, to avoid an unnecessary DT for a migration we cannot afford.
Let's patch the data access layer a bit more. We try launching in production — everything worked and the locks are gone!
Summarized in two words
We split one super-giant dataset into N pieces without changing the database topology. This removed the continuous locks at the database level and normalized data processing performance in the application.
A quick run through the pros and cons of the approach
Pros:
- applies when there is no ability to shard with the database's own means right now (a very large data volume, no infrastructure for a sharded database cluster, etc.);
- production is not stopped (no downtime);
- the approach can speed up inserts and other operations (reduce the degradation caused by data growth by lowering the number of locks);
- no initial data migration is required — data moves from the single enormous super-collection into separate (split) small collections gradually, as documents are accessed.
Cons:
- managing the split data has to be implemented at the application level rather than the database level;
- it is hard to search for data without special tools;
- there is a limit, since we are still bounded by the capabilities of vertical scaling.
It seems the problem is solved. But in fact, not fully.
Finding and solving problem #2
Sooner or later — and practice shows it will be sooner — we will hit the limits of vertical scaling. Then we will have to move to a full-fledged sharded database. All that's left is to figure out how to do it!
What if we run two databases in parallel — the old (non-sharded) one and the new (sharded configuration)? The idea is good; it just needs to be implemented somehow.
We start sketching ideas:
- The new database will hold all new data, starting from the moment parallel operation begins.
- The old one will keep holding the old data.
- Query routing between the old and the new database is needed.
The main question in the current setup is how to organize query routing depending on the input parameters? The traditional approach of building a large routing table helps here.
In our case we need to fill the table with conditional pointers to data in the old database, since its data set remains static from the perspective of access keys.
Obviously, storing billions of records in a file is impossible. So a separate database instance has to be deployed. To reduce the storage volume and speed up lookups, we write the hashed value of the original record's key into the routing table. To minimize collisions we use an algorithm from the SHA family.
Of course, we will have to initialize this giant hashset before operation starts. The good news is that during this time we keep using the old database as before — so there is no DT here.
A few notes on building a hashset for this data volume. Although SHA-family algorithms provide minimal collisions, collisions still exist. For such cases we store a list of the keys themselves next to the hash, as extra metadata to resolve the collisions that arise.
We patch the data access layer in the application a bit, and start parallel operation — after profiling the router on the load-testing rig, naturally. Everything works!
Short summary of the work done
We deployed a new sharded database cluster for new data. The old one was not shut down, and its data was not copied into the new one. Using a link in the form of a query router built on a separate database instance that holds a static routing table, we were able to run full parallel operation of two databases, with a path toward retiring the old database and moving completely to the target topology. All of this without DT.
Let's analyze the implemented solution
Pros:
- production is not stopped (no downtime);
- no need to re-shard data (which requires downtime);
- we start the sharded db with an empty dataset, or migrate the most in-demand subset of data (the freshest, the frequently updated, business-critical, etc.);
- the old non-sharded db keeps running in parallel.
Cons:
- extra operating costs for the extended storage architecture (two database installations plus a separate instance for the router);
- the static query routing table must be initialized before parallel operation starts (for all entities in the non-sharded database).
A few words about compaction
I will also touch on another problem of operating a database of this size: compaction. Spoiler: things are bleak here too, but there is a solution!
Compaction lets you compact data in the database, which shrinks the disk footprint — our data's trace on the physical medium, the size of the data files. That pushes problems like the one we just discussed as far out as possible.
- The stock MongoDB compact utility does not work at all at these volumes.
- Compression works only through the initial sync procedure — that is, by creating a new replica from scratch via replication, but it takes a very long time.
- The only way to create a new replica quickly is direct copying. Of course, the better replica to copy is the one obtained by compressing through the initial sync procedure.
There are other operational quirks too, for example how to back up such a large database. That is a topic for a separate article, because the task is, to put it mildly, non-trivial.
Drawing conclusions
It is time to sum things up.
We solved the problem of changing the solution architecture on the fly in two stages:
- put out the fire by sharding data at the application level, without changing the original database topology;
- solved the problem for good and for real by running two databases (old and new) in parallel, which will gradually allow the old database to be retired.
In the new configuration each shard holds no more than 2 billion documents, which means we won on server resources compared to the original solution.
After the talk I was asked what we would choose if we were building the storage today, with everything we now know about the system's requirements. The answer is unambiguous: we would stay with MongoDB. It behaves predictably and has been battle-tested.
A few tips based on the experience gained along the way
- Build load models before you go live, even when everything seems obvious.
- Forecast target data volumes in advance. This lets you size the infrastructure correctly and choose a solution topology with room for data growth over a 3–5 year planning horizon.
- Deploy the database in sharded form from the start (ideally at project launch), so that later you don't have to heroically fight off the avalanche of problems building up in production, under load, and without DT. This saves not only time and people, but also infrastructure costs, since you won't have to run three database clusters in parallel.
And one last thing: all the tricks and fixes I described were performed by professionals! Don't step on this rake in your own projects — it's dangerous for your nerves! And remember, there is always a solution!