Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ML: creating ML State write alias and pointing writes there #37483

Merged
merged 10 commits into from
Jan 18, 2019
Merged

ML: creating ML State write alias and pointing writes there #37483

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
9ccd34e
ML: creating ML State write alias and pointing writes there
benwtrent Jan 15, 2019
a98b2d3
style fix
benwtrent Jan 15, 2019
e194d8e
Moving alias check to openJob method
benwtrent Jan 15, 2019
a8bbd2e
Addressing PR comments
benwtrent Jan 16, 2019
5754435
Merge branch 'master' into feature/ml-add-state-write-alias
benwtrent Jan 16, 2019
694bc86
updating for master merge
benwtrent Jan 16, 2019
77749e6
Fixing tests after master merge
benwtrent Jan 16, 2019
3974ec9
adjusting concrete index lookup for ml-state
benwtrent Jan 17, 2019
eb50b03
Merge branch 'master' into feature/ml-add-state-write-alias
benwtrent Jan 17, 2019
bf23c35
Merge branch 'master' into feature/ml-add-state-write-alias
benwtrent Jan 17, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 80 additions & 4 deletions .../src/main/java/org/elasticsearch/xpack/core/ml/job/persistence/AnomalyDetectorsIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,22 @@
*/
package org.elasticsearch.xpack.core.ml.job.persistence;

import org.elasticsearch.ResourceAlreadyExistsException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.AliasOrIndex;

import java.util.SortedMap;

import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;

/**
* Methods for handling index naming related functions
*/
Expand Down Expand Up @@ -40,11 +56,11 @@ public static String resultsWriteAlias(String jobId) {
}

