Skip to content

Commit 181d1a0

Browse files
pugnascotiapugnascotia
authored
Migrate .tasks to be managed automatically (#68667)
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 9d38152 commit 181d1a0

File tree

3 files changed

+111
-153
lines changed

3 files changed

+111
-153
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
@@ -30,7 +30,7 @@
3030
import static java.util.Collections.singletonMap;
3131
import static java.util.Collections.unmodifiableList;
3232
import static java.util.Collections.unmodifiableMap;
33-
import static org.elasticsearch.tasks.TaskResultsService.TASK_INDEX;
33+
import static org.elasticsearch.tasks.TaskResultsService.TASKS_DESCRIPTOR;
3434

3535
/**
3636
* This class holds the {@link SystemIndexDescriptor} objects that represent system indices the
@@ -39,7 +39,7 @@
3939
*/
4040
public class SystemIndices {
4141
private static final Map<String, Collection<SystemIndexDescriptor>> SERVER_SYSTEM_INDEX_DESCRIPTORS = singletonMap(
42-
TaskResultsService.class.getName(), singletonList(new SystemIndexDescriptor(TASK_INDEX + "*", "Task Result Index"))
42+
TaskResultsService.class.getName(), singletonList(TASKS_DESCRIPTOR)
4343
);
4444

4545
private final CharacterRunAutomaton runAutomaton;

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

Lines changed: 109 additions & 91 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,14 +46,19 @@ public class TaskResultsService {
5546
private static final Logger logger = LogManager.getLogger(TaskResultsService.class);
5647

5748
public static final String TASK_INDEX = ".tasks";
58-
5949
public static final String TASK_TYPE = "task";
60-
61-
public static final String TASK_RESULT_INDEX_MAPPING_FILE = "task-index-mapping.json";
62-
6350
public static final String TASK_RESULT_MAPPING_VERSION_META_FIELD = "version";
6451

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

6763
/**
6864
* The backoff policy to use when saving a task result fails. The total wait
@@ -73,76 +69,16 @@ public class TaskResultsService {
7369

7470
private final Client client;
7571

76-
private final ClusterService clusterService;
77-
7872
private final ThreadPool threadPool;
7973

8074
@Inject
81-
public TaskResultsService(Client client, ClusterService clusterService, ThreadPool threadPool) {
75+
public TaskResultsService(Client client, ThreadPool threadPool) {
8276
this.client = new OriginSettingClient(client, TASKS_ORIGIN);
83-
this.clusterService = clusterService;
8477
this.threadPool = threadPool;
8578
}
8679

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

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

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

193+
builder.startObject("response");
194+
builder.field("type", "object");
195+
builder.field("enabled", false);
196+
builder.endObject();
197+
198+
builder.startObject("error");
199+
builder.field("type", "object");
200+
builder.field("enabled", false);
201+
builder.endObject();
202+
}
203+
builder.endObject();
204+
}
205+
builder.endObject();
206+
}
207+
208+
builder.endObject();
209+
return builder;
210+
} catch (IOException e) {
211+
throw new UncheckedIOException("Failed to build " + TASK_INDEX + " index mappings", e);
212+
}
195213
}
196214
}

‎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)