Executing prioritized replication requests for objects in a distributed storage system

• US 8,285,686 B2
• Filed: 02/09/2011
• Issued: 10/09/2012
• Est. Priority Date: 02/09/2010
• Status: Active Grant

• Current Assignee: Google LLC (Alphabet Inc.)
• Original Assignee: Google Inc. (Alphabet Inc.)
• Inventors: Kesselman, Alexander
• Primary Examiner(s): Ehichioya, Fred I
• Assistant Examiner(s): Mobin, Hasanul
• Application Number:
  US13/024,243
• Publication Number:
  US 20110196835A1
• Time in Patent Office: 608 Days
• Field of Search: None
• US Class Current: 707/679
• CPC Class Codes: G06F 16/27 Replication, distribution or synchronisation of data between databases or within a distributed database system; Distributed database system architectures therefor

Abstract

A system and method for executing replication requests for objects in a distributed storage system is provided. A replication queue is identified from a plurality of replication queues corresponding to a replication key. The replication key includes information related to at least a source storage device in a distributed storage system at which objects are located and a destination storage device in the distributed storage system to which the objects are to be replicated. A distributed database is scanned using an identifier of the replication queue to produce a list of replication requests corresponding to the replication queue. The records of the distributed database are distributed across a plurality of nodes of the distributed database. The replication requests in the list of replication requests are executed in priority order. Replication requests are deleted from the distributed database only when the replication requests are complete.

Claims

1. A computer-implemented method for executing replication requests for objects in a distributed storage system, comprising:

at a computer system including one or more processors and memory storing one or more programs, for execution by the one or more processors;

identifying a replication queue from a plurality of replication queues corresponding to a replication key, wherein the replication key includes information related to at least a source storage device in a distributed storage system at which objects are located and a destination storage device in the distributed storage system to which the objects are to be replicated;

scanning a distributed database using an identifier of the replication queue to produce a list of replication requests corresponding to the replication queue, wherein the list of replication requests is sorted by globally-determined priorities of the replication requests that are included in row keys corresponding to records of the distributed database for the replication requests in the list of replication requests, wherein a respective replication request includes a globally-determined profit value corresponding to the respective replication request, wherein the globally-determined profit value is based on a metric corresponding to a benefit of performing the respective replication request minus a metric corresponding to a cost of performing the respective replication request, wherein the records of the distributed database are distributed across a plurality of nodes of the distributed database, and wherein a location assignment daemon is configured to generate replication requests globally across instances of the distributed storage system based at least in part on a current state of the distributed storage system and replication policies for objects in the distributed storage system;

executing replication requests in the list of replication requests in priority order; and

deleting replication requests from the distributed database only when the replication requests are complete.

2. The method of claim 1, wherein a respective replication request further includes:

a replication key corresponding to a respective object;

a list of chunks of the respective object; and

a replication identifier corresponding to the respective replication request.

3. The method of claim 2, wherein a respective row key corresponding to a respective record of the distributed database that includes the respective replication request is generated based on the replication key of the respective replication request, the profit value of the respective replication request, and the replication identifier of the respective replication request.

4. The method of claim 2, wherein a respective row key corresponding to a respective record of the distributed database that includes the respective replication request is generated based on a hashed version of the replication key of the respective replication request, the profit value of the respective replication request, and the replication identifier of the respective replication request.

5. The method of claim 2, wherein the replication key includes:

a user identifier;

a quality of service metric;

an identifier for a source storage device in the distributed storage system; and

an identifier for a destination storage device in the distributed storage system.

6. The method of claim 5, wherein the source storage device and the destination storage device are located in separate and distinct instances of the distributed storage system.

7. The method of claim 2, wherein replication requests in the replication queue have the same replication key.

8. The method of claim 2, wherein a globally-determined priority of the respective replication request is based on the globally-determined profit value for the respective replication request.

9. The method of claim 1, wherein the identifier for the replication queue is a replication key.

10. The method of claim 1, wherein executing the replication requests corresponding to the replication queue in priority order includes transmitting the replication requests to a task server for execution, wherein the task server is separate and distinct from the computer system.

11. The method of claim 1, wherein the identifying, the scanning, the executing, and the deleting are performed for each replication queue in the plurality of replication queues.

12. The method of claim 1, wherein the distributed storage system includes a plurality of instances of the distributed storage system.

13. The method of claim 1, wherein a respective replication request is executable to add a replica of the respective object at an instance of the distributed storage system.

14. A system for executing replication requests for objects in a distributed storage system, comprising:

one or more processors;

memory; and

one or more programs stored in the memory, the one or more programs comprising instructions to;

identify a replication queue from a plurality of replication queues corresponding to a replication key, wherein the replication key includes information related to at least a source storage device in a distributed storage system at which objects are located and a destination storage device in the distributed storage system to which the objects are to be replicated;

