Read ec2 discovery address from aws instance tags

This PR adds a new option for `host_type`: `tag:TAGNAME` where `TAGNAME` is the tag field you defined for your ec2 instance.

For example if you defined a tag `my-elasticsearch-host` in ec2 and set it to `myhostname1.mydomain.com`, then
setting `host_type: tag:my-elasticsearch-host` will tell Discovery Ec2 plugin to read the host name from the
`my-elasticsearch-host` tag. In this case, it will be resolved to `myhostname1.mydomain.com`.

Closes #22566.

docs/plugins/discovery-ec2.asciidoc

@@ -172,9 +172,18 @@ The following are a list of settings (prefixed with `discovery.ec2`) that can fu
 
 `host_type`::
 
-    The type of host type to use to communicate with other instances. Can be
-    one of `private_ip`, `public_ip`, `private_dns`, `public_dns`. Defaults to
-    `private_ip`.
++
+--
+The type of host type to use to communicate with other instances. Can be
+one of `private_ip`, `public_ip`, `private_dns`, `public_dns` or `tag:TAGNAME` where
+`TAGNAME` is the tag field you defined for your ec2 instance.
+
+For example if you defined a tag `my-elasticsearch-host` in ec2 and set it to `myhostname1.mydomain.com`, then
+setting `host_type: tag:my-elasticsearch-host` will tell Discovery Ec2 plugin to read the host name from the
+`my-elasticsearch-host` tag. In this case, it will be resolved to `myhostname1.mydomain.com`.
+
+Defaults to `private_ip`.
+--
 
 `availability_zones`::
 

plugins/discovery-ec2/src/main/java/org/elasticsearch/cloud/aws/AwsEc2Service.java

