...

Motivation

When building a cluster of Ignite nodes, users need to be able to establish some restrictions on the member nodes based on cluster invariants in order to avoid breaking the consistency of the cluster. Such restrictions may include: having the same product version across the cluster, having consistent table and memory configurations, enforcing a particular cluster stateIgnite clusters are required to have a dynamically changing topology due to the nature of distributed systems - nodes can fail and need to be restarted at any time, so new nodes can be introduced or removed from the cluster. Users expect that topology changes do not break the consistency of the cluster and it remains operational.

Description

Problem statement

This document describes the process of creating a new node joining a cluster , which includes a validation step where a set of rules are applied to determine whether the incoming node is able to enter the current topology. These rules may include node-local information (e.g. node version) as well as cluster-wide information (e.g. data encryption algorithm), which means that the validation component may require access to the Meta Storage (it is assumed that Meta Storage contains the consistent cluster-wide information, unless some other mechanism is proposed). The problem is that, according to the Node Lifecycle description, a cluster can exist in a "zombie" state, during which the Meta Storage is unavailable. This means the the validation process can be split into 2 steps:

1. "Pre-init" validation: a joining node tries to enter the topology on the network level and gets validated against its local properties.
2. "Post-init" validation: the cluster has received the "init" command, which activates the Meta Storage, and the joining node can be validated against the cluster-wide properties.

Apart from the 2-step validation, there are also the following questions that need to be addressed:

1. Where will the whole process happen: on the joining node itself or on an arbitrary remote node.
2. How to negate possible security issues if a not yet fully validated node gets access to the Meta storage.
3. How to deal with different configurations of the Meta Storage: the "most recent" configuration should be consistently delivered to all nodes in a cluster.

Terminology

Init command

The "init" command is supposed to move the cluster from the "zombie" state into the "active" state. It is supposed to have the following characteristics (note that the "init" command has not been specified at the moment of writing and is out of scope of this document, so all statements are approximate and can change in the future):

1. It should deliver the following information: addresses of the nodes that host the Meta Storage Raft group, Meta Storage version and a cluster tag (described below).
2. It should deliver this information atomically, i.e. either all nodes enter the "active" state or none. As a possible solution, it can be implemented similarly to a two-phase commit: first, a "prepare" message is broadcasted to all nodes in the current topology. The initiator node remembers the topology members at the start of the prepare phase and restarts the operation (or sends additional messages) until the topology is stable. After the prepare phase is finished, the commit message is broadcasted. Until the commit phase finishes, no new nodes are allowed to enter the cluster.

Initialized and empty nodes

This document uses a notation of "initialized" and "empty" nodes. An initialized node is a node that has received the "init" message sometime in its lifetime and therefore possesses the cluster tag and the meta storage version. An empty node has never received the "init" command and does not possess the aforementioned properties.

Meta Storage version

Meta Storage version is a totally ordered property that should be used to compute the most "recent" state of the Meta Storage configuration. A possible implementation can be a monotonically increasing counter, which is increased every time the Meta Storage configuration (e.g. addresses of nodes that host the Meta Storage Raft group) is updated.

of Ignite nodes and adding new nodes to it. It describes the following concepts:

1. Cluster setup - the process of assembling a predefined set of nodes into a cluster and preparing it for the next lifecycle steps;
2. Cluster initialization - after setting up the cluster, it should be transferred to a state when it is ready to serve user requests;
3. Node validation - in order to maintain the cluster configuration in a consistent state, the joining nodes should be compatible with all of the existing cluster nodes.

Terminology

Meta Storage

Meta Storage is a subset of cluster nodes hosting a Raft group responsible for storing  a master copy of cluster metadata.

Cluster Management Group

Cluster Management Group or CMG is a subset of cluster nodes hosting a Raft group. CMG leader is responsible for orchestrating the node join process.

Init command

The init command is issued by a user with a CLI tool and moves a cluster from the idle state into the running state.

Idle and running cluster

