From Zero to Hero: Scaling up to million users

Let's investigate the request flow and traffic source!

1. Users access websites through domain names, such as api.mysite.com. Usually, the Domain Name System (DNS) is a paid service provided by 3rd parties and not hosted by our servers.
2. Internet Protocol (IP) address is returned to the browser or mobile app. In the example, IP address 15.125.23.214 is returned.
3. Once the IP address is obtained, Hypertext Transfer Protocol (HTTP) 1 requests are sent directly to your web server.
4. The web server returns HTML pages or JSON response for rendering.

Next, the traffic source. The traffic to the web server comes from two sources: web, mobile apps, and API.

1. Web: It uses a combination of server-side languages (Java, Python, etc.) to handle business logic, storage, etc., and client-side languages (HTML and JavaScript) for presentation.
2. Mobile: HTTP protocol is the communication protocol between the mobile app and the web server. JavaScript Object Notation (JSON) is commonly used API response format to transfer data due to its simplicity.
3. API: The web server exposes a RESTful API that allows the mobile app to access the data. The API is designed to be stateless, meaning that each request from the client contains all the information needed to process the request.

Separating web/mobile traffic (web tier) and database (data tier) servers allows them to be scaled independently. The web tier can be scaled horizontally by adding more web servers, while the data tier can be scaled vertically by upgrading the database server. This separation also allows for better performance and reliability, as each tier can be optimized for its specific workload.

Which databases to use?

• relational database
• or a non-relational database?

Relational Databases (RDBMS / SQL)

Relational databases store data in tables with rows and columns, using a fixed schema. They support complex join operations through SQL, making them ideal for structured data with well-defined relationships. They are best suited for apps with complex queries and transactions.

Popular relational databases:

• MySQL
• PostgreSQL
• Oracle Database
• Microsoft SQL Server

Use cases include applications requiring strong consistency, ACID transactions, and structured schemas.

Non-Relational Databases (NoSQL)

Non-relational databases, or NoSQL databases, do not rely on fixed table-based schemas and are better suited for unstructured or semi-structured data. Most NoSQL databases do not support joins, making them more scalable but sometimes requiring data denormalization.

They are grouped into four main categories:

• Key-Value Stores: Redis, Amazon DynamoDB
• Document Stores: MongoDB, CouchDB
• Column Stores: Apache Cassandra, HBase
• Graph Stores: Neo4j

Use cases include scalability, flexible schema, high-throughput, and big data scenarios.

Handling Failures in Replication

In the previous section, we discussed how a load balancer improves system availability. We ask a similar question here: what if one of the databases goes offline?

The architecture in Figure 5 addresses this:

• If a single slave database goes offline:
  Read operations are temporarily redirected to the master database.
  A new slave database is then added to replace the failed one.
• If multiple slave databases are available and one fails:
  Read operations are redirected to other healthy slave databases.
  The failed database is replaced with a new one.
• If the master database goes offline:
  A slave database is promoted to become the new master.
  All database operations temporarily execute on the promoted node.
  A new slave is added for replication purposes.

This caching strategy is called a read-through cache. Other caching strategies are available depending on the data type, size, and access patterns.

