Skip to content

Commit aa203af

Browse files
masseykeelasticsearchmachine
masseyke
and
elasticsearchmachine
authored
Connecting the reindex data stream persistent task to ReindexDataStreamIndexAction (#118978) (#119032)
* Connecting the reindex data stream persistent task to ReindexDataStreamIndexAction (#118978) * removing java 21 syntax * [CI] Auto commit changes from spotless * removing java 21 syntax * removing code that doesnt work in 7.17 * expecting no upgrades in clusters of the same major versoin * correcting method name after merge --------- Co-authored-by: elasticsearchmachine <[email protected]>
1 parent 3c0da86 commit aa203af

File tree

2 files changed

+269
-12
lines changed

2 files changed

+269
-12
lines changed

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

Lines changed: 112 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,15 @@
99

1010
import org.elasticsearch.ElasticsearchException;
1111
import org.elasticsearch.action.ActionListener;
12+
import org.elasticsearch.action.admin.indices.rollover.RolloverAction;
13+
import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
1214
import org.elasticsearch.action.datastreams.GetDataStreamAction;
15+
import org.elasticsearch.action.datastreams.ModifyDataStreamsAction;
16+
import org.elasticsearch.action.support.CountDownActionListener;
17+
import org.elasticsearch.action.support.master.AcknowledgedResponse;
1318
import org.elasticsearch.client.internal.Client;
19+
import org.elasticsearch.cluster.metadata.DataStream;
20+
import org.elasticsearch.cluster.metadata.DataStreamAction;
1421
import org.elasticsearch.cluster.service.ClusterService;
1522
import org.elasticsearch.core.TimeValue;
1623
import org.elasticsearch.index.Index;
@@ -20,7 +27,10 @@
2027
import org.elasticsearch.persistent.PersistentTasksExecutor;
2128
import org.elasticsearch.tasks.TaskId;
2229
import org.elasticsearch.threadpool.ThreadPool;
30+
import org.elasticsearch.xpack.migrate.action.ReindexDataStreamIndexAction;
2331

32+
import java.util.ArrayList;
33+
import java.util.Collections;
2434
import java.util.List;
2535
import java.util.Map;
2636

@@ -72,22 +82,109 @@ protected void nodeOperation(AllocatedPersistentTask task, ReindexDataStreamTask
7282
reindexClient.execute(GetDataStreamAction.INSTANCE, request, ActionListener.wrap(response -> {
7383
List<GetDataStreamAction.Response.DataStreamInfo> dataStreamInfos = response.getDataStreams();
7484
if (dataStreamInfos.size() == 1) {
75-
List<Index> indices = dataStreamInfos.get(0).getDataStream().getIndices();
76-
List<Index> indicesToBeReindexed = indices.stream()
77-
.filter(getReindexRequiredPredicate(clusterService.state().metadata()))
78-
.toList();
79-
reindexDataStreamTask.setPendingIndicesCount(indicesToBeReindexed.size());
80-
for (Index index : indicesToBeReindexed) {
81-
reindexDataStreamTask.incrementInProgressIndicesCount(index.getName());
82-
// TODO This is just a placeholder. This is where the real data stream reindex logic will go
83-
reindexDataStreamTask.reindexSucceeded(index.getName());
85+
DataStream dataStream = dataStreamInfos.get(0).getDataStream();
86+
if (getReindexRequiredPredicate(clusterService.state().metadata()).test(dataStream.getWriteIndex())) {
87+
reindexClient.execute(
88+
RolloverAction.INSTANCE,
89+
new RolloverRequest(sourceDataStream, null),
90+
ActionListener.wrap(
91+
rolloverResponse -> reindexIndices(dataStream, reindexDataStreamTask, reindexClient, sourceDataStream),
92+
e -> completeFailedPersistentTask(reindexDataStreamTask, e)
93+
)
94+
);
95+
} else {
96+
reindexIndices(dataStream, reindexDataStreamTask, reindexClient, sourceDataStream);
8497
}
85-
86-
completeSuccessfulPersistentTask(reindexDataStreamTask);
8798
} else {
8899
completeFailedPersistentTask(reindexDataStreamTask, new ElasticsearchException("data stream does not exist"));
89100
}
90-
}, reindexDataStreamTask::markAsFailed));
101+
}, exception -> completeFailedPersistentTask(reindexDataStreamTask, exception)));
102+
}
103+
104+
private void reindexIndices(
105+
DataStream dataStream,
106+
ReindexDataStreamTask reindexDataStreamTask,
107+
ExecuteWithHeadersClient reindexClient,
108+
String sourceDataStream
109+
) {
110+
List<Index> indices = dataStream.getIndices();
111+
List<Index> indicesToBeReindexed = indices.stream().filter(getReindexRequiredPredicate(clusterService.state().metadata())).toList();
112+
reindexDataStreamTask.setPendingIndicesCount(indicesToBeReindexed.size());
113+
// The CountDownActionListener is 1 more than the number of indices so that the count is not 0 if we have no indices
114+
CountDownActionListener listener = new CountDownActionListener(indicesToBeReindexed.size() + 1, ActionListener.wrap(response1 -> {
115+
completeSuccessfulPersistentTask(reindexDataStreamTask);
116+
}, exception -> { completeFailedPersistentTask(reindexDataStreamTask, exception); }));
117+
List<Index> indicesRemaining = Collections.synchronizedList(new ArrayList<>(indicesToBeReindexed));
118+
final int maxConcurrentIndices = 1;
119+
for (int i = 0; i < maxConcurrentIndices; i++) {
120+
maybeProcessNextIndex(indicesRemaining, reindexDataStreamTask, reindexClient, sourceDataStream, listener);
121+
}
122+
// This takes care of the additional latch count referenced above:
123+
listener.onResponse(null);
124+
}
125+
126+
private void maybeProcessNextIndex(
127+
List<Index> indicesRemaining,
128+
ReindexDataStreamTask reindexDataStreamTask,
129+
ExecuteWithHeadersClient reindexClient,
130+
String sourceDataStream,
131+
CountDownActionListener listener
132+
) {
133+
if (indicesRemaining.isEmpty()) {
134+
return;
135+
}
136+
Index index;
137+
try {
138+
index = indicesRemaining.remove(0);
139+
} catch (IndexOutOfBoundsException e) {
140+
return;
141+
}
142+
reindexDataStreamTask.incrementInProgressIndicesCount(index.getName());
143+
reindexClient.execute(
144+
ReindexDataStreamIndexAction.INSTANCE,
145+
new ReindexDataStreamIndexAction.Request(index.getName()),
146+
ActionListener.wrap(response1 -> {
147+
updateDataStream(sourceDataStream, index.getName(), response1.getDestIndex(), ActionListener.wrap(unused -> {
148+
reindexDataStreamTask.reindexSucceeded(index.getName());
149+
listener.onResponse(null);
150+
maybeProcessNextIndex(indicesRemaining, reindexDataStreamTask, reindexClient, sourceDataStream, listener);
151+
}, exception -> {
152+
reindexDataStreamTask.reindexFailed(index.getName(), exception);
153+
listener.onResponse(null);
154+
}), reindexClient);
155+
}, exception -> {
156+
reindexDataStreamTask.reindexFailed(index.getName(), exception);
157+
listener.onResponse(null);
158+
})
159+
);
160+
}
161+
162+
private void updateDataStream(
163+
String dataStream,
164+
String oldIndex,
165+
String newIndex,
166+
ActionListener<Void> listener,
167+
ExecuteWithHeadersClient reindexClient
168+
) {
169+
reindexClient.execute(
170+
ModifyDataStreamsAction.INSTANCE,
171+
new ModifyDataStreamsAction.Request(
172+
TimeValue.MAX_VALUE,
173+
TimeValue.MAX_VALUE,
174+
List.of(DataStreamAction.removeBackingIndex(dataStream, oldIndex), DataStreamAction.addBackingIndex(dataStream, newIndex))
175+
),
176+
new ActionListener<>() {
177+
@Override
178+
public void onResponse(AcknowledgedResponse response) {
179+
listener.onResponse(null);
180+
}
181+
182+
@Override
183+
public void onFailure(Exception e) {
184+
listener.onFailure(e);
185+
}
186+
}
187+
);
91188
}
92189

