Geographically-distributed file system using coordinated namespace replication over a wide area network

• US 9,495,381 B2
• Filed: 03/31/2014
• Issued: 11/15/2016
• Est. Priority Date: 01/12/2005
• Status: Active Grant

• Current Assignee: Cirata Incorporated
• Original Assignee: WANdisco, Inc.
• Inventors: Jeliazkov, Plamen Jeliazkov, Shvachko, Konstantin V, Aahlad, Yeturu, Sundar, Jagane
• Primary Examiner(s): Rostami, Mohammad S
• Application Number:
  US14/231,311
• Publication Number:
  US 20150278244A1
• Time in Patent Office: 960 Days
• US Class Current: 1/1
• CPC Class Codes: G06F 16/178 Techniques for file synchronisation in file systems
• CPC Class Codes: G06F 16/182 Distributed file systems
• 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 cluster of nodes implements a single distributed file system comprises at least first and second data centers and a coordination engine process. The first data center may comprise first DataNodes configured to store data blocks of client files, and first NameNodes configured to update a state of a namespace of the cluster. The second data center, geographically remote from and coupled to the first data center by a wide area network, may comprise second DataNodes configured to store data blocks of client files, and second NameNodes configured to update the state of the namespace. The first and second NameNodes are configured to update the state of the namespace responsive to data blocks being written to the DataNodes. The coordination engine process spans the first and second NameNodes and coordinates updates to the namespace stored such that the state thereof is maintained consistent across the first and second data centers.

Claims

1. A cluster of nodes comprising computing devices configured to implement a single geographically-distributed file system, the cluster comprising:

a first data center, comprising;

a plurality of first DataNode computing devices, each configured to store data blocks of client files;

a plurality of first local persistent storages;

a plurality of first NameNode computing devices, each configured to update a state of a namespace of the cluster and each configured to store the updated state of the namespace in a first local persistent storage of the plurality of first local persistent storages;

a second data center that is geographically remote from and coupled to the first data center by a wide area network, the second data center comprising;

a plurality of second DataNode computing devices, each configured to store data blocks of client files;

a plurality of second local persistent storages;

a plurality of second NameNode computing devices, each configured to update the state of the namespace of the cluster and each configured to store the updated state of the namespace in a second local persistent storage of the plurality of second local persistent storages;

wherein the plurality of first and second NameNode computing devices are configured to update the state of the namespace responsive to data blocks being written to the plurality of first and second DataNode computing devices; and

a coordination engine process spanning the plurality of first NameNode computing devices and the plurality of second NameNode computing devices, the coordination engine process being configured to coordinate updates to the state of the namespace stored by the plurality of first and second NameNode computing devices such that the state of the namespace is maintained consistent across the first and second data centers of the cluster,

wherein the coordination engine process is configured to receive proposals from the first and second plurality of NameNode computing devices to update the state of the namespace and to generate, in response, an ordered set of agreements that specifies an order in which the plurality of first and second NameNode computing devices are to update their respective stored state of the namespace, and wherein the plurality of first and second NameNode computing devices are configured to delay updates to the state of the namespace until the ordered set of agreements is received from the coordination engine process.

2. The cluster of claim 1, wherein each of the plurality of first NameNode computing devices is configured to update the state of the namespace while one or more others of the first NameNode computing devices in the first data center is also updating the state of the namespace.

3. The cluster of claim 1, wherein each of the plurality of second NameNode computing devices is configured to update the state of the namespace while one or more others of the second NameNode computing devices in the second data center is also updating the state of the namespace.

4. The cluster of claim 1, wherein each of the plurality of first NameNode computing devices in the first data center is configured to update the state of the namespace while any of the plurality of second NameNode computing devices in the second data center is also updating the state of the namespace.

5. The cluster of claim 1, wherein each of the plurality of second NameNode computing devices in the second data center is configured to update the state of the namespace while any of the plurality of first NameNode computing devices in the first data center is also updating the state of the namespace.

6. The cluster of claim 1, wherein each of the first DataNode computing devices is configured to communicate only with the plurality of first NameNode computing devices in the first data center and wherein each of the second DataNode computing devices is configured to communicate only with the plurality of second NameNode computing devices in the second data center.

