Distributed replication and deduplication of an object from a source site to a destination site

1. A method comprising:

storing, at a source site, a plurality of object recipes corresponding to a plurality of objects at the source site that have been divided into a plurality of segments, each object recipe referencing one or more slice recipes, each slice recipe comprising a plurality of fingerprints corresponding to segments into which an object has been divided, each fingerprint mapping to a respective segment and being calculated by applying a hash to content of the respective segment, and the plurality of objects being stored in object storage at the source site, the plurality of objects thereby being stored in a flat address space;

fetching, by a first source worker node at the source site, a first task from a message queue holding a plurality of tasks, the first task specifying replicating a first object at the source site to a destination site;

issuing, from the first source worker node to the destination site, a first request for a first connection, the first request being received by a load balancer at the destination site and assigned to a first destination worker node;

establishing the first connection between the first source and destination worker nodes;

determining that the destination site does not include an object having a same name, timestamp, size, and checksum as the first object;

upon the determining, replicating the first object to the destination site by:

transmitting from the first source worker node to the first destination worker node each of one or more slice recipes referenced by a first object recipe corresponding to the first object, the transmitting comprising grouping each of the one or more slice recipes into a package until a size of the package reaches a threshold size, and upon the size of the package reaching the threshold size, transferring the package of slice recipes from the first source worker node to the first destination worker node;

receiving at the first source worker node from the first destination worker node a list identifying at least a subset of fingerprints from the one or more slice recipes, the subset of fingerprints corresponding to segments of the first object that the first destination worker node has determined are missing at the destination site;

transmitting from the first source worker node to the first destination worker node a compression region comprising compressed segments of the first object missing at the destination site; and

transmitting from the first source worker node to the first destination worker node the first object recipe, copies of the first object and the first object recipe thereby now being at both the source and destination sites;

while the first object is being replicated to the destination site by the first source and destination worker nodes, allocating a new pair of worker nodes comprising a second source worker node at the source site and a second destination worker node at the destination site;

fetching, by the second source worker node, a second task from the message queue, the second task specifying replicating a second object at the source site to the destination site;

issuing, from the first source worker node to the destination site, a second request for a second connection, the second request being received by the load balancer at the destination site and assigned to the second destination worker node;

establishing the second connection between the second source and destination worker nodes;

determining that the destination site does not include an object having a same name, timestamp, size, and checksum as the second object;

upon the determining, using the second source and destination worker nodes to replicate the second object to the destination site while the first object is being replicated to the destination site using the first source and destination worker nodes;

after completing replication of the first object to the destination site, deallocating the first source and destination worker nodes; and

after completing replication of the second object to the destination site, deallocating the second source and destination worker nodes.

2. The method of claim 1 comprising:

fetching, by a third source worker node at the source site, a third task from the message queue, the third task specifying deleting a third object from the destination site;

issuing, from the third source worker node to the destination site, a third request for a third connection, the third request being received by the load balancer and assigned to a third destination worker node;

establishing the third connection between the third source and destination worker nodes; and

after the establishing the third connection, instructing the third destination worker node to delete the third object.

3. The method of claim 1 wherein the source and destination sites comprise cloud storage and the method further comprises:

providing a replication tool to be executed at a client device, the client device not offering the cloud storage;

receiving, from a user at the client, an instruction to replicate a local file on the client to the cloud storage; and

replicating, using the replication tool, the local file at the client to the cloud storage.

4. The method of claim 1 wherein the establishing the first connection between the first source and destination worker nodes comprises:

transmitting an authentication key from the first source worker node to the first destination worker node to authenticate the first source worker node.

5. The method of claim 1 wherein the first slice recipe is organized into a similarity group having other slice recipes, and during the replicating the first object, the first destination worker node does not deduplicate the first object using slice recipes outside the similarity group, thereby allowing duplicate segments of the first object to be replicated at the destination site when the duplicate segments are referenced by the slice recipes outside the similarity group.

6. The method of claim 1 wherein the first object recipe forms a one-to-one mapping to the first object.

7. The method of claim 1 further comprising:

generating, by a controller, the plurality of tasks for replicating from the source to destination site;

pushing the plurality of tasks onto the message queue, the message queue comprising a limit on a number of tasks that can be held in the message queue;

determining that the message queue is full; and

