feat: add dynamic cache limit for snapshot repository data based on heap size

- `opensearch.snapshot.cache.size.percentage` JVM flag to adjust cache size as a percentage of heap.
- Default remains 500KB to minimize impact on existing users.

Signed-off-by: inpink <[email protected]>

Author: inpink

CHANGELOG.md

@@ -33,6 +33,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for restoring from snapshot with search replicas ([#16111](https://github.com/opensearch-project/OpenSearch/pull/16111))
 - Add logic in master service to optimize performance and retain detailed logging for critical cluster operations. ([#14795](https://github.com/opensearch-project/OpenSearch/pull/14795))
 - Add Setting to adjust the primary constraint weights ([#16471](https://github.com/opensearch-project/OpenSearch/pull/16471))
+- Add dynamic cache limit for snapshot repository data based on heap size ([#16489](https://github.com/opensearch-project/OpenSearch/pull/16489))
 
 ### Dependencies
 - Bump `com.azure:azure-identity` from 1.13.0 to 1.13.2 ([#15578](https://github.com/opensearch-project/OpenSearch/pull/15578))

server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java

@@ -3050,12 +3050,15 @@ private void cacheRepositoryData(BytesReference updated, long generation) {
             try {
                 serialized = CompressorRegistry.defaultCompressor().compress(updated);
                 final int len = serialized.length();
-                if (len > ByteSizeUnit.KB.toBytes(500)) {
+                long dynamicCacheSizeLimit = determineCacheSizeLimit();
+
+                if (len > dynamicCacheSizeLimit) {
                     logger.debug(
-                        "Not caching repository data of size [{}] for repository [{}] because it is larger than 500KB in"
+                        "Not caching repository data of size [{}] for repository [{}] because it is larger than [{}] bytes in"
                             + " serialized size",
                         len,
-                        metadata.name()
+                        metadata.name(),
+                        dynamicCacheSizeLimit
                     );
                     if (len > ByteSizeUnit.MB.toBytes(5)) {
                         logger.warn(
@@ -3083,6 +3086,22 @@ private void cacheRepositoryData(BytesReference updated, long generation) {
         }
     }
 
+    protected long determineCacheSizeLimit() {
+        final String cacheSizeProperty = System.getProperty("opensearch.snapshot.cache.size.percentage");
+        long defaultCacheSizeLimit = ByteSizeUnit.KB.toBytes(500);
+
+        if (Strings.hasLength(cacheSizeProperty)) {
+            long maxHeapSize = Runtime.getRuntime().maxMemory();
+            double cacheSizeRatio = Double.parseDouble(cacheSizeProperty) / 100.0;
+
+            long heapBasedCacheSize = (long) (maxHeapSize * cacheSizeRatio);
+
+            return Math.max(defaultCacheSizeLimit, heapBasedCacheSize);
+        }
+
+        return defaultCacheSizeLimit;
+    }
+
     private RepositoryData repositoryDataFromCachedEntry(Tuple<Long, BytesReference> cacheEntry) throws IOException {
         try (InputStream input = CompressorRegistry.defaultCompressor().threadLocalInputStream(cacheEntry.v2().streamInput())) {
             return RepositoryData.snapshotsFromXContent(

server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java

@@ -111,6 +111,8 @@ public class BlobStoreRepositoryTests extends BlobStoreRepositoryHelperTests {
 
     static final String REPO_TYPE = "fsLike";
 
+    static final String CACHE_SIZE_PROPERTY = "opensearch.snapshot.cache.size.percentage";
+
     protected Collection<Class<? extends Plugin>> getPlugins() {
         return Arrays.asList(FsLikeRepoPlugin.class);
     }
@@ -653,4 +655,37 @@ public void testGetRestrictedSystemRepositorySettings() {
         assertTrue(settings.contains(BlobStoreRepository.REMOTE_STORE_INDEX_SHALLOW_COPY));
         repository.close();
     }
+
+    public void testCacheSizeUsesHeapPercentageWhenFlagEnabled() {
+        // given
+        BlobStoreRepository blobStoreRepository = setupRepo();
+        double percentage = 1.5;
+        System.setProperty(CACHE_SIZE_PROPERTY, String.valueOf(percentage));
+
+        long maxHeapSize = Runtime.getRuntime().maxMemory();
+        double cacheSizeRatio = percentage / 100.0;
+        long defaultCacheSize = 500 * 1024;
+        long heapBasedCacheSize = (long) (maxHeapSize * cacheSizeRatio);
+        long expectedCacheSize = Math.max(defaultCacheSize, heapBasedCacheSize);
+
+        // when
+        long actualCacheSize = blobStoreRepository.determineCacheSizeLimit();
+
+        // then
+        assertEquals(expectedCacheSize, actualCacheSize);
+    }
+
+    public void testDefaultCacheSizeWhenFlagDisabled() {
+        // given
+        System.clearProperty(CACHE_SIZE_PROPERTY);
+
+        BlobStoreRepository blobStoreRepository = setupRepo();
+        long expectedCacheSize = 500 * 1024;
+
+        // when
+        long actualCacheSize = blobStoreRepository.determineCacheSizeLimit();
+
+        // then
+        assertEquals(expectedCacheSize, actualCacheSize);
+    }
 }