Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add index and reindex request settings to speed up reindex #119780

Merged
merged 6 commits into from
Jan 10, 2025
Merged

Add index and reindex request settings to speed up reindex #119780

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ea30dac
Add index and reindexRequest settings to optimize reindex
parkertimmins Jan 8, 2025
8eaa71d
Update docs/changelog/119780.yaml
parkertimmins Jan 8, 2025
12652fe
Fix bugs when updating settings
parkertimmins Jan 8, 2025
b66e40a
Merge branch 'main' into reindex-datastream-reindex-args
parkertimmins Jan 8, 2025
00847ca
Better naming and comment
parkertimmins Jan 9, 2025
e4d20b9
Merge branch 'main' into reindex-datastream-reindex-args
parkertimmins Jan 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions docs/changelog/119780.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
pr: 119780
summary: Add index and reindex request settings to speed up reindex
area: Data streams
type: enhancement
issues: []
38 changes: 37 additions & 1 deletion .../java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.admin.indices.template.delete.DeleteIndexTemplateRequest;
import org.elasticsearch.action.admin.indices.template.delete.TransportDeleteIndexTemplateAction;
import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
Expand Down Expand Up @@ -155,7 +157,11 @@ public void testSettingsAddedBeforeReindex() throws Exception {

// update with a dynamic setting
var numReplicas = randomIntBetween(0, 10);
var dynamicSettings = Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numReplicas).build();
var refreshInterval = randomIntBetween(1, 100) + "s";
var dynamicSettings = Settings.builder()
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numReplicas)
.put(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(), refreshInterval)
.build();
indicesAdmin().updateSettings(new UpdateSettingsRequest(dynamicSettings, sourceIndex)).actionGet();

// call reindex
Expand All @@ -167,6 +173,7 @@ public void testSettingsAddedBeforeReindex() throws Exception {
var settingsResponse = indicesAdmin().getSettings(new GetSettingsRequest().indices(destIndex)).actionGet();
assertEquals(numReplicas, Integer.parseInt(settingsResponse.getSetting(destIndex, IndexMetadata.SETTING_NUMBER_OF_REPLICAS)));
assertEquals(numShards, Integer.parseInt(settingsResponse.getSetting(destIndex, IndexMetadata.SETTING_NUMBER_OF_SHARDS)));
assertEquals(refreshInterval, settingsResponse.getSetting(destIndex, IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey()));
}

