Scaling quorum based replication systems

A computer determines whether it has received user input or a node within a replica set has reached a capacity threshold. Based on receiving user input or determining that a node within a replica set has reached a capacity threshold, creating a snapshot of the data stored in the replica set and partitioning the data based on the created snapshot. The computer then initializing nodes within a new replica set and moves a partition from the original replica set to the new replica set before deleting the other partition from the old replica set.

BACKGROUND

The present invention relates generally to data storage, and more particular to quorum-based replication systems. Data storage is increasing in popularity as the world moves towards a digital age. Information once written in the pages of books and newspapers is now digitally available for searching, filtering, and analyzing from a variety of sources, such as the internet or a local database. Moreover, this information is more available than ever with the popularity of electronic devices, such as cell phones and computers, and the accessibility of such devices has resulted in vast amounts of user generated information, such as photos, comments, posts, and interactions with others. All of this data requires storage, and while data storage is relatively stable, it is possible for data servers storing such data to fail, losing potentially valuable information. In order to provide some fault tolerance, a system has been implemented in which identical data is stored in multiple locations forming what is known as a data replication service. In a data replication service, data is made redundant by replicating it across multiple storage devices, or nodes, within what is known as a replica set. If one of the nodes within a replica set fails, the data can be recovered from one of the other nodes within the replica set which has backed up the same data. The redundancy of the data within the cluster allows the data to remain accessible and the service to remain fully operational even in the face of some node failures.

SUMMARY

Embodiments of the present invention disclose a method, system, and computer program product for a database scaling system. A computer determines whether it has received user input or a node within a replica set has reached a capacity threshold. Based on receiving user input or determining that a node within a replica set has reached a capacity threshold, creating a snapshot of the data stored in the replica set and partitioning the data based on the created snapshot. The computer then initializing nodes within a new replica set and moves a partition from the original replica set to the new replica set before deleting the other partition from the old replica set.

DETAILED DESCRIPTION

As previously mentioned, data replication services store identical data redundantly across multiple storage devices (or nodes) within a replica set (or replication cluster) in order to back up and make available the data even if all but one node within the replica set fails. For example, data replication is often performed between a phone, computer, and “cloud storage” (remote storage) belonging to a user. In this example, each of the phone, cloud, and computer are considered nodes within a replication set and information stored on one node is backed up to other nodes within the replica set. For example, photos on the phone of a user may also be saved, or backed up, on a personal computer and cloud.

However, because the same data is replicated across all nodes within the replica set (the photo in the example above is stored on the phone, cloud, and computer), a replication service is limited in scalability by the node of the cluster with the least storage capacity. For example, if the phone of a user has 16 GB storage capacity and the cloud can store up to 2 GB of user data, then phone data exceeding 2 GB will not be replicated to the cloud because the cloud can only store up to 2 GB, reducing the effectiveness of the replication service. A partial solution to this problem is to add more nodes to the system, such as incorporating the computer of the user as a node in the replica set. Assuming the computer of the user has a storage capacity of 500 GB, if the user adds the computer to the replica set, now all 16 GB of data on the phone of the user can be backed up on the computer of the user. However, adding this additional node does not rectify the inability of another node in the replica set, i.e. the cloud storage, to replicate all 16 GB of phone data. Rather, the phone and computer of the user backup all 16 GB of data while the cloud storage only replicates up to 2 GB of data. Scaling replication systems such as that described above are a common problem as more and more electronic data is created and stored, except instead of a phone, cloud, and computer, the nodes may be servers designed to store large amounts of data. Often times, these servers are stored vast distances apart from each other for multiple reasons, making it difficult and expensive to physically add more storage capacity to an individual node (in the example, increasing cloud storage). This is because nodes in a replica set are stored apart from each other to avoid both nodes failing simultaneously, for instance in the case of a natural disaster or power outage. In addition, having the information stored in multiple locations allows a user to obtain the replicated data from the closest proximity node, saving time and increasing efficiency.

