System and method for augmenting consensus election in a distributed database

• US 10,713,275 B2
• Filed: 07/01/2016
• Issued: 07/14/2020
• Est. Priority Date: 07/02/2015
• Status: Active Grant

• Current Assignee: MongoDB, Inc.
• Original Assignee: MongoDB, Inc.
• Inventors: Merriman, Dwight, Horowitz, Eliot, Michalski, Andrew, Avitabile, Therese
• Primary Examiner(s): Peng, Huawen A
• Application Number:
  US15/200,975
• Publication Number:
  US 20170032007A1
• Time in Patent Office: 1,474 Days
• CPC Class Codes: G06F 16/21 Design, administration or maintenance of databases
• CPC Class Codes: G06F 16/25 Integrating or interfacing systems involving database management systems
• CPC Class Codes: G06F 16/27 Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor

Abstract

According to one aspect, a distributed database system is configured to manage write operations received from database clients and execute the write operations at primary nodes. The system then replicates received operations across a plurality of secondary nodes. Write operation can include safe write requests such that the database guaranties the operation against data loss once acknowledged. In some embodiments, the system incorporates an enhanced arbiter role the enables the arbiter to participate in cluster-wide commitment of data. In other embodiments, the enhanced arbiter role enables secondary nodes to evaluate arbiter operations logs when determining election criteria for new primary nodes.

Claims

1. A computer implemented method for managing a distributed database, the method comprising:

establishing at least one primary node within a plurality of nodes, wherein the plurality of nodes comprise a distributed database system and the distributed database system provides responses to database requests from database clients;

establishing at least one secondary node configured to host a replica of data of the at least one primary node from the plurality of nodes and update the replica responsive to received replicated operations from the at least one primary node;

establishing at least one arbiter node configured to host an operation log of operations executed by the at least one primary node, wherein the at least one arbiter node does not host a replica of the data; and

electing a new primary node responsive to detecting a failure of the at least one primary node, wherein electing the new primary node includes;

executing a consensus protocol to elect one of the at least one secondary node as the new primary node, wherein the consensus protocol includes participation of the at least one secondary node and the at least one arbiter node associated with the failed at least one primary node;

evaluating election criteria at the at least one secondary node during an election period;

communicating by a respective one of the at least one secondary node i) a self-vote responsive to determining that the respective one of the at least one secondary node meets the election criteria, or ii) a confirmation of a received vote of another secondary node responsive to determining that the another secondary node is more suitable to be elected as the new primary node based on the election criteria; and

evaluating by the respective one of the at least one secondary node the operation log of the at least one arbiter node as part of the acts of determining.

2. The method according to claim 1, further comprising restricting processing of write operations, received from the database clients, to the at least one primary node.

3. The method according to claim 1, further comprising an act of communicating the self-vote responsive to querying the operation log of the at least arbiter node and determining that the operation log of the at least arbiter node is more up to date than operation logs of one or more of the at least one secondary node based on received election information.

4. The method according to claim 1, wherein the election criteria include at least one of, or any combination of, or two or more of:

a most recent data requirement, a location requirement, a hardware requirement, and an uptime requirement.

5. The method according to claim 3, wherein responsive to the respective one of the at least one secondary node determining that the operation log of the at least arbiter node includes the most up to date data, triggering an update of a replica of the data hosted by the respective one of the at least one secondary node.

6. The method according to claim 1, further comprising an act of committing write operations responsive to safe write requests received from the database clients, wherein committing the write operations includes counting an act of logging the write operations at the at least one arbiter node towards any commitment requirements.

7. The method according to claim 1, wherein the at least one arbiter node is configured to limit participation in replication operations at the at least one arbiter node to updating the operation log without replication of the operations on respective data.

8. The method according to claim 1, wherein the at least one arbiter node is configured to provide operations stored in the operation log to one or more of the at least one secondary node.

9. A distributed database system, the system comprising:

at least one processor operatively connected to a memory, wherein the at least one processor when running is configured to execute a plurality of system components, wherein the plurality of system components comprise;

a configuration component configured to establish a role associated with each node in a plurality of nodes, wherein the configuration component is configured to establish at least one primary node with a primary role, at least one secondary node with a secondary role, and at least one arbiter node with an arbiter role;

the at least one primary node configured to;

host data;

execute write operations on at least a portion of the data and generate at least one log entry for execution of the write operations;

replicate the at least one log entry to the at least one secondary node and the at least one arbiter node;

the at least one secondary node configured to;

host a replica of the data of the at least one primary node;

execute the at least one log entry received from the at least one primary node to update a respective copy of the at least a portion of the data;

