Collection-based object replication

Collection-based object replication for a system that includes a client computing device (client) connected to a server and multiple data storage nodes. In certain cases, a data storage node generates a replica of multiple replicas of a collection. The collection is a unit of data placement, access, replication, and repair. Other data storage nodes are also configured with a respective replica of the multiple replicas. The data storage node verifies whether an object received directly from the client for storage in the collection has been fully replicated by the other data storage nodes in respective replicas.

BACKGROUND

A storage replication service is a managed service in which stored or archived data is duplicated among a number of data storage nodes. This provides a measure of redundancy that can be invaluable if a data storage node fails. To obtain storage replication services, including data object storage and replication, data access, and data removal services, a client computing device typically requests a central server to directly perform the service(s). One reason for this is because the server generally maintains a central data structure to index locations of each stored data object in the system to respective storage devices. As data objects are stored and replicated in the system, the server indexes the storage locations of the corresponding replicas so that they can be subsequently located, managed, removed, and repaired responsive to storage node failures. For example, responsive to losing copies of data objects on a failed storage node, the server refers to the index to verify that the lost data objects are fully replicated in the system. However, a centralized index represents a scalability and performance bottleneck in systems containing a very large number of data objects.

SUMMARY

Collection-based object replication is described in a system that includes a client computing device (client) connected to a server and to multiple data storage nodes. In one aspect, a data storage node contains a replica out of multiple replicas of a collection. The collection is a unit of data placement, access, replication, and repair. Other data storage nodes are also configured with a respective replica of the multiple replicas. The data storage node verifies whether an object received directly from the client for storage in the collection has been fully replicated by the other data storage nodes in respective replicas.

DETAILED DESCRIPTION

Overview

Collection-based object replication is described in reference to the systems and methods ofFIGS. 1-5. The systems and methods for collection-based object replication address the scalability and performance bottleneck limitations common in conventional systems that use a centralized index to index a very large number of data objects. This is accomplished, at least in part, by implementing decentralized management of data object replicas on the data storage nodes themselves. To this end, the systems and methods group objects together by application to form collections of objects. A collection is a unit of data placement, replication, and data repair. The systems and methods allow applications to create collections, store objects into collections, retrieve objects, delete objects, and delete collections.

To provide system reliability, the systems and methods replicate individual objects within a collection by storing multiple replicas of the collection across different data storage nodes. Object replication operations include data storage node-based determinations of whether objects are fully replicated across all collection locations. Additionally, responsive to data storage node-based determinations that a data object may not have been successfully replicated on a different target data storage node, a data storage node will attempt to ensure that the data object is successfully replicated on the different data storage node—and thus fully replicated in the system. Additionally, when an object is to be deleted from the system, data storage nodes work together to ensure that all copies of the object are removed from corresponding collection replicas, regardless of whether data storage nodes and/or communications fail during object removal operations.

These and other aspects of collection-based object replication are now described.

An Exemplary System

Although not required, collection-based object replication is described in the general context of computer-executable instructions (program modules) being executed by computing devices such as a general-purpose computer or a mobile handheld device. Program modules generally include routines, programs, objects, components, data structures, etc., that perform particular tasks or implement particular abstract data types. While collection-based object replication is described in the foregoing context, acts and operations described hereinafter may also be implemented in hardware.

FIG. 1shows an exemplary system100for collection-based replication in a storage area network, according to one embodiment. System100groups multiple data objects together by application to form respective collections of objects. A collection is a unit of data placement, access, replication, and repair. To these ends, system100includes client computing device(s)102coupled across network104to collection information server (CIS)106, and data storage nodes (“data nodes” and/or “nodes”)108-1through108-N. Network104may include any combination of a local area network (LAN) and a general wide area network (WAN) communication environments, such as those which are commonplace in offices, enterprise-wide computer networks, intranets, and the Internet.

Each of client computing device(s)102, CIS106, and data nodes108includes a respective processor coupled to a system memory. The respective processor is configured to fetch and execute computer-program instructions stored in the system memory to perform respective aspects of collection-based object replication associated with the particular component. For example, client computing device (“client”)102includes processor110coupled to system memory112. System memory112includes program modules114and program data116. In this implementation, program modules114include, for example, collection-based module118, one or more data I/O applications120, and other program modules122such as an operating system (OS), etc.

Collection-based module118provides data I/O application(s)120with collection-based object replication facilities for collection-based data object storage, replication, retrieval, removal, and repair facilities. To provide these facilities to data I/O application(s)120, collection-based module118exposes application program interface (API)124. An application120invokes respective interfaces of API124to request CIS106, and more particularly collection-based data management module126, to create collections128for subsequent data object storage, replication, retrieval, removal, and automated data object repair responsive to data node108failure. For purposes of exemplary illustration, such data objects are shown as a respective portion of “other program data”130.