/**
* The name of the default index where a job's state is stored
* @return The index name
* The name of the alias pointing to the appropriate index for writing job state
* @return The write alias name
*/
public static String jobStateIndexName() {
return AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX;
public static String jobStateIndexWriteAlias() {
return AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX + "-write";
}

/**
Expand All @@ -64,4 +80,64 @@ public static String configIndexName() {
return AnomalyDetectorsIndexFields.CONFIG_INDEX;
}

/**
* Create the .ml-state index (if necessary)
* Create the .ml-state-write alias for the .ml-state index (if necessary)
*/
public static void createStateIndexAndAliasIfNecessary(Client client, ClusterState state, final ActionListener<Boolean> finalListener) {

if (state.getMetaData().getAliasAndIndexLookup().containsKey(jobStateIndexWriteAlias())) {
finalListener.onResponse(false);
return;
}

final ActionListener<Void> createAliasListener = ActionListener.wrap(
r -> {
final IndicesAliasesRequest request = client.admin()
.indices()
.prepareAliases()
.addAlias(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX, jobStateIndexWriteAlias())
.request();
executeAsyncWithOrigin(client.threadPool().getThreadContext(),
ML_ORIGIN,
request,
ActionListener.<AcknowledgedResponse>wrap(
resp -> finalListener.onResponse(resp.isAcknowledged()),
finalListener::onFailure),
client.admin().indices()::aliases);
},
finalListener::onFailure
);

// Only create the index or aliases if some other ML index exists - saves clutter if ML is never used.
SortedMap<String, AliasOrIndex> mlLookup = state.getMetaData().getAliasAndIndexLookup().tailMap(".ml");
if (mlLookup.isEmpty() == false && mlLookup.firstKey().startsWith(".ml")) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We probably shouldn't do the clutter avoidance in this case. If this method is called we know we're intending to write to an ML index shortly even if we never have up to this time. The effect of not creating the alias is pretty bad - it results in creation of a concrete index called .ml-state-write, and that is hard to switch over to an alias of the same name.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Definitely, I agree. Additionally, This method should look for the concrete indices that match the prefix .ml-state*. Should be easy enough to adjust.

if (mlLookup.containsKey(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX)) {
createAliasListener.onResponse(null);
} else {
CreateIndexRequest createIndexRequest = client.admin()
.indices()
.prepareCreate(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX)
.addAlias(new Alias(jobStateIndexWriteAlias()))
.request();
executeAsyncWithOrigin(client.threadPool().getThreadContext(),
ML_ORIGIN,
createIndexRequest,
ActionListener.<CreateIndexResponse>wrap(
createIndexResponse -> finalListener.onResponse(true),
createIndexFailure -> {
// If it was created between our last check, and this request being handled, we should add the alias
// Adding an alias that already exists is idempotent, so, no need to double check if the alias exists
// as well.
if (createIndexFailure instanceof ResourceAlreadyExistsException) {
createAliasListener.onResponse(null);
} else {
finalListener.onFailure(createIndexFailure);
}
}),
client.admin().indices()::create);
}
}
}

}
5 changes: 2 additions & 3 deletions .../test/java/org/elasticsearch/xpack/core/security/authz/store/ReservedRolesStoreTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,6 @@
import org.elasticsearch.xpack.core.ml.action.UpdateProcessAction;
import org.elasticsearch.xpack.core.ml.action.ValidateDetectorAction;
import org.elasticsearch.xpack.core.ml.action.ValidateJobConfigAction;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
import org.elasticsearch.xpack.core.ml.notifications.AuditorField;
import org.elasticsearch.xpack.core.monitoring.action.MonitoringBulkAction;
Expand Down Expand Up @@ -762,7 +761,7 @@ public void testMachineLearningAdminRole() {

assertNoAccessAllowed(role, "foo");
assertOnlyReadAllowed(role, MlMetaIndex.INDEX_NAME);
assertOnlyReadAllowed(role, AnomalyDetectorsIndex.jobStateIndexName());
assertOnlyReadAllowed(role, AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX);
assertOnlyReadAllowed(role, AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + AnomalyDetectorsIndexFields.RESULTS_INDEX_DEFAULT);
assertOnlyReadAllowed(role, AuditorField.NOTIFICATIONS_INDEX);
}
Expand Down Expand Up @@ -814,7 +813,7 @@ public void testMachineLearningUserRole() {

assertNoAccessAllowed(role, "foo");
assertNoAccessAllowed(role, MlMetaIndex.INDEX_NAME);
assertNoAccessAllowed(role, AnomalyDetectorsIndex.jobStateIndexName());
assertNoAccessAllowed(role, AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX);
assertOnlyReadAllowed(role, AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + AnomalyDetectorsIndexFields.RESULTS_INDEX_DEFAULT);
assertOnlyReadAllowed(role, AuditorField.NOTIFICATIONS_INDEX);
}
Expand Down
5 changes: 3 additions & 2 deletions x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/test/rest/XPackRestTestHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.xpack.core.ml.MlMetaIndex;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
import org.elasticsearch.xpack.core.ml.notifications.AuditorField;

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

public static final List<String> ML_POST_V660_TEMPLATES = Collections.unmodifiableList(
Arrays.asList(AuditorField.NOTIFICATIONS_INDEX,
MlMetaIndex.INDEX_NAME,
AnomalyDetectorsIndex.jobStateIndexName(),
AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX,
AnomalyDetectorsIndex.jobResultsIndexPrefix(),
AnomalyDetectorsIndex.configIndexName()));

