Introduced split_clusters_during_rolling_update

- Fixes #33 (rolling upgrades)
- Set version to 1.0.3-SNAPSHOT, included Seb's changes (issue #33)
- Fix to #33

Author: slaskawi

README.adoc

@@ -167,6 +167,8 @@ will be thrown
 
 | 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
+
 |===============
 
 
@@ -312,4 +314,4 @@ The commands for running on https://cloud.google.com/container-engine/docs/[Goog
 as when running locally in https://github.com/kubernetes/minikube[minikube].
 
 The only difference is that on GKE, contrary to minikube, IP multicasting is not available. This means that the `probe.sh`
-command has to be run as `probe.sh -addr localhost` instead of simply running `probe.sh`.
+command has to be run as `probe.sh -addr localhost` instead of simply running `probe.sh`.

pom.xml

@@ -3,7 +3,7 @@
 
     <groupId>org.jgroups.kubernetes</groupId>
     <artifactId>jgroups-kubernetes</artifactId>
-    <version>1.0.2-SNAPSHOT</version>
+    <version>1.0.4-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>KUBE_PING</name>
     <url>https://github.com/jgroups-extras/jgroups-kubernetes</url>
@@ -56,6 +56,7 @@
         <version.jgroups>4.0.4.Final</version.jgroups>
         <version.oauth>20100527</version.oauth>
         <version.junit>4.12</version.junit>
+        <version.assertj>3.8.0</version.assertj>
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
 
