Tracking status and restarting distributed replication
A controller at a source site generates a set of tasks associated with a replication job. Each task involves a source worker node from among a set of source worker nodes at the source site, a destination worker node from among a set of destination worker nodes at the destination site, and includes one or more of copying an object from the source to destination site, or deleting an object from the destination site. Status update messages concerning the tasks are received at a message queue connected between the controller and the set of source worker nodes. The status update messages are logged into a persistent key-value store. Upon a failure to complete the replication job, the key-value store is accessed to identify tasks that were and were not completed before the failure. The tasks that were not completed are resent to the source worker nodes.
1. A method comprising:
generating, by a controller at a source site, a plurality of tasks associated with a replication job, each task involving a source worker node from among a plurality of source worker nodes at the source site, a destination worker node from among a plurality of destination worker nodes at a destination site, and comprising one or more of copying an object from the source site to the destination site, or deleting an object from the destination site;
receiving, at a message queue coupled between the controller and the plurality of source worker nodes, status update messages concerning the plurality of tasks from the source worker nodes;
logging the status update messages into a persistent key-value store;
upon a failure to complete the replication job, accessing the key-value store to identify tasks of the plurality of tasks that were and were not completed before the failure; and
resending the tasks that were not completed to the source worker nodes and not resending the tasks that were completed.
2. The method of claim 1 comprising:
calculating differences between a first snapshot created at the source site at a first time and a second snapshot created at the source site at a second time, after the first time, to generate the plurality of tasks;
after the failure to complete the replication job, recalculating the differences between the first and second snapshots;
determining, from an examination of the key-value store, that a first difference between the first and second snapshots and corresponding to a first task of the plurality of tasks was completed;
determining, from the examination of the key-value store, that a second difference between the first and second snapshots and corresponding to a second task of the plurality of tasks was not completed; and
resending the second task to a second message queue and not resending the first task.
3. The method of claim 1 comprising:
periodically checking the message queue for the status update messages;
determining that an amount of time elapsed since a particular source worker node last reported with a status update message associated with a particular task has exceeded a threshold time period; and
upon the determination, resending the particular task to the message queue.
4. The method of claim 1 comprising:
periodically checking the message queue for the status update messages;
determining, from the periodic check, that a particular source worker node has reported an error; and
sending an alert to an administrator concerning the error.
5. The method of claim 1 wherein a status update message specifies an object to be replicated from the source site to the destination site, and identifies portions of the object already replicated from the source site to the destination site.
6. The method of claim 1 wherein the resending the tasks that were not completed comprises:
resending the tasks to a message queue for retrieval by the source worker nodes.
7. A system comprising:
a processor-based system executed on a computer system and configured to:
generate, by a controller at a source site, a plurality of tasks associated with a replication job, each task involving a source worker node from among a plurality of source worker nodes at the source site, a destination worker node from among a plurality of destination worker nodes at a destination site, and comprising one or more of copying an object from the source site to the destination site, or deleting an object from the destination site;
receive, at a message queue coupled between the controller and the plurality of source worker nodes, status update messages concerning the plurality of tasks from the source worker nodes;
log the status update messages into a persistent key-value store;
upon a failure to complete the replication job, access the key-value store to identify tasks of the plurality of tasks that were and were not completed before the failure; and
resend the tasks that were not completed to the source worker nodes and not resend the tasks that were completed.
8. The system of claim 7 wherein the processor-based system is configured to:
calculate differences between a first snapshot created at the source site at a first time and a second snapshot created at the source site at a second time, after the first time, to generate the plurality of tasks;
after the failure to complete the replication job, recalculate the differences between the first and second snapshots;
determine, from an examination of the key-value store, that a first difference between the first and second snapshots and corresponding to a first task of the plurality of tasks was completed;
determine, from the examination of the key-value store, that a second difference between the first and second snapshots and corresponding to a second task of the plurality of tasks was not completed; and
resend the second task to a second message queue and not resend the first task.
9. The system of claim 7 wherein the processor-based system is configured to:
periodically check the message queue for the status update messages;
determine that an amount of time elapsed since a particular source worker node last reported with a status update message associated with a particular task has exceeded a threshold time period; and
upon the determination, resend the particular task to the message queue.
10. The system of claim 7 wherein the processor-based system is configured to:
periodically check the message queue for the status update messages;
determine, from the periodic check, that a particular source worker node has reported an error; and
send an alert to an administrator concerning the error.
11. The system of claim 7 wherein a status update message specifies an object to be replicated from the source site to the destination site, and identifies portions of the object already replicated from the source site to the destination site.
12. 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:
generating, by a controller at a source site, a plurality of tasks associated with a replication job, each task involving a source worker node from among a plurality of source worker nodes at the source site, a destination worker node from among a plurality of destination worker nodes at a destination site, and comprising one or more of copying an object from the source site to the destination site, or deleting an object from the destination site;
receiving, at a message queue coupled between the controller and the plurality of source worker nodes, status update messages concerning the plurality of tasks from the source worker nodes;
logging the status update messages into a persistent key-value store;
upon a failure to complete the replication job, accessing the key-value store to identify tasks of the plurality of tasks that were and were not completed before the failure; and
resending the tasks that were not completed to the source worker nodes and not resending the tasks that were completed.
13. The computer program product of claim 12 wherein the method comprises:
calculating differences between a first snapshot created at the source site at a first time and a second snapshot created at the source site at a second time, after the first time, to generate the plurality of tasks;
after the failure to complete the replication job, recalculating the differences between the first and second snapshots;
determining, from an examination of the key-value store, that a first difference between the first and second snapshots and corresponding to a first task of the plurality of tasks was completed;
determining, from the examination of the key-value store, that a second difference between the first and second snapshots and corresponding to a second task of the plurality of tasks was not completed; and
resending the second task to a second message queue and not resending the first task.
14. The computer program product of claim 12 wherein the method comprises:
periodically checking the message queue for the status update messages;
determining that an amount of time elapsed since a particular source worker node last reported with a status update message associated with a particular task has exceeded a threshold time period; and
upon the determination, resending the particular task to the message queue.
15. The computer program product of claim 12 wherein the method comprises:
periodically checking the message queue for the status update messages;
determining, from the periodic check, that a particular source worker node has reported an error; and
sending an alert to an administrator concerning the error.
16. The computer program product of claim 12 wherein a status update message specifies an object to be replicated from the source site to the destination site, and identifies portions of the object already replicated from the source site to the destination site.
17. The computer program product of claim 12 wherein the resending the tasks that were not completed comprises:
resending the tasks to a message queue for retrieval by the source worker nodes.