c5P4

Design object storage similar to Amazon S3. S3 launched 2006; 100+ trillion objects by 2021.

Storage System 101

• Block: raw blocks as volumes, physically or network-attached (FC, iSCSI). Single-server ownership.
• File: built on block storage, hierarchical directories, shared via SMB/NFS.
• Object: flat structure, RESTful API, "cold" data, archival/backup. Sacrifices performance for durability + scale + low cost.

Terminology:

• Bucket: logical container, globally unique name
• Object: payload + metadata (name-value pairs)
• Versioning: multiple variants of object in bucket
• URI: uniquely identifies resources
• SLA: e.g. S3 Standard-IA: 99.999999999% durability, 99.9% availability

Step 1 - Understand the Problem and Establish Design Scope

Requirements:

• Bucket creation, object upload/download, versioning, listing objects in bucket
• Both massive objects (few GBs+) and many small objects (tens of KBs)
• 100 PB data per year
• Durability: 6 nines (99.9999%), Availability: 4 nines (99.99%)
• Storage efficiency (reduce costs while maintaining reliability/performance)

Back-of-the-envelope estimation:

• Object distribution: 20% small (<1MB), 60% medium (1-64MB), 20% large (>64MB)
• HDD: 100-150 IOPS (7200 rpm SATA)
• Using median sizes (0.5MB, 32MB, 200MB), 40% storage usage:
  • 100 PB → 10¹¹ MB
  • ~0.68 billion objects
  • Metadata ~1KB each → 0.68 TB metadata

Step 2 - Propose High-Level Design and Get Buy-In

Key properties:

• Object immutability: no incremental changes; replace with new version
• Key-value store: URI = key, object data = value
• Write once, read many: 95% of requests are reads (LinkedIn)
• Support both small and large objects

Analogy: Like UNIX file system — inode (metadata) separate from file data (disk blocks). Object storage: metadata store (object metadata) + data store (object data), connected via object UUID over network.

Figure 2: Unix file system and object store

Figure 3: Bucket & object

High-level design #

Figure 4: High-level design

• Load balancer: distributes RESTful API requests
• API service: stateless, orchestrates IAM + metadata + data store calls
• IAM: authentication + authorization
• Data store: stores/retrieves object data by UUID
• Metadata store: stores object metadata

Uploading an object #

Figure 5: Uploading an object

1. Client: PUT to create bucket → API service
2. API → IAM: verify WRITE permission
3. API → metadata store: create bucket entry
4. Client: PUT to create object script.txt
5. API → IAM: verify WRITE on bucket
6. API → data store: persist payload → returns UUID
7. API → metadata store: create entry (object_name, object_id/UUID, bucket_id)

Sample upload request:

PUT /bucket-to-share/script.txt HTTP/1.1
Host: foo.s3example.org
Authorization: authorization string
Content-Type: text/plain
Content-Length: 4567
x-amz-meta-author: Alex
[4567 bytes of object data]

Downloading an object #

Figure 6: Downloading an object

1. Client: GET /bucket-to-share/script.txt
2. API → IAM: verify READ access
3. API → metadata store: fetch UUID by object name
4. API → data store: fetch data by UUID
5. Return object data to client

Logical hierarchy via naming: bucket-to-share/script.txt implies folder structure (no actual directories).

Step 3 - Design Deep Dive

Data store #

Figure 7: Upload and download an object

Data store components #

Figure 8: Data store components

Data routing service: stateless RESTful/gRPC service. Queries placement service for best data node; reads/writes data to nodes.

Placement service: determines which data nodes (primary + replicas) store an object. Maintains virtual cluster map (physical topology). Ensures replicas are physically separated for durability. Monitors nodes via heartbeats (15s grace period). Critical → 5-7 nodes with Paxos/Raft consensus (tolerates minority failure).

Figure 9: Virtual cluster map

Data node: stores actual object data. Replicates to multiple nodes (replication group). Daemon sends heartbeats (how many drives, how much data). On first heartbeat, placement assigns: unique ID, virtual cluster map, replication targets.

Data persistence flow #

Figure 10: Data persistence flow

1. API → data store: forward object data
2. Data routing: generate UUID, query placement → get primary data node (via consistent hashing)
3. Send data to primary + UUID
4. Primary saves locally → replicates to 2 secondary nodes → responds only after all replicas succeed
5. Return UUID to API service

Consistency vs latency trade-off:

Figure 11: Consistency vs latency

1. All 3 nodes: best consistency, highest latency
2. Primary + 1 secondary: medium consistency/latency (eventual)
3. Primary only: worst consistency, lowest latency (eventual)

How data is organized #

Storing each object as a separate file fails at scale:

• Wastes disk blocks (4KB block for tiny file)
• Exhausts inode capacity
• OS handles millions of inodes poorly

Solution: Merge small objects into larger files (WAL-like). Append to read-write file. At capacity (~few GBs) → mark read-only, create new read-write file. Serialize writes to read-write file. For multi-core servers: dedicated read-write file per core.

