Skip to content

Commit 6823d6a

Browse files
pugnascotiapugnascotia
authored
Migrate .tasks to be managed automatically (#67351)
Re-apply changes from 0c9b9c1, which migrates the `.tasks` system index to be managed automatically by the system indices infrastructure. Changes went into #67114 that, I hope, will avoid the problems we saw before in the BWC tests in CI.
1 parent 0ca6819 commit 6823d6a

File tree

3 files changed

+105
-151
lines changed

3 files changed

+105
-151
lines changed

‎server/src/main/java/org/elasticsearch/indices/SystemIndices.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@
2727
import java.util.stream.Collectors;
2828

2929
import static java.util.stream.Collectors.toUnmodifiableList;
30-
import static org.elasticsearch.tasks.TaskResultsService.TASK_INDEX;
30+
import static org.elasticsearch.tasks.TaskResultsService.TASKS_DESCRIPTOR;
3131

3232
/**
3333
* This class holds the {@link SystemIndexDescriptor} objects that represent system indices the
@@ -36,7 +36,7 @@
3636
*/
3737
public class SystemIndices {
3838
private static final Map<String, Collection<SystemIndexDescriptor>> SERVER_SYSTEM_INDEX_DESCRIPTORS = Map.of(
39-
TaskResultsService.class.getName(), List.of(new SystemIndexDescriptor(TASK_INDEX + "*", "Task Result Index"))
39+
TaskResultsService.class.getName(), List.of(TASKS_DESCRIPTOR)
4040
);
4141

4242
private final CharacterRunAutomaton runAutomaton;

‎server/src/main/java/org/elasticsearch/tasks/TaskResultsService.java

Lines changed: 103 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -11,41 +11,32 @@
1111
import org.apache.logging.log4j.Logger;
1212
import org.apache.logging.log4j.message.ParameterizedMessage;
1313
import org.elasticsearch.ElasticsearchException;
14-
import org.elasticsearch.ExceptionsHelper;
15-
import org.elasticsearch.ResourceAlreadyExistsException;
14+
import org.elasticsearch.Version;
1615
import org.elasticsearch.action.ActionListener;
17-
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
18-
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
1916
import org.elasticsearch.action.bulk.BackoffPolicy;
2017
import org.elasticsearch.action.index.IndexRequestBuilder;
2118
import org.elasticsearch.action.index.IndexResponse;
2219
import org.elasticsearch.client.Client;
2320
import org.elasticsearch.client.OriginSettingClient;
2421
import org.elasticsearch.client.Requests;
25-
import org.elasticsearch.cluster.ClusterState;
2622
import org.elasticsearch.cluster.metadata.IndexMetadata;
27-
import org.elasticsearch.cluster.metadata.MappingMetadata;
28-
import org.elasticsearch.cluster.service.ClusterService;
2923
import org.elasticsearch.common.inject.Inject;
3024
import org.elasticsearch.common.settings.Settings;
3125
import org.elasticsearch.common.unit.TimeValue;
3226
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
3327
import org.elasticsearch.common.xcontent.ToXContent;
3428
import org.elasticsearch.common.xcontent.XContentBuilder;
3529
import org.elasticsearch.common.xcontent.XContentFactory;
36-
import org.elasticsearch.common.xcontent.XContentType;
37-
import org.elasticsearch.core.internal.io.Streams;
30+
import org.elasticsearch.indices.SystemIndexDescriptor;
3831
import org.elasticsearch.threadpool.ThreadPool;
3932

40-
import java.io.ByteArrayOutputStream;
4133
import java.io.IOException;
42-
import java.io.InputStream;
43-
import java.nio.charset.StandardCharsets;
34+
import java.io.UncheckedIOException;
4435
import java.util.Iterator;
45-
import java.util.Map;
4636

4737
import static org.elasticsearch.action.admin.cluster.node.tasks.get.GetTaskAction.TASKS_ORIGIN;
4838
import static org.elasticsearch.common.unit.TimeValue.timeValueMillis;
39+
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
4940

5041
/**
5142
* Service that can store task results.
@@ -55,12 +46,17 @@ public class TaskResultsService {
5546
private static final Logger logger = LogManager.getLogger(TaskResultsService.class);
5647

5748
public static final String TASK_INDEX = ".tasks";
58-
59-
public static final String TASK_RESULT_INDEX_MAPPING_FILE = "task-index-mapping.json";
60-
6149
public static final String TASK_RESULT_MAPPING_VERSION_META_FIELD = "version";
6250

63-
public static final int TASK_RESULT_MAPPING_VERSION = 3;
51+
public static final SystemIndexDescriptor TASKS_DESCRIPTOR = SystemIndexDescriptor.builder()
52+
.setIndexPattern(TASK_INDEX + "*")
53+
.setPrimaryIndex(TASK_INDEX)
54+
.setDescription("Task Result Index")
55+
.setSettings(getTaskResultIndexSettings())
56+
.setMappings(getTaskResultIndexMappings())
57+
.setVersionMetaKey(TASK_RESULT_MAPPING_VERSION_META_FIELD)
58+
.setOrigin(TASKS_ORIGIN)
59+
.build();
6460

6561
/**
6662
* The backoff policy to use when saving a task result fails. The total wait
@@ -71,75 +67,15 @@ public class TaskResultsService {
7167

7268
private final Client client;
7369

74-
private final ClusterService clusterService;
75-
7670
private final ThreadPool threadPool;
7771

7872
@Inject
79-
public TaskResultsService(Client client, ClusterService clusterService, ThreadPool threadPool) {
73+
public TaskResultsService(Client client, ThreadPool threadPool) {
8074
this.client = new OriginSettingClient(client, TASKS_ORIGIN);
81-
this.clusterService = clusterService;
8275
this.threadPool = threadPool;
8376
}
8477

8578
public void storeResult(TaskResult taskResult, ActionListener<Void> listener) {
86-
87-
ClusterState state = clusterService.state();
88-
89-
if (state.routingTable().hasIndex(TASK_INDEX) == false) {
90-
CreateIndexRequest createIndexRequest = new CreateIndexRequest();
91-
createIndexRequest.settings(taskResultIndexSettings());
92-
createIndexRequest.index(TASK_INDEX);
93-
createIndexRequest.mapping(taskResultIndexMapping());
94-
createIndexRequest.cause("auto(task api)");
95-
96-
client.admin().indices().create(createIndexRequest, new ActionListener<CreateIndexResponse>() {
97-
@Override
98-
public void onResponse(CreateIndexResponse result) {
99-
doStoreResult(taskResult, listener);
100-
}
101-
102-
@Override
103-
public void onFailure(Exception e) {
104-
if (ExceptionsHelper.unwrapCause(e) instanceof ResourceAlreadyExistsException) {
105-
// we have the index, do it
106-
try {
107-
doStoreResult(taskResult, listener);
108-
} catch (Exception inner) {
109-
inner.addSuppressed(e);
110-
listener.onFailure(inner);
111-
}
112-
} else {
113-
listener.onFailure(e);
114-
}
115-
}
116-
});
117-
} else {
118-
IndexMetadata metadata = state.getMetadata().index(TASK_INDEX);
119-
if (getTaskResultMappingVersion(metadata) < TASK_RESULT_MAPPING_VERSION) {
120-
// The index already exists but doesn't have our mapping
121-
client.admin().indices().preparePutMapping(TASK_INDEX)
122-
.setSource(taskResultIndexMapping(), XContentType.JSON)
123-
.execute(ActionListener.delegateFailure(listener, (l, r) -> doStoreResult(taskResult, listener)));
124-
} else {
125-
doStoreResult(taskResult, listener);
126-
}
127-
}
128-
}
129-
130-
private int getTaskResultMappingVersion(IndexMetadata metadata) {
131-
MappingMetadata mappingMetadata = metadata.mapping();
132-
if (mappingMetadata == null) {
133-
return 0;
134-
}
135-
@SuppressWarnings("unchecked") Map<String, Object> meta = (Map<String, Object>) mappingMetadata.sourceAsMap().get("_meta");
136-
if (meta == null || meta.containsKey(TASK_RESULT_MAPPING_VERSION_META_FIELD) == false) {
137-
return 1; // The mapping was created before meta field was introduced
138-
}
139-
return (int) meta.get(TASK_RESULT_MAPPING_VERSION_META_FIELD);
140-
}
141-
142-
private void doStoreResult(TaskResult taskResult, ActionListener<Void> listener) {
14379
IndexRequestBuilder index = client.prepareIndex(TASK_INDEX).setId(taskResult.getTask().getTaskId().toString());
14480
try (XContentBuilder builder = XContentFactory.contentBuilder(Requests.INDEX_CONTENT_TYPE)) {
14581
taskResult.toXContent(builder, ToXContent.EMPTY_PARAMS);
@@ -171,24 +107,102 @@ public void onFailure(Exception e) {
171107
});
172108
}
173109

174-
private Settings taskResultIndexSettings() {
110+
private static Settings getTaskResultIndexSettings() {
175111
return Settings.builder()
176112
.put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
177113
.put(IndexMetadata.INDEX_AUTO_EXPAND_REPLICAS_SETTING.getKey(), "0-1")
178114
.put(IndexMetadata.SETTING_PRIORITY, Integer.MAX_VALUE)
179115
.build();
180116
}
181117

182-
public String taskResultIndexMapping() {
183-
try (InputStream is = getClass().getResourceAsStream(TASK_RESULT_INDEX_MAPPING_FILE)) {
184-
ByteArrayOutputStream out = new ByteArrayOutputStream();
185-
Streams.copy(is, out);
186-
return out.toString(StandardCharsets.UTF_8.name());
187-
} catch (Exception e) {
188-
logger.error(() -> new ParameterizedMessage(
189-
"failed to create tasks results index template [{}]", TASK_RESULT_INDEX_MAPPING_FILE), e);
190-
throw new IllegalStateException("failed to create tasks results index template [" + TASK_RESULT_INDEX_MAPPING_FILE + "]", e);
191-
}
118+
private static XContentBuilder getTaskResultIndexMappings() {
119+
try {
120+
final XContentBuilder builder = jsonBuilder();
121+
122+
builder.startObject();
123+
{
124+
builder.startObject("_meta");
125+
builder.field(TASK_RESULT_MAPPING_VERSION_META_FIELD, Version.CURRENT.toString());
126+
builder.endObject();
127+
128+
builder.field("dynamic", "strict");
129+
builder.startObject("properties");
130+
{
131+
builder.startObject("completed");
132+
builder.field("type", "boolean");
133+
builder.endObject();
134+
135+
builder.startObject("task");
136+
{
137+
builder.startObject("properties");
138+
{
139+
builder.startObject("action");
140+
builder.field("type", "keyword");
141+
builder.endObject();
142+
143+
builder.startObject("cancellable");
144+
builder.field("type", "boolean");
145+
builder.endObject();
146+
147+
builder.startObject("id");
148+
builder.field("type", "long");
149+
builder.endObject();
150+
151+
builder.startObject("parent_task_id");
152+
builder.field("type", "keyword");
153+
builder.endObject();
154+
155+
builder.startObject("node");
156+
builder.field("type", "keyword");
157+
builder.endObject();
158+
159+
builder.startObject("running_time_in_nanos");
160+
builder.field("type", "long");
161+
builder.endObject();
162+
163+
builder.startObject("start_time_in_millis");
164+
builder.field("type", "long");
165+
builder.endObject();
166+
167+
builder.startObject("type");
168+
builder.field("type", "keyword");
169+
builder.endObject();
170+
171+
builder.startObject("status");
172+
builder.field("type", "object");
173+
builder.field("enabled", false);
174+
builder.endObject();
175+
176+
builder.startObject("description");
177+
builder.field("type", "text");
178+
builder.endObject();
179+
180+
builder.startObject("headers");
181+
builder.field("type", "object");
182+
builder.field("enabled", false);
183+
builder.endObject();
184+
}
185+
builder.endObject();
186+
}
187+
builder.endObject();
188+
189+
builder.startObject("response");
190+
builder.field("type", "object");
191+
builder.field("enabled", false);
192+
builder.endObject();
192193

194+
builder.startObject("error");
195+
builder.field("type", "object");
196+
builder.field("enabled", false);
197+
builder.endObject();
198+
}
199+
builder.endObject();
200+
}
201+
202+
builder.endObject();
203+
return builder;
204+
} catch (IOException e) {
205+
throw new UncheckedIOException("Failed to build " + TASK_INDEX + " index mappings", e);
206+
}
193207
}
194208
}

‎server/src/main/resources/org/elasticsearch/tasks/task-index-mapping.json

Lines changed: 0 additions & 60 deletions
This file was deleted.

0 commit comments

Comments
 (0)