Interacting with cache servers is simple because most cache servers provide APIs for common programming languages. The following code snippet shows typical Memcached APIs:

  SECONDS = 1
  cache.set('myKey, 'hi there', 3600 * SECONDS)
  cache.get('myKey')

The cache tier is a high-speed, temporary data storage layer positioned between the application and the database. It serves frequently accessed or computationally expensive data, reducing the need to query the database repeatedly.

Benefits of a dedicated cache tier include:

1. Faster performance – Data loads quicker from cache than DB.
2. Lighter DB load – Fewer queries free the DB for key tasks.
3. Scales independently – Cache can grow without touching DB.
4. Cost savings – Cuts need for costly DB resources.
5. Flexible tech – Use Redis, Memcached, or in-memory stores.
6. Fault-tolerant – If cache fails, app still works via DB.
7. Efficient scaling – Cache scales alone for better performance.

1. When to Use Cache: Caching is ideal when data is read frequently but modified infrequently. Since cache data is stored in volatile memory, it is not suitable for persistent storage. For example, if a cache server restarts, all stored data is lost. Therefore, critical or persistent data should always be saved in a durable data store.
2. Expiration Policy: Implementing an expiration policy is essential to manage memory efficiently. When cached data expires, it is removed from memory. Without expiration, cached items remain indefinitely, which can waste resources. A good expiration policy strikes a balance: Too short: Frequent reloads from the database, adding load and latency. Or Too long: Data may become stale or outdated.
3. Data Consistency: Maintaining consistency between the cache and the data store is challenging. Since cache and database operations are often not performed in a single transaction, discrepancies may occur. This issue becomes more complex in distributed systems across multiple regions. For in-depth discussion, refer to the paper “Scaling Memcache at Facebook”.
4. Mitigating Failures: A single cache server introduces a Single Point of Failure (SPOF) - if it goes down, the entire system can be affected. To mitigate this risk: Deploy multiple cache servers across different data centers. Or Overprovision cache memory by a safety margin to account for growth or unexpected spikes.
5. Eviction Policy: When the cache is full, older or less important items must be removed to make space. This process is known as cache eviction. Common eviction policies include: LRU (Least Recently Used) – Removes items that haven’t been accessed recently. LFU (Least Frequently Used) – Removes items accessed the least. FIFO (First In, First Out) – Removes items in the order they were added.

Dynamic content caching is a relatively new concept and beyond the scope of this course. It enables the caching of HTML pages that are based on request path, query strings, cookies, and request headers. Refer to the article mentioned in reference material 9 for more about this. This course focuses on how to use CDN to cache static content.

Here is how CDN works at the high-level: when a user visits a website, a CDN server closest to the user will deliver static content. Intuitively, the further users are from CDN servers, the slower the website loads. For example, if CDN servers are in San Francisco, users in Los Angeles will get content faster than users in Europe.

1. User A tries to get image.png by using an image URL. The URL’s domain is provided by the CDN provider. The following two image URLs are samples used to demonstrate what image URLs look like on Amazon and Akamai CDNs:

• https://mysite.cloudfront.net/logo.jpg
• https://mysite.akamai.com/image-manager/img/logo.jpg

2. If the CDN server does not have image.png in the cache, the CDN server requests the file from the origin, which can be a web server or online storage like Amazon S3.
3. The origin returns image.png to the CDN server, which includes optional HTTP header Time-to-Live (TTL) which describes how long the image is cached.
4. The CDN caches the image and returns it to User A. The image remains cached in the CDN until the TTL expires.
5. User B sends a request to get the same image.
6. The image is returned from the cache as long as the TTL has not expired.

1. Static assets (JS, CSS, images, etc.,) are no longer served by web servers. They are fetched from the CDN for better performance.
2. The database load is lightened by caching data.

• Cost: CDNs are run by third-party providers, and you are charged for data transfers in and out of the CDN. Caching infrequently used assets provides no significant benefits so you should consider moving them out of the CDN.
• Setting an appropriate cache expiry: For time-sensitive content, setting a cache expiry time is important. The cache expiry time should neither be too long nor too short. If it is too long, the content might no longer be fresh. If it is too short, it can cause repeat reloading of content from origin servers to the CDN.
• CDN fallback: You should consider how your website/application copes with CDN failure. If there is a temporary CDN outage, clients should be able to detect the problem and request resources from the origin.
• Invalidating files: You can remove a file from the CDN before it expires by performing one of the following operations:
• Invalidate the CDN object using APIs provided by CDN vendors.
• Use object versioning to serve a different version of the object. To version an object, you can add a parameter to the URL, such as a version number. For example, version number 2 is added to the query string: image.png?v=2. Design after the CDN and cache are added.

A stateful server and stateless server has some key differences. A stateful server remembers client data (state) from one request to the next. A stateless server keeps no state information.

User A’s session data and profile image are stored in Server 1. To authenticate User A, HTTP requests must be routed to Server 1. If a request is sent to other servers like Server 2, authentication would fail because Server 2 does not contain User A’s session data. Similarly, all HTTP requests from User B must be routed to Server 2; all requests from User C must be sent to Server 3.

The issue is that every request from the same client must be routed to the same server. This can be done with sticky sessions in most load balancers 10; however, this adds the overhead. Adding or removing servers is much more difficult with this approach. It is also challenging to handle server failures.

In this stateless architecture, HTTP requests from users can be sent to any web servers, which fetch state data from a shared data store. State data is stored in a shared data store and kept out of web servers. A stateless system is simpler, more robust, and scalable.

Here, we move the session data out of the web tier and store them in the persistent data store. The shared data store could be a relational database, Memcached/Redis, NoSQL, etc. The NoSQL data store is chosen as it is easy to scale. Autoscaling means adding or removing web servers automatically based on the traffic load. After the state data is removed out of web servers, auto-scaling of the web tier is easily achieved by adding or removing servers based on traffic load.

Your website grows rapidly and attracts a significant number of users internationally. To improve availability and provide a better user experience across wider geographical areas, supporting multiple data centers is crucial.

This is an example setup with two data centers. In normal operation, users are geoDNS-routed, also known as geo-routed, to the closest data center, with a split traffic of x% in US-East and (100 – x)% in US-West. geoDNS is a DNS service that allows domain names to be resolved to IP addresses based on the location of a user.

In the event of any significant data center outage, we direct all traffic to a healthy data center. In Figure 16, data center 2 (US-West) is offline, and 100% of the traffic is routed to data center 1 (US-East).

Several technical challenges must be resolved to achieve multi-data center setup:

1. Traffic redirection: Effective tools are needed to direct traffic to the correct data center. GeoDNS can be used to direct traffic to the nearest data center depending on where a user is located.
2. Data synchronization: Users from different regions could use different local databases or caches. In failover cases, traffic might be routed to a data center where data is unavailable. A common strategy is to replicate data across multiple data centers. A previous study shows how Netflix implements asynchronous multi-data center replication 11.
3. Test and deployment: With multi-data center setup, it is important to test your website/application at different locations. Automated deployment tools are vital to keep services consistent through all the data centers 11.

A message queue is a durable component, stored in memory, that supports asynchronous communication. It serves as a buffer and distributes asynchronous requests. The basic architecture of a message queue is simple. Input services, called producers/publishers, create messages, and publish them to a message queue. Other services or servers, called consumers/subscribers, connect to the queue, and perform actions defined by the messages.

Decoupling makes the message queue a preferred architecture for building a scalable and reliable application. With the message queue, the producer can post a message to the queue when the consumer is unavailable to process it. The consumer can read messages from the queue even when the producer is unavailable.

When working with a small website that runs on a few servers, logging, metrics, and automation support are good practices but not a necessity. However, now that your site has grown to serve a large business, investing in those tools is essential.

1. Logging: Monitoring error logs is important because it helps to identify errors and problems in the system. You can monitor error logs at per server level or use tools to aggregate them to a centralized service for easy search and viewing.
2. Metrics: Collecting different types of metrics help us to gain business insights and understand the health status of the system. Some of the following metrics are useful:

• Host level metrics: CPU, Memory, disk I/O, etc.
• Aggregated level metrics: for example, the performance of the entire database tier, cache tier, etc
• Key business metrics: daily active users, retention, revenue, etc.

3. Automation: When a system gets big and complex, we need to build or leverage automation tools to improve productivity. Continuous integration is a good practice, in which each code check-in is verified through automation, allowing teams to detect problems early. Besides, automating your build, test, deploy process, etc. could improve developer productivity significantly.

Due to the space constraint, only one data center is shown in the figure.

1. The design includes a message queue, which helps to make the system more loosely coupled and failure resilient.
2. Logging, monitoring, metrics, and automation tools are included.

Horizontal scaling, also known as sharding, is the practice of adding more servers. Figure 20 compares vertical scaling with horizontal scaling.

Sharding separates large databases into smaller, more easily managed parts called shards. Each shard shares the same schema, though the actual data on each shard is unique to the shard.

User data is allocated to a database server based on user IDs. Anytime you access data, a hash function is used to find the corresponding shard. In our example, user_id % 4 is used as the hash function. If the result equals to 0, shard 0 is used to store and fetch data. If the result equals to 1, shard 1 is used. The same logic applies to other shards.

The most important factor to consider when implementing a sharding strategy is the choice of the sharding key. Sharding key (known as a partition key) consists of one or more columns that determine how data is distributed. As shown in Figure 22, “user_id” is the sharding key. A sharding key allows you to retrieve and modify data efficiently by routing database queries to the correct database. When choosing a sharding key, one of the most important criteria is to choose a key that can evenly distributed data.

Sharding is a great technique to scale the database but it is far from a perfect solution. It introduces complexities and new challenges to the system:

1. Resharding data: Resharding data is needed when 1) a single shard could no longer hold more data due to rapid growth. 2) Certain shards might experience shard exhaustion faster than others due to uneven data distribution. When shard exhaustion happens, it requires updating the sharding function and moving data around. Consistent hashing is a commonly used technique to solve this problem.
2. Celebrity problem: This is also called a hotspot key problem. Excessive access to a specific shard could cause server overload. Imagine data for Katy Perry, Justin Bieber, and Lady Gaga all end up on the same shard. For social applications, that shard will be overwhelmed with read operations. To solve this problem, we may need to allocate a shard for each celebrity. Each shard might even require further partition.
3. Join and de-normalization: Once a database has been sharded across multiple servers, it is hard to perform join operations across database shards. A common workaround is to de-normalize the database so that queries can be performed in a single table.

Here we shard databases to support rapidly increasing data traffic. At the same time, some of the non-relational functionalities are moved to a NoSQL data store to reduce the database load. Here is an article that covers many use cases of nosql