Once a collection has been created, the application120uses API124to receive information (i.e., received collection information132) specific to a particular collection128from management module126. In view of this received information, the application120directly communicates with at least a subset of data nodes108to store data objects into collections128, retrieve stored data objects, remove data objects from collections128, and delete collections128. Exemplary APIs124for an application120to create a collection128, to request and receive collection information134specific to a particular collection128, store data objects into a collection128, remove data objects from collections128, and delete collections128are respectively described below in the sections titled “Collection Creation”, “Data Object Storage/Check-In”, “Data Object Checkout and Removal”, and “Collection Deletion”.

Management module126maintains information indicating which data nodes108(collection locations) store replicas of collections128, and other information. This information is shown as centrally managed collection information134. In this implementation, for each collection128, centrally managed collection information134includes for example:a collection identifier (cid).k, replication degree, the desired number of replicas for this collection.t, the minimum number of successfully stored object replicas when a check-in operation is successful.Lc, a set of current locations for collection replicas (i.e., data node108identifiers).Lc′ a set of future locations; in steady state, Lc′ is the same as Lc.an execution_plan: when Lc′ differs from Lc, the execution plan includes steps to be taken by management module126to move collection locations from Lcto Lc′. A step, for example, describes a source data node (“source”)108and a destination data node (“destination”)108of data movement during data repair and load balancing operations. For example, Lc={A, B, C} and Lc′={A, B, D}, and execution_plan={(move C to D)}, which denotes a load balancing operation moving the collection replica in location C to a new location D (A, B, C, and D represent respective ones of data storage nodes108).instance number, a current instance number for the collection128. Every change to Lcor Lc′ causes management module126to increment the instance number.recycle, the set of locations for collection replicas (data nodes108) that are out-dated, and should be deleted from the set of current locations.

Using centrally managed collection information134, management module126determines whether the number of replicas of a collection128is the same as an indicated replication degree A, migrates collections128responsive to permanent failure of data node(s)108, and performs load balancing operations. These operations are respectively described in greater detail in the following sections titled “Exemplary Collection Repair for Permanent Data Node Failures”, “Transient Data Node Failures”, and “Load Balancing”.

Collection Creation

To create a collection128for subsequent storage of data objects, an application120calls create collection interface of API124. In one implementation, a create collection interface of API124for example is:

Status=CreateCollection(cid, k, t), whereincid represents the unique id (e.g., generated by client102) of a collection128k represents a replication degree of the collection128. In one implementation, all collections128have a same replication degree k. In another implementation, replication degree k may differ from one collection128to another collection128. In one implementation, API124includes an interface for a client102to change k and/or t over time.t represents a minimum replication degree (1≦t≦k) used for a successful completion of a data object storage/check-in operation to this collection128. In one implementation, if a check-in is successful, the data object is guaranteed to be stored on at least t data nodes108. It is not required that a check-in operation be always successful when there are t data nodes108storing replicas of a collection128.Return: success or failure.

In this implementation, for example, CreateCollection(cid, k, t) is executed by collection-based module118as follows:Collection-based module118sends a create collection request (including cid, k, and t) to management module126;Responsive to receiving the create collection request, management module126checks centrally managed collection information134to determine if cid already exists or is under creation. If so, failure status is returned. If not, management module126:selects k data nodes108for the locations of a new collection128;initializes a respective portion of centrally managed collection information134for the collection128. For example, the cid in data structure134is the cid passed in; Lcis the set of k data nodes108with enough free space; Lc′=Lc; execution_plan is empty; instance number=1; and recycle (i.e., delete) is empty. CIS106informs the k data nodes about this new collection128; sending the corresponding collection information.The k data nodes128store the respective portion of new collection information locally in the local information136and send respective acknowledgments back to management module126. If management module126does not receive acknowledgments from all k data nodes, then the instance number is incremented and the immediately previous step and this step is retried for a different set of data nodes108until k available nodes are selected.Return: success.

Before an application120stores data objects into a collection128, removes data objects from the collection128, and/or deletes the collection128, the application120determines data nodes108(collection locations) that maintain replicas of the collection. To this end, the application120calls the get collection interface of API124. In one implementation, the get location interface is implemented, for example, as(Status, k, t, Lc, Lc′, instance number) GetCollectionLocation(cid), whereincid represents the id of the collection128of interest;k is the replication degree of the collection128;t is the minimum number of successfully stored object replicas when a check-in operation is successful;Lcand Lc′ represent two sets of data node108locations; andinstance number represents the instance number of the collection128.
Lcis the set of current locations, and Lc′ is the set of future locations (|Lc′|=k, where k is the replication degree of the collection128). The two sets differ in the following situations: (1) when some collection location is lost and data is being repaired to a new data node108; or (2) for load balancing purpose the collection locations are being moved.

