Method and apparatus for monitoring and maintaining health in a searchable data service

• US 8,554,758 B1
• Filed: 03/29/2006
• Issued: 10/08/2013
• Est. Priority Date: 12/29/2005
• Status: Active Grant

• Current Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Original Assignee: Amazon Technologies, Inc. (Amazon.com, Inc.)
• Inventors: Larson, James S., Ransil, Patrick W., Martynov, Aleksey V., Collette, James R., Chu, Robert Wai-Chi, Saha, Partha
• Primary Examiner(s): Vital, Pierre
• Assistant Examiner(s): Mitiku, Berhanu
• Application Number:
  US11/392,480
• Time in Patent Office: 2,750 Days
• Field of Search: 707/219, 709/217, 709/219, 709/223, 709/224
• US Class Current: 707/707
• CPC Class Codes: G06F 11/1446 Point-in-time backing up or restoration of persistent data
• CPC Class Codes: G06F 11/3433 for load management allocation of a server based on load conditions G06F9/505; load rebalancing G06F9/5083; redistributing the load in a network by a load balancer H04L67/1029
• CPC Class Codes: G06F 16/2228 Indexing structures
• CPC Class Codes: G06F 16/2255 Hash tables
• CPC Class Codes: G06F 16/2272 Management thereof
• CPC Class Codes: G06F 16/245 Query processing
• CPC Class Codes: G06F 16/2455 Query execution
• CPC Class Codes: G06F 16/24552 Database cache management
• CPC Class Codes: G06F 16/24554 Unary operations; Data partitioning operations
• CPC Class Codes: G06F 16/24557 Efficient disk access during query execution
• 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
• CPC Class Codes: G06F 16/278 Data partitioning, e.g. horizontal or vertical partitioning
• CPC Class Codes: G06F 16/282 Hierarchical databases, e.g. IMS, LDAP data stores or Lotus Notes
• CPC Class Codes: G06F 16/951 Indexing; Web crawling techniques
• CPC Class Codes: G06F 9/5083 Techniques for rebalancing the load in a distributed system
• CPC Class Codes: Y10S 707/99933 Query processing, i.e. searching

Abstract

Method and apparatus for monitoring and maintaining group health and membership in a searchable data service. The searchable data service may provide a searchable index to a backend data store, and an interface to build and query the searchable index, that enables client applications to search for and retrieve locators for stored entities in the backend data store. Embodiments of the searchable data service may implement a group health mechanism that may allow one or more components of the searchable data service to identify sets of nodes that the components may monitor. A group health component may be queried by the other components for automatically-refreshed health information about those nodes. The group health component may perform as a node failure detection mechanism. Node health information may be communicated among nodes at least in part using a gossip protocol.

Claims

1. A searchable data service system, comprising:

a plurality of computer devices configured to implement a searchable data service, wherein the searchable data service is configured to process query requests from client applications to locate searchable data service objects in respective searchable indexes stored on the plurality of computer devices that satisfy the query requests, wherein each searchable data service object specifies two or more attributes of a particular entity in a particular one of a plurality of independent data stores on one or more storage devices each on a network and separate from the plurality of computer devices that implement the searchable data service, wherein each searchable index stores searchable data service objects for a particular one of the plurality of independent data stores, and wherein the attributes include a unique entity identifier for locating the particular entity in the particular data store;

wherein the searchable data service comprises a node group comprising two or more nodes implemented on the plurality of computer devices, wherein each node in the node group comprises an instance of a group health component of the searchable data service, wherein the group health component on each node is configured to;

send group health messages each including collected local health information for the node to at least one other node in the node group; and

receive group health messages originating from one or more others of the nodes in the node group, wherein each group health message includes local health information for the originating node.

2. The system as recited in claim 1, wherein each attribute is expressed as a {name, value} pair.