7. The cluster of claim 1, wherein the coordination engine process is configured to maintain the state of the namespace consistent upon a failure of one or more of the first and second NameNode computing devices or a failure of one or more of the first and second DataNode computing devices.

8. The cluster of claim 1, wherein the coordination engine process is configured to maintain the state of the namespace consistent upon a failure of the first or the second data center.

9. The cluster of claim 1, wherein the single geographically-distributed file system comprises a version of the Hadoop Distributed File System (HDFS).

10. The cluster of claim 1, wherein replicas of at least some of the data blocks of a file written by a client of the first data center are stored in selected ones of the plurality of second DataNode computing devices in the second data center and wherein replicas of at least some of the data blocks of a file of a client written by the second data center are stored in selected ones of the plurality of first DataNode computing devices in the first data center.

11. The cluster of claim 1, wherein each of the plurality of first DataNode computing devices of the first data center is configured to asynchronously send selected data blocks to a selected one of the plurality of second DataNode computing devices of the second data center over the wide area network.

12. The cluster of claim 11, wherein the selected data blocks are sent from the first data center to the second data center after a predetermined number of replicas of the selected data blocks are stored on selected ones of the plurality of first DataNode computing devices in the first data center.

13. The cluster of claim 1, wherein at least some of the plurality of first NameNode computing devices are configured to generate a block report that includes a list of all data blocks stored in the plurality of first DataNode computing devices for consumption by the plurality of second NameNode computing devices wherein at least some of the plurality of second NameNode computing devices are configured to generate a block report that includes a list of all data blocks stored in the plurality of second DataNode computing devices for consumption by the plurality of first NameNode computing devices.

14. The cluster of claim 13, wherein the generated block report is written as a block report file to the file system, and wherein each of the plurality of first and second NameNode computing devices in the first and second data centers are configured to periodically read the block report file from the file system and to correspondingly update their respective stored state of the namespace.

15. The cluster of claim 1, wherein the plurality of first NameNode computing devices and the plurality of first DataNode computing devices are configured to complete writing of data blocks of a client file of the first data center before any of the data blocks of the client file are sent to the second data center over the wide area network.

16. The cluster of claim 1, wherein the plurality of first NameNode computing devices and the plurality of first DataNode computing devices are configured to cause data blocks of a client file to be replicated a first predetermined and selectable number of times within the first data center and wherein the plurality of second NameNode computing devices and the plurality of second DataNode computing devices are configured to cause the data blocks of the client file to be replicated a second predetermined and selectable number of times within the second data center.

17. The cluster of claim 16, wherein the first predetermined and selectable number of times is the same as the second predetermined and selectable number of times.

18. The cluster of claim 16, wherein the first predetermined and selectable number of times is different than the second predetermined and selectable number of times.

19. The cluster of claim 1, wherein the coordination engine process is configured to run in each of the plurality of first and second NameNode computing devices.

20. The cluster of claim 1, wherein metadata and file system data is replicated across the first and second data centers.

21. A computer-implemented method, comprising:

establishing a single distributed file system that spans, over a wide area network, a cluster comprising a first data center and a geographically remote second data center, the first data center comprising a plurality of first NameNode computing devices and a plurality of first DataNode computing devices configured to store data blocks of client files, the second data center comprising a plurality of second NameNode computing devices and a plurality of second DataNode computing devices configured to store data blocks of client files;

storing, in local persistent storage accessible to each of the plurality of first NameNode computing devices and in each of the plurality of second NameNode computing devices, a state of a namespace of the cluster;

receiving proposals from the plurality of first and second NameNode computing devices to update the state of the namespace and generating, in response, an ordered set of agreements that specifies an order in which the plurality of first and second NameNode computing devices are to update their respective stored state of the namespace;

delaying, by the plurality of first and second NameNode computing devices, making updates to the state of the namespace until the ordered set of agreements is received;

updating, in the local persistent storage, the state of the namespace stored in the plurality of first NameNode computing devices and in the plurality of second NameNode computing devices, responsive to data blocks being written to the plurality of first and second DataNode computing devices; and