Responsive to a get collection location call (e.g., GetCollectionLocation(cid)), collection-based model118sends a request to management module126to retrieve a list of data nodes108that store the collection128. Responsive to receiving the request, management module126checks if the identified collection has been created successfully by the create collection interface portion of API124. If not, management module126returns an error code indicating the collection128has not been created successfully. Otherwise, management module126extracts the requested information and the instance number from centrally managed collection information134, and returns the information to collection-based module118. For purposes of exemplary illustration, the information returned by the get collection interface of API124(e.g., Status, k, t, Lc, Lc′, instance number, and/or so on) is shown on a client102as “received collection information”132.

Using at least a subset of received collection information132, an application120uses API124to store data objects into a collection128, remove data objects from the collection128, and/or delete the collection128. In this implementation, if returned Lcspecifies no data nodes108, the identified collection128is lost due to permanent failures of corresponding nodes108. The specified collection128is in a normal or stable state if Lc=Lc′; the collection128is in a transitional state otherwise. For example, during client102data object input operations to a collection128, a collection transitional state indicates that the data object has been successfully stored by a data node108into the collection128, and the data object may or may not have been successfully (or yet) stored by other identified data node(s)108. In this same scenario, a collection128in stable state indicates that a data object is successfully stored locally at a data node108and the object has also been successfully stored to other designated collection locations. Each data node108maintains local information136for each collection128(including collection state) stored on the data node108.

Data Object Storage/Check-In