3. The system as recited in claim 1, wherein each node in the node group is configured to compare collected local health information for the node to local health information for at least one of the one or more other nodes in the node group received in the group health messages to make local node health decisions for the node.

4. The system as recited in claim 1, wherein each node in a node group is configured to send group health messages and to receive group health messages in accordance with a gossip protocol.

5. The system as recited in claim 1, wherein the nodes in a node group are configured to propagate received group health messages to other nodes in the node group in accordance with a gossip protocol.

6. The system as recited in claim 1, wherein each node in a node group is configured to piggyback group health messages on other searchable data service messages to other nodes in the node group.

7. The system as recited in claim 1, wherein the group health component on each node is further configured to provide the local health information for other nodes received in group health messages to one or more other searchable data service components on the node.

8. The system as recited in claim 1, wherein the group health component on each node is further configured to determine that another node in the node group has gone offline if a new group health message from the other node is not received within a specified period after a last received group health message from the other node.

9. The system as recited in claim 1, wherein the group health component on each node is further configured to:

detect that a new group health message has not been received from a particular node in the node group for a specified period;

perform an anti-entropy operation with another one of the plurality of nodes to obtain updated health information for the nodes in the node group in response to said detection; and

if the anti-entropy operation does not produce updated health information for the particular node, determine that the particular node has gone offline.

10. The system as recited in claim 1, wherein the searchable data service comprises a plurality of nodes configured to organize into node groups each including a subset of the plurality of nodes, wherein each node in a node group is configured to communicate group health messages with other nodes in the node group in accordance with a gossip protocol.

11. The system as recited in claim 1, wherein the local health information includes:

one or more storage resource usage metrics for a local storage resource;

one or more processor usage metrics for a local processor resource; and

one or more network usage metrics for a network communications resource.

12. The system as recited in claim 1, wherein the searchable data service further comprises:

a plurality of storage nodes each configured to persistently store at least one partition of the searchable index;

one or more query nodes each configured to;

maintain a local query cache of responses to previous query requests;

access the local query cache on the query node to determine if a query request can be satisfied from the local query cache; and

if the query request cannot be satisfied from the local query cache, forward the query request to an appropriate one or more of the storage nodes;

one or more coordinator nodes each configured to;

receive service requests from the client application, wherein the service requests comprise storage requests and query requests;

forward each storage request to an appropriate one or more of the storage nodes; and

forward each query request to an appropriate one or more of the query nodes.

13. The system as recited in claim 12, wherein each of the one or more storage nodes is configured to:

receive a storage request from a coordinator node; and

modify a partition of the searchable index persistently stored by the storage node as specified in the storage request.

14. The system as recited in claim 12, wherein each of the one or more storage nodes is configured to:

receive a query request from a query node;

search a partition of the searchable index persistently stored by the storage node to locate a set of one or more searchable data service objects in the searchable index that satisfy the query request; and

return at least the entity identifiers from the set of one or more searchable data service objects that satisfy the query request to the client application.

15. The system as recited in claim 12, wherein the searchable data service comprises a plurality of nodes configured to organize into node groups each including a subset of the plurality of nodes, wherein each node in a node group is configured to communicate with other nodes in the node group in accordance with a gossip protocol to propagate searchable data service information within the node group;

wherein the storage nodes are configured to organize into one or more storage node groups;

wherein the query nodes are configured to organize into one or more query node groups; and

wherein the coordinator nodes are configured to organize into one or more coordinator node groups.

16. The system as recited in claim 15, wherein each query node and each coordinator node comprises an instance of a storage node locator component of the searchable data service, wherein the storage node locator component is configured to locate appropriate storage nodes to receive service requests, and wherein the query nodes and coordinator nodes are configured to self-organize into one or more storage node locator groups.

17. The system as recited in claim 12, wherein the searchable data service comprises a plurality of nodes distributed among two or more data centers, wherein each data center includes two or more of the storage nodes, one or more of the query nodes, and one or more of the coordinator nodes.

