intuitions for distributed consensus at dc systems

From DC Systems October 2024.

dcsystems.xyz

1. intuitions for distributed consensus @eatonphil

2. meetups started a little meetup with @ngeloxyz in nyc

3. meetups started a little meetup with @ngeloxyz in nyc turned up in cities around the us, india, germany

4. meetups started a little meetup with @ngeloxyz in nyc turned up in cities around the us, india, germany very special to be a part of dc systems

5. who i am i’m phil

6. who i am i’m phil developer for 10 years

7. who i am i’m phil developer for 10 years got interested in databases 4 years ago

8. who i am i’m phil developer for 10 years got interested in databases 4 years ago work for edb on distributed postgres product

9. who you are a number of experts in this room

10. who you are a number of experts in this room who’ve worked on some of the systems i’ll cover

11. who you are a number of experts in this room who’ve worked on some of the systems i’ll cover risky of me! 😂

12. setting expectations this talk is about basics and behavior

13. setting expectations this talk is about basics and behavior relevant for any (backend) developer

14. setting expectations this talk is about basics and behavior relevant for any (backend) developer (first talk i’ve given in ~5 years)

15. let’s go

16. you’ve got an app MyApp

17. and some data Key-Value Store

18. (on a single node) Key-Value Store

19. communication is simple Key-Value Store MyApp Reads / Writes

20. we might write Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" <pending>

21. we might write Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()>

22. and every time we read Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()> t1 get "id:1.name" <pending>

23. we get back what we wrote Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()> t1 get "id:1.name" ok<"I. Asimov">

24. a.k.a. consistency

25. consistency is a spectrum

26. where “consistent” = linearizable

27. most easily illustrated via counterexample

28. linearizability counterexample Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" <pending>

29. linearizability counterexample Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()>

30. linearizability counterexample Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()> t1 get "id:1.name" <pending>

31. linearizability counterexample: reading stale values Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()> t1 get "id:1.name" ok<()>

32. linearizability counterexample: reading stale values Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()> t1 get "id:1.name" ok<()> t2 get "id:1.name" <pending>

33. linearizability counterexample: reading stale values Time WhatsIpp Key-Value Store t0 set "id:1.name" "I. Asimov" ok<()> t1 get "id:1.name" ok<()> t2 get "id:1.name" ok<"I. Asimov">

34. stale reads: one example of being not linearizable

35. more formally

36. https://jepsen.io/consistency/models/linearizable “Linearizability is one of the strongest single-object consistency models, and implies that every operation appears to take place atomically, in some order, consistent with the real-time ordering of those operations”

37. linearizability for a single node, not super interesting

38. linearizability for a single node, not super interesting but as a property in general, very useful

39. linearizability for a single node, not super interesting but as a property in general, very useful linearizable system can be treated as if it were a single node

40. linearizability for a single node, not super interesting but as a property in general, very useful linearizable system can be treated as if it were a single node even if it consists of more than one node

41. single node crashes, entire system is unavailable

42. intuition #0: a single node is not highly available!

43. how to achieve high availability?

44. add nodes? MyApp Reads / Writes KV Store A KV Store B KV Store C ???

45. how to keep data in sync? WhatsIpp Reads / Writes KV Store A KV Store B KV Store C ???

46. read replicas!

47. read replicas are a thing MyApp KV Store A (Leader) KV Store B KV Store C

48. but on leader failure MyApp KV Store A (Leader) KV Store B KV Store C

49. but on leader failure MyApp KV Store A (Leader) KV Store B KV Store C

50. read replicas may not be up-to-date MyApp KV Store A (Leader) KV Store B KV Store C

51. thus not linearizable MyApp KV Store A (Leader) KV Store B KV Store C

52. some options for linearizability + availability chain replication (ebs)

53. some options for linearizability + availability chain replication (ebs) kafka replication protocol

54. some options for linearizability + availability chain replication (ebs) kafka replication protocol foundationdb replication protocol