An idle cluster is a cluster that has been assembled for the first time and has never received the init command, therefore the Meta Storage of this cluster does not exist. Acluster can be considered running if it has obtained the information about the Meta Storage and CMG location.

Empty and initialized node

Every Ignite node starts in the empty state. After joining the cluster and passing the validation step, a node obtains the location of the Meta Storage and transitions to the initialized state.

Cluster Tag

...

A cluster tag is a string that uniquely identifies a cluster (e.g. a UUID). It is generated once per cluster and is distributed across the nodes during the "init" phasenode join process. The purpose of a cluster tag is to understand whether a joining node used to be a member of another cluster, in which case its Meta Storage version is not comparable and the joining node it should be rejected.

Validation approaches

Local validation

Local validation approach requires the joining node to retrieve some information from a random node/Meta Storage and deciding to join the cluster based on that information.

This approach has the following pros and cons:

1. When using in the rolling upgrade scenario, it might be easier to maintain backward compatibility: a newer joining node already knows about the requirements of the older nodes in the cluster.
2. Possible security issues: if a node is able to allow itself to join, it might be easier to compromise the cluster.

Remote validation

Remote validation approach requires the joining node to send some information about itself  to a remote node, which decides whether to allow the new node to join or not.

This approach has the following pros and cons:

1. This approach is used in Ignite 2, which may be more familiar to users and developers.
2. It may be more secure, since a node can't join without notifying at least one valid node.
3. Harder to support backward compatibility.

Discussion needed: At the time of writing this document, it is assumed that validation protocol is going to be remote.

Implementation details

Based on the initialization status of the nodes in a cluster, there can be several possible scenarios of new nodes entering the topology:

1. An empty node enters a cluster where all nodes are empty.
2. An empty node enters a cluster where all nodes are initialized.
3. An initialized node enters a cluster where all nodes are empty.
4. An initialized node with a Meta Storage of version X enters a cluster where all nodes have the Meta Storage of version Y.

Common logic

A joining node tries to enter the topology. It is possible to piggyback on the transport of the membership protocol in order to exchange validation messages before allowing to send membership messages (similar to the handshake protocol). During this step it sends some information (cluster tag, Meta Storage version, node version) to a random node and gets validated (more details below). After this step is complete, the joining node becomes visible through the Topology Service, therefore establishing an invariant that visible topology will always consist of nodes that have passed the first validation step. Possible issues: there can be a race condition when multiple conflicting nodes join at the same time, in which case only the first node to join will be valid. This can be considered expected behavior, because such situations can only occur during the initial set up of a cluster, which is a manual process and requires manual intervention anyway.

Empty node joins a cluster

If an empty node tries to join a cluster the following process is proposed:

1. It connects to a random node, sends the available local validation information and enters the topology, if it gets accepted.
2. The following scenarios can then happen:
   1. The random node is initialized. The joining node should then retrieve the information, that was broadcasted by the "init" command, become initialized and finish the join process.
   2. The random node is empty because the cluster has not yet been initialized. In this case the node finishes the join process and remains in the "zombie" state until the "init" command arrives.
   3. The random node is empty because it hasn't finished the join process itself. In this case the random node should send a corresponding message, and the joining node should choose another random node and repeat the process.

Initialized node joins a cluster

If an initialized node tries to join a cluster the following process is proposed:

1. It connects to a random node and sends the available local validation information (including the cluster tag and the Meta Storage version).
2. The following scenarios can then happen:
   1. The random node is initialized and the cluster tags do not match. The joining node must be rejected.
   2. The random node is initialized, the cluster tags match, local Meta Storage version is "smaller" than the remote. The node joins the topology and updates its Meta Storage configuration, thus ending the joining process.
   3. The random node is initialized, the cluster tags match, local Meta Storage version is "larger" than the remote. Discussion needed: what is the expected behavior in this case?
      
   4. The random node is empty because the cluster has not yet been initialized. The joining node should initiate a process similar to sending the "init" command. Discussion needed: what to do if another "init" command is running in parallel?
   5. The random node is empty because it hasn't finished the join process itself. In this case the random node should send a corresponding message, and the joining node should choose another random node and repeat the process.