Expand Down
2 changes: 1 addition & 1 deletion ...-node-tests/src/test/java/org/elasticsearch/xpack/ml/integration/DeleteExpiredDataIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,7 @@ public void testDeleteExpiredData() throws Exception {
bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
for (int i = 0; i < 10010; i++) {
String docId = "non_existing_job_" + randomFrom("model_state_1234567#" + i, "quantiles", "categorizer_state#" + i);
IndexRequest indexRequest = new IndexRequest(AnomalyDetectorsIndex.jobStateIndexName(), "doc", docId);
IndexRequest indexRequest = new IndexRequest(AnomalyDetectorsIndex.jobStateIndexWriteAlias(), "doc", docId);
indexRequest.source(Collections.emptyMap());
bulkRequestBuilder.add(indexRequest);
}
Expand Down
9 changes: 5 additions & 4 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@
import org.elasticsearch.xpack.core.ml.action.ValidateDetectorAction;
import org.elasticsearch.xpack.core.ml.action.ValidateJobConfigAction;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
import org.elasticsearch.xpack.core.ml.job.persistence.ElasticsearchMappings;
import org.elasticsearch.xpack.core.ml.notifications.AuditMessage;
import org.elasticsearch.xpack.core.ml.notifications.AuditorField;
Expand Down Expand Up @@ -701,7 +702,7 @@ public UnaryOperator<Map<String, IndexTemplateMetaData>> getIndexTemplateMetaDat
}

