High Availability in Depth

A data distribution policy defines how application data is distributed, or partitioned, across multiple nodes. Partitioning of application data allows large application data sets to be distributed across multiple nodes, typically each running on a separate machine. This provides a simple mechanism for applications to grow horizontally by scaling the amount of machine resources available for storing and processing application data. Data distribution policies are associated with one or more availability zones.

To support partitioning application data, an application fragment must be associated with one or more data distribution policies. A data distribution policy defines the following characteristics:

Partitions can be distributed across the available machine resources, either dynamically, a dynamic data distribution policy, or statically, a static data distribution policy.

A dynamic data distribution policy automatically creates dynamic partitions based on the available nodes. There is no requirement to explicitly configure dynamic partitions or to map them to specific nodes. Data in a dynamic data distribution policy is automatically redistributed when a node joins or leaves a cluster without any explicit operator action.

Figure 2, “Dynamic data distribution policy” shows a dynamic data distribution policy that provides round-robin data distribution across three nodes. As data is created, it is assigned to the nodes using a round-robin algorithm: Data 1 is created on Node A, Data 2 is created on Node B, Data 3 is created on Node C, and the next data created is created back on Node A. There are dynamic partitions created to support this data distribution policy, but they are omitted from the diagram since they are implicitly created by the runtime. If additional nodes are added, the round-robin algorithm includes them when distributing data.

A static data distribution policy requires that static partitions be explicitly configured and mapped to nodes. Data is never repartitioned when using a static data distribution policy. Data can only be migrated to other nodes by loading updated configuration, but the data is never reassigned to a new static partition once it has been created.

Figure 3, “Static data distribution policy” shows a static data distribution policy that uses a data mapper that maps odd data to a partition named Odd and even data to a partition named Even. As data is created it is created on the active node for the partition, so all Odd data is created on Node A and all Even data is created on Node B. Adding a new node to a static distribution policy does not impact where the data is created; it will continue to be created on the nodes that are hosting the partition to which the data is mapped. Contrast this to the behavior of the dynamic data distribution policy — where new nodes affect the distribution of the data.

The number of copies of data, or replicas, is defined by a data distribution policy. The policy also defines whether replication should be done synchronously or asynchronously, and other runtime tuning and recovery properties.

Data is distributed across partitions at runtime using a data mapper. A data mapper maps application data to a specific partition at runtime, usually based on the content of the application data. Built-in data mappers are provided, and applications can provide their own data mappers to perform application-specific data mapping.

Figure 4, “Single data center availability zone” shows a simple example of a single cluster with three nodes deployed in a single data center. There is one availability zone defined, which contains all of the nodes in the cluster. The availability zone would define a data distribution policy appropriate for the deployed application, and possibly define a quorum policy that ensures that quorum is always maintained.

Figure 5, “Disaster recovery (DR) availability zone” shows a more complex example of a cluster that spans two data centers to provide disaster recovery for an application. This examples defines these availability zones:

Notice that nodes C and E are in multiple availability zones, with each availability zone having possibly different quorum policies.

The application data types that represent the critical application data must be in the same data distribution policy. This distribution policy is then associated with the Disaster Recovery availability zone to ensure that the data is redundant across data centers.

Data is partitioned by installing a data mapper on a managed object using configuration. A managed object with a data mapper installed is called a partitioned object. A data mapper is responsible for assigning a managed object to a partition. Partition assignment occurs:

Data mappers are inherited by all subtypes of a parent type. A child type can install a new data mapper to override a parent's data mapper.

A partitioned object is always associated with a single partition, but the partition it is associated with can change during the lifetime of the object.

The algorithm used by a data mapper to assign an object to a partition is application-specific. It can use any of the following criteria to make a partition assignment:

Built-in data mappers for consistent hashing and round-robin are provided.

Figure 9, “Foreign partition” shows three nodes, A, B, and C, and a partition P. Partition P is defined with an active node of A and a replica node of B. Partition P is also defined on node C but node C is not in the node list for the partition. On node C, partition P is considered a foreign partition.

The partition state and node list of foreign partitions is maintained as the partition definition changes in the cluster. However, no objects are replicated to these nodes, and these nodes cannot become the active node for the partition. When an object in a foreign partition is created or updated, the create and update is pushed to the active and any replica nodes in the partition.

Foreign partition definitions are useful for application-specific mechanisms that require a node to have a distributed view of partition state, without being the active node or participating in replication.

Partitions are defined directly by the application or an administrator on a running system. At minimum, TIBCO recommends defining and enabling partitions (see Enabling and Disabling Partitions) on all nodes on which the partition should be known. This allows an application to:

As an example, here are the steps to define a partition P in a cluster with an active node of A and a replica node of B.

You can also redefine partition definitions to allow partition migration to different nodes. See Migrating a Partition for details.

The only time that node list inconsistencies are detected is when object re-partitioning is done (see Updating Object Partition Mapping), or a foreign partition is being defined.

Once a partition is defined, it must be enabled. Enabling a partition causes the local node to transition the partition from the Initial state to the Active state. Partition activation can include migration of object data from other nodes to the local node. It can also include updating the active node for the partition in the cluster. Enabling an already Active partition has no effect.

Disabling a partition causes the local node to stop hosting the partition. The local node is removed from the node list in the partition definition on all nodes in the cluster. If the local node is the active node for a partition, the partition migrates to the next node in the node list and becomes active on that node. As part of migrating, all objects in the partition on the local node are removed from shared memory.