18. The system as recited in claim 1, further comprising a web services platform configured to provide a web service interface to the searchable data service, wherein the web services platform is configured to:

receive service requests from the client application in accordance with the web service interface to the searchable data service, wherein the service requests comprise query requests and storage requests; and

return searchable data service responses to the service request to the client application in accordance with the web service interface to the searchable data service.

19. The system as recited in claim 1, wherein, to send group health messages each including collected local health information for the node to at least one other node in the node group, the group health component on each node is further configured to periodically send a group health message including collected local health information for the node to at least one other node in the node group.

20. The method as recited in claim 19, further comprising:

comparing collected local health information for a node in the node group to local health information for at least one of the one or more other nodes in the node group received in group health messages;

making a local node health decision on the node in accordance with said comparing.

21. The system as recited in claim 1, wherein, to send group health messages including collected local health information for the node to at least one other node in the node group, the group health component on each node is further configured to aperiodically send a group health message including collected local health information for the node to at least one other node in the node group.

22. A method, comprising:

processing query requests received from client applications by a searchable data service implemented on a plurality of computer devices, wherein the searchable data service is configured to locate searchable data service objects in respective searchable indexes that satisfy the query requests, wherein each searchable data service object specifies two or more attributes of a particular entity in a articular one of a plurality of independent data stores on one or more storage devices on a network and separate from the plurality of computer devices that implement the searchable data service, wherein each searchable index stores searchable data service objects for a particular one of the plurality of independent data stores, and wherein the attributes include a unique entity identifier for locating the particular entity in the particular data store;

wherein the searchable data service comprises a plurality of nodes implemented on the plurality of computer devices, wherein the plurality of nodes are organized into one or more node groups each including a subset of the plurality of nodes, wherein each node in the node group comprises an instance of a group health component of the searchable data service;

on each node in a node group, the group health component on the node;

sending group health messages each including collected local health information for the node to at least one other node in the node group; and

receiving group health messages originating from one or more others of the nodes in the node group, wherein each group health message includes local health information for the originating node.

23. The method as recited in claim 22, wherein each attribute is expressed as a {name, value} pair.

24. The method as recited in claim 22, wherein the nodes in the node group are each configured to send group health messages and to receive group health messages in accordance with a gossip protocol.

25. The method as recited in claim 22, further comprising the nodes in the node group propagating received group health messages to other nodes in the node group in accordance with a gossip protocol.

26. The method as recited in claim 22, further comprising piggybacking group health messages on other searchable data service messages to be sent from a node in the node group to one or more other nodes in the node group.

27. The method as recited in claim 22, further comprising forwarding the local health information for other nodes received in group health messages from the group health component on the node to one or more other searchable data service components on the node.

28. The method as recited in claim 22, further comprising determining that another node in the node group has gone offline if a new group health message from the other node is not received within a specified period after a last received group health message from the other node.

29. The method as recited in claim 28, wherein said determining that another node in the node group has gone offline is performed by the group health component on the node.

30. The method as recited in claim 22, further comprising the group health component on a node in the node group:

detecting that a new group health message has not been received from a particular node in the node group for a specified period;

performing an anti-entropy operation with another one of the plurality of nodes to obtain updated health information for the nodes in the node group in response to said detection; and

determining that the particular node has gone offline if the anti-entropy operation does not produce updated health information for the particular node.

31. The method as recited in claim 22, wherein each node in a node group is configured to communicate with other nodes in the node group in accordance with a gossip protocol.

32. The method as recited in claim 22, wherein the local health information includes:

one or more storage resource usage metrics for a local storage resource;

one or more processor usage metrics for a local processor resource; and

one or more network usage metrics for a network communications resource.

33. The method as recited in claim 22, wherein the plurality of nodes comprises a plurality of storage nodes, one or more coordinator nodes, and one or more query nodes, and wherein the method further comprises:

