Repository: hazelcast/hazelcast-kubernetes
Pull request #57 (2017-11-30 12:58:07)

Add configurable support for add endpoints of pods not still ready on kubernetes
Configurable version of feature commented on https://github.com/hazelcast/hazelcast-kubernetes/pull/37


Patch:
diff --git a/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java b/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java
index fd6fb09..081be98 100644
--- a/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java
+++ b/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java
@@ -37,6 +37,7 @@ import static com.hazelcast.kubernetes.KubernetesProperties.SERVICE_DNS_TIMEOUT;
 import static com.hazelcast.kubernetes.KubernetesProperties.SERVICE_LABEL_NAME;
 import static com.hazelcast.kubernetes.KubernetesProperties.SERVICE_LABEL_VALUE;
 import static com.hazelcast.kubernetes.KubernetesProperties.SERVICE_NAME;
+import static com.hazelcast.kubernetes.KubernetesProperties.RESOLVE_NOT_READY_ADDRESSES;
 
 final class HazelcastKubernetesDiscoveryStrategy
         extends AbstractDiscoveryStrategy {
@@ -56,9 +57,10 @@ final class HazelcastKubernetesDiscoveryStrategy
         String serviceName = getOrNull(properties, KUBERNETES_SYSTEM_PREFIX, SERVICE_NAME);
         String serviceLabel = getOrNull(properties, KUBERNETES_SYSTEM_PREFIX, SERVICE_LABEL_NAME);
         String serviceLabelValue = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, SERVICE_LABEL_VALUE, "true");
-        String apiToken = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, KUBERNETES_API_TOKEN, null);
         String namespace = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, NAMESPACE, getNamespaceOrDefault());
+        Boolean resolveNotReadyAddresses = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, RESOLVE_NOT_READY_ADDRESSES, false);
         String kubernetesMaster = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, KUBERNETES_MASTER_URL, DEFAULT_MASTER_URL);
+        String apiToken = getOrDefault(properties, KUBERNETES_SYSTEM_PREFIX, KUBERNETES_API_TOKEN, null);
 
         logger.info("Kubernetes Discovery properties: { "
                 + "service-dns: " + serviceDns + ", "
@@ -66,14 +68,16 @@ final class HazelcastKubernetesDiscoveryStrategy
                 + "service-name: " + serviceName + ", "
                 + "service-label: " + serviceLabel + ", "
                 + "service-label-value: " + serviceLabelValue + ", "
-                + "namespace: " + namespace + ", " + "kubernetes-master: " + kubernetesMaster + "}");
+                + "namespace: " + namespace + ", "
+                + "resolve-not-ready-addresses: " + resolveNotReadyAddresses + ", "
+                + "kubernetes-master: " + kubernetesMaster + "}");
 
         EndpointResolver endpointResolver;
         if (serviceDns != null) {
             endpointResolver = new DnsEndpointResolver(logger, serviceDns, serviceDnsTimeout);
         } else {
-            endpointResolver = new ServiceEndpointResolver(logger, serviceName, serviceLabel, serviceLabelValue, namespace,
-                    kubernetesMaster, apiToken);
+            endpointResolver = new ServiceEndpointResolver(logger, serviceName, serviceLabel, serviceLabelValue,
+                    namespace, resolveNotReadyAddresses, kubernetesMaster, apiToken);
         }
         logger.info("Kubernetes Discovery activated resolver: " + endpointResolver.getClass().getSimpleName());
         this.endpointResolver = endpointResolver;
diff --git a/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyFactory.java b/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyFactory.java
index 1af96f8..eb3e9c5 100644
--- a/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyFactory.java
+++ b/src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategyFactory.java
@@ -39,9 +39,10 @@ public class HazelcastKubernetesDiscoveryStrategyFactory
         PROPERTY_DEFINITIONS = Collections.unmodifiableCollection(Arrays.asList(
                 KubernetesProperties.SERVICE_DNS,
                 KubernetesProperties.SERVICE_NAME,
-                KubernetesProperties.NAMESPACE,
                 KubernetesProperties.SERVICE_LABEL_NAME,
                 KubernetesProperties.SERVICE_LABEL_VALUE,
+                KubernetesProperties.NAMESPACE,
+                KubernetesProperties.RESOLVE_NOT_READY_ADDRESSES,
                 KubernetesProperties.KUBERNETES_MASTER_URL,
                 KubernetesProperties.KUBERNETES_API_TOKEN));
     }
diff --git a/src/main/java/com/hazelcast/kubernetes/KubernetesProperties.java b/src/main/java/com/hazelcast/kubernetes/KubernetesProperties.java
index a892171..02209d5 100644
--- a/src/main/java/com/hazelcast/kubernetes/KubernetesProperties.java
+++ b/src/main/java/com/hazelcast/kubernetes/KubernetesProperties.java
@@ -22,6 +22,7 @@ import com.hazelcast.core.TypeConverter;
 
 import static com.hazelcast.config.properties.PropertyTypeConverter.INTEGER;
 import static com.hazelcast.config.properties.PropertyTypeConverter.STRING;
