Merge pull request #76 from slaskawi/fix_split_during_rolling_upgrade

Fix splitting cluster on Rolling Update

Author: rhusar

README.adoc

@@ -169,7 +169,9 @@ namespace `"default"` will be returned.
 
 | dump_requests || Dumps all discovery requests and responses to the Kubernetes server to stdout when true
 
-| split_clusters_during_rolling_update || During the Rolling Update, prevents from putting all Pods into a single cluster
+| split_clusters_during_rolling_update | KUBERNETES_SPLIT_CLUSTERS_DURING_ROLLING_UPDATE | During the Rolling Update, prevents from putting all Pods into a single cluster
+
+| useNotReadyAddresses | KUBERNETES_USE_NOT_READY_ADDRESSES | True if initial discovery should take unready Pods into consideration. Default is `true`.
 
 |===============
 

src/main/java/org/jgroups/protocols/kubernetes/Client.java

@@ -91,7 +91,7 @@ protected String fetchFromKubernetes(String op, String namespace, String labels,
 
 
     public List<Pod> getPods(String namespace, String labels, boolean dump_requests) throws Exception {
-        String result=fetchFromKubernetes("pods", namespace, labels, dump_requests);
+        String result = fetchFromKubernetes("pods", namespace, labels, dump_requests);
         if(result == null)
             return Collections.emptyList();
         return parseJsonResult(result, namespace, labels);
@@ -107,20 +107,29 @@ String getPodGroup(Json pod) {
         Json labels = Optional.ofNullable(meta)
                 .map(podMetadata -> podMetadata.at("labels"))
                 .orElse(null);
+
+        // This works for Deployment Config
         String group = Optional.ofNullable(labels)
                 .map(l -> l.at("pod-template-hash"))
                 .map(Json::asString)
                 .orElse(null);
 
         if (group == null) {
-            log.warn("metadata.labels.pod-template-hash not found in pod json. Impossible to reliably determine pod group during Rolling Update");
-            // keep backward-compatible behavior
+            // Ok, maybe, it's a Deployment and has a valid deployment flag?
             group = Optional.ofNullable(labels)
                     .map(l -> l.at("deployment"))
                     .map(Json::asString)
                     .orElse(null);
         }
 
+        if (group == null) {
+            // Final check, maybe it's a StatefulSet?
+            group = Optional.ofNullable(labels)
+                    .map(l -> l.at("controller-revision-hash"))
+                    .map(Json::asString)
+                    .orElse(null);
+        }
+
         log.debug("pod %s, group %s", Optional.ofNullable(meta)
                 .map(m -> m.at("name"))
                 .map(Json::asString)
@@ -146,7 +155,6 @@ protected List<Pod> parseJsonResult(String input, String namespace, String label
         List<Pod> pods=new ArrayList<>();
         for(Json obj: items) {
             String parentDeployment = getPodGroup(obj);
-
             String name = Optional.ofNullable(obj.at("metadata"))
                   .map(podMetadata -> podMetadata.at("name"))
                   .map(Json::asString)

src/main/java/org/jgroups/protocols/kubernetes/KUBE_PING.java

@@ -105,9 +105,12 @@ public class KUBE_PING extends Discovery {
 
     @Property(description="The standard behavior during Rolling Update is to put all Pods in the same cluster. In" +
           " cases (application level incompatibility) this causes problems. One might decide to split clusters to" +
-          " 'old' and 'new' during that process")
+          " 'old' and 'new' during that process", systemProperty="KUBERNETES_SPLIT_CLUSTERS_DURING_ROLLING_UPDATE")
     protected boolean split_clusters_during_rolling_update;
 
+    @Property(description="Introduces similar behaviour to Kubernetes Services (using DNS) with publishNotReadyAddresses set to true. " +
+            "By default it's true", systemProperty="KUBERNETES_USE_NOT_READY_ADDRESSES")
+    protected boolean useNotReadyAddresses = true;
 
     protected Client  client;
 
@@ -217,7 +220,7 @@ public void findMembers(List<Address> members, boolean initial_discovery, Respon
             if(log.isTraceEnabled())
                 log.trace("%s: hosts fetched from Kubernetes: %s", local_addr, hosts);
             for(Pod host: hosts) {
-                if (!host.isReady())
+                if (!host.isReady() && !useNotReadyAddresses)
                     continue;
                 for(int i=0; i <= port_range; i++) {
                     try {
@@ -243,7 +246,11 @@ public void findMembers(List<Address> members, boolean initial_discovery, Respon
             if(physical_addr != null) {
                 String senderIp = ((IpAddress)physical_addr).getIpAddress().getHostAddress();
                 // Please note we search for sender parent group through all pods, ever not ready. It's because JGroup discovery is performed
-                // before Wildfly can respond to http liveness probe.
+                // before WildFly can respond to http readiness probe.
+                hosts.stream()
+                        .filter(p -> p.getPodGroup() == null)
+                        .forEach(p -> log.warn("Pod %s doesn't have group assigned. Impossible to reliably determine pod group during Rolling Update."));
+
                 String senderPodGroup = hosts.stream()
                       .filter(pod -> senderIp.contains(pod.getIp()))
                       .map(Pod::getPodGroup)

src/main/java/org/jgroups/protocols/kubernetes/Pod.java

@@ -1,5 +1,7 @@
 package org.jgroups.protocols.kubernetes;
 
+import java.util.Objects;
+
 public class Pod {
 
    private final String name;
@@ -15,10 +17,6 @@ public Pod(String name, String ip, String podGroup, boolean isReady) {
       this.isReady = isReady;
    }
 
-   public Pod(String name, String ip, String podGroup) {
-      this(name, ip, podGroup, false);
-   }
-
    public String getName() {
       return name;
    }
@@ -51,9 +49,9 @@ public boolean equals(Object o) {
 
       Pod pod = (Pod) o;
 
-      if (name != null ? !name.equals(pod.name) : pod.name != null) return false;
-      if (ip != null ? !ip.equals(pod.ip) : pod.ip != null) return false;
-      return podGroup != null ? podGroup.equals(pod.podGroup) : pod.podGroup == null;
+      if (!Objects.equals(name, pod.name)) return false;
+      if (!Objects.equals(ip, pod.ip)) return false;
+      return Objects.equals(podGroup, pod.podGroup);
    }
 
    @Override

src/test/java/org/jgroups/ping/kube/test/RollingUpdateTest.java

@@ -53,6 +53,7 @@ public void testPutOnlyNodesWithTheSameParentDuringRollingUpdateOpenShift() thro
       KUBE_PING_FOR_TESTING testedProtocol = new KUBE_PING_FOR_TESTING("/openshift_rolling_update.json");
       testedProtocol.setValue("split_clusters_during_rolling_update", true);
 
+      //when //then
       testPutOnlyNodesWithTheSameParentDuringRollingUpdate(testedProtocol);
    }
 
@@ -61,6 +62,18 @@ public void testPutOnlyNodesWithTheSameParentDuringRollingUpdateReplicaSet() thr
       //given
       KUBE_PING_FOR_TESTING testedProtocol = new KUBE_PING_FOR_TESTING("/replicaset_rolling_update.json");
       testedProtocol.setValue("split_clusters_during_rolling_update", true);
+
+      //when //then
+      testPutOnlyNodesWithTheSameParentDuringRollingUpdate(testedProtocol);
+   }
+
+   @Test
+   public void testPutOnlyNodesWithTheSameParentDuringRollingUpdateStatefulSet() throws Exception {
+      //given
+      KUBE_PING_FOR_TESTING testedProtocol = new KUBE_PING_FOR_TESTING("/statefulset_rolling_update.json");
+      testedProtocol.setValue("split_clusters_during_rolling_update", true);
+
+      //when //then
       testPutOnlyNodesWithTheSameParentDuringRollingUpdate(testedProtocol);
    }