System requirements

Functional:

• Support operations:
• Set
• Get
• Delete
• Conditional update

Non-Functional:

1. fast data access, high scalability, and high availability.
2. Low latency
3. Support data replication across multiple nodes to ensure fault tolerance
4. Support partitioning to distribute data evenly across the cluster
5. Support persistence to safegaurd data
6. The system must efficiently handle high volumes of read/write operations and provide consistency and durability guarantees
7. Support 1M concurrent users

Capacity estimation

Read / write

1. Each user makes average 100 read/write requests per day, then there are 100M requests
2. QPS = 100M / 24/ 3600 = 1200
3. For storage size, assuming the read write ratio is 4/1, then there are 20M write request, we assume average data size is 50M byte, then per day is 20M * 50 M = 1 PB
4. For 50 years, the storage size is 50 * 365 * 1 PB = 18250 PB

For handling the current QPS and storage scale as well as considering future scalability, we'll use distributed system architecture to design the k-v store

API design

1. Set
2. post v1/set
3. header: authorization: aut_token
4. body: {user_id, key, value}
5. Get
6. get v1/get:user_id?key=xxx
7. header: auth
8. Delete
9. post v1/delete
10. header: auth
11. body: {user_id, key}
12. Conditional update
13. post v1/conditional_update
14. header: auth
15. body: {user_id, key, value, condition}

Database design

• Data schema
• user_id
• key
• value
• The data is stored in memory for high availability
• For data not being accessed recently, they'll be evicted out and being persisted to hard disk of each machine. We'll discuss this later in detail.

High-level design

Client

• User send KV operations through client by sending APIs

Load Balancer

• Load balancer is responsible to route requests to dedicated machines, algorithm such as round robin can be used for load balancing

API Gateway

• Route API requests to a coordinator node
• Rate limiting
• Authentication check

Coordinator Nodes

• Find the key is on which kv node and send API request to the node for actual operation
• Maintain the consistent hash ring structure of kv nodes
• Check availability of each kv node, detect whether a node goes down or not

KV nodes 1...n

• KV nodes 1...n is structured as a consistent hashing ring, the direction is clock-wise
• The key is hashed, each node is responsible for handle hash value range starting from itself and clock wise to the start value of next node.

Request flows

Set value

1. The v1/set request is routed to a coordinator node after load balancer and API gateway
2. The coordinator checks the and user_id and key to see which kv node is responsible to handle the request on the consistent hashing ring, then send the set request to the kv node
3. The KV node checks the key exits or not, if exists, then update the value to the new value, if not, then add the key-value pair.
4. After the operation succeeds, the kv node sends a response back to coordinator and the coordinator send successful response to the client with http status 200 OK

Get value

1. The v1/get request eventually lands on a coordinator node after routing, the coordinator then checks the key to see which KV node is responsible for the key
2. Then coordinator route the request to the corresponding kv node
3. The kv node checks whether the key exists or not, if found, return the value to coordinator then as a successful response the client. If not found, the response should be successful but with information indicating not found
4. The kv node also first checks if the key is in memory or not, if not then checks if it's persisted on hard disk, if found on hard disk, the k-v pair will be loaded into memory to reduce latency for later calls

Delete value

1. The v1/delete api request is routed to the specific kv node after routed by load balancer, api gateway and coordinator.
2. The node checks if the key exists or not, if exists, the node delete it from both memory and hard disk and return deletion success response to the client

Conditional update

1. The v1/conditional_update request eventually land on the corresponding kv node.
2. The kv node checks whether the key exits or not, if exits, then checks whether the condition meets, if the condition meets, then update the value and return success response

Detailed component design

1. Node structure detail

A KV node machine contains below component:

1. Node service
2. Responsible to receive requests
3. Manges data in memory, and hard disk
4. Write and manage commit logs for merge conflict or data loss recovery
5. KV data in Memory
6. Cached KV data in memory to reduce latency and increase availability. Each data entry can have a TTL field so that old data without accessing will be evicted
7. KV data persisted on hard disk
8. The KV data is persisted in hard disk as SSTable (sorted string table) as key-value pair strings. The SSTable is commonly used data structure by popular noSQL database such as cassandra
9. Commit logs for operation history
10. Operation logs, it can be used to handle merge conflict and data loss recovery

2. Sharding strategy - consistent hashing

1. KV nodes are structured as a consistent hashing ring in the hash range 2^64-1, to further make sure the data is evenly distributed to each node, we can map each physical node to 3 virtual node on the hash ring
2. The coordinator node stores the hash key range data about each kv node, it also stores the mapping between physical node and virtual node so that when a request for a key comes, the coordinator knows which node should handle the request

3. Replication

Master / slave replication

1. For each node, in order to achieve high availability we uses master / slave replication. All write requests, i.e. v1/set and v1/conditional_update, v1/delete are handled by master and all read requests are handled by slaves, there are multiple slaves to increase the availability of read
2. The write operation of master will eventually propagate to slave to ensure eventual consistency

4. Support data replication across multiple nodes to ensure fault tolerance

1. To ensure fault tolerance, we can support data replication across multiple nodes, we can choose the next 3 nodes on the hash ring as the replication node of the current node, 2 to be used as write replica and 2 to be used as read replicas
2. As 2+2 > 3, we'll ensure eventually consistency on the node

Trade offs/Tech choices

Explain any trade offs you have made and why you made certain tech choices...

Failure scenarios/bottlenecks

1. Fault tolerance

• If one node goes down
• Quorum consensus can be used on the consistent hash ring to detect whether a node is down
• If a node is down, the coordinator needs to check it's a temporary down or a permanent down.
• A node is normally marked as temporary down first, in such case, the coordinator will try to use corresponding write replica and read replica on the hash ring to handle operations as backup
• If a node goes down too long, the coordinator will detects it and raise an on-call alert to notify related engineering team to respond to the incident
• If it's a permanent down, the data on the node needs to be migrated, in consistent hashing, we can find the node next to the current node anti-clock-wise, as the current goes down, it's removed from the ring, all data needs to be migrated to that next node
• If master replication goes down for a node
• One of the slave is promoted to master to handle write request

Future improvements

What are some future improvements you would make? How would you mitigate the failure scenario(s) you described above?