Skip to content

Commit 4c8ae24

Browse files
edsavageedsavage
authored
Merge branch 'main' into disable_ml_on_macos_x86_64
2 parents e965efd + 4e04a7b commit 4c8ae24

File tree

60 files changed

+3539
-524
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

60 files changed

+3539
-524
lines changed

‎benchmarks/src/main/java/org/elasticsearch/benchmark/script/ScriptScoreBenchmark.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import org.elasticsearch.index.mapper.NumberFieldMapper.NumberType;
3535
import org.elasticsearch.indices.breaker.CircuitBreakerService;
3636
import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
37+
import org.elasticsearch.plugins.PluginsLoader;
3738
import org.elasticsearch.plugins.PluginsService;
3839
import org.elasticsearch.plugins.ScriptPlugin;
3940
import org.elasticsearch.script.DocReader;
@@ -76,8 +77,7 @@ public class ScriptScoreBenchmark {
7677
private final PluginsService pluginsService = new PluginsService(
7778
Settings.EMPTY,
7879
null,
79-
null,
80-
Path.of(System.getProperty("plugins.dir"))
80+
new PluginsLoader(null, Path.of(System.getProperty("plugins.dir")))
8181
);
8282
private final ScriptModule scriptModule = new ScriptModule(Settings.EMPTY, pluginsService.filterPlugins(ScriptPlugin.class).toList());
8383

‎branches.json‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@
77
{
88
"branch": "8.16"
99
},
10+
{
11+
"branch": "8.17"
12+
},
1013
{
1114
"branch": "8.x"
1215
},

‎docs/changelog/116765.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 116765
2+
summary: Metrics for incremental bulk splits
3+
area: Distributed
4+
type: enhancement
5+
issues: []

‎docs/changelog/117182.yaml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
pr: 117182
2+
summary: Change synthetic source logic for `constant_keyword`
3+
area: Mapping
4+
type: bug
5+
issues:
6+
- 117083

‎modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/action/ReindexDataStreamTransportActionIT.java‎

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
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", the "GNU Affero General Public License v3.0 only", and the "Server Side
5+
* Public License v 1"; you may not use this file except in compliance with, at
6+
* your election, the "Elastic License 2.0", the "GNU Affero General Public
7+
* License v3.0 only", or the "Server Side Public License, v 1".
8+
*/
9+
10+
package org.elasticsearch.datastreams.action;
11+
12+
import org.elasticsearch.ResourceNotFoundException;
13+
import org.elasticsearch.action.ActionListener;
14+
import org.elasticsearch.action.ActionType;
15+
import org.elasticsearch.action.DocWriteRequest;
16+
import org.elasticsearch.action.DocWriteResponse;
17+
import org.elasticsearch.action.admin.indices.rollover.RolloverRequestBuilder;
18+
import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
19+
import org.elasticsearch.action.datastreams.CreateDataStreamAction;
20+
import org.elasticsearch.action.datastreams.ReindexDataStreamAction;
21+
import org.elasticsearch.action.datastreams.ReindexDataStreamAction.ReindexDataStreamRequest;
22+
import org.elasticsearch.action.datastreams.ReindexDataStreamAction.ReindexDataStreamResponse;
23+
import org.elasticsearch.action.index.IndexRequest;
24+
import org.elasticsearch.action.support.master.AcknowledgedResponse;
25+
import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
26+
import org.elasticsearch.cluster.metadata.Template;
27+
import org.elasticsearch.datastreams.DataStreamsPlugin;
28+
import org.elasticsearch.datastreams.task.ReindexDataStreamTask;
29+
import org.elasticsearch.plugins.Plugin;
30+
import org.elasticsearch.tasks.CancellableTask;
31+
import org.elasticsearch.tasks.TaskManager;
32+
import org.elasticsearch.test.ESIntegTestCase;
33+
import org.elasticsearch.transport.TransportService;
34+
import org.elasticsearch.xcontent.XContentType;
35+
36+
import java.util.Collection;
37+
import java.util.List;
38+
import java.util.Locale;
39+
import java.util.Map;
40+
import java.util.Optional;
41+
import java.util.concurrent.CountDownLatch;
42+
import java.util.concurrent.atomic.AtomicReference;
43+
44+
import static org.hamcrest.Matchers.equalTo;
45+
import static org.hamcrest.Matchers.is;
46+
47+
public class ReindexDataStreamTransportActionIT extends ESIntegTestCase {
48+
49+
@Override
50+
protected Collection<Class<? extends Plugin>> nodePlugins() {
51+
return List.of(DataStreamsPlugin.class);
52+
}
53+
54+
public void testNonExistentDataStream() {
55+
String nonExistentDataStreamName = randomAlphaOfLength(50);
56+
ReindexDataStreamRequest reindexDataStreamRequest = new ReindexDataStreamRequest(nonExistentDataStreamName);
57+
assertThrows(
58+
ResourceNotFoundException.class,
59+
() -> client().execute(new ActionType<ReindexDataStreamResponse>(ReindexDataStreamAction.NAME), reindexDataStreamRequest)
60+
.actionGet()
61+
);
62+
}
63+
64+
public void testAlreadyUpToDateDataStream() throws Exception {
65+
String dataStreamName = randomAlphaOfLength(50).toLowerCase(Locale.ROOT);
66+
ReindexDataStreamRequest reindexDataStreamRequest = new ReindexDataStreamRequest(dataStreamName);
67+
createDataStream(dataStreamName);
68+
ReindexDataStreamResponse response = client().execute(
69+
new ActionType<ReindexDataStreamResponse>(ReindexDataStreamAction.NAME),
70+
reindexDataStreamRequest
71+
).actionGet();
72+
String persistentTaskId = response.getTaskId();
73+
assertThat(persistentTaskId, equalTo("reindex-data-stream-" + dataStreamName));
74+
AtomicReference<ReindexDataStreamTask> runningTask = new AtomicReference<>();
75+
for (TransportService transportService : internalCluster().getInstances(TransportService.class)) {
76+
TaskManager taskManager = transportService.getTaskManager();
77+
Map<Long, CancellableTask> tasksMap = taskManager.getCancellableTasks();
78+
Optional<Map.Entry<Long, CancellableTask>> optionalTask = taskManager.getCancellableTasks()
79+
.entrySet()
80+
.stream()
81+
.filter(entry -> entry.getValue().getType().equals("persistent"))
82+
.filter(
83+
entry -> entry.getValue() instanceof ReindexDataStreamTask
84+
&& persistentTaskId.equals((((ReindexDataStreamTask) entry.getValue()).getPersistentTaskId()))
85+
)
86+
.findAny();
87+
optionalTask.ifPresent(
88+
longCancellableTaskEntry -> runningTask.compareAndSet(null, (ReindexDataStreamTask) longCancellableTaskEntry.getValue())
89+
);
90+
}
91+
ReindexDataStreamTask task = runningTask.get();
92+
assertNotNull(task);
93+
assertThat(task.getStatus().complete(), equalTo(true));
94+
assertNull(task.getStatus().exception());
95+
assertThat(task.getStatus().pending(), equalTo(0));
96+
assertThat(task.getStatus().inProgress(), equalTo(0));
97+
assertThat(task.getStatus().errors().size(), equalTo(0));
98+
}
99+
100+
private void createDataStream(String dataStreamName) {
101+
final TransportPutComposableIndexTemplateAction.Request putComposableTemplateRequest =
102+
new TransportPutComposableIndexTemplateAction.Request("my-template");
103+
putComposableTemplateRequest.indexTemplate(
104+
ComposableIndexTemplate.builder()
105+
.indexPatterns(List.of(dataStreamName))
106+
.dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
107+
.template(Template.builder().build())
108+
.build()
109+
);
110+
final AcknowledgedResponse putComposableTemplateResponse = safeGet(
111+
client().execute(TransportPutComposableIndexTemplateAction.TYPE, putComposableTemplateRequest)
112+
);
113+
assertThat(putComposableTemplateResponse.isAcknowledged(), is(true));
114+
115+
final CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(
116+
TEST_REQUEST_TIMEOUT,
117+
TEST_REQUEST_TIMEOUT,
118+
dataStreamName
119+
);
120+
final AcknowledgedResponse createDataStreamResponse = safeGet(
121+
client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest)
122+
);
123+
assertThat(createDataStreamResponse.isAcknowledged(), is(true));
124+
indexDocs(dataStreamName);
125+
safeGet(new RolloverRequestBuilder(client()).setRolloverTarget(dataStreamName).lazy(false).execute());
126+
indexDocs(dataStreamName);
127+
safeGet(new RolloverRequestBuilder(client()).setRolloverTarget(dataStreamName).lazy(false).execute());
128+
}
129+
130+
private void indexDocs(String dataStreamName) {
131+
int docs = randomIntBetween(5, 10);
132+
CountDownLatch countDownLatch = new CountDownLatch(docs);
133+
for (int i = 0; i < docs; i++) {
134+
var indexRequest = new IndexRequest(dataStreamName).opType(DocWriteRequest.OpType.CREATE);
135+
final String doc = "{ \"@timestamp\": \"2099-05-06T16:21:15.000Z\", \"message\": \"something cool happened\" }";
136+
indexRequest.source(doc, XContentType.JSON);
137+
client().index(indexRequest, new ActionListener<>() {
138+
@Override
139+
public void onResponse(DocWriteResponse docWriteResponse) {
140+
countDownLatch.countDown();
141+
}
142+
143+
@Override
144+
public void onFailure(Exception e) {
145+
fail("Indexing request should have succeeded eventually, failed with " + e.getMessage());
146+
}
147+
});
148+
}
149+
safeAwait(countDownLatch);
150+
}
151+
152+
}