Changes in API (WIP)

NetworkTopologyService

Current TopologyService  will be renamed to NetworkTopologyService . It is proposed to extend this service to add validation handlers that will validate the joining nodes on the network level.

/**
 * Class for working with the cluster topology on the network level.
 */
public interface NetworkTopologyService {
    /**
     * This topology member.
     */
    ClusterNode localMember();

    /**
     * All topology members.
     */
    Collection<ClusterNode> allMembers();

    /**
     * Handlers for topology events (join, leave).
     */
    void addEventHandler(TopologyEventHandler handler);

    /**
     * Returns a member by a network address
     */
    @Nullable ClusterNode getByAddress(NetworkAddress addr);

    /**
     * Handlers for validating a joining node.
     */
    void addValidationHandler(TopologyValidationHandler handler);
}

TopologyService

The new service will have the same API, but will work on top of the Meta Storage, and will provide methods to work with the list of validated nodes. In addition to that, it will perform the validation of incoming nodes against the Meta Storage, based on the registered validation handlers.

/**
 * Class for working with the cluster topology on the Meta Storage level. Only fully validated nodes are allowed to be present in such topology.
 */
public interface TopologyService {
    /**
     * This topology member.
     */
    ClusterNode localMember();

    /**
     * All topology members.
     */
    Collection<ClusterNode> allMembers();

    /**
     * Handlers for topology events (join, leave).
     */
    void addEventHandler(TopologyEventHandler handler);

    /**
     * Returns a member by a network address
     */
    @Nullable ClusterNode getByAddress(NetworkAddress addr);

    /**
     * Handlers for validating a joining node.
     */
    void addValidationHandler(TopologyValidationHandler handler);
}

TopologyService will depend on the MessagingService  (to respond and listen to validation requests) and on the MetaStorageManager (for interacting with the Meta Storage).

Risks and Assumptions

...

A cluster tag consists of two parts:

1. Human-readable part (Cluster Name): a string property that is set by the system administrator. Its purpose is to make the debugging and error reporting easier.
2. Unique part (Cluster ID): a generated unique string (e.g. a UUID). Its purpose is to ensure that Cluster Tags are different between different clusters.

Physical topology

Physical topology consists of nodes that can communicate with each other on the network level. Nodes enter the topology though a network discovery mechanism, currently provided by the SWIM protocol. However, such nodes may not yet have passed the validation step, so not all nodes from the physical topology can participate in cluster-wide activities. In order to do that, a node must enter the logical topology.

Logical topology

Logical topology consists of nodes that have passed the validation step and are therefore able to participate in cluster-wide activities. Logical topology is maintained by the Cluster Management Group.

Protocol description

Initial cluster setup

This section describes the process of assembling a new Ignite cluster from a set of empty nodes.

1. Initial set of nodes is started, providing the following properties:
   1. A subset of nodes (minimum 1, more can be specified to increase startup reliability) in the initial cluster setup, provided by an IP Finder. Concrete IP Finder implementations can be used to obtain the seed members, depending on the environment the cluster is running in (e.g. conventional networks or Kubernetes), and are specified either via the configuration or the CLI.
2. The nodes assemble into a physical topology using a network discovery protocol (e.g. SWIM), bootstrapped with the provided seed members.
3. An init command is sent by a user to a single node in the cluster, providing the following information:
   1. Consistent IDs (names) of the nodes that will host the Meta Storage;
   2. Consistent IDs (names) of the nodes that will comprise the Cluster Management Group. It is possible for both of these address sets to be the same.
4. The node, that has received the command, propagates it to all members of the physical topology that were specified in the init command. These members should start the corresponding Raft groups and, after the group leaders are elected, the initial node should return a response to the user. In case of errors, Raft groups should be removed and an error response will be returned to the user. If no response has been received, the user should retry sending the command with the same parameters to a different node.  
5. As soon as the CMG leader is elected, the leader initializes the CMG state by applying a Raft command (e.g. ClusterInitCommand), which includes:
   1. A generated Cluster Tag;
   2. Ignite product version.
