Confirm
358
edits
Changes
no edit summary
== Summary ==
This is a working proposal for the backend storage architecture of PiCL server. It's based on tries to take some of the good bits from the Firefox Sync backend, add in some lessons learned from running that in the field, simplify things a massively-sharded and cross-DC-replicated MySQL installationlittle, and make some adjustments towards stronger durability. It is far from final. All feedback welcome!
Goals:
* Scale to billions of users. Quickly. Easily.
* Don't lose user data. Even if a machine dies. Even if a meteor hits a data-center.
* Provide a relatively simple and well-understood Ops environment.
* Try to be low-cost, while maintaining acceptable levels of durability and availability.
* Provide for on-going infrasturcture experiments, refinements and upgrades
* The client-facing API is strongly consistent, and exposes an atomic check-and-set operation.
** This makes an eventually-consistent NoSQL store rather less attractive, unless coupled with a strongly-consistent control layer e.g. zookeeper. * Initial deployment will be into AWS.** Assuming PiCL succeeds in replacing sync, we can probably subsume some of the sync hardware over time. * It's OK to have brief periods of unavailability** This is, after all, a background service. There's no user in the loop most of the time.** The user-agent will be expected to deal gracefully with server unavailability. * Ops would like the ability to move users onto different levels of infrastructure, depending on their usage profile** For example, moving highly active users out of AWS and onto bare metal hardware.** Or, moving inactive users off onto lower-cost storage.** Or, just experimenting with a new setup for a select subset of users.
Basic Principles:
* Each user account gets an opaque, immutable user id.** This will only change if they completely delete and then re-create their account. * Each user account is explicitly assigned to a particular '''shardcluster'''.** Each cluster is a stand-alone piece of infrastructure with no links to other clusters.** Each cluster is responsible for its own durability, replication, scalability and so-on. * Each cluster is identified by an integera URL, at which it speaks a common protocol.** Different clusters may have different underlying technologies, e.g. one may be MySQL, one may be Cassandra.** But they all look the same from the outside. * A user's cluster assignment might change over time; this migration will require careful management.** Their shard This would be fairly infrequent, however. * The user-account and cluster-mapping information lives in a stand-alone piece of infra, the "userdb". Architecturally, the system winds up looking something like this: login handshake +--------+ +----------------------->| UserDB |<-------------------+ |+-----------------------| System | management api | || cluster URL +--------+ | || | || | |v | +--------+ storage protocol +----------------------+ | | client |<-------------------->| MySQL-Backed Cluster |<-----+ +--------+ +----------------------+ | | +----------------------+ | | MySQL-Backed Cluster |<-----+ +----------------------+ | | +-------------------------+ | | Casandra-Backed Cluster |<--+ +-------------------------+ == What the Client Sees == To begin a syncing session, the user-agent first "logs in" to the storage system, performing a handshake to exchange its BrowserID assertion for some short-lived Hawk access credentials. As part of this handshake, it will be told the base_url to which it should direct its storage operations. For simple third-party deployments, the base_url will point back to the originating server. For at-scale Mozilla deployments, it will point into the user's assigned cluster. In this example, the user has id "12345" and is assigned to the "mysql3" cluster: > POST https://storage.picl.services.mozilla.com HTTP/1.1 > { > "assertion": <browserid assertion>, > "device": <device UUID> > } . . < HTTP/1.1 200 OK < Content-Type: application/json < { < "base_url": "https://mysql3.storage.picl.services.mozilla.com/storage/12345", < "id": <hawk auth id>, < "key": <hawk auth secret key> < } < } The client then syncs away by talking to this base_url via the as-yet-undefined sync protocol: > GET https://mysql3.storage.picl.services.mozilla.com/storage/12345 HTTP/1.1 > Authorization: <hawk auth parameters> . . < HTTP/1.1 200 OK < Content-Type: application/json < { < "collections": { < "XXXXX": 42, < "YYYYY": 128 < } < } When the Hawk credentials expire, or when the user's cluster assignment is changed, it will receive a "401 Unauthorized" response from the storage server. To continue syncing, it will never change unless they delete have to perform a new handshake and get a new base_url. In this example, the user has been re-create assigned to the "cassandra1" cluster: > GET https://mysql3.storage.picl.services.mozilla.com/storage/12345 HTTP/1.1 > Authorization: <hawk auth parameters> . . < HTTP/1.1 401 Unauthorized < Content-Length: 0 . . > POST https://storage.picl.services.mozilla.com HTTP/1.1 > { > "assertion": <fresh browserid assertion>, > "device": <device UUID> > } . . < HTTP/1.1 200 OK < Content-Type: application/json < { < "base_url": "https://cassandra1.storage.picl.services.mozilla.com/storage/12345", < "id": <hawk auth id>, < "key": <hawk auth secret key> < } < } == The UserDB System == The UserDB system contains the mapping of user account emails to userids, and mapping of userids to clusters. This component has a lot of similarity to the TokenServer from the Sync2.0 architecture: https://wiki.mozilla.org/Services/Sagrada/TokenServer https://docs.services.mozilla.com/token/index.html However, we intend for it to manage a relatively small number of clusters, which each have their own internal sharding or other scaling techniques, rather than managing a large number of service node shards. We're also going to simplify some of the secrets/signing management, and not supporting multiple services from a single user account. It's not terribly write-heavy, but is very valuable data that must be kept strongly consistent - if we lose the ability to direct a user to the correct cluster, or send different devices to different clusters, the user is not going to be happy. It also needs to be highly available for reads, since if UserDB read capability goes down, then we lose the ability to access all clusters. To keep things simple and reliable and available, this will use a Multi-DC Replicated MySQL setup. It would be awesome if the write load is small enough to do '''synchronous''' replication here, using something like Galera cluster: http://codership.com/content/using-galera-cluster If not, then a standard master/slave setup should be OK. As long as we're careful about send users to stale cluster assignments. Example schema: CREATE TABLE users userid INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY email VARCHAR(128) NOT NULL clusterid INTEGER NOT NULL previous_clusterid INTEGER Each user is assigned to a particular cluster. We can also track the cluster they were previously assigned to, which might help with managing migration of users between clusters. CREATE TABLE clusters clusterid INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY base_url VARCHAR(128) NOT NULL assignment_weight INTEGER NOT NULL Each cluster as a base_url and an assignment_weight. When a new user accountgets created, we randomly assignment the to a cluster with probability proportional to the assignment_weight. Set it to zero to stop sending new users to a particular cluster. This service will need to have a user-facing API to support the login handshake dance, and some private management APIs for managing clusters, assignments, etc. Maybe even a nice friendly admin UI for the ops folks to use. == A Massively-Sharded MySQL Cluster == One of the leading options for storage is a massively-sharded MySQL setup, taking advantage of the highly shardable nature of the data set. Basic principles: * Each user is transaprently mapped to a shard via e.g.consistent hashing
* All reads and writes for a shard go to a single '''master''' MySQL database.
* Each master synchronously replicates to one or more '''hot standby''' dbs in the same DC, to guard against individual machine failure.
* The entire DC One of the standby dbs is asynchronously replicated to a '''warm standby''' setup in another regionperiodically snapshotted into S3, to guard against agaist data loss if the whole-DC failuregoes down.
* All sharding logic and management lives in a standThere is no cross-alone "db router" processDC replication; if the DC goes down, so that it's transparent the cluster becomes unavailable and we might have to the applicationrestore from S3.
* All sharding logic and management lives in a stand-alone "db router" process, so that it's transparent to the webapp code.
=== What the WebApp Sees === From the POV of the application webapp code, it's just talking to a regular old MySQL database:
+---------+ +--------------+
=== Transparent DB Router ===
The application code is actually talking to a "db router" server that speaks the MySQL wire protocol. In turn, the router is talking to the individual MySQL servers that are hosting each shard:
=== Intra-DC Redundancy ===
We need to guard against the loss of any individual server within a DC. There are separate redundancy schemes for the MySQL servers, and for the other supporting services.
==== MySQL Redundancy ====
To guard against the loss of any individual database server, each shard will also have a hot standby database, living in the same DC and configured for synchronous (semi-synchronous?) replication. For AWS it would be in a separate Availability Zone. The router monitors the health of the standby database, but does not forward it any queries. Its only job is to serve as a backup for the active master:
'''TODO:''' We could use the standby as a read slave, but I don't see the point. In a failure scenario the master needs to be able to handle the entire read load on its own, so it might as well do that all the time.
==== Other Service Redundancy ====
We don't want any single-point-of-failures, so we'll have to have multiple instances of the webapp talking to multiple instances of the router. These are connected via loadbalancing, virtual IPs, and whatever Ops wizardry is required to make single-machine failures in each tier be a non-event:
With multiple DB Router processes, we run into the problem of shared state. They must all agree on the current mapping of userids to shards, of shards to database machines, and which database machines are master versus standby. They'll One solution is to have them operate as a ZooKeeper (or similar) cluster to store this state in a consistent and highly-available fashion:
+----------------------------------------------+
=== Database Snapshots === For a final level of redundancy, we periodically snapshot each database into long-term storage, e.g. S3. Likely take the snapshot on the least up-to-date replica to minimize the chances that it would impact production capacity. As well as providing redundancy, these snapshots allow us to quickly bring up another DB for a particular shard. E.g. if we lose the hot standby, we can start a fresh one, restore it from a snapshot, then set it to work catching up from that point via standard replication. We'd use a similar process if we need to move or split shards - bring up a new replica from snapshot, get it up to date, then start sending traffic to it. === Inter-DC Redundancy ===
'''TODO:''' If we want to spend the money, we could keep replicas on standby in another DC. I doubt we'll want to spend the money.
=== Implications for the Client ===
''However'TODO:', since we're doing asynchronous replication, there' How many DCs? The principle should s a chance that recent database writes could be lost in the same regardless event of how many we havefailure. Nested Star Topology FTWThe client will see a consistent, but out-of-date view of its data. It must be able to recover from such a situation, although we hope this would be a very rare occurrence!
=== Implementing the Router ===
From the POV of the webapp code, it's just talking to ZooKeeper and Cassandra Storage Node as abstract systems:
'''TODO:''' In Try to use Route53 to provide consistent names for some of the presence of multiple clients and asynchronous replication and failovernodes, are we exposing any stronger guarantees to act as introducers even if the client than we'd get from an eventually-consistent store? E.g. client A writes, entire membership of the write is lost due to failover, client B writes, client A is now in an inconsistent state. Is this any different to client A and client B doing a conflicting writes in a NoSQL store, and us arbitrarily picking a winner?cluster has changed
The Cassandra cluster replicates out to another DC for durability, but everything else stays in the one DC. If that DC goes down, the cluster becomes unavailable but no data is lost. We can re-create it in the other DC, or wait for it to come back up.
== A Hibernation Cluster ==
* Needs There's a detailed and careful plan for how we'll bring up new DBs for existing shards, how we'll move dshards between DBsbit of management overhead in the API, with the handshake etc. We could consider factoring that out and how we'll split shards if that becomes necessaryjust doing the routing internally. All very doable, just fiddlyBut there's something to be said for explicitness.
