Skip to content

Commit 70064e2

Browse files
opensearch-trigger-bot[bot]github-actions[bot]
opensearch-trigger-bot[bot]
and
github-actions[bot]
authored
Fix file cache initialization (#14004) (#14566)
* fix file cache initialization * changelog * add test --------- (cherry picked from commit 2be25bb) Signed-off-by: panguixin <panguixin@bytedance.com> Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
1 parent 2caa7db commit 70064e2

File tree

9 files changed

+106
-41
lines changed

9 files changed

+106
-41
lines changed

‎CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
4747
- Fix fs info reporting negative available size ([#11573](https://github.com/opensearch-project/OpenSearch/pull/11573))
4848
- Add ListPitInfo::getKeepAlive() getter ([#14495](https://github.com/opensearch-project/OpenSearch/pull/14495))
4949
- Fix FuzzyQuery in keyword field will use IndexOrDocValuesQuery when both of index and doc_value are true ([#14378](https://github.com/opensearch-project/OpenSearch/pull/14378))
50+
- Fix file cache initialization ([#14004](https://github.com/opensearch-project/OpenSearch/pull/14004))
5051

5152
### Security
5253

‎server/src/internalClusterTest/java/org/opensearch/snapshots/SearchableSnapshotIT.java

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,13 +28,15 @@
2828
import org.opensearch.cluster.block.ClusterBlockException;
2929
import org.opensearch.cluster.metadata.IndexMetadata;
3030
import org.opensearch.cluster.node.DiscoveryNode;
31+
import org.opensearch.cluster.node.DiscoveryNodeRole;
3132
import org.opensearch.cluster.routing.GroupShardsIterator;
3233
import org.opensearch.cluster.routing.ShardIterator;
3334
import org.opensearch.cluster.routing.ShardRouting;
3435
import org.opensearch.cluster.routing.allocation.command.MoveAllocationCommand;
3536
import org.opensearch.common.Priority;
3637
import org.opensearch.common.io.PathUtils;
3738
import org.opensearch.common.settings.Settings;
39+
import org.opensearch.common.settings.SettingsException;
3840
import org.opensearch.common.unit.TimeValue;
3941
import org.opensearch.core.common.unit.ByteSizeUnit;
4042
import org.opensearch.core.index.Index;
@@ -65,10 +67,13 @@
6567
import java.util.stream.StreamSupport;
6668

6769
import static org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest.Metric.FS;
70+
import static org.opensearch.common.util.FeatureFlags.TIERED_REMOTE_INDEX;
6871
import static org.opensearch.core.common.util.CollectionUtils.iterableAsArrayList;
6972
import static org.opensearch.index.store.remote.filecache.FileCacheSettings.DATA_TO_FILE_CACHE_SIZE_RATIO_SETTING;
7073
import static org.opensearch.test.NodeRoles.clusterManagerOnlyNode;
7174
import static org.opensearch.test.NodeRoles.dataNode;
75+
import static org.opensearch.test.NodeRoles.onlyRole;
76+
import static org.opensearch.test.NodeRoles.onlyRoles;
7277
import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
7378
import static org.hamcrest.Matchers.contains;
7479
import static org.hamcrest.Matchers.containsString;
@@ -1009,6 +1014,26 @@ public void cleanup() throws Exception {
10091014
);
10101015
}
10111016

1017+
public void testStartSearchNode() throws Exception {
1018+
// test start dedicated search node
1019+
internalCluster().startNode(Settings.builder().put(onlyRole(DiscoveryNodeRole.SEARCH_ROLE)));
1020+
// test start node without search role
1021+
internalCluster().startNode(Settings.builder().put(onlyRole(DiscoveryNodeRole.DATA_ROLE)));
1022+
// test start non-dedicated search node with TIERED_REMOTE_INDEX feature enabled
1023+
internalCluster().startNode(
1024+
Settings.builder()
1025+
.put(onlyRoles(Set.of(DiscoveryNodeRole.SEARCH_ROLE, DiscoveryNodeRole.DATA_ROLE)))
1026+
.put(TIERED_REMOTE_INDEX, true)
1027+
);
1028+
// test start non-dedicated search node
1029+
assertThrows(
1030+
SettingsException.class,
1031+
() -> internalCluster().startNode(
1032+
Settings.builder().put(onlyRoles(Set.of(DiscoveryNodeRole.SEARCH_ROLE, DiscoveryNodeRole.DATA_ROLE)))
1033+
)
1034+
);
1035+
}
1036+
10121037
private void assertSearchableSnapshotIndexDirectoryExistence(String nodeName, Index index, boolean exists) throws Exception {
10131038
final Node node = internalCluster().getInstance(Node.class, nodeName);
10141039
final ShardId shardId = new ShardId(index, 0);

‎server/src/main/java/org/opensearch/cluster/node/DiscoveryNode.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,10 @@ public static boolean isSearchNode(Settings settings) {
132132
return hasRole(settings, DiscoveryNodeRole.SEARCH_ROLE);
133133
}
134134

135+
public static boolean isDedicatedSearchNode(Settings settings) {
136+
return getRolesFromSettings(settings).stream().allMatch(DiscoveryNodeRole.SEARCH_ROLE::equals);
137+
}
138+
135139
private final String nodeName;
136140
private final String nodeId;
137141
private final String ephemeralId;

‎server/src/main/java/org/opensearch/index/store/remote/utils/cache/SegmentedCache.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,15 +52,15 @@ private static final int ceilingNextPowerOfTwo(int x) {
5252
private final Weigher<V> weigher;
5353

5454
public SegmentedCache(Builder<K, V> builder) {
55-
this.capacity = builder.capacity;
5655
final int segments = ceilingNextPowerOfTwo(builder.concurrencyLevel);
5756
this.segmentMask = segments - 1;
5857
this.table = newSegmentArray(segments);
59-
this.perSegmentCapacity = (capacity + (segments - 1)) / segments;
58+
this.perSegmentCapacity = (builder.capacity + (segments - 1)) / segments;
6059
this.weigher = builder.weigher;
6160
for (int i = 0; i < table.length; i++) {
6261
table[i] = new LRUCache<>(perSegmentCapacity, builder.listener, builder.weigher);
6362
}
63+
this.capacity = perSegmentCapacity * segments;
6464
}
6565

6666
@SuppressWarnings("unchecked")

‎server/src/main/java/org/opensearch/monitor/fs/FsProbe.java

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,11 @@ public FsInfo stats(FsInfo previous) throws IOException {
8181
if (fileCache != null && dataLocations[i].fileCacheReservedSize != ByteSizeValue.ZERO) {
8282
paths[i].fileCacheReserved = adjustForHugeFilesystems(dataLocations[i].fileCacheReservedSize.getBytes());
8383
paths[i].fileCacheUtilized = adjustForHugeFilesystems(fileCache.usage().usage());
84-
paths[i].available -= (paths[i].fileCacheReserved - paths[i].fileCacheUtilized);
84+
// fileCacheFree will be less than zero if the cache being over-subscribed
85+
long fileCacheFree = paths[i].fileCacheReserved - paths[i].fileCacheUtilized;
86+
if (fileCacheFree > 0) {
87+
paths[i].available -= fileCacheFree;
88+
}
8589
// occurs if reserved file cache space is occupied by other files, like local indices
8690
if (paths[i].available < 0) {
8791
paths[i].available = 0;
@@ -215,4 +219,11 @@ public static FsInfo.Path getFSInfo(NodePath nodePath) throws IOException {
215219
return fsPath;
216220
}
217221

222+
public static long getTotalSize(NodePath nodePath) throws IOException {
223+
return adjustForHugeFilesystems(nodePath.fileStore.getTotalSpace());
224+
}
225+
226+
public static long getAvailableSize(NodePath nodePath) throws IOException {
227+
return adjustForHugeFilesystems(nodePath.fileStore.getUsableSpace());
228+
}
218229
}

‎server/src/main/java/org/opensearch/node/Node.java

Lines changed: 53 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
import org.opensearch.Build;
3939
import org.opensearch.ExceptionsHelper;
4040
import org.opensearch.OpenSearchException;
41+
import org.opensearch.OpenSearchParseException;
4142
import org.opensearch.OpenSearchTimeoutException;
4243
import org.opensearch.Version;
4344
import org.opensearch.action.ActionModule;
@@ -107,6 +108,7 @@
107108
import org.opensearch.common.settings.Settings;
108109
import org.opensearch.common.settings.SettingsException;
109110
import org.opensearch.common.settings.SettingsModule;
111+
import org.opensearch.common.unit.RatioValue;
110112
import org.opensearch.common.unit.TimeValue;
111113
import org.opensearch.common.util.BigArrays;
112114
import org.opensearch.common.util.FeatureFlags;
@@ -175,7 +177,6 @@
175177
import org.opensearch.ingest.IngestService;
176178
import org.opensearch.monitor.MonitorService;
177179
import org.opensearch.monitor.fs.FsHealthService;
178-
import org.opensearch.monitor.fs.FsInfo;
179180
import org.opensearch.monitor.fs.FsProbe;
180181
import org.opensearch.monitor.jvm.JvmInfo;
181182
import org.opensearch.node.remotestore.RemoteStoreNodeService;
@@ -371,9 +372,12 @@ public class Node implements Closeable {
371372
}
372373
}, Setting.Property.NodeScope);
373374

374-
public static final Setting<ByteSizeValue> NODE_SEARCH_CACHE_SIZE_SETTING = Setting.byteSizeSetting(
375+
private static final String ZERO = "0";
376+
377+
public static final Setting<String> NODE_SEARCH_CACHE_SIZE_SETTING = new Setting<>(
375378
"node.search.cache.size",
376-
ByteSizeValue.ZERO,
379+
s -> (FeatureFlags.isEnabled(FeatureFlags.TIERED_REMOTE_INDEX_SETTING) || DiscoveryNode.isDedicatedSearchNode(s)) ? "80%" : ZERO,
380+
Node::validateFileCacheSize,
377381
Property.NodeScope
378382
);
379383

@@ -1997,43 +2001,59 @@ DiscoveryNode getNode() {
19972001
* Initializes the search cache with a defined capacity.
19982002
* The capacity of the cache is based on user configuration for {@link Node#NODE_SEARCH_CACHE_SIZE_SETTING}.
19992003
* If the user doesn't configure the cache size, it fails if the node is a data + search node.
2000-
* Else it configures the size to 80% of available capacity for a dedicated search node, if not explicitly defined.
2004+
* Else it configures the size to 80% of total capacity for a dedicated search node, if not explicitly defined.
20012005
*/
20022006
private void initializeFileCache(Settings settings, CircuitBreaker circuitBreaker) throws IOException {
20032007
boolean isWritableRemoteIndexEnabled = FeatureFlags.isEnabled(FeatureFlags.TIERED_REMOTE_INDEX_SETTING);
2004-
if (DiscoveryNode.isSearchNode(settings) || isWritableRemoteIndexEnabled) {
2005-
NodeEnvironment.NodePath fileCacheNodePath = nodeEnvironment.fileCacheNodePath();
2006-
long capacity = NODE_SEARCH_CACHE_SIZE_SETTING.get(settings).getBytes();
2007-
FsInfo.Path info = ExceptionsHelper.catchAsRuntimeException(() -> FsProbe.getFSInfo(fileCacheNodePath));
2008-
long availableCapacity = info.getAvailable().getBytes();
2009-
2010-
// Initialize default values for cache if NODE_SEARCH_CACHE_SIZE_SETTING is not set.
2011-
if (capacity == 0) {
2012-
// If node is not a dedicated search node without configuration, prevent cache initialization
2013-
if (!isWritableRemoteIndexEnabled
2014-
&& DiscoveryNode.getRolesFromSettings(settings)
2015-
.stream()
2016-
.anyMatch(role -> !DiscoveryNodeRole.SEARCH_ROLE.equals(role))) {
2017-
throw new SettingsException(
2018-
"Unable to initialize the "
2019-
+ DiscoveryNodeRole.SEARCH_ROLE.roleName()
2020-
+ "-"
2021-
+ DiscoveryNodeRole.DATA_ROLE.roleName()
2022-
+ " node: Missing value for configuration "
2023-
+ NODE_SEARCH_CACHE_SIZE_SETTING.getKey()
2024-
);
2025-
} else {
2026-
capacity = 80 * availableCapacity / 100;
2027-
}
2008+
if (DiscoveryNode.isSearchNode(settings) == false && isWritableRemoteIndexEnabled == false) {
2009+
return;
2010+
}
2011+
2012+
String capacityRaw = NODE_SEARCH_CACHE_SIZE_SETTING.get(settings);
2013+
logger.info("cache size [{}]", capacityRaw);
2014+
if (capacityRaw.equals(ZERO)) {
2015+
throw new SettingsException(
2016+
"Unable to initialize the "
2017+
+ DiscoveryNodeRole.SEARCH_ROLE.roleName()
2018+
+ "-"
2019+
+ DiscoveryNodeRole.DATA_ROLE.roleName()
2020+
+ " node: Missing value for configuration "
2021+
+ NODE_SEARCH_CACHE_SIZE_SETTING.getKey()
2022+
);
2023+
}
2024+
2025+
NodeEnvironment.NodePath fileCacheNodePath = nodeEnvironment.fileCacheNodePath();
2026+
long totalSpace = ExceptionsHelper.catchAsRuntimeException(() -> FsProbe.getTotalSize(fileCacheNodePath));
2027+
long capacity = calculateFileCacheSize(capacityRaw, totalSpace);
2028+
if (capacity <= 0 || totalSpace <= capacity) {
2029+
throw new SettingsException("Cache size must be larger than zero and less than total capacity");
2030+
}
2031+
2032+
this.fileCache = FileCacheFactory.createConcurrentLRUFileCache(capacity, circuitBreaker);
2033+
fileCacheNodePath.fileCacheReservedSize = new ByteSizeValue(this.fileCache.capacity(), ByteSizeUnit.BYTES);
2034+
List<Path> fileCacheDataPaths = collectFileCacheDataPath(fileCacheNodePath);
2035+
this.fileCache.restoreFromDirectory(fileCacheDataPaths);
2036+
}
2037+
2038+
private static long calculateFileCacheSize(String capacityRaw, long totalSpace) {
2039+
try {
2040+
RatioValue ratioValue = RatioValue.parseRatioValue(capacityRaw);
2041+
return Math.round(totalSpace * ratioValue.getAsRatio());
2042+
} catch (OpenSearchParseException e) {
2043+
try {
2044+
return ByteSizeValue.parseBytesSizeValue(capacityRaw, NODE_SEARCH_CACHE_SIZE_SETTING.getKey()).getBytes();
2045+
} catch (OpenSearchParseException ex) {
2046+
ex.addSuppressed(e);
2047+
throw ex;
20282048
}
2029-
capacity = Math.min(capacity, availableCapacity);
2030-
fileCacheNodePath.fileCacheReservedSize = new ByteSizeValue(capacity, ByteSizeUnit.BYTES);
2031-
this.fileCache = FileCacheFactory.createConcurrentLRUFileCache(capacity, circuitBreaker);
2032-
List<Path> fileCacheDataPaths = collectFileCacheDataPath(fileCacheNodePath);
2033-
this.fileCache.restoreFromDirectory(fileCacheDataPaths);
20342049
}
20352050
}
20362051

2052+
private static String validateFileCacheSize(String capacityRaw) {
2053+
calculateFileCacheSize(capacityRaw, 0L);
2054+
return capacityRaw;
2055+
}
2056+
20372057
/**
20382058
* Returns the {@link FileCache} instance for remote search node
20392059
* Note: Visible for testing

‎server/src/test/java/org/opensearch/env/NodeRepurposeCommandTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,7 +95,7 @@ public void createNodePaths() throws IOException {
9595
dataClusterManagerSettings = buildEnvSettings(Settings.EMPTY);
9696
Settings defaultSearchSettings = Settings.builder()
9797
.put(dataClusterManagerSettings)
98-
.put(NODE_SEARCH_CACHE_SIZE_SETTING.getKey(), new ByteSizeValue(16, ByteSizeUnit.GB))
98+
.put(NODE_SEARCH_CACHE_SIZE_SETTING.getKey(), new ByteSizeValue(16, ByteSizeUnit.GB).toString())
9999
.build();
100100

101101
searchNoDataNoClusterManagerSettings = onlyRole(dataClusterManagerSettings, DiscoveryNodeRole.SEARCH_ROLE);

‎server/src/test/java/org/opensearch/node/NodeTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -380,7 +380,7 @@ public void testCreateWithFileCache() throws Exception {
380380
List<Class<? extends Plugin>> plugins = basePlugins();
381381
ByteSizeValue cacheSize = new ByteSizeValue(16, ByteSizeUnit.GB);
382382
Settings searchRoleSettingsWithConfig = baseSettings().put(searchRoleSettings)
383-
.put(Node.NODE_SEARCH_CACHE_SIZE_SETTING.getKey(), cacheSize)
383+
.put(Node.NODE_SEARCH_CACHE_SIZE_SETTING.getKey(), cacheSize.toString())
384384
.build();
385385
Settings onlySearchRoleSettings = Settings.builder()
386386
.put(searchRoleSettingsWithConfig)

‎test/framework/src/main/java/org/opensearch/test/InternalTestCluster.java

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,7 @@
165165
import static org.opensearch.test.NodeRoles.onlyRoles;
166166
import static org.opensearch.test.NodeRoles.removeRoles;
167167
import static org.opensearch.test.OpenSearchTestCase.assertBusy;
168+
import static org.opensearch.test.OpenSearchTestCase.randomBoolean;
168169
import static org.opensearch.test.OpenSearchTestCase.randomFrom;
169170
import static org.hamcrest.Matchers.equalTo;
170171
import static org.hamcrest.Matchers.greaterThan;
@@ -216,7 +217,8 @@ public final class InternalTestCluster extends TestCluster {
216217
nodeAndClient.node.settings()
217218
);
218219

219-
private static final ByteSizeValue DEFAULT_SEARCH_CACHE_SIZE = new ByteSizeValue(2, ByteSizeUnit.GB);
220+
private static final String DEFAULT_SEARCH_CACHE_SIZE_BYTES = "2gb";
221+
private static final String DEFAULT_SEARCH_CACHE_SIZE_PERCENT = "5%";
220222

221223
public static final int DEFAULT_LOW_NUM_CLUSTER_MANAGER_NODES = 1;
222224
public static final int DEFAULT_HIGH_NUM_CLUSTER_MANAGER_NODES = 3;
@@ -700,8 +702,10 @@ public synchronized void ensureAtLeastNumSearchAndDataNodes(int n) {
700702
logger.info("increasing cluster size from {} to {}", size, n);
701703
Set<DiscoveryNodeRole> searchAndDataRoles = Set.of(DiscoveryNodeRole.DATA_ROLE, DiscoveryNodeRole.SEARCH_ROLE);
702704
Settings settings = Settings.builder()
703-
.put(Settings.EMPTY)
704-
.put(Node.NODE_SEARCH_CACHE_SIZE_SETTING.getKey(), DEFAULT_SEARCH_CACHE_SIZE)
705+
.put(
706+
Node.NODE_SEARCH_CACHE_SIZE_SETTING.getKey(),
707+
randomBoolean() ? DEFAULT_SEARCH_CACHE_SIZE_PERCENT : DEFAULT_SEARCH_CACHE_SIZE_BYTES
708+
)
705709
.build();
706710
startNodes(n - size, Settings.builder().put(onlyRoles(settings, searchAndDataRoles)).build());
707711
validateClusterFormed();

0 commit comments

Comments
 (0)