Skip to content

Commit 887bdbc

Browse files
xuxiong1xuxiong1
authored
Add subdirectory-aware store module with recovery support (opensearch-project#19132)
* Add subdirectory-aware store module with recovery support Signed-off-by: xuxiong1 <xiongxug@outlook.com> * precommit Signed-off-by: xuxiong1 <xiongxug@outlook.com> * update Signed-off-by: xuxiong1 <xiongxug@outlook.com> * address comments Signed-off-by: xuxiong1 <xiongxug@outlook.com> * update Signed-off-by: xuxiong1 <xiongxug@outlook.com> * spotless Signed-off-by: xuxiong1 <xiongxug@outlook.com> * udpate Signed-off-by: xuxiong1 <xiongxug@outlook.com> * udpate Signed-off-by: xuxiong1 <xiongxug@outlook.com> * udpate Signed-off-by: xuxiong1 <xiongxug@outlook.com> --------- Signed-off-by: xuxiong1 <xiongxug@outlook.com> Signed-off-by: Xu Xiong <xiongxug@outlook.com>
1 parent 316e9b8 commit 887bdbc

9 files changed

Lines changed: 792 additions & 4 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
1414
- Use S3CrtClient for higher throughput while uploading files to S3 ([#18800](https://github.com/opensearch-project/OpenSearch/pull/18800))
1515
- Add a dynamic setting to change skip_cache_factor and min_frequency for querycache ([#18351](https://github.com/opensearch-project/OpenSearch/issues/18351))
1616
- Add overload constructor for Translog to accept Channel Factory as a parameter ([#18918](https://github.com/opensearch-project/OpenSearch/pull/18918))
17+
- Add subdirectory-aware store module with recovery support ([#19132](https://github.com/opensearch-project/OpenSearch/pull/19132))
1718

1819
### Changed
1920
- Add CompletionStage variants to methods in the Client Interface and default to ActionListener impl ([#18998](https://github.com/opensearch-project/OpenSearch/pull/18998))

‎modules/store-subdirectory/build.gradle‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
2+
opensearchplugin {
3+
description = 'OpenSearch Subdirectory Store plugin'
4+
classname = 'org.opensearch.plugin.store.subdirectory.SubdirectoryStorePlugin'
5+
}
6+
7+
dependencies {
8+
implementation project(':server')
9+
}
10+
11+
apply plugin: 'opensearch.internal-cluster-test'

‎modules/store-subdirectory/src/internalClusterTest/java/org/opensearch/plugin/store/subdirectory/SubdirectoryAwareRecoveryTests.java‎

Lines changed: 337 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,337 @@
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.plugin.store.subdirectory;
10+
11+
import org.apache.lucene.document.Document;
12+
import org.apache.lucene.document.Field;
13+
import org.apache.lucene.document.StringField;
14+
import org.apache.lucene.index.IndexCommit;
15+
import org.apache.lucene.index.IndexWriter;
16+
import org.apache.lucene.index.IndexWriterConfig;
17+
import org.apache.lucene.index.SegmentInfos;
18+
import org.apache.lucene.store.Directory;
19+
import org.apache.lucene.store.FSDirectory;
20+
import org.opensearch.common.concurrent.GatedCloseable;
21+
import org.opensearch.common.settings.Setting;
22+
import org.opensearch.common.settings.Settings;
23+
import org.opensearch.index.IndexService;
24+
import org.opensearch.index.IndexSettings;
25+
import org.opensearch.index.engine.Engine;
26+
import org.opensearch.index.engine.EngineConfig;
27+
import org.opensearch.index.engine.EngineException;
28+
import org.opensearch.index.engine.EngineFactory;
29+
import org.opensearch.index.engine.InternalEngine;
30+
import org.opensearch.index.shard.IndexShard;
31+
import org.opensearch.indices.IndicesService;
32+
import org.opensearch.plugins.EnginePlugin;
33+
import org.opensearch.plugins.Plugin;
34+
import org.opensearch.test.OpenSearchIntegTestCase;
35+
36+
import java.io.IOException;
37+
import java.nio.file.Files;
38+
import java.nio.file.Path;
39+
import java.util.Arrays;
40+
import java.util.Collection;
41+
import java.util.Collections;
42+
import java.util.HashMap;
43+
import java.util.HashSet;
44+
import java.util.List;
45+
import java.util.Map;
46+
import java.util.Optional;
47+
import java.util.Set;
48+
49+
@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 3)
50+
public class SubdirectoryAwareRecoveryTests extends OpenSearchIntegTestCase {
51+
52+
@Override
53+
protected Collection<Class<? extends Plugin>> nodePlugins() {
54+
return Arrays.asList(SubdirectoryStorePlugin.class, TestEnginePlugin.class);
55+
}
56+
57+
@Override
58+
protected Collection<Class<? extends Plugin>> getMockPlugins() {
59+
return super.getMockPlugins().stream()
60+
.filter(plugin -> !plugin.getName().contains("MockEngineFactoryPlugin"))
61+
.collect(java.util.stream.Collectors.toList());
62+
}
63+
64+
public void testSubdirectoryAwareRecovery() throws Exception {
65+
// Create index with custom store and engine
66+
Settings indexSettings = Settings.builder()
67+
.put("index.number_of_shards", 1)
68+
.put("index.number_of_replicas", 0)
69+
.put("index.store.factory", "subdirectory_store")
70+
.put(TestEnginePlugin.TEST_ENGINE_INDEX_SETTING.getKey(), true)
71+
.build();
72+
73+
prepareCreate("test_index").setSettings(indexSettings).get();
74+
ensureGreen("test_index");
75+
76+
// Index documents to create content
77+
for (int i = 0; i < 10; i++) {
78+
client().prepareIndex("test_index").setId(Integer.toString(i)).setSource("field", "value" + i).get();
79+
}
80+
81+
client().admin().indices().prepareFlush("test_index").get();
82+
client().admin().indices().prepareRefresh("test_index").get();
83+
84+
// Add replica to trigger recovery
85+
client().admin()
86+
.indices()
87+
.prepareUpdateSettings("test_index")
88+
.setSettings(Settings.builder().put("index.number_of_replicas", 2))
89+
.get();
90+
91+
// Verify recovery completes successfully
92+
ensureGreen("test_index");
93+
94+
// Verify subdirectory files were copied to both primary and replica
95+
verifySubdirectoryFilesOnAllNodes("test_index", 3);
96+
}
97+
98+
private void verifySubdirectoryFilesOnAllNodes(String indexName, int expectedCount) throws Exception {
99+
Map<String, Set<String>> nodeFiles = new HashMap<>();
100+
101+
for (String nodeName : internalCluster().getNodeNames()) {
102+
103+
IndicesService indicesService = internalCluster().getInstance(IndicesService.class, nodeName);
104+
IndexService indexService = indicesService.indexService(resolveIndex(indexName));
105+
if (indexService == null) {
106+
continue; // Index not on this node
107+
}
108+
IndexShard shard = indexService.getShard(0);
109+
Path subdirectoryPath = shard.shardPath().getDataPath().resolve(TestEngine.SUBDIRECTORY_NAME);
110+
111+
if (Files.exists(subdirectoryPath)) {
112+
try (Directory directory = FSDirectory.open(subdirectoryPath)) {
113+
SegmentInfos segmentInfos = SegmentInfos.readLatestCommit(directory);
114+
Collection<String> segmentFiles = segmentInfos.files(true);
115+
if (!segmentFiles.isEmpty()) {
116+
nodeFiles.put(nodeName, new HashSet<>(segmentFiles));
117+
}
118+
} catch (IOException e) {
119+
// corrupt index or no commit files, skip this node
120+
}
121+
}
122+
}
123+
124+
assertEquals(
125+
"Expected " + expectedCount + " nodes with subdirectory files, found: " + nodeFiles.keySet(),
126+
expectedCount,
127+
nodeFiles.size()
128+
);
129+
130+
// Verify all nodes have identical files
131+
if (nodeFiles.size() > 1) {
132+
Set<String> referenceFiles = nodeFiles.values().iterator().next();
133+
for (Map.Entry<String, Set<String>> entry : nodeFiles.entrySet()) {
134+
assertEquals("Node " + entry.getKey() + " should have identical files to other nodes", referenceFiles, entry.getValue());
135+
}
136+
}
137+
}
138+
139+
/**
140+
* Plugin that provides a custom engine for testing subdirectory recovery
141+
*/
142+
public static class TestEnginePlugin extends Plugin implements EnginePlugin {
143+
144+
static final Setting<Boolean> TEST_ENGINE_INDEX_SETTING = Setting.boolSetting(
145+
"index.use_test_engine",
146+
false,
147+
Setting.Property.IndexScope
148+
);
149+
150+
@Override
151+
public List<Setting<?>> getSettings() {
152+
return Collections.singletonList(TEST_ENGINE_INDEX_SETTING);
153+
}
154+
155+
@Override
156+
public Optional<EngineFactory> getEngineFactory(IndexSettings indexSettings) {
157+
if (TEST_ENGINE_INDEX_SETTING.get(indexSettings.getSettings())) {
158+
return Optional.of(new TestEngineFactory());
159+
}
160+
return Optional.empty();
161+
}
162+
}
163+
164+
/**
165+
* Factory for creating TestEngine instances
166+
*/
167+
static class TestEngineFactory implements EngineFactory {
168+
@Override
169+
public Engine newReadWriteEngine(EngineConfig config) {
170+
try {
171+
return new TestEngine(config);
172+
} catch (IOException e) {
173+
throw new EngineException(config.getShardId(), "Failed to create test engine", e);
174+
}
175+
}
176+
}
177+
178+
/**
179+
* Test engine that extends InternalEngine and creates a proper Lucene index in a subdirectory
180+
*/
181+
static class TestEngine extends InternalEngine {
182+
183+
static final String SUBDIRECTORY_NAME = "test_subdirectory";
184+
185+
private final Path subdirectoryPath;
186+
private final Directory subdirectoryDirectory;
187+
private final IndexWriter subdirectoryWriter;
188+
private final EngineConfig engineConfig;
189+
190+
TestEngine(EngineConfig config) throws IOException {
191+
super(config);
192+
this.engineConfig = config;
193+
194+
// Set up subdirectory path and writer
195+
Path shardPath = config.getStore().shardPath().getDataPath();
196+
subdirectoryPath = shardPath.resolve(SUBDIRECTORY_NAME);
197+
Files.createDirectories(subdirectoryPath);
198+
subdirectoryDirectory = FSDirectory.open(subdirectoryPath);
199+
IndexWriterConfig writerConfig = new IndexWriterConfig();
200+
writerConfig.setOpenMode(IndexWriterConfig.OpenMode.CREATE_OR_APPEND);
201+
subdirectoryWriter = new IndexWriter(subdirectoryDirectory, writerConfig);
202+
}
203+
204+
@Override
205+
public IndexResult index(Index index) throws IOException {
206+
// First, index the document normally
207+
IndexResult result = super.index(index);
208+
209+
// Only add to subdirectory if is a primary shard
210+
if (result.getResultType() == Engine.Result.Type.SUCCESS && engineConfig.getStartedPrimarySupplier().getAsBoolean()) {
211+
addDocumentToSubdirectory(index);
212+
}
213+
return result;
214+
}
215+
216+
private void addDocumentToSubdirectory(Index index) throws IOException {
217+
Document doc = new Document();
218+
doc.add(new StringField("source_id", index.id(), Field.Store.YES));
219+
subdirectoryWriter.addDocument(doc);
220+
}
221+
222+
@Override
223+
public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
224+
// First flush the main engine
225+
super.flush(force, waitIfOngoing);
226+
// Then commit the subdirectory
227+
try {
228+
subdirectoryWriter.commit();
229+
} catch (IOException e) {
230+
throw new EngineException(shardId, "Failed to commit subdirectory during flush", e);
231+
}
232+
}
233+
234+
@Override
235+
public void close() throws IOException {
236+
subdirectoryWriter.close();
237+
subdirectoryDirectory.close();
238+
super.close();
239+
}
240+
241+
@Override
242+
public GatedCloseable<IndexCommit> acquireLastIndexCommit(boolean flushFirst) throws EngineException {
243+
if (flushFirst) {
244+
flush(false, true);
245+
}
246+
try {
247+
GatedCloseable<IndexCommit> realCommit = super.acquireLastIndexCommit(false);
248+
IndexCommit originalCommit = realCommit.get();
249+
TestIndexCommit testCommit = new TestIndexCommit(originalCommit, subdirectoryPath);
250+
realCommit.close();
251+
return new GatedCloseable<>(testCommit, () -> {});
252+
} catch (Exception e) {
253+
throw new EngineException(shardId, "Failed to acquire last index commit", e);
254+
}
255+
}
256+
257+
@Override
258+
public GatedCloseable<IndexCommit> acquireSafeIndexCommit() throws EngineException {
259+
try {
260+
GatedCloseable<IndexCommit> realCommit = super.acquireSafeIndexCommit();
261+
IndexCommit originalCommit = realCommit.get();
262+
TestIndexCommit testCommit = new TestIndexCommit(originalCommit, subdirectoryPath);
263+
realCommit.close();
264+
return new GatedCloseable<>(testCommit, () -> {});
265+
} catch (Exception e) {
266+
throw new EngineException(shardId, "Failed to acquire safe index commit", e);
267+
}
268+
}
269+
}
270+
271+
/**
272+
* Custom IndexCommit that includes subdirectory files in recovery
273+
*/
274+
static class TestIndexCommit extends IndexCommit {
275+
276+
private final IndexCommit delegate;
277+
private final Path subdirectoryPath;
278+
279+
TestIndexCommit(IndexCommit delegate, Path subdirectoryPath) {
280+
this.delegate = delegate;
281+
this.subdirectoryPath = subdirectoryPath;
282+
}
283+
284+
@Override
285+
public String getSegmentsFileName() {
286+
return delegate.getSegmentsFileName();
287+
}
288+
289+
@Override
290+
public Collection<String> getFileNames() throws IOException {
291+
Set<String> allFiles = new HashSet<>(delegate.getFileNames());
292+
293+
if (Files.exists(subdirectoryPath)) {
294+
try (Directory directory = FSDirectory.open(subdirectoryPath)) {
295+
SegmentInfos segmentInfos = SegmentInfos.readLatestCommit(directory);
296+
Collection<String> segmentFiles = segmentInfos.files(true);
297+
298+
for (String fileName : segmentFiles) {
299+
String relativePath = Path.of(TestEngine.SUBDIRECTORY_NAME, fileName).toString();
300+
allFiles.add(relativePath);
301+
}
302+
}
303+
}
304+
return allFiles;
305+
}
306+
307+
@Override
308+
public Directory getDirectory() {
309+
return delegate.getDirectory();
310+
}
311+
312+
@Override
313+
public void delete() {
314+
delegate.delete();
315+
}
316+
317+
@Override
318+
public int getSegmentCount() {
319+
return delegate.getSegmentCount();
320+
}
321+
322+
@Override
323+
public long getGeneration() {
324+
return delegate.getGeneration();
325+
}
326+
327+
@Override
328+
public Map<String, String> getUserData() throws IOException {
329+
return delegate.getUserData();
330+
}
331+
332+
@Override
333+
public boolean isDeleted() {
334+
return delegate.isDeleted();
335+
}
336+
}
337+
}

0 commit comments

Comments
 (0)