You have to consider the tail latencies of the system responding plus the network in between. The p99 is typically much higher than the average. Also, may have to account for GC as was mentioned in the article. 500ms gets used up pretty fast.
500ms is actually a very short interval for heartbeats in modern distributed systems. Kubernetes nodes out of the box send heartbeats every 10s, and Kubernetes only declares a node as dead when there's no heartbeat for 40s.
The relevant timescale here is not CPU time but network time. There's so much jitter in networks that if your heartbeats are on CPU scale (even, say, 100ms) and you wait for 4 missed before declaring dead, you'd just be constantly failing over.
Speak for your own network. On a densely interconnected datacenter network there would be no appreciable jitter.
4 x 10s heartbeats sounds like an incredibly conservative decision by whoever chose the default, and I cant imagine any critical service keeping those timeouts.
> Consider a system with 1000 nodes where each node sends heartbeats to a central monitor every 500 milliseconds. This results in 2000 heartbeat messages per second just for health monitoring. In a busy production environment, this overhead can interfere with actual application traffic.
If your 1000-node busy production environment is run so close to the edge that 2000 heartbeat messages per second, push it into overload, that's impressive resource scheduling.
Really, setting the interval balances speed of detection/cost of slow detection vs cost of reacting to a momentary interruption. If the node actually dies, you'd like to react as soon as possible; but if it's something like a link flap or system pause (GC or otherwise), most applications would prefer to wait and not transition state; some applications like live broadcast are better served by moving very rapidly and 500 ms might be too long.
Re: network partitioning, the author left out the really fun splits. Say you have servers in DC, TX, and CA. If there's a damaged (but not severed) link between TX and CA, there's a good chance that DC can talk to everyone, but TX and CA can't communicate. You can have that inside a datacenter too, maybe each node can only reach 75% of the other nodes, but A can reach B and B can reach C does not indicate A can reach C. Lots of fun times there.
> You can have that inside a datacenter too, maybe each node can only reach 75% of the other nodes, but A can reach B and B can reach C does not indicate A can reach C. Lots of fun times there
At BigCloud in the early days, things went berserk with a gossip system when A could reach B but B couldn't reach A.
Small example from CZMQ - High-level C Binding for ZeroMQ: http://czmq.zeromq.org/manual:zgossip . Perhaps not self-contained, one might have to read the manual of other resources, such as zactor, zsock or of the application Zyre - Local Area Clustering for Peer-to-Peer Applications https://github.com/zeromq/zyre .
I’ve worked with a few, and seen a few built. IME they are never really worth it. They make the system much more complex, fragile and hard to debug. The scale limits they aim to help are just not a big deal anymore. A small group of “coordinator” nodes to do this is much simpler and easy to build and scale.
10 years ago I've implemented SCAMP (a gossip protocol) in Clojure, you might find it interesting, the implementation is quite small https://github.com/cipherself/gossip
I've been noodling a lot on how IP/ARP works as a "distributed system". Are there any reference distributed systems that have a similar setup of "optimistic"/best effort delivery? IPv6 and NDP seem like they could scale a lot, what would be the negatives about using a similar design for RPC?
Some fuzzy thinking in here. "A heartbeat sent from a node in California to a monitor in Virginia might take 80 milliseconds under normal conditions, but could spike to 200 milliseconds during periods of congestion." This is not really the effect of congestion, or at best this sentence misleads the reader. The mechanism that causes high latency during congestion is dropped frames, which are retried at the protocol level based on timers. You can get a 200ms delay between two nodes even if they are adjacent, because the TCP minimum RTO is 200ms.
A few years ago, I created a simple Broker (middle layer) program with NetMQ (C# implementation of ZeroMQ)
I did not have to worry about a "Heartbeat" because the way the...
Client <--> Broker <--> Worker
...would communicate already handled it for me.
As for the Worker would send a message to the Broker like "Do you have anything for me?" and the Broker would either respond with "Nothing at the moment" or "Yes.. here you go!"
Either way, we have confirmation between the two services. The Broker keeps a log of the last interactions.. and so does the Worker. If the Worker has not heard back from the Broker in about 10 mins, would attempt to reconnect. If this fails 3 times, would send out an email.
Same for the Broker. If it has not heard anything from the Worker, will send out an email.
As for the Client, that will always send new messages to the Broker. It also keeps tabs on all messages, ensuring the Broker confirms the message is Completed before removing it from the Client system. Every now and then the Client asks "if this was processed" to which the Brokers responds with "No.. still processing" or "Successful/Error"
Regardless, it has a "heartbeat" by default, as well as reliability and low chance of data loss (well, data can get lost... trying to work out a perfect system is difficult but we had something decent beyond the scope of this post)
It was a good system. It was light and requests from the client or worker was responded fast by the broker. Was pretty fast compared to the old system which was relying on SQL Server queries to check states of data, with SQL Server replication on the client systems. It was a "quick and easy" solution when you had 10 clients. Once it got to over 40 it concerned me a lot.
20 comments
[ 4.3 ms ] story [ 45.5 ms ] thread500 milliseconds is a very long interval, on a CPU timescale. Funny how we all tend to judge intervals based on human timescales
Of course the best way to choose heartbeat intervals is based on metrics like transaction failure rate or latency
7200
That is two hours in seconds.
The relevant timescale here is not CPU time but network time. There's so much jitter in networks that if your heartbeats are on CPU scale (even, say, 100ms) and you wait for 4 missed before declaring dead, you'd just be constantly failing over.
4 x 10s heartbeats sounds like an incredibly conservative decision by whoever chose the default, and I cant imagine any critical service keeping those timeouts.
A dead server is much better for a distributed system than a misbehaving one. The latter can bring down your entire application.
If your 1000-node busy production environment is run so close to the edge that 2000 heartbeat messages per second, push it into overload, that's impressive resource scheduling.
Really, setting the interval balances speed of detection/cost of slow detection vs cost of reacting to a momentary interruption. If the node actually dies, you'd like to react as soon as possible; but if it's something like a link flap or system pause (GC or otherwise), most applications would prefer to wait and not transition state; some applications like live broadcast are better served by moving very rapidly and 500 ms might be too long.
Re: network partitioning, the author left out the really fun splits. Say you have servers in DC, TX, and CA. If there's a damaged (but not severed) link between TX and CA, there's a good chance that DC can talk to everyone, but TX and CA can't communicate. You can have that inside a datacenter too, maybe each node can only reach 75% of the other nodes, but A can reach B and B can reach C does not indicate A can reach C. Lots of fun times there.
At BigCloud in the early days, things went berserk with a gossip system when A could reach B but B couldn't reach A.
Cloudflare hit something similar though they misclassified the failure mode: https://blog.cloudflare.com/a-byzantine-failure-in-the-real-...
I have been more interested in learning about gossip protocols and how they are used, different tradeoffs, etc.
I did not have to worry about a "Heartbeat" because the way the...
Client <--> Broker <--> Worker
...would communicate already handled it for me.
As for the Worker would send a message to the Broker like "Do you have anything for me?" and the Broker would either respond with "Nothing at the moment" or "Yes.. here you go!"
Either way, we have confirmation between the two services. The Broker keeps a log of the last interactions.. and so does the Worker. If the Worker has not heard back from the Broker in about 10 mins, would attempt to reconnect. If this fails 3 times, would send out an email.
Same for the Broker. If it has not heard anything from the Worker, will send out an email.
As for the Client, that will always send new messages to the Broker. It also keeps tabs on all messages, ensuring the Broker confirms the message is Completed before removing it from the Client system. Every now and then the Client asks "if this was processed" to which the Brokers responds with "No.. still processing" or "Successful/Error"
Regardless, it has a "heartbeat" by default, as well as reliability and low chance of data loss (well, data can get lost... trying to work out a perfect system is difficult but we had something decent beyond the scope of this post)
It was a good system. It was light and requests from the client or worker was responded fast by the broker. Was pretty fast compared to the old system which was relying on SQL Server queries to check states of data, with SQL Server replication on the client systems. It was a "quick and easy" solution when you had 10 clients. Once it got to over 40 it concerned me a lot.