Configurable JsonPath for pod owner

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

@@ -6,7 +6,12 @@
 import org.jgroups.util.Util;
 
 import java.io.InputStream;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
 
 import static org.jgroups.protocols.kubernetes.Utils.openStream;
 import static org.jgroups.protocols.kubernetes.Utils.urlencode;
@@ -92,6 +97,37 @@ public List<Pod> getPods(String namespace, String labels, boolean dump_requests)
         return parseJsonResult(result, namespace, labels);
     }
 
+    /**
+     * get pod group during Rolling Update
+     * @param pod - json returned by k8s
+     * @return
+     */
+    String getPodGroup(Json pod) {
+        Json meta = Optional.ofNullable(pod.at("metadata")).orElse(null);
+        Json labels = Optional.ofNullable(meta)
+                .map(podMetadata -> podMetadata.at("labels"))
+                .orElse(null);
+        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
+            group = Optional.ofNullable(labels)
+                    .map(l -> l.at("deployment"))
+                    .map(Json::asString)
+                    .orElse(null);
+        }
+
+        log.debug("pod %s, group %s", Optional.ofNullable(meta)
+                .map(m -> m.at("name"))
+                .map(Json::asString)
+                .orElse(null), group);
+        return group;
+    }
+
     protected List<Pod> parseJsonResult(String input, String namespace, String labels) {
         if(input == null)
             return Collections.emptyList();
@@ -109,11 +145,8 @@ protected List<Pod> parseJsonResult(String input, String namespace, String label
         List<Json> items=json.at("items").asJsonList();
         List<Pod> pods=new ArrayList<>();
         for(Json obj: items) {
-            String parentDeployment = Optional.ofNullable(obj.at("metadata"))
-                  .map(podMetadata -> podMetadata.at("labels"))
-                  .map(podLabels -> podLabels.at("deployment"))
-                  .map(Json::asString)
-                  .orElse(null);
+            String parentDeployment = getPodGroup(obj);
+
             String name = Optional.ofNullable(obj.at("metadata"))
                   .map(podMetadata -> podMetadata.at("name"))
                   .map(Json::asString)
@@ -126,10 +159,10 @@ protected List<Pod> parseJsonResult(String input, String namespace, String label
                   .orElse(null);
             }
             boolean running = podRunning(podStatus);
-            if(podIP == null || !running) {
-                log.trace("Skipping pod %s since it's IP is %s or running is %s", name, podIP, Boolean.toString(running));
+            if(podIP == null) {
+                log.trace("Skipping pod %s since it's IP is %s", name, podIP);
             } else {
-                pods.add(new Pod(name, podIP, parentDeployment));
+                pods.add(new Pod(name, podIP, parentDeployment, running));
             }
         }
         log.trace("getPods(%s, %s) = %s", namespace, labels, pods);

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

@@ -108,6 +108,7 @@ public class KUBE_PING extends Discovery {
           " 'old' and 'new' during that process")
     protected boolean split_clusters_during_rolling_update;
 
+
     protected Client  client;
 
     protected int     tp_bind_port;
@@ -194,15 +195,18 @@ private boolean isPropertyDefined(String property_name) {
         super.destroy();
     }
 
+    private PhysicalAddress getCurrentPhysicalAddress(Address addr) {
+        return (PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, addr));
+    }
+
     public void findMembers(List<Address> members, boolean initial_discovery, Responses responses) {
         List<Pod>             hosts=readAll();
         List<PhysicalAddress> cluster_members=new ArrayList<>(hosts != null? hosts.size() : 16);
         PhysicalAddress       physical_addr=null;
         PingData              data=null;
 
         if(!use_ip_addrs || !initial_discovery) {
-            physical_addr=(PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, local_addr));
-
+            physical_addr = getCurrentPhysicalAddress(local_addr);
             // https://issues.jboss.org/browse/JGRP-1670
             data=new PingData(local_addr, false, NameCache.get(local_addr), physical_addr);
             if(members != null && members.size() <= max_members_in_discovery_request)
@@ -213,6 +217,8 @@ 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())
+                    continue;
                 for(int i=0; i <= port_range; i++) {
                     try {
                         IpAddress addr=new IpAddress(host.getIp(), tp_bind_port + i);
@@ -236,19 +242,21 @@ public void findMembers(List<Address> members, boolean initial_discovery, Respon
         if (split_clusters_during_rolling_update) {
             if(physical_addr != null) {
                 String senderIp = ((IpAddress)physical_addr).getIpAddress().getHostAddress();
-                String senderParentDeployment = hosts.stream()
+                // 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.
+                String senderPodGroup = hosts.stream()
                       .filter(pod -> senderIp.contains(pod.getIp()))
-                      .map(Pod::getParentDeployment)
+                      .map(Pod::getPodGroup)
                       .findFirst().orElse(null);
-                if(senderParentDeployment != null) {
+                if(senderPodGroup != null) {
                     Set<String> allowedAddresses = hosts.stream()
-                          .filter(pod -> senderParentDeployment.equals(pod.getParentDeployment()))
+                          .filter(pod -> senderPodGroup.equals(pod.getPodGroup()))
                           .map(Pod::getIp)
                           .collect(Collectors.toSet());
                     for(Iterator<PhysicalAddress> memberIterator = cluster_members.iterator(); memberIterator.hasNext();) {
                         IpAddress podAddress = (IpAddress) memberIterator.next();
                         if(!allowedAddresses.contains(podAddress.getIpAddress().getHostAddress())) {
-                            log.trace("removing pod %s from cluster members list since its parent domain is different than senders (%s). Allowed hosts: %s", podAddress, senderParentDeployment, allowedAddresses);
+                            log.trace("removing pod %s from cluster members list since its parent domain is different than senders (%s). Allowed hosts: %s", podAddress, senderPodGroup, allowedAddresses);
                             memberIterator.remove();
                         }
                     }

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

@@ -4,13 +4,19 @@ public class Pod {
 
    private final String name;
    private final String ip;
-   private final String parentDeployment;
+   private final String podGroup;    // name of group of Pods during Rolling Update. There is two groups: new pods and old pods
+   private final boolean isReady;
 
 
-   public Pod(String name, String ip, String parentDeployment) {
+   public Pod(String name, String ip, String podGroup, boolean isReady) {
       this.name = name;
       this.ip = ip;
-      this.parentDeployment = parentDeployment;
+      this.podGroup = podGroup;
+      this.isReady = isReady;
+   }
+
+   public Pod(String name, String ip, String podGroup) {
+      this(name, ip, podGroup, false);
    }
 
    public String getName() {
@@ -21,16 +27,20 @@ public String getIp() {
       return ip;
    }
 
-   public String getParentDeployment() {
-      return parentDeployment;
+   public String getPodGroup() {
+      return podGroup;
+   }
+
+   public boolean isReady() {
+      return isReady;
    }
 
    @Override
    public String toString() {
       return "Pod{" +
             "name='" + name + '\'' +
             ", ip='" + ip + '\'' +
-            ", parentDeployment='" + parentDeployment + '\'' +
+            ", podGroup='" + podGroup + '\'' +
             '}';
    }
 
@@ -43,14 +53,14 @@ public boolean equals(Object 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 parentDeployment != null ? parentDeployment.equals(pod.parentDeployment) : pod.parentDeployment == null;
+      return podGroup != null ? podGroup.equals(pod.podGroup) : pod.podGroup == null;
    }
 
    @Override
    public int hashCode() {
       int result = name != null ? name.hashCode() : 0;
       result = 31 * result + (ip != null ? ip.hashCode() : 0);
-      result = 31 * result + (parentDeployment != null ? parentDeployment.hashCode() : 0);
+      result = 31 * result + (podGroup != null ? podGroup.hashCode() : 0);
       return result;
    }
 }

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

@@ -37,4 +37,29 @@ public void testParsingPortWithoutNames() throws Exception {
         assertEquals(2, numberOfPods);
     }
 
+    @Test
+    public void testParsingPodGroupPodTemplateHash() throws Exception {
+        //given
+        Client client = new TestClient("/pods_without_ports.json");
+
+        //when
+        String podGroup = client.getPods(null, null, false).get(0).getPodGroup();
+
+        //then
+        assertEquals("infinispan-simple-tutorials-kubernetes-5", podGroup);
+    }
+
+    @Test
+    public void testParsingPodGroupOpenshift() throws Exception {
+        //given
+        Client client = new TestClient("/replicaset_rolling_update.json");
+
+        //when
+        String podGroup = client.getPods(null, null, false).get(0).getPodGroup();
+
+        //then
+        assertEquals("6569c544b", podGroup);
+    }
+
+
 }

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

@@ -48,22 +48,34 @@ public void testPuttingAllNodesInTheSameClusterDuringRollingUpdate() throws Exce
    }
 
    @Test
-   public void testPutOnlyNodesWithTheSameParentDuringRollingUpdate() throws Exception {
+   public void testPutOnlyNodesWithTheSameParentDuringRollingUpdateOpenShift() throws Exception {
       //given
       KUBE_PING_FOR_TESTING testedProtocol = new KUBE_PING_FOR_TESTING("/openshift_rolling_update.json");
       testedProtocol.setValue("split_clusters_during_rolling_update", true);
 
+      testPutOnlyNodesWithTheSameParentDuringRollingUpdate(testedProtocol);
+   }
+
+   @Test
+   public void testPutOnlyNodesWithTheSameParentDuringRollingUpdateReplicaSet() throws Exception {
+      //given
+      KUBE_PING_FOR_TESTING testedProtocol = new KUBE_PING_FOR_TESTING("/replicaset_rolling_update.json");
+      testedProtocol.setValue("split_clusters_during_rolling_update", true);
+      testPutOnlyNodesWithTheSameParentDuringRollingUpdate(testedProtocol);
+   }
+
+   private void testPutOnlyNodesWithTheSameParentDuringRollingUpdate(KUBE_PING_FOR_TESTING testedProtocol) throws Exception {
       //when
       sendInitialDiscovery(testedProtocol);
       String senderParentDeployment = testedProtocol.getPods().stream()
             .filter(pod -> "127.0.0.1".equals(pod.getIp()))
-            .map(Pod::getParentDeployment)
+            .map(Pod::getPodGroup)
             .findFirst().get();
       Set<String> membersUsedForDiscovery = testedProtocol.getCollectedMessages().stream()
-            .map(e -> ((IpAddress)e.getDest()).getIpAddress().getHostAddress())
-            .collect(Collectors.toSet());
+              .map(e -> ((IpAddress)e.getDest()).getIpAddress().getHostAddress())
+              .collect(Collectors.toSet());
       List<String> allowedPodsFromKubernetesApi = testedProtocol.getPods().stream()
-            .filter(pod -> senderParentDeployment.equals(pod.getParentDeployment()))
+            .filter(pod -> senderParentDeployment.equals(pod.getPodGroup()))
             .map(Pod::getIp)
             .collect(Collectors.toList());
 

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

@@ -1,14 +1,14 @@
 package org.jgroups.ping.kube.test;
 
-import static org.junit.Assert.assertEquals;
-
 import java.util.List;
 
 import org.jgroups.protocols.kubernetes.Client;
 import org.jgroups.protocols.kubernetes.Pod;
 import org.junit.Assert;
 import org.junit.Test;
 
+import static org.junit.Assert.*;
+
 /**
  * @author <a href="mailto:ulrich.romahn@gmail.com">Ulrich Romahn</a>
  */
@@ -30,7 +30,10 @@ public void testOnePodNotRunning() throws Exception {
         Client client = new TestClient(jsonFile);
         List<Pod> pods = client.getPods(null, null, false);
         Assert.assertNotNull(pods);
-        assertEquals(2, pods.size());
+        assertEquals(3, pods.size());
+        assertTrue(pods.get(0).isReady());
+        assertTrue(pods.get(1).isReady());
+        assertFalse(pods.get(2).isReady());
         String pod = pods.get(0).getIp();
         Assert.assertNotNull(pod);
     }