System requirements

Functional:

• User can create, delete, update and read large files
• System is optimized for large files (MB - GB)
• System is optimized for appending writes and sequential reads, with rare random reads and writes

Non-Functional:

• System is high available (99.99%)
• System is strongly consistent
• System support high read and write throughput
• Scale to 10M DAU, billions of files

Capacity estimation

• 10M DAU
• Assume 1B files of 100MB each = 100PB
• Assume each user writes once daily, and read 5 times
• 10M / 86400 = 100 writes/sec, and 500 reads/sec

API design

createDirectory(name, parent)

deleteDirectory(path)

listDirectory(path)

open(name, parent, mode) -> File // mode = overwrite | append

write(file, content)

read(file, offset, length)

close(file)

delete(path)

Database design

// Table for directory hierarchy

• id (PK)
• name
• parentId (self reference)
• type (file | dir)
• owner
• createTime

// Table for file metadata

• id (PK)
• size
• chunkSize (64MB)
• chunks [(chunkNumber, chunkHash]

// In memory chunk information

• chunkHash
• chunkServers [server1, server2, server3]
• fileId
• offset
• length

Assume 200B per chunk, and each chunk is 64MB, total metadata size is 100PB / 64MB * 200 = 320GB which can fit a DB and can also into memory of one server.

High-level design

API Gateway

• Responsible for authentication, rate limiting, etc.

API server

• Implements the file system API

Metadata

• Persisted DB that stores file system hierarchy and file metadata

ChunkManager

• Keeps track of information of Chunk to ChunkServer mapping

ChunkServer

• The file servers that stores the file content

MonitoringService

• Checks for failed chunks servers, corrupted files, and over-busy servers in the background.
• Instructs the ChunkManager to migrate or replicate chunks.

Request flows

Directory management

• Client talks to APIServer via APIGateway
• API server validates the request (e.g. parent directory must exist, user has permission, etc.)
• It then updates the directory hierarchy table in the Metadata DB

Create file

• Client talks to API server by opening a file for write
• API server validates request (e.g. it's not a directory)
• It then creates a new file DB, and obtain a lease (i.e. distributed lock) for that client.
• Based on the size, the APIServer creates N chunks for the file in the DB. It then asks ChunkManager to allocate the first m chunks with each chunk replicated on 3 servers.
• API server returns first m chunk information to the client, including offset and length
• Using this information, the client writes to the ChunkServer directly.
• Client keeps asking for more chunks as it finishes uploading known chunks.

Read file

• Client opens a file for read. API server validates the correctness of the parameters (e.g. file must exist)
• API server asks the ChunkManager for the chunk server that should serve the request for each chunk. Then it returns this information to the client.
• The client reads the chunks directly from ChunkServer.

Detailed component design

ChunkManager

• It know the mapping of chunks to the 3 replicated servers.
• In order for it to have the most updated information, the ChunkServers should periodically report their status to the ChunkManager, including what chunks are present, and disk utility, I/O utility, etc.
• When placing new chunks, ChunkManager should try to pick healthy, under-utilized ChunkServers (e.g. low disk utilization, best recent performance).
• When picking ChunkServers for read, it should consider the health the server, and recent load on the service.

Write consistency

• When writing a file, the client acquires a distributed lock via the APIServer, with a TTL that can be extended by heartbeat
• When replicating, in order to speed up the write, we can consider a pipelined approach (closest server1 -> server2 -> server3)
• The write is considered successful when quorum write is successful (2 out of 3)
• Therefore for each chunk, the ChunkManager should pick a master server (server1) that coordinates the write
• In this design we disable concurrent writes

Trade offs/Tech choices

Instead of using a database to store metadata, we can also store it in memory. In this case we want to have write-ahead log, with replication and periodical checkpointing, so we can quickly reconstruct the state if the in-memory state goes down.

Failure scenarios/bottlenecks

APIServer is stateless and can be horizontally scaled.

The Metadata is a relational database. We should enable high availability mode (read replicas, active-active or active-passive with strong consistency). We can improve reliability by partitioning the database by subtrees of the directory hierarchy.

ChunkManager has information about chunks in memory (e.g. Redis). We should have a standby.

Assume each ChunkServer can store 100TB of data. We need 100PB / 100TB = 1000 servers. ChunkServers might fail, and new servers are can be brought up. We can consider putting the ChunkServers on a consistent hash ring to minimize the amount of chunks that needs to be migrated.

Hotspot problem

• The chunk manager should be smart about load balancing reads for the same chunk
• Additional caching can alleviate the load (e.g. cache in memory / SSD)

Future improvements

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