Failover mechanism in a distributed computing system

• US 9,983,957 B2
• Filed: 05/12/2016
• Issued: 05/29/2018
• Est. Priority Date: 09/12/2014
• Status: Active Grant

• Current Assignee: Meta Platforms, Inc. (f/k/a Facebook, Inc.)
• Original Assignee: Meta Platforms, Inc. (f/k/a Facebook, Inc.)
• Inventors: Tang, Liyin, Shroff, Rishit Manit, Aiyer, Amitanand S., Roodselaar, Arjen
• Primary Examiner(s): Ko, Chae M
• Application Number:
  US15/152,769
• Publication Number:
  US 20160253249A1
• Time in Patent Office: 747 Days
• Field of Search: 714 411, 714 63
• CPC Class Codes: G06F 11/1471 involving logging of persistent data for recovery
• CPC Class Codes: G06F 11/2028 eliminating a faulty processor or activating a spare
• CPC Class Codes: G06F 11/2033 switching over of hardware resources
• CPC Class Codes: G06F 11/2046 where the redundant components share persistent storage G06F11/2043 takes precedence
• CPC Class Codes: G06F 11/2097 maintaining the standby controller/processing unit updated initialisation or re-synchronisation thereof G06F11/1658 and subgroups
• CPC Class Codes: G06F 2201/805 Real-time
• CPC Class Codes: G06F 2201/84 Using snapshots, i.e. a logical point-in-time copy of the data
• CPC Class Codes: H04L 67/1095 Replication or mirroring of data, e.g. scheduling or transport for data synchronisation between network nodes
• CPC Class Codes: H04L 67/1097 for distributed storage of data in networks, e.g. transport arrangements for network file system [NFS], storage area networks [SAN] or network attached storage [NAS]

Abstract

The disclosure is directed to failover mechanisms in a distributed computing system. A region of data is managed by multiple region servers. One of the region servers is elected as a “leader” and the remaining are “followers.” The leader serves the read/write requests from a client. The leader writes the data received from the client into the in-memory store and a local write-ahead log (“WAL”), and synchronously replicates the WAL to the followers. A region server designated as an “active” region server synchronizes a distributed data store with the data from the WAL. Active witness followers apply the data from the WAL to their in-memory store while shadow witness followers do not. Different types of servers provide failover mechanisms with different characteristics. A leader is elected based on their associated ranks—higher the rank, higher the likelihood of electing itself as a leader.

Claims

1. A method performed by a computing system, comprising:

receiving a data access request including a dataset from a client computing system at a distributed computing system, the distributed computing system including multiple computing servers hosting a region of data associated with the distributed computing system, the computing servers distributed across multiple clusters, wherein one of the computing servers is a leader server and a remaining of the computing servers are follower servers, and wherein some of the computing servers are active servers, active witness servers and shadow witness servers;

writing, by the leader server, the dataset to a log file at a data storage system associated with the leader server;

replicating, by the leader server, the log file to the follower servers; and

updating, by an active server of the active servers, a distributed data storage system associated with a first cluster of the clusters to which the leader server belongs with the dataset from the log file.

2. The method of claim 1, wherein the leader server is one of the computing servers assigned to serve read-write requests received for the region.

3. The method of claim 1, wherein the active server is one of the computing servers in the first cluster assigned to update the distributed data storage system.

4. The method of claim 3, wherein the active server and the leader server are the same computing server of the computing servers.

5. The method of claim 1, wherein replicating the log file to the follower servers includes:

transmitting, by the leader server, the log file to the follower servers synchronously,

receiving, by the leader server, an acknowledgement from a majority of the follower servers regarding a receipt of the log file at the corresponding follower server, and

sending, by the leader server, an indication to the client computing system regarding a receipt of the dataset.

6. The method of claim 1, wherein replicating the log file to the follower servers includes:

replicating, by the leader server, an in-memory store of the leader server and the log file to the active witness servers, the in-memory store including data from the log file, and

replicating, by the leader server, the log file to the shadow witness servers.

7. The method of claim 1 further comprising:

receiving a read request at the distributed computing system from the client computing system for a specified data set in the region;

obtaining, by the leader server and in response to the read request, the specified data set from the distributed data storage system; and

transmitting, by the leader server, the specified data set to the client computing system.

