Skip to content

Commit 151a671

Browse files
ashking94ashking94
authored
[Backport 2.x] Refactor remote store flow to support any path type opensearch-project#12822 opensearch-project#12920 (opensearch-project#13405)
* Refactor remote store flow to support any path type (opensearch-project#12822) Signed-off-by: Ashish Singh <ssashish@amazon.com> * Add missed API visibility annotations for public APIs (opensearch-project#12920) Signed-off-by: Ashish Singh <ssashish@amazon.com> --------- Signed-off-by: Ashish Singh <ssashish@amazon.com>
1 parent de875c5 commit 151a671

21 files changed

+427
-111
lines changed

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

+16-8
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/common/blobstore/BlobPath.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
package org.opensearch.common.blobstore;
3434

3535
import org.opensearch.common.Nullable;
36+
import org.opensearch.common.annotation.PublicApi;
3637

3738
import java.util.ArrayList;
3839
import java.util.Collections;
@@ -42,8 +43,9 @@
4243
/**
4344
* The list of paths where a blob can reside. The contents of the paths are dependent upon the implementation of {@link BlobContainer}.
4445
*
45-
* @opensearch.internal
46+
* @opensearch.api
4647
*/
48+
@PublicApi(since = "1.0.0")
4749
public class BlobPath implements Iterable<String> {
4850

4951
private static final String SEPARATOR = "/";

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

+2-1
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

+9
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

+69
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
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 org.opensearch.common.annotation.PublicApi;
12+
13+
import java.util.Set;
14+
15+
import static org.opensearch.index.remote.RemoteStoreDataEnums.DataType.DATA;
16+
import static org.opensearch.index.remote.RemoteStoreDataEnums.DataType.METADATA;
17+
18+
/**
19+
* This class contains the different enums related to remote store data categories and types.
20+
*
21+
* @opensearch.api
22+
*/
23+
public class RemoteStoreDataEnums {
24+
25+
/**
26+
* Categories of the data in Remote store.
27+
*/
28+
@PublicApi(since = "2.14.0")
29+
public enum DataCategory {
30+
SEGMENTS("segments", Set.of(DataType.values())),
31+
TRANSLOG("translog", Set.of(DATA, METADATA));
32+
33+
private final String name;
34+
private final Set<DataType> supportedDataTypes;
35+
36+
DataCategory(String name, Set<DataType> supportedDataTypes) {
37+
this.name = name;
38+
this.supportedDataTypes = supportedDataTypes;
39+
}
40+
41+
public boolean isSupportedDataType(DataType dataType) {
42+
return supportedDataTypes.contains(dataType);
43+
}
44+
45+
public String getName() {
46+
return name;
47+
}
48+
}
49+
50+
/**
51+
* Types of data in remote store.
52+
*/
53+
@PublicApi(since = "2.14.0")
54+
public enum DataType {
55+
DATA("data"),
56+
METADATA("metadata"),
57+
LOCK_FILES("lock_files");
58+
59+
private final String name;
60+
61+
DataType(String name) {
62+
this.name = name;
63+
}
64+
65+
public String getName() {
66+
return name;
67+
}
68+
}
69+
}

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

+42-3
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,11 @@
88

99
package org.opensearch.index.remote;
1010

11+
import org.opensearch.common.annotation.PublicApi;
12+
import org.opensearch.common.blobstore.BlobPath;
13+
import org.opensearch.index.remote.RemoteStoreDataEnums.DataCategory;
14+
import org.opensearch.index.remote.RemoteStoreDataEnums.DataType;
15+
1116
import java.util.Locale;
1217

1318
/**
@@ -16,15 +21,49 @@
1621
*
1722
* @opensearch.internal
1823
*/
24+
@PublicApi(since = "2.14.0")
1925
public enum RemoteStorePathType {
2026

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

2461
public static RemoteStorePathType parseString(String remoteStoreBlobPathType) {
2562
try {
2663
return RemoteStorePathType.valueOf(remoteStoreBlobPathType.toUpperCase(Locale.ROOT));
27-
} catch (IllegalArgumentException e) {
64+
} catch (IllegalArgumentException | NullPointerException e) {
65+
// IllegalArgumentException is thrown when the input does not match any enum name
66+
// NullPointerException is thrown when the input is null
2867
throw new IllegalArgumentException("Could not parse RemoteStorePathType for [" + remoteStoreBlobPathType + "]");
2968
}
3069
}

‎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/IndexShard.java

+9-2
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

+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;
@@ -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

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

0 commit comments

Comments
 (0)