Skip to content

Commit 560bed5

Browse files
benwtrentbenwtrent
committed
ML: creating ML State write alias and pointing writes there (#37483)
* ML: creating ML State write alias and pointing writes there * Moving alias check to openJob method * adjusting concrete index lookup for ml-state
1 parent df3df4a commit 560bed5

File tree

16 files changed

+266
-121
lines changed

16 files changed

+266
-121
lines changed

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/job/persistence/AnomalyDetectorsIndex.java‎

Lines changed: 84 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,25 @@
55
*/
66
package org.elasticsearch.xpack.core.ml.job.persistence;
77

8+
import org.elasticsearch.ResourceAlreadyExistsException;
9+
import org.elasticsearch.action.ActionListener;
10+
import org.elasticsearch.action.admin.indices.alias.Alias;
11+
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
12+
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
13+
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
14+
import org.elasticsearch.action.support.IndicesOptions;
15+
import org.elasticsearch.action.support.master.AcknowledgedResponse;
16+
import org.elasticsearch.client.Client;
17+
import org.elasticsearch.cluster.ClusterState;
18+
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
19+
import org.elasticsearch.common.settings.Settings;
20+
21+
import java.util.Arrays;
22+
import java.util.Collections;
23+
24+
import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
25+
import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
26+
827
/**
928
* Methods for handling index naming related functions
1029
*/
@@ -40,11 +59,11 @@ public static String resultsWriteAlias(String jobId) {
4059
}
4160

4261
/**
43-
* The name of the default index where a job's state is stored
44-
* @return The index name
62+
* The name of the alias pointing to the appropriate index for writing job state
63+
* @return The write alias name
4564
*/
46-
public static String jobStateIndexName() {
47-
return AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX;
65+
public static String jobStateIndexWriteAlias() {
66+
return AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX + "-write";
4867
}
4968

5069
/**
@@ -64,4 +83,65 @@ public static String configIndexName() {
6483
return AnomalyDetectorsIndexFields.CONFIG_INDEX;
6584
}
6685

86+
/**
87+
* Create the .ml-state index (if necessary)
88+
* Create the .ml-state-write alias for the .ml-state index (if necessary)
89+
*/
90+
public static void createStateIndexAndAliasIfNecessary(Client client, ClusterState state, final ActionListener<Boolean> finalListener) {
91+
92+
if (state.getMetaData().getAliasAndIndexLookup().containsKey(jobStateIndexWriteAlias())) {
93+
finalListener.onResponse(false);
94+
return;
95+
}
96+
97+
final ActionListener<String> createAliasListener = ActionListener.wrap(
98+
concreteIndexName -> {
99+
final IndicesAliasesRequest request = client.admin()
100+
.indices()
101+
.prepareAliases()
102+
.addAlias(concreteIndexName, jobStateIndexWriteAlias())
103+
.request();
104+
executeAsyncWithOrigin(client.threadPool().getThreadContext(),
105+
ML_ORIGIN,
106+
request,
107+
ActionListener.<AcknowledgedResponse>wrap(
108+
resp -> finalListener.onResponse(resp.isAcknowledged()),
109+
finalListener::onFailure),
110+
client.admin().indices()::aliases);
111+
},
112+
finalListener::onFailure
113+
);
114+
115+
IndexNameExpressionResolver indexNameExpressionResolver = new IndexNameExpressionResolver(Settings.EMPTY);
116+
String[] stateIndices = indexNameExpressionResolver.concreteIndexNames(state,
117+
IndicesOptions.lenientExpandOpen(),
118+
jobStateIndexPattern());
119+
if (stateIndices.length > 0) {
120+
Arrays.sort(stateIndices, Collections.reverseOrder());
121+
createAliasListener.onResponse(stateIndices[0]);
122+
} else {
123+
CreateIndexRequest createIndexRequest = client.admin()
124+
.indices()
125+
.prepareCreate(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX)
126+
.addAlias(new Alias(jobStateIndexWriteAlias()))
127+
.request();
128+
executeAsyncWithOrigin(client.threadPool().getThreadContext(),
129+
ML_ORIGIN,
130+
createIndexRequest,
131+
ActionListener.<CreateIndexResponse>wrap(
132+
createIndexResponse -> finalListener.onResponse(true),
133+
createIndexFailure -> {
134+
// If it was created between our last check, and this request being handled, we should add the alias
135+
// Adding an alias that already exists is idempotent. So, no need to double check if the alias exists
136+
// as well.
137+
if (createIndexFailure instanceof ResourceAlreadyExistsException) {
138+
createAliasListener.onResponse(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX);
139+
} else {
140+
finalListener.onFailure(createIndexFailure);
141+
}
142+
}),
143+
client.admin().indices()::create);
144+
}
145+
}
146+
67147
}

‎x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/store/ReservedRolesStoreTests.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,6 @@
7373
import org.elasticsearch.xpack.core.ml.action.UpdateProcessAction;
7474
import org.elasticsearch.xpack.core.ml.action.ValidateDetectorAction;
7575
import org.elasticsearch.xpack.core.ml.action.ValidateJobConfigAction;
76-
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
7776
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
7877
import org.elasticsearch.xpack.core.ml.notifications.AuditorField;
7978
import org.elasticsearch.xpack.core.monitoring.action.MonitoringBulkAction;
@@ -761,7 +760,7 @@ public void testMachineLearningAdminRole() {
761760

762761
assertNoAccessAllowed(role, "foo");
763762
assertOnlyReadAllowed(role, MlMetaIndex.INDEX_NAME);
764-
assertOnlyReadAllowed(role, AnomalyDetectorsIndex.jobStateIndexName());
763+
assertOnlyReadAllowed(role, AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX);
765764
assertOnlyReadAllowed(role, AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + AnomalyDetectorsIndexFields.RESULTS_INDEX_DEFAULT);
766765
assertOnlyReadAllowed(role, AuditorField.NOTIFICATIONS_INDEX);
767766
}
@@ -813,7 +812,7 @@ public void testMachineLearningUserRole() {
813812

814813
assertNoAccessAllowed(role, "foo");
815814
assertNoAccessAllowed(role, MlMetaIndex.INDEX_NAME);
816-
assertNoAccessAllowed(role, AnomalyDetectorsIndex.jobStateIndexName());
815+
assertNoAccessAllowed(role, AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX);
817816
assertOnlyReadAllowed(role, AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + AnomalyDetectorsIndexFields.RESULTS_INDEX_DEFAULT);
818817
assertOnlyReadAllowed(role, AuditorField.NOTIFICATIONS_INDEX);
819818
}

‎x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/test/rest/XPackRestTestHelper.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import org.elasticsearch.test.ESTestCase;
1717
import org.elasticsearch.xpack.core.ml.MlMetaIndex;
1818
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
19+
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
1920
import org.elasticsearch.xpack.core.ml.notifications.AuditorField;
2021

2122
import java.io.IOException;
@@ -30,13 +31,13 @@ public final class XPackRestTestHelper {
3031
public static final List<String> ML_PRE_V660_TEMPLATES = Collections.unmodifiableList(
3132
Arrays.asList(AuditorField.NOTIFICATIONS_INDEX,
3233
MlMetaIndex.INDEX_NAME,
33-
AnomalyDetectorsIndex.jobStateIndexName(),
34+
AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX,
3435
AnomalyDetectorsIndex.jobResultsIndexPrefix()));
3536

3637
public static final List<String> ML_POST_V660_TEMPLATES = Collections.unmodifiableList(
3738
Arrays.asList(AuditorField.NOTIFICATIONS_INDEX,
3839
MlMetaIndex.INDEX_NAME,
39-
AnomalyDetectorsIndex.jobStateIndexName(),
40+
AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX,
4041
AnomalyDetectorsIndex.jobResultsIndexPrefix(),
4142
AnomalyDetectorsIndex.configIndexName()));
4243

‎x-pack/plugin/ml/qa/native-multi-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/DeleteExpiredDataIT.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -181,7 +181,7 @@ public void testDeleteExpiredData() throws Exception {
181181
bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
182182
for (int i = 0; i < 10010; i++) {
183183
String docId = "non_existing_job_" + randomFrom("model_state_1234567#" + i, "quantiles", "categorizer_state#" + i);
184-
IndexRequest indexRequest = new IndexRequest(AnomalyDetectorsIndex.jobStateIndexName(), "doc", docId);
184+
IndexRequest indexRequest = new IndexRequest(AnomalyDetectorsIndex.jobStateIndexWriteAlias(), "doc", docId);
185185
indexRequest.source(Collections.emptyMap());
186186
bulkRequestBuilder.add(indexRequest);
187187
}

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@
109109
import org.elasticsearch.xpack.core.ml.action.ValidateDetectorAction;
110110
import org.elasticsearch.xpack.core.ml.action.ValidateJobConfigAction;
111111
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
112+
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
112113
import org.elasticsearch.xpack.core.ml.job.persistence.ElasticsearchMappings;
113114
import org.elasticsearch.xpack.core.ml.notifications.AuditMessage;
114115
import org.elasticsearch.xpack.core.ml.notifications.AuditorField;
@@ -698,7 +699,7 @@ public UnaryOperator<Map<String, IndexTemplateMetaData>> getIndexTemplateMetaDat
698699
}
699700

700701
try (XContentBuilder stateMapping = ElasticsearchMappings.stateMapping()) {
701-
IndexTemplateMetaData stateTemplate = IndexTemplateMetaData.builder(AnomalyDetectorsIndex.jobStateIndexName())
702+
IndexTemplateMetaData stateTemplate = IndexTemplateMetaData.builder(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX)
702703
.patterns(Collections.singletonList(AnomalyDetectorsIndex.jobStateIndexPattern()))
703704
// TODO review these settings
704705
.settings(Settings.builder()
@@ -707,9 +708,9 @@ public UnaryOperator<Map<String, IndexTemplateMetaData>> getIndexTemplateMetaDat
707708
.putMapping(ElasticsearchMappings.DOC_TYPE, Strings.toString(stateMapping))
708709
.version(Version.CURRENT.id)
709710
.build();
710-
templates.put(AnomalyDetectorsIndex.jobStateIndexName(), stateTemplate);
711+
templates.put(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX, stateTemplate);
711712
} catch (IOException e) {
712-
logger.error("Error loading the template for the " + AnomalyDetectorsIndex.jobStateIndexName() + " index", e);
713+
logger.error("Error loading the template for the " + AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX + " index", e);
713714
}
714715

715716
try (XContentBuilder docMapping = ElasticsearchMappings.resultsMapping()) {
@@ -739,7 +740,7 @@ public UnaryOperator<Map<String, IndexTemplateMetaData>> getIndexTemplateMetaDat
739740
public static boolean allTemplatesInstalled(ClusterState clusterState) {
740741
boolean allPresent = true;
741742
List<String> templateNames = Arrays.asList(AuditorField.NOTIFICATIONS_INDEX, MlMetaIndex.INDEX_NAME,
742-
AnomalyDetectorsIndex.jobStateIndexName(), AnomalyDetectorsIndex.jobResultsIndexPrefix(),
743+
AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX, AnomalyDetectorsIndex.jobResultsIndexPrefix(),
743744
AnomalyDetectorsIndex.configIndexName());
744745
for (String templateName : templateNames) {
745746
allPresent = allPresent && TemplateUtils.checkTemplateExistsAndVersionIsGTECurrentVersion(templateName, clusterState);

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlConfigMigrator.java‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -349,7 +349,7 @@ public void snapshotMlMeta(MlMetadata mlMetadata, ActionListener<Boolean> listen
349349

350350
logger.debug("taking a snapshot of ml_metadata");
351351
String documentId = "ml-config";
352-
IndexRequestBuilder indexRequest = client.prepareIndex(AnomalyDetectorsIndex.jobStateIndexName(),
352+
IndexRequestBuilder indexRequest = client.prepareIndex(AnomalyDetectorsIndex.jobStateIndexWriteAlias(),
353353
ElasticsearchMappings.DOC_TYPE, documentId)
354354
.setOpType(DocWriteRequest.OpType.CREATE);
355355

@@ -366,8 +366,10 @@ public void snapshotMlMeta(MlMetadata mlMetadata, ActionListener<Boolean> listen
366366
return;
367367
}
368368

369-
executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, indexRequest.request(),
370-
ActionListener.<IndexResponse>wrap(
369+
AnomalyDetectorsIndex.createStateIndexAndAliasIfNecessary(client, clusterService.state(), ActionListener.wrap(
370+
r -> {
371+
executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, indexRequest.request(),
372+
ActionListener.<IndexResponse>wrap(
371373
indexResponse -> {
372374
listener.onResponse(indexResponse.getResult() == DocWriteResponse.Result.CREATED);
373375
},
@@ -379,8 +381,11 @@ public void snapshotMlMeta(MlMetadata mlMetadata, ActionListener<Boolean> listen
379381
listener.onFailure(e);
380382
}
381383
}),
382-
client::index
383-
);
384+
client::index
385+
);
386+
},
387+
listener::onFailure
388+
));
384389
}
385390

386391
private void createConfigIndex(ActionListener<Boolean> listener) {

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportOpenJobAction.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -617,7 +617,7 @@ public void onFailure(Exception e) {
617617
// Try adding state doc mapping
618618
ActionListener<Boolean> resultsPutMappingHandler = ActionListener.wrap(
619619
response -> {
620-
addDocMappingIfMissing(AnomalyDetectorsIndex.jobStateIndexName(), ElasticsearchMappings::stateMapping,
620+
addDocMappingIfMissing(AnomalyDetectorsIndex.jobStateIndexWriteAlias(), ElasticsearchMappings::stateMapping,
621621
state, missingMappingsListener);
622622
}, listener::onFailure
623623
);
@@ -800,6 +800,7 @@ public static class OpenJobPersistentTasksExecutor extends PersistentTasksExecut
800800
private volatile int maxConcurrentJobAllocations;
801801
private volatile int maxMachineMemoryPercent;
802802
private volatile int maxLazyMLNodes;
803+
private volatile ClusterState clusterState;
803804

804805
public OpenJobPersistentTasksExecutor(Settings settings, ClusterService clusterService,
805806
AutodetectProcessManager autodetectProcessManager, MlMemoryTracker memoryTracker,
@@ -817,6 +818,7 @@ public OpenJobPersistentTasksExecutor(Settings settings, ClusterService clusterS
817818
clusterService.getClusterSettings()
818819
.addSettingsUpdateConsumer(MachineLearning.MAX_MACHINE_MEMORY_PERCENT, this::setMaxMachineMemoryPercent);
819820
clusterService.getClusterSettings().addSettingsUpdateConsumer(MachineLearning.MAX_LAZY_ML_NODES, this::setMaxLazyMLNodes);
821+
clusterService.addListener(event -> clusterState = event.state());
820822
}
821823

822824
@Override
@@ -878,7 +880,7 @@ protected void nodeOperation(AllocatedPersistentTask task, OpenJobAction.JobPara
878880
}
879881

880882
String jobId = jobTask.getJobId();
881-
autodetectProcessManager.openJob(jobTask, e2 -> {
883+
autodetectProcessManager.openJob(jobTask, clusterState, e2 -> {
882884
if (e2 == null) {
883885
FinalizeJobExecutionAction.Request finalizeRequest = new FinalizeJobExecutionAction.Request(new String[]{jobId});
884886
executeAsyncWithOrigin(client, ML_ORIGIN, FinalizeJobExecutionAction.INSTANCE, finalizeRequest,

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportRevertModelSnapshotAction.java‎

Lines changed: 30 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.xpack.core.ml.action.RevertModelSnapshotAction;
2525
import org.elasticsearch.xpack.core.ml.job.config.JobState;
2626
import org.elasticsearch.xpack.core.ml.job.messages.Messages;
27+
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
2728
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSnapshot;
2829
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
2930
import org.elasticsearch.xpack.ml.MlConfigMigrationEligibilityCheck;
@@ -79,26 +80,38 @@ protected void masterOperation(RevertModelSnapshotAction.Request request, Cluste
7980
logger.debug("Received request to revert to snapshot id '{}' for job '{}', deleting intervening results: {}",
8081
request.getSnapshotId(), request.getJobId(), request.getDeleteInterveningResults());
8182

82-
jobManager.jobExists(request.getJobId(), ActionListener.wrap(
83-
exists -> {
84-
PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
85-
JobState jobState = MlTasks.getJobState(request.getJobId(), tasks);
8683

87-
if (jobState.equals(JobState.CLOSED) == false) {
88-
throw ExceptionsHelper.conflictStatusException(Messages.getMessage(Messages.REST_JOB_NOT_CLOSED_REVERT));
84+
// 3. Revert the state
85+
ActionListener<Boolean> jobExistsListener = ActionListener.wrap(
86+
exists -> {
87+
PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
88+
JobState jobState = MlTasks.getJobState(request.getJobId(), tasks);
89+
90+
if (jobState.equals(JobState.CLOSED) == false) {
91+
throw ExceptionsHelper.conflictStatusException(Messages.getMessage(Messages.REST_JOB_NOT_CLOSED_REVERT));
92+
}
93+
94+
getModelSnapshot(request, jobResultsProvider, modelSnapshot -> {
95+
ActionListener<RevertModelSnapshotAction.Response> wrappedListener = listener;
96+
if (request.getDeleteInterveningResults()) {
97+
wrappedListener = wrapDeleteOldDataListener(wrappedListener, modelSnapshot, request.getJobId());
98+
wrappedListener = wrapRevertDataCountsListener(wrappedListener, modelSnapshot, request.getJobId());
8999
}
100+
jobManager.revertSnapshot(request, wrappedListener, modelSnapshot);
101+
}, listener::onFailure);
102+
},
103+
listener::onFailure
104+
);
105+
106+
107+
// 2. Verify the job exists
108+
ActionListener<Boolean> createStateIndexListener = ActionListener.wrap(
109+
r -> jobManager.jobExists(request.getJobId(), jobExistsListener),
110+
listener::onFailure
111+
);
90112

91-
getModelSnapshot(request, jobResultsProvider, modelSnapshot -> {
92-
ActionListener<RevertModelSnapshotAction.Response> wrappedListener = listener;
93-
if (request.getDeleteInterveningResults()) {
94-
wrappedListener = wrapDeleteOldDataListener(wrappedListener, modelSnapshot, request.getJobId());
95-
wrappedListener = wrapRevertDataCountsListener(wrappedListener, modelSnapshot, request.getJobId());
96-
}
97-
jobManager.revertSnapshot(request, wrappedListener, modelSnapshot);
98-
}, listener::onFailure);
99-
},
100-
listener::onFailure
101-
));
113+
// 1. Verify/Create the state index and its alias exists
114+
AnomalyDetectorsIndex.createStateIndexAndAliasIfNecessary(client, state, createStateIndexListener);
102115
}
103116

104117
private void getModelSnapshot(RevertModelSnapshotAction.Request request, JobResultsProvider provider, Consumer<ModelSnapshot> handler,

0 commit comments

Comments
 (0)