https://www.facebook.com/notes/facebook-engineering/scaling-the-messages-application-back-end/10150148835363920/
To overcome all these challenges, we started laying down a new architecture. At the heart of the application back end are the application servers. Application servers are responsible for answering all queries and take all the writes into the system. They also interact with a number of services to achieve this.
Each application server comprises:
- API: The entry point for all get and set operations, which every client calls. An application server is the sole entry point for any given user into the system. Any data written to or read from the system needs to go through this API.
- Distributed logic: To understand the distributed logic we need to understand what a cell is. The entire system is divided into cells, and each cell contains only a subset of users. A cell looks like this:
Understanding Cells
Cells give us many advantages:
- They help us scale incrementally while limiting failure scenarios
- Easy upgrades
- Metadata store failures affect only a few users
- Easy rollout
- Flexibility to host cells in different data centers with multi-homing for disaster recovery
Each cell consists of a single cluster of application servers, and each application server cluster is controlled by a set of ZooKeeper machines.
ZooKeeper is open source software that we use mainly for two purposes: as the controller for implementing sharding and failover of application servers, and as a store for our discovery service. Since ZooKeeper provides us with a highly available repository and notification mechanism, it goes a long way towards helping us build a highly available service.
Each application server registers itself in ZooKeeper by generating N tokens. The server uses these tokens to take N virtual positions on a consistent hash ring. This is used to shard users across these nodes. In case of failures, the neighboring nodes take over the load for those users, hence distributing the load evenly. This also allows for easy addition and removal of nodes into and from the application server cluster.
https://www.facebook.com/notes/facebook-engineering/chat-stability-and-scalability/51412338919/
Eugene wrote a great post detailing the inner workings of Chat, and recently we gave a talk (video: 1 2 3 4) about the product from front to back.
https://www.facebook.com/note.php?note_id=14218138919&id=9445547199&index=9
Distribution, Isolation, and Failover:
Fault tolerance is a desirable characteristic of any big system: if an error happens, the system should try its best to recover without human intervention before giving up and informing the user. The results of inevitable programming bugs, hardware failures, et al., should be hidden from the user as much as possible and isolated from the rest of the system.
The way this is typically accomplished in a web application is by separating the model and the view: data is persisted in a database (perhaps with a separate in-memory cache), with each short-lived request retrieving only the parts relevant to that request. Because the data is persisted, a failed read request can be re-attempted. Cache misses and database failure can be detected by the non-database layers and either reported to the user or worked around using replication.
While this architecture works pretty well in general, it isn't as successful in a chat application due to the high volume of long-lived requests, the non-relational nature of the data involved, and the statefulness of each request.
For Facebook Chat, we rolled our own subsystem for logging chat messages (in C++) as well as an epoll-driven web server (in Erlang) that holds online users' conversations in-memory and serves the long-polled HTTP requests. Both subsystems are clustered and partitioned for reliability and efficient failover. Why Erlang? In short, because the problem domain fits Erlang like a glove. Erlang is a functional concurrency-oriented language with extremely low-weight user-space "processes", share-nothing message-passing semantics, built-in distribution, and a "crash and recover" philosophy proven by two decades of deployment on large soft-realtime production systems.
- Application business logic: This is where the magic happens. The business logic is responsible for making sense of all user data, storing and retrieving it, and applying all the complex product operations to it to perform various functions. It also has a dedicated cache that acts as a write-through cache, since the application servers are the only entry points to read/write data for any given user. This cache stores the entire recent image for the user and gives us a very high cache hit rate. The business logic also interacts with the Web servers to respect user privacy and also apply any policies.
- Data access layer: The data access layer is the schema used to store the user’s metadata. It consists mainly of a time sequenced log, which is the absolute source of truth for the user’s data, and is used to back up, retrieve, and regenerate user data. The schema also consists of snapshots that represent the serialized user objects understood by the business logic. This layer is designed to present a generic interface to the application servers while making the underlying store pluggable.
- Metadata store: Each cell also has a dedicated metadata store. We use HBase as our metadata store. The data access layer interacts with HBase to provide storage functionality. Late last year we talked about our Messages storage infrastructure, which is built on top of Apache HBase.
Finally, the whole system has a number of cells, and looks like this:
Other Messages Services
The Messages application back end needs to parse email messages and attachments, and also provide discovery of the right application servers for the given user. This is achieved with the following services:
- MTA proxy: This service receives all incoming email messages and is responsible for parsing the email RFCs, attachments, large bodies of email, and so forth. These parsed out values are stored in a dedicated Haystack cluster (which is the same key/value store that we use for photos). Once the proxy has created a lightweight email object, it talks to the appropriate application server and delivers the message. But talking to the appropriate application server involves figuring out the cell and machine a particular user resides on, which brings us to the discovery service.
- Discovery service: This consists of a map of user-to-cell mappings. Every client needs to talk to the discovery service before it can contact an application server for any request. Given the stringent requirements, this service needs to be very highly available, scalable, and performant.
- Distributed logic client: These clients listen for ZooKeeper notifications and watch for any changes in the application server cluster state. Each application server cluster or cell has a dedicated client. These clients live in the discovery service process, and once the discovery service has mapped the user’s cell, it queries that cell’s client, which executes the consistent hash algorithm to figure out the correct application server node for the user.
The Messages application back end also relies on the following services:
- Memcache dirty service: The application servers query message counts from the home page very frequently to accurately display the message notification jewels. These counts are cached in memcache in order to display the home page as quickly as possible. As new messages arrive, these entries need to be dirtied from the application servers. Thus, this dedicated service runs to dirty these caches in every data center.
- User index service: This provides the social information for each user, like friends, friends of friends, and so forth. This information is used to implement the social features of messaging. For example, on every message that is added to the system, the application server node queries this service to determine if this message is from a friend or a friend of friend and directs it to the appropriate folder.
https://www.facebook.com/note.php?note_id=454991608919
By monitoring usage, two general data patterns emerged:
- A short set of temporal data that tends to be volatile
- An ever-growing set of data that rarely gets accessed
We spent a few weeks setting up a test framework to evaluate clusters of MySQL, Apache Cassandra, Apache HBase, and a couple of other systems. We ultimately chose HBase. MySQL proved to not handle the long tail of data well; as indexes and data sets grew large, performance suffered. We found Cassandra's eventual consistency model to be a difficult pattern to reconcile for our new Messages infrastructure.
HBase comes with very good scalability and performance for this workload and a simpler consistency model than Cassandra. While we’ve done a lot of work on HBase itself over the past year, when we started we also found it to be the most feature rich in terms of our requirements (auto load balancing and failover, compression support, multiple shards per server, etc.). HDFS, the underlying filesystem used by HBase, provides several nice features such as replication, end-to-end checksums, and automatic rebalancing. Additionally, our technical teams already had a lot of development and operational expertise in HDFS from data processing with Hadoop. Since we started working on HBase, we've been focused on committing our changes back to HBase itself and working closely with the community. The open source release of HBase is what we’re running today.
TODO
https://www.facebook.com/notes/facebook-engineering/inside-facebook-messages-application-server/10150162742108920/