coordinating updates to the state of the namespace stored in the plurality of first NameNode computing devices and stored in the plurality of second NameNode computing devices to maintain the state of the namespace consistent across the first and second data centers of the cluster.

22. The computer-implemented method of claim 21, wherein updating the state of the namespace stored in the plurality of first NameNode computing devices is carried out with each of the plurality of first NameNode computing devices being configured to update the state of the namespace while one or more others of the first NameNode computing devices in the first data center is also updating the state of the namespace.

23. The computer-implemented method of claim 21, wherein updating the state of the namespace stored in the plurality of second NameNode computing devices is carried out with each of the plurality of second NameNode computing devices being configured to update the state of the namespace while one or more others of the second NameNode computing devices in the second data center is also updating the state of the namespace.

24. The computer-implemented method of claim 21, wherein updating the state of the namespace stored in the plurality of first NameNode computing devices is carried out while updating the state of the namespace stored in the plurality of second NameNode computing devices.

25. The computer-implemented method of claim 21, further comprising enabling each of the first DataNode computing devices to communicate only with the plurality of first NameNode computing devices in the first data center and enabling each of the second DataNode computing devices to communicate only with the plurality of second NameNode computing devices in the second data center.

26. The computer-implemented method of claim 21, further comprising maintaining the state of the namespace consistent upon a failure of one or more of the plurality of first and second NameNode computing devices or a failure of one or more of the plurality of first and second DataNode computing devices.

27. The computer-implemented method of claim 21, further comprising maintaining the state of the namespace consistent upon a failure of the first or the second data center.

28. The computer-implemented method of claim 21, wherein the distributed file system comprises a version of the Hadoop Distributed File System (HDFS).

29. The computer-implemented method of claim 21, further comprising storing replicas of at least some of the data blocks of a file of a client of the first data center in selected ones of the plurality of second DataNode computing devices in the second data center and storing replicas of at least some of the data blocks of a file of a client of the second data center in selected ones of the plurality of first DataNode computing devices in the first data center.

30. The computer-implemented method of claim 21, further comprising sending, by each of the plurality of first DataNode computing devices of the first data center, selected data blocks to a selected one of the plurality of second DataNode computing devices of the second data center over the wide area network.

31. The computer-implemented method of claim 30, further comprising sending the selected data blocks from the first data center to the second data center after a predetermined number of replicas of the selected data blocks are stored on selected ones of the plurality of first DataNode computing devices in the first data center.

32. The computer-implemented method of claim 21, further comprising:

generating, by at least some of the plurality of first NameNode computing devices, a block report that includes a list of all data blocks stored in the plurality of first DataNode computing devices, for consumption by the plurality of second NameNode computing devices; and

generating, by at least some of the plurality of second NameNode computing devices, a block report that includes a list of all data blocks stored in the plurality of second DataNode computing devices, for consumption by the plurality of first NameNode computing devices.

33. The computer-implemented method of claim 32, further comprising writing the generated block report as a block report file to the file system, and periodically reading, by each of the plurality of first and second NameNode computing devices in the first and second data centers, the block report file from the file system.

34. The computer-implemented method of claim 21, further comprising the plurality of first NameNode computing devices and the plurality of first DataNode computing devices completing writing data blocks of a client file of the first data center before sending any of the data blocks of the client file to the second data center over the wide area network.

35. The computer-implemented method of claim 21, further comprising the plurality of first NameNode computing devices and the plurality of first DataNode computing devices causing data blocks of a client file to be replicated a first predetermined and selectable number of times within the first data center and the plurality of second NameNode computing devices and the plurality of second DataNode computing devices causing the data blocks of the client file to be replicated a second predetermined and selectable number of times within the second data center.

36. The computer-implemented method of claim 35, wherein the first predetermined and selectable number of times is the same as the second predetermined and selectable number of times.

37. The computer-implemented method of claim 35, wherein the first predetermined and selectable number of times is different than the second predetermined and selectable number of times.

38. The computer-implemented method of claim 21, wherein coordinating is performed in each of the plurality of first and second NameNode computing devices.

39. The computer-implemented method of claim 21, further comprising replicating metadata and file system data across the first and second data centers.