Skip to content

Commit eb322a2

Browse files
carlosdelestcarlosdelest
authored
Merge branch 'main' into tests/mute-test-120965
2 parents 133244d + d60e6fa commit eb322a2

File tree

7 files changed

+106
-58
lines changed

7 files changed

+106
-58
lines changed

‎docs/changelog/120883.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 120883
2+
summary: Add `?master_timeout` to `POST /_ilm/migrate_to_data_tiers`
3+
area: Indices APIs
4+
type: bug
5+
issues: []

‎muted-tests.yml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -283,6 +283,9 @@ tests:
283283
- class: org.elasticsearch.datastreams.DataStreamsClientYamlTestSuiteIT
284284
method: test {p0=data_stream/80_resolve_index_data_streams/Resolve index with hidden and closed indices}
285285
issue: https://github.com/elastic/elasticsearch/issues/120965
286+
- class: org.elasticsearch.datastreams.DataStreamsClientYamlTestSuiteIT
287+
method: test {p0=data_stream/140_data_stream_aliases/Create data stream alias with filter}
288+
issue: https://github.com/elastic/elasticsearch/issues/121014
286289

287290
# Examples:
288291
#

‎rest-api-spec/src/main/resources/rest-api-spec/api/ilm.migrate_to_data_tiers.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@
2121
]
2222
},
2323
"params": {
24+
"master_timeout":{
25+
"type":"time",
26+
"description":"Explicit operation timeout for connection to master node"
27+
},
2428
"dry_run": {
2529
"type": "boolean",
2630
"description": "If set to true it will simulate the migration, providing a way to retrieve the ILM policies and indices that need to be migrated. The default is false"

‎server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/TransportGetSnapshotsAction.java‎

Lines changed: 66 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -276,55 +276,67 @@ void runOperation(ActionListener<GetSnapshotsResponse> listener) {
276276
*/
277277
private void populateResults(ActionListener<Void> listener) {
278278
try (var listeners = new RefCountingListener(listener)) {
279-
for (final RepositoryMetadata repository : repositories) {
280-
final String repositoryName = repository.name();
281-
if (skipRepository(repositoryName)) {
282-
continue;
283-
}
284279

285-
if (listeners.isFailing()) {
286-
return;
287-
}
280+
final BooleanSupplier failFastSupplier = () -> cancellableTask.isCancelled() || listeners.isFailing();
281+
282+
final Iterator<AsyncSnapshotInfoIterator> asyncSnapshotInfoIterators = Iterators.failFast(
283+
Iterators.map(
284+
Iterators.filter(
285+
Iterators.map(repositories.iterator(), RepositoryMetadata::name),
286+
repositoryName -> skipRepository(repositoryName) == false
287+
),
288+
repositoryName -> asyncRepositoryContentsListener -> SubscribableListener
289+
290+
.<RepositoryData>newForked(l -> maybeGetRepositoryData(repositoryName, l))
291+
.andThenApply(repositoryData -> {
292+
assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.MANAGEMENT);
293+
cancellableTask.ensureNotCancelled();
294+
ensureRequiredNamesPresent(repositoryName, repositoryData);
295+
return getAsyncSnapshotInfoIterator(repositoriesService.repository(repositoryName), repositoryData);
296+
})
297+
.addListener(asyncRepositoryContentsListener)
298+
),
299+
failFastSupplier
300+
);
288301

289-
maybeGetRepositoryData(repositoryName, listeners.acquire(repositoryData -> {
290-
assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.MANAGEMENT);
291-
cancellableTask.ensureNotCancelled();
292-
ensureRequiredNamesPresent(repositoryName, repositoryData);
293-
ThrottledIterator.run(
294-
Iterators.failFast(
295-
getAsyncSnapshotInfoIterator(repositoriesService.repository(repositoryName), repositoryData),
296-
() -> cancellableTask.isCancelled() || listeners.isFailing()
297-
),
298-
(ref, asyncSnapshotInfo) -> ActionListener.run(
299-
ActionListener.runBefore(listeners.acquire(), ref::close),
300-
refListener -> asyncSnapshotInfo.getSnapshotInfo(new ActionListener<>() {
301-
@Override
302-
public void onResponse(SnapshotInfo snapshotInfo) {
303-
if (matchesPredicates(snapshotInfo)) {
304-
totalCount.incrementAndGet();
305-
if (afterPredicate.test(snapshotInfo)) {
306-
allSnapshotInfos.add(snapshotInfo.maybeWithoutIndices(indices));
302+
// TODO if the request parameters allow it, modify asyncSnapshotInfoIterators to skip unnecessary GET calls here
303+
304+
asyncSnapshotInfoIterators.forEachRemaining(
305+
asyncSnapshotInfoIteratorSupplier -> asyncSnapshotInfoIteratorSupplier.getAsyncSnapshotInfoIterator(
306+
listeners.acquire(
307+
asyncSnapshotInfoIterator -> ThrottledIterator.run(
308+
Iterators.failFast(asyncSnapshotInfoIterator, failFastSupplier),
309+
(ref, asyncSnapshotInfo) -> ActionListener.run(
310+
ActionListener.runBefore(listeners.acquire(), ref::close),
311+
refListener -> asyncSnapshotInfo.getSnapshotInfo(new ActionListener<>() {
312+
@Override
313+
public void onResponse(SnapshotInfo snapshotInfo) {
314+
if (matchesPredicates(snapshotInfo)) {
315+
totalCount.incrementAndGet();
316+
if (afterPredicate.test(snapshotInfo)) {
317+
allSnapshotInfos.add(snapshotInfo.maybeWithoutIndices(indices));
318+
}
307319
}
320+
refListener.onResponse(null);
308321
}
309-
refListener.onResponse(null);
310-
}
311322

312-
@Override
313-
public void onFailure(Exception e) {
314-
if (ignoreUnavailable) {
315-
logger.warn(Strings.format("failed to fetch snapshot info for [%s]", asyncSnapshotInfo), e);
316-
refListener.onResponse(null);
317-
} else {
318-
refListener.onFailure(e);
323+
@Override
324+
public void onFailure(Exception e) {
325+
if (ignoreUnavailable) {
326+
logger.warn(Strings.format("failed to fetch snapshot info for [%s]", asyncSnapshotInfo), e);
327+
refListener.onResponse(null);
328+
} else {
329+
refListener.onFailure(e);
330+
}
319331
}
320-
}
321-
})
322-
),
323-
getSnapshotInfoExecutor.getMaxRunningTasks(),
324-
() -> {}
325-
);
326-
}));
327-
}
332+
})
333+
),
334+
getSnapshotInfoExecutor.getMaxRunningTasks(),
335+
() -> {}
336+
)
337+
)
338+
)
339+
);
328340
}
329341
}
330342

@@ -383,6 +395,17 @@ private interface AsyncSnapshotInfo {
383395
void getSnapshotInfo(ActionListener<SnapshotInfo> listener);
384396
}
385397

398+
/**
399+
* An asynchronous supplier of the collection of snapshots contained in a repository, as an iterator over snapshots each represented
400+
* as an {@link AsyncSnapshotInfo}.
401+
*/
402+
private interface AsyncSnapshotInfoIterator {
403+
/**
404+
* @param listener completed, possibly asynchronously, with the appropriate iterator over {@link AsyncSnapshotInfo} instances.
405+
*/
406+
void getAsyncSnapshotInfoIterator(ActionListener<Iterator<AsyncSnapshotInfo>> listener);
407+
}
408+
386409
/**
387410
* @return an {@link AsyncSnapshotInfo} for the given in-progress snapshot entry.
388411
*/

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/cluster/action/MigrateToDataTiersRequest.java‎

Lines changed: 15 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import org.elasticsearch.common.io.stream.StreamInput;
1212
import org.elasticsearch.common.io.stream.StreamOutput;
1313
import org.elasticsearch.core.Nullable;
14+
import org.elasticsearch.core.TimeValue;
1415
import org.elasticsearch.xcontent.ConstructingObjectParser;
1516
import org.elasticsearch.xcontent.ParseField;
1617
import org.elasticsearch.xcontent.XContentParser;
@@ -23,11 +24,14 @@ public class MigrateToDataTiersRequest extends AcknowledgedRequest<MigrateToData
2324
private static final ParseField LEGACY_TEMPLATE_TO_DELETE = new ParseField("legacy_template_to_delete");
2425
private static final ParseField NODE_ATTRIBUTE_NAME = new ParseField("node_attribute");
2526

26-
@SuppressWarnings("unchecked")
27-
public static final ConstructingObjectParser<MigrateToDataTiersRequest, Void> PARSER = new ConstructingObjectParser<>(
27+
public interface Factory {
28+
MigrateToDataTiersRequest create(@Nullable String legacyTemplateToDelete, @Nullable String nodeAttributeName);
29+
}
30+
31+
public static final ConstructingObjectParser<MigrateToDataTiersRequest, Factory> PARSER = new ConstructingObjectParser<>(
2832
"index_template",
2933
false,
30-
a -> new MigrateToDataTiersRequest((String) a[0], (String) a[1])
34+
(a, factory) -> factory.create((String) a[0], (String) a[1])
3135
);
3236

3337
static {
@@ -48,20 +52,20 @@ public class MigrateToDataTiersRequest extends AcknowledgedRequest<MigrateToData
4852
private final String legacyTemplateToDelete;
4953
private boolean dryRun = false;
5054

51-
public static MigrateToDataTiersRequest parse(XContentParser parser) throws IOException {
52-
return PARSER.parse(parser, null);
55+
public static MigrateToDataTiersRequest parse(Factory factory, XContentParser parser) throws IOException {
56+
return PARSER.parse(parser, factory);
5357
}
5458

55-
public MigrateToDataTiersRequest(@Nullable String legacyTemplateToDelete, @Nullable String nodeAttributeName) {
56-
super(TRAPPY_IMPLICIT_DEFAULT_MASTER_NODE_TIMEOUT, DEFAULT_ACK_TIMEOUT);
59+
public MigrateToDataTiersRequest(
60+
TimeValue masterNodeTimeout,
61+
@Nullable String legacyTemplateToDelete,
62+
@Nullable String nodeAttributeName
63+
) {
64+
super(masterNodeTimeout, DEFAULT_ACK_TIMEOUT);
5765
this.legacyTemplateToDelete = legacyTemplateToDelete;
5866
this.nodeAttributeName = nodeAttributeName;
5967
}
6068

61-
public MigrateToDataTiersRequest() {
62-
this(null, null);
63-
}
64-
6569
public MigrateToDataTiersRequest(StreamInput in) throws IOException {
6670
super(in);
6771
dryRun = in.readBoolean();

‎x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/cluster/action/MigrateToDataTiersRequestTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ protected Writeable.Reader<MigrateToDataTiersRequest> instanceReader() {
1919

2020
@Override
2121
protected MigrateToDataTiersRequest createTestInstance() {
22-
return new MigrateToDataTiersRequest(randomAlphaOfLength(10), randomAlphaOfLength(10));
22+
return new MigrateToDataTiersRequest(TEST_REQUEST_TIMEOUT, randomAlphaOfLength(10), randomAlphaOfLength(10));
2323
}
2424

2525
@Override

‎x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/action/RestMigrateToDataTiersAction.java‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import org.elasticsearch.client.internal.node.NodeClient;
1111
import org.elasticsearch.rest.BaseRestHandler;
1212
import org.elasticsearch.rest.RestRequest;
13+
import org.elasticsearch.rest.RestUtils;
1314
import org.elasticsearch.rest.action.RestToXContentListener;
1415
import org.elasticsearch.xpack.cluster.action.MigrateToDataTiersAction;
1516
import org.elasticsearch.xpack.cluster.action.MigrateToDataTiersRequest;
@@ -33,13 +34,21 @@ public List<Route> routes() {
3334

3435
@Override
3536
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
36-
MigrateToDataTiersRequest migrateRequest;
37+
final var masterNodeTimeout = RestUtils.getMasterNodeTimeout(request);
38+
final MigrateToDataTiersRequest migrateRequest;
3739
if (request.hasContent()) {
3840
try (var parser = request.contentParser()) {
39-
migrateRequest = MigrateToDataTiersRequest.parse(parser);
41+
migrateRequest = MigrateToDataTiersRequest.parse(
42+
(legacyTemplateToDelete, nodeAttributeName) -> new MigrateToDataTiersRequest(
43+
masterNodeTimeout,
44+
legacyTemplateToDelete,
45+
nodeAttributeName
46+
),
47+
parser
48+
);
4049
}
4150
} else {
42-
migrateRequest = new MigrateToDataTiersRequest();
51+
migrateRequest = new MigrateToDataTiersRequest(masterNodeTimeout, null, null);
4352
}
4453
migrateRequest.setDryRun(request.paramAsBoolean("dry_run", false));
4554
return channel -> client.execute(MigrateToDataTiersAction.INSTANCE, migrateRequest, new RestToXContentListener<>(channel));

0 commit comments

Comments
 (0)