Skip to content

Commit 21636af

Browse files
ashking94ashking94
committed
Refactor remote store flow to support any path type with bwc
Signed-off-by: Ashish Singh <ssashish@amazon.com>
1 parent f3d2bee commit 21636af

13 files changed

+148
-65
lines changed

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

+10-8
Original file line numberDiff line numberDiff line change
@@ -88,8 +88,8 @@
8888
import org.opensearch.index.mapper.MapperService;
8989
import org.opensearch.index.mapper.MapperService.MergeReason;
9090
import org.opensearch.index.query.QueryShardContext;
91-
import org.opensearch.index.remote.RemoteStorePathResolver;
9291
import org.opensearch.index.remote.RemoteStorePathType;
92+
import org.opensearch.index.remote.RemoteStorePathTypeResolver;
9393
import org.opensearch.index.shard.IndexSettingProvider;
9494
import org.opensearch.index.translog.Translog;
9595
import org.opensearch.indices.IndexCreationException;
@@ -113,6 +113,7 @@
113113
import java.util.List;
114114
import java.util.Locale;
115115
import java.util.Map;
116+
import java.util.Objects;
116117
import java.util.Optional;
117118
import java.util.Set;
118119
import java.util.concurrent.atomic.AtomicInteger;
@@ -170,7 +171,7 @@ public class MetadataCreateIndexService {
170171
private AwarenessReplicaBalance awarenessReplicaBalance;
171172

172173
@Nullable
173-
private final RemoteStorePathResolver remoteStorePathResolver;
174+
private final RemoteStorePathTypeResolver remoteStorePathTypeResolver;
174175

175176
public MetadataCreateIndexService(
176177
final Settings settings,
@@ -203,8 +204,8 @@ public MetadataCreateIndexService(
203204

204205
// Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
205206
createIndexTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.CREATE_INDEX_KEY, true);
206-
remoteStorePathResolver = isRemoteDataAttributePresent(settings)
207-
? new RemoteStorePathResolver(clusterService.getClusterSettings())
207+
remoteStorePathTypeResolver = isRemoteDataAttributePresent(settings)
208+
? new RemoteStorePathTypeResolver(clusterService.getClusterSettings())
208209
: null;
209210
}
210211

@@ -553,7 +554,7 @@ IndexMetadata buildAndValidateTemporaryIndexMetadata(
553554
tmpImdBuilder.setRoutingNumShards(routingNumShards);
554555
tmpImdBuilder.settings(indexSettings);
555556
tmpImdBuilder.system(isSystem);
556-
addRemoteCustomData(tmpImdBuilder);
557+
addRemoteStorePathTypeInCustomData(tmpImdBuilder, true);
557558

558559
// Set up everything, now locally create the index to see that things are ok, and apply
559560
IndexMetadata tempMetadata = tmpImdBuilder.build();
@@ -562,17 +563,18 @@ IndexMetadata buildAndValidateTemporaryIndexMetadata(
562563
return tempMetadata;
563564
}
564565

565-
public void addRemoteCustomData(IndexMetadata.Builder tmpImdBuilder) {
566-
if (remoteStorePathResolver != null) {
566+
public void addRemoteStorePathTypeInCustomData(IndexMetadata.Builder tmpImdBuilder, boolean assertNullOldType) {
567+
if (remoteStorePathTypeResolver != null) {
567568
// It is possible that remote custom data exists already. In such cases, we need to only update the path type
568569
// in the remote store custom data map.
569570
Map<String, String> existingRemoteCustomData = tmpImdBuilder.removeCustom(IndexMetadata.REMOTE_STORE_CUSTOM_KEY);
570571
Map<String, String> remoteCustomData = existingRemoteCustomData == null
571572
? new HashMap<>()
572573
: new HashMap<>(existingRemoteCustomData);
573574
// Determine the path type for use using the remoteStorePathResolver.
574-
String newPathType = remoteStorePathResolver.resolveType().toString();
575+
String newPathType = remoteStorePathTypeResolver.getType().toString();
575576
String oldPathType = remoteCustomData.put(RemoteStorePathType.NAME, newPathType);
577+
assert !assertNullOldType || Objects.isNull(oldPathType);
576578
logger.trace(() -> new ParameterizedMessage("Added new path type {}, replaced old path type {}", newPathType, oldPathType));
577579
tmpImdBuilder.putCustom(IndexMetadata.REMOTE_STORE_CUSTOM_KEY, remoteCustomData);
578580
}

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

+2-1
Original file line numberDiff line numberDiff line change
@@ -507,7 +507,8 @@ public synchronized IndexShard createShard(
507507
remoteDirectory = ((RemoteSegmentStoreDirectoryFactory) remoteDirectoryFactory).newDirectory(
508508
RemoteStoreNodeAttribute.getRemoteStoreSegmentRepo(this.indexSettings.getNodeSettings()),
509509
this.indexSettings.getUUID(),
510-
shardId
510+
shardId,
511+
this.indexSettings.getRemoteStorePathType()
511512
);
512513
}
513514
remoteStore = new Store(shardId, this.indexSettings, remoteDirectory, lock, Store.OnClose.EMPTY, path);

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

+9
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
import org.opensearch.core.common.unit.ByteSizeUnit;
4949
import org.opensearch.core.common.unit.ByteSizeValue;
5050
import org.opensearch.core.index.Index;
51+
import org.opensearch.index.remote.RemoteStorePathType;
5152
import org.opensearch.index.translog.Translog;
5253
import org.opensearch.indices.replication.common.ReplicationType;
5354
import org.opensearch.ingest.IngestService;
@@ -59,6 +60,7 @@
5960
import java.util.Collections;
6061
import java.util.List;
6162
import java.util.Locale;
63+
import java.util.Map;
6264
import java.util.Optional;
6365
import java.util.concurrent.TimeUnit;
6466
import java.util.function.Consumer;
@@ -1905,4 +1907,11 @@ public double getDocIdFuzzySetFalsePositiveProbability() {
19051907
public void setDocIdFuzzySetFalsePositiveProbability(double docIdFuzzySetFalsePositiveProbability) {
19061908
this.docIdFuzzySetFalsePositiveProbability = docIdFuzzySetFalsePositiveProbability;
19071909
}
1910+
1911+
public RemoteStorePathType getRemoteStorePathType() {
1912+
Map<String, String> remoteCustomData = indexMetadata.getCustomData(IndexMetadata.REMOTE_STORE_CUSTOM_KEY);
1913+
return remoteCustomData != null && remoteCustomData.containsKey(RemoteStorePathType.NAME)
1914+
? RemoteStorePathType.parseString(remoteCustomData.get(RemoteStorePathType.NAME))
1915+
: RemoteStorePathType.FIXED;
1916+
}
19081917
}

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

+43-3
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,13 @@
88

99
package org.opensearch.index.remote;
1010

11+
import org.opensearch.common.blobstore.BlobPath;
12+
1113
import java.util.Locale;
1214

15+
import static org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory.LOCK_FILES;
16+
import static org.opensearch.index.translog.RemoteFsTranslog.TRANSLOG;
17+
1318
/**
1419
* Enumerates the types of remote store paths resolution techniques supported by OpenSearch.
1520
* For more information, see <a href="https://github.com/opensearch-project/OpenSearch/issues/12567">Github issue #12567</a>.
@@ -18,13 +23,48 @@
1823
*/
1924
public enum RemoteStorePathType {
2025

21-
FIXED,
22-
HASHED_PREFIX;
26+
FIXED {
27+
@Override
28+
public BlobPath validateAndGeneratePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType) {
29+
return basePath.add(indexUUID).add(shardId).add(dataCategory).add(dataType);
30+
}
31+
},
32+
HASHED_PREFIX {
33+
@Override
34+
public BlobPath validateAndGeneratePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType) {
35+
// TODO - We need to implement this, keeping the same path as Fixed for sake of multiple tests that can fail otherwise.
36+
// throw new UnsupportedOperationException("Not implemented"); --> Not using this for some tests
37+
return basePath.add(indexUUID).add(shardId).add(dataCategory).add(dataType);
38+
}
39+
};
40+
41+
/**
42+
* @param basePath base path of the underlying blob store repository
43+
* @param indexUUID of the index
44+
* @param shardId shard id
45+
* @param dataCategory is either translog or segment
46+
* @param dataType can be one of data, metadata or lock_files.
47+
* @return the blob path for the underlying remote store path type.
48+
*/
49+
public BlobPath generatePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType) {
50+
assertDataCategoryAndTypeCombination(dataCategory, dataType);
51+
return validateAndGeneratePath(basePath, indexUUID, shardId, dataCategory, dataType);
52+
}
53+
54+
abstract BlobPath validateAndGeneratePath(BlobPath basePath, String indexUUID, String shardId, String dataCategory, String dataType);
55+
56+
/**
57+
* This method verifies that if the data category is translog, then the data type can not be lock_files. All other
58+
* combination of data categories and data types are possible.
59+
*/
60+
private static void assertDataCategoryAndTypeCombination(String dataCategory, String dataType) {
61+
assert dataCategory.equals(TRANSLOG) == false || dataType.equals(LOCK_FILES) == false;
62+
}
2363

2464
public static RemoteStorePathType parseString(String remoteStoreBlobPathType) {
2565
try {
2666
return RemoteStorePathType.valueOf(remoteStoreBlobPathType.toUpperCase(Locale.ROOT));
27-
} catch (IllegalArgumentException e) {
67+
} catch (IllegalArgumentException | NullPointerException e) {
2868
throw new IllegalArgumentException("Could not parse RemoteStorePathType for [" + remoteStoreBlobPathType + "]");
2969
}
3070
}

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

+11-6
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/StoreRecovery.java

+3-1
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;
@@ -409,7 +410,8 @@ void recoverFromSnapshotAndRemoteStore(
409410
RemoteSegmentStoreDirectory sourceRemoteDirectory = (RemoteSegmentStoreDirectory) directoryFactory.newDirectory(
410411
remoteStoreRepository,
411412
indexUUID,
412-
shardId
413+
shardId,
414+
RemoteStorePathType.FIXED // TODO - The path type needs to be obtained from RemoteStoreShardShallowCopySnapshot
413415
);
414416
sourceRemoteDirectory.initializeToSpecificCommit(
415417
primaryTerm,

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

+5-2
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();

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

+22-9
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import org.opensearch.common.blobstore.BlobPath;
1313
import org.opensearch.core.index.shard.ShardId;
1414
import org.opensearch.index.IndexSettings;
15+
import org.opensearch.index.remote.RemoteStorePathType;
1516
import org.opensearch.index.shard.ShardPath;
1617
import org.opensearch.index.store.lockmanager.RemoteStoreLockManager;
1718
import org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory;
@@ -23,6 +24,7 @@
2324
import org.opensearch.threadpool.ThreadPool;
2425

2526
import java.io.IOException;
27+
import java.util.Objects;
2628
import java.util.function.Supplier;
2729

2830
/**
@@ -32,6 +34,8 @@
3234
*/
3335
public class RemoteSegmentStoreDirectoryFactory implements IndexStorePlugin.DirectoryFactory {
3436
private static final String SEGMENTS = "segments";
37+
private final static String DATA_DIR = "data";
38+
private final static String METADATA_DIR = "metadata";
3539

3640
private final Supplier<RepositoriesService> repositoriesService;
3741

@@ -46,29 +50,38 @@ public RemoteSegmentStoreDirectoryFactory(Supplier<RepositoriesService> reposito
4650
public Directory newDirectory(IndexSettings indexSettings, ShardPath path) throws IOException {
4751
String repositoryName = indexSettings.getRemoteStoreRepository();
4852
String indexUUID = indexSettings.getIndex().getUUID();
49-
return newDirectory(repositoryName, indexUUID, path.getShardId());
53+
return newDirectory(repositoryName, indexUUID, path.getShardId(), indexSettings.getRemoteStorePathType());
5054
}
5155

52-
public Directory newDirectory(String repositoryName, String indexUUID, ShardId shardId) throws IOException {
56+
public Directory newDirectory(String repositoryName, String indexUUID, ShardId shardId, RemoteStorePathType pathType)
57+
throws IOException {
58+
assert Objects.nonNull(pathType);
5359
try (Repository repository = repositoriesService.get().repository(repositoryName)) {
60+
5461
assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
5562
BlobStoreRepository blobStoreRepository = ((BlobStoreRepository) repository);
56-
BlobPath commonBlobPath = blobStoreRepository.basePath();
57-
commonBlobPath = commonBlobPath.add(indexUUID).add(String.valueOf(shardId.id())).add(SEGMENTS);
63+
BlobPath repositoryBasePath = blobStoreRepository.basePath();
64+
String shardIdStr = String.valueOf(shardId.id());
5865

66+
// Derive the path for data directory of SEGMENTS
67+
BlobPath dataBlobPath = pathType.generatePath(repositoryBasePath, indexUUID, shardIdStr, SEGMENTS, DATA_DIR);
5968
RemoteDirectory dataDirectory = new RemoteDirectory(
60-
blobStoreRepository.blobStore().blobContainer(commonBlobPath.add("data")),
69+
blobStoreRepository.blobStore().blobContainer(dataBlobPath),
6170
blobStoreRepository::maybeRateLimitRemoteUploadTransfers,
6271
blobStoreRepository::maybeRateLimitRemoteDownloadTransfers
6372
);
64-
RemoteDirectory metadataDirectory = new RemoteDirectory(
65-
blobStoreRepository.blobStore().blobContainer(commonBlobPath.add("metadata"))
66-
);
73+
74+
// Derive the path for metadata directory of SEGMENTS
75+
BlobPath mdBlobPath = pathType.generatePath(repositoryBasePath, indexUUID, shardIdStr, SEGMENTS, METADATA_DIR);
76+
RemoteDirectory metadataDirectory = new RemoteDirectory(blobStoreRepository.blobStore().blobContainer(mdBlobPath));
77+
78+
// The path for lock is derived within the RemoteStoreLockManagerFactory
6779
RemoteStoreLockManager mdLockManager = RemoteStoreLockManagerFactory.newLockManager(
6880
repositoriesService.get(),
6981
repositoryName,
7082
indexUUID,
71-
String.valueOf(shardId.id())
83+
shardIdStr,
84+
pathType
7285
);
7386

7487
return new RemoteSegmentStoreDirectory(dataDirectory, metadataDirectory, mdLockManager, threadPool, shardId);

‎server/src/main/java/org/opensearch/index/store/lockmanager/RemoteStoreLockManagerFactory.java

+11-24
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,13 @@
1111
import org.opensearch.common.annotation.PublicApi;
1212
import org.opensearch.common.blobstore.BlobContainer;
1313
import org.opensearch.common.blobstore.BlobPath;
14+
import org.opensearch.index.remote.RemoteStorePathType;
1415
import org.opensearch.index.store.RemoteBufferedOutputDirectory;
1516
import org.opensearch.repositories.RepositoriesService;
1617
import org.opensearch.repositories.Repository;
1718
import org.opensearch.repositories.RepositoryMissingException;
1819
import org.opensearch.repositories.blobstore.BlobStoreRepository;
1920

20-
import java.io.IOException;
2121
import java.util.function.Supplier;
2222

2323
/**
@@ -28,33 +28,30 @@
2828
@PublicApi(since = "2.8.0")
2929
public class RemoteStoreLockManagerFactory {
3030
private static final String SEGMENTS = "segments";
31-
private static final String LOCK_FILES = "lock_files";
31+
public static final String LOCK_FILES = "lock_files";
3232
private final Supplier<RepositoriesService> repositoriesService;
3333

3434
public RemoteStoreLockManagerFactory(Supplier<RepositoriesService> repositoriesService) {
3535
this.repositoriesService = repositoriesService;
3636
}
3737

38-
public RemoteStoreLockManager newLockManager(String repositoryName, String indexUUID, String shardId) throws IOException {
39-
return newLockManager(repositoriesService.get(), repositoryName, indexUUID, shardId);
38+
public RemoteStoreLockManager newLockManager(String repositoryName, String indexUUID, String shardId, RemoteStorePathType pathType) {
39+
return newLockManager(repositoriesService.get(), repositoryName, indexUUID, shardId, pathType);
4040
}
4141

4242
public static RemoteStoreMetadataLockManager newLockManager(
4343
RepositoriesService repositoriesService,
4444
String repositoryName,
4545
String indexUUID,
46-
String shardId
47-
) throws IOException {
46+
String shardId,
47+
RemoteStorePathType pathType
48+
) {
4849
try (Repository repository = repositoriesService.repository(repositoryName)) {
4950
assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
50-
BlobPath shardLevelBlobPath = ((BlobStoreRepository) repository).basePath().add(indexUUID).add(shardId).add(SEGMENTS);
51-
RemoteBufferedOutputDirectory shardMDLockDirectory = createRemoteBufferedOutputDirectory(
52-
repository,
53-
shardLevelBlobPath,
54-
LOCK_FILES
55-
);
56-
57-
return new RemoteStoreMetadataLockManager(shardMDLockDirectory);
51+
BlobPath repositoryBasePath = ((BlobStoreRepository) repository).basePath();
52+
BlobPath lockDirectoryPath = pathType.generatePath(repositoryBasePath, indexUUID, shardId, SEGMENTS, LOCK_FILES);
53+
BlobContainer lockDirectoryBlobContainer = ((BlobStoreRepository) repository).blobStore().blobContainer(lockDirectoryPath);
54+
return new RemoteStoreMetadataLockManager(new RemoteBufferedOutputDirectory(lockDirectoryBlobContainer));
5855
} catch (RepositoryMissingException e) {
5956
throw new IllegalArgumentException("Repository should be present to acquire/release lock", e);
6057
}
@@ -65,14 +62,4 @@ public static RemoteStoreMetadataLockManager newLockManager(
6562
public Supplier<RepositoriesService> getRepositoriesService() {
6663
return repositoriesService;
6764
}
68-
69-
private static RemoteBufferedOutputDirectory createRemoteBufferedOutputDirectory(
70-
Repository repository,
71-
BlobPath commonBlobPath,
72-
String extention
73-
) {
74-
BlobPath extendedPath = commonBlobPath.add(extention);
75-
BlobContainer dataBlobContainer = ((BlobStoreRepository) repository).blobStore().blobContainer(extendedPath);
76-
return new RemoteBufferedOutputDirectory(dataBlobContainer);
77-
}
7865
}

0 commit comments

Comments
 (0)