scan a distributed database using an identifier of the replication queue to produce a list of replication requests corresponding to the replication queue, wherein the list of replication requests is sorted by globally-determined priorities of the replication requests that are included in row keys corresponding to records of the distributed database for the replication requests in the list of replication requests, wherein a respective replication request includes a globally-determined profit value corresponding to the respective replication request, wherein the globally-determined profit value is based on a metric corresponding to a benefit of performing the respective replication request minus a metric corresponding to a cost of performing the respective replication request, wherein the records of the distributed database are distributed across a plurality of nodes of the distributed database, and wherein a location assignment daemon is configured to generate replication requests globally across instances of the distributed storage system based at least in part on a current state of the distributed storage system and replication policies for objects in the distributed storage system;

execute replication requests in the list of replication requests in priority order; and

delete replication requests from the distributed database only when the replication requests are complete.

15. The system of claim 14, wherein a respective replication request further includes:

a replication key corresponding to a respective object;

a list of chunks of the respective object; and

a replication identifier corresponding to the respective replication request.

16. The system of claim 15, wherein a respective row key corresponding to a respective record of the distributed database that includes the respective replication request is generated based on the replication key of the respective replication request, the profit value of the respective replication request, and the replication identifier of the respective replication request.

17. The system of claim 15, wherein a respective row key corresponding to a respective record of the distributed database that includes the respective replication request is generated based on a hashed version of the replication key of the respective replication request, the profit value of the respective replication request, and the replication identifier of the respective replication request.

18. The system of claim 15, wherein the replication key includes:

a user identifier;

a quality of service metric;

an identifier for a source storage device in the distributed storage system; and

an identifier for a destination storage device in the distributed storage system.

19. The system of claim 15, wherein a globally-determined priority of the respective replication request is based on the globally-determined profit value for the respective replication request.

20. The system of claim 14, wherein the instructions to execute the replication requests corresponding to the replication queue in priority order include instructions to transmit the replication requests to a task server for execution, wherein the task server is separate and distinct from the computer system.

21. The system of claim 14, wherein a respective replication request is executable to add a replica of the respective object at an instance of the distributed storage system.

22. A non-transitory computer readable storage medium storing one or more programs configured for execution by a computer, the one or more programs comprising instructions to:

identify a replication queue from a plurality of replication queues corresponding to a replication key, wherein the replication key includes information related to at least a source storage device in a distributed storage system at which objects are located and a destination storage device in the distributed storage system to which the objects are to be replicated;

scan a distributed database using an identifier of the replication queue to produce a list of replication requests corresponding to the replication queue, wherein the list of replication requests is sorted by globally-determined priorities of the replication requests that are included in row keys corresponding to records of the distributed database for the replication requests in the list of replication requests, wherein a respective replication request includes a globally-determined profit value corresponding to the respective replication request, wherein the globally-determined profit value is based on a metric corresponding to a benefit of performing the respective replication request minus a metric corresponding to a cost of performing the respective replication request, wherein the records of the distributed database are distributed across a plurality of nodes of the distributed database, and wherein a location assignment daemon is configured to generate replication requests globally across instances of the distributed storage system based at least in part on a current state of the distributed storage system and replication policies for objects in the distributed storage system;

execute replication requests in the list of replication requests in priority order; and

delete replication requests from the distributed database only when the replication requests are complete.

23. The computer readable storage medium of claim 22, wherein a respective replication request further includes:

a replication key corresponding to a respective object;

a list of chunks of the respective object; and

a replication identifier corresponding to the respective replication request.

24. The computer readable storage medium of claim 23, wherein a respective row key corresponding to a respective record of the distributed database that includes the respective replication request is generated based on the replication key of the respective replication request, the profit value of the respective replication request, and the replication identifier of the respective replication request.

25. The computer readable storage medium of claim 23, wherein a respective row key corresponding to a respective record of the distributed database that includes the respective replication request is generated based on a hashed version of the replication key of the respective replication request, the profit value of the respective replication request, and the replication identifier of the respective replication request.

26. The computer readable storage medium of claim 23, wherein the replication key includes:

a user identifier;

a quality of service metric;

an identifier for a source storage device in the distributed storage system; and

an identifier for a destination storage device in the distributed storage system.

27. The computer readable storage medium of claim 23, wherein a globally-determined priority of the respective replication request is based on the globally-determined profit value for the respective replication request.

28. The computer readable storage medium of claim 22, wherein the instructions to execute the replication requests corresponding to the replication queue in priority order include instructions to transmit the replication requests to a task server for execution, wherein the task server is separate and distinct from the computer system.

29. The computer readable storage medium of claim 22, wherein a respective replication request is executable to add a replica of the respective object at an instance of the distributed storage system.