‎modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java‎

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,19 +19,23 @@
1919
import org.elasticsearch.action.datastreams.MigrateToDataStreamAction;
2020
import org.elasticsearch.action.datastreams.ModifyDataStreamsAction;
2121
import org.elasticsearch.action.datastreams.PromoteDataStreamAction;
22+
import org.elasticsearch.action.datastreams.ReindexDataStreamAction;
2223
import org.elasticsearch.action.datastreams.lifecycle.ExplainDataStreamLifecycleAction;
2324
import org.elasticsearch.action.datastreams.lifecycle.GetDataStreamLifecycleAction;
2425
import org.elasticsearch.action.datastreams.lifecycle.PutDataStreamLifecycleAction;
26+
import org.elasticsearch.client.internal.Client;
2527
import org.elasticsearch.client.internal.OriginSettingClient;
2628
import org.elasticsearch.cluster.metadata.DataStream;
2729
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
2830
import org.elasticsearch.cluster.node.DiscoveryNodes;
31+
import org.elasticsearch.cluster.service.ClusterService;
2932
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
3033
import org.elasticsearch.common.settings.ClusterSettings;
3134
import org.elasticsearch.common.settings.IndexScopedSettings;
3235
import org.elasticsearch.common.settings.Setting;
3336
import org.elasticsearch.common.settings.Settings;
3437
import org.elasticsearch.common.settings.SettingsFilter;
38+
import org.elasticsearch.common.settings.SettingsModule;
3539
import org.elasticsearch.core.IOUtils;
3640
import org.elasticsearch.core.TimeValue;
3741
import org.elasticsearch.datastreams.action.CreateDataStreamTransportAction;
@@ -40,6 +44,7 @@
4044
import org.elasticsearch.datastreams.action.MigrateToDataStreamTransportAction;
4145
import org.elasticsearch.datastreams.action.ModifyDataStreamsTransportAction;
4246
import org.elasticsearch.datastreams.action.PromoteDataStreamTransportAction;
47+
import org.elasticsearch.datastreams.action.ReindexDataStreamTransportAction;
4348
import org.elasticsearch.datastreams.action.TransportGetDataStreamsAction;
4449
import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleErrorStore;
4550
import org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService;
@@ -73,14 +78,27 @@
7378
import org.elasticsearch.datastreams.rest.RestMigrateToDataStreamAction;
7479
import org.elasticsearch.datastreams.rest.RestModifyDataStreamsAction;
7580
import org.elasticsearch.datastreams.rest.RestPromoteDataStreamAction;
81+
import org.elasticsearch.datastreams.task.ReindexDataStreamPersistentTaskExecutor;
82+
import org.elasticsearch.datastreams.task.ReindexDataStreamPersistentTaskState;
83+
import org.elasticsearch.datastreams.task.ReindexDataStreamStatus;
84+
import org.elasticsearch.datastreams.task.ReindexDataStreamTask;
85+
import org.elasticsearch.datastreams.task.ReindexDataStreamTaskParams;
7686
import org.elasticsearch.features.NodeFeature;
7787
import org.elasticsearch.health.HealthIndicatorService;
7888
import org.elasticsearch.index.IndexSettingProvider;
89+
import org.elasticsearch.persistent.PersistentTaskParams;
90+
import org.elasticsearch.persistent.PersistentTaskState;
91+
import org.elasticsearch.persistent.PersistentTasksExecutor;
7992
import org.elasticsearch.plugins.ActionPlugin;
8093
import org.elasticsearch.plugins.HealthPlugin;
94+
import org.elasticsearch.plugins.PersistentTaskPlugin;
8195
import org.elasticsearch.plugins.Plugin;
8296
import org.elasticsearch.rest.RestController;
8397
import org.elasticsearch.rest.RestHandler;
98+
import org.elasticsearch.tasks.Task;
99+
import org.elasticsearch.threadpool.ThreadPool;
100+
import org.elasticsearch.xcontent.NamedXContentRegistry;
101+
import org.elasticsearch.xcontent.ParseField;
84102

