Skip to content

Commit a355f19

Browse files
masseykemasseyke
authored
Reindex status api updates (#118803) (#118906)
* Reindex status api updates (#118803) * randomSet does not handle 101 elements * making sure min set size is not more than max * making sure min set size is not more than max
1 parent f1eac4c commit a355f19

File tree

11 files changed

+526
-191
lines changed

11 files changed

+526
-191
lines changed

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

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,14 +29,15 @@
2929
import org.elasticsearch.xpack.migrate.MigratePlugin;
3030
import org.elasticsearch.xpack.migrate.action.ReindexDataStreamAction.ReindexDataStreamRequest;
3131
import org.elasticsearch.xpack.migrate.action.ReindexDataStreamAction.ReindexDataStreamResponse;
32-
import org.elasticsearch.xpack.migrate.task.ReindexDataStreamStatus;
32+
import org.elasticsearch.xpack.migrate.task.ReindexDataStreamEnrichedStatus;
3333
import org.elasticsearch.xpack.migrate.task.ReindexDataStreamTask;
3434

3535
import java.util.Collection;
3636
import java.util.List;
3737
import java.util.Locale;
3838
import java.util.Map;
3939
import java.util.Optional;
40+
import java.util.Set;
4041
import java.util.concurrent.CountDownLatch;
4142
import java.util.concurrent.atomic.AtomicReference;
4243

@@ -100,20 +101,20 @@ public void testAlreadyUpToDateDataStream() throws Exception {
100101
assertThat(task.getStatus().complete(), equalTo(true));
101102
assertNull(task.getStatus().exception());
102103
assertThat(task.getStatus().pending(), equalTo(0));
103-
assertThat(task.getStatus().inProgress(), equalTo(0));
104+
assertThat(task.getStatus().inProgress(), equalTo(Set.of()));
104105
assertThat(task.getStatus().errors().size(), equalTo(0));
105106

106107
assertBusy(() -> {
107108
GetMigrationReindexStatusAction.Response statusResponse = client().execute(
108109
new ActionType<GetMigrationReindexStatusAction.Response>(GetMigrationReindexStatusAction.NAME),
109110
new GetMigrationReindexStatusAction.Request(dataStreamName)
110111
).actionGet();
111-
ReindexDataStreamStatus status = (ReindexDataStreamStatus) statusResponse.getTask().getTask().status();
112+
ReindexDataStreamEnrichedStatus status = statusResponse.getEnrichedStatus();
112113
assertThat(status.complete(), equalTo(true));
113114
assertThat(status.errors(), equalTo(List.of()));
114115
assertThat(status.exception(), equalTo(null));
115116
assertThat(status.pending(), equalTo(0));
116-
assertThat(status.inProgress(), equalTo(0));
117+
assertThat(status.inProgress().size(), equalTo(0));
117118
assertThat(status.totalIndices(), equalTo(backingIndexCount));
118119
assertThat(status.totalIndicesToBeUpgraded(), equalTo(0));
119120
});

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

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,9 @@
1616
import org.elasticsearch.common.Strings;
1717
import org.elasticsearch.common.io.stream.StreamInput;
1818
import org.elasticsearch.common.io.stream.StreamOutput;
19-
import org.elasticsearch.tasks.Task;
20-
import org.elasticsearch.tasks.TaskResult;
2119
import org.elasticsearch.xcontent.ToXContentObject;
2220
import org.elasticsearch.xcontent.XContentBuilder;
21+
import org.elasticsearch.xpack.migrate.task.ReindexDataStreamEnrichedStatus;
2322

2423
import java.io.IOException;
2524
import java.util.Objects;
@@ -36,46 +35,43 @@ public GetMigrationReindexStatusAction() {
3635
}
3736

3837
public static class Response extends ActionResponse implements ToXContentObject {
39-
private final TaskResult task;
38+
private final ReindexDataStreamEnrichedStatus enrichedStatus;
4039

41-
public Response(TaskResult task) {
42-
this.task = requireNonNull(task, "task is required");
40+
public Response(ReindexDataStreamEnrichedStatus enrichedStatus) {
41+
this.enrichedStatus = requireNonNull(enrichedStatus, "status is required");
4342
}
4443

4544
public Response(StreamInput in) throws IOException {
4645
super(in);
47-
task = in.readOptionalWriteable(TaskResult::new);
46+
enrichedStatus = in.readOptionalWriteable(ReindexDataStreamEnrichedStatus::new);
4847
}
4948

5049
@Override
5150
public void writeTo(StreamOutput out) throws IOException {
52-
out.writeOptionalWriteable(task);
51+
out.writeOptionalWriteable(enrichedStatus);
5352
}
5453

5554
/**
5655
* Get the actual result of the fetch.
5756
*/
58-
public TaskResult getTask() {
59-
return task;
57+
public ReindexDataStreamEnrichedStatus getEnrichedStatus() {
58+
return enrichedStatus;
6059
}
6160

6261
@Override
6362
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
64-
Task.Status status = task.getTask().status();
65-
if (status != null) {
66-
task.getTask().status().toXContent(builder, params);
67-
}
63+
enrichedStatus.toXContent(builder, params);
6864
return builder;
6965
}
7066

7167
@Override
7268
public int hashCode() {
73-
return Objects.hashCode(task);
69+
return Objects.hashCode(enrichedStatus);
7470
}
7571

7672
@Override
7773
public boolean equals(Object other) {
78-
return other instanceof Response && task.equals(((Response) other).task);
74+
return other instanceof Response && enrichedStatus.equals(((Response) other).enrichedStatus);
7975
}
8076

8177
@Override

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

Lines changed: 108 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -11,40 +11,55 @@
1111
import org.elasticsearch.ResourceNotFoundException;
1212
import org.elasticsearch.action.ActionListener;
1313
import org.elasticsearch.action.ActionListenerResponseHandler;
14+
import org.elasticsearch.action.admin.indices.stats.IndexStats;
15+
import org.elasticsearch.action.admin.indices.stats.IndicesStatsAction;
16+
import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
17+
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
1418
import org.elasticsearch.action.support.ActionFilters;
1519
import org.elasticsearch.action.support.HandledTransportAction;
20+
import org.elasticsearch.action.support.IndicesOptions;
21+
import org.elasticsearch.client.internal.Client;
1622
import org.elasticsearch.cluster.node.DiscoveryNode;
1723
import org.elasticsearch.cluster.service.ClusterService;
1824
import org.elasticsearch.common.util.concurrent.EsExecutors;
1925
import org.elasticsearch.core.Strings;
26+
import org.elasticsearch.core.Tuple;
27+
import org.elasticsearch.index.shard.DocsStats;
2028
import org.elasticsearch.injection.guice.Inject;
2129
import org.elasticsearch.persistent.AllocatedPersistentTask;
2230
import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
2331
import org.elasticsearch.tasks.CancellableTask;
2432
import org.elasticsearch.tasks.Task;
2533
import org.elasticsearch.tasks.TaskInfo;
26-
import org.elasticsearch.tasks.TaskResult;
2734
import org.elasticsearch.transport.TransportRequestOptions;
2835
import org.elasticsearch.transport.TransportService;
2936
import org.elasticsearch.xpack.migrate.action.GetMigrationReindexStatusAction.Request;
3037
import org.elasticsearch.xpack.migrate.action.GetMigrationReindexStatusAction.Response;
38+
import org.elasticsearch.xpack.migrate.task.ReindexDataStreamEnrichedStatus;
39+
import org.elasticsearch.xpack.migrate.task.ReindexDataStreamStatus;
3140

41+
import java.util.HashMap;
3242
import java.util.Map;
3343
import java.util.Optional;
44+
import java.util.Set;
45+
import java.util.stream.Stream;
3446

3547
public class GetMigrationReindexStatusTransportAction extends HandledTransportAction<Request, Response> {
3648
private final ClusterService clusterService;
3749
private final TransportService transportService;
50+
private final Client client;
3851

3952
@Inject
4053
public GetMigrationReindexStatusTransportAction(
4154
ClusterService clusterService,
4255
TransportService transportService,
43-
ActionFilters actionFilters
56+
ActionFilters actionFilters,
57+
Client client
4458
) {
4559
super(GetMigrationReindexStatusAction.NAME, transportService, actionFilters, Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
4660
this.clusterService = clusterService;
4761
this.transportService = transportService;
62+
this.client = client;
4863
}
4964

5065
@Override
@@ -60,9 +75,9 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
6075
} else if (persistentTask.isAssigned()) {
6176
String nodeId = persistentTask.getExecutorNode();
6277
if (clusterService.localNode().getId().equals(nodeId)) {
63-
getRunningTaskFromNode(persistentTaskId, listener);
78+
fetchAndReportStatusForTaskOnThisNode(persistentTaskId, listener);
6479
} else {
65-
runOnNodeWithTaskIfPossible(task, request, nodeId, listener);
80+
fetchAndReportStatusForTaskOnRemoteNode(task, request, nodeId, listener);
6681
}
6782
} else {
6883
listener.onFailure(new ElasticsearchException("Persistent task with id [{}] is not assigned to a node", persistentTaskId));
@@ -82,7 +97,7 @@ private Task getRunningPersistentTaskFromTaskManager(String persistentTaskId) {
8297
return optionalTask.<Task>map(Map.Entry::getValue).orElse(null);
8398
}
8499

85-
void getRunningTaskFromNode(String persistentTaskId, ActionListener<Response> listener) {
100+
void fetchAndReportStatusForTaskOnThisNode(String persistentTaskId, ActionListener<Response> listener) {
86101
Task runningTask = getRunningPersistentTaskFromTaskManager(persistentTaskId);
87102
if (runningTask == null) {
88103
listener.onFailure(
@@ -96,11 +111,97 @@ void getRunningTaskFromNode(String persistentTaskId, ActionListener<Response> li
96111
);
97112
} else {
98113
TaskInfo info = runningTask.taskInfo(clusterService.localNode().getId(), true);
99-
listener.onResponse(new Response(new TaskResult(false, info)));
114+
ReindexDataStreamStatus status = (ReindexDataStreamStatus) info.status();
115+
Set<String> inProgressIndices = status.inProgress();
116+
if (inProgressIndices.isEmpty()) {
117+
// We have no reason to fetch index stats since there are no in progress indices
118+
reportStatus(Map.of(), status, listener);
119+
} else {
120+
fetchInProgressStatsAndReportStatus(inProgressIndices, status, listener);
121+
}
100122
}
101123
}
102124

103-
private void runOnNodeWithTaskIfPossible(Task thisTask, Request request, String nodeId, ActionListener<Response> listener) {
125+
/*
126+
* The status is enriched with the information from inProgressMap to create a new ReindexDataStreamEnrichedStatus, which is used in the
127+
* response sent to the listener.
128+
*/
129+
private void reportStatus(
130+
Map<String, Tuple<Long, Long>> inProgressMap,
131+
ReindexDataStreamStatus status,
132+
ActionListener<Response> listener
133+
) {
134+
ReindexDataStreamEnrichedStatus enrichedStatus = new ReindexDataStreamEnrichedStatus(
135+
status.persistentTaskStartTime(),
136+
status.totalIndices(),
137+
status.totalIndicesToBeUpgraded(),
138+
status.complete(),
139+
status.exception(),
140+
inProgressMap,
141+
status.pending(),
142+
status.errors()
143+
);
144+
listener.onResponse(new Response(enrichedStatus));
145+
}
146+
147+
/*
148+
* This method feches doc counts for all indices in inProgressIndices (and the indices they are being reindexed into). After
149+
* successfully fetching those, reportStatus is called.
150+
*/
151+
private void fetchInProgressStatsAndReportStatus(
152+
Set<String> inProgressIndices,
153+
ReindexDataStreamStatus status,
154+
ActionListener<Response> listener
155+
) {
156+
IndicesStatsRequest indicesStatsRequest = new IndicesStatsRequest();
157+
String[] indices = inProgressIndices.stream()
158+
.flatMap(index -> Stream.of(index, ReindexDataStreamIndexTransportAction.generateDestIndexName(index)))
159+
.toList()
160+
.toArray(new String[0]);
161+
indicesStatsRequest.indices(indices);
162+
/*
163+
* It is possible that the destination index will not exist yet, so we want to ignore the fact that it is missing
164+
*/
165+
indicesStatsRequest.indicesOptions(IndicesOptions.fromOptions(true, true, true, true));
166+
client.execute(IndicesStatsAction.INSTANCE, indicesStatsRequest, new ActionListener<IndicesStatsResponse>() {
167+
@Override
168+
public void onResponse(IndicesStatsResponse indicesStatsResponse) {
169+
Map<String, Tuple<Long, Long>> inProgressMap = new HashMap<>();
170+
for (String index : inProgressIndices) {
171+
IndexStats sourceIndexStats = indicesStatsResponse.getIndex(index);
172+
final long totalDocsInIndex;
173+
if (sourceIndexStats == null) {
174+
totalDocsInIndex = 0;
175+
} else {
176+
DocsStats totalDocsStats = sourceIndexStats.getTotal().getDocs();
177+
totalDocsInIndex = totalDocsStats == null ? 0 : totalDocsStats.getCount();
178+
}
179+
IndexStats migratedIndexStats = indicesStatsResponse.getIndex(
180+
ReindexDataStreamIndexTransportAction.generateDestIndexName(index)
181+
);
182+
final long reindexedDocsInIndex;
183+
if (migratedIndexStats == null) {
184+
reindexedDocsInIndex = 0;
185+
} else {
186+
DocsStats reindexedDocsStats = migratedIndexStats.getTotal().getDocs();
187+
reindexedDocsInIndex = reindexedDocsStats == null ? 0 : reindexedDocsStats.getCount();
188+
}
189+
inProgressMap.put(index, Tuple.tuple(totalDocsInIndex, reindexedDocsInIndex));
190+
}
191+
reportStatus(inProgressMap, status, listener);
192+
}
193+
194+
@Override
195+
public void onFailure(Exception e) {
196+
listener.onFailure(e);
197+
}
198+
});
199+
}
200+
201+
/*
202+
* The task and its status exist on some other node, so this method forwards the request to that node.
203+
*/
204+
private void fetchAndReportStatusForTaskOnRemoteNode(Task thisTask, Request request, String nodeId, ActionListener<Response> listener) {
104205
DiscoveryNode node = clusterService.state().nodes().get(nodeId);
105206
if (node == null) {
106207
listener.onFailure(

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

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
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.task;
9+
10+
import org.elasticsearch.common.io.stream.StreamInput;
11+
import org.elasticsearch.common.io.stream.StreamOutput;
12+
import org.elasticsearch.common.io.stream.Writeable;
13+
import org.elasticsearch.core.Tuple;
14+
import org.elasticsearch.xcontent.ToXContentObject;
15+
import org.elasticsearch.xcontent.XContentBuilder;
16+
17+
import java.io.IOException;
18+
import java.util.List;
19+
import java.util.Map;
20+
import java.util.Objects;
21+
22+
/*
23+
* This class represents information similar to that in ReindexDataStreamStatus, but enriched from other sources besides just the task
24+
* itself.
25+
*/
26+
public record ReindexDataStreamEnrichedStatus(
27+
long persistentTaskStartTime,
28+
int totalIndices,
29+
int totalIndicesToBeUpgraded,
30+
boolean complete,
31+
Exception exception,
32+
Map<String, Tuple<Long, Long>> inProgress,
33+
int pending,
34+
List<Tuple<String, Exception>> errors
35+
) implements ToXContentObject, Writeable {
36+
public ReindexDataStreamEnrichedStatus {
37+
Objects.requireNonNull(inProgress);
38+
Objects.requireNonNull(errors);
39+
}
40+
41+
public ReindexDataStreamEnrichedStatus(StreamInput in) throws IOException {
42+
this(
43+
in.readLong(),
44+
in.readInt(),
45+
in.readInt(),
46+
in.readBoolean(),
47+
in.readException(),
48+
in.readMap(StreamInput::readString, in2 -> Tuple.tuple(in2.readLong(), in2.readLong())),
49+
in.readInt(),
50+
in.readCollectionAsList(in1 -> Tuple.tuple(in1.readString(), in1.readException()))
51+
);
52+
}
53+
54+
@Override
55+
public void writeTo(StreamOutput out) throws IOException {
56+
out.writeLong(persistentTaskStartTime);
57+
out.writeInt(totalIndices);
58+
out.writeInt(totalIndicesToBeUpgraded);
59+
out.writeBoolean(complete);
60+
out.writeException(exception);
61+
out.writeMap(inProgress, StreamOutput::writeString, (out2, tuple) -> {
62+
out2.writeLong(tuple.v1());
63+
out2.writeLong(tuple.v2());
64+
});
65+
out.writeInt(pending);
66+
out.writeCollection(errors, (out1, tuple) -> {
67+
out1.writeString(tuple.v1());
68+
out1.writeException(tuple.v2());
69+
});
70+
}
71+
72+
@Override
73+
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
74+
builder.startObject();
75+
builder.timestampFieldsFromUnixEpochMillis("start_time_millis", "start_time", persistentTaskStartTime);
76+
builder.field("complete", complete);
77+
builder.field("total_indices_in_data_stream", totalIndices);
78+
builder.field("total_indices_requiring_upgrade", totalIndicesToBeUpgraded);
79+
builder.field("successes", totalIndicesToBeUpgraded - (inProgress.size() + pending + errors.size()));
80+
builder.startArray("in_progress");
81+
for (Map.Entry<String, Tuple<Long, Long>> inProgressEntry : inProgress.entrySet()) {
82+
builder.startObject();
83+
builder.field("index", inProgressEntry.getKey());
84+
builder.field("total_doc_count", inProgressEntry.getValue().v1());
85+
builder.field("reindexed_doc_count", inProgressEntry.getValue().v2());
86+
builder.endObject();
87+
}
88+
builder.endArray();
89+
builder.field("pending", pending);
90+
builder.startArray("errors");
91+
for (Tuple<String, Exception> error : errors) {
92+
builder.startObject();
93+
builder.field("index", error.v1());
94+
builder.field("message", error.v2() == null ? "unknown" : error.v2().getMessage());
95+
builder.endObject();
96+
}
97+
builder.endArray();
98+
if (exception != null) {
99+
builder.field("exception", exception.getMessage());
100+
}
101+
builder.endObject();
102+
return builder;
103+
}
104+
}

0 commit comments

Comments
 (0)