add resolution against node IPs

Author: adwk67

src/main/java/tech/stackable/hadoop/StackableTopologyProvider.java

@@ -119,24 +119,30 @@ private List<String> tryResolveFromCache(List<String> names) {
   private List<String> performFullResolution(List<String> names) {
     LOG.debug("Performing full topology resolution for: {}", names);
 
-    // Step 1: Gather all dataNodes
+    // Pre-requisites : fetch all dataNodes and build label lookup maps from them
     List<Pod> dataNodes = fetchDataNodes();
-
-    // Step 2: Resolve listeners to actual datanode IPs
-    List<String> resolvedNames = resolveListeners(names);
-
-    // Step 3: Build label lookup maps
     Map<String, Map<String, String>> podLabels = buildPodLabelMap(dataNodes);
     Map<String, Map<String, String>> nodeLabels = buildNodeLabelMap(dataNodes);
 
-    // Step 4: Build node-to-datanode map for O(1) colocated lookups
+    // Build node-to-datanode map for O(1) colocated lookups
     Map<String, String> nodeToDatanodeIp = buildNodeToDatanodeMap(dataNodes);
 
-    // Step 5: Resolve client pods to co-located dataNodes
+    // Resolve masqueraded IPs to nodes: do this before inspecting possible listener-
+    // or other pod-IPs as we don't want to mistakenly treat a masqueraded IP as a
+    // cache-miss
+    List<String> resolvedNames = tryResolveNodes(names, nodeToDatanodeIp);
+
+    // Resolve dataNode listeners to datanode IPs
+    resolvedNames = tryResolveListeners(resolvedNames);
+
+    // Step : Resolve client pods to co-located dataNodes
     List<String> datanodeIps =
-        resolveClientPodsToDataNodes(resolvedNames, podLabels, nodeToDatanodeIp);
+        tryResolveClientPodsToDataNodes(resolvedNames, podLabels, nodeToDatanodeIp);
 
-    // Step 6: Build topology strings and cache results
+    // Step : Build topology strings and cache results
+    // IP-masquerading can mean that the advertised IP is either a client Pod's node IP,
+    // or an IP that is nto easily associated with a node (e.g. if the veth interface is
+    // used).
     return buildAndCacheTopology(names, datanodeIps, podLabels, nodeLabels);
   }
 
@@ -183,6 +189,25 @@ private List<Pod> fetchDataNodes() {
     return dataNodes;
   }
 
+  // ============================================================================
+  // NODE RESOLUTION
+  // ============================================================================
+
+  private List<String> tryResolveNodes(List<String> names, Map<String, String> nodeToDatanodeIp) {
+    List<String> result = new ArrayList<>();
+
+    for (String name : names) {
+      String dataNodeIp = nodeToDatanodeIp.get(name);
+      if (dataNodeIp == null) {
+        result.add(name);
+      } else {
+        LOG.debug("Returning dataNode {} for {}", name, dataNodeIp);
+        result.add(dataNodeIp);
+      }
+    }
+    return result;
+  }
+
   // ============================================================================
   // LISTENER RESOLUTION
   // ============================================================================
@@ -221,10 +246,10 @@ private String getListenerVersion() {
     }
   }
 