public void testMappingsAddedToDestIndex() throws Exception {
Expand Down Expand Up @@ -229,6 +236,35 @@ public void testReadOnlyAddedBack() {
removeReadOnly(destIndex);
}

public void testUpdateSettingsDefaultsRestored() {
assumeTrue("requires the migration reindex feature flag", REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled());

// random_index_template sets value for number_of_replicas, remove template so default value is used instead
assertAcked(
indicesAdmin().execute(TransportDeleteIndexTemplateAction.TYPE, new DeleteIndexTemplateRequest("random_index_template"))
);
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a bit hacky, but not sure of a better way to force number_of_replicas to use the default value.


var sourceIndex = randomAlphaOfLength(20).toLowerCase(Locale.ROOT);
assertAcked(indicesAdmin().create(new CreateIndexRequest(sourceIndex)));

// call reindex
var destIndex = client().execute(ReindexDataStreamIndexAction.INSTANCE, new ReindexDataStreamIndexAction.Request(sourceIndex))
.actionGet()
.getDestIndex();

var settingsResponse = indicesAdmin().getSettings(new GetSettingsRequest().indices(sourceIndex, destIndex)).actionGet();
var destSettings = settingsResponse.getIndexToSettings().get(destIndex);

assertEquals(
IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.getDefault(destSettings),
IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.get(destSettings)
);
assertEquals(
IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getDefault(destSettings),
IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.get(destSettings)
);
}

public void testSettingsAndMappingsFromTemplate() throws IOException {
assumeTrue("requires the migration reindex feature flag", REINDEX_DATA_STREAM_FEATURE_FLAG.isEnabled());

Expand Down
35 changes: 35 additions & 0 deletions ...in/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockRequest;
import org.elasticsearch.action.admin.indices.readonly.AddIndexBlockResponse;
import org.elasticsearch.action.admin.indices.readonly.TransportAddIndexBlockAction;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.HandledTransportAction;
import org.elasticsearch.action.support.IndicesOptions;
Expand All @@ -25,6 +26,7 @@
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.IndexSettings;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.ReindexAction;
import org.elasticsearch.index.reindex.ReindexRequest;
Expand Down Expand Up @@ -95,6 +97,7 @@ protected void doExecute(
.<AcknowledgedResponse>andThen(l -> deleteDestIfExists(destIndexName, l, taskId))
.<AcknowledgedResponse>andThen(l -> createIndex(sourceIndex, destIndexName, l, taskId))
.<BulkByScrollResponse>andThen(l -> reindex(sourceIndexName, destIndexName, l, taskId))
.<AcknowledgedResponse>andThen(l -> updateSettings(settingsBefore, destIndexName, l, taskId))
.<AddIndexBlockResponse>andThen(l -> addBlockIfFromSource(WRITE, settingsBefore, destIndexName, l, taskId))
.<AddIndexBlockResponse>andThen(l -> addBlockIfFromSource(READ_ONLY, settingsBefore, destIndexName, l, taskId))
.andThenApply(ignored -> new ReindexDataStreamIndexAction.Response(destIndexName))
Expand Down Expand Up @@ -147,6 +150,8 @@ private void createIndex(
var removeReadOnlyOverride = Settings.builder()
.putNull(IndexMetadata.SETTING_READ_ONLY)
.putNull(IndexMetadata.SETTING_BLOCKS_WRITE)
.put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
.put(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(), -1)
.build();

var request = new CreateIndexFromSourceAction.Request(
Expand All @@ -168,6 +173,7 @@ private void reindex(String sourceIndexName, String destIndexName, ActionListene
reindexRequest.getSearchRequest().source().fetchSource(true);
reindexRequest.setDestIndex(destIndexName);
reindexRequest.setParentTask(parentTaskId);
reindexRequest.setSlices(0); // equivalent to slices=auto in rest api
client.execute(ReindexAction.INSTANCE, reindexRequest, listener);
}

Expand All @@ -186,6 +192,35 @@ private void addBlockIfFromSource(
}
}

private void updateSettings(
Settings settingsBefore,
String destIndexName,
ActionListener<AcknowledgedResponse> listener,
TaskId parentTaskId
) {
logger.debug("Updating settings on destination index after reindex completes");

var settings = Settings.builder();
copySettingOrUnset(settingsBefore, settings, IndexMetadata.SETTING_NUMBER_OF_REPLICAS);
copySettingOrUnset(settingsBefore, settings, IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey());

var updateSettingsRequest = new UpdateSettingsRequest(settings.build(), destIndexName);
updateSettingsRequest.setParentTask(parentTaskId);
var errorMessage = String.format(Locale.ROOT, "Could not update settings on index [%s]", destIndexName);
client.admin().indices().updateSettings(updateSettingsRequest, failIfNotAcknowledged(listener, errorMessage));
}

private static void copySettingOrUnset(Settings settingsBefore, Settings.Builder builder, String setting) {
// if setting was explicitly added to the source index
if (settingsBefore.get(setting) != null) {
// copy it back to the dest index
builder.copy(setting, settingsBefore);
} else {
// otherwise, delete from dest index so that it loads from the settings default
builder.putNull(setting);
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Alternatively, we could get the current setting value using one of the methods that falls back to the default, then set this value on the dest index. But I prefer the current version as it avoids adding a settings explicitly to the dest index which had been unset (and using the default) on the source index.

}
}

public static String generateDestIndexName(String sourceIndex) {
return "migrated-" + sourceIndex;
}
Expand Down