55. some options for linearizability + availability chain replication (ebs) kafka replication protocol foundationdb replication protocol client-side quorums

56. some options for linearizability + availability chain replication (ebs) kafka replication protocol foundationdb replication protocol client-side quorums distributed consensus

57. further reading: data replication design spectrum

58. but most popular

59. and for this talk

60. distributed consensus

61. distributed consensus linearizability + availability

62. distributed consensus linearizability + availability examples: raft multipaxos viewstamped replication

63. used everywhere

64. kubernetes

65. elasticsearch

66. mongodb

67. cockroachdb

68. tigerbeetle

69. edb postgres distributed

70. etcd, redpanda, scylla, consul, hazelcast, yugabyte, clickhouse, nats, kafka, tidb, neo4j, rabbitmq, etc. https://en.wikipedia.org/wiki/Raft_(algorithm)

71. zooming in on

72. raft

73. at a high level

74. nodes form a cluster KV Store A KV Store B KV Store C

75. cluster elects a leader by majority vote KV Store A KV Store B (Leader) KV Store C

76. if a leader becomes unavailable KV Store A KV Store B (Leader) KV Store C

77. the cluster elects a new leader KV Store A (Leader) KV Store B KV Store C

78. client talks with the (current) leader KV Store A (Leader) KV Store B KV Store C MyApp

79. state changes modeled as commands MyApp KV Store A (Leader) KV Store B KV Store C exec: 'set "id:1.name" "I. Asimov"'

80. state changes modeled as commands MyApp KV Store A (Leader) KV Store B KV Store C exec: 'get "id:1.name"'

81. note: log replication pauses while leader election is happening

82. commands stored in a log KV Store A (Leader) log index value additional log entry metadata (null) (null) 1 set "id:1.name" "I. Asimov" … 2 get "id:1.name" …

83. leader replicates logs in order

84. example state KV Store A (Leader) log index value additional log entry metadata (null) (null) 1 set "id:1.name" "I. Asimov" … 2 get "id:1.name" … 3 get "id:2.name" KV Store B log index value additional log entry metadata (null) (null) 1 set "id:1.name" "I. Asimov" … KV Store C log index value additional log entry metadata (null) (null) 1 set "id:1.name" "I. Asimov" … 2 get "id:1.name" …

85. once majority replicates a log entry, the entry is “committed”

86. once majority replicates a log entry, the entry is “committed” the entry is durable

87. committed index = index replicated by majority KV Store A (Leader) log index value additional log entry metadata (null) (null) 1 set "id:1.name" "I. Asimov" … 2 get "id:1.name" … 3 get "id:2.name" KV Store B log index value additional log entry metadata (null) (null) 1 set "id:1.name" "I. Asimov" … KV Store C log index value additional log entry metadata (null) (null) 1 set "id:1.name" "I. Asimov" … 2 get "id:1.name" … committed replicated not-replicated

88. nodes apply committed entry to state machine, leader returns result to client

89. example kv store state machine def apply_log ( state : Map < string , u64 >, command : [] u8 ) -> Result < Option < u64 >>: command_type = get_command_type ( command ) if command_type == " set ": state [ get_set_command_key ( command )] = get_set_command_value ( command ) return Ok ( None ) if command_type == " get ": return Ok ( Some ( state [ get_get_command_key ( command )])) return Err ( " Unknown command {command_type} " )

90. note: reads need not be replicated like this

91. intuition #1: consensus involves more work than using a single node

92. intuition #1: consensus involves more work than using a single node consensus means higher latency, lower throughput than a single node

93. easily encapsulated

94. distributed consensus libraries go https://github.com/etcd-io/raft https://github.com/lni/dragonboat https://github.com/hashicorp/raft rust https://github.com/tikv/raft-rs https://github.com/databendlabs/openraft

95. pedagogical walkthrough

96. but state machine is always application-specific

97. how is everyone using consensus?