the at least one arbiter node configured to update an operation log of operations responsive to receipt of the at least one log entry for the write operations from the at least one primary node, wherein the at least one arbiter node does not host a replica of the data;

an election component configured to elect a new primary node responsive to detecting a failure of the at least one primary node, wherein the election component is further configured to;

execute a consensus protocol to elect one of the at least one secondary node as the new primary node, wherein the consensus protocol includes participation of the at least one secondary node and the at least one arbiter node associated with the failed at least one primary node;

evaluate election criteria at the at least one secondary node during an election period;

communicate i) a self-vote for a respective one of the at least one secondary node responsive to determining that the respective one of the at least one secondary node meets the election criteria, or ii) a confirmation of a received vote of another secondary node responsive to determining that the another secondary node is more suitable to be elected as the new primary node based on the election criteria; and

evaluate the operation log of the at least one arbiter node as part of the acts of determining.

10. The system of claim 9, further comprising a replication component configured to restrict the write operations received from client computer systems to the at least one primary node having the primary role.

11. The system according to claim 9, wherein the election component is distributed through one or more nodes comprising the distributed database system.

12. The system according to claim 11, wherein the election component is distributed at least through the at least one secondary node and the at least one arbiter node.

13. A computer implemented method for managing a distributed database, the method comprising:

establishing at least one primary node configured to host data;

establishing at least one secondary node configured to host a replica of the data hosted by the at least one primary node;

establishing at least one arbiter node, the at least one arbiter node configured to host an operation log of operations executed by the at least one primary node, wherein the at least one arbiter node does not host a replica of the data hosted by the at least one primary node or the at least one secondary node;

electing a new primary node responsive to detecting a failure of the at least one primary node, wherein electing the new primary node includes;

executing a consensus protocol to elect one of the at least one secondary node as the new primary node, wherein the consensus protocol includes participation of the at least one secondary node and the at least one arbiter node associated with the failed at least one primary node;

evaluating election criteria at the at least one secondary node during an election period;

communicating by a respective one of the at least one secondary node i) a self-vote responsive to determining that the respective one of the at least one secondary node meets the election criteria, or ii) a confirmation of a received vote of another secondary node responsive to determining that the another secondary node is more suitable to be elected as the new primary node based on the election criteria; and

evaluating, by the respective one of the at least one secondary node, the operation log of the at least one arbiter node as part of the acts of determining.

14. The method according to claim 13, further comprising an act of communicating the self-vote responsive to querying the operation log of the at least arbiter node and determining that the operation log of the at least arbiter node is more up to date than operation logs of one or more of the at least one secondary node based on received election information.

15. The method according to claim 13, wherein the election criteria include at least one of, or any combination of, or two or more of:

a most recent data requirement, a location requirement, a hardware requirement, and an uptime requirement.

16. The method according to claim 14, wherein responsive to the respective one of the at least one secondary node determining that the operation log of the at least arbiter node includes the most up to date data, triggering an update of a replica of the data hosted by the respective one of the at least one secondary node replica.

17. The method according to claim 13, further comprising an act of committing write operations responsive to safe write requests received from database clients, wherein committing the write operations includes counting an act of logging the write operations at the at least one arbiter node towards any commitment requirements.

18. A distributed database system, the system comprising:

at least one processor operatively connected to a memory, wherein the at least one processor when running is configured to execute a plurality of system components, wherein the plurality of system components comprise;

a configuration component configured to;

establish at least one primary node configured to host data;

establish at least one secondary node configured to host a replica of the data of the at least one primary node; and

establish at least one arbiter node, the at least one arbiter node configured to host an operation log of operations executed by the at least one primary node, wherein the at least one arbiter node does not host a replica of the data;

an election component configured to elect a new primary node responsive to detecting a failure of the at least one primary node, wherein the election component is further configured to;

execute a consensus protocol to elect one of the at least one secondary node as the new primary node, wherein the consensus protocol includes participation of the at least one secondary node and the at least one arbiter node;

evaluate election criteria at the at least one secondary node during an election period;

communicate i) a self-vote for a respective one of the at least one secondary node responsive to determining that the respective one of the at least one secondary node meets the election criteria, or ii) communicate a confirmation of a received vote of another secondary node responsive to determining that the another secondary node is more suitable to be elected as the new primary node based on the election criteria; and

evaluate the operation log of the at least one arbiter node as part of the acts of determining.

19. The system of claim 18, further comprising a replication component configured to restrict write operations received from client computer systems to the at least one primary node having a primary role.

20. The system according to claim 18, wherein the election component is distributed through one or more nodes comprising the distributed database system.

21. The system according to claim 20, wherein the election component is distributed at least through the at least one secondary node and the at least one arbiter node.