try (XContentBuilder stateMapping = ElasticsearchMappings.stateMapping()) {
IndexTemplateMetaData stateTemplate = IndexTemplateMetaData.builder(AnomalyDetectorsIndex.jobStateIndexName())
IndexTemplateMetaData stateTemplate = IndexTemplateMetaData.builder(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX)
.patterns(Collections.singletonList(AnomalyDetectorsIndex.jobStateIndexPattern()))
// TODO review these settings
.settings(Settings.builder()
Expand All @@ -710,9 +711,9 @@ public UnaryOperator<Map<String, IndexTemplateMetaData>> getIndexTemplateMetaDat
.putMapping(ElasticsearchMappings.DOC_TYPE, Strings.toString(stateMapping))
.version(Version.CURRENT.id)
.build();
templates.put(AnomalyDetectorsIndex.jobStateIndexName(), stateTemplate);
templates.put(AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX, stateTemplate);
} catch (IOException e) {
logger.error("Error loading the template for the " + AnomalyDetectorsIndex.jobStateIndexName() + " index", e);
logger.error("Error loading the template for the " + AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX + " index", e);
}

try (XContentBuilder docMapping = ElasticsearchMappings.resultsMapping()) {
Expand Down Expand Up @@ -742,7 +743,7 @@ public UnaryOperator<Map<String, IndexTemplateMetaData>> getIndexTemplateMetaDat
public static boolean allTemplatesInstalled(ClusterState clusterState) {
boolean allPresent = true;
List<String> templateNames = Arrays.asList(AuditorField.NOTIFICATIONS_INDEX, MlMetaIndex.INDEX_NAME,
AnomalyDetectorsIndex.jobStateIndexName(), AnomalyDetectorsIndex.jobResultsIndexPrefix());
AnomalyDetectorsIndexFields.STATE_INDEX_PREFIX, AnomalyDetectorsIndex.jobResultsIndexPrefix());
for (String templateName : templateNames) {
allPresent = allPresent && TemplateUtils.checkTemplateExistsAndVersionIsGTECurrentVersion(templateName, clusterState);
}
Expand Down
2 changes: 1 addition & 1 deletion x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlConfigMigrator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -347,7 +347,7 @@ public void snapshotMlMeta(MlMetadata mlMetadata, ActionListener<Boolean> listen

logger.debug("taking a snapshot of ml_metadata");
String documentId = "ml-config";
IndexRequestBuilder indexRequest = client.prepareIndex(AnomalyDetectorsIndex.jobStateIndexName(),
IndexRequestBuilder indexRequest = client.prepareIndex(AnomalyDetectorsIndex.jobStateIndexWriteAlias(),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's no guarantee that an autodetect process will have been started on the newer version of the product at the point when this call is made - if all ML jobs are closed prior to upgrading from 6.5 to 6.7 then that will definitely trigger this situation. So this method needs to call AnomalyDetectorsIndex.createStateIndexAndAliasIfNecessary() first.

ElasticsearchMappings.DOC_TYPE, documentId)
.setOpType(DocWriteRequest.OpType.CREATE);

Expand Down
6 changes: 4 additions & 2 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportOpenJobAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -529,7 +529,7 @@ public void onFailure(Exception e) {
// Try adding state doc mapping
ActionListener<Boolean> resultsPutMappingHandler = ActionListener.wrap(
response -> {
addDocMappingIfMissing(AnomalyDetectorsIndex.jobStateIndexName(), ElasticsearchMappings::stateMapping,
addDocMappingIfMissing(AnomalyDetectorsIndex.jobStateIndexWriteAlias(), ElasticsearchMappings::stateMapping,
state, jobUpdateListener);
}, listener::onFailure
);
Expand Down Expand Up @@ -673,6 +673,7 @@ public static class OpenJobPersistentTasksExecutor extends PersistentTasksExecut
private volatile int maxConcurrentJobAllocations;
private volatile int maxMachineMemoryPercent;
private volatile int maxLazyMLNodes;
private volatile ClusterState clusterState;

public OpenJobPersistentTasksExecutor(Settings settings, ClusterService clusterService,
AutodetectProcessManager autodetectProcessManager, MlMemoryTracker memoryTracker,
Expand All @@ -689,6 +690,7 @@ public OpenJobPersistentTasksExecutor(Settings settings, ClusterService clusterS
clusterService.getClusterSettings()
.addSettingsUpdateConsumer(MachineLearning.MAX_MACHINE_MEMORY_PERCENT, this::setMaxMachineMemoryPercent);
clusterService.getClusterSettings().addSettingsUpdateConsumer(MachineLearning.MAX_LAZY_ML_NODES, this::setMaxLazyMLNodes);
clusterService.addListener(event -> clusterState = event.state());
}

@Override
Expand Down Expand Up @@ -741,7 +743,7 @@ protected void nodeOperation(AllocatedPersistentTask task, OpenJobAction.JobPara
}

String jobId = jobTask.getJobId();
autodetectProcessManager.openJob(jobTask, e2 -> {
autodetectProcessManager.openJob(jobTask, clusterState, e2 -> {
if (e2 == null) {
FinalizeJobExecutionAction.Request finalizeRequest = new FinalizeJobExecutionAction.Request(new String[]{jobId});
executeAsyncWithOrigin(client, ML_ORIGIN, FinalizeJobExecutionAction.INSTANCE, finalizeRequest,
Expand Down
4 changes: 2 additions & 2 deletions ...ugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobResultsPersister.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,7 @@ public void persistCategoryDefinition(CategoryDefinition category) {
*/
public void persistQuantiles(Quantiles quantiles) {
Persistable persistable = new Persistable(quantiles.getJobId(), quantiles, Quantiles.documentId(quantiles.getJobId()));
persistable.persist(AnomalyDetectorsIndex.jobStateIndexName()).actionGet();
persistable.persist(AnomalyDetectorsIndex.jobStateIndexWriteAlias()).actionGet();
}

/**
Expand All @@ -237,7 +237,7 @@ public void persistQuantiles(Quantiles quantiles) {
public void persistQuantiles(Quantiles quantiles, WriteRequest.RefreshPolicy refreshPolicy, ActionListener<IndexResponse> listener) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method can be called when reverting a model snapshot, and there's no guarantee that an autodetect process will have been started on the newer version of the product at the point when a model snapshot is reverted. The call chain is TransportRevertModelSnapshotAction.masterOperation() -> JobManager.revertSnapshot() -> this method. So one of those two calls needs to call AnomalyDetectorsIndex.createStateIndexAndAliasIfNecessary() first.

Persistable persistable = new Persistable(quantiles.getJobId(), quantiles, Quantiles.documentId(quantiles.getJobId()));
persistable.setRefreshPolicy(refreshPolicy);
persistable.persist(AnomalyDetectorsIndex.jobStateIndexName(), listener);
persistable.persist(AnomalyDetectorsIndex.jobStateIndexWriteAlias(), listener);
}

/**
Expand Down
Loading