preventing the read of stale index entries

Signed-off-by: Steve Hawkins <shawkins@redhat.com>

Author: shawkins

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/impl/cache/CacheImpl.java

@@ -22,14 +22,14 @@
 import io.fabric8.kubernetes.client.informers.cache.ItemStore;
 import io.fabric8.kubernetes.client.utils.Utils;
 
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -47,36 +47,39 @@
  */
 public class CacheImpl<T extends HasMetadata> implements Cache<T> {
 
-  private static class Index {
-    private Map<Object, Set<String>> values = new ConcurrentHashMap<Object, Set<String>>();
+  private static class Index<T extends HasMetadata> {
+    
+    private final Function<T, List<String>> indexer;
+    private Map<Object, Map<String, String>> values = new ConcurrentHashMap<Object, Map<String, String>>();
+    
+    public Index(Function<T, List<String>> indexer) {
+      this.indexer = indexer;
+    }
 
-    public void update(String indexKey, String key, boolean remove) {
+    public void update(String indexKey, String key, String resourceVersion, boolean remove) {
       if (remove) {
         values.computeIfPresent(indexKey == null ? this : indexKey, (k, v) -> {
           v.remove(key);
           return v.isEmpty() ? null : v;
         });
       } else {
-        values.computeIfAbsent(indexKey == null ? this : indexKey, k -> ConcurrentHashMap.newKeySet()).add(key);
+        values.compute(indexKey == null ? this : indexKey, (k, v) -> v == null ? new ConcurrentHashMap<>() : v).put(key, resourceVersion);
       }
     }
 
-    public Set<String> get(String indexKey) {
-      return values.getOrDefault(indexKey == null ? this : indexKey, Collections.emptySet());
+    public Map<String, String> get(String indexKey) {
+      return values.getOrDefault(indexKey == null ? this : indexKey, Map.of());
     }
   }
 
   // NAMESPACE_INDEX is the default index function for caching objects
   public static final String NAMESPACE_INDEX = "namespace";
 
-  // indexers stores index functions by their names
-  private final Map<String, Function<T, List<String>>> indexers = Collections.synchronizedMap(new HashMap<>());
-
   // items stores object instances
   private ItemStore<T> items;
 
   // indices stores objects' key by their indices
-  private final ConcurrentMap<String, Index> indices = new ConcurrentHashMap<>();
+  private final ConcurrentMap<String, Index<T>> indices = new ConcurrentHashMap<>();
 
   public CacheImpl() {
     this(NAMESPACE_INDEX, Cache::metaNamespaceIndexFunc, Cache::metaNamespaceKeyFunc);
@@ -97,15 +100,13 @@ public void setItemStore(ItemStore<T> items) {
    * @return registered indexers
    */
   @Override
-  public Map<String, Function<T, List<String>>> getIndexers() {
-    synchronized (indexers) {
-      return Collections.unmodifiableMap(indexers);
-    }
+  public synchronized Map<String, Function<T, List<String>>> getIndexers() {
+    return Collections.unmodifiableMap(indices.entrySet().stream().collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue().indexer)));
   }
 
   @Override
   public synchronized void addIndexers(Map<String, Function<T, List<String>>> indexersNew) {
-    Set<String> intersection = new HashSet<>(indexers.keySet());
+    Set<String> intersection = new HashSet<>(indices.keySet());
     intersection.retainAll(indexersNew.keySet());
     if (!intersection.isEmpty()) {
       throw new IllegalArgumentException("Indexer conflict: " + intersection);
@@ -208,29 +209,17 @@ public T getByKey(String key) {
    */
   @Override
   public List<T> index(String indexName, T obj) {
-    Function<T, List<String>> indexFunc = this.indexers.get(indexName);
-    if (indexFunc == null) {
-      throw new IllegalArgumentException(String.format("index %s doesn't exist!", indexName));
-    }
-    Index index = getIndex(indexName);
-    List<String> indexKeys = indexFunc.apply(obj);
-    Set<String> returnKeySet = new HashSet<>();
+    Index<T> index = getIndex(indexName);
+    List<String> indexKeys = index.indexer.apply(obj);
+    Map<String, T> result = new LinkedHashMap<>();
     for (String indexKey : indexKeys) {
-      returnKeySet.addAll(index.get(indexKey));
+      byIndex(indexName, indexKey, result);
     }
 
-    return getItems(returnKeySet);
-  }
-
-  private List<T> getItems(Set<String> returnKeySet) {
-    List<T> items = new ArrayList<>(returnKeySet.size());
-    for (String absoluteKey : returnKeySet) {
-      Optional.ofNullable(this.items.get(absoluteKey)).ifPresent(items::add);
-    }
-    return items;
+    return List.copyOf(result.values());
   }
 
-  private Index getIndex(String indexName) {
+  private Index<T> getIndex(String indexName) {
     return Optional.ofNullable(this.indices.get(indexName))
         .orElseThrow(() -> new IllegalArgumentException(String.format("index %s doesn't exist!", indexName)));
   }
@@ -244,8 +233,7 @@ private Index getIndex(String indexName) {
    */
   @Override
   public List<String> indexKeys(String indexName, String indexKey) {
-    Index index = getIndex(indexName);
-    return new ArrayList<>(index.get(indexKey));
+    return byIndex(indexName, indexKey).stream().map(this::getKey).collect(Collectors.toList());
   }
 
   /**
@@ -257,8 +245,29 @@ public List<String> indexKeys(String indexName, String indexKey) {
    */
   @Override
   public List<T> byIndex(String indexName, String indexKey) {
-    Index index = getIndex(indexName);
-    return getItems(index.get(indexKey));
+    Map<String, T> result = new LinkedHashMap<>();
+    byIndex(indexName, indexKey, result);
+    return List.copyOf(result.values());
+  }
+  
+  private void byIndex(String indexName, String indexKey, Map<String, T> result) {
+    Index<T> index = getIndex(indexName);
+    Map<String, String> objs = index.get(indexKey);
+    for (Map.Entry<String, String> entry : objs.entrySet()) {
+      if (result.containsKey(entry.getKey())) {
+        continue;
+      }
+      T item = this.items.get(entry.getKey());
+      if (item != null) {
+        if (!Objects.equals(item.getMetadata().getResourceVersion(), entry.getValue())) {
+          List<String> values = index.indexer.apply(item);
+          if (values == null || !values.contains(indexKey)) {
+            continue; // out-of-date
+          }
+        }
+        result.put(entry.getKey(), item);
+      }
+    }
   }
 
   /**
@@ -271,26 +280,19 @@ public List<T> byIndex(String indexName, String indexKey) {
    * @param key the key
    */
   private void updateIndices(T oldObj, T newObj, String key) {
-    for (Map.Entry<String, Function<T, List<String>>> indexEntry : indexers.entrySet()) {
-      String indexName = indexEntry.getKey();
-      Function<T, List<String>> indexFunc = indexEntry.getValue();
-      Index index = this.indices.get(indexName);
-      if (index != null) {
-        updateIndex(key, oldObj, newObj, indexFunc, index);
-      }
-    }
+    indices.values().forEach(i -> updateIndex(key, oldObj, newObj, i));
   }
 
-  private void updateIndex(String key, T oldObj, T newObj, Function<T, List<String>> indexFunc, Index index) {
-    List<String> oldValues = getIndexValues(oldObj, indexFunc);
-    Collection<String> newIndexValues = new LinkedHashSet<>(getIndexValues(newObj, indexFunc));
+  private void updateIndex(String key, T oldObj, T newObj, Index<T> index) {
+    List<String> oldValues = getIndexValues(oldObj, index.indexer);
+    Collection<String> newIndexValues = new LinkedHashSet<>(getIndexValues(newObj, index.indexer));
     for (String indexValue : oldValues) {
-      if (!newIndexValues.remove(indexValue)) {
-        index.update(indexValue, key, true);
+      if (!newIndexValues.contains(indexValue)) {
+        index.update(indexValue, key, null, true);
       }
     }
     for (String indexValue : newIndexValues) {
-      index.update(indexValue, key, false);
+      index.update(indexValue, key, newObj.getMetadata().getResourceVersion(), false);
     }
   }
 
@@ -314,11 +316,10 @@ public synchronized CacheImpl<T> addIndexFunc(String indexName, Function<T, List
     if (this.indices.containsKey(indexName)) {
       throw new IllegalArgumentException("Indexer conflict: " + indexName);
     }
-    Index index = new Index();
+    Index<T> index = new Index<>(indexFunc);
     this.indices.put(indexName, index);
-    this.indexers.put(indexName, indexFunc);
 
-    items.values().forEach(v -> updateIndex(getKey(v), null, v, indexFunc, index));
+    items.values().forEach(v -> updateIndex(getKey(v), null, v, index));
     return this;
   }
 
@@ -382,7 +383,6 @@ public static List<String> metaNamespaceIndexFunc(Object obj) {
   @Override
   public synchronized void removeIndexer(String name) {
     this.indices.remove(name);
-    this.indexers.remove(name);
   }
 
   public boolean isFullState() {