Distributed object replication architecture
A source site includes a controller, a set of source worker nodes, and a message queue connected between the controller and source worker nodes. The message queue receives messages and stores the messages for retrieval. A destination site includes a set of destination worker nodes. Tasks are generated to replicate changes to objects at the source site to the destination site. The controller pushes messages corresponding to the tasks onto the message queue. A source worker node retrieves a message corresponding to a task from the message queue for processing in conjunction with a destination worker node. The message is indicated as having been retrieved from the message queue.
1. A method comprising:
providing, at a source site, a controller, a plurality of source worker nodes, and a message queue, coupled between the controller and the plurality of source worker nodes, wherein the source site comprises an object storage system storing objects, the objects having been split into a plurality of segments;
providing, at a destination site, a plurality of destination worker nodes;
as objects are written to and deleted from the object storage system at the source site, maintaining a log listing, by user-generated names given to the objects, the objects that have been written to the object storage system at the source site and the objects that have been deleted from the object storage system at the source site;
upon starting a replication job, generating a set of tasks comprising tasks to replicate objects written to the object storage system at the source site to the destination site according to the log, and tasks to delete objects from the destination site that have been deleted from the object storage system at the source site according to the log;
pushing, by the controller, the tasks onto the message queue;
retrieving, by a first source worker node, a first task from the message queue for processing in conjunction with a first destination worker node;
when the first task is to delete an object from the destination site, making a remote procedure call (RPC) to the first destination worker node requesting that the object be deleted; and
when the first task is to replicate an object from the source to destination site,
determining which segments of the object exist at the destination site and which segments of the object do not exist at the destination site, the segments of the object not existing at the destination site thereby being missing segments; and
sending the missing segments of the object to the destination site,
wherein the object storage system storing objects at the source site comprises object recipes corresponding to the objects, each object recipe referencing one or more slice recipes, each slice recipe comprising a plurality of fingerprints corresponding to the segments into which an object has been split, and each fingerprint mapping to a respective segment, and
wherein replication of the object from the source to destination site further comprises:
issuing, to the first destination worker node, a query comprising a batch of slice recipes from an object recipe corresponding to the object to be replicated;
checking, by the first destination worker node, which segments of the object identified by fingerprints in the batch of slice recipes already exist at the destination site;
based on the checking, replying to the first source worker node with a listing of fingerprints from the batches of slice recipes, wherein the listing of fingerprints correspond to segments of the object that are missing at the destination site thereby allowing the first source worker node to retrieve and send the missing segments to the destination site; and
repeating the issuing a query comprising a batch of slice recipes until all slice recipes of the object recipe have been sent to the first destination worker node.
2. The method of claim 1 further comprising:
holding the tasks on the message queue for retrieval by the plurality of source worker nodes;
determining that a number of tasks on the message queue has reached a threshold limit; and
pausing the generating tasks and the pushing of the tasks onto the message queue while allowing the plurality of source worker nodes to continue retrieving tasks from the message queue, thereby freeing space on the message queue.
3. The method of claim 2 further comprising:
after the pausing, continuing the generating tasks and the pushing of the tasks onto the message queue.
4. The method of claim 1 further comprising
logging in a key-value store the batch of one or more slice recipes that the first source worker node has sent to the first destination worker node; and
upon a failure during the replication of the object, restarting the replication from a last logged position of the batch of one or more slice recipes that the first source worker node has sent to the first destination worker node according to the key-value store.
5. The method of claim 1 further comprising:
designating different controllers to handle different portions of a replication job, the replication job comprising replicating the changes to the objects at the source site to the destination site.
6. The method of claim 1 wherein names of the objects listed in the log comprise timestamps appended to the user-generated names indicating when the objects were written to the object storage system at the source site.
7. The method of claim 1 wherein the message queue resides in volatile memory.
8. The method of claim 1 wherein a cost of the replication job is subject to spot pricing and the method further comprises:
monitoring a price associated with renting a source working node, a destination worker node, or both;
when the price is above a user-configured threshold price, pausing the replication job; and
when the price falls below the user-configured threshold price, resuming the replication job.
9. A system for replicating data, the system comprising: a processor, and memory storing one or more sequences of instructions which, when executed by the processor, cause the processor to carry out the steps of:
providing, at a source site, a controller, a plurality of source worker nodes, and a message queue, coupled between the controller and the plurality of source worker nodes, wherein the source site comprises an object storage system storing objects, the objects having been split into a plurality of segments;
providing, at a destination site, a plurality of destination worker nodes;
as objects are written to and deleted from the object storage system at the source site, maintaining a log of the objects that have been written and the objects that have been deleted;
upon starting a replication job, generating a set of tasks comprising tasks to replicate objects written to the object storage system at the source site to the destination site according to the log, and tasks to delete objects from the destination site that have been deleted from the object storage system at the source site according to the log;
pushing, by the controller, the tasks onto the message queue;
retrieving, by a first source worker node, a first task from the message queue for processing in conjunction with a first destination worker node;
when the first task is to delete an object from the destination site, making a remote procedure call (RPC) to the first destination worker node requesting that the object be deleted; and
when the first task is to replicate an object from the source to destination site,
determining which segments of the object exist at the destination site and which segments of the object do not exist at the destination site, the segments of the object not existing at the destination site thereby being missing segments; and
sending the missing segments of the object to the destination site,
wherein the object storage system storing objects at the source site comprises object recipes corresponding to the objects, each object recipe referencing one or more slice recipes, each slice recipe comprising a plurality of fingerprints corresponding to the segments into which an object has been split, and each fingerprint mapping to a respective segment, and
wherein replication of the object from the source to destination site further comprises:
sending, by the first source worker node to the first destination worker node, one or more slice recipes from an object recipe corresponding to the object to be replicated, wherein the first destination worker node checks the fingerprints in the slice recipes against fingerprints of segments already existing at the destination site to determine which segments of the object are missing at the destination site;
logging in a key-value store the one or more slice recipes that the first source worker node has sent to the first destination worker node; and
upon a failure during the replication of the object, restarting the replication from a last logged position of the one or more slice recipes that the first source worker node has sent to the first destination worker node according to the key-value store.
10. The system of claim 9 wherein the processor further carries out the steps of:
holding the tasks on the message queue for retrieval by the plurality of source worker nodes;
determining that a number of tasks on the message queue has reached a threshold limit; and
pausing the generating tasks and the pushing of the tasks onto the message queue while allowing the plurality of source worker nodes to continue retrieving tasks from the message queue, thereby freeing space on the message queue.
11. The system of claim 10 wherein the processor further carries out the steps of:
after the pausing, continuing the generating tasks and the pushing of the tasks onto the message queue.
12. The system of claim 9 wherein the processor further carries out the steps of:
designating different controllers to handle different portions of a replication job, the replication job comprising replicating the changes to the objects at the source site to the destination site.
13. The system of claim 9 wherein names of the objects listed in the log comprise user-generated names given to the objects and timestamps appended to the user-generated names indicating when the objects were written to the object storage system at the source site.
14. A computer program product comprising a non-transitory computer-readable medium having a computer-readable program code embodied therein, the computer-readable program code adapted to be executed by one or more processors to implement a method comprising:
providing, at a source site, a controller, a plurality of source worker nodes, and a message queue, coupled between the controller and the plurality of source worker nodes, wherein the source site comprises an object storage system storing objects, the objects having been split into a plurality of segments;
providing, at a destination site, a plurality of destination worker nodes;
as objects are written to and deleted from the object storage system at the source site, maintaining a log of the objects that have been written and the objects that have been deleted;
upon starting a replication job, generating a set of tasks comprising tasks to replicate objects written to the object storage system at the source site to the destination site according to the log, and tasks to delete objects from the destination site that have been deleted from the object storage system at the source site according to the log;
pushing, by the controller, the tasks onto the message queue;
retrieving, by a first source worker node, a first message corresponding to a first task from the message queue for processing in conjunction with a first destination worker node;
when the first task is to delete an object from the destination site, making a remote procedure call (RPC) to the first destination worker node requesting that the object be deleted;
when the first task is to replicate an object from the source to destination site,
determining which segments of the object exist at the destination site and which segments of the object do not exist at the destination site, the segments of the object not existing at the destination site thereby being missing segments; and
sending the missing segments of the object to the destination site;
tracking a depth of the message queue; and
based on the depth of the message queue, adjusting a number of source worker nodes and a number of destination worker nodes.
15. The computer program product of claim 14 wherein the method further comprises:
holding the tasks on the message queue for retrieval by the plurality of source worker nodes;
determining that a number of tasks on the message queue has reached a threshold limit; and
pausing the generating tasks and the pushing of the tasks onto the message queue while allowing the plurality of source worker nodes to continue retrieving tasks from the message queue, thereby freeing space on the message queue.
16. The computer program product of claim 15 wherein the method further comprises:
after the pausing, continuing the generating tasks and the pushing of the tasks onto the message queue.
17. The computer program product of claim 14 wherein the object storage system storing objects at the source site comprises object recipes corresponding to the objects, each object recipe referencing one or more slice recipes, each slice recipe comprising a plurality of fingerprints corresponding to the segments into which an object has been split, and each fingerprint mapping to a respective segment, and
wherein replication of the object from the source to destination site further comprises:
sending, by the first source worker node to the first destination worker node, one or more slice recipes from an object recipe corresponding to the object to be replicated, wherein the first destination worker node checks the fingerprints in the slice recipes against fingerprints of segments already existing at the destination site to determine which segments of the object are missing at the destination site;
logging in a key-value store the one or more slice recipes that the first source worker node has sent to the first destination worker node; and
upon a failure during the replication of the object, restarting the replication from a last logged position of the one or more slice recipes that the first source worker node has sent to the first destination worker node according to the key-value store.
18. The computer program product of claim 14 wherein the method further comprises:
designating different controllers to handle different portions of a replication job, the replication job comprising replicating the changes to the objects at the source site to the destination site.
19. The computer program product of claim 14 wherein names of the objects listed in the log comprise user-generated names given to the objects and timestamps appended to the user-generated names indicating when the objects were written to the object storage system at the source site.