Abstract:
A node failure detector for use in a distributed database that is accessed through a plurality of interconnected transactional and archival nodes. Each node is selected as an informer node that tests communications with each other node. Each informer node generates a list of suspicious nodes that is resident in one node designated as a leader node. The leader node analyzes the data from all of the informer nodes to designate each node that should be designated for removal with appropriate failover procedures.

Description:
CROSS-REFERENCE TO RELATED APPLICATIONS 
       [0001]    This application is a continuation of U.S. application Ser. No. 14/215,372 filed on Mar. 17, 2014, and entitled “Distributed Database Management System With Node Failure Detection,” which claims priority from U.S. Application No. 61/789,370 filed on Mar. 15, 2013. Each of these applications is incorporated in its entirety by reference. 
         [0002]    U.S. Pat. No. 8,224,860 granted Jul. 17, 2012, for a Database Management System is incorporated in its entirety herein by reference. 
     
    
     BACKGROUND 
       [0003]    Field of the Invention 
         [0004]    This invention generally relates to database management systems and more specifically to detecting failures during the processing of a distributed database system. 
         [0005]    Description of Related Art 
         [0006]    The above-identified U.S. Pat. No. 8,224,860 discloses a distributed database management system comprising a network of transactional nodes and archival nodes. Archival nodes act as storage managers for all the data in the database. Each user connects to a transactional node to perform operations on the database by generating queries for being processed at that transactional node. A given transactional node need only contain that data and metadata as required to process queries from users connected to that node. This distributed database is defmed by an array of atom classes, such as an index class, and atoms where each atom corresponds to a different instance of the class, such as index atom for a specific index. Replications or copies of a single atom may reside in multiple nodes wherein the atom copy in a given node is processed in that node. 
         [0007]    In an implementation of such a distributed database asynchronous messages transfer among the different nodes to maintain the integrity of the database in a consistent and a concurrent state. Specifically each node in the database network has a unique communication path to every other node. When one node generates a message involving a specific atom, that message may be sent to every node that contains a replication of that specific atom. Each node generates these messages independently of other nodes. So at any given instant multiple nodes will contain copies of a given atom and different nodes may be at various stages of processing that atom. As the operations in different nodes are not synchronized it is important that the database be in a consistent and concurrent state at all times. 
         [0008]    A major characteristic of such distributed databases is that all nodes be in communication with each other at all times so the database is completely connected. If a communications break occurs, the database is not considered to be connected. One or more nodes must be identified and may be removed from the network in an orderly manner. Such identification and removal must consider that any node can fail at any given time and that a communications break can occur only between two nodes or that multiple breaks can occur among several nodes. The identification of a node or nodes for removal must be accomplished in a reliable manner. Moreover such an identification should enable failure processes to resolve a failure with minimal interruption to users. 
       SUMMARY 
       [0009]    Therefore it is an object of this invention to provide a method for detecting a node failure in a distributed database management system. 
         [0010]    Another object of this invention is to provide a method for detecting a node failure and for designating a node for failure. 
         [0011]    Still another object of this invention is to provide a method for detecting a node failure and for designating a node for failure on a reliable basis. 
         [0012]    Yet another object of this invention is to provide a method for detecting a node failure and for designating a node for failure with minimal interruption to users. 
     
    
     
       BRIEF DESCRIPTIONS OF THE DRAWINGS 
         [0013]    The appended claims particularly point out and distinctly claim the subject matter of this invention. The various objects, advantages and novel features of this invention will be more fully apparent from a reading of the following detailed description in conjunction with the accompanying drawings in which like reference numerals refer to like parts, and in which: 
           [0014]      FIG. 1  is a diagram in schematic form of one embodiment of an elastic, scalable, on-demand, distributed database management system that incorporates this invention and that includes interconnected transactional and archival nodes; 
           [0015]      FIG. 2  depicts the organization of a transactional node; 
           [0016]      FIG. 3  depicts the organization of an archival node; 
           [0017]      FIG. 4  depicts the syntax of an exemplary asynchronous message that is transferred among the transaction and archival nodes of the system of  FIG. 1 . 
           [0018]      FIG. 5  depicts certain asynchronous messages that are specifically involved with the implementation of this invention; 
           [0019]      FIG. 6  is a flow diagram of a node failure system that incorporates this invention; 
           [0020]      FIG. 7  is a flow diagram of a node failure detection system that identifies possible failures and that is useful in implementing the node failure system of  FIG. 6 ; 
           [0021]      FIG. 8  is a flow diagram of an operation that processes the information obtained in the node failure detection system of  FIG. 7 ; and 
           [0022]      FIG. 9  is a flow diagram of a process for choosing a node to fail based upon information obtained from the operation disclosed in  FIG. 8 . 
       
    
    
     DETAILED DESCRIPTION 
       [0023]      FIG. 1  depicts one embodiment of an elastic, scalable, on-demand, distributed database management system  30  with a plurality of data processing nodes that incorporates this invention. Nodes N 1  through N 6  are “transactional nodes” that provide user applications access to the database; nodes A 1  and A 2 , “archival nodes” that function to maintain a disk archive of the entire database at each archival node. While an archival node normally stores the entire database, a single transactional node contains only that portion of the database it determines to be necessary to support transactions being performed at that node at that time. 
         [0024]    Each node in  FIG. 1  can communicate directly with every other node in the system  30  through a database system network  31 . For example, node N 1  can establish a communications path with each of nodes N 2  through N 6 , A 1  and A 2 . Communications between any two nodes is by way of serialized messages. In a preferred embodiment, the messaging is performed in an asynchronous manner to maximize the bandwidth used by the system thereby to perform various operations in a timely and prompt manner. Typically the database system network  31  operates with a combination of high-bandwidth, low-latency paths (e.g., an Ethernet network) and high-bandwidth, high-latency paths (e.g., a WAN network). Each node has the capability to restrict use of a low-latency path to time-critical communications (e.g., fetching an atom). The high-latency path can be used for non-critical communications (e.g. a request to update information for a table). Also and preferably, the data processing network of this invention incorporates a messaging protocol, such as the Transmission Control Protocol (TCP) that assures that each node processes messages in the same sequence in which they were sent to it by other nodes. 
         [0025]      FIG. 2  depicts a representative transactional node  32  that links to the database system network  31  and various end users  33 . The transactional node  32  includes a central processing system (CP)  34  that communicates with the database system network  31  through a network interface  35  and with the various users through and user network interface  37 . The central processing system  34  also interacts with RAM memory  38  that contains a copy of the database management program that implements a preferred embodiment of this invention. This program functions to provide a remote interface  40 , a database request engine  41  and a set  42  of classes or objects. 
         [0026]    The database request engine  41  only exists on transactional nodes and is the interface between the high-level input and output commands at the user level and system level input and output commands at the system level. In general terms, its database request engine parses, compiles and optimizes user queries such as SQL queries into commands that are interpreted by the various classes or objects in the set  42 . The classes/objects set  42  is divided into a subset  43  of “atom classes,” a subset  44  of “message classes” and a subset  45  of “helper classes.” 
         [0027]    Referring to  FIG. 3 , each archival node  50 , such as archival node A 1  or A 2  in  FIG. 1 , also connects to the database system network  31 . However, in place of end users  33  associated with a transactional node  32  in  FIG. 2 , an archival node connects only to persistent storage  51 , typically a disk-based storage system or a key value store. The archival node  50  includes a central processing system  54  that communicates with the persistent storage  51  through an I/O channel  52  and with the database system network  31  through a network interface  55 . The central processing system  54  also interacts with RAM memory  57  that contains a set  62  of classes or objects. Similarly to the transactional node  32  in  FIG. 2 , the classes/objects set  62  in  FIG. 3  includes a set  63  of “atom classes,” a set  64  of “message classes” and a set  65  of “helper classes.” 
         [0028]    The atom classes collectively defme atoms that contain fragments of the database including metadata and data. Atoms are replicated from one node to another so that users at different nodes are dealing with consistent and concurrent versions of the database, even during actions taken to modify the information in any atom. At any given time there is no guarantee that all replicated copies of a single atom will be in the same state because there is no guarantee that messages from one node to other nodes will be processed concurrently. 
         [0029]    As previously indicated, communications between any two nodes is by way of serialized messages which are transmitted asynchronously using the TCP or other protocol with controls to maintain message sequences.  FIG. 4  depicts the basic syntax of a typical message  90  that includes a variable length header  91  and a variable length body  92 . The header  91  includes a message identifier code  93  that specifies the message and its function. The header  91  also includes a software version identification  94 , a local identification  95  of the sender and information  96  for the destination of the message as an added identification. From this information the recipient node can de-serialize, decode and process the message. 
         [0030]      FIG. 5  depicts a subset of messages having the syntax of  FIG. 4  for implementing this invention. As discussed previously, when a message is to be sent, there are different communications paths to different nodes. For example, if one node requests an existing atom, replications of that atom may be obtained from a number of other nodes. In this embodiment, a known “pinging” operation one node sends a “Ping” message to a receiving node. The receiving node returns a “Ping Acknowledge” message to the sending node with time information. In one embodiment each node periodically uses a helper&#39;s class to send a Ping message to each of the other nodes to which it connects. Each receiving node uses a helper class to return a Ping Acknowledge message  111  that contains the ping time. 
         [0031]    The failure to receive a ping acknowledgement within a predetermined time indicates a break in communications with respect to messages being sent between the requesting or sending node and the recipient or receiving node. In the context of this invention a first node transmits the Ping message  110  to another node and functions as an “informer node” or I-Node if the Ping Acknowledge signal is not received. If there is a failure, the I-Node identifies the receiving node as being suspicious (e.g. an “S-Node”) by means of a Suspicious Node message  159 . A Leader Acknowledge message  160  triggers a request for each I-Node to respond with information about any suspicious nodes that connect to that specific I-Node. 
         [0032]    A purpose of this invention is to detect a node failure and enable corrective action.  FIG. 6  depicts an overall failure system  200  that includes a failure monitor  201  for detecting various types of failure. A node failure detection process  202  incorporates this invention to provide such an indication. If a failure is detected, step  203  diverts the failure system to a detected failure process  204 . 
         [0033]    Referring to  FIG. 7 , during an iterative node failure detection process  202 , each node, generally in sequence, acts as a sending node of I-node that uses step  211  to select a receiving node and step  212  to ping the selected receiving node. For example, when N 1  acts as an I-Node, steps  211  and  212  select an initial receiving node to receive a Ping message at step  212 . During successive iterations the I-node will ping nodes N 2  through N 6 , A 1  and A 2  using step  211  in successive iterations to select each selected receiving node in sequence. If a Ping Acknowledge message is received in a timely fashion at step  213 , step  214  selects a next node, such as node N 3 , for receiving a Ping message. 
         [0034]    If no Ping Acknowledge message is received within a defmed time interval, it is assumed that a communications break exists. Step  215  marks that receiving node as being “suspicious” with respect to the sending I-Node. Step  216  sends a Suspicious Node message  159  that includes the I-Node identification and Suspicious Node identification in  216  to a Leader Node. 
         [0035]    A Leader Node is responsible for analyzing the information received from all the I-Nodes. Only one L-Node can exist at a time and it must be a non-suspicious node. A node can only act as an L-Node if it has received a response for a given set of S-Nodes from a majority of the database as represented by other I-Nodes (i.e., majority of active, non-suspicious nodes). If the active I-Node receives a Leader Acknowledgement message in a timely manner, step  217  uses step  214  to select a next node to be tested by the I-node and control returns to step  212 . Otherwise, there is no certainty as to whether the I-node or the node being tested is causing the communications break. Step  220  selects the next non-suspicious node as the new Leader Node. If it is available, step  221  returns control to step  216  and the message is resent to the new L-Node. If no node qualifies as an L-Node, an error state is generated in step  222 . 
         [0036]    With respect to the process shown in  FIG. 8 , whenever a Leader Node receives a suspicious node message that identifies both an I-Node and a suspicious node at step  251 , step  252  establishes a first time interval. Step  253  identifies the number of I-Nodes that identify the reported Suspicious Node. If a majority of the active nodes identify the Suspicious Node, step  254  immediately transfers control to a Choose Node to Fail process  255  that chooses the node that is to fail. Step  256  sends a Leader Node Acknowledge message to all I-Nodes. Control then returns to step  251  to await receipt of a the next Suspicious Node message. 
         [0037]    If a majority does not exist at the instant of step  253 , step  254  transfers control to step  257  that times out the first time interval. If the majority is reached prior to the expiration of that time interval, step  257  diverts control to steps  255  and  256 . Otherwise step  257  transfers control to step  260  that sends a Leader Acknowledge message to all I-Nodes and then waits in step  261  for a second time interval to determine whether a majority of I-Nodes respond. At the end of that interval control passes through step  262  to step  255  if a majority of I-Nodes has been identified. If the second time interval expires without obtaining a majority, step  262  diverts to establish an error state at step  263 . 
         [0038]      FIG. 9  depicts the process  204  in  FIG. 6  for designating nodes that should fail. The process selects one I-Node at step  271  that is paired with a Suspicious Node in step  272 . Then the system processes step  273  to determine whether the selected Suspicious Node is only suspicious to the selected I-Node. If only the selected I-Node identifies a Suspicious Node, a standoff exists and step  274  diverts control to step  276  that designates the node with the highest node ID as being disabled. If the standoff does not exist, step  274  transfers control to step  275  that designates all the suspicious nodes associated with the selected I-Node to be disabled. 
         [0039]    After either step  275 , step  276  completes its process and control returns to step  271  to execute the process for another I-Node. When completed all the I-Nodes have been processed the node designations are made available to the failure system  200  in  FIG. 6  for being processed in step  204 . 
         [0040]    Therefore there has been disclosed an embodiment of this invention wherein each node operating with a distributed database monitors communication paths with all other nodes in the network. Any communications break is noted and the network is analyzed to determine nodes that need to be failed. This information is reported for processing whereby failures are handled in orderly fashion with minimal interruption to user activities and in a manner in which data remains consistent and concurrent. More specifically, this invention enhances user access because it detects node failure in an orderly and efficient manner to enable appropriate failure system to maintain the database in a concurrent and consistent manner. 
         [0041]    It will be apparent that many modifications can be made to the disclosed apparatus without departing from the invention. For example, this invention has been described with a “majority” defmed as a majority of a subset of all the active nodes. In other applications, the “majority” might be defmed as a majority of the archival nodes. Still other subsets of nodes could be defmed as the basis for determining a majority. Specific messages and processes have been given different titles for purposes of explanation. It will be apparent that equivalent messages and processes could be designed by different names while still performing the same or equivalent functions. Therefore, it is the intent of the appended claims to cover all such variations and modifications as come within the true spirit and scope of this invention.