upon determining that the message queue is full, pausing the generating and pushing until there is space on the message queue as source worker nodes complete their tasks and pull new tasks from the message queue.

8. A 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:

storing, at a source site, a plurality of object recipes corresponding to a plurality of objects at the source site that have been divided into a plurality of segments, each object recipe referencing one or more slice recipes, each slice recipe comprising a plurality of fingerprints corresponding to segments into which an object has been divided, each fingerprint mapping to a respective segment and being calculated by applying a hash to content of the respective segment, and the plurality of objects being stored in object storage at the source site, the plurality of objects thereby being stored in a flat address space;

fetching, by a first source worker node at the source site, a first task from a message queue holding a plurality of tasks, the first task specifying replicating a first object at the source site to a destination site;

issuing, from the first source worker node to the destination site, a first request for a first connection, the first request being received by a load balancer at the destination site and assigned to a first destination worker node;

establishing the first connection between the first source and destination worker nodes;

determining that the destination site does not include an object having a same name, timestamp, size, and checksum as the first object;

upon the determination, replicating the first object to the destination site by:

transmitting from the first source worker node to the first destination worker node each of one or more slice recipes referenced by a first object recipe corresponding to the first object, the transmitting comprising grouping each of the one or more slice recipes into a package until a size of the package reaches a threshold size, and upon the size of the package reaching the threshold size, transferring the package of slice recipes from the first source worker node to the first destination worker node;

receiving at the first source worker node from the first destination worker node a list identifying at least a subset of fingerprints from the one or more slice recipes, the subset of fingerprints corresponding to segments of the first object that the first destination worker node has determined are missing at the destination site;

transmitting from the first source worker node to the first destination worker node a compression region comprising compressed segments of the first object missing at the destination site; and

transmitting from the first source worker node to the first destination worker node the first object recipe, copies of the first object and the first object recipe thereby now being at both the source and destination sites;

generating, by a controller, the plurality of tasks for replicating from the source to destination site;

pushing the plurality of tasks onto the message queue, the message queue comprising a limit on a number of tasks that can be held in the message queue;

determining that the message queue is full; and

upon determining that the message queue is full, pausing the generating and pushing until there is space on the message queue as source worker nodes complete their tasks and pull new tasks from the message queue.

9. The system of claim 8 wherein the processor further carries out the steps of:

fetching, by a second source worker node at the source site, a second task from the message queue, the second task specifying deleting a second object from the destination site;

issuing, from the second source worker node to the destination site, a second request for a second connection, the second request being received by the load balancer and assigned to a second destination worker node;

establishing the second connection between the second source and destination worker nodes; and

after the establishing the second connection, instructing the second destination worker node to delete the second object.

10. The system of claim 8 wherein the source and destination sites comprise cloud storage and the processor further carries out the steps of:

providing a replication tool to be executed at a client device, the client device not offering the cloud storage;

receiving, from a user at the a client, an instruction to replicate a local file on the client to the cloud storage; and

receiving, via the replication tool, segments of the local file at the cloud storage.

11. The system of claim 8 wherein the processor further carries out the step of:

transmitting an authentication key from the first source worker node to the first destination worker node to authenticate the first source worker node.

12. The system of claim 8 wherein the first slice recipe is organized into a similarity group having other slice recipes, and during the replicating the first object, the first destination worker node does not deduplicate the first object using slice recipes outside the similarity group, thereby allowing duplicate segments of the first object to be replicated at the destination site when the duplicate segments are referenced by the slice recipes outside the similarity group.

13. The system of claim 8 wherein the processor further carries out the steps of:

while the first object is being replicated to the destination site by the first source and destination worker nodes, allocating a new pair of worker nodes comprising a second source worker node at the source site and a second destination worker node at the destination site;

fetching, by the second source worker node, a second task from the message queue, the second task specifying replicating a second object at the source site to the destination site;

issuing, from the first source worker node to the destination site, a second request for a second connection, the second request being received by the load balancer at the destination site and assigned to the second destination worker node;

establishing the second connection between the second source and destination worker nodes;

determining that the destination site does not include an object having a same name, timestamp, size, and checksum as the second object;

upon the determining, using the second source and destination worker nodes to replicate the second object to the destination site while the first object is being replicated to the destination site using the first source and destination worker nodes;

after completing replication of the first object to the destination site, deallocating the first source and destination worker nodes; and

