Add changes to support pinning in FileCache

CHANGELOG.md

@@ -30,10 +30,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for `matched_fields` with the unified highlighter ([#18164](https://github.com/opensearch-project/OpenSearch/issues/18164))
 - [repository-s3] Add support for SSE-KMS and S3 bucket owner verification ([#18312](https://github.com/opensearch-project/OpenSearch/pull/18312))
 - Added File Cache Stats - Involves Block level as well as full file level stats ([#17538](https://github.com/opensearch-project/OpenSearch/issues/17479))
+- Added File Cache Pinning ([#17617](https://github.com/opensearch-project/OpenSearch/issues/13648))
 
 ### Changed
 - Create generic DocRequest to better categorize ActionRequests ([#18269](https://github.com/opensearch-project/OpenSearch/pull/18269)))
-    
+
 ### Dependencies
 - Update Apache Lucene from 10.1.0 to 10.2.1 ([#17961](https://github.com/opensearch-project/OpenSearch/pull/17961))
 - Bump `com.google.code.gson:gson` from 2.12.1 to 2.13.1 ([#17923](https://github.com/opensearch-project/OpenSearch/pull/17923), [#18266](https://github.com/opensearch-project/OpenSearch/pull/18266))

server/src/main/java/org/opensearch/index/store/CompositeDirectory.java

@@ -327,19 +327,15 @@ public String toString() {
      */
     public void afterSyncToRemote(String file) {
         ensureOpen();
-        /*
-        Decrementing the refCount here for the path so that it becomes eligible for eviction
-        This is a temporary solution until pinning support is added
-        TODO - Unpin the files here from FileCache so that they become eligible for eviction, once pinning/unpinning support is added in FileCache
-        Uncomment the below commented line(to remove the file from cache once uploaded) to test block based functionality
-         */
+
         logger.trace(
             "Composite Directory[{}]: File {} uploaded to Remote Store and now can be eligible for eviction in FileCache",
             this::toString,
             () -> file
         );
-        fileCache.decRef(getFilePath(file));
-        // fileCache.remove(getFilePath(fileName));
+        final Path filePath = getFilePath(file);
+        fileCache.unpin(filePath);
+        // fileCache.remove(filePath);
     }
 
     // Visibility public since we need it in IT tests
@@ -390,12 +386,10 @@ private String[] getRemoteFiles() throws IOException {
 
     protected void cacheFile(String name) throws IOException {
         Path filePath = getFilePath(name);
-        // put will increase the refCount for the path, making sure it is not evicted, will decrease the ref after it is uploaded to Remote
-        // so that it can be evicted after that
-        // this is just a temporary solution, will pin the file once support for that is added in FileCache
-        // TODO : Pin the above filePath in the file cache once pinning support is added so that it cannot be evicted unless it has been
-        // successfully uploaded to Remote
+
         fileCache.put(filePath, new CachedFullFileIndexInput(fileCache, filePath, localDirectory.openInput(name, IOContext.DEFAULT)));
+        fileCache.pin(filePath);
+        fileCache.decRef(filePath);
     }
 
 }

server/src/main/java/org/opensearch/index/store/remote/filecache/AggregateFileCacheStats.java

@@ -39,24 +39,28 @@ public class AggregateFileCacheStats implements Writeable, ToXContentFragment {
     private final FileCacheStats overallFileCacheStats;
     private final FileCacheStats fullFileCacheStats;
     private final FileCacheStats blockFileCacheStats;
+    private final FileCacheStats pinnedFileCacheStats;
 
     public AggregateFileCacheStats(
         final long timestamp,
         final FileCacheStats overallFileCacheStats,
         final FileCacheStats fullFileCacheStats,
-        final FileCacheStats blockFileCacheStats
+        final FileCacheStats blockFileCacheStats,
+        FileCacheStats pinnedFileCacheStats
     ) {
         this.timestamp = timestamp;
         this.overallFileCacheStats = overallFileCacheStats;
         this.fullFileCacheStats = fullFileCacheStats;
         this.blockFileCacheStats = blockFileCacheStats;
+        this.pinnedFileCacheStats = pinnedFileCacheStats;
     }
 
     public AggregateFileCacheStats(final StreamInput in) throws IOException {
         this.timestamp = in.readLong();
         this.overallFileCacheStats = new FileCacheStats(in);
         this.fullFileCacheStats = new FileCacheStats(in);
         this.blockFileCacheStats = new FileCacheStats(in);
+        this.pinnedFileCacheStats = new FileCacheStats(in);
     }
 
     public static short calculatePercentage(long used, long max) {
@@ -69,6 +73,7 @@ public void writeTo(final StreamOutput out) throws IOException {
         overallFileCacheStats.writeTo(out);
         fullFileCacheStats.writeTo(out);
         blockFileCacheStats.writeTo(out);
+        pinnedFileCacheStats.writeTo(out);
     }
 
     public long getTimestamp() {
@@ -91,6 +96,10 @@ public ByteSizeValue getUsed() {
         return new ByteSizeValue(overallFileCacheStats.getUsed());
     }
 
+    public ByteSizeValue getPinnedUsage() {
+        return new ByteSizeValue(overallFileCacheStats.getPinnedUsage());
+    }
+
     public short getUsedPercent() {
         return calculatePercentage(getUsed().getBytes(), getTotal().getBytes());
     }
@@ -114,6 +123,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.humanReadableField(Fields.ACTIVE_IN_BYTES, Fields.ACTIVE, getActive());
         builder.humanReadableField(Fields.TOTAL_IN_BYTES, Fields.TOTAL, getTotal());
         builder.humanReadableField(Fields.USED_IN_BYTES, Fields.USED, getUsed());
+        builder.humanReadableField(Fields.PINNED_IN_BYTES, Fields.PINNED, getPinnedUsage());
         builder.humanReadableField(Fields.EVICTIONS_IN_BYTES, Fields.EVICTIONS, getEvicted());
         builder.field(Fields.ACTIVE_PERCENT, getActivePercent());
         builder.field(Fields.USED_PERCENT, getUsedPercent());
@@ -122,7 +132,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         overallFileCacheStats.toXContent(builder, params);
         fullFileCacheStats.toXContent(builder, params);
         blockFileCacheStats.toXContent(builder, params);
-
+        pinnedFileCacheStats.toXContent(builder, params);
         builder.endObject();
         return builder;
     }
@@ -134,6 +144,8 @@ static final class Fields {
         static final String ACTIVE_IN_BYTES = "active_in_bytes";
         static final String USED = "used";
         static final String USED_IN_BYTES = "used_in_bytes";
+        static final String PINNED = "pinned";
+        static final String PINNED_IN_BYTES = "pinned_in_bytes";
         static final String EVICTIONS = "evictions";
         static final String EVICTIONS_IN_BYTES = "evictions_in_bytes";
         static final String TOTAL = "total";
@@ -153,7 +165,8 @@ static final class Fields {
     public enum FileCacheStatsType {
         FULL_FILE_STATS("full_file_stats"),
         BLOCK_FILE_STATS("block_file_stats"),
-        OVER_ALL_STATS("over_all_stats");
+        OVER_ALL_STATS("over_all_stats"),
+        PINNED_FILE_STATS("pinned_file_stats");
 
         private final String fileCacheStatsType;
 

server/src/main/java/org/opensearch/index/store/remote/filecache/FileCache.java

@@ -124,6 +124,26 @@
         theCache.decRef(key);
     }
 
+    /**
+     * Pins the key in the cache, preventing it from being evicted.
+     *
+     * @param key
+     */
+    @Override
+    public void pin(Path key) {
+        theCache.pin(key);
+    }
+
+    /**
+     * Unpins the key in the cache, allowing it to be evicted.
+     *
+     * @param key
+     */
+    @Override
+    public void unpin(Path key) {
+        theCache.unpin(key);
+    }
+
     @Override
     public Integer getRef(Path key) {
         return theCache.getRef(key);
@@ -141,14 +161,22 @@
 
     @Override
     public long usage() {
-        long l = theCache.usage();
-        return l;
+        return theCache.usage();
     }
 
     @Override
     public long activeUsage() {
-        long l = theCache.activeUsage();
-        return l;
+        return theCache.activeUsage();
+    }
+
+    /**
+     * Returns the pinned usage of this cache.
+     *
+     * @return the combined pinned weight of the values in this cache.
+     */
+    @Override
+    public long pinnedUsage() {
+        return theCache.pinnedUsage();
     }
 
     @Override
@@ -225,12 +253,14 @@
         final RefCountedCacheStats overallCacheStats = stats.getOverallCacheStats();
         final RefCountedCacheStats fullFileCacheStats = stats.getFullFileCacheStats();
         final RefCountedCacheStats blockFileCacheStats = stats.getBlockFileCacheStats();
+        final RefCountedCacheStats pinnedFileCacheStats = stats.getPinnedFileCacheStats();
         return new AggregateFileCacheStats(
             System.currentTimeMillis(),
             new FileCacheStats(
                 overallCacheStats.activeUsage(),
                 capacity(),
                 overallCacheStats.usage(),
+                overallCacheStats.pinnedUsage(),
                 overallCacheStats.evictionWeight(),
                 overallCacheStats.hitCount(),
                 overallCacheStats.missCount(),
@@ -240,6 +270,7 @@
                 fullFileCacheStats.activeUsage(),
                 capacity(),
                 fullFileCacheStats.usage(),
+                fullFileCacheStats.pinnedUsage(),
                 fullFileCacheStats.evictionWeight(),
                 fullFileCacheStats.hitCount(),
                 fullFileCacheStats.missCount(),
@@ -249,10 +280,21 @@
                 blockFileCacheStats.activeUsage(),
                 capacity(),
                 blockFileCacheStats.usage(),
+                blockFileCacheStats.pinnedUsage(),
                 blockFileCacheStats.evictionWeight(),
                 blockFileCacheStats.hitCount(),
                 blockFileCacheStats.missCount(),
                 FileCacheStatsType.BLOCK_FILE_STATS
+            ),
+            new FileCacheStats(
+                pinnedFileCacheStats.activeUsage(),
+                capacity(),
+                pinnedFileCacheStats.usage(),
+                pinnedFileCacheStats.pinnedUsage(),
+                pinnedFileCacheStats.evictionWeight(),
+                pinnedFileCacheStats.hitCount(),
+                pinnedFileCacheStats.missCount(),
+                FileCacheStatsType.PINNED_FILE_STATS
             )
         );
     }

server/src/main/java/org/opensearch/index/store/remote/filecache/FileCacheStats.java

@@ -38,6 +38,7 @@ public class FileCacheStats implements Writeable, ToXContentFragment {
     private final long active;
     private final long total;
     private final long used;
+    private final long pinned;
     private final long evicted;
     private final long hits;
     private final long misses;
@@ -47,6 +48,7 @@ public FileCacheStats(
         final long active,
         long total,
         final long used,
+        final long pinned,
         final long evicted,
         final long hits,
         long misses,
@@ -55,6 +57,7 @@ public FileCacheStats(
         this.active = active;
         this.total = total;
         this.used = used;
+        this.pinned = pinned;
         this.evicted = evicted;
         this.hits = hits;
         this.misses = misses;
@@ -66,6 +69,7 @@ public FileCacheStats(final StreamInput in) throws IOException {
         this.active = in.readLong();
         this.total = in.readLong();
         this.used = in.readLong();
+        this.pinned = in.readLong();
         this.evicted = in.readLong();
         this.hits = in.readLong();
         this.misses = in.readLong();
@@ -77,6 +81,7 @@ public void writeTo(final StreamOutput out) throws IOException {
         out.writeLong(active);
         out.writeLong(total);
         out.writeLong(used);
+        out.writeLong(pinned);
         out.writeLong(evicted);
         out.writeLong(hits);
         out.writeLong(misses);
@@ -106,6 +111,10 @@ public long getTotal() {
         return total;
     }
 
+    public long getPinnedUsage() {
+        return pinned;
+    }
+
     public short getUsedPercent() {
         return calculatePercentage(getUsed(), total);
     }
@@ -122,6 +131,7 @@ static final class Fields {
         static final String ACTIVE = "active";
         static final String ACTIVE_IN_BYTES = "active_in_bytes";
         static final String USED = "used";
+        static final String PINNED = "pinned";
         static final String USED_IN_BYTES = "used_in_bytes";
         static final String EVICTIONS = "evictions";
         static final String EVICTIONS_IN_BYTES = "evictions_in_bytes";
@@ -134,6 +144,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.startObject(statsType.toString());
         builder.humanReadableField(FileCacheStats.Fields.ACTIVE_IN_BYTES, FileCacheStats.Fields.ACTIVE, new ByteSizeValue(getActive()));
         builder.humanReadableField(FileCacheStats.Fields.USED_IN_BYTES, FileCacheStats.Fields.USED, new ByteSizeValue(getUsed()));
+        builder.humanReadableField(FileCacheStats.Fields.USED_IN_BYTES, Fields.PINNED, new ByteSizeValue(getPinnedUsage()));
         builder.humanReadableField(
             FileCacheStats.Fields.EVICTIONS_IN_BYTES,
             FileCacheStats.Fields.EVICTIONS,