[Tiered Caching] Stats rework (1/3): Interfaces and implementations for individual tiers

modules/cache-common/src/main/java/org/opensearch/cache/common/tier/TieredSpilloverCache.java

@@ -11,9 +11,11 @@
 import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.common.cache.CacheType;
 import org.opensearch.common.cache.ICache;
+import org.opensearch.common.cache.ICacheKey;
 import org.opensearch.common.cache.LoadAwareCacheLoader;
 import org.opensearch.common.cache.RemovalListener;
 import org.opensearch.common.cache.RemovalNotification;
+import org.opensearch.common.cache.stats.CacheStats;
 import org.opensearch.common.cache.store.config.CacheConfig;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
@@ -44,7 +46,12 @@ public class TieredSpilloverCache<K, V> implements ICache<K, V> {
 
     private final ICache<K, V> diskCache;
     private final ICache<K, V> onHeapCache;
-    private final RemovalListener<K, V> removalListener;
+
+    // The listener for removals from the spillover cache as a whole
+    // TODO: In TSC stats PR, each tier will have its own separate removal listener.
+    private final RemovalListener<ICacheKey<K>, V> removalListener;
+    private final CacheStats stats;
+    private final List<String> dimensionNames;
     ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
     ReleasableLock readLock = new ReleasableLock(readWriteLock.readLock());
     ReleasableLock writeLock = new ReleasableLock(readWriteLock.writeLock());
@@ -59,9 +66,9 @@ public class TieredSpilloverCache<K, V> implements ICache<K, V> {
         this.removalListener = Objects.requireNonNull(builder.removalListener, "Removal listener can't be null");
 
         this.onHeapCache = builder.onHeapCacheFactory.create(
-            new CacheConfig.Builder<K, V>().setRemovalListener(new RemovalListener<K, V>() {
+            new CacheConfig.Builder<K, V>().setRemovalListener(new RemovalListener<ICacheKey<K>, V>() {
                 @Override
-                public void onRemoval(RemovalNotification<K, V> notification) {
+                public void onRemoval(RemovalNotification<ICacheKey<K>, V> notification) {
                     try (ReleasableLock ignore = writeLock.acquire()) {
                         diskCache.put(notification.getKey(), notification.getValue());
                     }
@@ -71,13 +78,25 @@ public void onRemoval(RemovalNotification<K, V> notification) {
                 .setValueType(builder.cacheConfig.getValueType())
                 .setSettings(builder.cacheConfig.getSettings())
                 .setWeigher(builder.cacheConfig.getWeigher())
+                .setDimensionNames(builder.cacheConfig.getDimensionNames())
                 .build(),
             builder.cacheType,
             builder.cacheFactories
 
         );
-        this.diskCache = builder.diskCacheFactory.create(builder.cacheConfig, builder.cacheType, builder.cacheFactories);
+        this.diskCache = builder.diskCacheFactory.create(
+            new CacheConfig.Builder<K, V>().setRemovalListener(removalListener) // TODO: change
+                .setKeyType(builder.cacheConfig.getKeyType())
+                .setValueType(builder.cacheConfig.getValueType())
+                .setSettings(builder.cacheConfig.getSettings())
+                .setWeigher(builder.cacheConfig.getWeigher())
+                .build(),
+            builder.cacheType,
+            builder.cacheFactories
+        );
         this.cacheList = Arrays.asList(onHeapCache, diskCache);
+        this.stats = null; // TODO - in next stats rework PR
+        this.dimensionNames = builder.cacheConfig.getDimensionNames();
     }
 
     // Package private for testing
@@ -91,19 +110,19 @@ ICache<K, V> getDiskCache() {
     }
 
     @Override
-    public V get(K key) {
+    public V get(ICacheKey<K> key) {
         return getValueFromTieredCache().apply(key);
     }
 
     @Override
-    public void put(K key, V value) {
+    public void put(ICacheKey<K> key, V value) {
         try (ReleasableLock ignore = writeLock.acquire()) {
             onHeapCache.put(key, value);
         }
     }
 
     @Override
-    public V computeIfAbsent(K key, LoadAwareCacheLoader<K, V> loader) throws Exception {
+    public V computeIfAbsent(ICacheKey<K> key, LoadAwareCacheLoader<ICacheKey<K>, V> loader) throws Exception {
 
         V cacheValue = getValueFromTieredCache().apply(key);
         if (cacheValue == null) {
@@ -120,7 +139,7 @@ public V computeIfAbsent(K key, LoadAwareCacheLoader<K, V> loader) throws Except
     }
 
     @Override
-    public void invalidate(K key) {
+    public void invalidate(ICacheKey<K> key) {
         // We are trying to invalidate the key from all caches though it would be present in only of them.
         // Doing this as we don't know where it is located. We could do a get from both and check that, but what will
         // also trigger a hit/miss listener event, so ignoring it for now.
@@ -146,7 +165,7 @@ public void invalidateAll() {
      */
     @SuppressWarnings("unchecked")
     @Override
-    public Iterable<K> keys() {
+    public Iterable<ICacheKey<K>> keys() {
         return Iterables.concat(onHeapCache.keys(), diskCache.keys());
     }
 
@@ -175,7 +194,12 @@ public void close() throws IOException {
         }
     }
 
-    private Function<K, V> getValueFromTieredCache() {
+    @Override
+    public CacheStats stats() {
+        return stats;
+    }
+
+    private Function<ICacheKey<K>, V> getValueFromTieredCache() {
         return key -> {
             try (ReleasableLock ignore = readLock.acquire()) {
                 for (ICache<K, V> cache : cacheList) {
@@ -253,7 +277,7 @@ public String getCacheName() {
     public static class Builder<K, V> {
         private ICache.Factory onHeapCacheFactory;
         private ICache.Factory diskCacheFactory;
-        private RemovalListener<K, V> removalListener;
+        private RemovalListener<ICacheKey<K>, V> removalListener;
         private CacheConfig<K, V> cacheConfig;
         private CacheType cacheType;
         private Map<String, ICache.Factory> cacheFactories;
@@ -288,7 +312,7 @@ public Builder<K, V> setDiskCacheFactory(ICache.Factory diskCacheFactory) {
          * @param removalListener Removal listener
          * @return builder
          */
-        public Builder<K, V> setRemovalListener(RemovalListener<K, V> removalListener) {
+        public Builder<K, V> setRemovalListener(RemovalListener<ICacheKey<K>, V> removalListener) {
             this.removalListener = removalListener;
             return this;
         }

modules/cache-common/src/test/java/org/opensearch/cache/common/tier/MockDiskCache.java

@@ -10,10 +10,12 @@
 
 import org.opensearch.common.cache.CacheType;
 import org.opensearch.common.cache.ICache;
+import org.opensearch.common.cache.ICacheKey;
 import org.opensearch.common.cache.LoadAwareCacheLoader;
 import org.opensearch.common.cache.RemovalListener;
 import org.opensearch.common.cache.RemovalNotification;
 import org.opensearch.common.cache.RemovalReason;
+import org.opensearch.common.cache.stats.CacheStats;
 import org.opensearch.common.cache.store.builders.ICacheBuilder;
 import org.opensearch.common.cache.store.config.CacheConfig;
 
@@ -22,27 +24,27 @@
 
 public class MockDiskCache<K, V> implements ICache<K, V> {
 
-    Map<K, V> cache;
+    Map<ICacheKey<K>, V> cache;
     int maxSize;
     long delay;
 
-    private final RemovalListener<K, V> removalListener;
+    private final RemovalListener<ICacheKey<K>, V> removalListener;
 
-    public MockDiskCache(int maxSize, long delay, RemovalListener<K, V> removalListener) {
+    public MockDiskCache(int maxSize, long delay, RemovalListener<ICacheKey<K>, V> removalListener) {
         this.maxSize = maxSize;
         this.delay = delay;
         this.removalListener = removalListener;
-        this.cache = new ConcurrentHashMap<K, V>();
+        this.cache = new ConcurrentHashMap<ICacheKey<K>, V>();
     }
 
     @Override
-    public V get(K key) {
+    public V get(ICacheKey<K> key) {
         V value = cache.get(key);
         return value;
     }
 
     @Override
-    public void put(K key, V value) {
+    public void put(ICacheKey<K> key, V value) {
         if (this.cache.size() >= maxSize) { // For simplification
             this.removalListener.onRemoval(new RemovalNotification<>(key, value, RemovalReason.EVICTED));
         }
@@ -55,7 +57,7 @@ public void put(K key, V value) {
     }
 
     @Override
-    public V computeIfAbsent(K key, LoadAwareCacheLoader<K, V> loader) {
+    public V computeIfAbsent(ICacheKey<K> key, LoadAwareCacheLoader<ICacheKey<K>, V> loader) {
         V value = cache.computeIfAbsent(key, key1 -> {
             try {
                 return loader.load(key);
@@ -67,7 +69,7 @@ public V computeIfAbsent(K key, LoadAwareCacheLoader<K, V> loader) {
     }
 
     @Override
-    public void invalidate(K key) {
+    public void invalidate(ICacheKey<K> key) {
         this.cache.remove(key);
     }
 
@@ -77,7 +79,7 @@ public void invalidateAll() {
     }
 
     @Override
-    public Iterable<K> keys() {
+    public Iterable<ICacheKey<K>> keys() {
         return this.cache.keySet();
     }
 
@@ -89,6 +91,11 @@ public long count() {
     @Override
     public void refresh() {}
 
+    @Override
+    public CacheStats stats() {
+        return null;
+    }
+
     @Override
     public void close() {