Skip to content

Commit dc3aaf1

Browse files
lukewhitinglukewhiting
authored
Make requests_per_second configurable to throttle reindexing (elastic#120207) (elastic#120300)
* Make requests_per_second configurable to throttle reindexing * Update docs/changelog/120207.yaml * Add restrictions to prevent zero or negative rate limit Also allow -1 as infinite * PR Changes - Switch to cluster settings for rate limit retrieval
1 parent 2b1210f commit dc3aaf1

File tree

5 files changed

+246
-28
lines changed

5 files changed

+246
-28
lines changed

‎docs/changelog/120207.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 120207
2+
summary: Make `requests_per_second` configurable to throttle reindexing
3+
area: Data streams
4+
type: enhancement
5+
issues: []

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

Lines changed: 0 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -493,31 +493,4 @@ private static String getIndexUUID(String index) {
493493
.get(IndexMetadata.SETTING_INDEX_UUID);
494494
}
495495

496-
public void testGenerateDestIndexName_noDotPrefix() {
497-
String sourceIndex = "sourceindex";
498-
String expectedDestIndex = "migrated-sourceindex";
499-
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
500-
assertEquals(expectedDestIndex, actualDestIndex);
501-
}
502-
503-
public void testGenerateDestIndexName_withDotPrefix() {
504-
String sourceIndex = ".sourceindex";
505-
String expectedDestIndex = ".migrated-sourceindex";
506-
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
507-
assertEquals(expectedDestIndex, actualDestIndex);
508-
}
509-
510-
public void testGenerateDestIndexName_withHyphen() {
511-
String sourceIndex = "source-index";
512-
String expectedDestIndex = "migrated-source-index";
513-
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
514-
assertEquals(expectedDestIndex, actualDestIndex);
515-
}
516-
517-
public void testGenerateDestIndexName_withUnderscore() {
518-
String sourceIndex = "source_index";
519-
String expectedDestIndex = "migrated-source_index";
520-
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
521-
assertEquals(expectedDestIndex, actualDestIndex);
522-
}
523496
}

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@
5959
import java.util.function.Supplier;
6060

6161
import static org.elasticsearch.xpack.migrate.action.ReindexDataStreamAction.REINDEX_DATA_STREAM_FEATURE_FLAG;
62+
import static org.elasticsearch.xpack.migrate.action.ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING;
6263
import static org.elasticsearch.xpack.migrate.task.ReindexDataStreamPersistentTaskExecutor.MAX_CONCURRENT_INDICES_REINDEXED_PER_DATA_STREAM_SETTING;
6364