Another known solution to providing extensibility and scalability to such systems is to create a master cluster for keeping the information about replica sets, similar to a catalog service, where replica sets can be added when needed. However, this approach requires the master cluster be predefined upfront and include more components within the system, increasing system complexity and resource consumption while decreasing manageability. Not only do these additional components cost time and money, but they are also required for each replica set and, thus, the process of predefining a master is compounded by the amount of replica sets desired. Similarly, predefining the master cluster complicates the client code which must be first introduced into the master cluster and, thus, as the number of clients are increased, as is the cost of predefining the master cluster. Reclustering alleviates these problems by allowing a user to split existing clusters into multiple partitions rather than predefining the master cluster.

Another benefit of reclustering is that creating cluster IDs (or replica IDs) for each cluster and the manner in which the data is split allows for replication services to remain operational despite moving some of the data to a new cluster. Many replication services require a period of time where requests cannot be served due to the fact that state and control are moved from one cluster to another. Conversely, reclustering defines new cluster IDs prior to initializing the new cluster which allows for the identification and modification of data stored in either cluster throughout the reclustering process. This reclustering process is described in greater detail with reference to the following Figures.

FIG. 1illustrates a replication service scaling system100, in accordance with an embodiment of the invention. In the example embodiment, replication service scaling system100includes computing device110as well as nodes120A,120B,120C, and120D, all interconnected via network108.

In the example embodiment, network108may be the Internet, representing a worldwide collection of networks and gateways to support communications between devices connected to the Internet. Network108may include, for example, wired, wireless, or fiber optic connections. In other embodiments, network108may be implemented as an intranet, a local area network (LAN), or a wide area network (WAN). In general, network108can be any combination of connections and protocols that will support communications between computing device110and nodes120.

In the example embodiment, nodes120(node120A, node120B, node120C, and node120D) are a set of computing devices which store data, such as replicated data122, and are capable of being configured to perform data replication. Data replication is advantageous because the redundancy of the replicated data allows it to remain accessible on at least one node, such as node120A, in the event other nodes within the cluster, such as nodes120B-D, were to fail. In the example embodiment, nodes120may be grouped together into a single replica set where replicated data within the replica set, such as replicated data122A stored on node120A, is in theory identical to the replication data of another node within the same cluster, such as replication data122B stored on node120B. In the example embodiment, one node, such as node120A, may be designated as a replica set primary node while the remaining nodes, such as nodes120B-D, are designated as replica set secondary nodes. The replica set primary node (in this example, node120A) receives all write operations to replicated data122A and logs all instances in an operations log, or oplog. After writes are executed on the primary node (node120A), secondary nodes (nodes120B-D) reference the oplog to determine which writes to make to the secondary nodes, thereby making the data in the primary node (node120A) redundant across the secondary nodes (nodes120A-D). This process is known as asynchronous replication, while synchronous replication applies writes to all nodes120simultaneously. While the example embodiment utilizes asynchronous replication, in other embodiments, synchronous replication may be implemented. It should be noted, however, that nodes within the replica set do fail and it is possible for writes logged in the oplog to not be reflected in the secondary nodes120(it is also possible for the replica set primary node to fail, in which case a new replica set primary node is elected from the replica set secondary nodes). In order to provide fault-tolerance for instances in which a secondary node fails, replica sets may be configured for quorum-based replication (or consensus-based replication) wherein a write to the primary node is only carried through in secondary nodes if the majority of the secondary nodes (quorum) accept the command. Thus, quorum-based data replication systems like that formed by nodes120will tolerate failures in a minority of the nodes while still making progress (read/writes) in the majority of the nodes. While operational nodes continue to make progress, nodes which have failed are then recovered and incorporated back into the replica set.

