(Below is a blog post I started in late 2014 but never found time to finish. The research is still good, although since this was written, bolo became a real thing...)
Bolo is the working title of a new breed of monitoring system I've been writing for fun, in my spare time. It aims to be simple, scalable and easy-to-understand.
(This article summarizes a ton of research I've been doing before I begin designing the longterm data storage backends for bolo)
When it comes to ops work, dealing with data is the worst. It's sad, really, since without the data, most technology offerings are worthless. What's a Twitter without tweets? Google without search results? Data is so hard to deal with precisely because it is so apparent to the end users. They notice when it's gone; they know when it's wrong.
Often, data cannot be recreated. If it could, we would just store the logic for reconstruction on disk, and get the data on-demand. Some compression schemes work this way.
Consider a monitoring system collecting state information (is the web cluster up? are the disks healthy?) and metric data (how many requests is the caching layer deflecting, per second?). Having recorded the state or metric, all that remains is that record. Without it, it is impossible to know with any certainty what happened in the past.
Sure, that server is up now, but has it always been? You could check the
uptime, that would tell you when it last booted, but how long was it down
for? The last database could help fill in some more blanks, but what if
the box was powered on, but inaccessible? That counts as an outage, but
isn't recorded anywhere on the box.
Data is fragile, so we build redundancy into the data storage layer. Lots of redundancy. Hopefully, as much as we can afford. Disks fail, so we RAID lots of them together. Servers fail, so we copy the data around to multiple servers. Data centers go offline, so we backup the data somewhere else.
The other problem with data is that there's just so dang much of it. If information wants to be free, data wants to multiply. That's usually fine; disks are getting larger and cheaper every day. However, some systems, like metric engines, exhibit startling growth trends. Adding servers increases the number of tracked states and metrics linearly, but most of the sprawl in monitoring data comes from finding new things to track.
I've seen this firsthand myself, countless times in my professional life. Start with 1,000 servers. Assume that, on average, you're tracking 20 states and 60 metrics. Overall, that's 62,000 things to keep track of.
If the web cluster gets two new nodes, the monitoring system picks up 160 new things to track, a +0.26% gain. On the other hand, if the ops team finds a new failure case that can only be detected by watching 3 new metrics and 1 new state (across all servers), the monitoring system picks up 4,000 new things to track, a +6.5% gain.
On top of that, the monitoring system tends to become a victim of its own success. Provide people with decent insight into how their applications and databases are behaving, and they will inevitably ask for more data, not less.
TBD. Needs some pull
A hash function is some computation that maps arbitrarily large inputs to fixed size outputs. We could implement a hash function that turns a NULL-terminated string of ASCII characters (8-bit wide) into a single 32-bit number by adding up each byte and ignoring the overflow.
#include <stdint.h>
uint32_t hash(const char *s)
{
uint32_t v = 0;
while (*s)
v += *s++;
return v;
}
This is a simple, if naïve, design, but it does qualify as a hash function. It is also stable, because it always returns the same output value when given the same input string, and one-way, since it is difficult to recover any information about the original string from the hash value.
Here are some example runs of our hash function:
"distributed" = 0x04a3
"hashing" = 0x02e2
"can be fun" = 0x0382
Sadly, our little hash function exhibits some pretty bad behavior when it comes to collisions. Consider the two input strings "team" and "mate", and their corresponding hashed values:
"team" = 0x01a7
"mate" = 0x01a7
No emphasis is placed on the order the characters are in; as long as the input strings have the same number and types of characters, they will hash to the same value. Our hash function cannot be described as robust. Proving that a hash-function is robust is non-trivial and involves a fair bit of higher-level maths. For this reason, It is best to stick with well-studied, cryptographically-secure hash functions like the SHA family.
Here's an implementation of Dan Bernstein's djb2 hash function. libvigor uses a range-controlled variant of this algorithm for its hash_t implementation.
#include <stdint.h>
uint32_t djb2(const char *s)
{
uint32_t v = 81;
while (*s)
v += (v * 32) + *s++;
return v;
}
The multiplication step (usually implemented as a 5-bit left-shift
operation, i.e. v << 5) serves to perturb the hashed value v, and helps
emphasize character order. Here are the djb2 hash values for "team" and
"mate":
"team" = 0x5fb1758
"mate" = 0x5f73218
Before we can jump into consistent hashing, we need to re-acquaint ourselves with our old friend, π.
3.1415926535897932384626433832795028841971…
As you'll recall from high school geometry classes, π is the ratio of the diameter of circle to its circumference. It's decimal expansion continues ad infinitum, and exhibits no discernable digit pattern.
Where there's π, there's angles, and angles can be measured three ways: in degrees, in gradients and radians.
Degrees seem pretty natural for most people. perpendicular lines meet at 90° angles, a right triangle is composed of a right angle and two 45° angles, and a circle contains 360° worth of angles.
θ
Using a hash function, we can divide up a large dataset across multiple servers. If we have a hash function that ranges over 0 … 255 (i.e. 8-bits) but we only have 4 servers in our storage pool, we can take the modulo of the hashed value by 4 and get a number between 0 and 3. This number can be used as an index into a table of storage pool nodes:
| 0 | storage1.pool.fq.dn |
|---|---|
| 1 | storage2.pool.fq.dn |
| 2 | storage3.pool.fq.dn |
| 3 | storage4.pool.fq.dn |
More formally, if $n$ is the total number of nodes, $S$ is the set of nodes from $S_0…S_{n-1}$, $L(k)$ is the locator function responsible for finding the node on which content for key $k$ is stored, and $H(s)$ is the hash function:
$$L(k) = S_{H(k)\:\:mod\:\:n}$$
As long as all clients agree on the definition of $H$, $S$ and $n$, they can determine reliably what node contains the content they are interested in.
But this is not a consistent hashing scheme, because it depends so completely on the value of $n$ and the definition of $S$. Changes to either will invalidate previous computations of $L(k)$. While a caching system may be able to recover (following a period of intense utilization), storage systems will just lose data.
To illustrate this, let's consider storing 5 values across these four servers, using the keys "foo", "bar", "baz", "quux" and "norf". First, we'll calculate their 8-bit hashes (using the djb2 algorithm):
"foo" = 0x02e2b55 = 0x55 = 85
"bar" = 0x02e1886 = 0x86 = 134
"baz" = 0x02e188e = 0x8e = 142
"quux" = 0x5f9b8e4 = 0xe4 = 228
"norf" = 0x5f7f9c6 = 0xc6 = 198
The third column contains the same hash values, modulo 256 to make them easier to work with. The last column contains the decimal representation of the 8-bit quantities. With $n = 4$, we end up with the following modified hash values ($H(k)\:\:mod\:\:n$):
"foo" = 85 mod 4 = 1 (storage2)
"bar" = 134 mod 4 = 2 (storage3)
"baz" = 142 mod 4 = 2
"quux" = 228 mod 4 = 0 (storage1)
"norf" = 198 mod 4 = 2
What happens when storage2 goes down because of hardware failure? We could remove it from the list, but that changes both $S$ and $n$, which affects our modulo calculations quite a bit:
"foo" = 85 mod 3 = 1 (storage3)
"bar" = 134 mod 3 = 2 (storage4)
"baz" = 142 mod 3 = 1
"quux" = 228 mod 3 = 0 (storage1)
"norf" = 198 mod 3 = 0
The only key that is still in the same place "quux", Keys that weren't even on storage2 have been affected, including "bar" (which moved from storage3 to storage4) and "norf" (which moved from storage3 to storage1).
Another way to look at this (although it may seem unnatural, at first) is to think of the entire data set as a large circle, divided into segments. All keys map to points on the circle, and the distribution via set $S$ is represented by assigning an equal-sized segment of the circle to each node.
If we remove one node, we must remove its corresponding segment, effectively resizing the other segments by 1/3 of their original length. Similarly, adding a node requires that we resize the other segments to make room for the new one.
These scenarios are bad for data sharding for a few reasons. Whenever segment boundaries change, nodes will need to communicate between themselves to get authority handed over for stored data. In the above example, when we took storage2 out of the ring, the responsibility for the "bar" key migrated from storage3 to storage4. However, neither storage3 nor storage4 can be consciously aware of this,