Figure 12: Store multiple small objects in one big file

Object lookup #

To locate object by UUID: need data file name, start offset, object size.

Database choice: Read-heavy (write once, read many). RocksDB (SSTable-based, fast writes) vs. Relational (B+ tree, fast reads). Relational wins (better read perf). Deploy SQLite per data node (local, file-based, isolated per node).

Tables 3-4: Object_mapping table

Updated persistence flow #

Figure 13: Updated data persistence flow

1. API sends object 4
2. Data node appends to /data/c
3. Insert record into object_mapping table
4. Return UUID

Durability #

Hardware failure & failure domains #

• Annual disk failure rate: ~0.81%. 3 copies → 1-(0.0081³) ≈ 99.9999% reliability
• Failure domains: node-level (server components), rack-level (shared switches/power), AZ-level (separate infrastructure)
• Replicate data across AZs for extreme-case resilience

Figure 14: Multi-Datacenter replication

Erasure coding #

Alternative to replication. (4+2) example: data split into 4 equal chunks (d1-d4), 2 parities (p1, p2) computed mathematically. Lose d3, d4 → reconstruct from d1, d2, p1, p2.

Figure 15: Erasure coding

(8+4): 8 data chunks + 4 parities across 12 failure domains. Can lose up to 4 nodes and reconstruct.

Figure 16: (8+4) erasure coding

Storage overhead: replication 3× → 200%. Erasure coding (8+4) → 50%.

Figure 17: Extra space comparison

Durability from 0.81% AFR: erasure coding → 11 nines.

Replication: latency-sensitive apps. Erasure coding: cost efficiency, archival. This design focuses on replication.

Correctness verification (checksums) #

In-memory data corruption common at scale. Append checksum (MD5) to each object + file-level checksum when marked read-only.

Figures 18-20: Checksum generation, comparison, data node layout

Read flow: fetch data + checksum → compute → mismatched? → try other failure domains → reconstruct from 8 valid pieces.

Metadata data model #

Schema #

Figure 21: Database tables

Two tables: bucket (small, ~10GB for 1M users × 10 buckets × 1KB) and object.

Scale bucket table: single server fits. Spread reads via replicas.

Scale object table: shard by hash(bucket_name, object_name) — most operations are URI-based.

Three queries supported:

1. Find object ID by name
2. Insert/delete object by name
3. List objects in bucket sharing prefix

Listing objects in a bucket #

Flat structure, organized by prefixes. Path: s3://mybucket/abc/d/e/f/file.txt → bucket: mybucket, object: abc/d/e/f/file.txt, prefix: abc/d/e/f/.

Three uses:

1. aws s3 list-buckets — list all user's buckets
2. aws s3 ls s3://mybucket/abc/ — list at same prefix level (rolls up deeper prefixes)
3. aws s3 ls s3://mybucket/abc/ --recursive — all objects sharing prefix

Single DB: SELECT * FROM object WHERE bucket_id = "123" AND object_name LIKE 'abc/%'

Sharded DB (distributed): query all shards → aggregate. Paging is complex: each shard returns different counts, offsets vary per shard, cursor must track all offsets. Solution: denormalize listing data into a separate table sharded by bucket_id → isolates listing to single DB. Acceptable trade-off (listing performance is rarely priority).

Object versioning #

Figure 22: Object versioning

Upload flow for versioned object: same as normal, but metadata store inserts new row (not overwrite) with same bucket_id + object_name, new object_id (UUID) + object_version (TIMEUUID). Current version = largest TIMEUUID.

Figure 23: Versioned metadata

Deletion: all versions remain; insert a delete marker as new version. GET on delete marker → 404.

Figure 24: Delete with delete marker

Optimizing uploads of large files — Multipart upload #

Figure 25: Multipart upload

1. Initiate multipart upload → get uploadID
2. Split file into parts (e.g. 1.6GB → 8 × 200MB). Upload each part with uploadID.
3. Each part returns ETag (MD5 checksum)
4. Complete multipart upload: send uploadID + part numbers + ETags
5. Data store reassembles object (may take minutes for large files)

Old parts after reassembly → garbage collected.

Garbage collection #

Reclaims space from: lazy deletion (marked deleted), orphan data (half-uploaded/abandoned multipart), corrupted data (checksum failures).

Compaction: Copy non-deleted objects from read-only files to new file. Skip objects with delete_flag = true. Update object_mapping table (file_name, start_offset). Wrap updates in transaction.

Figure 26: Compaction

Wait until many read-only files accumulate to compact → avoid creating many small files.

Step 4 - Wrap Up

Designed S3-like object storage: storage taxonomy → high-level design (data store + metadata store separation) → deep dive into data persistence (WAL-like merging, SQLite per node), durability (replication vs. erasure coding, checksums), metadata sharding (hash of bucket+object name), listing (denormalized table), versioning (TIMEUUID + delete markers), multipart upload, garbage collection (compaction).