In the example embodiment, replicated data122(replicated data122A, replicated data122B, replicated data122C, and replicated data122D) is information contained in computer storage, such as files, folders, images, documents, audio clips, video files, etc. In the example embodiment, replicated data122may be redundant such that replicated data122A may be identical to replicated data122B and so forth depending on how nodes120are clustered. Alternatively, replicated data within a single node, such as replicated data122A, may differ from one another, such as node120B, again, based on how nodes120are clustered (described in more detail below).

In the example embodiment, computing device110includes scaling program112. In the example embodiment, computing device110may be a laptop computer, a notebook, tablet computer, netbook computer, personal computer (PC), a desktop computer, a personal digital assistant (PDA), a smart phone, a thin client, or any other electronic device or computing system capable of receiving and sending data to and from other computing devices. While computing device110is shown as a single device, in other embodiments, computing device110may be comprised of a cluster or plurality of computing devices, working together or working separately. Computing device110is described in more detail with reference toFIG. 6.

In the example embodiment, scaling program112is a program on computing device110which is capable of determining when information, such as replicated data122, within a node, such as one of nodes120, of an original cluster reaches a threshold storage capacity. Scaling program112is additionally capable of receiving a user input indicating a request to “recluster” the original cluster and, in response to determining that a node has reached the threshold storage capacity and/or receiving a request to recluster, scaling program112is capable of capturing a snapshot of replicated data122within the original cluster. Scaling program112is further capable of partitioning the original cluster such that replicated data122is split between the original cluster and one or more clusters and initializing the new cluster(s) with data available from the snapshot. Moreover, scaling program112is capable of determining whether a request has been transmitted to the incorrect cluster and, if so, forwarding the request to the appropriate cluster.

FIG. 2is a flowchart depicting the operation of scaling program112in providing on-demand extensibility of a quorum-based replication server via “reclustering” without the need to predefine a master cluster nor temporarily shut down the replication service. In the example embodiment, reclustering is the process of partitioning the nodes and data contained in a single replica set into multiple replica sets. For example, if a replica set stores tenants with last names A-Z, then scaling program112splits the nodes within the original cluster to create an additional new cluster and stores half of the data (tenants A-M) on the original cluster of nodes and the other half of the data (tenants N-Z) on the new cluster of nodes.

With reference toFIG. 2, scaling program112determines whether a node within nodes120of an original replica set has reached a threshold capacity (decision202). In the example embodiment, scaling program112determines whether a node of nodes120has reached a threshold capacity, for example 90% of dedicated replication disk space, by communicating with the operating systems of nodes120. In the example embodiment, a user may alter the threshold capacity such that nodes120may be filled to desired amounts up to 100%. Furthermore, in the example embodiment, nodes120have the same storage capacities. Nodes within the same replica set tend to have the same storage capacities because replication sets are designed to store the same data in as many places as possible. Therefore, if all nodes within a replica set are responsible for storing the same exact data, for example 100 GB of tenancy information, then all of the nodes within the cluster need capacities of 100 GB or more. For a node to have less capacity than others in the cluster would render it less useful and for a node to have greater capacity would be wasteful. It is because of this need for consistent capacities that nodes within the same replica sets tend to have common storage capacity. If, for example, one of the nodes within the replica set had only 64 GB capacity, then it would not be capable of replicating (backing up) the 100 GB of tenancy information and would reduce the effectiveness of the replica set. Therefore, when a single node120reaches storage capacity, such as node120A, it is likely that other nodes120within the same cluster will reach storage capacity. For example, using a threshold of 90% capacity, if a replication set comprising nodes120A-D having 100 GB capacity per node are each storing a replicated (identical) 90 GB of tenancy information, then scaling program112determines that 90% of the capacity of nodes120A-D has been reached. Similarly, using the same example, even if nodes120had differing storage capacities and only node120A of the replica set reaches 90% capacity, then scaling program112determines that one of the nodes of nodes120has reached 90% capacity.