persistently storing at least one partition of the searchable index on each of the plurality of storage nodes;

on each of the one or more coordinator nodes;

receiving service requests from the client application, wherein the service requests comprise storage requests and query requests;

forwarding each storage request to an appropriate one or more of the storage nodes; and

forwarding each query request to an appropriate one or more of the query nodes;

on each of the one or more query nodes;

maintaining a local query cache of responses to previous query requests;

accessing the local query cache on the query node to determine if a query request can be satisfied from the local query cache;

forwarding the query request to an appropriate one or more of the storage nodes if the query request cannot be satisfied from the local query cache.

34. The method as recited in claim 33, further comprising, on each of the one or more storage nodes:

receiving a storage request from a coordinator node; and

modifying a partition of the searchable index persistently stored by the storage node as specified in the storage request.

35. The method as recited in claim 33, further comprising, on each of the one or more storage nodes:

receiving a query request from a query node;

searching a partition of the searchable index persistently stored by the storage node to locate a set of one or more searchable data service objects in the searchable index that satisfy the query request; and

returning at least the entity identifiers from the set of one or more searchable data service objects that satisfy the query request to the client application.

36. The method as recited in claim 33, wherein the two or more node groups include:

one or more storage node groups each comprising a subset of the storage nodes;

one or more query node groups each comprising a subset of the query nodes; and

one or more coordinator node groups each comprising a subset of the coordinator nodes.

37. The method as recited in claim 36, wherein each query node and each coordinator node comprises an instance of a storage node locator component of the searchable data service configured to locate appropriate storage nodes to receive service requests, and wherein the two or more node groups include one or more storage node locator groups each including a subset of the query nodes and the coordinator nodes.

38. The method as recited in claim 33, wherein the plurality of nodes is distributed among two or more data centers, wherein each data center includes two or more of the storage nodes, one or more of the query nodes, and one or more of the coordinator nodes.

39. The method as recited in claim 22, further comprising a web services platform configured to provide a web service interface to the searchable data service, wherein the method further comprises the web services platform:

providing a web service interface to the searchable data service;

receiving service requests from the client application in accordance with the web service interface to the searchable data service, wherein the service requests comprise query requests and storage requests;

forwarding the received service requests to one or more of the plurality of nodes; and

returning searchable data service responses to the service request to the client application in accordance with the web service interface to the searchable data service.

40. The method as recited in claim 22, wherein said sending group health messages each including collected local health information for the node to at least one other node in the node group comprises periodically sending a group health message including collected local health information for the node to at least one other node in the node group.

41. The method as recited in claim 22, wherein said sending group health messages each including collected local health information for the node to at least one other node in the node group comprises aperiodically sending a group health message including collected local health information for the node to at least one other node in the node group.

42. A computer-accessible memory medium storing program instructions, wherein the program instructions are computer-executable to implement a searchable data service on a plurality of nodes, wherein the searchable data service is configured to:

process query requests received from client applications by a searchable data service implemented on a plurality of computer devices, wherein the searchable data service is configured to locate searchable data service objects in respective searchable indexes that satisfy the query requests, wherein each searchable data service object specifies two or more attributes of a particular entity in a particular one of a plurality of independent data stores on one or more storage devices on a network and separate from the plurality of computer devices that implement the searchable data service, wherein each searchable index stores searchable data service objects for a particular one of the plurality of independent data stores, and wherein the attributes include a unique entity identifier for locating the particular entity in the particular data store;

wherein the searchable data service comprises a plurality of nodes implemented on the plurality of computer devices, wherein the plurality of nodes are organized into one or more node groups each including a subset of the plurality of nodes, wherein each node in the node group comprises an instance of a group health component of the searchable data service;

wherein, on each node in a node group, the group health component of the searchable data service is configured to;

send group health messages each including collected local health information for the node to at least one other node in the node group; and

receive group health messages originating from one or more others of the nodes in the node group, wherein each group health message includes local health information for the originating node.

