The Key Value Store should support set, get, delete and conditional update operation
Non-Functional:
Fault tolerance: Data should not be loss if some node fails
Latency: The operations should be within 100 ms.
Performance: The system should handle high volumes of write and read: 50000 QPS
Scalability: The system should be able to handle increased volume.
Consistency: The system should be consistent among nodes
Capacity estimation
The system should support 10 million key-value pairs: 50 bytes for keys, 200 bytes for values, the volume should be 10 TB
The system should support 50000 QPS, and peak traffic can be 5 times, which is 250 k
API design
Set: PUT /set
Request:
byte key
byte value
Response:
bool success
String messages
Get: POST /get
Request:
byte key
Response:
byte value
String message
Delete: POST /delete
Request:
byte key
Response: string message
Conditional update: POST /update
Request:
byte key
Database design
data storage: Data can not be stored on the single machine, so we need data partition; To prevent data from loss, we need data replication. To keep consistency of data replication, we can use the leader-follower pattern to first update data to leader, and asynchronously update to followers to meet the eventual consistency requirement. When a leader fails, we need a leader election algorithm to select a new leader. This election and service discovery can be kept in Zookeeper. The data sharding can be determined by the hash ring, both key and service are mapped on the same hash ring, and only the closest leader handles the hash ring. The calculation of hash ring should be done at the API gateway.
byte key
byte data
datetime modified_at
High-level design
API gateway: load balancing, rate limiting, authentication, node discovery
Request handler: handles the operation
Data Storage: Keeps the key value pairs, considering the large size, requires sharding and considering the fault tolerant requirement, requires replica.
Storage cache: in memory cache for frequently accessed keys
Coordination service: The data storage and storage cache are distributed, so the coordination service can be used to determine the acknowledgement strategy and the leader election. Coordination service also holds the shard map.
Monitoring service
Request flows
User sends the request to API gateway
API gateway checks authentication and rate limiting.
API gateway routes the request to a request handler.
Request handler translates the request to the certain operation, and looks up the key in the shard map in zookeeper
Request handler takes certain operations to the data storage.
After confirmation that the operation is success, request handler get back to user
Detailed component design
Data storage: As mentioned earlier, to reach higher throughput, the storage can be split in shards. for the sake of fault tolerance, shards can be replicated. Each node holds several replicas from different shards. The coordination service is responsible for maintaining the shard map and also elect the leader among the nodes holding the same replica of a shard. Once the leader is not responding, coordination service need to elect a new leader and bring up a new node and copy the replica to the new node. We can use the consistent hash ring to determine the sharding, we hash the node and keys to the same hash ring, and each node handles the range of the following range. This way adding a new node or removing a node only a small set of keys need to be rehashed to other nodes. This improves the scalability.
To handle the high QPS, we also need multiple instances of Request handler, the load balancing can be based on CPU cost.
Storage cache: We can use the fingerprint of a kay as the cache key. It is unique, determined and more compact than the original key to be more memory efficient. The invalidation algorithm I think LRU cache meets the requirements in this case. The request handler first looks up the key in the cache layer, if found update the access time and return. Otherwise first lookup in the storage layer, then write it to the cache.
Trade offs/Tech choices
Keeping hash locally at request handler vs a separate cache layer: a separate cache layer enables the independant scalability. Also makes request handler stateless.
Chose leader-follower pattern over quorum based solution to meed consistency because we have a comparatively strong latency requirement.
Failure scenarios/bottlenecks
Read and write conflict: introduce version number to solve the conflict
Request handler fail before write the data in storage: retry at client
Request handler fail after data is written to the storage: introduce the transaction and introduce the idempotation process: when it happens, revert the operations.
Future improvements
What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?