93190
private void completeSuccessfulPersistentTask(ReindexDataStreamTask persistentTask) {
@@ -105,6 +202,9 @@ private TimeValue getTimeToLive(ReindexDataStreamTask reindexDataStreamTask) {
105202
PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = persistentTasksCustomMetadata.getTask(
106203
reindexDataStreamTask.getPersistentTaskId()
107204
);
205+
if (persistentTask == null) {
206+
return TimeValue.timeValueMillis(0);
207+
}
108208
PersistentTaskState state = persistentTask.getState();
109209
final long completionTime;
110210
if (state == null) {

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

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,19 +7,29 @@
77
package org.elasticsearch.upgrades;
88

99
import org.apache.http.util.EntityUtils;
10+
import org.elasticsearch.Build;
11+
import org.elasticsearch.Version;
1012
import org.elasticsearch.client.Request;
1113
import org.elasticsearch.client.Response;
1214
import org.elasticsearch.cluster.metadata.DataStream;
1315
import org.elasticsearch.cluster.metadata.DataStreamTestHelper;
16+
import org.elasticsearch.common.time.DateFormatter;
17+
import org.elasticsearch.common.time.FormatNames;
18+
import org.elasticsearch.common.xcontent.XContentHelper;
1419
import org.elasticsearch.core.Booleans;
1520
import org.elasticsearch.core.Strings;
21+
import org.elasticsearch.xcontent.json.JsonXContent;
1622
import org.hamcrest.Matchers;
1723

1824
import java.io.IOException;
1925
import java.nio.charset.StandardCharsets;
26+
import java.time.Instant;
2027
import java.util.List;
28+
import java.util.Map;
29+
import java.util.concurrent.TimeUnit;
2130

2231
import static org.elasticsearch.upgrades.IndexingIT.assertCount;
32+
import static org.hamcrest.Matchers.equalTo;
2333

2434
public class DataStreamsUpgradeIT extends AbstractUpgradeTestCase {
2535

@@ -164,4 +174,151 @@ public void testDataStreamValidationDoesNotBreakUpgrade() throws Exception {
164174
}
165175
}
166176

177+
public void testUpgradeDataStream() throws Exception {
178+
String dataStreamName = "reindex_test_data_stream";
179+
int numRollovers = 5;
180+
if (CLUSTER_TYPE == ClusterType.OLD) {
181+
createAndRolloverDataStream(dataStreamName, numRollovers);
182+
} else if (CLUSTER_TYPE == ClusterType.UPGRADED) {
183+
upgradeDataStream(dataStreamName, numRollovers);
184+
}
185+
}
186+
187+
private static void createAndRolloverDataStream(String dataStreamName, int numRollovers) throws IOException {
188+
// We want to create a data stream and roll it over several times so that we have several indices to upgrade
189+
final String template = """
190+
{
191+
"mappings":{
192+
"dynamic_templates": [
193+
{
194+
"labels": {
195+
"path_match": "pod.labels.*",
196+
"mapping": {
197+
"type": "keyword",
198+
"time_series_dimension": true
199+
}
200+
}
201+
}
202+
],
203+
"properties": {
204+
"@timestamp" : {
205+
"type": "date"
206+
},
207+
"metricset": {
208+
"type": "keyword",
209+
"time_series_dimension": true
210+
},
211+
"k8s": {
212+
"properties": {
213+
"pod": {
214+
"properties": {
215+
"name": {
216+
"type": "keyword"
217+
},
218+
"network": {
219+
"properties": {
220+
"tx": {
221+
"type": "long"
222+
},
223+
"rx": {
224+
"type": "long"
225+
}
226+
}
227+
}
228+
}
229+
}
230+
}
231+
}
232+
}
233+
}
234+
}
235+
""";
236+
final String indexTemplate = """
237+
{
238+
"index_patterns": ["$PATTERN"],
239+
"template": $TEMPLATE,
240+
"data_stream": {
241+
}
242+
}""";
243+
var putIndexTemplateRequest = new Request("POST", "/_index_template/reindex_test_data_stream_template");
244+
putIndexTemplateRequest.setJsonEntity(indexTemplate.replace("$TEMPLATE", template).replace("$PATTERN", dataStreamName));
245+
assertOK(client().performRequest(putIndexTemplateRequest));
246+
bulkLoadData(dataStreamName);
247+
for (int i = 0; i < numRollovers; i++) {
248+
rollover(dataStreamName);
249+
bulkLoadData(dataStreamName);
250+
}
251+
}
252+
253+
private void upgradeDataStream(String dataStreamName, int numRollovers) throws Exception {
254+
Request reindexRequest = new Request("POST", "/_migration/reindex");
255+
reindexRequest.setJsonEntity(Strings.format("""
256+
{
257+
"mode": "upgrade",
258+
"source": {
259+
"index": "%s"
260+
}
261+
}""", dataStreamName));
262+
Response reindexResponse = client().performRequest(reindexRequest);
263+
assertOK(reindexResponse);
264+
assertBusy(() -> {
265+
Request statusRequest = new Request("GET", "_migration/reindex/" + dataStreamName + "/_status");
266+
Response statusResponse = client().performRequest(statusRequest);
267+
Map<String, Object> statusResponseMap = XContentHelper.convertToMap(
268+
JsonXContent.jsonXContent,
269+
statusResponse.getEntity().getContent(),
270+
false
271+
);
272+
assertOK(statusResponse);
273+
assertThat(statusResponseMap.get("complete"), equalTo(true));
274+
if (isOriginalClusterSameMajorVersionAsCurrent()) {
275+
// If the original cluster was the same as this one, we don't want any indices reindexed:
276+
assertThat(statusResponseMap.get("successes"), equalTo(0));
277+
} else {
278+
assertThat(statusResponseMap.get("successes"), equalTo(numRollovers + 1));
279+
}
280+
}, 60, TimeUnit.SECONDS);
281+
Request cancelRequest = new Request("POST", "_migration/reindex/" + dataStreamName + "/_cancel");
282+
Response cancelResponse = client().performRequest(cancelRequest);
283+
assertOK(cancelResponse);
284+
}
285+
286+
private boolean isOriginalClusterSameMajorVersionAsCurrent() {
287+
return Version.fromString(UPGRADE_FROM_VERSION).major == Version.fromString(Build.current().version()).major;
288+
}
289+
290+
private static void bulkLoadData(String dataStreamName) throws IOException {
291+
final String bulk = """
292+
{"create": {}}
293+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "cat", "network": {"tx": 2001818691, "rx": 802133794}}}}
294+
{"create": {}}
295+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "hamster", "network": {"tx": 2005177954, "rx": 801479970}}}}
296+
{"create": {}}
297+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "cow", "network": {"tx": 2006223737, "rx": 802337279}}}}
298+
{"create": {}}
299+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "rat", "network": {"tx": 2012916202, "rx": 803685721}}}}
300+
{"create": {}}
301+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "dog", "network": {"tx": 1434521831, "rx": 530575198}}}}
302+
{"create": {}}
303+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "tiger", "network": {"tx": 1434577921, "rx": 530600088}}}}
304+
{"create": {}}
305+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "lion", "network": {"tx": 1434587694, "rx": 530604797}}}}
306+
{"create": {}}
307+
{"@timestamp": "$now", "metricset": "pod", "k8s": {"pod": {"name": "elephant", "network": {"tx": 1434595272, "rx": 530605511}}}}
308+
""";
309+
var bulkRequest = new Request("POST", "/" + dataStreamName + "/_bulk");
310+
bulkRequest.setJsonEntity(bulk.replace("$now", formatInstant(Instant.now())));
311+
var response = client().performRequest(bulkRequest);
312+
assertOK(response);
313+
}
314+
315+
static String formatInstant(Instant instant) {
316+
return DateFormatter.forPattern(FormatNames.STRICT_DATE_OPTIONAL_TIME.getName()).format(instant);
317+
}
318+
319+
private static void rollover(String dataStreamName) throws IOException {
320+
Request rolloverRequest = new Request("POST", "/" + dataStreamName + "/_rollover");
321+
Response rolloverResponse = client().performRequest(rolloverRequest);
322+
assertOK(rolloverResponse);
323+
}
167324
}

0 commit comments

Comments
 (0)