Skip to content

Commit 2cc86b3

Browse files
parkertimminsparkertimmins
authored
Add read-block to source index during data stream reindex (#122887)
When reindexing a data stream, we currently add a write block to the source indices so that new documents cannot be added to the index while it is being reindexed. A write block still allows the index to be deleted and for the metadata to be updated. It is possible that ILM could delete a backing index or update a backing index's lifecycle metadata while it is being reindexed. To avoid this, this PR sets a read-only block on the source index. This block must be removed before source index can be deleted after it is replaced with the destination index.
1 parent af364a4 commit 2cc86b3

File tree

3 files changed

+46
-80
lines changed

3 files changed

+46
-80
lines changed

‎x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java‎

Lines changed: 10 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77

88
package org.elasticsearch.xpack.migrate.action;
99

10-
import org.elasticsearch.ElasticsearchException;
1110
import org.elasticsearch.ResourceNotFoundException;
1211
import org.elasticsearch.action.DocWriteRequest;
1312
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
@@ -16,7 +15,6 @@
1615
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
1716
import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
1817
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
19-
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsResponse;
2018
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
2119
import org.elasticsearch.action.admin.indices.template.delete.DeleteIndexTemplateRequest;
2220
import org.elasticsearch.action.admin.indices.template.delete.TransportDeleteIndexTemplateAction;
@@ -30,11 +28,9 @@
3028
import org.elasticsearch.action.ingest.PutPipelineTransportAction;
3129
import org.elasticsearch.action.support.IndicesOptions;
3230
import org.elasticsearch.action.support.master.AcknowledgedRequest;
33-
import org.elasticsearch.cluster.block.ClusterBlockException;
3431
import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
3532
import org.elasticsearch.cluster.metadata.IndexMetadata;
3633
import org.elasticsearch.cluster.metadata.MappingMetadata;
37-
import org.elasticsearch.cluster.metadata.MetadataIndexStateService;
3834
import org.elasticsearch.cluster.metadata.Template;
3935
import org.elasticsearch.common.bytes.BytesArray;
4036
import org.elasticsearch.common.compress.CompressedXContent;
@@ -82,7 +78,6 @@
8278
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
8379
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
8480
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
85-
import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
8681
import static org.hamcrest.Matchers.equalTo;
8782

8883
public class ReindexDatastreamIndexTransportActionIT extends ESIntegTestCase {
@@ -274,8 +269,7 @@ public void testDestIndexNameSet_noDotPrefix() throws Exception {
274269
assertEquals(expectedDestIndexName, response.getDestIndex());
275270
}
276271

277-
public void testDestIndexNameSet_withDotPrefix() throws Exception {
278-
272+
public void testDestIndexNameSet_withDotPrefix() {
279273
var sourceIndex = "." + randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
280274
safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex)));
281275

@@ -288,13 +282,19 @@ public void testDestIndexNameSet_withDotPrefix() throws Exception {
288282
assertEquals(expectedDestIndexName, response.getDestIndex());
289283
}
290284

291-
public void testDestIndexContainsDocs() throws Exception {
285+
public void testDestIndexContainsDocs() {
292286
// source index with docs
293287
var numDocs = randomIntBetween(1, 100);
294288
var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
295289
safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex)));
296290
indexDocs(sourceIndex, numDocs);
297291

292+
var settings = Settings.builder()
293+
.put(IndexMetadata.SETTING_BLOCKS_METADATA, randomBoolean())
294+
.put(IndexMetadata.SETTING_READ_ONLY, randomBoolean())
295+
.build();
296+
safeGet(indicesAdmin().updateSettings(new UpdateSettingsRequest(settings, sourceIndex)));
297+
298298
// call reindex
299299
var response = safeGet(
300300
client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex))
@@ -305,31 +305,6 @@ public void testDestIndexContainsDocs() throws Exception {
305305
assertHitCount(prepareSearch(response.getDestIndex()).setSize(0), numDocs);
306306
}
307307