+import static com.hazelcast.config.properties.PropertyTypeConverter.BOOLEAN;
 
 /**
  * <p>Configuration class of the Hazelcast Discovery Plugin for <a href="http://kubernetes.io">Kubernetes</a>.</p>
@@ -84,6 +85,12 @@ public final class KubernetesProperties {
      */
     public static final PropertyDefinition NAMESPACE = property("namespace", STRING);
 
+    /**
+     *  <p>Configuration key: <tt>resolve-notready-addresses</tt></p>
+     *  Defines if not ready addresses should be evaluated to be discovered on startup.
+     */
+    public static final PropertyDefinition RESOLVE_NOT_READY_ADDRESSES = property("resolve-not-ready-addresses", BOOLEAN);
+
     /**
      * <p>Configuration key: <tt>kubernetes-master</tt></p>
      * Defines an alternative address for the kubernetes master. Defaults to: <tt>https://kubernetes.default.svc</tt>
diff --git a/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java b/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java
index 00c1ae4..7a42676 100644
--- a/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java
+++ b/src/main/java/com/hazelcast/kubernetes/ServiceEndpointResolver.java
@@ -49,11 +49,12 @@ class ServiceEndpointResolver extends HazelcastKubernetesDiscoveryStrategy.Endpo
     private final String serviceLabel;
     private final String serviceLabelValue;
     private final String namespace;
+    private final Boolean resolveNotReadyAddresses;
 
     private final KubernetesClient client;
 
     ServiceEndpointResolver(ILogger logger, String serviceName, String serviceLabel, String serviceLabelValue,
-            String namespace, String kubernetesMaster, String apiToken) {
+                            String namespace, Boolean resolveNotReadyAddresses, String kubernetesMaster, String apiToken) {
 
         super(logger);
 
@@ -61,6 +62,7 @@ class ServiceEndpointResolver extends HazelcastKubernetesDiscoveryStrategy.Endpo
         this.namespace = namespace;
         this.serviceLabel = serviceLabel;
         this.serviceLabelValue = serviceLabelValue;
+        this.resolveNotReadyAddresses = resolveNotReadyAddresses;
         this.client = buildKubernetesClient(apiToken, kubernetesMaster);
     }
 
@@ -112,22 +114,39 @@ class ServiceEndpointResolver extends HazelcastKubernetesDiscoveryStrategy.Endpo
         }
         List<DiscoveryNode> discoveredNodes = new ArrayList<DiscoveryNode>();
         for (EndpointSubset endpointSubset : endpointsSubsets) {
-            List<EndpointAddress> endpointAddresses = endpointSubset.getAddresses();
-            if (endpointAddresses == null) {
-                continue;
-            }
-            for (EndpointAddress endpointAddress : endpointAddresses) {
-                Map<String, Object> properties = endpointAddress.getAdditionalProperties();
-                String ip = endpointAddress.getIp();
-                InetAddress inetAddress = mapAddress(ip);
-                int port = getServicePort(properties);
-                Address address = new Address(inetAddress, port);
-                discoveredNodes.add(new SimpleDiscoveryNode(address, properties));
-            }
+            resolveNotReadyAddresses(discoveredNodes, endpointSubset);
+            resolveAddresses(discoveredNodes, endpointSubset);
         }
         return discoveredNodes;
     }
 
+    private void resolveNotReadyAddresses(List<DiscoveryNode> discoveredNodes, EndpointSubset endpointSubset) {
+        if (Boolean.TRUE.equals(resolveNotReadyAddresses)) {
+            for (EndpointAddress endpointAddress : endpointSubset.getNotReadyAddresses()) {
+                addAddress(discoveredNodes, endpointAddress);
+            }
+        }
+    }
+
+    private void resolveAddresses(List<DiscoveryNode> discoveredNodes, EndpointSubset endpointSubset) {
+        List<EndpointAddress> endpointAddresses = endpointSubset.getAddresses();
+        if (endpointAddresses == null) {
+            return;
+        }
+        for (EndpointAddress endpointAddress : endpointAddresses) {
+            addAddress(discoveredNodes, endpointAddress);
+        }
+    }
+
+    private void addAddress(List<DiscoveryNode> discoveredNodes, EndpointAddress endpointAddress) {
+        Map<String, Object> properties = endpointAddress.getAdditionalProperties();
+        String ip = endpointAddress.getIp();
+        InetAddress inetAddress = mapAddress(ip);
+        int port = getServicePort(properties);
+        Address address = new Address(inetAddress, port);
+        discoveredNodes.add(new SimpleDiscoveryNode(address, properties));
+    }
+
     @Override
     void destroy() {
         client.close();


Test patch:
diff --git a/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java b/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java
index af7f512..a97e05f 100644
--- a/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java
+++ b/src/test/java/com/hazelcast/kubernetes/ServiceEndpointResolverTest.java
@@ -12,7 +12,6 @@ import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.fabric8.kubernetes.client.dsl.base.BaseOperation;
 import org.junit.Assert;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
@@ -41,6 +40,7 @@ public class ServiceEndpointResolverTest {
     private static final String SERVICE_NAME = "";
     private static final String SERVICE_LABEL = "theLabel";
     private static final String SERVICE_LABEL_VALUE = "serviceLabelValue";
+    private static final Boolean RESOLVE_NOT_READY_ADDRESSES = true;
     private static final String NAMESPACE = "theNamespace";
     private static final String KUBERNETES_MASTER_URL = "http://bla";
     private static final String API_TOKEN = "token";
@@ -73,7 +73,7 @@ public class ServiceEndpointResolverTest {
 
     @Test
     public void resolveWithNamespaceAndNoNodeInNamespace() {
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, KUBERNETES_MASTER_URL, API_TOKEN);
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
         List<DiscoveryNode> nodes = sut.resolve();
 
         assertEquals(0, nodes.size());
@@ -84,7 +84,7 @@ public class ServiceEndpointResolverTest {
         Endpoints discoveryNode = createEndpoints(1);
         nodesInNamespace.getItems().add(discoveryNode);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, KUBERNETES_MASTER_URL, API_TOKEN);
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
         List<DiscoveryNode> nodes = sut.resolve();
 
         assertEquals(1, nodes.size());
@@ -97,7 +97,7 @@ public class ServiceEndpointResolverTest {
         discoveryNode.getSubsets().get(0).setAddresses(null);
         nodesInNamespace.getItems().add(discoveryNode);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, KUBERNETES_MASTER_URL, API_TOKEN);
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
         List<DiscoveryNode> nodes = sut.resolve();
 
         assertEquals(0, nodes.size());
@@ -109,7 +109,7 @@ public class ServiceEndpointResolverTest {
         discoveryNode.setSubsets(null);
         nodesInNamespace.getItems().add(discoveryNode);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, KUBERNETES_MASTER_URL, API_TOKEN);
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
         List<DiscoveryNode> nodes = sut.resolve();
 
         assertEquals(0, nodes.size());
@@ -119,7 +119,7 @@ public class ServiceEndpointResolverTest {
     public void resolveWithServiceLabelAndNodeInNamespace() {
         nodesInNamespace.getItems().add(createEndpoints(1));
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, SERVICE_LABEL, SERVICE_LABEL_VALUE, NAMESPACE, KUBERNETES_MASTER_URL, API_TOKEN);
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, SERVICE_LABEL, SERVICE_LABEL_VALUE, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
         List<DiscoveryNode> nodes = sut.resolve();
 
         assertEquals(0, nodes.size());
@@ -131,13 +131,24 @@ public class ServiceEndpointResolverTest {
         Endpoints discoveryNode = createEndpoints(2);
         nodesWithLabel.getItems().add(discoveryNode);
 
-        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, SERVICE_LABEL, SERVICE_LABEL_VALUE, NAMESPACE, KUBERNETES_MASTER_URL, API_TOKEN);
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, SERVICE_LABEL, SERVICE_LABEL_VALUE, NAMESPACE, null, KUBERNETES_MASTER_URL, API_TOKEN);
         List<DiscoveryNode> nodes = sut.resolve();
 
         assertEquals(1, nodes.size());
         assertEquals(2, nodes.get(0).getPrivateAddress().getPort());
     }
 
+    @Test
+    public void resolveWithNamespaceAndNotReadyAddresses() {
+        nodesInNamespace.getItems().add(createEndpoints(1));
+        nodesInNamespace.getItems().add(createNotReadyEndpoints(2));
+
+        ServiceEndpointResolver sut = new ServiceEndpointResolver(LOGGER, SERVICE_NAME, null, null, NAMESPACE, RESOLVE_NOT_READY_ADDRESSES, KUBERNETES_MASTER_URL, API_TOKEN);
+        List<DiscoveryNode> nodes = sut.resolve();
+
+        assertEquals(2, nodes.size());
+    }
+
     @Test
     public void testReadFileContents() throws IOException {
         String expectedContents = "Hello, world!\nThis is a test with Unicode ✓.";
@@ -169,4 +180,15 @@ public class ServiceEndpointResolverTest {
         address.getAdditionalProperties().put("hazelcast-service-port", String.valueOf(id));
         return endpoints;
     }
+
+    private Endpoints createNotReadyEndpoints(int id) {
+        Endpoints endpoints = new Endpoints();
+        EndpointSubset subset = new EndpointSubset();
+        endpoints.getSubsets().add(subset);
+        EndpointAddress address = new EndpointAddress();
+        subset.getNotReadyAddresses().add(address);
+        address.setIp("1.1.1.1");
+        address.getAdditionalProperties().put("hazelcast-service-port", String.valueOf(id));
+        return endpoints;
+    }
 }
\ No newline at end of file