An application120uses API124to store (check-in) data objects into a collection128. In one implementation, for example, a check-in interface is:Status Check-in (object, cid, oid, wherein:object represents content of the data object to be replicated in a collection128across at least a subset of data nodes108;oid represents a unique id of the object within a collection (oid, for example, can be assigned by the application120, etc.);cid represents a unique id of the collection128to which the object belongs; andReturn: success or failure.

Responsive to invocation of Check-in (object, cid, old), collection-based module118sends a list of the collection locations and an instance number associated with the collection128to the first data node108in the collection locations, and streams the object to the first data node108identified in the collection location list. In this implementation, it is assumed that the collection list is equivalent to Lcfor the collection128. In another implementation, the collection list is equivalent to Lc′ to anticipate new locations.

Responsive to receiving this information, the data node108checks if the instance number is the same as what is stored locally for the collection128. If not, the data node108returns an error code to client102. If the local instance number is lower, the data node108initiates a collection refresh operation (please see the section below titled “Collection Refresh”) to synchronize local information136associated with the collection128with centrally managed collection information134, and obtains the matching instance of the collection128. When the instance number matches, the first data node108stores the object locally. The data node108stores at least a subset of the information in a corresponding portion of local information136.

In this object check-in scenario, and before an object is successfully stored on a data node108(e.g., successful in view of a communicated checksum of the object, and/or other criteria), local information136associated with the object (i.e., the object's corresponding collection128) is in a temporary state. Temporary state is logically equivalent to no permanent object record. That is, if data node108failure occurs while the object state is still temporary, the object can be deleted by the data node108and the space re-collected. If an object is successfully stored on a data node108, and if the data node108is not the last data node108in the collection list, the data node108sets the local information136associated with the object to a partial state, and communicates the following to the next data node108in the collection location list: the object, the list, and the instance number. The partial state of the object on a data node means that the object has been successfully stored locally but it is not yet sure if the object replicas have also been stored successfully in other designated collection locations. These check-in operations are repeated by each data node108in the ordered collection location list until each data node108has successfully stored the object, or incurred failure. Indication of such incurred failure is propagated by the failing data node108or its immediately preceding data node108in reverse order down the list of collection locations to the first data node108in the list, which in-turn, notifies the client102.

After the last data node108in the ordered collection list has successfully stored the object, the last data node108sets the local information136associated with the object to a complete state. The complete state of the object on a data node means that it is sure that the object has also been replicated to all designated collection locations. The data node only updates the object state to complete when the corresponding collection is in the stable state. If the collection is in the transitional state, the data node keeps the object in the partial state. Then, the last data node108sends an acknowledgement138in reverse order direction to the immediately preceding data node108in the chain. Responsive to receiving such an acknowledgement138, the data node108updates (changes) local information136pertaining to the object from partial to complete, again only when the collection is in the stable state. The data node108then sends an acknowledgement138to any sequentially previous data node108in the chain so that the sequentially previous data node108may similarly update object status associated with the data object from partial to complete when the collection is in the stable state.

When the collection is in the stable state, this process is repeated in reverse chain direction until each of the data nodes108in the chain has updated the locally maintained status of the data object from partial to complete. When the first data node108in the list receives such an acknowledgement, the first data node108returns a success status to the requesting client102.

When the collection is in the transitional state, the current collection locations Lcmay not have k nodes. If Lchas at least t nodes, where t is the minimum number of successfully stored object replicas specified by the create collection API124, then when the first data node108in the list receives an acknowledgement from the second node in the list, the first data node108still returns a success status to the requesting client102.

Object state on a data node is only changed to complete when the collection state is in the stable state, i.e., Lc=Lc′. If the collection state is transitional (Lc≠Lc′), then the object state on a data node is set to partial.

In one implementation, if during object check-in operations a data node108determines that a locally maintained status associated with the object has remained partial for a configurable period of time, and that an acknowledgement to change the status to stable has not been received from a data node108subsequent in a chain of nodes108, the data node100actively probes (e.g., pings) other data node(s)108downstream in the chain to make sure that the other data node(s)108have a copy of the data object being stored. If so, the other data node(s)108can turn the state associated with the local copy of the object to complete and send a corresponding acknowledgement138, as indicated above. Otherwise, the data node108provides the object to at least the next data node108in the chain. An exemplary implementation of these operations is described below in the section titled “Partial Object Repair”.

In view of the above, it is clear that although management module126creates and maintains a record of data nodes108that host respective collections128, management module126does not need to: (a) index individual data objects grouped within a collection128to identify which collections store which objects; or (b) maintain indications of whether objects in a collection128have been fully replicated to collection locations. The first aspect (a) is determined according to the particular implementation of the application120that checks data objects into a collection128. The second aspect (b) is determined by respective ones of the data nodes108that store replicas of the collection128.

Partial Object Repair

An object replica stored on a data node108being in the partial state indicates that the data node108storing the collection128does not know if the object stored in the collection128has also successfully stored at other collection locations. To ensure data reliability, a data node108is responsible for turning all objects stored locally with the partial state into the complete state. To this end, the data node108executes, for example, the following partial object repair operations that are carried out for a particular object that is in partial state, and this partial object repair operation is initiated only when the data node sees that the collection is in the stable state (Lc=Lc′):1. The data node108(i.e., the originator) uses its current collection location information (a respective portion of local information136) to contact each other data node108in Lcto request that the other data node check if it has a corresponding copy of the data object.2. When a data node108receives such a request, the data node checks whether the locally maintained instance number for the collection128matches the received instance number. If the locally maintained number is lower than the received number, the data node108initiates a collection refresh operation (please see the section below titled “Collection Refresh”). If the locally maintained number is higher than the received number, the data node returns an error code to the originating data node108(causing the originating node to initiate a collection refresh operation). If the instance number matches, then the data node108checks if a replica of the data object exists locally and returns the finding to the originating data node108.3. When the originator receives a status from the other data node108, if the result is no object on the other data node108, then the originator streams the object to the other data node. When the receiving data node has completely stored the object locally into a replica of the collection128, the data object is marked as partial, and a success acknowledgment is communicated to the originator.4. When the originator learns that all other data nodes108in Lchave the object stored, then it turns its local object replica to the complete state. If the originator cannot successfully contact all other data nodes in Lcin a configurable amount of time, it retries steps 1 to 3 periodically until the repair process is successful. During the retry, it is possible that the collection location Lcis changed. In this case, the originator will contact the new set of data nodes in Lc.

In one implementation, partial object repair is implemented by a chained protocol analogous to the described data object check-in protocol, but the chain of communication is initiated individually by each data node108that has an object in partial state. Also, for all partial objects in one collection128, the protocol can be batch-executed. The partial object repair operations use local collection information (local information136) on data nodes108. If a collection128is experiencing changes (e.g., due to collection repair or load balancing operations), partial object repair operations can be aborted to yield to the collection level changes. Partial object repair can be re-initiated after collection level changes are done and the collection is back to the stable state, to ensure correctness of the transitional and stable state semantics of collections128that encapsulate objects.

Data Object Checkout and Removal

After an application120has retrieved collection information132indicating locations of data nodes108storing a particular collection128, the application can use a respective interface of API124to check-out (retrieve) data object(s) from the particular collection128. For example, in one implementation, a data object check-out interface is:(status, object)=Checkout(cid, oid), whereincid represents the id of the collection128;oid represents the id of the object in the collection;status represents the status operation, either successful or failure; andobject: the content of the object if a successful status is returned.

Responsive to such a data object check-out call, collection-based module118contacts the first data node108in the associated ordered collection locations list to retrieve the object. In one implementation, if a data node108is not accessible, or the request result is not returned during a configurable period of time, collection-based module118will automatically try to obtain the object from a next data node108in the list. Responsive to a data node108receiving a check-out request, the data node108checks local information136to determine if the instance number of the collection128comprising the object is correct. If the local instance number is lower, the data node108refreshes the associated collection128(please see the section titled “Collection Refresh”). If instance number matches, the data node108checks if object exists in the collection. If not, an error code (“object not found”) is returned to the client102. Otherwise, the data node108checks if the object has been deleted (please refer to the “Delete” protocol below). If so, the data node108returns an error code (e.g., “object has been deleted”). Otherwise, the data node108streams the object to the requester and returns a success code.

After an application120has retrieved collection information132indicating locations of data nodes108storing a particular collection128, the application can use a respective interface of API124to delete data object(s) from the particular collection128. For example, in one implementation, a data object delete interface is:status=Delete(cid, oid), whereincid represents the id of the collection128;oid represents the id of the object to be removed; andstatus: success or failure status.

Responsive to such a data object delete interface invocation, collection-based module118contacts a first data node108in the associated and ordered collection locations (list) to remove the object. For each object to be deleted from a respective data node108, the data node108creates a tombstone object (shown as a respective portion of “other data”140on data node108-1). The tombstone object identifies the object and includes a same partial or complete state as maintained in local information136pertaining to the object. The delete object protocol flow is analogous to the described object check-in object flow, with the exception that the target operation is to: remove the object upon which a last node108turns status of the tombstone object to complete, and communicate a removal acknowledgement138for migration in reverse chain order of the ordered collection locations. Responsive to a data node receiving the removal acknowledgement, the data node turns the status of the local tombstone object to complete. When a tombstone object is complete on a collection location, the object and its corresponding tombstone object are garbage-collected at that location.

To ensure that all copies of an object marked for deletion are removed, in one implementation a collection location data node108waits until the associated tombstone object and the object for deletion are both in complete state. At this point, the data node108proceeds with local garbage collection. This may result in data copying to turn the object complete even if the object has already been tombstoned. In another implementation, when the tombstone object turns complete on a data node108, the data node108contacts other collection locations to make sure all tombstone object replicas are complete, and then if the data node108has a copy of the object, the data node108garbage collects both the tombstone and the object. If the local node only has the tombstone but not the object, then the local node waits for all other collection locations with the object to complete garbage collection. This ensures that after garbage collection, the object cannot resurface, for example, as a result of the described partial object recovery process.

Collection Refresh

A collection128is synchronized (refreshed) responsive to a request from a data node108, or automatically (e.g., at periodic intervals) by management module126. For example, collection refresh operations could be triggered when a data node108determines local information136(local collection information) is out of date, recovers from a crash, or is rebooted. In one implementation, management module126triggers a collection refresh periodically to guarantee that the collection states on all data nodes eventually converge to a same state.

In one implementation, management module126triggers a collection refresh operation whenever it changes the centrally managed collection information134. To these ends, and for each collection128represented in centrally managed collection information134:management module126checks if a data node108still belongs to Lcor Lc′. If not, the data node108is outdated and is marked for garbage collection. If the data node108is not out-dated, but a locally maintained collection128instance number is lower than it should be, then management module126sends the current state of the collection134to all data nodes in Lcand Lc′ to synch up.all data nodes108update their local collection information136when receiving a synch-up message from management module126. After update, data nodes108send acknowledgment back to management module126. Such an acknowledgement is shown as a respective portion of “other data”142of CIS106.

Collection Deletion

An application120uses a delete collection portion of API124(e.g., DeleteCollection(cid)) to delete a collection128. Responsive to this invocation, collection-based module118communicates the request to management module126, which in turn marks the collection128in centrally managed collection information134as deleted (the final state of the collection). Management module126broadcasts the cid and the final state to all collection locations. When a data node108receives the final state of the collection128, the data node108stops any ongoing operation for the collection128, rejects any future operation on the collection128, and marks the collection128in local information136as deleted (e.g., tombstones the collection128). The data node108then sends an acknowledgment to management module126indicated that the collection has been tombstoned. When management module126receives acknowledgments from all data nodes108listed in the collection locations, management module126also tombstones the collection128. CIS106initiates a garbage collection of all such tombstoned collections128.

Exemplary Collection Repair for Permanent Data Node Failures

When a data node108(data node x) permanently fails, or a disk scan of the data on x discovers data corruption, management module126decommissions x and repairs a collection c128a copy of which was stored on x to other data nodes128. In this implementation, for each collection c, collection location based criteria determine repair operations. Such repair criteria include, for example, the following scenarios.

Case 1. xεLc∩Lc′. In this scenario, when an existing collection location108permanently fails, management module126selects a new collection location y108. In one implementation, for example, this is accomplished as follows:1. Management module126selects a new location y for a collection128; selects z in Lc\{x} to be a source108for collection repair (if Lc\{X} is the empty set, then all collection locations are compromised and data loss occurs).2. Management module126updates centrally managed collection information134as follows: Lcis Lc\{x}; Lc′ is changed to Lc′\{x}∪{y}; execution_plan is added to the data structure (132) with (copy z to y); and instance number is incremented by one; recycle remains the same. Management module126syncs up the new represented state with the data nodes108in Lc∪Lc′.3. Management module126informs the source z to copy all of its objects in the collection128to y. Data node z copies its objects in the collection128to y, and it also passes the collection state and the object states (e.g., a respective portion of local information136) to y.4. It is possible that meanwhile new objects are checked into z. Node z can terminate the copying process as described, for example, in steps 5 and 6.5. Once node y successfully locally stores object replicas received from z, node y sends positive acknowledgments back to z. For objects stored, y marks the object as complete if the object is marked complete on z, and marks it partial otherwise. Node y also locally stores collection local information136.6. Node z performs the following to terminate the copying process. After starting the copying process, z determines a cutoff time t. A convenient time is when z receives the new collection state and is asked to copy data to y. For all object replicas successfully stored on z by time t, node z has to successfully copy all of them to node y regardless of whether the replicas are in complete or partial states. For any object that is being stored on z after time t (due to new check-ins or partial object repair operations at other nodes), node z will not copy the object over to y, but it will follow the described check-in protocols for these new objects. Note that for these new objects, since the collection is in a transitional state, the new objects are stored as partial replicas and will not be marked complete. So for these new replicas, their check-ins may still return success status but their replicas are in the partial state. Node z is now in a transitional state.7. After node z successfully copies all object replicas before the cutoff time, the copying process completes, and node z informs management module126. At this time, management module126changes the centrally managed collection information134for the collection128as follows. The new collection state Lcis changed to Lc\{x}∪{y}, Lc′ is unchanged (it was already changed in step 2), execution_plan is changed by removing the (copy z to y) entry, instance number is incremented by one, and recycle is unchanged. (Note that Lc′ may not be equal to Lchere because there might be other ongoing repairs for this collection128. However, this particular repair from z to y is complete). Management module126informs all data nodes in Lcand Lc′ about the new collection state.

Case 2. xεLc′\Lc. In this case, a new node108selected for a future collection location fails before it becomes a current collection location. Management module126selects a new node y replacing x. The procedure is analogous to that in Case 1.

Case 3. xεLc\Lc′. In this scenario, a data node108that is not going to be in a future collection location has failed. This location is garbage collected and management module126removes x from Lc, increments the instance number, and syncs up with the data nodes108. The repair source z could also fail in the middle of repair. This would leave y partially repaired. In this latter case, management module126checks that y is in Lc′ and there is a pending execution_plan (copy from z to y, but z failed). Then management module126selects a new repair source108, changes the execution_plan, and resumes the repair.

Transient Data Node Failures

To allow new data object check-ins when there are transient data node108failures, management module126implements, for example, the following process. After determining a data node108is in a transient failure state, for every collection128hosted on the transiently failed node108, management module126changes the collection state in centrally managed collection information134such that Lcexcludes the node, the instance number is changed, but Lc′ remains unchanged (and thus including the failed node). Management module126syncs up this new state with the data nodes108in Lc∪Lc′. With this change, the check-in protocol is analogous to the protocol followed by “Collection Refresh”. With new node108excluded in Lcthe check-in will not store the node in the failed node, and thus check-in can be successful if the number of nodes in Lcis at least t, the minimum number of successfully stored replicas in a check-in operation. Since the collection128is now in a transitional state, newly checked-in object replicas will all be partial. If management module126later determines the node failure is permanent, then it will start the described process for repair in view of permanent data node failure.

Load Balancing

In this implementation, management module126performs load balancing by moving collection128replicas from a node x108to a new node y108. The source x is in Lc∩Lc′. The protocol is analogous to that described for data repair. Lcis not changed (so node x can still be used for checkouts), and Lc′ is changed to Lc′\{x}∪{y}. The execution_plan is move from x to y. Termination criteria of the copying process and special transition state for a copying source are the same as discussed for collection repair. At the end of this load balancing process, management module126switches the new collection instance and syncs up with data nodes108. In this scenario, if it is desired for the repair source x not to serve checkout operations during data movement, the repair source can be removed from Lcwhen the data movement starts to simplify collection state transition.

Geographically Distributed Replication

FIG. 2shows an exemplary system200for geographically distributed collection-based replication (“geo-reptication”), according to one embodiment. In this implementation, system200includes two geographically distributed collection-based replication sites202and204. Each geographically distributed site includes, for example, at least a subset of the components of system100ofFIG. 1. For example, in one implementation, a geographically distributed site includes a collection information server (CIS) (e.g., analogous to CIS106ofFIG. 1) coupled to multiple data storage nodes (e.g., analogous to data storage nodes108ofFIG. 1). At least one geographically distributed site202or204also includes one or more client computing devices102coupled to the local and/or remote site CIS and data storage nodes. AlthoughFIG. 2shows only two geographically separated sites for collection-based replication, system200could include any number of such sites.

For purposes of exemplary description, the operations of a geo-replication site202or204are described with respect to the components ofFIG. 1, site202is referred to as a local site, and site204is referred to as a remote site. For example, using API124, a client102in a local site202specifies when a collection128is to be geo-replicated (e.g., with the CreateCollection( ) interface) to a different site204. For a geo-replicated collection128, the local CIS106(and more particularly, management module126) records this in collection state information associated with the collection128. Such collection state information is shown as a respective portion of centrally managed collection information134. Management module126initiates a create collection operation (e.g., CreateCollection( )) at the remote site204. If the connection to the remote site204is down, in one implementation management module126retries until the collection128is created on the remote site204, or until some other criteria is satisfied (e.g., a maximum number of tries, etc). In one implementation, at least a subset of APIs124include two versions, a first version for local site202access by a client102, and a second version for remote site204access by a client102.

When an object in a geo-replicated collection128is checked into one site202, it is replicated to the remote site204(or vice versa). Among collection locations in the local site202, a single data node108is dedicated (e.g., by lexicographic order, etc.) as responsible for geo-replication to the remote site204. For purposes of differentiation, the single data node is referred to as a primary data node for the collection128. In this scenario, each object replica is associated with a status indicating that the object replica has been successfully checked into the remote site204(e.g., a remote-complete status with respect to the remote site check-in operations). When an object is first checked into the local site202, the local check-in protocol is the same described above in the section titled “Data Object Storage/Check-In”, with the object being initially associated with a partial state, which is changed to complete after successful check-in of the object at the local site202.

In one implementation, after an object replica is in the complete state, if the hosting node108is a primary node, the hosting node108checks the object into the remote site204for geo-replication (e.g., by calling the CreateCollection( ) interface on the remote site204). After the remote check-in operations returns successfully (e.g., an acknowledgment is sent from the remote site in204to the local site202), the hosting node108marks its local copy of the object as remote-complete and broadcasts this status change to other data nodes108associated with the corresponding collection128. If the hosting node108of a complete object is not the primary node108, the hosting node108periodically checks with the primary node108to see if the object has turned remote-complete. If so, the hosting node108turns its local copy to remote-complete.

For a remote site204, when a new object is checked in by a different site202, the object replicas stored locally at the remote site204are first in the partial state, and subsequently remote-complete state once the object is already reliably stored at the originating site202.

Exemplary Procedures

An Exemplary Procedure for a Data Storage Node

FIG. 3shows an exemplary procedure300for collection-based replication operations of a data storage node108ofFIG. 1in a storage system100(FIG. 1), according to one embodiment. For purposes of discussion, the operations ofFIG. 3are described in reference to components ofFIG. 1. For instance, in the description, the left-most digit of a component reference number identifies the particular Figure in which the component first appears. Additionally, although the operations of procedure300are illustrated with respect to a sequential ordering of operations associated with blocks302through316, operations of respective ones of these blocks could be implemented in a different order. For example, in one implementation the operations of block308and310are performed before the operation of block306, etc. In view of this, exemplary operations of procedure300are now described.

At block302, data storage nodes108create a respective replica of multiple replicas of a collection128. The collection128represents a unit of data placement, access, replication, and repair in the system100. At block304, a particular node of the data storage nodes108receives a request (e.g., directly from a client102or from the server106) in the system100. The request is to perform an action with respect to the multiple replicas of the collection128. Such an action includes, for example, a request to check-in a data object into a replica of the collection128, access the data object from the replica, remove the data object from the replica, and/or remove the replica from the node. At block306, the node processes the request in a manner that ensures each of the data storage nodes associated with the collection128has successfully performed the action with respect to a corresponding replica of the collection128before any success status is returned to the client102. The operations of block306may include operations by the node to facilitate partial data object repair (as described above in the section titled “partial Data Object Repair”) for collection locations that have not indicated (within a configurable amount of time) that the action was successfully completed. The operations of this block also include return of an error status to the requesting client102if the action is not successfully performed by each of the collection locations.

At block308, the data node108determines that the local information136of the collection128maintained by the data node108is out of date/not current. In one implementation, this determination is made in view of a locally maintained instance number of the collection and a different instance number of the collection received from a different entity (e.g., the server106and/or a different data storage node108) in a request). At block310, and responsive to determining that the collection information is out of date, the data node108requests synchronization (a collection refresh operation) from the server106to obtain a current version of the collection information136.

At block312, a data storage node108receives a request from server106to repair a collection128lost by a permanently failed data storage node108to a new collection location (i.e., a data storage node108that was not previously part of the collection locations). At block314, and responsive to this repair request, the data storage node copies the collection128, excluding any new objects recently checked-into the collection128during the repair process, to the new collection location. At block316, and responsive to data object replication operations of any origin (e.g., a data object storage request from a client102, a collection repair request from a server106, etc.), the data storage node determines whether objects in a collection128have been fully replicated by all other locations (i.e., data storage of108) associated with the collection128. This is performed independently of server maintained replication information. In this implementation, server106does not maintain any indications of whether individual data objects associated with collections128, or collections128as a whole, are fully replicated across respective ones of the data storage nodes108in system100.

An Exemplary Procedure for a Client Computing Device

FIG. 4shows an exemplary procedure400for collection-based replication operations of a client computing device102ofFIG. 1in a storage system100(FIG. 1), according to one embodiment. For purposes of discussion, the operations ofFIG. 4are described in reference to components ofFIG. 1. For instance, in the description, the left-most digit of a component reference number identifies the particular Figure in which the component first appears.

At block402, a program module (i.e., collection-based model118) requests server106to create a collection128for replication across multiple data storage nodes108. At block404, and responsive to the create collection request of block402, the program module receives a list of collection locations (an ordered array of data storage node108identifiers) from the server106. Each collection location represents a respective data storage node108that stores a respective replica of the collection120. At block406, the program module directly contacts the first data storage node108(“first node”) identified in ordered array of collection locations to request the first node to perform an action to manage content associated with the replica of the collection stored by the first node. Such an action includes, for example, a request to store the data object into the replica, retrieve the data object from the replica, remove/delete the data object from the replica, and/or remove the replica from the system100, Responsive to successfully completing the action, the first node forwards the request to a next node in the ordered array of collection locations. This forwarding is implemented according to criteria that ensure that each node listed in the collection locations, if the node is properly operating, will have the opportunity to successfully complete the action. Thus, the client102directly contacts a single data storage node108to manage content across all replicas of the collection128independent of any object-to-collection mapping maintained by the server106. In this implementation, server106does not maintain any such mapping. In an alternate implementation, the client102may contact all data storage nodes108in the ordered array of collection locations itself.

In one implementation, the program module of the immediately preceding paragraph performs the operations associated with blocks402and406responsive to a respective requests received (e.g., via an API124) from a different application120executing on the client computing device102.

An Exemplary Procedure for Data Repair by a Collection-Based Index Server

FIG. 5shows an exemplary procedure500for data repair operations by a collection-based index server (“server”)106ofFIG. 1in a storage area network, according to one embodiment. For purposes of discussion, the operations ofFIG. 5are described in reference to components ofFIG. 1. For instance, in the description, the left-most digit of a component reference number identifies the particular Figure in which the component first appears.

At block502, a collection-based index server (“server”)106(FIG. 1) identifies/detects a permanently failed data storage node108in a data storage system100. At block504, server106selects a particular replication scheme to repair a collection128that was lost on the permanently failed data storage node108. The particular replication scheme is selected based on whether the failed node108represented an existing location for the lost collection128, or a new (in progress or future) storage location for the lost collection. At block506, the server106requests a single data storage node108that maintains an existing replica of the lost collection128to replicate the lost collection to a new data storage in the108(i.e., begin repair operations according to the select replication scheme). In this implementation, server106does not make any determination of whether the lost collection128is eventually fully replicated in the system100. Rather, the single data storage node108makes that determination.

CONCLUSION

Although collection-based object replication has been described in language specific to structural features and/or methodological operations or actions, it is understood that the implementations defined in the appended claims are not necessarily limited to the specific features or actions described. Rather, the specific features and operations are disclosed as exemplary forms of implementing the claimed subject matter.