If scaling program112determines that storage capacity of a node within a cluster has not been reached (decision202“NO” branch), then scaling program112determines whether computing device110has received a user input indicating a request to recluster the original cluster (decision204). In the example embodiment, a recluster command is received by scaling program112from a user via a user interface of computing device110and denotes a function which partitions replicated data122of a single, original cluster into two or more clusters (described in greater detail below).

If scaling program112determines that it has not received a recluster command from a user (decision204“NO” branch), then scaling program112continues to determine whether a node within the original cluster has reached storage capacity (decision202).

If scaling program112determines that the storage capacity of a node within an original replica set has been reached (decision202“YES” branch) or scaling program112receives a user command to recluster (decision204“YES” branch), then scaling program112captures a snapshot of the original cluster (step206). In the example embodiment, the snapshot of the original cluster provides scaling program112with the contents, structure, metadata, and organization of replicated data122within the original cluster. In the example embodiment, the snapshot provides an overview of replicated data122such that scaling program112can determine the most effective and efficient means to partition/split the contents between the original cluster and a new cluster. Continuing the example above where replicated data122details tenancy information for an apartment complex, scaling program112captures a snapshot of the 100 GB of tenancy information within the original cluster to identify a table which categorizes the tenants of the apartment complex by name, year, building, unit, and unit type.

Scaling program112determines an optimal partition of the original cluster (step208). Because replicated data122is being partitioned from an original cluster into two or more clusters (e.g. rather than storing all of replicated data122on each of multiple nodes making up a replica set, the nodes making up the original cluster are split between the original cluster and a new cluster and a portion of replicated data122is stored on the original cluster nodes while the other portion of replicated data122is stored on the new cluster nodes), scaling program112determines an optimal split of replicated data122between the original partition and the new partition such that related information is kept in the same cluster. Storing related replicated data122within the same clusters is advantageous organizationally and functionally because not only is it easier and quicker to collect/modify/add to large groups of related information from the same location, but rationally grouping related information together is more likely to result in less requests reaching the incorrect cluster following partitioning. While, in some embodiments, all information within replicated data122may be related, scaling program112seeks to determine an optimal partition of related information by analyzing the cluster snapshot for several different characteristics (described in greater detail below). In the example embodiment, scaling program112determines whether information contained replicated data122, such as folders and files, are related to one another using natural language processing techniques in conjunction with reference to a relational database.

In the example embodiment, scaling program112first identifies terms within the names and/or metadata associated with the largest container(s) within replicated data122, such as folders (step208continued). Containers that are identified to have similarities in name/metadata through character matching techniques, such as determining whether matching characters exceed a threshold percentage relative to word length (position sensitive), are assumed to be related and assigned a relation value indicating a relationship likelihood. For example, the folders “Tenants_2014” and “Tenants_2015” both share more than 90% of the same characters relative to word length ( 11/12 matching characters in both folder names) and are, therefore, considered related by scaling program112. Furthermore, the assigned relation value may be weighted based on the amount of matching characters within terms such that more character matches within terms result in a greater relation value and thus a greater relationship likelihood. For example, terms which have ≥50% matching characters are assigned a relation value of 1, terms which have ≥75% matching characters are assigned a relation value of 3, and terms which have ≥95% matching characters is assigned a relation value of 5. Moreover, scaling program112may further utilize natural language processing techniques to recognize patterns and/or conventions within the names and metadata of identified containers to assign relation values. Using the example above, scaling program112may determine that the “2014” and “2015” of the folders “Tenants_2014” and “Tenants_2015” are indicative of a year and assign a relation value of 3 based on the identified pattern as well as assign a relation value of 5 based on the remaining matching characters of “Tenants_” and “Tenants_” matching ≥95% (100%), summing to a total relation value of 8.