@@ -146,6 +147,12 @@
             <version>${version.junit}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${version.assertj}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -71,17 +71,18 @@ protected String fetchFromKubernetes(String op, String namespace, String labels,
 
 
 
-    public List<String> getPods(String namespace, String labels, boolean dump_requests) throws Exception {
+    public List<Pod> getPods(String namespace, String labels, boolean dump_requests) throws Exception {
         String result=fetchFromKubernetes("pods", namespace, labels, dump_requests);
         if(result == null)
             return Collections.emptyList();
         return parseJsonResult(result, namespace, labels);
     }
 
-    protected List<String> parseJsonResult(String input, String namespace, String labels) {
+    protected List<Pod> parseJsonResult(String input, String namespace, String labels) {
         if(input == null)
             return Collections.emptyList();
         Json json=Json.read(input);
+
         if(json == null || !json.isObject()) {
             log.error("JSON is not a map: %s", json);
             return Collections.emptyList();
@@ -92,22 +93,28 @@ protected List<String> parseJsonResult(String input, String namespace, String la
             return Collections.emptyList();
         }
         List<Json> items=json.at("items").asJsonList();
-        List<String> pods=new ArrayList<>();
+        List<Pod> pods=new ArrayList<>();
         for(Json obj: items) {
-            if(obj.isObject() && obj.has("status")) {
-                Json status=obj.at("status");
-                if(status.isObject() && status.has("podIP")) {
-                    String podIP=status.at("podIP").asString();
-                    if(status.has("phase")) {
-                        Json phase=status.at("phase");
-                        if(phase != null && phase.isString() && !"Running".equals(phase.asString())) {
-                            log.trace("skipped pod with IP=%s as it is not running (%s)", podIP, phase);
-                            continue;
-                        }
-                    }
-                    if(!pods.contains(podIP))
-                        pods.add(podIP);
-                }
+            String parentDeployment = Optional.ofNullable(obj.at("metadata"))
+                  .map(podMetadata -> podMetadata.at("labels"))
+                  .map(podLabels -> podLabels.at("deployment"))
+                  .map(Json::asString)
+                  .orElse(null);
+            String name = Optional.ofNullable(obj.at("metadata"))
+                  .map(podMetadata -> podMetadata.at("name"))
+                  .map(Json::asString)
+                  .orElse(null);
+            String podIP = Optional.ofNullable(obj.at("status"))
+                  .map(podStatus -> podStatus.at("podIP"))
+                  .map(Json::asString)
+                  .orElse(null);
+            if(podIP == null) {
+                //Previously we did checks on phase. But from my observations, it is extremely rare to have a container
+                //listed by Kubernetes API with any other status but Running (I might imagine it will hang in scheduled).
+                //However in both cases, its IP address will be null. So it is much better to stick to that.
+                log.trace("Skipping pod %s since it has no IP %s", name, podIP);
+            } else {
+                pods.add(new Pod(name, podIP, parentDeployment));
             }
         }
         log.trace("getPods(%s, %s) = %s", namespace, labels, pods);

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

@@ -101,6 +101,11 @@ public class KUBE_PING extends Discovery {
     @Property(description="Dumps all discovery requests and responses to the Kubernetes server to stdout when true")
     protected boolean dump_requests;
 
+    @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")
+    protected boolean split_clusters_during_rolling_update;
+
     protected Client  client;
 
     protected int     tp_bind_port;
@@ -168,7 +173,7 @@ public void init() throws Exception {
     }
 
     public void findMembers(List<Address> members, boolean initial_discovery, Responses responses) {
-        List<String>          hosts=readAll();
+        List<Pod>             hosts=readAll();
         List<PhysicalAddress> cluster_members=new ArrayList<>(hosts != null? hosts.size() : 16);
         PhysicalAddress       physical_addr=null;
         PingData              data=null;
@@ -185,10 +190,10 @@ public void findMembers(List<Address> members, boolean initial_discovery, Respon
         if(hosts != null) {
             if(log.isTraceEnabled())
                 log.trace("%s: hosts fetched from Kubernetes: %s", local_addr, hosts);
-            for(String host: hosts) {
+            for(Pod host: hosts) {
                 for(int i=0; i <= port_range; i++) {
                     try {
-                        IpAddress addr=new IpAddress(host, tp_bind_port + i);
+                        IpAddress addr=new IpAddress(host.getIp(), tp_bind_port + i);
                         if(!cluster_members.contains(addr))
                             cluster_members.add(addr);
                     }
@@ -206,6 +211,33 @@ public void findMembers(List<Address> members, boolean initial_discovery, Respon
                 list.stream().filter(phys_addr -> !cluster_members.contains(phys_addr)).forEach(cluster_members::add);
         }
 
+        if (split_clusters_during_rolling_update) {
+            if(physical_addr != null) {
+                String senderIp = ((IpAddress)physical_addr).getIpAddress().getHostAddress();
+                String senderParentDeployment = hosts.stream()
+                      .filter(pod -> senderIp.contains(pod.getIp()))
+                      .map(Pod::getParentDeployment)
+                      .findFirst().orElse(null);
+                if(senderParentDeployment != null) {
+                    Set<String> allowedAddresses = hosts.stream()
+                          .filter(pod -> senderParentDeployment.equals(pod.getParentDeployment()))
+                          .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);
+                            memberIterator.remove();
+                        }
+                    }
+                } else {
+                    log.warn("split_clusters_during_rolling_update is set to 'true' but can't obtain local node parent deployment. All nodes will be placed in the same cluster.");
+                }
+            } else {
+                log.warn("split_clusters_during_rolling_update is set to 'true' but can't obtain local node IP address. All nodes will be placed in the same cluster.");
+            }
+        }
+
         if(log.isTraceEnabled())
             log.trace("%s: sending discovery requests to %s", local_addr, cluster_members);
         PingHeader hdr=new PingHeader(PingHeader.GET_MBRS_REQ).clusterName(cluster_name).initialDiscovery(initial_discovery);
@@ -229,12 +261,12 @@ public void findMembers(List<Address> members, boolean initial_discovery, Respon
 
     @ManagedOperation(description="Asks Kubernetes for the IP addresses of all pods")
     public String fetchFromKube() {
-        List<String> list=readAll();
-        return list.stream().collect(Collectors.joining(", "));
+        List<Pod> list=readAll();
+        return list.toString();
     }
 
 
-    protected List<String> readAll() {
+    protected List<Pod> readAll() {
         if(isClusteringEnabled() && client != null) {
             try {
                 return client.getPods(namespace, labels, dump_requests);

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

@@ -0,0 +1,56 @@
+package org.jgroups.protocols.kubernetes;
+
+public class Pod {
+
+   private final String name;
+   private final String ip;
+   private final String parentDeployment;
+
+
+   public Pod(String name, String ip, String parentDeployment) {
+      this.name = name;
+      this.ip = ip;
+      this.parentDeployment = parentDeployment;
+   }
+
+   public String getName() {
+      return name;
+   }
+
+   public String getIp() {
+      return ip;
+   }
+
+   public String getParentDeployment() {
+      return parentDeployment;
+   }
+
+   @Override
+   public String toString() {
+      return "Pod{" +
+            "name='" + name + '\'' +
+            ", ip='" + ip + '\'' +
+            ", parentDeployment='" + parentDeployment + '\'' +
+            '}';
+   }
+
+   @Override
+   public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+
+      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 parentDeployment != null ? parentDeployment.equals(pod.parentDeployment) : pod.parentDeployment == 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);
+      return result;
+   }
+}

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

@@ -1,13 +1,14 @@
 
 package org.jgroups.ping.kube.test;
 
-import org.jgroups.protocols.kubernetes.Client;
-import org.junit.Assert;
-import org.junit.Test;
+import static org.junit.Assert.assertEquals;
 
 import java.util.List;
 
-import static org.junit.Assert.assertEquals;
+import org.jgroups.protocols.kubernetes.Client;
+import org.jgroups.protocols.kubernetes.Pod;
+import org.junit.Assert;
+import org.junit.Test;
 
 /**
  * @author <a href="mailto:[email protected]">Ales Justin</a>
@@ -17,10 +18,10 @@ public class ClientTest {
     @Test
     public void testPods() throws Exception {
         Client client = new TestClient();
-        List<String> pods = client.getPods(null, null, false);
+        List<Pod> pods = client.getPods(null, null, false);
         Assert.assertNotNull(pods);
         assertEquals(2, pods.size());
-        String pod = pods.get(0);
+        String pod = pods.get(0).getIp();
         Assert.assertNotNull(pod);
     }
 

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

@@ -0,0 +1,114 @@
+package org.jgroups.ping.kube.test;
+
+import static org.jgroups.ping.kube.test.util.FreePortFinder.findFreePort;
+
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.assertj.core.api.Assertions;
+import org.jgroups.JChannel;
+import org.jgroups.Message;
+import org.jgroups.protocols.TCP;
+import org.jgroups.protocols.kubernetes.KUBE_PING;
+import org.jgroups.protocols.kubernetes.Pod;
+import org.jgroups.protocols.pbcast.GMS;
+import org.jgroups.protocols.pbcast.NAKACK2;
+import org.jgroups.stack.IpAddress;
+import org.junit.Test;
+
+/**
+ * Tests Rolling update scenarios.
+ *
+ * <p>
+ *    The idea of this tests is to mock the Kubernetes API response and check which hosts were queried during
+ *    initial discovery. This way we will know which hosts where put into the cluster.
+ * </p>
+ */
+public class RollingUpdateTest {
+
+   @Test
+   public void testPuttingAllNodesInTheSameClusterDuringRollingUpdate() throws Exception {
+      //given
+      KUBE_PING_FOR_TESTING testedProtocol = new KUBE_PING_FOR_TESTING("/openshift_rolling_update.json");
+
+      //when
+      sendInitialDiscovery(testedProtocol);
+      Set<String> membersUsedForDiscovery = testedProtocol.getCollectedMessages().stream()
+            .map(e -> ((IpAddress)e.getDest()).getIpAddress().getHostAddress())
+            .collect(Collectors.toSet());
+      List<String> allPodsFromKubernetesApi = testedProtocol.getPods().stream()
+            .map(pod -> pod.getIp())
+            .collect(Collectors.toList());
+
+      //then
+      Assertions.assertThat(membersUsedForDiscovery).hasSameElementsAs(allPodsFromKubernetesApi);
+   }
+
+   @Test
+   public void testPutOnlyNodesWithTheSameParentDuringRollingUpdate() 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);
+
+      //when
+      sendInitialDiscovery(testedProtocol);
+      String senderParentDeployment = testedProtocol.getPods().stream()
+            .filter(pod -> "127.0.0.1".equals(pod.getIp()))
+            .map(pod -> pod.getParentDeployment())
+            .findFirst().get();
+      Set<String> membersUsedForDiscovery = testedProtocol.getCollectedMessages().stream()
+            .map(e -> ((IpAddress)e.getDest()).getIpAddress().getHostAddress())
+            .collect(Collectors.toSet());
+      List<String> allowedPodsFromKubernetesApi = testedProtocol.getPods().stream()
+            .filter(pod -> senderParentDeployment.equals(pod.getParentDeployment()))
+            .map(pod -> pod.getIp())
+            .collect(Collectors.toList());
+
+      //then
+      Assertions.assertThat(allowedPodsFromKubernetesApi).containsAll(membersUsedForDiscovery);
+   }
+
+   private void sendInitialDiscovery(KUBE_PING kubePingProtocol) throws Exception {
+      new JChannel(
+            new TCP().setValue("bind_addr", InetAddress.getLoopbackAddress()).setValue("bind_port", findFreePort()),
+            kubePingProtocol,
+            new NAKACK2(),
+            new GMS().setValue("join_timeout", 1)
+      ).connect("RollingUpdateTest").disconnect();
+   }
+
+   static class KUBE_PING_FOR_TESTING extends KUBE_PING {
+
+      private final String resourceFile;
+      private List<Message> collectedMessages = new ArrayList<>();
+      private List<Pod> pods;
+
+      public KUBE_PING_FOR_TESTING(String resourceFile) {
+         this.resourceFile = resourceFile;
+      }
+
+      @Override
+      public void init() throws Exception {
+         super.init();
+         client = new TestClient(resourceFile);
+         pods = client.getPods(namespace, labels, true);
+      }
+
+      @Override
+      protected void sendDiscoveryRequest(Message req) {
+         collectedMessages.add(req);
+      }
+
+      public List<Message> getCollectedMessages() {
+         return collectedMessages;
+      }
+
+      public List<Pod> getPods() {
+         return pods;
+      }
+   }
+
+}