43. The computer-accessible memory medium as recited in claim 42, wherein each attribute is expressed as a {name, value} pair.

44. The computer-accessible memory medium as recited in claim 42, wherein each node in a node group is configured to compare collected local health information for the node to local health information for at least one of the one or more other nodes in the node group received in the group health messages to make local node health decisions for the node.

45. The computer-accessible memory medium as recited in claim 42, wherein each node in a node group is configured to send group health messages and to receive group health messages in accordance with a gossip protocol.

46. The computer-accessible memory medium as recited in claim 42, wherein the nodes in a node group are configured to propagate received group health messages to other nodes in the node group in accordance with a gossip protocol.

47. The computer-accessible memory medium as recited in claim 42, wherein each node in a node group is configured to piggyback group health messages on other searchable data service messages to other nodes in the node group.

48. The computer-accessible memory medium system as recited in claim 42, wherein the group health component on each node is further configured to provide the local health information for other nodes received in group health messages to one or more other searchable data service components on the node.

49. The computer-accessible memory medium as recited in claim 42, wherein, on each node in the node group, the group health component of the searchable data service is further configured to determine that another node in the node group has gone offline if a new group health message from the other node is not received within a specified period after a last received group health message from the other node.

50. The computer-accessible memory medium as recited in claim 42, wherein, on each node in the node group, the group health component of the searchable data service is configured to:

detect that a new group health message has not been received from a particular node in the node group for a specified period;

perform an anti-entropy operation with another one of the plurality of nodes to obtain updated health information for the nodes in the node group in response to said detection; and

if the anti-entropy operation does not produce updated health information for the particular node, determine that the particular node has gone offline.

51. The computer-accessible memory medium as recited in claim 42, wherein each node in a node group is configured to communicate with other nodes in the node group in accordance with a gossip protocol and an anti-entropy protocol.

52. The computer-accessible memory medium as recited in claim 42, wherein the local health information includes:

one or more storage resource usage metrics for a local storage resource;

one or more processor usage metrics for a local processor resource; and

one or more network usage metrics for a network communications resource.

53. The computer-accessible memory medium as recited in claim 42, wherein the program instructions are configured to implement the searchable data service on the plurality of nodes as:

a plurality of storage nodes each configured to persistently store at least one partition of the searchable index;

one or more query nodes each configured to;

maintain a local query cache of responses to previous query requests;

access the local query cache on the query node to determine if a query request can be satisfied from the local query cache; and

if the query request cannot be satisfied from the local query cache, forward the query request to an appropriate one or more of the storage nodes;

one or more coordinator nodes each configured to;

receive service requests from the client application, wherein the service requests comprise storage requests and query requests;

forward each storage request to an appropriate one or more of the storage nodes; and

forward each query request to an appropriate one or more of the query nodes.

54. The computer-accessible memory medium as recited in claim 53, wherein each of the one or more storage nodes is configured to:

receive a storage request from a coordinator node; and

modify a partition of the searchable index persistently stored by the storage node as specified in the storage request.

55. The computer-accessible memory medium as recited in claim 53, wherein each of the one or more storage nodes is configured to:

receive a query request from a query node;

search a partition of the searchable index persistently stored by the storage node to locate a set of one or more searchable data service objects in the searchable index that satisfy the query request; and

return at least the entity identifiers from the set of one or more searchable data service objects that satisfy the query request to the client application.

56. The computer-accessible memory medium as recited in claim 53, wherein the two or more node groups include:

one or more storage node groups each comprising a subset of the storage nodes;

one or more query node groups each comprising a subset of the query nodes; and

one or more coordinator node groups each comprising a subset of the coordinator nodes.

57. The computer-accessible memory medium as recited in claim 56, wherein each query node and each coordinator node comprises an instance of a storage node locator component of the searchable data service configured to locate appropriate storage nodes to receive service requests, and wherein the two or more node groups include one or more storage node locator groups each including a subset of the query nodes and the coordinator nodes.