Conversely, containers which do not have similar names and/or metadata, for example terms with <50% matching characters, are categorized by referencing a relational database to determine whether the names and/or metadata associated with the container belong to any of the same or similar categories (step208continued). For example, the folder “Tenants_Studio” and the folder “Tenants_OneBedroom” contain terms “Tenant,” “Studio,” and “OneBedroom” which may all be associated with the category of housing. If scaling program112determines that one or more containers are likely to be related based on associated name/metadata, scaling program112may be configured to perform a similar analysis to that above on the contents within the related containers, such as additional containers and files, to ensure the accuracy of and further define the relation value. In the example embodiment, if the relationship identified between the related containers is bolstered by relationships found within the contents of the related containers, then scaling program112may increase the relation value to reflect the greater likelihood of relatedness. Continuing the example above where the folders “Tenants_2014” and “Tenants_2015” are assigned a relationship value of 8, if the folders “Tenants_2014” and “Tenants_2015” contain files named John_Smith and Kathy_Johnson, respectively, and scaling program112compares John Smith and Kathy Johnson to the relational database to determine that they are both names, then scaling program112increases the relation value from 8 to 10 because both folders contain files related by the category “names.” Conversely, if the terms within the contents of the related containers are not related, scaling program112maintains or reduces the relationship value. It should be noted that the aforementioned methods of partitioning replicated data122are merely examples and other methods may be implemented.

The relation values are then compared to one another and the contents of replicated data122which exhibit the highest relation value are grouped together to be partitioned to the same cluster (step208continued). In the example embodiment, scaling program112may be further configured to partition replicated data122by other metrics, such as alphabetical order, creation date, last use date (hotness), file size, file type, author, etc. Furthermore, in the example embodiment, scaling program112attempts to tailor the partitioning of replicated data122such that half of the replicated data in the original cluster is partitioned into a new cluster. For example, if nodes120reach storage capacity and replicated data122is tenancy data of tenants A-Z, then scaling program112partitions the data such that nodes120A and120B contain tenant information for tenants A-M while nodes120C and120D store tenant information for tenants N-Z. By partitioning half of the replicated data to a new cluster, it is more likely that all nodes120have freed up similar capacities to receive new replicated data122.

Scaling program112initializes new cluster(s) based on the information detailed by the snapshot and the determined partitioning of replicated data122(step210). For a more thorough explanation of initializing a new cluster (step210), please refer toFIGS. 3-5throughout this discussion as directed.

With reference now toFIG. 3,FIG. 3illustrates original cluster302just prior to initializing a new cluster. Note that prior to reclustering, nodes120all contain both data X and data Y as nodes within original cluster302, analogous to the tenancy information A-Z in the example introduced earlier. In addition, data X and data Y are visually depicted to occupy all of replicated data122, indicating that the nodes are reaching maximum storage capacity and a recluster is necessary. It should be further noted that although data X is identical in all instances shown, the location at which data X is stored (i.e. node120A, node120B, etc.) varies and the same reference characters are used merely for simplicity.

Now with reference toFIG. 4, when nodes120within original cluster302reach storage capacity, scaling program112initiates one or more nodes from original cluster302within new cluster404, as shown by nodes120C and120D. During this transfer, data X and data Y (tenant data A-Z) are maintained on all nodes120, allowing it to remain accessible throughout the reclustering process despite which replica set the node is part of. Moreover, the cluster snapshot obtained in step206and determined partitioning in step208allow scaling program112to determine in which cluster data is now stored.

