implement DefaultSharedIndexInformer#addIndexers

shakeelrao · shakeelrao · commit 4951b85372f9 · 2019-12-15T16:46:22.000-08:00
diff --git a/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java b/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java
@@ -319,6 +319,38 @@ public List<ApiType> byIndex(String indexName, String indexKey) {
     }
   }
 
+  /**
+   * Return the indexers registered with the cache.
+   *
+   * @return registered indexers
+   */
+  @Override
+  public Map<String, Function<ApiType, List<String>>> getIndexers() {
+    return indexers;
+  }
+
+  /**
+   * Add additional indexers to the cache.
+   *
+   * @param newIndexers indexers to add
+   */
+  @Override
+  public void addIndexers(Map<String, Function<ApiType, List<String>>> newIndexers) {
+    if (!items.isEmpty()) {
+      throw new IllegalStateException("cannot add indexers to a non-empty cache");
+    }
+    Set<String> oldKeys = indexers.keySet();
+    Set<String> newKeys = newIndexers.keySet();
+    Set<String> intersection = new HashSet<>(oldKeys);
+    intersection.retainAll(newKeys);
+    if (!intersection.isEmpty()) {
+      throw new IllegalArgumentException("indexer conflict: " + intersection);
+    }
+    for (Map.Entry<String, Function<ApiType, List<String>>> indexEntry : newIndexers.entrySet()) {
+      addIndexFunc(indexEntry.getKey(), indexEntry.getValue());
+    }
+  }
+
   /**
    * updateIndices modifies the objects location in the managed indexes, if this is an update, you
    * must provide an oldObj.
@@ -381,13 +413,8 @@ private void deleteFromIndices(ApiType oldObj, String key) {
     }
   }
 
-  /**
-   * Add index func.
-   *
-   * @param indexName the index name
-   * @param indexFunc the index func
-   */
-  public void addIndexFunc(String indexName, Function<ApiType, List<String>> indexFunc) {
+  /** Protected for testing. */
+  protected void addIndexFunc(String indexName, Function<ApiType, List<String>> indexFunc) {
     this.indices.put(indexName, new HashMap<>());
     this.indexers.put(indexName, indexFunc);
   }
diff --git a/util/src/main/java/io/kubernetes/client/informer/cache/Indexer.java b/util/src/main/java/io/kubernetes/client/informer/cache/Indexer.java
@@ -1,6 +1,8 @@
 package io.kubernetes.client.informer.cache;
 
 import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
 
 /** Indexer extends Store interface and adds index/de-index methods. */
 public interface Indexer<ApiType> extends Store<ApiType> {
@@ -30,4 +32,18 @@ public interface Indexer<ApiType> extends Store<ApiType> {
    * @return matched objects
    */
   List<ApiType> byIndex(String indexName, String indexKey);
+
+  /**
+   * Return the indexers registered with the store.
+   *
+   * @return registered indexers
+   */
+  Map<String, Function<ApiType, List<String>>> getIndexers();
+
+  /**
+   * Add additional indexers to the store.
+   *
+   * @param indexers indexers to add
+   */
+  void addIndexers(Map<String, Function<ApiType, List<String>>> indexers);
 }
diff --git a/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java b/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java
@@ -12,6 +12,7 @@
 import java.util.Deque;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.tuple.MutablePair;
 import org.slf4j.Logger;
@@ -223,8 +224,11 @@ private void handleDeltas(Deque<MutablePair<DeltaFIFO.DeltaType, Object>> deltas
   }
 
   @Override
-  public void addIndexers(Map indexers) {
-    throw new RuntimeException("unimplemented!");
+  public void addIndexers(Map<String, Function<ApiType, List<String>>> indexers) {
+    if (started) {
+      throw new IllegalStateException("cannot add indexers to a running informer");
+    }
+    indexer.addIndexers(indexers);
   }
 
   @Override
diff --git a/util/src/test/java/io/kubernetes/client/informer/cache/CacheTest.java b/util/src/test/java/io/kubernetes/client/informer/cache/CacheTest.java
@@ -5,9 +5,8 @@
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodSpec;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
+import java.util.*;
+import java.util.function.Function;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.contrib.java.lang.system.EnvironmentVariables;
@@ -144,4 +143,44 @@ public void testMultiIndexFuncCacheStore() {
     List<V1Pod> nodeNameIndexedPods = podCache.byIndex(testIndexFuncName, "node1");
     assertEquals(1, nodeNameIndexedPods.size());
   }
+
+  @Test
+  public void testAddIndexers() {
+    Cache<V1Pod> podCache = new Cache<>();
+
+    String nodeIndex = "node-index";
+    String clusterIndex = "cluster-index";
+
+    Map<String, Function<V1Pod, List<String>>> indexers = new HashMap<>();
+
+    indexers.put(
+        nodeIndex,
+        (V1Pod pod) -> {
+          return Arrays.asList(pod.getSpec().getNodeName());
+        });
+
+    indexers.put(
+        clusterIndex,
+        (V1Pod pod) -> {
+          return Arrays.asList(pod.getMetadata().getClusterName());
+        });
+
+    podCache.addIndexers(indexers);
+
+    V1Pod testPod =
+        new V1Pod()
+            .metadata(new V1ObjectMeta().namespace("ns").name("n").clusterName("cluster1"))
+            .spec(new V1PodSpec().nodeName("node1"));
+
+    podCache.add(testPod);
+
+    List<V1Pod> namespaceIndexedPods = podCache.byIndex(Caches.NAMESPACE_INDEX, "ns");
+    assertEquals(1, namespaceIndexedPods.size());
+
+    List<V1Pod> nodeNameIndexedPods = podCache.byIndex(nodeIndex, "node1");
+    assertEquals(1, nodeNameIndexedPods.size());
+
+    List<V1Pod> clusterNameIndexedPods = podCache.byIndex(clusterIndex, "cluster1");
+    assertEquals(1, clusterNameIndexedPods.size());
+  }
 }