after completing replication of the second object to the destination site, deallocating the second source and destination worker nodes.

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:

storing, at a source site, a plurality of object recipes corresponding to a plurality of objects at the source site that have been divided into a plurality of segments, each object recipe referencing one or more slice recipes, each slice recipe comprising a plurality of fingerprints corresponding to segments into which an object has been divided, each fingerprint mapping to a respective segment and being calculated by applying a hash to content of the respective segment, and the plurality of objects being stored in object storage at the source site, the plurality of objects thereby being stored in a flat address space;

fetching, by a first source worker node at the source site, a first task from a message queue holding a plurality of tasks, the first task specifying replicating a first object at the source site to a destination site;

issuing, from the first source worker node to the destination site, a first request for a first connection, the first request being received by a load balancer at the destination site and assigned to a first destination worker node;

establishing the first connection between the first source and destination worker nodes;

determining that the destination site does not include an object having a same name, timestamp, size, and checksum as the first object;

upon the determining, replicating the first object to the destination site by:

transmitting from the first source worker node to the first destination worker node each of one or more slice recipes referenced by a first object recipe corresponding to the first object, the transmitting comprising grouping each of the one or more slice recipes into a package until a size of the package reaches a threshold size, and upon the size of the package reaching the threshold size, transferring the package of slice recipes from the first source worker node to the first destination worker node;

receiving at the first source worker node from the first destination worker node a list identifying at least a subset of fingerprints from the one or more slice recipes, the subset of fingerprints corresponding to segments of the first object that the first destination worker node has determined are missing at the destination site;

transmitting from the first source worker node to the first destination worker node a compression region comprising compressed segments of the first object missing at the destination site; and

transmitting from the first source worker node to the first destination worker node the first object recipe, copies of the first object and the first object recipe thereby now being at both the source and destination sites;

generating, by a controller, the plurality of tasks for replicating from the source to destination site;

pushing the plurality of tasks onto the message queue, the message queue comprising a limit on a number of tasks that can be held in the message queue;

determining that the message queue is full;

upon determining that the message queue is full, pausing the generating and pushing until there is space on the message queue as source worker nodes complete their tasks and pull new tasks from the message queue;

while the first object is being replicated to the destination site by the first source and destination worker nodes, allocating a new pair of worker nodes comprising a second source worker node at the source site and a second destination worker node at the destination site;

fetching, by the second source worker node, a second task from the message queue, the second task specifying replicating a second object at the source site to the destination site;

issuing, from the first source worker node to the destination site, a second request for a second connection, the second request being received by the load balancer at the destination site and assigned to the second destination worker node;

establishing the second connection between the second source and destination worker nodes;

determining that the destination site does not include an object having a same name, timestamp, size, and checksum as the second object;

upon the determining, using the second source and destination worker nodes to replicate the second object to the destination site while the first object is being replicated to the destination site using the first source and destination worker nodes;

after completing replication of the first object to the destination site, deallocating the first source and destination worker nodes; and

after completing replication of the second object to the destination site, deallocating the second source and destination worker nodes.

15. The computer program product of claim 14 wherein the method comprises:

fetching, by a third source worker node at the source site, a third task from the message queue, the third task specifying deleting a third object from the destination site;

issuing, from the third source worker node to the destination site, a third request for a third connection, the third request being received by the load balancer and assigned to a third destination worker node;

establishing the third connection between the third source and destination worker nodes; and

after the establishing the third connection, instructing the third destination worker node to delete the third object.

16. The computer program product of claim 14 wherein the source and destination sites comprise cloud storage and the method further comprises:

providing a replication tool to be executed at a client device, the client device not offering the cloud storage;

receiving, from a user at the client, an instruction to replicate a local file on the client to the cloud storage; and

replicating, using the replication tool, the local file at the client to the cloud storage.

17. The computer program product of claim 14 wherein the establishing the first connection between the first source and destination worker nodes comprises:

transmitting an authentication key from the first source worker node to the first destination worker node to authenticate the first source worker node.

18. The computer program product of claim 14 wherein the first slice recipe is organized into a similarity group having other slice recipes, and during the replicating the first object, the first destination worker node does not deduplicate the first object using slice recipes outside the similarity group, thereby allowing duplicate segments of the first object to be replicated at the destination site when the duplicate segments are referenced by the slice recipes outside the similarity group.