85103
import java.io.IOException;
86104
import java.time.Clock;
@@ -93,7 +111,7 @@
93111

94112
import static org.elasticsearch.cluster.metadata.DataStreamLifecycle.DATA_STREAM_LIFECYCLE_ORIGIN;
95113

96-
public class DataStreamsPlugin extends Plugin implements ActionPlugin, HealthPlugin {
114+
public class DataStreamsPlugin extends Plugin implements ActionPlugin, HealthPlugin, PersistentTaskPlugin {
97115

98116
public static final Setting<TimeValue> TIME_SERIES_POLL_INTERVAL = Setting.timeSetting(
99117
"time_series.poll_interval",
@@ -244,6 +262,7 @@ public Collection<?> createComponents(PluginServices services) {
244262
actions.add(new ActionHandler<>(PutDataStreamOptionsAction.INSTANCE, TransportPutDataStreamOptionsAction.class));
245263
actions.add(new ActionHandler<>(DeleteDataStreamOptionsAction.INSTANCE, TransportDeleteDataStreamOptionsAction.class));
246264
}
265+
actions.add(new ActionHandler<>(ReindexDataStreamAction.INSTANCE, ReindexDataStreamTransportAction.class));
247266
return actions;
248267
}
249268

@@ -302,4 +321,48 @@ public void close() throws IOException {
302321
public Collection<HealthIndicatorService> getHealthIndicatorServices() {
303322
return List.of(dataStreamLifecycleHealthIndicatorService.get());
304323
}
324+
325+
@Override
326+
public List<NamedXContentRegistry.Entry> getNamedXContent() {
327+
return List.of(
328+
new NamedXContentRegistry.Entry(
329+
PersistentTaskState.class,
330+
new ParseField(ReindexDataStreamPersistentTaskState.NAME),
331+
ReindexDataStreamPersistentTaskState::fromXContent
332+
),
333+
new NamedXContentRegistry.Entry(
334+
PersistentTaskParams.class,
335+
new ParseField(ReindexDataStreamTaskParams.NAME),
336+
ReindexDataStreamTaskParams::fromXContent
337+
)
338+
);
339+
}
340+
341+
@Override
342+
public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
343+
return List.of(
344+
new NamedWriteableRegistry.Entry(
345+
PersistentTaskState.class,
346+
ReindexDataStreamPersistentTaskState.NAME,
347+
ReindexDataStreamPersistentTaskState::new
348+
),
349+
new NamedWriteableRegistry.Entry(
350+
PersistentTaskParams.class,
351+
ReindexDataStreamTaskParams.NAME,
352+
ReindexDataStreamTaskParams::new
353+
),
354+
new NamedWriteableRegistry.Entry(Task.Status.class, ReindexDataStreamStatus.NAME, ReindexDataStreamStatus::new)
355+
);
356+
}
357+
358+
@Override
359+
public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(
360+
ClusterService clusterService,
361+
ThreadPool threadPool,
362+
Client client,
363+
SettingsModule settingsModule,
364+
IndexNameExpressionResolver expressionResolver
365+
) {
366+
return List.of(new ReindexDataStreamPersistentTaskExecutor(client, clusterService, ReindexDataStreamTask.TASK_NAME, threadPool));
367+
}
305368
}

0 commit comments

Comments
 (0)