Gridgain leader election pattern

Question

What is the advised approach to ensuring a particular process is running exactly once within the grid?

Usecase would be subscription of a single multiplexed stream from a remote source in order to update data across the grid. We need to elect the node to subscribe and elect a new node to subscribe when that node fails.

Is there any pre-built pattern for this in gridgain or is it solved by a combination of listening to grid lifecycle events and a distributed CAS operation?

Another usecase would be a singleton job that must run forever, migrating to a new node on failure.

Thanks.

Answer 1

You can simply take the oldest node in the cluster and start your operation on that node (Grid.nodes() will return all the nodes in the grid). You should also subscribe a discovery event listener on other nodes and have the next oldest node take over in case if oldest node fails.

To check if a node is oldest or node you can use GridNode.order() method. Node with smallest order will be the oldest.

To listen to discovery events, you can use this code:

        grid.events().localListen(new GridPredicate<GridEvent>() {
            @Override public boolean apply(GridEvent event) {
                System.out.println("Event: " + event.name());

                return true;
            }
        }, GridEventType.EVTS_DISCOVERY);

Answer 2

Starting with GridGain 6.2.0-rc2 release, GridGain has several ways for leader election:

1. GridProjection.oldest() will return you a dynamic projection over the oldest node in the cluster. If oldest node leaves cluster for whatever reason, then the next oldest node is automatically picked, so user can continue using the oldest node without interruption.
2. GridGain added DistributedServices feature which provides ability over controlled service deployment in the Grid. Some cool features include cluster-singleton-service, per-node-singleton-service, or per-cache-key-singleton-service. This is not leader election per se, but it may remove the need for leader election altogether as GridGain, for example, will guarantee for cluster-singleton-service that there is only one instance of that service available in the grid at any time.

For more information on distributed services refer to Distributed Services documentation.

Answer 3

Please, check the code bellow. I tested it using GridGain 6.1.8.

import org.gridgain.grid.*;
import org.gridgain.grid.cache.GridCache;
import org.gridgain.grid.cache.GridCacheConfiguration;
import org.gridgain.grid.cache.GridCacheMode;

import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;

public class GridGainMain {

    public static void main(String[] args) throws GridException {

        GridConfiguration config = new GridConfiguration();
        // Give a name to your grid.
        config.setGridName("MyCoolGrid");

        // Configure the cache that will be used by the leader election algorithm.
        GridCacheConfiguration leaderConf = new GridCacheConfiguration();
        leaderConf.setName("leader");
        leaderConf.setCacheMode(GridCacheMode.REPLICATED);

        config.setCacheConfiguration(leaderConf);

        // Start the grid!
        try (Grid grid = GridGain.start(config)) {

            // Get the local node.
            final GridNode localNode = grid.localNode();
            // Get the leader cache.
            final GridCache<String, String> leaderCache = grid.cache("leader");

            // Elect this member as the leader, if no other node was elected yet.
            putIfAbsent("leader", localNode.id().toString(), leaderCache);

            // ================================================
            // Schedule the leader election algorithm. 
            // The leader election algorithm will elect the oldest grid node as the leader.
            // ================================================
            new Timer().scheduleAtFixedRate(new TimerTask() {
                                                @Override
                                                public void run() {

                                                    // Get the self ID.
                                                    final String selfId = localNode.id().toString();
                                                    // Get the cached leader ID.
                                                    final String cachedLeaderId = get("leader", leaderCache);
                                                    // Get all nodes.
                                                    List<GridNode> list = new ArrayList<>(grid.nodes());
                                                    // Sort all nodes by natural order.
                                                    list.sort((o1, o2) -> (int) (o1.order() - o2.order()));
                                                    // Get the ID of the oldest node, which is the leader ID.
                                                    final String leaderId = list.get(0).id().toString();

                                                    // If the leader ID is not equals to the cached leader ID,
                                                    if (!leaderId.equals(cachedLeaderId)) {
                                                        // Put the leader ID into cache.
                                                        put("leader", leaderId, leaderCache);
                                                    }

                                                    // If this node is the leader,
                                                    if (selfId.equals(leaderId)) {
                                                        // =====================================
                                                        // Do something! Only this grid node will execute this code.
                                                        // =====================================
                                                    }

                                                    System.out.println("### Self ID: " + selfId
                                                            + ", Order: " + localNode.order()
                                                            + ", Leader ID: " + leaderId);
                                                }
                                            },
                    // Schedule now.
                    0L,
                    // Run the algorithm once every five seconds.
                    TimeUnit.SECONDS.toMillis(5));

            // Remove this in production.
            sleep(1, TimeUnit.DAYS);
        }
    }

    private static <T> T get(String key, GridCache<String, T> cache) {
        try {
            return cache.get(key);
        } catch (GridException e) {
            return null;
        }
    }

    private static <T> T putIfAbsent(String key, T value, GridCache<String, T> cache) {
        try {
            return cache.putIfAbsent(key, value);
        } catch (GridException e) {
            return null;
        }
    }

    private static <T> T put(String key, T value, GridCache<String, T> cache) {
        try {
            return cache.put(key, value);
        } catch (GridException e) {
            return null;
        }
    }

    public static void sleep(long duration, TimeUnit unit) {
        try {
            unit.sleep(duration);
        } catch (InterruptedException e) {
            // Ignore.
        }
    }

}