[Tiered Caching] Clear up disk cache(ehcache) files during node shutd…

…own (#12734)

* Adding logic to clear up the disk cache files during close()
* Adding logic to update entries count after invalidateAll()
* Removing unneeded system log statement
* Added comment in test for readability
* Fixing issue where we were sending compacted byte[] array to ehcache but calculating size with padded byte[]

---------

Signed-off-by: Sagar Upadhyaya <[email protected]>
Signed-off-by: Sagar <[email protected]>

libs/core/src/main/java/org/opensearch/core/common/bytes/BytesReference.java

@@ -82,6 +82,11 @@ static byte[] toBytes(BytesReference reference) {
         return ArrayUtil.copyOfSubArray(bytesRef.bytes, bytesRef.offset, bytesRef.offset + bytesRef.length);
     }
 
+    static byte[] toBytesWithoutCompact(BytesReference reference) {
+        final BytesRef bytesRef = reference.toBytesRef();
+        return bytesRef.bytes;
+    }
+
     /**
      * Returns an array of byte buffers from the given BytesReference.
      */

plugins/cache-ehcache/src/main/java/org/opensearch/cache/store/disk/EhcacheDiskCache.java

@@ -10,6 +10,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.opensearch.OpenSearchException;
 import org.opensearch.cache.EhcacheDiskCacheSettings;
 import org.opensearch.common.SuppressForbidden;
@@ -28,9 +29,14 @@
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.util.io.IOUtils;
 
 import java.io.File;
+import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Iterator;
@@ -363,7 +369,10 @@ public void invalidate(K key) {
     }
 
     @Override
-    public void invalidateAll() {}
+    public void invalidateAll() {
+        cache.clear();
+        this.entries.dec(this.entries.count()); // reset to zero.
+    }
 
     /**
      * Provides a way to iterate over disk cache keys.
@@ -389,13 +398,21 @@ public void refresh() {
     }
 
     @Override
+    @SuppressForbidden(reason = "Ehcache uses File.io")
     public void close() {
         cacheManager.removeCache(this.diskCacheAlias);
         cacheManager.close();
         try {
             cacheManager.destroyCache(this.diskCacheAlias);
+            // Delete all the disk cache related files/data
+            Path ehcacheDirectory = Paths.get(this.storagePath);
+            if (Files.exists(ehcacheDirectory)) {
+                IOUtils.rm(ehcacheDirectory);
+            }
         } catch (CachePersistenceException e) {
             throw new OpenSearchException("Exception occurred while destroying ehcache and associated data", e);
+        } catch (IOException e) {
+            logger.error(() -> new ParameterizedMessage("Failed to delete ehcache disk cache data under path: {}", this.storagePath));
         }
     }
 

plugins/cache-ehcache/src/test/java/org/opensearch/cache/store/disk/EhCacheDiskCacheTests.java

@@ -554,6 +554,41 @@ public void testEhcacheKeyIteratorWithRemove() throws IOException {
 
     }
 
+    public void testInvalidateAll() throws Exception {
+        Settings settings = Settings.builder().build();
+        MockRemovalListener<String, String> removalListener = new MockRemovalListener<>();
+        try (NodeEnvironment env = newNodeEnvironment(settings)) {
+            ICache<String, String> ehcacheTest = new EhcacheDiskCache.Builder<String, String>().setThreadPoolAlias("ehcacheTest")
+                .setStoragePath(env.nodePaths()[0].indicesPath.toString() + "/request_cache")
+                .setIsEventListenerModeSync(true)
+                .setKeyType(String.class)
+                .setValueType(String.class)
+                .setKeySerializer(new StringSerializer())
+                .setValueSerializer(new StringSerializer())
+                .setCacheType(CacheType.INDICES_REQUEST_CACHE)
+                .setSettings(settings)
+                .setExpireAfterAccess(TimeValue.MAX_VALUE)
+                .setMaximumWeightInBytes(CACHE_SIZE_IN_BYTES)
+                .setRemovalListener(removalListener)
+                .build();
+            int randomKeys = randomIntBetween(10, 100);
+            Map<String, String> keyValueMap = new HashMap<>();
+            for (int i = 0; i < randomKeys; i++) {
+                keyValueMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+            }
+            for (Map.Entry<String, String> entry : keyValueMap.entrySet()) {
+                ehcacheTest.put(entry.getKey(), entry.getValue());
+            }
+            ehcacheTest.invalidateAll(); // clear all the entries.
+            for (Map.Entry<String, String> entry : keyValueMap.entrySet()) {
+                // Verify that value is null for a removed entry.
+                assertNull(ehcacheTest.get(entry.getKey()));
+            }
+            assertEquals(0, ehcacheTest.count());
+            ehcacheTest.close();
+        }
+    }
+
     public void testBasicGetAndPutBytesReference() throws Exception {
         Settings settings = Settings.builder().build();
         try (NodeEnvironment env = newNodeEnvironment(settings)) {

server/src/main/java/org/opensearch/common/cache/serializer/BytesReferenceSerializer.java

@@ -24,7 +24,7 @@ public BytesReferenceSerializer() {}
 
     @Override
     public byte[] serialize(BytesReference object) {
-        return BytesReference.toBytes(object);
+        return BytesReference.toBytesWithoutCompact(object);
     }
 
     @Override

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

@@ -185,8 +185,9 @@ public final class IndicesRequestCache implements RemovalListener<IndicesRequest
     }
 
     @Override
-    public void close() {
+    public void close() throws IOException {
         cache.invalidateAll();
+        cache.close();
         cacheCleanupManager.close();
     }