EC2 node discovery for Hazelcast

README.md

Alternative implementation of TcpIpJoinerOverAWS

This is a workaround for this Hazelcast issue.

To use the alternative TcpIpJoinerOverAWS:

• Remove hazelcast-cloud from your project's classpath,
• Add this class in your project or JAR it up if you prefer,
• Do not configure an accessKey nor a secretKey in Hazelcast's AwsConfig.

Note that it depends on:

• com.amazonaws:aws-java-sdk-ec2
• org.apache.commons:commons-lang3

We're using it at Unbounce and are pretty happy with it.

TcpIpJoinerOverAWS.java

package com.hazelcast.cluster;

import static com.amazonaws.regions.RegionUtils.getRegion;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.commons.lang3.Validate;

import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import com.amazonaws.internal.StaticCredentialsProvider;
import com.amazonaws.services.ec2.AmazonEC2;
import com.amazonaws.services.ec2.AmazonEC2Client;
import com.amazonaws.services.ec2.model.DescribeInstancesRequest;
import com.amazonaws.services.ec2.model.DescribeInstancesResult;
import com.amazonaws.services.ec2.model.Filter;
import com.amazonaws.services.ec2.model.Instance;
import com.amazonaws.services.ec2.model.Reservation;
import com.hazelcast.config.AwsConfig;
import com.hazelcast.instance.Node;
import com.hazelcast.logging.ILogger;
import com.hazelcast.util.ExceptionUtil;

/**
 * An Hazelcast {@link TcpIpJoiner} for EC2, which uses the official AWS SDK. Forked and rewritten
 * from the original class.
 */
public class TcpIpJoinerOverAWS extends TcpIpJoiner
{
    private final ILogger logger;
    private final AwsConfig awsConfig;
    private final AmazonEC2 ec2;

    public TcpIpJoinerOverAWS(final Node node)
    {
        super(Validate.notNull(node, "node can't be null"));

        logger = node.getLogger(getClass());
        awsConfig = node.getConfig().getNetworkConfig().getJoin().getAwsConfig();
        ec2 = new AmazonEC2Client(newAwsCredentialsProvider(awsConfig.getAccessKey(),
            awsConfig.getSecretKey()));

        if (isNotBlank(awsConfig.getRegion()))
        {
            ec2.setRegion(getRegion(awsConfig.getRegion()));
        }
    }

    private static AWSCredentialsProvider newAwsCredentialsProvider(final String accessKey,
                                                                    final String secretKey)
    {
        if (isBlank(accessKey) && isBlank(secretKey))
        {
            return new DefaultAWSCredentialsProviderChain();
        }

        return new StaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey));
    }

    @Override
    protected Collection<String> getMembers()
    {
        try
        {
            final Collection<String> list = getPrivateIpAddresses();
            if (list.isEmpty())
            {
                logger.warning("No EC2 instances found!");
            }
            else
            {
                if (logger.isFinestEnabled())
                {
                    final StringBuilder sb = new StringBuilder("Found the following EC2 instances:\n");
                    for (final String ip : list)
                    {
                        sb.append("    ").append(ip).append("\n");
                    }
                    logger.finest(sb.toString());
                }
            }
            return list;
        }
        catch (final Exception e)
        {
            logger.warning(e);
            throw ExceptionUtil.rethrow(e);
        }
    }

    private Collection<String> getPrivateIpAddresses()
    {
        final DescribeInstancesRequest request = new DescribeInstancesRequest();

        if (isNotBlank(awsConfig.getSecurityGroupName()))
        {
            request.getFilters().add(
                new Filter().withName("group-name").withValues(awsConfig.getSecurityGroupName()));
        }

        if (isNotBlank(awsConfig.getTagKey()) && isNotBlank(awsConfig.getTagValue()))
        {
            request.getFilters().add(
                new Filter().withName("tag:" + awsConfig.getTagKey()).withValues(awsConfig.getTagValue()));
        }

        final List<String> privateIpAddresses = new ArrayList<>();

        do
        {
            logger.finest("Describing instances with request: " + request);

            final DescribeInstancesResult result = ec2.describeInstances(request);
            handleDescribeInstancesResult(result, privateIpAddresses);

            request.setNextToken(result.getNextToken());
        }
        while (isNotBlank(request.getNextToken()));

        return privateIpAddresses;
    }

    private void handleDescribeInstancesResult(final DescribeInstancesResult result,
                                               final List<String> privateIpAddresses)
    {
        for (final Reservation reservation : result.getReservations())
        {
            handleReservation(reservation, privateIpAddresses);
        }
    }

    private void handleReservation(final Reservation reservation, final List<String> privateIpAddresses)
    {
        for (final Instance instance : reservation.getInstances())
        {
            handleInstance(instance, privateIpAddresses);
        }
    }

    private void handleInstance(final Instance instance, final List<String> privateIpAddresses)
    {
        if ("running".equals(instance.getState().getName()))
        {
            privateIpAddresses.add(instance.getPrivateIpAddress());

            logger.finest("Accepted instance: " + instance);
        }
        else
        {
            logger.finest("Ignoring not-running instance: " + instance);
        }
    }

    @Override
    protected int getConnTimeoutSeconds()
    {
        return awsConfig.getConnectionTimeoutSeconds();
    }

    @Override
    public String getType()
    {
        return "aws";
    }
}

@unixunion When you don't configure the AWS credentials in the Hz config, the above code will use the DefaultAWSCredentialsProviderChain to locate credentials on the instance. Read this to understand how this mechanism works.

FYI, Hazelcast will now throw an error if accessKey and/or secretKey are left empty or not specified. I'm using version 3.4.2 for reference. Also, the package for this class has changed to com.hazelcast.cluster.impl.

What permissions are required for the IAM role to be used with this code?

FYI, required permissions are listed at https://www.linkedin.com/grp/post/49531-5849077519474524163?goback=%2Egmp_49531