6465
public class MigratePlugin extends Plugin implements ActionPlugin, PersistentTaskPlugin {
@@ -160,6 +161,7 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(
160161
public List<Setting<?>> getSettings() {
161162
List<Setting<?>> pluginSettings = new ArrayList<>();
162163
pluginSettings.add(MAX_CONCURRENT_INDICES_REINDEXED_PER_DATA_STREAM_SETTING);
164+
pluginSettings.add(REINDEX_MAX_REQUESTS_PER_SECOND_SETTING);
163165
return pluginSettings;
164166
}
165167
}

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

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.cluster.block.ClusterBlockException;
2525
import org.elasticsearch.cluster.metadata.IndexMetadata;
2626
import org.elasticsearch.cluster.service.ClusterService;
27+
import org.elasticsearch.common.settings.Setting;
2728
import org.elasticsearch.common.settings.Settings;
2829
import org.elasticsearch.core.TimeValue;
2930
import org.elasticsearch.index.IndexSettings;
@@ -46,8 +47,37 @@ public class ReindexDataStreamIndexTransportAction extends HandledTransportActio
4647
ReindexDataStreamIndexAction.Request,
4748
ReindexDataStreamIndexAction.Response> {
4849

50+
public static final String REINDEX_MAX_REQUESTS_PER_SECOND_KEY = "migrate.data_stream_reindex_max_request_per_second";
51+
52+
public static final Setting<Float> REINDEX_MAX_REQUESTS_PER_SECOND_SETTING = new Setting<>(
53+
REINDEX_MAX_REQUESTS_PER_SECOND_KEY,
54+
Float.toString(10f),
55+
s -> {
56+
if (s.equals("-1")) {
57+
return Float.POSITIVE_INFINITY;
58+
} else {
59+
return Float.parseFloat(s);
60+
}
61+
},
62+
value -> {
63+
if (value <= 0f) {
64+
throw new IllegalArgumentException(
65+
"Failed to parse value ["
66+
+ value
67+
+ "] for setting ["
68+
+ REINDEX_MAX_REQUESTS_PER_SECOND_KEY
69+
+ "] "
70+
+ "must be greater than 0 or -1 for infinite"
71+
);
72+
}
73+
},
74+
Setting.Property.Dynamic,
75+
Setting.Property.NodeScope
76+
);
77+
4978
private static final Logger logger = LogManager.getLogger(ReindexDataStreamIndexTransportAction.class);
5079
private static final IndicesOptions IGNORE_MISSING_OPTIONS = IndicesOptions.fromOptions(true, true, false, false);
80+
5181
private final ClusterService clusterService;
5282
private final Client client;
5383

@@ -176,14 +206,16 @@ private void createIndex(
176206
client.execute(CreateIndexFromSourceAction.INSTANCE, request, failIfNotAcknowledged(listener, errorMessage));
177207
}
178208

179-
private void reindex(String sourceIndexName, String destIndexName, ActionListener<BulkByScrollResponse> listener, TaskId parentTaskId) {
209+
// Visible for testing
210+
void reindex(String sourceIndexName, String destIndexName, ActionListener<BulkByScrollResponse> listener, TaskId parentTaskId) {
180211
logger.debug("Reindex to destination index [{}] from source index [{}]", destIndexName, sourceIndexName);
181212
var reindexRequest = new ReindexRequest();
182213
reindexRequest.setSourceIndices(sourceIndexName);
183214
reindexRequest.getSearchRequest().allowPartialSearchResults(false);
184215
reindexRequest.getSearchRequest().source().fetchSource(true);
185216
reindexRequest.setDestIndex(destIndexName);
186217
reindexRequest.setParentTask(parentTaskId);
218+
reindexRequest.setRequestsPerSecond(clusterService.getClusterSettings().get(REINDEX_MAX_REQUESTS_PER_SECOND_SETTING));
187219
reindexRequest.setSlices(0); // equivalent to slices=auto in rest api
188220
client.execute(ReindexAction.INSTANCE, reindexRequest, listener);
189221
}

‎x-pack/plugin/migrate/src/test/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportActionTests.java‎

Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,206 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License
4+
* 2.0; you may not use this file except in compliance with the Elastic License
5+
* 2.0.
6+
*/
7+
8+
package org.elasticsearch.xpack.migrate.action;
9+
10+
import org.elasticsearch.action.ActionListener;
11+
import org.elasticsearch.action.support.ActionFilters;
12+
import org.elasticsearch.client.internal.Client;
13+
import org.elasticsearch.cluster.service.ClusterService;
14+
import org.elasticsearch.common.settings.ClusterSettings;
15+
import org.elasticsearch.common.settings.Settings;
16+
import org.elasticsearch.index.reindex.BulkByScrollResponse;
17+
import org.elasticsearch.index.reindex.ReindexAction;
18+
import org.elasticsearch.index.reindex.ReindexRequest;
19+
import org.elasticsearch.tasks.TaskId;
20+
import org.elasticsearch.test.ESTestCase;
21+
import org.elasticsearch.transport.TransportService;
22+
import org.junit.After;
23+
import org.junit.Before;
24+
import org.mockito.Answers;
25+
import org.mockito.ArgumentCaptor;
26+
import org.mockito.Captor;
27+
import org.mockito.InjectMocks;
28+
import org.mockito.Mock;
29+
import org.mockito.MockitoAnnotations;
30+
31+
import java.util.Collections;
32+
33+
import static org.mockito.ArgumentMatchers.eq;
34+
import static org.mockito.Mockito.doNothing;
35+
import static org.mockito.Mockito.when;
36+
37+
public class ReindexDataStreamIndexTransportActionTests extends ESTestCase {
38+
39+
@Mock(answer = Answers.RETURNS_DEEP_STUBS)
40+
private TransportService transportService;
41+
@Mock
42+
private ClusterService clusterService;
43+
@Mock
44+
private ActionFilters actionFilters;
45+
@Mock
46+
private Client client;
47+
48+
@InjectMocks
49+
private ReindexDataStreamIndexTransportAction action;
50+
51+
@Captor
52+
private ArgumentCaptor<ReindexRequest> request;
53+
54+
private AutoCloseable mocks;
55+
56+
@Before
57+
public void setUp() throws Exception {
58+
super.setUp();
59+
mocks = MockitoAnnotations.openMocks(this);
60+
}
61+
62+
@After
63+
public void tearDown() throws Exception {
64+
super.tearDown();
65+
mocks.close();
66+
}
67+
68+
public void testGenerateDestIndexName_noDotPrefix() {
69+
String sourceIndex = "sourceindex";
70+
String expectedDestIndex = "migrated-sourceindex";
71+
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
72+
assertEquals(expectedDestIndex, actualDestIndex);
73+
}
74+
75+
public void testGenerateDestIndexName_withDotPrefix() {
76+
String sourceIndex = ".sourceindex";
77+
String expectedDestIndex = ".migrated-sourceindex";
78+
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
79+
assertEquals(expectedDestIndex, actualDestIndex);
80+
}
81+
82+
public void testGenerateDestIndexName_withHyphen() {
83+
String sourceIndex = "source-index";
84+
String expectedDestIndex = "migrated-source-index";
85+
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
86+
assertEquals(expectedDestIndex, actualDestIndex);
87+
}
88+
89+
public void testGenerateDestIndexName_withUnderscore() {
90+
String sourceIndex = "source_index";
91+
String expectedDestIndex = "migrated-source_index";
92+
String actualDestIndex = ReindexDataStreamIndexTransportAction.generateDestIndexName(sourceIndex);
93+
assertEquals(expectedDestIndex, actualDestIndex);
94+
}
95+
96+
public void testReindexIncludesRateLimit() {
97+
var targetRateLimit = randomFloatBetween(1, 100, true);
98+
Settings settings = Settings.builder()
99+
.put(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING.getKey(), targetRateLimit)
100+
.build();
101+
102+
String sourceIndex = randomAlphanumericOfLength(10);
103+
String destIndex = randomAlphanumericOfLength(10);
104+
ActionListener<BulkByScrollResponse> listener = ActionListener.noop();
105+
TaskId taskId = TaskId.EMPTY_TASK_ID;
106+
107+
when(clusterService.getClusterSettings()).thenReturn(
108+
new ClusterSettings(
109+
settings,
110+
Collections.singleton(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING)
111+
)
112+
);
113+
114+
doNothing().when(client).execute(eq(ReindexAction.INSTANCE), request.capture(), eq(listener));
115+
116+
action.reindex(sourceIndex, destIndex, listener, taskId);
117+
118+
ReindexRequest requestValue = request.getValue();
119+
120+
assertEquals(targetRateLimit, requestValue.getRequestsPerSecond(), 0.0);
121+
}
122+
123+
public void testReindexIncludesInfiniteRateLimit() {
124+
Settings settings = Settings.builder()
125+
.put(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING.getKey(), "-1")
126+
.build();
127+
128+
String sourceIndex = randomAlphanumericOfLength(10);
129+
String destIndex = randomAlphanumericOfLength(10);
130+
ActionListener<BulkByScrollResponse> listener = ActionListener.noop();
131+
TaskId taskId = TaskId.EMPTY_TASK_ID;
132+
133+
when(clusterService.getClusterSettings()).thenReturn(
134+
new ClusterSettings(
135+
settings,
136+
Collections.singleton(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING)
137+
)
138+
);
139+
doNothing().when(client).execute(eq(ReindexAction.INSTANCE), request.capture(), eq(listener));
140+
141+
action.reindex(sourceIndex, destIndex, listener, taskId);
142+
143+
ReindexRequest requestValue = request.getValue();
144+
145+
assertEquals(Float.POSITIVE_INFINITY, requestValue.getRequestsPerSecond(), 0.0);
146+
}
147+
148+
public void testReindexZeroRateLimitThrowsError() {
149+
Settings settings = Settings.builder()
150+
.put(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING.getKey(), "0")
151+
.build();
152+
153+
String sourceIndex = randomAlphanumericOfLength(10);
154+
String destIndex = randomAlphanumericOfLength(10);
155+
ActionListener<BulkByScrollResponse> listener = ActionListener.noop();
156+
TaskId taskId = TaskId.EMPTY_TASK_ID;
157+
158+
when(clusterService.getClusterSettings()).thenReturn(
159+
new ClusterSettings(
160+
settings,
161+
Collections.singleton(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING)
162+
)
163+
);
164+
165+
IllegalArgumentException e = expectThrows(
166+
IllegalArgumentException.class,
167+
() -> action.reindex(sourceIndex, destIndex, listener, taskId)
168+
);
169+
assertEquals(
170+
"Failed to parse value [0.0] for setting [migrate.data_stream_reindex_max_request_per_second]"
171+
+ " must be greater than 0 or -1 for infinite",
172+
e.getMessage()
173+
);
174+
}
175+
176+
public void testReindexNegativeRateLimitThrowsError() {
177+
float targetRateLimit = randomFloatBetween(-100, -1, true);
178+
Settings settings = Settings.builder()
179+
.put(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING.getKey(), targetRateLimit)
180+
.build();
181+
182+
String sourceIndex = randomAlphanumericOfLength(10);
183+
String destIndex = randomAlphanumericOfLength(10);
184+
ActionListener<BulkByScrollResponse> listener = ActionListener.noop();
185+
TaskId taskId = TaskId.EMPTY_TASK_ID;
186+
187+
when(clusterService.getClusterSettings()).thenReturn(
188+
new ClusterSettings(
189+
settings,
190+
Collections.singleton(ReindexDataStreamIndexTransportAction.REINDEX_MAX_REQUESTS_PER_SECOND_SETTING)
191+
)
192+
);
193+
194+
IllegalArgumentException e = expectThrows(
195+
IllegalArgumentException.class,
196+
() -> action.reindex(sourceIndex, destIndex, listener, taskId)
197+
);
198+
assertEquals(
199+
"Failed to parse value ["
200+
+ targetRateLimit
201+
+ "] for setting [migrate.data_stream_reindex_max_request_per_second]"
202+
+ " must be greater than 0 or -1 for infinite",
203+
e.getMessage()
204+
);
205+
}
206+
}

0 commit comments

Comments
 (0)