Add NODE_AWARE metadata support (#278)

* Add NODE_AWARE metadata support

With 3.12.11 IMDG release, new metadata introduced to define partition group
based on Kubernetes node that Hazelcast cluster runs on.The discovery plugin
should pass name of the node to the core SPI via using discoverLocalMetadata method
like zone info. This method passes both zone and node name metadata. User can not
enable tow group types at the same time so cluster will decide which metadata
will be used based on cluster config.

Author: Hasan Çelik

README.md

@@ -186,6 +186,8 @@ There are 2 properties to configure the plugin:
 **Note**: In this README, only YAML configurations are presented, however you can achieve exactly the same effect using 
 XML or Java-based configurations.
 
+## High Availability
+
 ### Zone Aware
 
 When using `ZONE_AWARE` configuration, backups are created in the other availability zone. This feature is available only for the Kubernetes API mode.
@@ -224,6 +226,42 @@ env:
         fieldPath: metadata.name
  ``` 
 
+### Node Aware
+
+When using `NODE_AWARE` configuration, backups are created in the other Kubernetes nodes. This feature is available only for the Kubernetes API mode.
+
+**Note**: Your Kubernetes cluster must orchestrate Hazelcast Member PODs equally between the nodes, otherwise Node Aware feature may not work correctly.
+
+#### YAML Configuration
+
+```yaml
+partition-group:
+  enabled: true
+  group-type: NODE_AWARE
+```
+
+#### Java-based Configuration
+
+```java
+config.getPartitionGroupConfig()
+    .setEnabled(true)
+    .setGroupType(MemberGroupType.NODE_AWARE);
+```
+
+Note the following aspects of `NODE_AWARE`:
+ * Retrieving name of the node uses Kubernetes API, so RBAC must be configured as described [here](#granting-permissions-to-use-kubernetes-api)
+ * `NODE_AWARE` feature works correctly when Hazelcast members are distributed equally in all nodes, so your Kubernetes cluster must orchestrate PODs equally.
+ 
+Note also that retrieving name of the node assumes that your container's hostname is the same as POD Name, which is almost always true. If you happen to change your hostname in the container, then please define the following environment variable:
+ 
+ ```yaml
+env:
+  - name: POD_NAME
+    valueFrom:
+      fieldRef:
+        fieldPath: metadata.name
+ ``` 
+
 ## Hazelcast Client Configuration
 
 Depending on whether your Hazelcast Client runs inside or outside the Kubernetes cluster, its configuration looks different.

src/main/java/com/hazelcast/kubernetes/HazelcastKubernetesDiscoveryStrategy.java

@@ -70,6 +70,7 @@ public void start() {
     public Map<String, Object> discoverLocalMetadata() {
         if (memberMetadata.isEmpty()) {
             memberMetadata.put(PartitionGroupMetaData.PARTITION_GROUP_ZONE, discoverZone());
+            memberMetadata.put("hazelcast.partition.group.node", discoverNodeName());
         }
         return memberMetadata;
     }
@@ -82,14 +83,7 @@ public Map<String, Object> discoverLocalMetadata() {
     private String discoverZone() {
         if (DiscoveryMode.KUBERNETES_API.equals(config.getMode())) {
             try {
-                String podName = System.getenv("POD_NAME");
-                if (podName == null) {
-                    podName = System.getenv("HOSTNAME");
-                }
-                if (podName == null) {
-                    podName = InetAddress.getLocalHost().getHostName();
-                }
-                String zone = client.zone(podName);
+                String zone = client.zone(podName());
                 if (zone != null) {
                     getLogger().info(String.format("Kubernetes plugin discovered availability zone: %s", zone));
                     return zone;
@@ -103,6 +97,39 @@ private String discoverZone() {
         return "unknown";
     }
 
+    /**
+     * Discovers the name of the node which the current Hazelcast member pod is running on.
+     * <p>
+     * Note: NODE_AWARE is available only for the Kubernetes API Mode.
+     */
+    private String discoverNodeName() {
+        if (DiscoveryMode.KUBERNETES_API.equals(config.getMode())) {
+            try {
+                String nodeName = client.nodeName(podName());
+                if (nodeName != null) {
+                    getLogger().info(String.format("Kubernetes plugin discovered node name: %s", nodeName));
+                    return nodeName;
+                }
+            } catch (Exception e) {
+                // only log the exception and the message, Hazelcast should still start
+                getLogger().finest(e);
+            }
+            getLogger().warning("Cannot fetch name of the node, NODE_AWARE feature is disabled");
+        }
+        return "unknown";
+    }
+
+    private String podName() throws UnknownHostException {
+        String podName = System.getenv("POD_NAME");
+        if (podName == null) {
+            podName = System.getenv("HOSTNAME");
+        }
+        if (podName == null) {
+            podName = InetAddress.getLocalHost().getHostName();
+        }
+        return podName;
+    }
+
     @Override
     public Iterable<DiscoveryNode> discoverNodes() {
         return endpointResolver.resolve();

src/main/java/com/hazelcast/kubernetes/KubernetesClient.java

@@ -137,7 +137,6 @@ List<Endpoint> endpointsByPodLabel(String podLabel, String podLabelValue) {
         }
     }
 
-
     /**
      * Retrieves zone name for the specified {@code namespace} and the given {@code podName}.
      * <p>
@@ -149,13 +148,22 @@ List<Endpoint> endpointsByPodLabel(String podLabel, String podLabelValue) {
      * @see <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.11">Kubernetes Endpoint API</a>
      */
     String zone(String podName) {
-        String podUrlString = String.format("%s/api/v1/namespaces/%s/pods/%s", kubernetesMaster, namespace, podName);
-        String nodeName = extractNodeName(callGet(podUrlString));
-
-        String nodeUrlString = String.format("%s/api/v1/nodes/%s", kubernetesMaster, nodeName);
+        String nodeUrlString = String.format("%s/api/v1/nodes/%s", kubernetesMaster, nodeName(podName));
         return extractZone(callGet(nodeUrlString));
     }
 
+    /**
+     * Retrieves node name for the specified {@code namespace} and the given {@code podName}.
+     *
+     * @param podName POD name
+     * @return Node name
+     * @see <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.11">Kubernetes Endpoint API</a>
+     */
+    String nodeName(String podName) {
+        String podUrlString = String.format("%s/api/v1/namespaces/%s/pods/%s", kubernetesMaster, namespace, podName);
+        return extractNodeName(callGet(podUrlString));
+    }
+
     private static List<Endpoint> parsePodsList(JsonObject podsListJson) {
         List<Endpoint> addresses = new ArrayList<Endpoint>();
 

src/test/java/com/hazelcast/kubernetes/KubernetesClientTest.java

@@ -361,6 +361,28 @@ public void zone() {
         assertEquals("us-central1-a", zone);
     }
 
+
+    @Test
+    public void nodeName() {
+        // given
+        String podName = "pod-name";
+
+        //language=JSON
+        String podResponse = "{\n"
+                + "  \"kind\": \"Pod\",\n"
+                + "  \"spec\": {\n"
+                + "    \"nodeName\": \"kubernetes-node-f0bbd602-f7cw\"\n"
+                + "  }\n"
+                + "}";
+        stub(String.format("/api/v1/namespaces/%s/pods/%s", NAMESPACE, podName), podResponse);
+
+        // when
+        String nodeName = kubernetesClient.nodeName(podName);
+
+        // then
+        assertEquals("kubernetes-node-f0bbd602-f7cw", nodeName);
+    }
+
     @Test
     public void endpointsByNamespaceWithLoadBalancerPublicIp() {
         // given