-  private List<String> resolveListeners(List<String> names) {
+  private List<String> tryResolveListeners(List<String> names) {
     refreshListenerCacheIfNeeded(names);
 
-    return names.stream().map(this::resolveListenerToDatanode).collect(Collectors.toList());
+    return names.stream().map(this::tryResolveListenerToDatanode).collect(Collectors.toList());
   }
 
   private void refreshListenerCacheIfNeeded(List<String> names) {
@@ -268,7 +293,7 @@ private void cacheListenerByNameAndAddresses(GenericKubernetesResource listener)
    *     listener
    * @return either the name (for non-listener) or the dataNode IP to which this listener resolves
    */
-  private String resolveListenerToDatanode(String name) {
+  private String tryResolveListenerToDatanode(String name) {
     GenericKubernetesResource listener = cache.getListener(name);
     if (listener == null) {
       LOG.debug("Not a listener: {}", name);
@@ -314,7 +339,7 @@ private GenericKubernetesResourceList fetchListeners(String listenerVersion) {
   // CLIENT POD RESOLUTION
   // ============================================================================
 
-  private List<String> resolveClientPodsToDataNodes(
+  private List<String> tryResolveClientPodsToDataNodes(
       List<String> names,
       Map<String, Map<String, String>> podLabels,
       Map<String, String> nodeToDatanodeIp) {
@@ -408,7 +433,8 @@ private String resolveToIpAddress(String hostname) {
 
   /**
    * Build a map from Kubernetes node name to datanode IP. This enables O(1) lookup when finding
-   * co-located dataNodes for client pods.
+   * co-located dataNodes for client pods. This map will contain as keys both the node name and all
+   * its addresses (as the address may be used by pods with IP masquerading).
    *
    * <p>Note: If multiple dataNodes run on the same node, the last one wins. This is acceptable
    * because all dataNodes on the same node have the same topology.
@@ -421,11 +447,17 @@ private Map<String, String> buildNodeToDatanodeMap(List<Pod> dataNodes) {
       String dataNodeIp = dataNode.getStatus().getPodIP();
 
       if (nodeName != null && dataNodeIp != null) {
+        LOG.debug("Assigned to node-name [{}/{}]", nodeName, dataNodeIp);
         nodeToDatanode.put(nodeName, dataNodeIp);
+        Node node = getOrFetchNode(nodeName);
+        for (NodeAddress nodeAddress : node.getStatus().getAddresses()) {
+          LOG.debug("Assigned to node-address [{}/{}]", nodeAddress.getAddress(), dataNodeIp);
+          nodeToDatanode.put(nodeAddress.getAddress(), dataNodeIp);
+        }
       }
     }
 
-    LOG.debug("Built node-to-datanode map with {} entries", nodeToDatanode.size());
+    LOG.debug("Built node-to-datanode map {}", nodeToDatanode);
     return nodeToDatanode;
   }
 
@@ -474,27 +506,32 @@ private String extractLabelValue(
    * Given a list of dataNodes this function will resolve which dataNodes run on which node as well
    * as all the ips assigned to a dataNodes. It will then return a mapping of every ip address to
    * the labels that are attached to the "physical" node running the dataNodes that this ip belongs
-   * to.
+   * to. It will also do this for the node addresses as calling pods may masquerade as node IPs.
    *
    * @param dataNodes List of all in-scope dataNodes (datanode pods in this namespace)
    * @return Map of ip addresses to labels of the node running the pod that the ip address belongs
    *     to
    */
   private Map<String, Map<String, String>> buildNodeLabelMap(List<Pod> dataNodes) {
     Map<String, Map<String, String>> result = new HashMap<>();
-    for (Pod pod : dataNodes) {
-      String nodeName = pod.getSpec().getNodeName();
+    for (Pod dataNode : dataNodes) {
+      String nodeName = dataNode.getSpec().getNodeName();
 
       if (nodeName == null) {
-        LOG.warn("Pod [{}] not yet assigned to node, retrying", pod.getMetadata().getName());
+        LOG.warn("Pod [{}] not yet assigned to node, retrying", dataNode.getMetadata().getName());
         return result;
       }
 
       Node node = getOrFetchNode(nodeName);
       Map<String, String> nodeLabels = node.getMetadata().getLabels();
       LOG.debug("Labels for node [{}]:[{}]....", nodeName, nodeLabels);
 
-      for (PodIP podIp : pod.getStatus().getPodIPs()) {
+      for (NodeAddress nodeAddress : node.getStatus().getAddresses()) {
+        LOG.debug("...assigned to node address [{}]", nodeAddress.getAddress());
+        result.put(nodeAddress.getAddress(), nodeLabels);
+      }
+
+      for (PodIP podIp : dataNode.getStatus().getPodIPs()) {
         LOG.debug("...assigned to IP [{}]", podIp.getIp());
         result.put(podIp.getIp(), nodeLabels);
       }

test/topology-provider/stack/01-install-krb5-kdc.yaml

@@ -31,7 +31,7 @@ spec:
     spec:
       initContainers:
         - name: init
-          image: oci.stackable.tech/sdp/krb5:1.18.2-stackable0.0.0-dev
+          image: oci.stackable.tech/sdp/krb5:1.21.1-stackable0.0.0-dev
           args:
             - sh
             - -euo
@@ -52,7 +52,7 @@ spec:
               name: data
       containers:
         - name: kdc
-          image: oci.stackable.tech/sdp/krb5:1.18.2-stackable0.0.0-dev
+          image: oci.stackable.tech/sdp/krb5:1.21.1-stackable0.0.0-dev
           args:
             - krb5kdc
             - -n
@@ -65,7 +65,7 @@ spec:
             - mountPath: /var/kerberos/krb5kdc
               name: data
         - name: kadmind
-          image: oci.stackable.tech/sdp/krb5:1.18.2-stackable0.0.0-dev
+          image: oci.stackable.tech/sdp/krb5:1.21.1-stackable0.0.0-dev
           args:
             - kadmind
             - -nofork
@@ -78,7 +78,7 @@ spec:
             - mountPath: /var/kerberos/krb5kdc
               name: data
         - name: client
-          image: oci.stackable.tech/sdp/krb5:1.18.2-stackable0.0.0-dev
+          image: oci.stackable.tech/sdp/krb5:1.21.1-stackable0.0.0-dev
           tty: true
           stdin: true
           env:

test/topology-provider/stack/03-hdfs.yaml

@@ -5,51 +5,55 @@ metadata:
   name: simple-zk
 spec:
   image:
-    productVersion: 3.8.3
+    productVersion: 3.9.4
+    pullPolicy: IfNotPresent
   servers:
     roleGroups:
       default:
         replicas: 1
 ---
-apiVersion: zookeeper.stackable.tech/v1alpha1
-kind: ZookeeperZnode
-metadata:
-  name: simple-hdfs-znode
-spec:
-  clusterRef:
-    name: simple-zk
----
 apiVersion: hdfs.stackable.tech/v1alpha1
 kind: HdfsCluster
 metadata:
   name: simple-hdfs
 spec:
   image:
-    productVersion: 3.4.0
-    custom: hdfs # updated by tilt
+    productVersion: 3.4.2
+    custom: oci.stackable.tech/sandbox/andrew/hadoop:3.4.2-stackable0.0.0-topprov
     pullPolicy: IfNotPresent
   clusterConfig:
     dfsReplication: 1
-    zookeeperConfigMapName: simple-hdfs-znode
+    zookeeperConfigMapName: simple-zk
     rackAwareness:
-      - labelType: node
-        labelName: kubernetes.io/hostname
-      - labelType: pod
-        labelName: app.kubernetes.io/role-group
+      - nodeLabel: kubernetes.io/hostname
+      - podLabel: app.kubernetes.io/role-group
     authentication:
       tlsSecretClass: tls
       kerberos:
         secretClass: kerberos-default
   nameNodes:
     config:
-      listenerClass: external-stable # We want to access the Web UI
+      listenerClass: external-stable
+      logging:
+        enableVectorAgent: false
+        containers:
+          hdfs:
+            console:
+              level: DEBUG
+            file:
+              level: DEBUG
+            loggers:
+              ROOT:
+                level: DEBUG
     configOverrides: &configOverrides
       core-site.xml:
         hadoop.user.group.static.mapping.overrides: "dr.who=;nn=;nm=;jn=;testuser=supergroup;"
     roleGroups:
       default:
         replicas: 2
   dataNodes:
+    config:
+      listenerClass: external-stable
     configOverrides: *configOverrides
     roleGroups:
       default:

…st/topology-provider/stack/07-spark.yaml …st/topology-provider/stack/04-spark.yamltest/topology-provider/stack/07-spark.yaml renamed to test/topology-provider/stack/04-spark.yaml test/topology-provider/stack/07-spark.yaml renamed to test/topology-provider/stack/04-spark.yaml

@@ -5,14 +5,14 @@ metadata:
   name: spark-teragen
 spec:
   sparkImage:
-    custom: docker.stackable.tech/stackable/spark-k8s-with-teragen:3.5.0-stackable0.0.0-dev
-    productVersion: 3.5.0
+    custom: oci.stackable.tech/sandbox/spark-k8s:3.5.6-stackable0.0.0-terasort
+    productVersion: 3.5.6
     pullPolicy: IfNotPresent
   mode: cluster
   mainApplicationFile: local:///tmp/spark-terasort-1.2-SNAPSHOT.jar
   mainClass: com.github.ehiggs.spark.terasort.TeraGen
   args:
-    - "10M"
+    - "100M"
     - "hdfs://simple-hdfs/user/stackable/teragen_output"
   sparkConf:
     "spark.driver.extraClassPath": "/etc/hadoop/conf/:/stackable/spark/extra-jars/*"

test/topology-provider/stack/05-access-hdfs.yaml

@@ -0,0 +1,63 @@
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: access-hdfs
+spec:
+  template:
+    spec:
+      containers:
+        - name: access-hdfs
+          image: oci.stackable.tech/sandbox/andrew/hadoop:3.4.2-stackable0.0.0-topprov
+          imagePullPolicy: IfNotPresent
+          env:
+            - name: HADOOP_CONF_DIR
+              value: /stackable/conf/hdfs
+            - name: KRB5_CONFIG
+              value: /stackable/kerberos/krb5.conf
+            - name: HADOOP_OPTS
+              value: -Djava.security.krb5.conf=/stackable/kerberos/krb5.conf
+          command:
+            - /bin/bash
+            - -c
+            - |
+              set -ex
+              klist -k /stackable/kerberos/keytab
+              kinit -kt /stackable/kerberos/keytab testuser/access-hdfs.default.svc.cluster.local
+              klist
+
+              #bin/hdfs dfs -mkdir /stackable
+              #bin/hdfs dfs -chown -R access-hive /stackable
+              #bin/hdfs dfs -mkdir /access-hive
+              #bin/hdfs dfs -chown -R access-hive /access-hive
+              bin/hdfs dfs -ls /
+              #bin/hdfs dfs -ls -d /
+              
+              sleep infinity
+          volumeMounts:
+            - name: hdfs-config
+              mountPath: /stackable/conf/hdfs
+            - name: kerberos
+              mountPath: /stackable/kerberos
+      volumes:
+        - name: hdfs-config
+          configMap:
+            name: simple-hdfs
+        - name: kerberos
+          ephemeral:
+            volumeClaimTemplate:
+              metadata:
+                annotations:
+                  secrets.stackable.tech/class: kerberos-default
+                  secrets.stackable.tech/scope: service=access-hdfs
+                  secrets.stackable.tech/kerberos.service.names: testuser
+              spec:
+                storageClassName: secrets.stackable.tech
+                accessModes:
+                  - ReadWriteOnce
+                resources:
+                  requests:
+                    storage: "1"
+      securityContext:
+        fsGroup: 1000
+      restartPolicy: OnFailure

test/topology-provider/teragen/Dockerfile

@@ -1,9 +1,9 @@
 #
 # cd test/stack/teragen
-# docker build . -t docker.stackable.tech/stackable/spark-k8s-with-teragen:3.5.0-stackable0.0.0-dev
+# docker build . -t oci.stackable.tech/sandbox/spark-k8s:3.5.6-stackable0.0.0-terasort
 #
 
-FROM oci.stackable.tech/sdp/spark-k8s:3.5.1-stackable0.0.0-dev
+FROM oci.stackable.tech/sdp/spark-k8s:3.5.6-stackable0.0.0-dev
 
 # this .jar is compiled from the code here: https://github.com/ehiggs/spark-terasort
-COPY --chown=stackable:stackable ./spark-terasort-1.2-SNAPSHOT.jar /tmp/
+COPY --chown=stackable:stackable ./spark-terasort-1.2-SNAPSHOT.jar /tmp/