Cleanup

server/src/main/java/org/opensearch/indices/EhcacheDiskCachingTier.java

@@ -34,7 +34,7 @@
 public class EhcacheDiskCachingTier<K, V> implements CachingTier<K, V>, RemovalListener<K, V> {
 
     private final PersistentCacheManager cacheManager;
-    public final Cache<K, V> cache; // make private after debug
+    private final Cache<K, V> cache; // make private after debug
 
     private final Class<K> keyType; // I think these are needed to pass to newCacheConfigurationBuilder
     private final Class<V> valueType;
@@ -48,6 +48,7 @@ public class EhcacheDiskCachingTier<K, V> implements CachingTier<K, V>, RemovalL
     private static final String cacheAlias = "diskTier";
     private final boolean isPersistent;
     private CounterMetric count; // number of entries in cache
+    private EhcacheEventListener listener;
     // private RBMIntKeyLookupStore keystore;
     // private CacheTierPolicy[] policies;
     // private IndicesRequestCacheDiskTierPolicy policy;
@@ -57,17 +58,18 @@ public EhcacheDiskCachingTier(boolean isPersistent, long maxWeightInBytes, long
         this.valueType = valueType;
         this.isPersistent = isPersistent;
         this.count = new CounterMetric();
+        this.listener = new EhcacheEventListener<K, V>(this, this.count);
         statsService = new DefaultStatisticsService();
 
         // our EhcacheEventListener should receive events every time an entry is changed
         CacheEventListenerConfigurationBuilder listenerConfig = CacheEventListenerConfigurationBuilder
-            .newEventListenerConfiguration(new EhcacheEventListener<K, V>(this, this.count),
+            .newEventListenerConfiguration(listener,
                 EventType.EVICTED,
                 EventType.EXPIRED,
                 EventType.REMOVED,
                 EventType.UPDATED,
-                EventType.CREATED);
-            //.ordered().asynchronous(); // ordered() has some performance penalty as compared to unordered(), we can also use synchronous()
+                EventType.CREATED)
+            .ordered().asynchronous(); // ordered() has some performance penalty as compared to unordered(), we can also use synchronous()
 
         PooledExecutionServiceConfiguration threadConfig = PooledExecutionServiceConfigurationBuilder.newPooledExecutionServiceConfigurationBuilder()
             .defaultPool("default", MIN_WRITE_THREADS, MAX_WRITE_THREADS)
@@ -79,7 +81,7 @@ public EhcacheDiskCachingTier(boolean isPersistent, long maxWeightInBytes, long
             .with(CacheManagerBuilder.persistence(DISK_CACHE_FP))
             .withCache(cacheAlias, CacheConfigurationBuilder.newCacheConfigurationBuilder(
                 keyType, valueType, ResourcePoolsBuilder.newResourcePoolsBuilder().disk(maxWeightInBytes, MemoryUnit.B, isPersistent))
-                .withService(listenerConfig) // stackoverflow shows .add(), but IDE says this is deprecated. idk
+                .withService(listenerConfig)
             ).build(true);
         this.cache = cacheManager.getCache(cacheAlias, keyType, valueType);
         this.getTimeMillisEWMA = new ExponentiallyWeightedMovingAverage(GET_TIME_EWMA_ALPHA, 10);
@@ -91,7 +93,7 @@ public EhcacheDiskCachingTier(boolean isPersistent, long maxWeightInBytes, long
 
     @Override
     public V get(K key) {
-        // do we need to do the future stuff? I don't think so?
+        // I don't think we need to do the future stuff as the cache is threadsafe
 
         // if (keystore.contains(key.hashCode()) {
         long now = System.nanoTime();
@@ -101,7 +103,6 @@ public V get(K key) {
         return value;
         // }
         // return null;
-
     }
 
     @Override
@@ -111,7 +112,6 @@ public void put(K key, V value) {
         // CheckDataResult policyResult = policy.checkData(value)
         // if (policyResult.isAccepted()) {
         cache.put(key, value);
-        //count++;
         // keystore.add(key.hashCode());
         // else { do something with policyResult.deniedReason()? }
         // }
@@ -129,7 +129,6 @@ public void invalidate(K key) {
 
         // if (keystore.contains(key.hashCode()) {
         cache.remove(key);
-        //count--;
         // keystore.remove(key.hashCode());
         // }
     }
@@ -159,7 +158,6 @@ public Iterable<K> keys() {
     @Override
     public int count() {
         return (int) count.count();
-        //return (int) getTierStats().getMappings();
     }
 
     protected void countInc() {
@@ -183,35 +181,6 @@ public double getTimeMillisEWMA() {
         return getTimeMillisEWMA.getAverage();
     }
 
-    // these aren't really needed, ShardRequestCache handles it
-    // Also, it seems that ehcache doesn't have functioning statistics anyway!
-
-    /*public TierStatistics getTierStats() {
-        return statsService.getCacheStatistics(cacheAlias).getTierStatistics().get("Disk");
-    }
-
-    public long getHits() {
-        return getTierStats().getHits();
-    }
-
-    public long getMisses() {
-        return getTierStats().getMisses();
-    }
-
-    public long getWeightBytes() {
-        return getTierStats().getOccupiedByteSize();
-    }
-
-    public long getEvictions() {
-        return getTierStats().getEvictions();
-    }
-
-    public double getHitRatio() {
-        TierStatistics ts = getTierStats();
-        long hits = ts.getHits();
-        return hits / (hits + ts.getMisses());
-    }*/
-
     public boolean isPersistent() {
         return isPersistent;
     }
@@ -222,12 +191,4 @@ public void close() {
         cacheManager.removeCache(cacheAlias);
         cacheManager.close();
     }
-
-
-    // See https://stackoverflow.com/questions/45827753/listenerobject-not-found-in-imports-for-ehcache-3 for API reference
-    // it's not actually documented by ehcache :(
-    // This class is used to get the old value from mutating calls to the cache, and it uses those to create a RemovalNotification
-    // It also handles incrementing and decrementing the count for the disk tier, since ehcache's statistics functionality
-    // does not seem to work
-
 }

server/src/main/java/org/opensearch/indices/IndicesRequestCache.java

@@ -46,7 +46,6 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.concurrent.ConcurrentCollections;
-import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.unit.ByteSizeValue;
 
@@ -107,8 +106,8 @@ public final class IndicesRequestCache implements TieredCacheEventListener<Indic
     private final TimeValue expire;
     // private final Cache<Key, BytesReference> cache;
 
-    //private final TieredCacheHandler<Key, BytesReference> tieredCacheHandler; // made public TieredCacheSpilloverStrategyHandler for testing
-    public final TieredCacheSpilloverStrategyHandler<Key, BytesReference> tieredCacheHandler;
+    //private final TieredCacheHandler<Key, BytesReference> tieredCacheHandler;
+    public final TieredCacheSpilloverStrategyHandler<Key, BytesReference> tieredCacheHandler; // Change this back after done debugging serialization issues
     IndicesRequestCache(Settings settings) {
         this.size = INDICES_CACHE_QUERY_SIZE.get(settings);
         this.expire = INDICES_CACHE_QUERY_EXPIRE.exists(settings) ? INDICES_CACHE_QUERY_EXPIRE.get(settings) : null;

server/src/main/java/org/opensearch/indices/TieredCacheSpilloverStrategyHandler.java

@@ -36,7 +36,7 @@ public class TieredCacheSpilloverStrategyHandler<K, V> implements TieredCacheHan
     private TieredCacheSpilloverStrategyHandler(
         OnHeapCachingTier<K, V> onHeapCachingTier,
         EhcacheDiskCachingTier<K, V> diskCachingTier,
-        // changed to EhcacheDiskCachingTier from CachingTier, to enable close() method, which is needed for tests. Implement close() in CachingTier or DiskCachingTier or something?
+        // changed to EhcacheDiskCachingTier from CachingTier, to enable close() method, which is needed for tests. Implement close() in CachingTier or DiskCachingTier?
         TieredCacheEventListener<K, V> tieredCacheEventListener
     ) {
         this.onHeapCachingTier = Objects.requireNonNull(onHeapCachingTier);

server/src/test/java/org/opensearch/indices/IndicesRequestCacheTests.java

@@ -142,6 +142,9 @@ public void testBasicOperationsCache() throws Exception {
     }
 
     public void testAddDirectToEhcache() throws Exception {
+        // this test is for debugging serialization issues and can eventually be removed
+        // Put breakpoints at line 260 of AbstractOffHeapStore to catch serialization errors
+        // that would otherwise fail silently
         ShardRequestCache requestCacheStats = new ShardRequestCache();
         Settings.Builder settingsBuilder = Settings.builder();
         long heapSizeBytes = 1000;
@@ -162,7 +165,9 @@ public void testAddDirectToEhcache() throws Exception {
         IndicesRequestCache.Key key = new IndicesRequestCache.Key(entity, reader.getReaderCacheHelper().getKey(), termBytes);
 
         TestBytesReference value = new TestBytesReference(124);
-        cache.tieredCacheHandler.getDiskCachingTier().cache.put(key, value);
+        cache.tieredCacheHandler.getDiskCachingTier().put(key, value);
+
+        System.out.println("Size: " + cache.tieredCacheHandler.getDiskCachingTier().count());
 
         IOUtils.close(reader, writer, dir, cache);
         cache.closeDiskTier();