6. After the command has been applied, the leader sends a message to all nodes in the physical topology, containing the location of the CMG nodes. At this point the cluster can be considered as running.
7. Upon receiving the message, each node sends a join request to the CMG leader, which consists of:
   1. Protocol version (an integer which is increased every time the join procedure is changed, needed to support backwards compatibility);
   2. Ignite product version;
8. Information from the join requests gets validated on the leader and, if the properties are equal to the CMG state (in case of the protocol version, a different comparison algorithm might be used), a successful response is sent, containing:
   
   1. Consistent IDs (names) of the Meta Storage nodes;
   2. Cluster Tag.
      If the properties do not match, an error response is sent and the joining node is rejected.
9. If the joining node has passed the validation and received the validation response, it starts some local recovery procedures (if necessary) and sends a response to the CMG leader, indicating that it is ready to be added to the logical topology.
10. The CMG leader issues a Raft command (e.g. AddNodeCommand), which adds the node to the logical topology.

An example of this flow can be found on the diagram below. Some initialization steps are omitted for Node C as they are identical to the corresponding steps on Node B.

Image Added

Adding a new node to a running cluster

Depending on the state of the node or the cluster there exist 4 possible scenarios:

Empty node joins an idle cluster

This scenario is equivalent to the initial cluster setup.

Initialized node joins an idle cluster

Such nodes should never be able to join the cluster, because it will fail the cluster tag validation step.

Empty node joins a running cluster

1. The new node enters the physical topology;
2. CMG leader discovers the new node and sends a message to it, containing the location of the CMG nodes;
3. Upon receiving the message, the joining node should execute the validation procedure and be added to the logical topology, as described by steps 7-9 of the Initial cluster setup section.

Initialized node joins a running cluster

1. The new node starts the CMG group server or client, depending on the existing local node configuration;
2. The new node enters the physical topology;
3. If the new node is elected CMG leader, it should use the local CMG state to validate itself, otherwise see point 4.
4. CMG leader discovers the new node and sends a message to it, containing the location of the CMG nodes;
5. Upon receiving the message, the joining node should execute the validation procedure and be added to the logical topology, as described by steps 7-9 of the Initial cluster setup section.

Implementation details

Node start flow

The following changes are proposed to the node start scenario in regards to the changes to the join protocol:

Image Added

Each blue rectangle represents a start of a component, changes are marked in red and notable action points are marked in green.

According to the diagram, the following changes are proposed:

1. RESTManager component is started earlier.
2. CMGManager component, responsible for managing CMG interactions, introduced.
   
3. nodeRecoveryFinished action item introduced. It’s a step within components' start process that denotes that a given node has finished its recovery and is ready to be included in a logical topology.

RestManager changes

RestManager should be started earlier, since it is required to register REST message handlers early to handle the init command.

Image Added

CMGManager

CMGManager is responsible for interacting with the CMG and should perform the following actions:

1. Launch the local CMG Raft server in case an existing local configuration exists;
2. Register a REST message handler for the init command;
3. Register a message handler that will be listening for messages from the CMG leader and sending a join request.

Changes in API

TODO

Risks and Assumptions

1. Proposed implementation does not discuss message encryption and security credentials.;
2. Protocol for nodes leaving the topology is out of scope of this document;
3. Protocol for migrating CMG and Meta Storage to different nodes is out of scope of this documentTwo-layered topology view may be confusing to use.

Discussion Links

https:// Links to discussions on the devlist, if applicable.lists.apache.org/thread/4lor2vxkg6x94thprvcr0h19rkm6j1gt

Reference Links

1. IEP-73: Node startup
2. https://github.com/apache/ignite-3/blob/main/modules/runner/README.md
3. IEP-67: Networking module
4. IEP-61: Common Replication Infrastructure

Tickets