When a partition is disabled with only the local node in the node list, there is no impact to the objects contained in the partition on the local node since a partition migration does not occur. The application can continue reading these objects. However, unless the partition mapper is removed, no new objects can be created in the disabled partition because there is no active node for the partition.

A partition with one or more replica nodes defined in its node list fails over if its current active node fails. The next highest priority available node in the node list takes over processing for this partition.

When a node fails, it is removed from the node list for the partition definition in the cluster. All undiscovered nodes in the node list for the partition are also removed from the partition definition. For example, if node A fails with the partition definitions in Figure 6, “Partition definitions” active, the node list is updated to remove node A leaving these partition definitions active in the cluster.

Once a node is removed from the node list for a partition, no communication occurs to that node for the partition.

A node is restored to service by defining and enabling all partitions that will be hosted on the node. This includes partitions for which the node being restored is the active or replica node. When a partition is enabled on the node being restored partition migration occurs, which copies all objects in the hosted partitions to the node.

Restoring node A to service after the failure in Failover requires the following steps:

After these steps execute and partition migration completes, node A is back online and the partition definitions are back to the original definitions in Figure 6, “Partition definitions”.

Partitions have one of the following states:

Figure 11, “Partition state machine” shows the state machine that controls the transitions between all of these states.

The external events in the state machine map to an API call or an epadmin command. The internal events are generated as part of node processing.

Partitions also have a status, which defines how the local definition of the partition was done, and whether it was enabled. The valid states are defined in Partition status.

Partition status values are controlled by an administrative operation, or API, on the local node except for the RemoteEnabled and RemoteDefined statuses. The RemoteEnabled and RemoteDefined statuses occurs when local partition state was not defined and enabled on the local node; it was only updated on a remote node.

If the local node leaves the cluster and is restarted, it must redefine and enable a partition locally before rejoining the cluster to rejoin as a member of the partition. For this reason TIBCO recommends that all nodes perform define and enable for all partitions in which they participate, even if they are a replica node in the partition.

If an I/O error is detected when attempting to send creates, updates, or deletes to a replica node, an error is logged on the node initiating the replication and the object modifications for the replica node are discarded. The replica node is removed from the node list for the partition. These errors include:

The replica node is then re-synchronized with the active node when the replica node is restored (see Restore).

The valid node quorum states are defined in Node quorum states.

Figure 16, “Quorum state machine” shows the state machine that controls the transitions between all of the node quorum states.

There are cases where disabling node quorum is desired. Examples are:

To support these cases, the node quorum mechanism can be disabled using configuration (see the StreamBase Administration Guide). When node quorum is disabled, high availability services are never disabled on a node because of a lack of quorum. With the node quorum mechanism disabled, a node can only be in the Active or Partial node quorum states defined in Node quorum states — it never transitions to the Disabled state. Because of this, it is possible that partitions may have multiple active nodes simultaneously.

This section describes how to restore a cluster following a multi-master scenario. These terms are used to describe the roles nodes play in restoring after a multi-master scenario:

To recover partitions that were active on multiple nodes, support is provided for merging objects using an application implemented compensation trigger. If a conflict is detected, the compensation trigger is executed on the source node to allow conflict resolution.

The detectable conflicts types are:

The application-implemented compensation trigger is always executed on the source node. The compensation trigger has access to data from the initiating and source nodes.

Figure 17, “Active cluster” shows an example cluster with a single partition, P, that has node A as the active node and node B as the replica node.

Figure 18, “Failed cluster” shows the same cluster after connectivity is lost between node A and node B with node quorum disabled. The partition P definition on node A has been updated to remove node B as a replica because it is no longer possible to communicate with node B. Node B removes node A from the partition definition because it believes that node A failed, so it has taken over responsibility for partition P.

Once connectivity was restored between all nodes in the cluster, and the nodes have discovered each other, the operator can initiate the restore of the cluster. The restore (see Restore) is initiated on the initiating node which is node A in this example. All partitions on the initiating node are merged with the same partitions on the source nodes on which the partitions are also active. In the case where a partition was active on multiple remote nodes, the node to merge from can be specified per partition, when the restore is initiated. If no remote node is specified for a partition, the last remote node to respond to the Is partition(n) active? request (see Figure 19, “Merge operation - using broadcast partition discovery”) will be the source node.

Figure 19, “Merge operation - using broadcast partition discovery” shows the steps taken to restore the nodes in Figure 18, “Failed cluster”. The restore command is executed on node A which is acting as the initiating node. Node B is acting as the source node in this example.

The steps in Figure 19, “Merge operation - using broadcast partition discovery” are:

The steps to restore a node, when the restore from node was specified in the restore operation are very similar to the ones above, except that instead of a broadcast to find the source node, a request is sent directly to the specified source node.

The example in this section has the A node as the final active node for the partition. However, there is no requirement that this is the case. The active node for a partition could be any other node in the cluster after the restore completes, including the source node.

Figure 20, “Split cluster” shows another possible multi-master scenario where the network outage causes a cluster to be split into multiple sub-clusters. In this diagram there are two sub-clusters:

To restore this cluster, the operator must decide which sub-cluster nodes should be treated as the initiating nodes and restore from the source nodes in the other sub-cluster. The steps to restore the individual nodes are identical to the ones described above.