98. modeling with distributed consensus consensus in the control plane & data plane

99. modeling with distributed consensus consensus in the control plane & data plane replicate statements (e.g. rqlite)

100. modeling with distributed consensus consensus in the control plane & data plane replicate statements (e.g. rqlite) replicate data (e.g. cockroach, tigerbeetle, etcd)

101. modeling with distributed consensus consensus in the control plane & data plane replicate statements (e.g. rqlite) replicate data (e.g. cockroach, tigerbeetle, etcd) consensus in the control plane, data plane replicated separately

102. modeling with distributed consensus consensus in the control plane & data plane replicate statements (e.g. rqlite) replicate data (e.g. cockroach, tigerbeetle, etcd) consensus in the control plane, data plane replicated separately chain replication (ebs, delta by meta)

103. modeling with distributed consensus consensus in the control plane & data plane replicate statements (e.g. rqlite) replicate data (e.g. cockroach, tigerbeetle, etcd) consensus in the control plane, data plane replicated separately chain replication (ebs, delta by meta) edb postgres distributed

104. some code

105. why again are we doing this?

106. fault tolerance a fault is when a service is unavailable for any reason

107. fault tolerance a fault is when a service is unavailable for any reason e.g. crashed process

108. fault tolerance a fault is when a service is unavailable for any reason e.g. crashed process e.g. network partition

109. fault tolerance a fault is when a service is unavailable for any reason e.g. crashed process e.g. network partition e.g. general slowness ( gray failure )

110. fault tolerance and raft handling f failures

111. fault tolerance and raft handling f failures requires 2 f +1 nodes

112. fault tolerance and raft handling f failures requires 2 f +1 nodes 3-node cluster can handle 1 failure

113. fault tolerance and raft handling f failures requires 2 f +1 nodes 3-node cluster can handle 1 failure 5-node cluster can handle 2 failures

114. fault tolerance and raft handling f failures requires 2 f +1 nodes 3-node cluster can handle 1 failure 5-node cluster can handle 2 failures 101-node cluster can handle 50 failures

115. worst case scenarios

116. imagine processes constantly crashing

117. imagine processes constantly crashing due to bugs, oom killer, etc.

118. imagine processes constantly crashing due to bugs, oom killer, etc. disks may be slow

119. imagine processes constantly crashing due to bugs, oom killer, etc. disks may be slow network may be down or slow

120. impact takes longer to achieve consensus

121. impact takes longer to achieve consensus takes longer to replicate the log

122. impact takes longer to achieve consensus takes longer to replicate the log elections happen more frequently

123. impact takes longer to achieve consensus takes longer to replicate the log elections happen more frequently and take longer to succeed

124. impact takes longer to achieve consensus takes longer to replicate the log elections happen more frequently and take longer to succeed bonus! leader elections block replication

125. for clients: worst case means worse throughput, worse latency

126. see: sim.tigerbeetle.com

127. sim.tigerbeetle.com

128. best case scenarios

129. imagine processes are stable

130. imagine processes are stable disks are fast

131. imagine processes are stable disks are fast network is fast and reliable

132. impact leader elected quickly

133. impact leader elected quickly leader is stable

134. impact leader elected quickly leader is stable logs replicated quickly

135. for clients: best case means better throughput, lower latency

136. intuition #2: throughput and latency deteriorate as the environment worsens

137. as you add nodes?

138. just add nodes! more nodes means more fault tolerance

139. just add nodes! more nodes means more fault tolerance 5-node cluster only tolerates 2 faults

140. just add nodes! more nodes means more fault tolerance 5-node cluster only tolerates 2 faults 101-node cluster tolerates 50 faults

141. just add nodes! more nodes means more fault tolerance 5-node cluster only tolerates 2 faults 101-node cluster tolerates 50 faults certainly highly available

142. but more communication is its own penalty

143. more nodes, more problems 5-node cluster

144. more nodes, more problems 5-node cluster leader makes 4 requests for every log entry

