code cleanup

Author: adwk67

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

@@ -16,8 +16,8 @@
  * An implementation of the org.apache.hadoop.net.DNSToSwitchMapping that is used to create a
  * topology out of dataNodes.
  *
- * <p>This class is intended to be run as part of the NameNode process and will be used by the
- * nameNode to retrieve topology strings for dataNodes.
+ * <p>This class is intended to be run as part of the NameNode process (in the same namespace) and
+ * will be used by the nameNode to retrieve topology strings for dataNodes.
  */
 public class StackableTopologyProvider implements DNSToSwitchMapping {
   private final Logger LOG = LoggerFactory.getLogger(StackableTopologyProvider.class);
@@ -47,11 +47,10 @@ public StackableTopologyProvider() {
   @Override
   public void reloadCachedMappings() {
     // TODO: According to the upstream comment we should rebuild all cache entries after
-    // invalidating them
-    //  this may mean trying to resolve ip addresses that do not exist any more and things like that
-    // though and
-    //  require some more thought, so we will for now just invalidate the cache.
-    this.cache.invalidateAll();
+    // invalidating them. This may mean trying to resolve ip addresses that do not exist
+    // any more and things like that though and require some more thought, so we will for
+    // now just invalidate the cache.
+    this.cache.invalidateAllTopologyKeys();
   }
 
   @Override
@@ -73,10 +72,7 @@ private void logInitializationStatus() {
           labels.stream().map(TopologyLabel::getName).collect(Collectors.toList());
       LOG.info("Initialized with topology labels: {}", labelNames);
     }
-    LOG.debug(
-        "Client namespaces {} and configuration {}",
-        client.getNamespace(),
-        client.getConfiguration());
+    LOG.debug("Client namespace {}", client.getNamespace());
   }
 
   @Override
@@ -106,8 +102,7 @@ private List<String> createDefaultRackList(List<String> names) {
 
   private List<String> tryResolveFromCache(List<String> names) {
     // We need to check if we have cached values for all dataNodes contained in this request.
-    // Unless we can answer everything from the cache we have to talk to k8s anyway and can just
-    // recalculate everything
+    // Unless we can answer everything from the cache we will perform a full resolution.
     List<String> cached = names.stream().map(cache::getTopology).collect(Collectors.toList());
     LOG.debug("Cached topologyKeyCache values [{}]", cached);
 
@@ -263,10 +258,10 @@ private String resolveListenerEndpoint(GenericKubernetesResource listener) {
   }
 
   private GenericKubernetesResourceList fetchListeners() {
+    // no version is specified here as we are not always going to be on v1alpha1
     ResourceDefinitionContext listenerCrd =
         new ResourceDefinitionContext.Builder()
             .withGroup("listeners.stackable.tech")
-            .withVersion("v1alpha1")
             .withPlural("listeners")
             .withNamespaced(true)
             .build();
@@ -488,8 +483,10 @@ private Map<String, Map<String, String>> buildPodLabelMap(List<Pod> dataNodes) {
     Map<String, Map<String, String>> result = new HashMap<>();
     for (Pod pod : dataNodes) {
       Map<String, String> podLabels = pod.getMetadata().getLabels();
+      LOG.debug("Labels for pod [{}]:[{}]....", pod.getMetadata().getName(), podLabels);
 
       for (PodIP podIp : pod.getStatus().getPodIPs()) {
+        LOG.debug("...assigned to pod IP [{}]", podIp.getIp());
         result.put(podIp.getIp(), podLabels);
       }
     }

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

@@ -6,7 +6,6 @@
 import io.fabric8.kubernetes.api.model.Node;
 import io.fabric8.kubernetes.api.model.Pod;
 import java.util.List;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 
 /** Manages all caching layers for the topology provider. */
@@ -38,7 +37,7 @@ void putTopology(String key, String value) {
     topology.put(key, value);
   }
 
-  void invalidateAll() {
+  void invalidateAllTopologyKeys() {
     topology.invalidateAll();
   }
 
@@ -58,26 +57,14 @@ GenericKubernetesResource getListener(String name) {
     return listeners.getIfPresent(name);
   }
 
-  ConcurrentMap<String, GenericKubernetesResource> getListenerMap() {
-    return listeners.asMap();
-  }
-
   void putListener(String name, GenericKubernetesResource listener) {
     listeners.put(name, listener);
   }
 
-  boolean hasAllListeners(List<String> names) {
-    return names.stream().noneMatch(name -> listeners.getIfPresent(name) == null);
-  }
-
   Pod getPod(String name) {
     return pods.getIfPresent(name);
   }
 
-  ConcurrentMap<String, Pod> getPodMap() {
-    return pods.asMap();
-  }
-
   void putPod(String name, Pod pod) {
     pods.put(name, pod);
   }

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

@@ -10,7 +10,7 @@
 public class TopologyLabel {
   private static final Logger LOG = LoggerFactory.getLogger(TopologyLabel.class);
   public static final String VARNAME_LABELS = "TOPOLOGY_LABELS";
-  public static final String VARNAME_MAXLEVELS = "TOPOLOGY_MAX_LEVELS";
+  public static final String VARNAME_MAX_LEVELS = "TOPOLOGY_MAX_LEVELS";
   private static final int MAX_LEVELS_DEFAULT = 2;
 
   public enum Type {
@@ -32,7 +32,7 @@ public enum Type {
     String[] parts = config.toLowerCase(Locale.ROOT).split(":", 2);
 
     if (parts.length != 2) {
-      LOG.warn("Invalid topology label format '{}' - expected '[node|pod]:<labelname>'", config);
+      LOG.warn("Invalid topology label format '{}' - expected '[node|pod]:<label>'", config);
       this.type = Type.UNDEFINED;
       this.name = null;
       return;
@@ -102,7 +102,7 @@ public static List<TopologyLabel> initializeTopologyLabels() {
 
     if (labels.stream().anyMatch(TopologyLabel::isUndefined)) {
       LOG.error(
-          "Invalid topology label configuration - labels must be in format '[pod|node]:<labelname>'");
+          "Invalid topology label configuration - labels must be in format '[pod|node]:<label>'");
       throw new RuntimeException("Invalid topology label configuration");
     }
 
@@ -111,6 +111,6 @@ public static List<TopologyLabel> initializeTopologyLabels() {
 
   private static int getMaxLabels() {
     return TopologyUtils.parseIntFromEnv(
-        VARNAME_MAXLEVELS, MAX_LEVELS_DEFAULT, "maximum topology levels");
+        VARNAME_MAX_LEVELS, MAX_LEVELS_DEFAULT, "maximum topology levels");
   }
 }

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

@@ -29,6 +29,7 @@ public static List<String> getIngressAddresses(GenericKubernetesResource listene
   public static int parseIntFromEnv(String varName, int defaultValue, String description) {
     String value = System.getenv(varName);
     if (value == null || value.isEmpty()) {
+      LOG.info("Set {} to default value {}", description, defaultValue);
       return defaultValue;
     }