308-
public void testSetSourceToBlockWrites() throws Exception {
309-
var settings = randomBoolean() ? Settings.builder().put(IndexMetadata.SETTING_BLOCKS_WRITE, true).build() : Settings.EMPTY;
310-
311-
// empty source index
312-
var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
313-
safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex, settings)));
314-
315-
// call reindex
316-
safeGet(client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex)));
317-
318-
// Assert that source index is now read-only but not verified read-only
319-
GetSettingsResponse getSettingsResponse = safeGet(
320-
admin().indices().getSettings(new GetSettingsRequest(TEST_REQUEST_TIMEOUT).indices(sourceIndex))
321-
);
322-
assertTrue(parseBoolean(getSettingsResponse.getSetting(sourceIndex, IndexMetadata.SETTING_BLOCKS_WRITE)));
323-
assertFalse(
324-
parseBoolean(getSettingsResponse.getSetting(sourceIndex, MetadataIndexStateService.VERIFIED_READ_ONLY_SETTING.getKey()))
325-
);
326-
327-
// assert that write to source fails
328-
var indexReq = new IndexRequest(sourceIndex).source(jsonBuilder().startObject().field("field", "1").endObject());
329-
expectThrows(ClusterBlockException.class, client().index(indexReq));
330-
assertHitCount(prepareSearch(sourceIndex).setSize(0), 0);
331-
}
332-
333308
public void testMissingSourceIndex() {
334309
var nonExistentSourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
335310
expectThrows(
@@ -387,34 +362,6 @@ public void testMappingsAddedToDestIndex() {
387362
assertEquals("text", XContentMapValues.extractValue("properties.foo1.type", destMappings));
388363
}
389364

390-
public void testFailIfMetadataBlockSet() {
391-
var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
392-
var settings = Settings.builder().put(IndexMetadata.SETTING_BLOCKS_METADATA, true).build();
393-
safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex, settings)));
394-
395-
ElasticsearchException e = expectThrows(
396-
ElasticsearchException.class,
397-
client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex))
398-
);
399-
assertTrue(e.getMessage().contains("Cannot reindex index") || e.getCause().getMessage().equals("Cannot reindex index"));
400-
401-
cleanupMetadataBlocks(sourceIndex);
402-
}
403-
404-
public void testFailIfReadBlockSet() {
405-
var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
406-
var settings = Settings.builder().put(IndexMetadata.SETTING_BLOCKS_READ, true).build();
407-
safeGet(indicesAdmin().create(new CreateIndexRequest(sourceIndex, settings)));
408-
409-
ElasticsearchException e = expectThrows(
410-
ElasticsearchException.class,
411-
client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex))
412-
);
413-
assertTrue(e.getMessage().contains("Cannot reindex index") || e.getCause().getMessage().equals("Cannot reindex index"));
414-
415-
cleanupMetadataBlocks(sourceIndex);
416-
}
417-
418365
public void testReadOnlyBlocksNotAddedBack() {
419366
var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
420367
var settings = Settings.builder()
@@ -434,7 +381,6 @@ public void testReadOnlyBlocksNotAddedBack() {
434381
assertFalse(parseBoolean(settingsResponse.getSetting(destIndex, IndexMetadata.SETTING_READ_ONLY_ALLOW_DELETE)));
435382
assertFalse(parseBoolean(settingsResponse.getSetting(destIndex, IndexMetadata.SETTING_BLOCKS_WRITE)));
436383

437-
cleanupMetadataBlocks(sourceIndex);
438384
cleanupMetadataBlocks(destIndex);
439385
}
440386

@@ -752,9 +698,8 @@ private static void cleanupMetadataBlocks(String index) {
752698
var settings = Settings.builder()
753699
.putNull(IndexMetadata.SETTING_READ_ONLY)
754700
.putNull(IndexMetadata.SETTING_READ_ONLY_ALLOW_DELETE)
755-
.putNull(IndexMetadata.SETTING_BLOCKS_METADATA)
756-
.build();
757-
safeGet(indicesAdmin().updateSettings(new UpdateSettingsRequest(settings, index)));
701+
.putNull(IndexMetadata.SETTING_BLOCKS_METADATA);
702+
updateIndexSettings(settings, index);
758703
}
759704

760705
private static void indexDocs(String index, int numDocs) {

‎x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java‎

Lines changed: 33 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.action.admin.indices.readonly.TransportAddIndexBlockAction;
2525
import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
2626
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
27+
import org.elasticsearch.action.admin.indices.settings.put.TransportUpdateSettingsAction;
2728
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
2829
import org.elasticsearch.action.bulk.BulkItemResponse;
2930
import org.elasticsearch.action.search.SearchRequest;
@@ -55,11 +56,13 @@
5556
import org.elasticsearch.xpack.core.deprecation.DeprecatedIndexPredicate;
5657
import org.elasticsearch.xpack.migrate.MigrateTemplateRegistry;
5758

59+
import java.util.Arrays;
5860
import java.util.Locale;
5961
import java.util.Map;
6062
import java.util.Objects;
6163

62-
import static org.elasticsearch.cluster.metadata.IndexMetadata.APIBlock.WRITE;
64+
import static org.elasticsearch.cluster.metadata.IndexMetadata.APIBlock.METADATA;
65+
import static org.elasticsearch.cluster.metadata.IndexMetadata.APIBlock.READ_ONLY;
6366

6467
public class ReindexDataStreamIndexTransportAction extends HandledTransportAction<
6568
ReindexDataStreamIndexAction.Request,
@@ -145,19 +148,10 @@ protected void doExecute(
145148
);
146149
}
147150

148-
if (settingsBefore.getAsBoolean(IndexMetadata.SETTING_BLOCKS_READ, false)) {
149-
var errorMessage = String.format(Locale.ROOT, "Cannot reindex index [%s] which has a read block.", destIndexName);
150-
listener.onFailure(new ElasticsearchException(errorMessage));
151-
return;
152-
}
153-
if (settingsBefore.getAsBoolean(IndexMetadata.SETTING_BLOCKS_METADATA, false)) {
154-
var errorMessage = String.format(Locale.ROOT, "Cannot reindex index [%s] which has a metadata block.", destIndexName);
155-
listener.onFailure(new ElasticsearchException(errorMessage));
156-
return;
157-
}
158151
final boolean wasClosed = isClosed(sourceIndex);
159-
SubscribableListener.<AcknowledgedResponse>newForked(l -> setBlockWrites(sourceIndexName, l, taskId))
152+
SubscribableListener.<AcknowledgedResponse>newForked(l -> removeMetadataBlocks(sourceIndexName, taskId, l))
160153
.<OpenIndexResponse>andThen(l -> openIndexIfClosed(sourceIndexName, wasClosed, l, taskId))
154+
.<AcknowledgedResponse>andThen(l -> setReadOnly(sourceIndexName, l, taskId))
161155
.<BroadcastResponse>andThen(l -> refresh(sourceIndexName, l, taskId))
162156
.<AcknowledgedResponse>andThen(l -> deleteDestIfExists(destIndexName, l, taskId))
163157
.<AcknowledgedResponse>andThen(l -> createIndex(sourceIndex, destIndexName, l, taskId))
@@ -166,6 +160,7 @@ protected void doExecute(
166160
.<AcknowledgedResponse>andThen(l -> copyIndexMetadataToDest(sourceIndexName, destIndexName, l, taskId))
167161
.<AcknowledgedResponse>andThen(l -> sanityCheck(sourceIndexName, destIndexName, l, taskId))
168162
.<CloseIndexResponse>andThen(l -> closeIndexIfWasClosed(destIndexName, wasClosed, l, taskId))
163+
.<AcknowledgedResponse>andThen(l -> removeAPIBlocks(sourceIndexName, taskId, l, READ_ONLY))
169164
.andThenApply(ignored -> new ReindexDataStreamIndexAction.Response(destIndexName))
170165
.addListener(listener);
171166
}
@@ -201,9 +196,9 @@ private static boolean isClosed(IndexMetadata indexMetadata) {
201196
return indexMetadata.getState().equals(IndexMetadata.State.CLOSE);
202197
}
203198

204-
private void setBlockWrites(String sourceIndexName, ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
205-
logger.debug("Setting write block on source index [{}]", sourceIndexName);
206-
addBlockToIndex(WRITE, sourceIndexName, new ActionListener<>() {
199+
private void setReadOnly(String sourceIndexName, ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
200+
logger.debug("Setting read-only on source index [{}]", sourceIndexName);
201+
addBlockToIndex(READ_ONLY, sourceIndexName, new ActionListener<>() {
207202
@Override
208203
public void onResponse(AddIndexBlockResponse response) {
209204
if (response.isAcknowledged()) {
@@ -399,6 +394,29 @@ private void addBlockToIndex(
399394
client.admin().indices().execute(TransportAddIndexBlockAction.TYPE, addIndexBlockRequest, listener);
400395
}
401396

397+
/**
398+
* All metadata blocks need to be removed at the start for the following reasons:
399+
* 1) If the source index has a metadata only block, the read-only block can't be added.
400+
* 2) If the source index is read-only and closed, it can't be opened.
401+
*/
402+
private void removeMetadataBlocks(String indexName, TaskId parentTaskId, ActionListener<AcknowledgedResponse> listener) {
403+
logger.debug("Removing metadata blocks from index [{}]", indexName);
404+
removeAPIBlocks(indexName, parentTaskId, listener, METADATA, READ_ONLY);
405+
}
406+
407+
private void removeAPIBlocks(
408+
String indexName,
409+
TaskId parentTaskId,
410+
ActionListener<AcknowledgedResponse> listener,
411+
IndexMetadata.APIBlock... blocks
412+
) {
413+
Settings.Builder settings = Settings.builder();
414+
Arrays.stream(blocks).forEach(b -> settings.putNull(b.settingName()));
415+
var updateSettingsRequest = new UpdateSettingsRequest(settings.build(), indexName);
416+
updateSettingsRequest.setParentTask(parentTaskId);
417+
client.execute(TransportUpdateSettingsAction.TYPE, updateSettingsRequest, listener);
418+
}
419+
402420
private void getIndexDocCount(String index, TaskId parentTaskId, ActionListener<Long> listener) {
403421
SearchRequest countRequest = new SearchRequest(index);
404422
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().size(0).trackTotalHits(true);

‎x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/DataStreamsUpgradeIT.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -548,6 +548,9 @@ private void upgradeDataStream(String dataStreamName, int numRolloversOnOldClust
548548
if (randomBoolean()) {
549549
closeIndex(oldIndexName);
550550
}
551+
if (randomBoolean()) {
552+
assertOK(client().performRequest(new Request("PUT", oldIndexName + "/_block/read_only")));
553+
}
551554
}
552555
Request reindexRequest = new Request("POST", "/_migration/reindex");
553556
reindexRequest.setJsonEntity(Strings.format("""

0 commit comments

Comments
 (0)