58. The computer-accessible memory medium as recited in claim 53, wherein the plurality of nodes is distributed among two or more data centers, wherein each data center includes two or more of the storage nodes, one or more of the query nodes, and one or more of the coordinator nodes.

59. The computer-accessible memory medium as recited in claim 42, wherein the program instructions are further configured to implement a web service interface to the searchable data service, wherein the web service interface is configured to:

receive service requests from the client application in accordance with the web service interface, wherein the service requests comprise query requests and storage requests;

forward the received service requests to one or more of the plurality of nodes; and

return searchable data service responses to the service request to the client application in accordance with the web service interface.

60. The computer-accessible memory medium as recited in claim 42, wherein, to send group health messages each including collected local health information for the node to at least one other node in the node group, the group health component on each node is further configured to periodically send a group health message including collected local health information for the node to at least one other node in the node group.

61. The computer-accessible memory medium as recited in claim 42, wherein, to send group health messages including collected local health information for the node to at least one other node in the node group, the group health component on each node is further configured to aperiodically send a group health message including collected local health information for the node to at least one other node in the node group.

62. A searchable data service node, comprising:

one or more processors;

a network interface configured to communicatively couple to a network; and

a memory comprising program instructions, wherein the program instructions are executable by the one or more processors to;

persistently store a partition of a particular one of a plurality of searchable indexes maintained by a searchable data service implemented on a plurality of computer devices, wherein the searchable data service node is implemented on one of the computer devices, wherein the particular searchable index corresponds to a particular one of a plurality of independent data stores on one or more storage devices on the network and separate from the plurality of computer devices that implement the searchable data service, and wherein the partition is configured to store a set of searchable data service objects stored in the particular searchable index;

process query requests received from client applications via the network to locate searchable data service objects that satisfy the query requests in the particular searchable index, wherein each searchable data service object specifies two or more attributes of a particular entity in the data store corresponding to the particular searchable index, and wherein the attributes include a unique entity identifier for locating the particular entity in the corresponding data store;

send group health messages each including collected local health information for the searchable data service node to another searchable data service node implemented on another one of the computer devices via the network; and

receive group health messages via the network and originating from the other searchable data service node, wherein each received group health message includes local health information for the other searchable data service node.

63. The searchable data service node as recited in claim 62, wherein each attribute is expressed as a {name, value} pair.

64. The searchable data service node as recited in claim 62, wherein the program instructions are further executable by the one or more processors to compare collected local health information for the searchable data service node to local health information for the other searchable data service node received in the group health messages to make local node health decisions for the searchable data service node.

65. The searchable data service node as recited in claim 62, wherein the program instructions are further executable by the one or more processors to send group health messages and to receive group health messages via the network in accordance with a gossip protocol.

66. The searchable data service node as recited in claim 62, wherein, to periodically or aperiodically send a group health message including collected local health information for the searchable data service node to another searchable data service node via the network, the program instructions are further executable by the one or more processors to piggyback the group health message including collected local health information for the searchable data service node on another searchable data service message to the other searchable data service node.

67. The searchable data service node as recited in claim 62, wherein the program instructions are further executable by the one or more processors to determine that the other searchable data service node has gone offline if a new group health message from the other searchable data service node is not received within a specified period after a last received group health message from the other searchable data service node.

68. The searchable data service node as recited in claim 62, wherein, send group health messages each including collected local health information for the searchable data service node to another searchable data service node via the network, the program instructions are further executable by the one or more processors to periodically send a group health message including collected local health information for the node to at least one other node in the node group.

69. The searchable data service node as recited in claim 62, wherein, send group health messages each including collected local health information for the searchable data service node to another searchable data service node via the network, the program instructions are further executable by the one or more processors to aperiodically send a group health message including collected local health information for the node to at least one other node in the node group.