Skip to content

Commit 103ab73

Browse files
github-actions[bot]github-actions[bot]
committed
Refactor remote store flow to support any path type (#12822)
Signed-off-by: Ashish Singh <ssashish@amazon.com> (cherry picked from commit 3907ec9) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
1 parent de875c5 commit 103ab73

20 files changed

Lines changed: 418 additions & 110 deletions

‎server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java‎

Lines changed: 16 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -90,8 +90,8 @@
9090
import org.opensearch.index.mapper.MapperService;
9191
import org.opensearch.index.mapper.MapperService.MergeReason;
9292
import org.opensearch.index.query.QueryShardContext;
93-
import org.opensearch.index.remote.RemoteStorePathResolver;
9493
import org.opensearch.index.remote.RemoteStorePathType;
94+
import org.opensearch.index.remote.RemoteStorePathTypeResolver;
9595
import org.opensearch.index.shard.IndexSettingProvider;
9696
import org.opensearch.index.translog.Translog;
9797
import org.opensearch.indices.IndexCreationException;
@@ -115,6 +115,7 @@
115115
import java.util.List;
116116
import java.util.Locale;
117117
import java.util.Map;
118+
import java.util.Objects;
118119
import java.util.Optional;
119120
import java.util.Set;
120121
import java.util.concurrent.atomic.AtomicInteger;
@@ -174,7 +175,7 @@ public class MetadataCreateIndexService {
174175
private AwarenessReplicaBalance awarenessReplicaBalance;
175176

176177
@Nullable
177-
private final RemoteStorePathResolver remoteStorePathResolver;
178+
private final RemoteStorePathTypeResolver remoteStorePathTypeResolver;
178179

179180
public MetadataCreateIndexService(
180181
final Settings settings,
@@ -207,8 +208,8 @@ public MetadataCreateIndexService(
207208

208209
// Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
209210
createIndexTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.CREATE_INDEX_KEY, true);
210-
remoteStorePathResolver = isRemoteDataAttributePresent(settings)
211-
? new RemoteStorePathResolver(clusterService.getClusterSettings())
211+
remoteStorePathTypeResolver = isRemoteDataAttributePresent(settings)
212+
? new RemoteStorePathTypeResolver(clusterService.getClusterSettings())
212213
: null;
213214
}
214215

@@ -557,7 +558,7 @@ IndexMetadata buildAndValidateTemporaryIndexMetadata(
557558
tmpImdBuilder.setRoutingNumShards(routingNumShards);
558559
tmpImdBuilder.settings(indexSettings);
559560
tmpImdBuilder.system(isSystem);
560-
addRemoteCustomData(tmpImdBuilder);
561+
addRemoteStorePathTypeInCustomData(tmpImdBuilder, true);
561562

562563
// Set up everything, now locally create the index to see that things are ok, and apply
563564
IndexMetadata tempMetadata = tmpImdBuilder.build();
@@ -566,17 +567,24 @@ IndexMetadata buildAndValidateTemporaryIndexMetadata(
566567
return tempMetadata;
567568
}
568569

569-
public void addRemoteCustomData(IndexMetadata.Builder tmpImdBuilder) {
570-
if (remoteStorePathResolver != null) {
570+
/**
571+
* Adds the remote store path type information in custom data of index metadata.
572+
*
573+
* @param tmpImdBuilder index metadata builder.
574+
* @param assertNullOldType flag to verify that the old remote store path type is null
575+
*/
576+
public void addRemoteStorePathTypeInCustomData(IndexMetadata.Builder tmpImdBuilder, boolean assertNullOldType) {
577+
if (remoteStorePathTypeResolver != null) {
571578
// It is possible that remote custom data exists already. In such cases, we need to only update the path type
572579
// in the remote store custom data map.
573580
Map<String, String> existingRemoteCustomData = tmpImdBuilder.removeCustom(IndexMetadata.REMOTE_STORE_CUSTOM_KEY);
574581
Map<String, String> remoteCustomData = existingRemoteCustomData == null
575582
? new HashMap<>()
576583
: new HashMap<>(existingRemoteCustomData);
577584
// Determine the path type for use using the remoteStorePathResolver.
578-
String newPathType = remoteStorePathResolver.resolveType().toString();
585+
String newPathType = remoteStorePathTypeResolver.getType().toString();
579586
String oldPathType = remoteCustomData.put(RemoteStorePathType.NAME, newPathType);
587+
assert !assertNullOldType || Objects.isNull(oldPathType);
580588
logger.trace(() -> new ParameterizedMessage("Added new path type {}, replaced old path type {}", newPathType, oldPathType));
581589
tmpImdBuilder.putCustom(IndexMetadata.REMOTE_STORE_CUSTOM_KEY, remoteCustomData);
582590
}

‎server/src/main/java/org/opensearch/index/IndexService.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -510,7 +510,8 @@ public synchronized IndexShard createShard(
510510
remoteDirectory = ((RemoteSegmentStoreDirectoryFactory) remoteDirectoryFactory).newDirectory(
511511
RemoteStoreNodeAttribute.getRemoteStoreSegmentRepo(this.indexSettings.getNodeSettings()),
512512
this.indexSettings.getUUID(),
513-
shardId
513+
shardId,
514+
this.indexSettings.getRemoteStorePathType()
514515
);
515516
}
516517
remoteStore = new Store(shardId, this.indexSettings, remoteDirectory, lock, Store.OnClose.EMPTY, path);

‎server/src/main/java/org/opensearch/index/IndexSettings.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@
4949
import org.opensearch.core.common.unit.ByteSizeUnit;
5050
import org.opensearch.core.common.unit.ByteSizeValue;
5151
import org.opensearch.core.index.Index;
52+
import org.opensearch.index.remote.RemoteStorePathType;
5253
import org.opensearch.index.translog.Translog;
5354
import org.opensearch.indices.replication.common.ReplicationType;
5455
import org.opensearch.ingest.IngestService;
@@ -60,6 +61,7 @@
6061
import java.util.Collections;
6162
import java.util.List;
6263
import java.util.Locale;
64+
import java.util.Map;
6365
import java.util.Optional;
6466
import java.util.concurrent.TimeUnit;
6567
import java.util.function.Consumer;
@@ -1903,4 +1905,11 @@ public double getDocIdFuzzySetFalsePositiveProbability() {
19031905
public void setDocIdFuzzySetFalsePositiveProbability(double docIdFuzzySetFalsePositiveProbability) {
19041906
this.docIdFuzzySetFalsePositiveProbability = docIdFuzzySetFalsePositiveProbability;
19051907
}
1908+
1909+
public RemoteStorePathType getRemoteStorePathType() {
1910+
Map<String, String> remoteCustomData = indexMetadata.getCustomData(IndexMetadata.REMOTE_STORE_CUSTOM_KEY);
1911+
return remoteCustomData != null && remoteCustomData.containsKey(RemoteStorePathType.NAME)
1912+
? RemoteStorePathType.parseString(remoteCustomData.get(RemoteStorePathType.NAME))
1913+
: RemoteStorePathType.FIXED;
1914+
}
19061915
}

‎server/src/main/java/org/opensearch/index/remote/RemoteStoreDataEnums.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.index.remote;
10+
11+
import java.util.Set;
12+
13+
import static org.opensearch.index.remote.RemoteStoreDataEnums.DataType.DATA;
14+
import static org.opensearch.index.remote.RemoteStoreDataEnums.DataType.METADATA;
15+
16+
/**
17+
* This class contains the different enums related to remote store data categories and types.
18+
*
19+
* @opensearch.internal
20+
*/
21+
public class RemoteStoreDataEnums {
22+
23+
/**
24+
* Categories of the data in Remote store.
25+
*/
26+
public enum DataCategory {
27+
SEGMENTS("segments", Set.of(DataType.values())),
28+
TRANSLOG("translog", Set.of(DATA, METADATA));
29+
30+
private final String name;
31+
private final Set<DataType> supportedDataTypes;
32+
33+
DataCategory(String name, Set<DataType> supportedDataTypes) {
34+
this.name = name;
35+
this.supportedDataTypes = supportedDataTypes;
36+
}
37+
38+
public boolean isSupportedDataType(DataType dataType) {
39+
return supportedDataTypes.contains(dataType);
40+
}
41+
42+
public String getName() {
43+
return name;
44+
}
45+
}
46+
47+
/**
48+
* Types of data in remote store.
49+
*/
50+
public enum DataType {
51+
DATA("data"),
52+
METADATA("metadata"),
53+
LOCK_FILES("lock_files");
54+
55+
private final String name;
56+
57+
DataType(String name) {
58+
this.name = name;
59+
}
60+
61+
public String getName() {
62+
return name;
63+
}
64+
}
65+
}

‎server/src/main/java/org/opensearch/index/remote/RemoteStorePathType.java‎

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,10 @@
88

99
package org.opensearch.index.remote;
1010

11+
import org.opensearch.common.blobstore.BlobPath;
12+
import org.opensearch.index.remote.RemoteStoreDataEnums.DataCategory;
13+
import org.opensearch.index.remote.RemoteStoreDataEnums.DataType;
14+
1115
import java.util.Locale;
1216

1317
/**
@@ -18,13 +22,46 @@
1822
*/
1923
public enum RemoteStorePathType {
2024

21-
FIXED,
22-
HASHED_PREFIX;
25+
FIXED {
26+
@Override
27+
public BlobPath generatePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType) {
28+
return basePath.add(indexUUID).add(shardId).add(dataCategory).add(dataType);
29+
}
30+
},
31+
HASHED_PREFIX {
32+
@Override
33+
public BlobPath generatePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType) {
34+
// TODO - We need to implement this, keeping the same path as Fixed for sake of multiple tests that can fail otherwise.
35+
// throw new UnsupportedOperationException("Not implemented"); --> Not using this for unblocking couple of tests.
36+
return basePath.add(indexUUID).add(shardId).add(dataCategory).add(dataType);
37+
}
38+
};
39+
40+
/**
41+
* @param basePath base path of the underlying blob store repository
42+
* @param indexUUID of the index
43+
* @param shardId shard id
44+
* @param dataCategory is either translog or segment
45+
* @param dataType can be one of data, metadata or lock_files.
46+
* @return the blob path for the underlying remote store path type.
47+
*/
48+
public BlobPath path(BlobPath basePath, String indexUUID, String shardId, DataCategory dataCategory, DataType dataType) {
49+
assert dataCategory.isSupportedDataType(dataType) : "category:"
50+
+ dataCategory
51+
+ " type:"
52+
+ dataType
53+
+ " are not supported together";
54+
return generatePath(basePath, indexUUID, shardId, dataCategory.getName(), dataType.getName());
55+
}
56+
57+
abstract BlobPath generatePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType);
2358

2459
public static RemoteStorePathType parseString(String remoteStoreBlobPathType) {
2560
try {
2661
return RemoteStorePathType.valueOf(remoteStoreBlobPathType.toUpperCase(Locale.ROOT));
27-
} catch (IllegalArgumentException e) {
62+
} catch (IllegalArgumentException | NullPointerException e) {
63+
// IllegalArgumentException is thrown when the input does not match any enum name
64+
// NullPointerException is thrown when the input is null
2865
throw new IllegalArgumentException("Could not parse RemoteStorePathType for [" + remoteStoreBlobPathType + "]");
2966
}
3067
}

server/src/main/java/org/opensearch/index/remote/RemoteStorePathResolver.java renamed to server/src/main/java/org/opensearch/index/remote/RemoteStorePathTypeResolver.java

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,15 +16,20 @@
1616
*
1717
* @opensearch.internal
1818
*/
19-
public class RemoteStorePathResolver {
19+
public class RemoteStorePathTypeResolver {
2020

21-
private final ClusterSettings clusterSettings;
21+
private volatile RemoteStorePathType type;
2222

23-
public RemoteStorePathResolver(ClusterSettings clusterSettings) {
24-
this.clusterSettings = clusterSettings;
23+
public RemoteStorePathTypeResolver(ClusterSettings clusterSettings) {
24+
type = clusterSettings.get(IndicesService.CLUSTER_REMOTE_STORE_PATH_PREFIX_TYPE_SETTING);
25+
clusterSettings.addSettingsUpdateConsumer(IndicesService.CLUSTER_REMOTE_STORE_PATH_PREFIX_TYPE_SETTING, this::setType);
2526
}
2627

27-
public RemoteStorePathType resolveType() {
28-
return clusterSettings.get(IndicesService.CLUSTER_REMOTE_STORE_PATH_PREFIX_TYPE_SETTING);
28+
public RemoteStorePathType getType() {
29+
return type;
30+
}
31+
32+
public void setType(RemoteStorePathType type) {
33+
this.type = type;
2934
}
3035
}

‎server/src/main/java/org/opensearch/index/shard/IndexShard.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4988,7 +4988,7 @@ public void deleteTranslogFilesFromRemoteTranslog() throws IOException {
49884988
TranslogFactory translogFactory = translogFactorySupplier.apply(indexSettings, shardRouting);
49894989
assert translogFactory instanceof RemoteBlobStoreInternalTranslogFactory;
49904990
Repository repository = ((RemoteBlobStoreInternalTranslogFactory) translogFactory).getRepository();
4991-
RemoteFsTranslog.cleanup(repository, shardId, getThreadPool());
4991+
RemoteFsTranslog.cleanup(repository, shardId, getThreadPool(), indexSettings.getRemoteStorePathType());
49924992
}
49934993

49944994
/*
@@ -5005,7 +5005,14 @@ public void syncTranslogFilesFromRemoteTranslog() throws IOException {
50055005
TranslogFactory translogFactory = translogFactorySupplier.apply(indexSettings, shardRouting);
50065006
assert translogFactory instanceof RemoteBlobStoreInternalTranslogFactory;
50075007
Repository repository = ((RemoteBlobStoreInternalTranslogFactory) translogFactory).getRepository();
5008-
RemoteFsTranslog.download(repository, shardId, getThreadPool(), shardPath().resolveTranslog(), logger);
5008+
RemoteFsTranslog.download(
5009+
repository,
5010+
shardId,
5011+
getThreadPool(),
5012+
shardPath().resolveTranslog(),
5013+
indexSettings.getRemoteStorePathType(),
5014+
logger
5015+
);
50095016
}
50105017

50115018
/**

‎server/src/main/java/org/opensearch/index/shard/StoreRecovery.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import org.opensearch.index.engine.Engine;
5959
import org.opensearch.index.engine.EngineException;
6060
import org.opensearch.index.mapper.MapperService;
61+
import org.opensearch.index.remote.RemoteStorePathType;
6162
import org.opensearch.index.seqno.SequenceNumbers;
6263
import org.opensearch.index.snapshots.IndexShardRestoreFailedException;
6364
import org.opensearch.index.snapshots.blobstore.RemoteStoreShardShallowCopySnapshot;
@@ -402,7 +403,8 @@ void recoverFromSnapshotAndRemoteStore(
402403
RemoteSegmentStoreDirectory sourceRemoteDirectory = (RemoteSegmentStoreDirectory) directoryFactory.newDirectory(
403404
remoteStoreRepository,
404405
indexUUID,
405-
shardId
406+
shardId,
407+
RemoteStorePathType.FIXED // TODO - The path type needs to be obtained from RemoteStoreShardShallowCopySnapshot
406408
);
407409
sourceRemoteDirectory.initializeToSpecificCommit(
408410
primaryTerm,

‎server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.opensearch.common.lucene.store.ByteArrayIndexInput;
3232
import org.opensearch.core.action.ActionListener;
3333
import org.opensearch.core.index.shard.ShardId;
34+
import org.opensearch.index.remote.RemoteStorePathType;
3435
import org.opensearch.index.remote.RemoteStoreUtils;
3536
import org.opensearch.index.store.lockmanager.FileLockInfo;
3637
import org.opensearch.index.store.lockmanager.RemoteStoreCommitLevelLockManager;
@@ -897,13 +898,15 @@ public static void remoteDirectoryCleanup(
897898
RemoteSegmentStoreDirectoryFactory remoteDirectoryFactory,
898899
String remoteStoreRepoForIndex,
899900
String indexUUID,
900-
ShardId shardId
901+
ShardId shardId,
902+
RemoteStorePathType pathType
901903
) {
902904
try {
903905
RemoteSegmentStoreDirectory remoteSegmentStoreDirectory = (RemoteSegmentStoreDirectory) remoteDirectoryFactory.newDirectory(
904906
remoteStoreRepoForIndex,
905907
indexUUID,
906-
shardId
908+
shardId,
909+
pathType
907910
);
908911
remoteSegmentStoreDirectory.deleteStaleSegments(0);
909912
remoteSegmentStoreDirectory.deleteIfEmpty();

0 commit comments

Comments
 (0)