Skip to content

Commit a607693

Browse files
inpinkinpink
committed
feat: add vertical scaling and SoftReference for snapshot repository data cache
- Applies `SoftReference` to cached repository data for efficient memory management under heap pressure. - Enables cache size configuration in `opensearch.yml`, adjustable within a range of 500KB to 1% of heap memory. - Sets the default cache size to `Math.max(ByteSizeUnit.KB.toBytes(500), CACHE_MAX_THRESHOLD / 2)` so it’s generally proportional to heap size. In cases where 1% of the heap is less than 1000KB, indicating a low-memory environment, the default reverts to 500KB as before. - Since `BytesReference` internally uses `byte[]`, the compressed array size is capped at `Integer.MAX_VALUE - 8` to ensure compatibility with JDK limitations on array sizes. Therefore, the maximum cache size cannot exceed this limit. Signed-off-by: inpink <[email protected]>
1 parent e688388 commit a607693

File tree

5 files changed

+208
-17
lines changed

5 files changed

+208
-17
lines changed

‎CHANGELOG.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
1313
- Switch from `buildSrc/version.properties` to Gradle version catalog (`gradle/libs.versions.toml`) to enable dependabot to perform automated upgrades on common libs ([#16284](https://github.com/opensearch-project/OpenSearch/pull/16284))
1414
- Add dynamic setting allowing size > 0 requests to be cached in the request cache ([#16483](https://github.com/opensearch-project/OpenSearch/pull/16483))
1515
- Make IndexStoreListener a pluggable interface ([#16583](https://github.com/opensearch-project/OpenSearch/pull/16583))
16+
- Add dynamic setting allowing size > 0 requests to be cached in the request cache ([#16483](https://github.com/opensearch-project/OpenSearch/pull/16483/files))
17+
- Add vertical scaling and SoftReference for snapshot repository data cache ([#16489](https://github.com/opensearch-project/OpenSearch/pull/16489))
1618

1719
### Dependencies
1820
- Bump `com.azure:azure-storage-common` from 12.25.1 to 12.27.1 ([#16521](https://github.com/opensearch-project/OpenSearch/pull/16521))

‎server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -786,6 +786,7 @@ public void apply(Settings value, Settings current, Settings previous) {
786786
// Snapshot related Settings
787787
BlobStoreRepository.SNAPSHOT_SHARD_PATH_PREFIX_SETTING,
788788
BlobStoreRepository.SNAPSHOT_ASYNC_DELETION_ENABLE_SETTING,
789+
BlobStoreRepository.SNAPSHOT_REPOSITORY_DATA_CACHE_THRESHOLD,
789790

790791
SearchService.CLUSTER_ALLOW_DERIVED_FIELD_SETTING,
791792

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

Lines changed: 96 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -142,6 +142,7 @@
142142
import org.opensearch.indices.RemoteStoreSettings;
143143
import org.opensearch.indices.recovery.RecoverySettings;
144144
import org.opensearch.indices.recovery.RecoveryState;
145+
import org.opensearch.monitor.jvm.JvmInfo;
145146
import org.opensearch.node.remotestore.RemoteStorePinnedTimestampService;
146147
import org.opensearch.repositories.IndexId;
147148
import org.opensearch.repositories.IndexMetaDataGenerations;
@@ -167,6 +168,7 @@
167168
import java.io.FilterInputStream;
168169
import java.io.IOException;
169170
import java.io.InputStream;
171+
import java.lang.ref.SoftReference;
170172
import java.nio.file.NoSuchFileException;
171173
import java.util.ArrayList;
172174
import java.util.Arrays;
@@ -196,6 +198,7 @@
196198
import java.util.stream.LongStream;
197199
import java.util.stream.Stream;
198200

201+
import static org.opensearch.common.unit.MemorySizeValue.parseBytesSizeValueOrHeapRatio;
199202
import static org.opensearch.index.remote.RemoteStoreEnums.PathHashAlgorithm.FNV_1A_COMPOSITE_1;
200203
import static org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo.canonicalName;
201204
import static org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat.SNAPSHOT_ONLY_FORMAT_PARAMS;
@@ -253,6 +256,21 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
253256
*/
254257
public static final String VIRTUAL_DATA_BLOB_PREFIX = "v__";
255258

259+
public static final String SNAPSHOT_REPOSITORY_DATA_CACHET_THRESHOLD_SETTING_NAME = "snapshot.repository_data.cache.threshold";
260+
261+
public static final double SNAPSHOT_REPOSITORY_DATA_CACHE_THRESHOLD_DEFAULT_PERCENTAGE = 0.01;
262+
263+
public static final long CACHE_MIN_THRESHOLD = ByteSizeUnit.KB.toBytes(500);
264+
265+
public static final long CACHE_MAX_THRESHOLD = calculateMaxSnapshotRepositoryDataCacheThreshold();
266+
267+
public static final long CACHE_DEFAULT_THRESHOLD = calculateDefaultSnapshotRepositoryDataCacheThreshold();
268+
269+
/**
270+
* Set to Integer.MAX_VALUE - 8 to prevent OutOfMemoryError due to array header requirements, following the limit used in certain JDK versions. This ensures compatibility across various JDK versions. For a practical usage example, see this link: https://github.com/openjdk/jdk11u/blob/cee8535a9d3de8558b4b5028d68e397e508bef71/src/jdk.zipfs/share/classes/jdk/nio/zipfs/ByteArrayChannel.java#L226
271+
*/
272+
private static final int MAX_SAFE_ARRAY_SIZE = Integer.MAX_VALUE - 8;
273+
256274
/**
257275
* When set to {@code true}, {@link #bestEffortConsistency} will be set to {@code true} and concurrent modifications of the repository
258276
* contents will not result in the repository being marked as corrupted.
@@ -275,6 +293,58 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
275293
Setting.Property.Deprecated
276294
);
277295

296+
/**
297+
* Sets the cache size for snapshot repository data: the valid range is within 500Kb ... 1% of the node heap memory.
298+
*/
299+
public static final Setting<ByteSizeValue> SNAPSHOT_REPOSITORY_DATA_CACHE_THRESHOLD = new Setting<>(
300+
SNAPSHOT_REPOSITORY_DATA_CACHET_THRESHOLD_SETTING_NAME,
301+
CACHE_DEFAULT_THRESHOLD + "b",
302+
(s) -> {
303+
ByteSizeValue userDefinedLimit = parseBytesSizeValueOrHeapRatio(s, SNAPSHOT_REPOSITORY_DATA_CACHET_THRESHOLD_SETTING_NAME);
304+
long userDefinedLimitBytes = userDefinedLimit.getBytes();
305+
306+
if (userDefinedLimitBytes > CACHE_MAX_THRESHOLD) {
307+
throw new IllegalArgumentException(
308+
"["
309+
+ SNAPSHOT_REPOSITORY_DATA_CACHET_THRESHOLD_SETTING_NAME
310+
+ "] cannot be larger than ["
311+
+ CACHE_MAX_THRESHOLD
312+
+ "] bytes."
313+
);
314+
}
315+
316+
if (userDefinedLimitBytes < CACHE_MIN_THRESHOLD) {
317+
throw new IllegalArgumentException(
318+
"["
319+
+ SNAPSHOT_REPOSITORY_DATA_CACHET_THRESHOLD_SETTING_NAME
320+
+ "] cannot be smaller than ["
321+
+ CACHE_MIN_THRESHOLD
322+
+ "] bytes."
323+
);
324+
}
325+
326+
return userDefinedLimit;
327+
},
328+
Setting.Property.NodeScope
329+
);
330+
331+
public static long calculateDefaultSnapshotRepositoryDataCacheThreshold() {
332+
return Math.max(ByteSizeUnit.KB.toBytes(500), CACHE_MAX_THRESHOLD / 2);
333+
}
334+
335+
public static long calculateMaxSnapshotRepositoryDataCacheThreshold() {
336+
long jvmHeapSize = JvmInfo.jvmInfo().getMem().getHeapMax().getBytes();
337+
long defaultThresholdOfHeap = (long) (jvmHeapSize * SNAPSHOT_REPOSITORY_DATA_CACHE_THRESHOLD_DEFAULT_PERCENTAGE);
338+
long defaultAbsoluteThreshold = ByteSizeUnit.KB.toBytes(500);
339+
long maxThreshold = calculateMaxWithinIntLimit(defaultThresholdOfHeap, defaultAbsoluteThreshold);
340+
341+
return maxThreshold;
342+
}
343+
344+
protected static long calculateMaxWithinIntLimit(long defaultThresholdOfHeap, long defaultAbsoluteThreshold) {
345+
return Math.min(Math.max(defaultThresholdOfHeap, defaultAbsoluteThreshold), MAX_SAFE_ARRAY_SIZE);
346+
}
347+
278348
/**
279349
* Size hint for the IO buffer size to use when reading from and writing to the repository.
280350
*/
@@ -461,6 +531,8 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
461531

462532
private volatile boolean enableAsyncDeletion;
463533

534+
protected final long repositoryDataCacheThreshold;
535+
464536
/**
465537
* Flag that is set to {@code true} if this instance is started with {@link #metadata} that has a higher value for
466538
* {@link RepositoryMetadata#pendingGeneration()} than for {@link RepositoryMetadata#generation()} indicating a full cluster restart
@@ -515,6 +587,7 @@ protected BlobStoreRepository(
515587
this.snapshotShardPathPrefix = SNAPSHOT_SHARD_PATH_PREFIX_SETTING.get(clusterService.getSettings());
516588
this.enableAsyncDeletion = SNAPSHOT_ASYNC_DELETION_ENABLE_SETTING.get(clusterService.getSettings());
517589
clusterService.getClusterSettings().addSettingsUpdateConsumer(SNAPSHOT_ASYNC_DELETION_ENABLE_SETTING, this::setEnableAsyncDeletion);
590+
this.repositoryDataCacheThreshold = SNAPSHOT_REPOSITORY_DATA_CACHE_THRESHOLD.get(clusterService.getSettings()).getBytes();
518591
}
519592

520593
@Override
@@ -1132,7 +1205,8 @@ private RepositoryData safeRepositoryData(long repositoryStateId, Map<String, Bl
11321205
cached = null;
11331206
} else {
11341207
genToLoad = latestKnownRepoGen.get();
1135-
cached = latestKnownRepositoryData.get();
1208+
SoftReference<Tuple<Long, BytesReference>> softRef = latestKnownRepositoryData.get();
1209+
cached = (softRef != null) ? softRef.get() : null;
11361210
}
11371211
if (genToLoad > generation) {
11381212
// It's always a possibility to not see the latest index-N in the listing here on an eventually consistent blob store, just
@@ -2926,15 +3000,19 @@ public void endVerification(String seed) {
29263000
private final AtomicLong latestKnownRepoGen = new AtomicLong(RepositoryData.UNKNOWN_REPO_GEN);
29273001

29283002
// Best effort cache of the latest known repository data and its generation, cached serialized as compressed json
2929-
private final AtomicReference<Tuple<Long, BytesReference>> latestKnownRepositoryData = new AtomicReference<>();
3003+
private final AtomicReference<SoftReference<Tuple<Long, BytesReference>>> latestKnownRepositoryData = new AtomicReference<>(
3004+
new SoftReference<>(null)
3005+
);
29303006

29313007
@Override
29323008
public void getRepositoryData(ActionListener<RepositoryData> listener) {
29333009
if (latestKnownRepoGen.get() == RepositoryData.CORRUPTED_REPO_GEN) {
29343010
listener.onFailure(corruptedStateException(null));
29353011
return;
29363012
}
2937-
final Tuple<Long, BytesReference> cached = latestKnownRepositoryData.get();
3013+
final SoftReference<Tuple<Long, BytesReference>> softRef = latestKnownRepositoryData.get();
3014+
final Tuple<Long, BytesReference> cached = (softRef != null) ? softRef.get() : null;
3015+
29383016
// Fast path loading repository data directly from cache if we're in fully consistent mode and the cache matches up with
29393017
// the latest known repository generation
29403018
if (bestEffortConsistency == false && cached != null && cached.v1() == latestKnownRepoGen.get()) {
@@ -2983,7 +3061,8 @@ private void doGetRepositoryData(ActionListener<RepositoryData> listener) {
29833061
genToLoad = latestKnownRepoGen.get();
29843062
}
29853063
try {
2986-
final Tuple<Long, BytesReference> cached = latestKnownRepositoryData.get();
3064+
final SoftReference<Tuple<Long, BytesReference>> softRef = latestKnownRepositoryData.get();
3065+
final Tuple<Long, BytesReference> cached = (softRef != null) ? softRef.get() : null;
29873066
final RepositoryData loaded;
29883067
// Caching is not used with #bestEffortConsistency see docs on #cacheRepositoryData for details
29893068
if (bestEffortConsistency == false && cached != null && cached.v1() == genToLoad) {
@@ -3050,19 +3129,22 @@ private void cacheRepositoryData(BytesReference updated, long generation) {
30503129
try {
30513130
serialized = CompressorRegistry.defaultCompressor().compress(updated);
30523131
final int len = serialized.length();
3053-
if (len > ByteSizeUnit.KB.toBytes(500)) {
3132+
long cacheWarningThreshold = Math.min(repositoryDataCacheThreshold * 10, MAX_SAFE_ARRAY_SIZE);
3133+
if (len > repositoryDataCacheThreshold) {
30543134
logger.debug(
3055-
"Not caching repository data of size [{}] for repository [{}] because it is larger than 500KB in"
3135+
"Not caching repository data of size [{}] for repository [{}] because it is larger than [{}] bytes in"
30563136
+ " serialized size",
30573137
len,
3058-
metadata.name()
3138+
metadata.name(),
3139+
repositoryDataCacheThreshold
30593140
);
3060-
if (len > ByteSizeUnit.MB.toBytes(5)) {
3141+
if (len > cacheWarningThreshold) {
30613142
logger.warn(
3062-
"Your repository metadata blob for repository [{}] is larger than 5MB. Consider moving to a fresh"
3143+
"Your repository metadata blob for repository [{}] is larger than [{}] bytes. Consider moving to a fresh"
30633144
+ " repository for new snapshots or deleting unneeded snapshots from your repository to ensure stable"
30643145
+ " repository behavior going forward.",
3065-
metadata.name()
3146+
metadata.name(),
3147+
cacheWarningThreshold
30663148
);
30673149
}
30683150
// Set empty repository data to not waste heap for an outdated cached value
@@ -3074,11 +3156,12 @@ private void cacheRepositoryData(BytesReference updated, long generation) {
30743156
logger.warn("Failed to serialize repository data", e);
30753157
return;
30763158
}
3077-
latestKnownRepositoryData.updateAndGet(known -> {
3159+
latestKnownRepositoryData.updateAndGet(knownRef -> {
3160+
Tuple<Long, BytesReference> known = (knownRef != null) ? knownRef.get() : null;
30783161
if (known != null && known.v1() > generation) {
3079-
return known;
3162+
return knownRef;
30803163
}
3081-
return new Tuple<>(generation, serialized);
3164+
return new SoftReference<>(new Tuple<>(generation, serialized));
30823165
});
30833166
}
30843167
}

‎server/src/test/java/org/opensearch/common/settings/MemorySizeSettingsTests.java

Lines changed: 59 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,13 +34,15 @@
3434

3535
import org.opensearch.common.settings.Setting.Property;
3636
import org.opensearch.common.util.PageCacheRecycler;
37+
import org.opensearch.core.common.unit.ByteSizeUnit;
3738
import org.opensearch.core.common.unit.ByteSizeValue;
3839
import org.opensearch.indices.IndexingMemoryController;
3940
import org.opensearch.indices.IndicesQueryCache;
4041
import org.opensearch.indices.IndicesRequestCache;
4142
import org.opensearch.indices.breaker.HierarchyCircuitBreakerService;
4243
import org.opensearch.indices.fielddata.cache.IndicesFieldDataCache;
4344
import org.opensearch.monitor.jvm.JvmInfo;
45+
import org.opensearch.repositories.blobstore.BlobStoreRepository;
4446
import org.opensearch.test.OpenSearchTestCase;
4547

4648
import static org.hamcrest.Matchers.equalTo;
@@ -127,22 +129,75 @@ public void testIndicesFieldDataCacheSetting() {
127129
);
128130
}
129131

132+
public void testSnapshotRepositoryDataCacheSizeSetting() {
133+
assertMemorySizeSettingInRange(
134+
BlobStoreRepository.SNAPSHOT_REPOSITORY_DATA_CACHE_THRESHOLD,
135+
"snapshot.repository_data.cache.threshold",
136+
new ByteSizeValue(BlobStoreRepository.calculateDefaultSnapshotRepositoryDataCacheThreshold()),
137+
ByteSizeUnit.KB.toBytes(500),
138+
1.0
139+
);
140+
}
141+
130142
private void assertMemorySizeSetting(Setting<ByteSizeValue> setting, String settingKey, ByteSizeValue defaultValue) {
131143
assertMemorySizeSetting(setting, settingKey, defaultValue, Settings.EMPTY);
132144
}
133145

134146
private void assertMemorySizeSetting(Setting<ByteSizeValue> setting, String settingKey, ByteSizeValue defaultValue, Settings settings) {
147+
assertMemorySizeSetting(setting, settingKey, defaultValue, 25.0, 1024, settings);
148+
}
149+
150+
private void assertMemorySizeSetting(
151+
Setting<ByteSizeValue> setting,
152+
String settingKey,
153+
ByteSizeValue defaultValue,
154+
double availablePercentage,
155+
long availableBytes,
156+
Settings settings
157+
) {
135158
assertThat(setting, notNullValue());
136159
assertThat(setting.getKey(), equalTo(settingKey));
137160
assertThat(setting.getProperties(), hasItem(Property.NodeScope));
138161
assertThat(setting.getDefault(settings), equalTo(defaultValue));
139-
Settings settingWithPercentage = Settings.builder().put(settingKey, "25%").build();
162+
Settings settingWithPercentage = Settings.builder().put(settingKey, percentageAsString(availablePercentage)).build();
140163
assertThat(
141164
setting.get(settingWithPercentage),
142-
equalTo(new ByteSizeValue((long) (JvmInfo.jvmInfo().getMem().getHeapMax().getBytes() * 0.25)))
165+
equalTo(
166+
new ByteSizeValue((long) (JvmInfo.jvmInfo().getMem().getHeapMax().getBytes() * percentageAsFraction(availablePercentage)))
167+
)
143168
);
144-
Settings settingWithBytesValue = Settings.builder().put(settingKey, "1024b").build();
145-
assertThat(setting.get(settingWithBytesValue), equalTo(new ByteSizeValue(1024)));
169+
Settings settingWithBytesValue = Settings.builder().put(settingKey, availableBytes + "b").build();
170+
assertThat(setting.get(settingWithBytesValue), equalTo(new ByteSizeValue(availableBytes)));
146171
}
147172

173+
private void assertMemorySizeSettingInRange(
174+
Setting<ByteSizeValue> setting,
175+
String settingKey,
176+
ByteSizeValue defaultValue,
177+
long minBytes,
178+
double maxPercentage
179+
) {
180+
assertMemorySizeSetting(setting, settingKey, defaultValue, maxPercentage, minBytes, Settings.EMPTY);
181+
182+
assertThrows(IllegalArgumentException.class, () -> {
183+
Settings settingWithTooSmallValue = Settings.builder().put(settingKey, minBytes - 1).build();
184+
setting.get(settingWithTooSmallValue);
185+
});
186+
187+
assertThrows(IllegalArgumentException.class, () -> {
188+
double unavailablePercentage = maxPercentage + 0.1;
189+
Settings settingWithPercentageExceedingLimit = Settings.builder()
190+
.put(settingKey, percentageAsString(unavailablePercentage))
191+
.build();
192+
setting.get(settingWithPercentageExceedingLimit);
193+
});
194+
}
195+
196+
private double percentageAsFraction(double availablePercentage) {
197+
return availablePercentage / 100.0;
198+
}
199+
200+
private String percentageAsString(double availablePercentage) {
201+
return availablePercentage + "%";
202+
}
148203
}

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@
9292
import java.util.stream.Collectors;
9393

9494
import static org.opensearch.repositories.RepositoryDataTests.generateRandomRepoData;
95+
import static org.opensearch.repositories.blobstore.BlobStoreRepository.calculateMaxWithinIntLimit;
9596
import static org.hamcrest.Matchers.equalTo;
9697
import static org.hamcrest.Matchers.nullValue;
9798
import static org.mockito.ArgumentMatchers.any;
@@ -653,4 +654,53 @@ public void testGetRestrictedSystemRepositorySettings() {
653654
assertTrue(settings.contains(BlobStoreRepository.REMOTE_STORE_INDEX_SHALLOW_COPY));
654655
repository.close();
655656
}
657+
658+
public void testSnapshotRepositoryDataCacheDefaultSetting() {
659+
// given
660+
BlobStoreRepository repository = setupRepo();
661+
long maxThreshold = repository.calculateMaxSnapshotRepositoryDataCacheThreshold();
662+
663+
// when
664+
long expectedThreshold = Math.max(ByteSizeUnit.KB.toBytes(500), maxThreshold / 2);
665+
666+
// then
667+
assertEquals(repository.repositoryDataCacheThreshold, expectedThreshold);
668+
}
669+
670+
public void testHeapThresholdUsed() {
671+
// given
672+
long defaultThresholdOfHeap = ByteSizeUnit.GB.toBytes(1);
673+
long defaultAbsoluteThreshold = ByteSizeUnit.KB.toBytes(500);
674+
675+
// when
676+
long expectedThreshold = calculateMaxWithinIntLimit(defaultThresholdOfHeap, defaultAbsoluteThreshold);
677+
678+
// then
679+
assertEquals(defaultThresholdOfHeap, expectedThreshold);
680+
}
681+
682+
public void testAbsoluteThresholdUsed() {
683+
// given
684+
long defaultThresholdOfHeap = ByteSizeUnit.KB.toBytes(499);
685+
long defaultAbsoluteThreshold = ByteSizeUnit.KB.toBytes(500);
686+
687+
// when
688+
long result = calculateMaxWithinIntLimit(defaultThresholdOfHeap, defaultAbsoluteThreshold);
689+
690+
// then
691+
assertEquals(defaultAbsoluteThreshold, result);
692+
}
693+
694+
public void testThresholdCappedAtIntMax() {
695+
// given
696+
int maxSafeArraySize = Integer.MAX_VALUE - 8;
697+
long defaultThresholdOfHeap = (long) maxSafeArraySize + 1;
698+
long defaultAbsoluteThreshold = ByteSizeUnit.KB.toBytes(500);
699+
700+
// when
701+
long expectedThreshold = calculateMaxWithinIntLimit(defaultThresholdOfHeap, defaultAbsoluteThreshold);
702+
703+
// then
704+
assertEquals(maxSafeArraySize, expectedThreshold);
705+
}
656706
}

0 commit comments

Comments
 (0)