8. The method of claim 1, wherein one of the computing servers is elected as the leader server as a function of a rank associated with the corresponding computing server.

9. The method of claim 8, wherein electing the leader server includes:

setting a timer of a specified duration for each of the computing servers, the specified duration being a function of the rank of the corresponding computing server, wherein the timer of a higher ranked computing server expires before that of a lower ranked computing server,

determining that a timer of the one of the computing servers expired before that of a remaining of the computing servers, and

selecting the one of the computing servers as the leader server.

10. The method of claim 8, wherein a computing server of the computing servers that is geographically closest to the client computing system is assigned a highest rank among the computing servers.

11. The method of claim 1 further comprising:

determining that the leader server in the first cluster failed;

determining if the clusters include one or more of the active witness servers; and

responsive to a determination that the clusters include one or more active witness servers, electing one of the one or more active witness servers as the leader server based on a rank associated with the one or more active witness servers.

12. The method of claim 11 further comprising:

responsive to a determination that the clusters do not include the active witness servers, electing one of the shadow witness servers as the leader server.

13. The method of claim 12 further comprising:

obtaining, by the one of the shadow witness servers, information from a replicated copy of the log file stored at a data storage system associated with the one of the shadow witness servers; and

updating, by the one of the shadow witness servers, an in-memory store of the one of the shadow witness servers with the information prior to serving a read-write request from the client computing system.

14. The method of claim 1, wherein the region is a portion of the data associated with distributed computing system.

15. A computer-readable storage medium storing computer-executable instructions, comprising:

instructions for selecting one of multiple computing servers in a distributed computing system as a leader server, wherein the leader server is a computing server assigned to serve a read-write request from a client computing system, wherein a remaining of the computing servers are follower servers, and wherein the leader server replicates a log file associated with the leader server to the follower servers, the log file including a data set received from the client computing system;

instructions for selecting one of the computing servers as an active server, wherein the active server is a computing server assigned to update a distributed data storage system with a log file associated with the active server; and

instructions for selecting some of the computing servers as active witness servers and some of the computing servers as shadow witness servers, wherein the leader server is configured to replicate the log file to the active witness servers and the shadow witness servers, wherein one of the active witness servers is configured to be elected as the leader server in an event the leader server and/or the active server becomes unavailable.

16. The computer-readable storage medium of claim 15, wherein the instructions for selecting one of the computing servers as the leader server includes:

instructions for assigning a rank to each of the computing servers,

instructions for setting a timer of a specified duration for each of the computing servers, the specified duration being a function of the rank of the corresponding computing server, wherein the timer of a higher ranked computing server expires before that of a lower ranked computing server,

instructions for determining that a timer of the one of the computing servers expired before that of a remaining of the computing servers, and

instructions for selecting the one of the computing servers as the leader server.

17. The computer-readable storage medium of claim 16, wherein the instructions for electing the one of the active witness servers include instructions for electing an active witness server of the one or more active witness servers as the leader server based on a rank associated with the one or more active witness servers.

18. The computer-readable storage medium of claim 15 further comprising:

instructions for determining that the leader server failed;

instructions for electing a shadow witness server of the one or more shadow witness servers as the leader server based on a rank associated with the one or more active witness servers;

instructions for obtaining information from a replicated copy of the log file stored at a data storage system associated with the shadow witness server; and

instructions for updating an in-memory store of the shadow witness server with the information prior to serving a read-write request from the client computing system.

19. The computer-readable storage medium of claim 15, wherein the instructions for selecting some of the computing servers as active witness servers further include replicating the log file to the active witness servers and instructing the active witness servers to update an in-memory store of the active witness servers with the data set from the log file.

20. A system, comprising:

a processor;

a first module configured to select;

one of multiple computing servers of a distributed computing system as a leader server and a remaining of the computing servers as follower servers, the leader server assigned to serve a data access request from a client computing system,

one of the computing servers as an active server,

some of the follower servers as active witness servers, and

some of the follower servers as shadow witness servers;

a second module configured to cause the leader server to process a data access request received at the distributed computing from a client computing system, the data access request including a dataset;

a third module configured to cause the leader server to write the dataset to a log file stored at a data storage system associated with the leader server;

a fourth module configured to cause the leader server to replicate the log file to the active witness servers and the shadow witness servers; and

a fifth module configured to cause the active server to update a distributed data storage system associated with the active server with the dataset from the log file.