145. more nodes, more problems 5-node cluster leader makes 4 requests for every log entry waits for 2 responses

146. more nodes, more problems 5-node cluster leader makes 4 requests for every log entry waits for 2 responses 101-node cluster

147. more nodes, more problems 5-node cluster leader makes 4 requests for every log entry waits for 2 responses 101-node cluster leader makes 100 requests for every log entry

148. more nodes, more problems 5-node cluster leader makes 4 requests for every log entry waits for 2 responses 101-node cluster leader makes 100 requests for every log entry Waits for 50 responses

149. we’re as slow as our slowest sub-request

150. Page 17 Designing Data Intensive Applications

151. and consider tail latency

152. and consider tail latency fancy term for variability

153. numbers everyone should know Caption https://static.googleusercontent.com/media/research.google.com/en/us/people/jeff/stanford-295-talk.pdf

154. kind of misleading? Caption https://static.googleusercontent.com/media/research.google.com/en/us/people/jeff/stanford-295-talk.pdf

155. latency is a distribution

156. network

157. Caption https://www.evanjones.ca/network-latencies-2021.html

158. disk

159. Caption fio --name=fiotest --ioengine=sync --size 1Gb --rw=read --bs=1M --direct=1 --numjobs=4 --runtime=60 --startdelay=60 --group_reporting

160. we become less predictable as the number of requests grow

161. intuition #3: throughput and latency deteriorate as the size of the cluster grows

162. how to scale?

163. distributed consensus as a building block some databases shard on top of consensus

164. distributed consensus as a building block some databases shard on top of consensus cockroach, yugabyte, tidb

165. distributed consensus as a building block some databases shard on top of consensus cockroach, yugabyte, tidb each shard is replicated with consensus

166. distributed consensus as a building block some databases shard on top of consensus cockroach, yugabyte, tidb each shard is replicated with consensus at the same time, some do not!

167. distributed consensus as a building block some databases shard on top of consensus cockroach, yugabyte, tidb each shard is replicated with consensus at the same time, some do not! tigerbeetle, etcd, consul

168. example: cockroach

169. https://github.com/cockroachdb/cockroach/blob/master/docs/design.md

170. key range sharding

171. https://github.com/cockroachdb/cockroach/blob/master/docs/design.md

172. each shard is replicated

173. so what does “distributed” mean? horizontal scaling means sharding

174. so what does “distributed” mean? horizontal scaling means sharding distributed means sharding?

175. so what does “distributed” mean? horizontal scaling means sharding distributed means sharding? not necessarily

176. so what does “distributed” mean? horizontal scaling means sharding distributed means sharding? not necessarily consensus means sharding?

177. so what does “distributed” mean? horizontal scaling means sharding distributed means sharding? not necessarily consensus means sharding? definitely not

178. intuition #4: distributed consensus has nothing to do with horizontal scaling

179. recapping

180. intuition takeaways #0: a single node is not highly available

181. intuition takeaways #0: a single node is not highly available #1: distributed consensus is not free

182. intuition takeaways #0: a single node is not highly available #1: distributed consensus is not free #2: latency and throughput of distributed consensus get worse as the environment worsens

183. intuition takeaways #0: a single node is not highly available #1: distributed consensus is not free #2: latency and throughput of distributed consensus get worse as the environment worsens #3: latency and throughput of distributed consensus get worse as the size of the cluster grows

184. intuition takeaways #0: a single node is not highly available #1: distributed consensus is not free #2: latency and throughput of distributed consensus get worse as the environment worsens #3: latency and throughput of distributed consensus get worse as the size of the cluster grows #4: distributed consensus has nothing to do with horizontal scaling

185. with thanks to alex miller (@alexmillerdb) jack vanlightly (@vanlightly) paul nowoczynski (@00pauln00) daniel chia (@DanielChiaJH) alex petrov (@ifesdjeen)

186. thank you