Skip to content

Commit 4abcf39

Browse files
alchemist51alchemist51
authored
Refactor remote-routing-table service inline with remote state interfaces (opensearch-project#14668)
--------- Signed-off-by: Arpit Bandejiya <abandeji@amazon.com> Signed-off-by: Arpit-Bandejiya <abandeji@amazon.com>
1 parent b3b743d commit 4abcf39

19 files changed

+799
-663
lines changed

‎CHANGELOG.md

+1
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
1919
- Print reason why parent task was cancelled ([#14604](https://github.com/opensearch-project/OpenSearch/issues/14604))
2020
- Add matchesPluginSystemIndexPattern to SystemIndexRegistry ([#14750](https://github.com/opensearch-project/OpenSearch/pull/14750))
2121
- Add Plugin interface for loading application based configuration templates (([#14659](https://github.com/opensearch-project/OpenSearch/issues/14659)))
22+
- Refactor remote-routing-table service inline with remote state interfaces([#14668](https://github.com/opensearch-project/OpenSearch/pull/14668))
2223

2324
### Dependencies
2425
- Bump `org.gradle.test-retry` from 1.5.8 to 1.5.9 ([#13442](https://github.com/opensearch-project/OpenSearch/pull/13442))

‎server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java

+41-191
Original file line numberDiff line numberDiff line change
@@ -11,35 +11,24 @@
1111
import org.apache.logging.log4j.LogManager;
1212
import org.apache.logging.log4j.Logger;
1313
import org.apache.logging.log4j.message.ParameterizedMessage;
14-
import org.apache.lucene.store.IndexInput;
1514
import org.opensearch.action.LatchedActionListener;
16-
import org.opensearch.cluster.ClusterState;
1715
import org.opensearch.cluster.DiffableUtils;
1816
import org.opensearch.cluster.routing.IndexRoutingTable;
1917
import org.opensearch.cluster.routing.RoutingTable;
2018
import org.opensearch.common.CheckedRunnable;
21-
import org.opensearch.common.blobstore.AsyncMultiStreamBlobContainer;
22-
import org.opensearch.common.blobstore.BlobContainer;
2319
import org.opensearch.common.blobstore.BlobPath;
24-
import org.opensearch.common.blobstore.stream.write.WritePriority;
25-
import org.opensearch.common.blobstore.transfer.RemoteTransferContainer;
26-
import org.opensearch.common.blobstore.transfer.stream.OffsetRangeIndexInputStream;
27-
import org.opensearch.common.io.stream.BytesStreamOutput;
2820
import org.opensearch.common.lifecycle.AbstractLifecycleComponent;
29-
import org.opensearch.common.lucene.store.ByteArrayIndexInput;
21+
import org.opensearch.common.remote.RemoteWritableEntityStore;
3022
import org.opensearch.common.settings.ClusterSettings;
31-
import org.opensearch.common.settings.Setting;
3223
import org.opensearch.common.settings.Settings;
3324
import org.opensearch.common.util.io.IOUtils;
3425
import org.opensearch.core.action.ActionListener;
35-
import org.opensearch.core.common.bytes.BytesReference;
36-
import org.opensearch.core.index.Index;
26+
import org.opensearch.core.compress.Compressor;
3727
import org.opensearch.gateway.remote.ClusterMetadataManifest;
3828
import org.opensearch.gateway.remote.RemoteStateTransferException;
29+
import org.opensearch.gateway.remote.model.RemoteRoutingTableBlobStore;
3930
import org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable;
40-
import org.opensearch.index.remote.RemoteStoreEnums;
41-
import org.opensearch.index.remote.RemoteStorePathStrategy;
42-
import org.opensearch.index.remote.RemoteStoreUtils;
31+
import org.opensearch.index.translog.transfer.BlobStoreTransferService;
4332
import org.opensearch.node.Node;
4433
import org.opensearch.node.remotestore.RemoteStoreNodeAttribute;
4534
import org.opensearch.repositories.RepositoriesService;
@@ -52,12 +41,10 @@
5241
import java.util.List;
5342
import java.util.Map;
5443
import java.util.Optional;
55-
import java.util.concurrent.ExecutorService;
5644
import java.util.function.Function;
5745
import java.util.function.Supplier;
5846
import java.util.stream.Collectors;
5947

60-
import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER;
6148
import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
6249

6350
/**
@@ -67,64 +54,29 @@
6754
*/
6855
public class InternalRemoteRoutingTableService extends AbstractLifecycleComponent implements RemoteRoutingTableService {
6956

70-
/**
71-
* This setting is used to set the remote routing table store blob store path type strategy.
72-
*/
73-
public static final Setting<RemoteStoreEnums.PathType> REMOTE_ROUTING_TABLE_PATH_TYPE_SETTING = new Setting<>(
74-
"cluster.remote_store.routing_table.path_type",
75-
RemoteStoreEnums.PathType.HASHED_PREFIX.toString(),
76-
RemoteStoreEnums.PathType::parseString,
77-
Setting.Property.NodeScope,
78-
Setting.Property.Dynamic
79-
);
80-
81-
/**
82-
* This setting is used to set the remote routing table store blob store path hash algorithm strategy.
83-
* This setting will come to effect if the {@link #REMOTE_ROUTING_TABLE_PATH_TYPE_SETTING}
84-
* is either {@code HASHED_PREFIX} or {@code HASHED_INFIX}.
85-
*/
86-
public static final Setting<RemoteStoreEnums.PathHashAlgorithm> REMOTE_ROUTING_TABLE_PATH_HASH_ALGO_SETTING = new Setting<>(
87-
"cluster.remote_store.routing_table.path_hash_algo",
88-
RemoteStoreEnums.PathHashAlgorithm.FNV_1A_BASE64.toString(),
89-
RemoteStoreEnums.PathHashAlgorithm::parseString,
90-
Setting.Property.NodeScope,
91-
Setting.Property.Dynamic
92-
);
93-
94-
public static final String INDEX_ROUTING_PATH_TOKEN = "index-routing";
95-
public static final String INDEX_ROUTING_FILE_PREFIX = "index_routing";
96-
public static final String INDEX_ROUTING_METADATA_PREFIX = "indexRouting--";
97-
9857
private static final Logger logger = LogManager.getLogger(InternalRemoteRoutingTableService.class);
9958
private final Settings settings;
10059
private final Supplier<RepositoriesService> repositoriesService;
60+
private Compressor compressor;
61+
private RemoteWritableEntityStore<IndexRoutingTable, RemoteIndexRoutingTable> remoteIndexRoutingTableStore;
62+
private final ClusterSettings clusterSettings;
10163
private BlobStoreRepository blobStoreRepository;
102-
private RemoteStoreEnums.PathType pathType;
103-
private RemoteStoreEnums.PathHashAlgorithm pathHashAlgo;
104-
private ThreadPool threadPool;
64+
private final ThreadPool threadPool;
65+
private final String clusterName;
10566

10667
public InternalRemoteRoutingTableService(
10768
Supplier<RepositoriesService> repositoriesService,
10869
Settings settings,
10970
ClusterSettings clusterSettings,
110-
ThreadPool threadpool
71+
ThreadPool threadpool,
72+
String clusterName
11173
) {
11274
assert isRemoteRoutingTableEnabled(settings) : "Remote routing table is not enabled";
11375
this.repositoriesService = repositoriesService;
11476
this.settings = settings;
115-
this.pathType = clusterSettings.get(REMOTE_ROUTING_TABLE_PATH_TYPE_SETTING);
116-
this.pathHashAlgo = clusterSettings.get(REMOTE_ROUTING_TABLE_PATH_HASH_ALGO_SETTING);
117-
clusterSettings.addSettingsUpdateConsumer(REMOTE_ROUTING_TABLE_PATH_TYPE_SETTING, this::setPathTypeSetting);
118-
clusterSettings.addSettingsUpdateConsumer(REMOTE_ROUTING_TABLE_PATH_HASH_ALGO_SETTING, this::setPathHashAlgoSetting);
11977
this.threadPool = threadpool;
120-
}
121-
122-
private void setPathTypeSetting(RemoteStoreEnums.PathType pathType) {
123-
this.pathType = pathType;
124-
}
125-
126-
private void setPathHashAlgoSetting(RemoteStoreEnums.PathHashAlgorithm pathHashAlgo) {
127-
this.pathHashAlgo = pathHashAlgo;
78+
this.clusterName = clusterName;
79+
this.clusterSettings = clusterSettings;
12880
}
12981

13082
public List<IndexRoutingTable> getIndicesRouting(RoutingTable routingTable) {
@@ -151,43 +103,32 @@ public DiffableUtils.MapDiff<String, IndexRoutingTable, Map<String, IndexRouting
151103

152104
/**
153105
* Create async action for writing one {@code IndexRoutingTable} to remote store
154-
* @param clusterState current cluster state
106+
* @param term current term
107+
* @param version current version
108+
* @param clusterUUID current cluster UUID
155109
* @param indexRouting indexRoutingTable to write to remote store
156110
* @param latchedActionListener listener for handling async action response
157-
* @param clusterBasePath base path for remote file
158111
* @return returns runnable async action
159112
*/
160-
public CheckedRunnable<IOException> getIndexRoutingAsyncAction(
161-
ClusterState clusterState,
113+
@Override
114+
public CheckedRunnable<IOException> getAsyncIndexRoutingWriteAction(
115+
String clusterUUID,
116+
long term,
117+
long version,
162118
IndexRoutingTable indexRouting,
163-
LatchedActionListener<ClusterMetadataManifest.UploadedMetadata> latchedActionListener,
164-
BlobPath clusterBasePath
119+
LatchedActionListener<ClusterMetadataManifest.UploadedMetadata> latchedActionListener
165120
) {
166121

167-
BlobPath indexRoutingPath = clusterBasePath.add(INDEX_ROUTING_PATH_TOKEN);
168-
BlobPath path = pathType.path(
169-
RemoteStorePathStrategy.PathInput.builder().basePath(indexRoutingPath).indexUUID(indexRouting.getIndex().getUUID()).build(),
170-
pathHashAlgo
171-
);
172-
final BlobContainer blobContainer = blobStoreRepository.blobStore().blobContainer(path);
173-
174-
final String fileName = getIndexRoutingFileName(clusterState.term(), clusterState.version());
122+
RemoteIndexRoutingTable remoteIndexRoutingTable = new RemoteIndexRoutingTable(indexRouting, clusterUUID, compressor, term, version);
175123

176124
ActionListener<Void> completionListener = ActionListener.wrap(
177-
resp -> latchedActionListener.onResponse(
178-
new ClusterMetadataManifest.UploadedIndexMetadata(
179-
indexRouting.getIndex().getName(),
180-
indexRouting.getIndex().getUUID(),
181-
path.buildAsString() + fileName,
182-
INDEX_ROUTING_METADATA_PREFIX
183-
)
184-
),
125+
resp -> latchedActionListener.onResponse(remoteIndexRoutingTable.getUploadedMetadata()),
185126
ex -> latchedActionListener.onFailure(
186127
new RemoteStateTransferException("Exception in writing index to remote store: " + indexRouting.getIndex().toString(), ex)
187128
)
188129
);
189130

190-
return () -> uploadIndex(indexRouting, fileName, blobContainer, completionListener);
131+
return () -> remoteIndexRoutingTableStore.writeAsync(remoteIndexRoutingTable, completionListener);
191132
}
192133

193134
/**
@@ -214,111 +155,21 @@ public List<ClusterMetadataManifest.UploadedIndexMetadata> getAllUploadedIndices
214155
return new ArrayList<>(allUploadedIndicesRouting.values());
215156
}
216157

217-
private void uploadIndex(
218-
IndexRoutingTable indexRouting,
219-
String fileName,
220-
BlobContainer blobContainer,
221-
ActionListener<Void> completionListener
222-
) {
223-
RemoteIndexRoutingTable indexRoutingInput = new RemoteIndexRoutingTable(indexRouting);
224-
BytesReference bytesInput = null;
225-
try (BytesStreamOutput streamOutput = new BytesStreamOutput()) {
226-
indexRoutingInput.writeTo(streamOutput);
227-
bytesInput = streamOutput.bytes();
228-
} catch (IOException e) {
229-
logger.error("Failed to serialize IndexRoutingTable for [{}]: [{}]", indexRouting, e);
230-
completionListener.onFailure(e);
231-
return;
232-
}
233-
234-
if (blobContainer instanceof AsyncMultiStreamBlobContainer == false) {
235-
try {
236-
blobContainer.writeBlob(fileName, bytesInput.streamInput(), bytesInput.length(), true);
237-
completionListener.onResponse(null);
238-
} catch (IOException e) {
239-
logger.error("Failed to write IndexRoutingTable to remote store for indexRouting [{}]: [{}]", indexRouting, e);
240-
completionListener.onFailure(e);
241-
}
242-
return;
243-
}
244-
245-
try (IndexInput input = new ByteArrayIndexInput("indexrouting", BytesReference.toBytes(bytesInput))) {
246-
try (
247-
RemoteTransferContainer remoteTransferContainer = new RemoteTransferContainer(
248-
fileName,
249-
fileName,
250-
input.length(),
251-
true,
252-
WritePriority.URGENT,
253-
(size, position) -> new OffsetRangeIndexInputStream(input, size, position),
254-
null,
255-
false
256-
)
257-
) {
258-
((AsyncMultiStreamBlobContainer) blobContainer).asyncBlobUpload(
259-
remoteTransferContainer.createWriteContext(),
260-
completionListener
261-
);
262-
} catch (IOException e) {
263-
logger.error("Failed to write IndexRoutingTable to remote store for indexRouting [{}]: [{}]", indexRouting, e);
264-
completionListener.onFailure(e);
265-
}
266-
} catch (IOException e) {
267-
logger.error(
268-
"Failed to create transfer object for IndexRoutingTable for remote store upload for indexRouting [{}]: [{}]",
269-
indexRouting,
270-
e
271-
);
272-
completionListener.onFailure(e);
273-
}
274-
}
275-
276158
@Override
277159
public CheckedRunnable<IOException> getAsyncIndexRoutingReadAction(
160+
String clusterUUID,
278161
String uploadedFilename,
279-
Index index,
280162
LatchedActionListener<IndexRoutingTable> latchedActionListener
281163
) {
282-
int idx = uploadedFilename.lastIndexOf("/");
283-
String blobFileName = uploadedFilename.substring(idx + 1);
284-
BlobContainer blobContainer = blobStoreRepository.blobStore()
285-
.blobContainer(BlobPath.cleanPath().add(uploadedFilename.substring(0, idx)));
286164

287-
return () -> readAsync(
288-
blobContainer,
289-
blobFileName,
290-
index,
291-
threadPool.executor(ThreadPool.Names.REMOTE_STATE_READ),
292-
ActionListener.wrap(
293-
response -> latchedActionListener.onResponse(response.getIndexRoutingTable()),
294-
latchedActionListener::onFailure
295-
)
165+
ActionListener<IndexRoutingTable> actionListener = ActionListener.wrap(
166+
latchedActionListener::onResponse,
167+
latchedActionListener::onFailure
296168
);
297-
}
298169

299-
private void readAsync(
300-
BlobContainer blobContainer,
301-
String name,
302-
Index index,
303-
ExecutorService executorService,
304-
ActionListener<RemoteIndexRoutingTable> listener
305-
) {
306-
executorService.execute(() -> {
307-
try {
308-
listener.onResponse(read(blobContainer, name, index));
309-
} catch (Exception e) {
310-
listener.onFailure(e);
311-
}
312-
});
313-
}
170+
RemoteIndexRoutingTable remoteIndexRoutingTable = new RemoteIndexRoutingTable(uploadedFilename, clusterUUID, compressor);
314171

315-
private RemoteIndexRoutingTable read(BlobContainer blobContainer, String path, Index index) {
316-
try {
317-
return new RemoteIndexRoutingTable(blobContainer.readBlob(path), index);
318-
} catch (IOException | AssertionError e) {
319-
logger.error(() -> new ParameterizedMessage("RoutingTable read failed for path {}", path), e);
320-
throw new RemoteStateTransferException("Failed to read RemoteRoutingTable from Manifest with error ", e);
321-
}
172+
return () -> remoteIndexRoutingTableStore.readAsync(remoteIndexRoutingTable, actionListener);
322173
}
323174

324175
@Override
@@ -335,16 +186,6 @@ public List<ClusterMetadataManifest.UploadedIndexMetadata> getUpdatedIndexRoutin
335186
}).collect(Collectors.toList());
336187
}
337188

338-
private String getIndexRoutingFileName(long term, long version) {
339-
return String.join(
340-
DELIMITER,
341-
INDEX_ROUTING_FILE_PREFIX,
342-
RemoteStoreUtils.invertLong(term),
343-
RemoteStoreUtils.invertLong(version),
344-
RemoteStoreUtils.invertLong(System.currentTimeMillis())
345-
);
346-
}
347-
348189
@Override
349190
protected void doClose() throws IOException {
350191
if (blobStoreRepository != null) {
@@ -362,6 +203,16 @@ protected void doStart() {
362203
final Repository repository = repositoriesService.get().repository(remoteStoreRepo);
363204
assert repository instanceof BlobStoreRepository : "Repository should be instance of BlobStoreRepository";
364205
blobStoreRepository = (BlobStoreRepository) repository;
206+
compressor = blobStoreRepository.getCompressor();
207+
208+
this.remoteIndexRoutingTableStore = new RemoteRoutingTableBlobStore<>(
209+
new BlobStoreTransferService(blobStoreRepository.blobStore(), threadPool),
210+
blobStoreRepository,
211+
clusterName,
212+
threadPool,
213+
ThreadPool.Names.REMOTE_STATE_READ,
214+
clusterSettings
215+
);
365216
}
366217

367218
@Override
@@ -377,5 +228,4 @@ public void deleteStaleIndexRoutingPaths(List<String> stalePaths) throws IOExcep
377228
throw e;
378229
}
379230
}
380-
381231
}

‎server/src/main/java/org/opensearch/cluster/routing/remote/NoopRemoteRoutingTableService.java

+6-8
Original file line numberDiff line numberDiff line change
@@ -9,14 +9,11 @@
99
package org.opensearch.cluster.routing.remote;
1010

1111
import org.opensearch.action.LatchedActionListener;
12-
import org.opensearch.cluster.ClusterState;
1312
import org.opensearch.cluster.DiffableUtils;
1413
import org.opensearch.cluster.routing.IndexRoutingTable;
1514
import org.opensearch.cluster.routing.RoutingTable;
1615
import org.opensearch.common.CheckedRunnable;
17-
import org.opensearch.common.blobstore.BlobPath;
1816
import org.opensearch.common.lifecycle.AbstractLifecycleComponent;
19-
import org.opensearch.core.index.Index;
2017
import org.opensearch.gateway.remote.ClusterMetadataManifest;
2118

2219
import java.io.IOException;
@@ -42,11 +39,12 @@ public DiffableUtils.MapDiff<String, IndexRoutingTable, Map<String, IndexRouting
4239
}
4340

4441
@Override
45-
public CheckedRunnable<IOException> getIndexRoutingAsyncAction(
46-
ClusterState clusterState,
42+
public CheckedRunnable<IOException> getAsyncIndexRoutingWriteAction(
43+
String clusterUUID,
44+
long term,
45+
long version,
4746
IndexRoutingTable indexRouting,
48-
LatchedActionListener<ClusterMetadataManifest.UploadedMetadata> latchedActionListener,
49-
BlobPath clusterBasePath
47+
LatchedActionListener<ClusterMetadataManifest.UploadedMetadata> latchedActionListener
5048
) {
5149
// noop
5250
return () -> {};
@@ -64,8 +62,8 @@ public List<ClusterMetadataManifest.UploadedIndexMetadata> getAllUploadedIndices
6462

6563
@Override
6664
public CheckedRunnable<IOException> getAsyncIndexRoutingReadAction(
65+
String clusterUUID,
6766
String uploadedFilename,
68-
Index index,
6967
LatchedActionListener<IndexRoutingTable> latchedActionListener
7068
) {
7169
// noop

0 commit comments

Comments
 (0)