Skip to content

Commit bc1e75e

Browse files
gwbrownpgomulka
gwbrown
and
pgomulka
authored
Register Feature migration persistent task state named XContent (#84192)
This PR properly registers `NamedXContentRegistry` entries for `SystemIndexMigrationTaskParams` and `SystemIndexMigrationTaskState`. It also adds tests for the XContent de/serialization for those classes, and fixes a bug revealed by these tests in `SystemIndexMigrationTaskState`'s parser. Finally, it adds an integration test which simulates the conditions in which #84115 occurs: A node restart while the migration is in progress. This ensures that we have fixed that particular bug. Co-authored-by: David Turner <[email protected]> Co-authored-by: Gordon Brown <[email protected]> (cherry picked from commit 4c3b7b1) Co-authored-by: Przemyslaw Gomulka <[email protected]>
1 parent fbcf5bc commit bc1e75e

File tree

13 files changed

+542
-265
lines changed

13 files changed

+542
-265
lines changed

‎docs/changelog/84192.yaml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
pr: 84192
2+
summary: Registration of `SystemIndexMigrationTask` named xcontent objects
3+
area: Infra/Core
4+
type: bug
5+
issues:
6+
- 84115

‎modules/reindex/src/internalClusterTest/java/org/elasticsearch/migration/AbstractFeatureMigrationIntegTest.java‎

Lines changed: 274 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,274 @@
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 and the Server Side Public License, v 1; you may not use this file except
5+
* in compliance with, at your election, the Elastic License 2.0 or the Server
6+
* Side Public License, v 1.
7+
*/
8+
9+
package org.elasticsearch.migration;
10+
11+
import org.elasticsearch.Version;
12+
import org.elasticsearch.action.ActionListener;
13+
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
14+
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
15+
import org.elasticsearch.action.admin.indices.stats.IndexStats;
16+
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
17+
import org.elasticsearch.action.index.IndexRequestBuilder;
18+
import org.elasticsearch.action.support.ActiveShardCount;
19+
import org.elasticsearch.client.Client;
20+
import org.elasticsearch.cluster.ClusterState;
21+
import org.elasticsearch.cluster.metadata.IndexMetadata;
22+
import org.elasticsearch.cluster.metadata.Metadata;
23+
import org.elasticsearch.cluster.service.ClusterService;
24+
import org.elasticsearch.common.Strings;
25+
import org.elasticsearch.common.settings.Settings;
26+
import org.elasticsearch.indices.AssociatedIndexDescriptor;
27+
import org.elasticsearch.indices.SystemIndexDescriptor;
28+
import org.elasticsearch.plugins.Plugin;
29+
import org.elasticsearch.plugins.SystemIndexPlugin;
30+
import org.elasticsearch.test.ESIntegTestCase;
31+
import org.elasticsearch.xcontent.XContentBuilder;
32+
import org.elasticsearch.xcontent.json.JsonXContent;
33+
import org.junit.Assert;
34+
import org.junit.Before;
35+
36+
import java.io.IOException;
37+
import java.util.ArrayList;
38+
import java.util.Arrays;
39+
import java.util.Collection;
40+
import java.util.Collections;
41+
import java.util.List;
42+
import java.util.Map;
43+
import java.util.Optional;
44+
import java.util.Set;
45+
import java.util.concurrent.atomic.AtomicReference;
46+
import java.util.function.BiConsumer;
47+
import java.util.function.Function;
48+
49+
import static org.hamcrest.Matchers.containsInAnyOrder;
50+
import static org.hamcrest.Matchers.equalTo;
51+
import static org.hamcrest.Matchers.is;
52+
53+
public abstract class AbstractFeatureMigrationIntegTest extends ESIntegTestCase {
54+
55+
static final String VERSION_META_KEY = "version";
56+
static final Version META_VERSION = Version.CURRENT;
57+
static final String DESCRIPTOR_MANAGED_META_KEY = "desciptor_managed";
58+
static final String DESCRIPTOR_INTERNAL_META_KEY = "descriptor_internal";
59+
static final String FEATURE_NAME = "A-test-feature"; // Sorts alphabetically before the feature from MultiFeatureMigrationIT
60+
static final String ORIGIN = AbstractFeatureMigrationIntegTest.class.getSimpleName();
61+
static final String FlAG_SETTING_KEY = IndexMetadata.INDEX_PRIORITY_SETTING.getKey();
62+
static final String INTERNAL_MANAGED_INDEX_NAME = ".int-man-old";
63+
static final int INDEX_DOC_COUNT = 100; // arbitrarily chosen
64+
static final int INTERNAL_MANAGED_FLAG_VALUE = 1;
65+
public static final Version NEEDS_UPGRADE_VERSION = Version.V_7_0_0;
66+
67+
static final SystemIndexDescriptor EXTERNAL_UNMANAGED = SystemIndexDescriptor.builder()
68+
.setIndexPattern(".ext-unman-*")
69+
.setType(SystemIndexDescriptor.Type.EXTERNAL_UNMANAGED)
70+
.setOrigin(ORIGIN)
71+
.setVersionMetaKey(VERSION_META_KEY)
72+
.setAllowedElasticProductOrigins(Collections.singletonList(ORIGIN))
73+
.setMinimumNodeVersion(NEEDS_UPGRADE_VERSION)
74+
.setPriorSystemIndexDescriptors(Collections.emptyList())
75+
.build();
76+
static final SystemIndexDescriptor INTERNAL_UNMANAGED = SystemIndexDescriptor.builder()
77+
.setIndexPattern(".int-unman-*")
78+
.setType(SystemIndexDescriptor.Type.INTERNAL_UNMANAGED)
79+
.setOrigin(ORIGIN)
80+
.setVersionMetaKey(VERSION_META_KEY)
81+
.setAllowedElasticProductOrigins(Collections.emptyList())
82+
.setMinimumNodeVersion(NEEDS_UPGRADE_VERSION)
83+
.setPriorSystemIndexDescriptors(Collections.emptyList())
84+
.build();
85+
86+
static final SystemIndexDescriptor INTERNAL_MANAGED = SystemIndexDescriptor.builder()
87+
.setIndexPattern(".int-man-*")
88+
.setAliasName(".internal-managed-alias")
89+
.setPrimaryIndex(INTERNAL_MANAGED_INDEX_NAME)
90+
.setType(SystemIndexDescriptor.Type.INTERNAL_MANAGED)
91+
.setSettings(createSimpleSettings(NEEDS_UPGRADE_VERSION, INTERNAL_MANAGED_FLAG_VALUE))
92+
.setMappings(createSimpleMapping(true, true))
93+
.setOrigin(ORIGIN)
94+
.setVersionMetaKey(VERSION_META_KEY)
95+
.setAllowedElasticProductOrigins(Collections.emptyList())
96+
.setMinimumNodeVersion(NEEDS_UPGRADE_VERSION)
97+
.setPriorSystemIndexDescriptors(Collections.emptyList())
98+
.build();
99+
static final int INTERNAL_UNMANAGED_FLAG_VALUE = 2;
100+
static final int EXTERNAL_MANAGED_FLAG_VALUE = 3;
101+
static final SystemIndexDescriptor EXTERNAL_MANAGED = SystemIndexDescriptor.builder()
102+
.setIndexPattern(".ext-man-*")
103+
.setAliasName(".external-managed-alias")
104+
.setPrimaryIndex(".ext-man-old")
105+
.setType(SystemIndexDescriptor.Type.EXTERNAL_MANAGED)
106+
.setSettings(createSimpleSettings(NEEDS_UPGRADE_VERSION, EXTERNAL_MANAGED_FLAG_VALUE))
107+
.setMappings(createSimpleMapping(true, false))
108+
.setOrigin(ORIGIN)
109+
.setVersionMetaKey(VERSION_META_KEY)
110+
.setAllowedElasticProductOrigins(Collections.singletonList(ORIGIN))
111+
.setMinimumNodeVersion(NEEDS_UPGRADE_VERSION)
112+
.setPriorSystemIndexDescriptors(Collections.emptyList())
113+
.build();
114+
static final int EXTERNAL_UNMANAGED_FLAG_VALUE = 4;
115+
static final String ASSOCIATED_INDEX_NAME = ".my-associated-idx";
116+
117+
@Before
118+
public void setupTestPlugin() {
119+
TestPlugin.preMigrationHook.set((state) -> Collections.emptyMap());
120+
TestPlugin.postMigrationHook.set((state, metadata) -> {});
121+
}
122+
123+
public void createSystemIndexForDescriptor(SystemIndexDescriptor descriptor) throws InterruptedException {
124+
Assert.assertTrue(
125+
"the strategy used below to create index names for descriptors without a primary index name only works for simple patterns",
126+
descriptor.getIndexPattern().endsWith("*")
127+
);
128+
String indexName = Optional.ofNullable(descriptor.getPrimaryIndex()).orElse(descriptor.getIndexPattern().replace("*", "old"));
129+
CreateIndexRequestBuilder createRequest = prepareCreate(indexName);
130+
createRequest.setWaitForActiveShards(ActiveShardCount.ALL);
131+
if (SystemIndexDescriptor.DEFAULT_SETTINGS.equals(descriptor.getSettings())) {
132+
// unmanaged
133+
createRequest.setSettings(
134+
createSimpleSettings(
135+
NEEDS_UPGRADE_VERSION,
136+
descriptor.isInternal() ? INTERNAL_UNMANAGED_FLAG_VALUE : EXTERNAL_UNMANAGED_FLAG_VALUE
137+
)
138+
);
139+
} else {
140+
// managed
141+
createRequest.setSettings(
142+
Settings.builder()
143+
.put("index.version.created", Version.CURRENT)
144+
.put(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 0)
145+
.build()
146+
);
147+
}
148+
if (descriptor.getMappings() == null) {
149+
createRequest.setMapping(createSimpleMapping(false, descriptor.isInternal()));
150+
}
151+
CreateIndexResponse response = createRequest.get();
152+
Assert.assertTrue(response.isShardsAcknowledged());
153+
154+
List<IndexRequestBuilder> docs = new ArrayList<>(INDEX_DOC_COUNT);
155+
for (int i = 0; i < INDEX_DOC_COUNT; i++) {
156+
docs.add(ESIntegTestCase.client().prepareIndex(indexName).setId(Integer.toString(i)).setSource("some_field", "words words"));
157+
}
158+
indexRandom(true, docs);
159+
IndicesStatsResponse indexStats = ESIntegTestCase.client().admin().indices().prepareStats(indexName).setDocs(true).get();
160+
Assert.assertThat(indexStats.getIndex(indexName).getTotal().getDocs().getCount(), is((long) INDEX_DOC_COUNT));
161+
}
162+
163+
static Settings createSimpleSettings(Version creationVersion, int flagSettingValue) {
164+
return Settings.builder()
165+
.put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
166+
.put(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 0)
167+
.put(FlAG_SETTING_KEY, flagSettingValue)
168+
.put("index.version.created", creationVersion)
169+
.build();
170+
}
171+
172+
static String createSimpleMapping(boolean descriptorManaged, boolean descriptorInternal) {
173+
try (XContentBuilder builder = JsonXContent.contentBuilder()) {
174+
builder.startObject();
175+
{
176+
builder.startObject("_meta");
177+
builder.field(VERSION_META_KEY, META_VERSION);
178+
builder.field(DESCRIPTOR_MANAGED_META_KEY, descriptorManaged);
179+
builder.field(DESCRIPTOR_INTERNAL_META_KEY, descriptorInternal);
180+
builder.endObject();
181+
182+
builder.field("dynamic", "strict");
183+
builder.startObject("properties");
184+
{
185+
builder.startObject("some_field");
186+
builder.field("type", "keyword");
187+
builder.endObject();
188+
}
189+
builder.endObject();
190+
}
191+
builder.endObject();
192+
return Strings.toString(builder);
193+
} catch (IOException e) {
194+
// Just rethrow, it should be impossible for this to throw here
195+
throw new AssertionError(e);
196+
}
197+
}
198+
199+
public void assertIndexHasCorrectProperties(
200+
Metadata metadata,
201+
String indexName,
202+
int settingsFlagValue,
203+
boolean isManaged,
204+
boolean isInternal,
205+
Collection<String> aliasNames
206+
) {
207+
IndexMetadata imd = metadata.index(indexName);
208+
assertThat(imd.getSettings().get(FlAG_SETTING_KEY), equalTo(Integer.toString(settingsFlagValue)));
209+
final Map<String, Object> mapping = imd.mapping().getSourceAsMap();
210+
@SuppressWarnings("unchecked")
211+
final Map<String, Object> meta = (Map<String, Object>) mapping.get("_meta");
212+
assertThat(meta.get(DESCRIPTOR_MANAGED_META_KEY), is(isManaged));
213+
assertThat(meta.get(DESCRIPTOR_INTERNAL_META_KEY), is(isInternal));
214+
215+
assertThat(imd.isSystem(), is(true));
216+
217+
Set<String> actualAliasNames = imd.getAliases().keySet();
218+
assertThat(actualAliasNames, containsInAnyOrder(aliasNames.toArray()));
219+
220+
IndicesStatsResponse indexStats = client().admin().indices().prepareStats(imd.getIndex().getName()).setDocs(true).get();
221+
assertNotNull(indexStats);
222+
final IndexStats thisIndexStats = indexStats.getIndex(imd.getIndex().getName());
223+
assertNotNull(thisIndexStats);
224+
assertNotNull(thisIndexStats.getTotal());
225+
assertNotNull(thisIndexStats.getTotal().getDocs());
226+
assertThat(thisIndexStats.getTotal().getDocs().getCount(), is((long) INDEX_DOC_COUNT));
227+
}
228+
229+
public static class TestPlugin extends Plugin implements SystemIndexPlugin {
230+
public static final AtomicReference<Function<ClusterState, Map<String, Object>>> preMigrationHook = new AtomicReference<>();
231+
public static final AtomicReference<BiConsumer<ClusterState, Map<String, Object>>> postMigrationHook = new AtomicReference<>();
232+
233+
public TestPlugin() {
234+
235+
}
236+
237+
@Override
238+
public String getFeatureName() {
239+
return FEATURE_NAME;
240+
}
241+
242+
@Override
243+
public String getFeatureDescription() {
244+
return "a plugin for testing system index migration";
245+
}
246+
247+
@Override
248+
public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings settings) {
249+
return Arrays.asList(INTERNAL_MANAGED, INTERNAL_UNMANAGED, EXTERNAL_MANAGED, EXTERNAL_UNMANAGED);
250+
}
251+
252+
@Override
253+
public Collection<AssociatedIndexDescriptor> getAssociatedIndexDescriptors() {
254+
255+
return Collections.singletonList(new AssociatedIndexDescriptor(ASSOCIATED_INDEX_NAME, TestPlugin.class.getCanonicalName()));
256+
}
257+
258+
@Override
259+
public void prepareForIndicesMigration(ClusterService clusterService, Client client, ActionListener<Map<String, Object>> listener) {
260+
listener.onResponse(preMigrationHook.get().apply(clusterService.state()));
261+
}
262+
263+
@Override
264+
public void indicesMigrationComplete(
265+
Map<String, Object> preUpgradeMetadata,
266+
ClusterService clusterService,
267+
Client client,
268+
ActionListener<Boolean> listener
269+
) {
270+
postMigrationHook.get().accept(clusterService.state(), preUpgradeMetadata);
271+
listener.onResponse(true);
272+
}
273+
}
274+
}

0 commit comments

Comments
 (0)