With reference now toFIG. 5,FIG. 5illustrates original cluster302and new cluster404following initialization of new cluster404. Having split nodes120into original cluster302and new cluster404, scaling program112now removes redundant information from nodes120or, in other words, removes the replicated data that is to be partitioned to another cluster. For example, if original cluster302is to store data X (tenant data A-M), then scaling program112removes data Y (tenant data N-Z) from nodes120A and120B. Similarly, if new cluster404is to store data Y (tenant data N-Z), then scaling program removes data X (tenant information A-M) from nodes120C and120D. It is worth noting that although replicated data122is now only replicated on two nodes (data X on nodes120A and120B (2 instances), data Y on nodes120C and120D (2 instances) rather than data X and data Y on nodes120A-D (4 instances)), additional nodes can be added to original cluster302or new cluster404to increase the number of backups. Moreover, having partitioned the entirety of the data, nodes120are now only storing a portion of the data, as depicted by the removal of data X and data Y from respective clusters inFIG. 5, and are no longer at maximum storage capacity. Therefore, the need to increase storage capacity across all nodes within the replica set has been eliminated by reducing the amount of data stored by each cluster. Moreover, the existing nodes are capable of replicating more data and additional backup nodes can be added at a later time to either of original cluster302or new cluster404. It is worth noting that new replica sets, such as new cluster404, may be initialized with new nodes, such as a node not included within nodes120, in a similar manner to that above by replicating data from nodes120to the new node prior to removing redundant information. Moreover, replica sets, including newly created replica sets, can be reclustered any number of times.

With reference now toFIG. 2, scaling program112determines whether a read, write, or other request has reached the wrong cluster (decision212). In the example embodiment, clusters are identified by a cluster ID which catalogs the replicated data122which is stored within the nodes of the cluster. In order to determine whether a request has reached the incorrect cluster, scaling program112analyses the request to identify the file, folder, etc. which is to be written, read, or modified otherwise and compares it to the cluster ID associated with the cluster receiving the request. If the request is associated with a file matching a file described by the cluster ID, then the request has reached the correct cluster.

If scaling program112determines that a request has reached the wrong cluster (decision212“YES” branch), then scaling program112references the cluster IDs of alternative clusters to determine in which cluster the replicated data122containing the requested file exists. Having identified the appropriate cluster for the request, scaling program112then forwards the request to the appropriate cluster (step214). In addition, scaling program112may edit the cluster ID's corresponding to the relevant clusters to prevent further mix-ups.

FIG. 3illustrates an original configuration of a cluster within a data replication system.

FIG. 4illustrates a multiple cluster configuration following the reclustering of the original cluster illustrated inFIG. 3.

FIG. 5illustrates an alternative multiple cluster configuration following the reclustering of the original cluster illustrated inFIG. 3.

Computing device110may include one or more processors02, one or more computer-readable RAMs04, one or more computer-readable ROMs06, one or more computer readable storage media08, device drivers12, read/write drive or interface14, network adapter or interface16, all interconnected over a communications fabric18. Communications fabric18may be implemented with any architecture designed for passing data and/or control information between processors (such as microprocessors, communications and network processors, etc.), system memory, peripheral devices, and any other hardware components within a system.

One or more operating systems10, and one or more application programs11, for example, scaling program112, are stored on one or more of the computer readable storage media08for execution by one or more of the processors02via one or more of the respective RAMs04(which typically include cache memory). In the illustrated embodiment, each of the computer readable storage media08may be a magnetic disk storage device of an internal hard drive, CD-ROM, DVD, memory stick, magnetic tape, magnetic disk, optical disk, a semiconductor storage device such as RAM, ROM, EPROM, flash memory or any other computer-readable tangible storage device that can store a computer program and digital information.

Computing device110may also include a R/W drive or interface14to read from and write to one or more portable computer readable storage media26. Application programs11on computing device110may be stored on one or more of the portable computer readable storage media26, read via the respective R/W drive or interface14and loaded into the respective computer readable storage media08.

Computing device110may also include a display screen20, a keyboard or keypad22, and a computer mouse or touchpad24. Device drivers12interface to display screen20for imaging, to keyboard or keypad22, to computer mouse or touchpad24, and/or to display screen20for pressure sensing of alphanumeric character entry and user selections. The device drivers12, R/W drive or interface14and network adapter or interface16may comprise hardware and software (stored on computer readable storage media08and/or ROM06).

Characteristics are as follows:

Service Models are as follows:

Deployment Models are as follows: