4 — Capacity estimation

An estimate decides the shape of a design before anything is built. It shows whether a system is one server or a hundred, whether the data fits in memory, and which part runs out first. Chapter 3 introduced the idea with one short example. This chapter covers each quantity in turn, and then works two examples from start to finish.

An estimate only needs to be right to within a factor of two or so. What matters more is that every input is written down, so that each one can be checked and replaced with a measurement later. In this chapter, anything that isn’t measured or sourced is labeled as an assumption.

The chain of quantities

Most estimates follow the same chain, where each step feeds the next.

users and their actions
  -> requests a day -> average rate -> peak rate
       -> reads and writes
            -> storage growth (with replication, indexes, backups)
            -> bandwidth in and out
            -> working set -> cache size -> load left for the database
            -> servers needed (with headroom)
            -> concurrency (Little's law) -> threads, connections, pools

Requests per second

The starting point is usually daily active users (DAU) and how many actions each one takes:

$\text{average rate} = \frac{\text{DAU} \times \text{actions per user per day}}{86{,}400}$

There are 86,400 seconds in a day. Rounding to 100,000 makes mental arithmetic easy and understates the rate by about 14%, which is fine for a first pass.

The peak factor

Traffic isn’t spread evenly. It follows the users’ day, and events cause sharper spikes. A system has to handle its peak, not its average, so the estimate needs a peak factor:

$\text{peak rate} = \text{average rate} \times \text{peak factor}$

The peak factor is an assumption until there’s traffic to measure, and it varies a great deal by product. A better way to state it is the share of daily traffic in the busiest hour, which can be read straight from logs once they exist. If the busiest hour carries a share $h$ of the day’s requests:

$\text{peak factor} = \frac{h \times 86{,}400}{3{,}600} = 24h$

The busiest hour still hides spikes within it, such as the top of the hour or a push notification sent to every user at once. A design that expects such spikes should either size for them or shed load gracefully (chapter 16).

Read and write ratio

Reads and writes scale in different ways, so they should be estimated separately. Reads can be served from caches and replicas. Writes all go to a primary, unless the data is partitioned (chapter 11).

A product where each user views 30 pages a day and posts twice has a read-to-write ratio of 15 to 1. A logging or metrics pipeline is the opposite: nearly all writes, read by a few dashboards. The ratio decides where the design effort goes. A read-heavy system needs caching and replicas, and a write-heavy one needs batching, partitioning and cheap appends.

Storage growth

Storage is the rate of new records multiplied by their size and by how long they’re kept. The size of a record is where estimates usually go wrong, because the stored size is larger than the data.

Record size

In PostgreSQL, every table row has a fixed header of 23 bytes on most machines, and each row on a page also needs a 4-byte item identifier pointing to it. Tables and indexes are stored in pages of 8 kB, and columns are aligned, which adds padding. So a row with 300 bytes of text and four 8-byte columns takes about:

$23 + 4 + 300 + 4 \times 8 = 359 \text{ bytes}$

before padding and free space. For small rows, the overhead is a large share: a row with 40 bytes of data takes 67 bytes, so 40% of what’s stored is overhead. Other stores have their own overheads, and the reliable way to know is to load a sample of realistic data and measure it, in PostgreSQL with pg_total_relation_size.

The multipliers

The general form is:

$\text{storage} = \text{records per day} \times \text{bytes per record} \times \text{days kept} \times (1 + \text{index share}) \times \text{copies}$

Backups are usually kept on separate, cheaper storage, so they’re best estimated separately.

Bandwidth

Bandwidth is the request rate multiplied by the bytes per request, in each direction:

$\text{egress} = \text{peak read rate} \times \text{response size}$

Network links are rated in bits a second, and storage and memory in bytes, so a factor of 8 appears when moving between them. 167 MB/s of responses is 1.33 Gbit/s. Protocol headers, TLS records and retransmissions add to the payload, so a link shouldn’t be planned near its rated speed. A CDN removes cached responses from the origin’s egress altogether (chapter 5).

Cache size

A cache only has to hold the working set, the data that’s actually being requested, not all of the data. How much of the working set it needs depends on how uneven the requests are.

Request popularity in many systems is close to a Zipf distribution: the item of rank $r$ is requested with probability proportional to $1/r^s$. The exponent $s$ sets how skewed it is. A larger $s$ means the top items get a bigger share.

The often-quoted 80/20 rule, that caching 20% of items serves 80% of requests, is one point on one such curve. Whether it holds depends on $s$ and on the number of items. Simulating the cache is more reliable than assuming it.

Simulating it

This program measures two things for one million items. The ideal hit ratio is what a cache gets if it always holds the most popular items. The LRU hit ratio is what a real least-recently-used cache achieves, after a warm-up long enough to fill it.

import random
from collections import OrderedDict
from itertools import accumulate

N = 1_000_000                      # distinct items
SHARES = [0.001, 0.01, 0.05, 0.10, 0.20]

def ideal_hit_ratio(cum, k):
    """Cache permanently holds the k most popular items."""
    return cum[k - 1] / cum[-1]

def lru_hit_ratio(cum, k, requests, warmup, rng):
    cache, hits = OrderedDict(), 0
    keys = rng.choices(range(N), cum_weights=cum, k=warmup + requests)
    for i, key in enumerate(keys):
        if key in cache:
            cache.move_to_end(key)
            if i >= warmup:
                hits += 1
        else:
            cache[key] = True
            if len(cache) > k:
                cache.popitem(last=False)
    return hits / requests

for s in (0.8, 1.0, 1.2):
    cum = list(accumulate(1 / r**s for r in range(1, N + 1)))
    rng = random.Random(42)
    for share in SHARES:
        k = int(N * share)
        print(f"s={s}  cache {share:>6.1%}  ideal {ideal_hit_ratio(cum, k):6.1%}  "
              f"LRU {lru_hit_ratio(cum, k, 2_000_000, 2_000_000, rng):6.1%}")

The results show three things.

1. The exponent matters more than anything else. Caching 1% of items served 23% of requests at $s = 0.8$ and 88% at $s = 1.2$. The same cache size can be nearly useless or nearly perfect.
2. The 80/20 rule is one case among many. Caching 20% of items gave between 61% and 98% of requests here.
3. LRU falls short of ideal, by about 13 points at the most in this run, because it sometimes evicts a popular item to make room for one that’s requested once. Policies that track frequency close part of the gap (chapter 7).

Sizing for a target

With $s = 1$, the ideal hit ratio has a simple form. Caching the top $k$ of $N$ items gives $H(k) / H(N)$, where $H(n) = 1 + 1/2 + \dots + 1/n$, and $H(n) \approx \ln n + 0.5772$ for large $n$. The approximation agrees with the exact sum for one million items to six decimal places. Solving for $k$ gives the number of items needed for a target:

import math
G = 0.5772156649                       # Euler's constant
def harmonic(n):
    return math.log(n) + G + 1 / (2 * n)

for n in (1_000_000, 10_000_000, 100_000_000):
    for target in (0.80, 0.90, 0.95):
        k = math.exp(target * harmonic(n) - G)    # solve H(k) = target * H(n)
        print(n, target, round(k), f"{k / n:.2%}")

Each step up in hit ratio costs much more than the last. For 100 million items, going from 80% to 90% needs almost seven times the cache, and 95% more than twice that again. These are ideal numbers, so a real LRU cache needs somewhat more.

Finding the exponent

Once there are request logs, $s$ can be measured instead of assumed. Count requests per item, sort the counts in descending order, and plot the logarithm of the count against the logarithm of the rank. A Zipf workload gives roughly a straight line, and its slope is $-s$. Real workloads often bend at the ends, so the fit should cover the range of ranks that the cache will actually hold.

What the cache leaves for the database

The hit ratio sets the load that reaches the database:

$\text{database reads} = \text{read rate} \times (1 - \text{hit ratio})$

So a cache sized for 90% instead of 95% doubles the database’s read load. The database has to be sized for the misses, and also for the cache being empty after a restart, when the miss rate is briefly 100%.

Servers

The number of application servers follows from the peak rate and what one server can handle at an acceptable latency:

$\text{servers} = \left\lceil \frac{\text{peak rate}}{\text{capacity per server} \times \text{target utilization}} \right\rceil$

Capacity per server should come from a load test of the real service, not from a guess. Chapter 2 measured a minimal Python server at 135,854 trivial requests a second on one core, falling to 3,836 when each request did 290 µs of CPU work. A production service with a database call, TLS and logging will be lower again, and the load test is the only way to know by how much.

Target utilization is kept well below 100%, because latency rises steeply as a server approaches full load (chapter 10). Somewhere around 50% to 70% at peak is a common choice, and the right value depends on how fast latency degrades for the particular service.

Headroom for failure is a separate requirement. If servers run in three zones and the design must survive losing one, the remaining two zones have to carry the full peak. Each zone is then sized for half the peak, not a third, which means 1.5 times the servers that normal operation would need.

Concurrency and Little’s law

Little’s law (chapter 10) links the arrival rate $\lambda$, the time each request spends in the system $W$, and the average number in the system $L$:

$L = \lambda W$

For estimation, it turns rates and latencies into concurrency, and concurrency is what limits threads, connections and pool sizes. At 2,000 requests a second and 50 ms each, 100 requests are in progress on average.

The law holds whatever the distribution of request times, but it gives the average. Pools have to be sized for the peaks around that average. A simulation with random arrivals at 2,000 a second and skewed request durations shows the gap:

import random, statistics
rng = random.Random(7)
RATE, T = 2_000, 200.0                     # requests per second, seconds simulated
events, t, durations = [], 0.0, []
while t < T:
    t += rng.expovariate(RATE)
    d = rng.lognormvariate(-3.4, 0.8)      # median about 33 ms, long right tail
    durations.append(d)
    events += [(t, +1), (t + d, -1)]
events.sort()
area, inflight, last, peak = 0.0, 0, 0.0, 0
for when, step in events:
    if when > T:
        break
    area += inflight * (when - last)
    inflight += step; last = when; peak = max(peak, inflight)
print(RATE * statistics.fmean(durations), area / T, peak)

The average matched the law. The peak was 50% higher, so a pool of 92 would have been too small at the busiest moments of those 200 seconds. Latency matters as much as rate here: if a slow dependency doubles $W$, the number in flight doubles at the same request rate, and a pool sized for normal latency runs out.

Little’s law applies at every layer. For a database behind a cache, the rate is the miss rate and the time is the query time, which usually gives a much smaller number than the application servers need.

Numbers worth knowing

These are the numbers used in this track’s estimates. Measured ones come from the programs in chapter 1 and chapter 2, run on one Apple M3 Pro laptop, and will differ on other hardware. Sourced ones come from the named specification or documentation.

Worked example 1: comments on a content site

A site lets users read and post comments under articles.

Inputs

Rates

Storage

A comment row holds the text and four 8-byte columns: an ID, the article ID, the user ID and a timestamp. With PostgreSQL’s row header and item identifier, that’s about 359 bytes.

A year of comments fits on one large PostgreSQL primary, with its replicas. Growth is the issue: after two or three years, the table is large enough that partitioning by time, archiving old threads or sharding by article (chapter 11) needs planning. Nothing has to be built for that on day one, but the choice of keys should leave it possible.

Bandwidth

Peak egress is 20,833 × 8 KB = 167 MB/s, or 1.33 Gbit/s, before protocol overhead. That’s spread across the application servers and doesn’t constrain the design.

Cache

Comments for one article are read together, so the cache holds one entry per article: a rendered page of comments, 8 KB. With 50 million threads and $s = 1$, the sizing formula gives:

A cache of around 64 GB, on one large node or spread across a few, takes the database from 20,833 reads a second to about 2,000. That figure is for an ideal cache, and the simulation above showed LRU doing several points worse at the same size, so the plan should allow either more memory or more database reads. The exponent is the biggest uncertainty. If real traffic turns out to have $s = 0.8$, the same cache serves far less, so measuring $s$ from the first weeks of logs is worth doing early.

Servers

The capacity per server is an assumption here, and stands in for a load test: 2,000 requests a second per server at acceptable latency.

For normal operation at 60% utilization:

$\left\lceil \frac{20{,}833}{2{,}000 \times 0.6} \right\rceil = \lceil 17.4 \rceil = 18 \text{ servers}$

To survive the loss of one of three zones at peak, with the remaining servers at no more than 80%, the two surviving zones must carry 20,833 requests a second between them, or 10,417 each:

$\left\lceil \frac{10{,}417}{2{,}000 \times 0.8} \right\rceil = \lceil 6.5 \rceil = 7 \text{ servers per zone}$

That’s 21 servers in all. On a normal day they run at about 20,833 ÷ (21 × 2,000), or 50% utilization, which is the cost of surviving a zone failure.

Concurrency

With 40 ms average request time (an assumption), the application tier has $20{,}833 \times 0.04 \approx 833$ requests in flight at peak, or about 40 per server. That’s comfortable for a thread pool or an event loop, with room for the peaks above average.

At the database, cache misses at 90% arrive at about 2,083 a second. At 5 ms per query (an assumption), that’s about 10 queries in flight on average. A connection pool of a few dozen across the whole fleet is enough, far below PostgreSQL’s default limit of typically 100 connections. The danger is the cold cache: at a 0% hit ratio, the database would see 20,833 reads a second, ten times the planned load, so a restarted cache needs warming or a gradual return of traffic (chapter 7).

Writes peak at 1,389 a second. On the laptop in chapter 2, a durable flush took 2.9 ms, about 345 flushes a second. At 1,389 commits a second, group commit would need about four commits to share each flush. The server’s own flush time should be measured with pg_test_fsync, and the write rate confirmed with pgbench.

Result

A modest system: one PostgreSQL primary with two replicas, a cache of about 64 GB, and around 21 application servers across three zones. The numbers that most need checking are the server capacity, the Zipf exponent and the index share.

Worked example 2: device telemetry

A fleet of devices sends a reading every few seconds, and the readings are kept for a month for dashboards and alerts.

Inputs

Rates

The traffic here doesn’t follow a daily cycle, because devices report all day. The peak comes from bursts instead.

This is a write-heavy system, the opposite of the first example.

Bandwidth

Ingest is 200,000 × 200 bytes = 40 MB/s, or 320 Mbit/s, and 640 Mbit/s in a burst. That’s well within a single 10 Gbit/s link, before protocol overhead.

Storage

This doesn’t fit on one machine. There are three levers, and an estimate shows which ones are needed:

• Compression. Readings from one device change slowly, so time-series data usually compresses well. The ratio has to be measured on real readings. At an assumed 10×, three copies take 31 TB.
• Downsampling. Keeping full resolution for a few days and per-minute averages after that cuts the older data by a factor of 6, at one reading per 10 seconds.
• Partitioning. Splitting the data by device across several machines (chapter 11), each holding a share.

Row overhead matters for small records. Stored as one PostgreSQL row per reading, each 200-byte reading would carry 27 bytes of header and item identifier, adding 13.5%. Storing readings in batches, one row per device per minute, spreads that overhead across six readings. A time-series store or a columnar format avoids most of it.

Writes and flushes

At one commit per reading, 200,000 commits a second would need 200,000 flushes a second, or extremely good group commit. Batching changes the picture completely:

At 1,000 readings per commit, 200 commits a second is within even the slow 345 flushes a second measured on the laptop in chapter 2. So the ingest path should collect readings into batches, usually in a queue or log in front of the store (chapter 8). Batching adds latency: a reading waits for its batch, so the batch size is bounded by how fresh the dashboards must be.

Concurrency

If each device sends its reading over a short request that takes 20 ms at the gateway (an assumption), Little’s law gives $200{,}000 \times 0.02 = 4{,}000$ requests in flight on average, and 8,000 in a burst. If devices keep a connection open instead, the gateways hold 2 million connections, whatever the request rate. Either way, this calls for event-loop servers, which handle many idle connections cheaply, rather than a thread per connection (chapter 2).

Result

Network bandwidth is no concern. The two hard parts are the write rate, which needs batching, and the storage volume, which needs compression, downsampling or partitioning, probably all three. The design should start from measured compression ratios and a decision on how long full resolution is really needed.

Common errors

The takeaway

An estimate is a chain from users to rates, storage, bandwidth, cache, servers and concurrency, and every link is a number that can be written down and checked. Peak factors and Zipf exponents are assumptions until they’re measured, and the cache size a workload needs depends on its exponent far more than on any rule of thumb. Little’s law turns rates into concurrency, but it gives an average, and the limits are hit at the peaks.