@@ -164,20 +164,21 @@ public interface AwsEc2Service {
      * Defines discovery settings for ec2. Starting with discovery.ec2.
      */
     interface DISCOVERY_EC2 {
-        enum HostType {
-            PRIVATE_IP,
-            PUBLIC_IP,
-            PRIVATE_DNS,
-            PUBLIC_DNS
+        class HostType {
+            public static final String PRIVATE_IP = "private_ip";
+            public static final String PUBLIC_IP = "public_ip";
+            public static final String PRIVATE_DNS = "private_dns";
+            public static final String PUBLIC_DNS = "public_dns";
+            public static final String TAG_PREFIX = "tag:";
         }
 
         /**
          * discovery.ec2.host_type: The type of host type to use to communicate with other instances.
-         * Can be one of private_ip, public_ip, private_dns, public_dns. Defaults to private_ip.
+         * Can be one of private_ip, public_ip, private_dns, public_dns or meta:XXXX where
+         * XXXX is the metadata field name we will read the address from. Defaults to private_ip.
          */
-        Setting<HostType> HOST_TYPE_SETTING =
-            new Setting<>("discovery.ec2.host_type", HostType.PRIVATE_IP.name(), s -> HostType.valueOf(s.toUpperCase(Locale.ROOT)),
-                Property.NodeScope);
+        Setting<String> HOST_TYPE_SETTING =
+            new Setting<>("discovery.ec2.host_type", HostType.PRIVATE_IP, Function.identity(), Property.NodeScope);
         /**
          * discovery.ec2.any_group: If set to false, will require all security groups to be present for the instance to be used for the
          * discovery. Defaults to true.

plugins/discovery-ec2/src/main/java/org/elasticsearch/discovery/ec2/AwsEc2UnicastHostsProvider.java

@@ -27,6 +27,7 @@ import com.amazonaws.services.ec2.model.Filter;
 import com.amazonaws.services.ec2.model.GroupIdentifier;
 import com.amazonaws.services.ec2.model.Instance;
 import com.amazonaws.services.ec2.model.Reservation;
+import com.amazonaws.services.ec2.model.Tag;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.Version;
@@ -51,6 +52,11 @@ import java.util.Set;
 import static java.util.Collections.disjoint;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.emptySet;
+import static org.elasticsearch.cloud.aws.AwsEc2Service.DISCOVERY_EC2.HostType.TAG_PREFIX;
+import static org.elasticsearch.cloud.aws.AwsEc2Service.DISCOVERY_EC2.HostType.PRIVATE_DNS;
+import static org.elasticsearch.cloud.aws.AwsEc2Service.DISCOVERY_EC2.HostType.PRIVATE_IP;
+import static org.elasticsearch.cloud.aws.AwsEc2Service.DISCOVERY_EC2.HostType.PUBLIC_DNS;
+import static org.elasticsearch.cloud.aws.AwsEc2Service.DISCOVERY_EC2.HostType.PUBLIC_IP;
 
 public class AwsEc2UnicastHostsProvider extends AbstractComponent implements UnicastHostsProvider {
 
@@ -66,7 +72,7 @@ public class AwsEc2UnicastHostsProvider extends AbstractComponent implements Uni
 
     private final Set<String> availabilityZones;
 
-    private final DISCOVERY_EC2.HostType hostType;
+    private final String hostType;
 
     private final DiscoNodesCache discoNodes;
 
@@ -149,19 +155,25 @@ public class AwsEc2UnicastHostsProvider extends AbstractComponent implements Uni
                 }
 
                 String address = null;
-                switch (hostType) {
-                    case PRIVATE_DNS:
-                        address = instance.getPrivateDnsName();
-                        break;
-                    case PRIVATE_IP:
-                        address = instance.getPrivateIpAddress();
-                        break;
-                    case PUBLIC_DNS:
-                        address = instance.getPublicDnsName();
-                        break;
-                    case PUBLIC_IP:
-                        address = instance.getPublicIpAddress();
-                        break;
+                if (hostType.equals(PRIVATE_DNS)) {
+                    address = instance.getPrivateDnsName();
+                } else if (hostType.equals(PRIVATE_IP)) {
+                    address = instance.getPrivateIpAddress();
+                } else if (hostType.equals(PUBLIC_DNS)) {
+                    address = instance.getPublicDnsName();
+                } else if (hostType.equals(PUBLIC_IP)) {
+                    address = instance.getPublicIpAddress();
+                } else if (hostType.startsWith(TAG_PREFIX)) {
+                    // Reading the node host from its metadata
+                    String tagName = hostType.substring(TAG_PREFIX.length());
+                    logger.debug("reading hostname from [{}] instance tag", tagName);
+                    List<Tag> tags = instance.getTags();
+                    for (Tag tag : tags) {
+                        if (tag.getKey().equals(tagName)) {
+                            address = tag.getValue();
+                            logger.debug("using [{}] as the instance address", address);
+                        }
+                    }
                 }
                 if (address != null) {
                     try {

plugins/discovery-ec2/src/test/java/org/elasticsearch/discovery/ec2/Ec2DiscoveryTests.java

@@ -41,6 +41,7 @@ import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 
+import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -51,6 +52,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.isOneOf;
 
 public class Ec2DiscoveryTests extends ESTestCase {
 
@@ -259,6 +261,37 @@ public class Ec2DiscoveryTests extends ESTestCase {
         assertThat(discoveryNodes, hasSize(prodInstances));
     }
 
+    public void testReadHostFromTag() throws InterruptedException, UnknownHostException {
+        int nodes = randomIntBetween(5, 10);
+
+        String[] addresses = new String[nodes];
+
+        for (int node = 0; node < nodes; node++) {
+            addresses[node] = "192.168.0." + (node + 1);
+            poorMansDNS.put("bar_" + node, new TransportAddress(InetAddress.getByName(addresses[node]), 9300));
+        }
+
+        Settings nodeSettings = Settings.builder()
+            .put(DISCOVERY_EC2.HOST_TYPE_SETTING.getKey(), "tag:foo")
+            .build();
+
+        List<List<Tag>> tagsList = new ArrayList<>();
+
+        for (int node = 0; node < nodes; node++) {
+            List<Tag> tags = new ArrayList<>();
+            tags.add(new Tag("foo", "bar_" + node));
+            tagsList.add(tags);
+        }
+
+        logger.info("started [{}] instances", nodes);
+        List<DiscoveryNode> discoveryNodes = buildDynamicNodes(nodeSettings, nodes, tagsList);
+        assertThat(discoveryNodes, hasSize(nodes));
+        for (DiscoveryNode discoveryNode : discoveryNodes) {
+            assertThat(discoveryNode.getHostName(), isOneOf(addresses));
+        }
+    }
+
+
     abstract class DummyEc2HostProvider extends AwsEc2UnicastHostsProvider {
         public int fetchCount = 0;
         public DummyEc2HostProvider(Settings settings, TransportService transportService, AwsEc2Service service) {