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] Prefer cluster state config to index documents #36014

Merged
merged 4 commits into from
Nov 29, 2018
Merged

[ML] Prefer cluster state config to index documents #36014

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a54cef8
Revert "Prefer index config documents to cluster state config"
davidkyle Nov 28, 2018
a893ef7
Reinstate reverted tests that are valid
davidkyle Nov 29, 2018
61f2ae2
Fix finalization and test
davidkyle Nov 29, 2018
38b996a
Check cluster state config for running datafeed
davidkyle Nov 29, 2018
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
24 changes: 9 additions & 15 deletions ...gin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteDatafeedAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,21 +131,15 @@ private void deleteDatafeedConfig(DeleteDatafeedAction.Request request, ClusterS
return;
}


datafeedConfigProvider.deleteDatafeedConfig(request.getDatafeedId(), ActionListener.wrap(
deleteResponse -> listener.onResponse(new AcknowledgedResponse(true)),
e -> {
if (e.getClass() == ResourceNotFoundException.class) {
// is the datafeed in the clusterstate
MlMetadata mlMetadata = MlMetadata.getMlMetadata(state);
if (mlMetadata.getDatafeed(request.getDatafeedId()) != null) {
deleteDatafeedFromMetadata(request, listener);
return;
}
}
listener.onFailure(e);
}
));
MlMetadata mlMetadata = MlMetadata.getMlMetadata(state);
if (mlMetadata.getDatafeed(request.getDatafeedId()) != null) {
deleteDatafeedFromMetadata(request, listener);
} else {
datafeedConfigProvider.deleteDatafeedConfig(request.getDatafeedId(), ActionListener.wrap(
deleteResponse -> listener.onResponse(new AcknowledgedResponse(true)),
listener::onFailure
));
}
}

private void deleteDatafeedFromMetadata(DeleteDatafeedAction.Request request, ActionListener<AcknowledgedResponse> listener) {
Expand Down
81 changes: 37 additions & 44 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 @@ -11,7 +11,6 @@
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.ResourceAlreadyExistsException;
import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingAction;
Expand Down Expand Up @@ -693,53 +692,47 @@ public void onTimeout(TimeValue timeout) {

private void clearJobFinishedTime(String jobId, ActionListener<AcknowledgedResponse> listener) {

JobUpdate update = new JobUpdate.Builder(jobId).setClearFinishTime(true).build();
boolean jobIsInClusterState = ClusterStateJobUpdate.jobIsInClusterState(clusterService.state(), jobId);
if (jobIsInClusterState) {
clusterService.submitStateUpdateTask("clearing-job-finish-time-for-" + jobId, new ClusterStateUpdateTask() {
@Override
public ClusterState execute(ClusterState currentState) {
MlMetadata mlMetadata = MlMetadata.getMlMetadata(currentState);
MlMetadata.Builder mlMetadataBuilder = new MlMetadata.Builder(mlMetadata);
Job.Builder jobBuilder = new Job.Builder(mlMetadata.getJobs().get(jobId));
jobBuilder.setFinishedTime(null);

mlMetadataBuilder.putJob(jobBuilder.build(), true);
ClusterState.Builder builder = ClusterState.builder(currentState);
return builder.metaData(new MetaData.Builder(currentState.metaData())
.putCustom(MlMetadata.TYPE, mlMetadataBuilder.build()))
.build();
}

jobConfigProvider.updateJob(jobId, update, null, ActionListener.wrap(
job -> listener.onResponse(new AcknowledgedResponse(true)),
e -> {
if (e.getClass() == ResourceNotFoundException.class) {
// Maybe the config is in the clusterstate
if (ClusterStateJobUpdate.jobIsInClusterState(clusterService.state(), jobId)) {
clearJobFinishedTimeClusterState(jobId, listener);
return;
}
}
logger.error("[" + jobId + "] Failed to clear finished_time", e);
// Not a critical error so continue
@Override
public void onFailure(String source, Exception e) {
logger.error("[" + jobId + "] Failed to clear finished_time; source [" + source + "]", e);
listener.onResponse(new AcknowledgedResponse(true));
}
));
}

private void clearJobFinishedTimeClusterState(String jobId, ActionListener<AcknowledgedResponse> listener) {
clusterService.submitStateUpdateTask("clearing-job-finish-time-for-" + jobId, new ClusterStateUpdateTask() {
@Override
public ClusterState execute(ClusterState currentState) {
MlMetadata mlMetadata = MlMetadata.getMlMetadata(currentState);
MlMetadata.Builder mlMetadataBuilder = new MlMetadata.Builder(mlMetadata);
Job.Builder jobBuilder = new Job.Builder(mlMetadata.getJobs().get(jobId));
jobBuilder.setFinishedTime(null);

mlMetadataBuilder.putJob(jobBuilder.build(), true);
ClusterState.Builder builder = ClusterState.builder(currentState);
return builder.metaData(new MetaData.Builder(currentState.metaData())
.putCustom(MlMetadata.TYPE, mlMetadataBuilder.build()))
.build();
}

@Override
public void onFailure(String source, Exception e) {
logger.error("[" + jobId + "] Failed to clear finished_time; source [" + source + "]", e);
listener.onResponse(new AcknowledgedResponse(true));
}

@Override
public void clusterStateProcessed(String source, ClusterState oldState,
ClusterState newState) {
listener.onResponse(new AcknowledgedResponse(true));
}
});
@Override
public void clusterStateProcessed(String source, ClusterState oldState,
ClusterState newState) {
listener.onResponse(new AcknowledgedResponse(true));
}
});
} else {
JobUpdate update = new JobUpdate.Builder(jobId).setClearFinishTime(true).build();

jobConfigProvider.updateJob(jobId, update, null, ActionListener.wrap(
job -> listener.onResponse(new AcknowledgedResponse(true)),
e -> {
logger.error("[" + jobId + "] Failed to clear finished_time", e);
// Not a critical error so continue
listener.onResponse(new AcknowledgedResponse(true));
}
));
}
}

private void cancelJobStart(PersistentTasksCustomMetaData.PersistentTask<OpenJobAction.JobParams> persistentTask, Exception exception,
Expand Down
32 changes: 14 additions & 18 deletions ...gin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateDatafeedAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
*/
package org.elasticsearch.xpack.ml.action;

import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
Expand Down Expand Up @@ -70,6 +69,19 @@ protected PutDatafeedAction.Response newResponse() {
protected void masterOperation(UpdateDatafeedAction.Request request, ClusterState state,
ActionListener<PutDatafeedAction.Response> listener) throws Exception {

MlMetadata mlMetadata = MlMetadata.getMlMetadata(state);
boolean datafeedConfigIsInClusterState = mlMetadata.getDatafeed(request.getUpdate().getId()) != null;
if (datafeedConfigIsInClusterState) {
updateDatafeedInClusterState(request, listener);
} else {
updateDatafeedInIndex(request, state, listener);
}
}

private void updateDatafeedInIndex(UpdateDatafeedAction.Request request, ClusterState state,
ActionListener<PutDatafeedAction.Response> listener) throws Exception {
final Map<String, String> headers = threadPool.getThreadContext().getHeaders();

// Check datafeed is stopped
PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
if (MlTasks.getDatafeedTask(request.getUpdate().getId(), tasks) != null) {
Expand All @@ -79,30 +91,14 @@ protected void masterOperation(UpdateDatafeedAction.Request request, ClusterStat
return;
}

updateDatafeedInIndex(request, state, listener);
}

private void updateDatafeedInIndex(UpdateDatafeedAction.Request request, ClusterState state,
ActionListener<PutDatafeedAction.Response> listener) throws Exception {
final Map<String, String> headers = threadPool.getThreadContext().getHeaders();
String datafeedId = request.getUpdate().getId();

CheckedConsumer<Boolean, Exception> updateConsumer = ok -> {
datafeedConfigProvider.updateDatefeedConfig(request.getUpdate().getId(), request.getUpdate(), headers,
jobConfigProvider::validateDatafeedJob,
ActionListener.wrap(
updatedConfig -> listener.onResponse(new PutDatafeedAction.Response(updatedConfig)),
e -> {
if (e.getClass() == ResourceNotFoundException.class) {
// try the clusterstate
MlMetadata mlMetadata = MlMetadata.getMlMetadata(state);
if (mlMetadata.getDatafeed(request.getUpdate().getId()) != null) {
updateDatafeedInClusterState(request, listener);
return;
}
}
listener.onFailure(e);
}
listener::onFailure
));
};

Expand Down
42 changes: 18 additions & 24 deletions x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedConfigReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
*/
package org.elasticsearch.xpack.ml.datafeed;

import org.elasticsearch.ResourceNotFoundException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterState;
Expand Down Expand Up @@ -51,22 +50,17 @@ public DatafeedConfigReader(DatafeedConfigProvider datafeedConfigProvider) {
* @param listener DatafeedConfig listener
*/
public void datafeedConfig(String datafeedId, ClusterState state, ActionListener<DatafeedConfig> listener) {

datafeedConfigProvider.getDatafeedConfig(datafeedId, ActionListener.wrap(
builder -> listener.onResponse(builder.build()),
e -> {
if (e.getClass() == ResourceNotFoundException.class) {
// look in the clusterstate
MlMetadata mlMetadata = MlMetadata.getMlMetadata(state);
DatafeedConfig config = mlMetadata.getDatafeed(datafeedId);
if (config != null) {
listener.onResponse(config);
return;
}
}
listener.onFailure(e);
}
));
MlMetadata mlMetadata = MlMetadata.getMlMetadata(state);
DatafeedConfig config = mlMetadata.getDatafeed(datafeedId);

if (config != null) {
listener.onResponse(config);
} else {
datafeedConfigProvider.getDatafeedConfig(datafeedId, ActionListener.wrap(
builder -> listener.onResponse(builder.build()),
listener::onFailure
));
}
}

/**
Expand All @@ -76,16 +70,14 @@ public void datafeedConfig(String datafeedId, ClusterState state, ActionListener
public void expandDatafeedIds(String expression, boolean allowNoDatafeeds, ClusterState clusterState,
ActionListener<SortedSet<String>> listener) {

Set<String> clusterStateDatafeedIds = MlMetadata.getMlMetadata(clusterState).expandDatafeedIds(expression);
ExpandedIdsMatcher requiredMatches = new ExpandedIdsMatcher(expression, allowNoDatafeeds);
requiredMatches.filterMatchedIds(clusterStateDatafeedIds);

datafeedConfigProvider.expandDatafeedIdsWithoutMissingCheck(expression, ActionListener.wrap(
expandedDatafeedIds -> {
requiredMatches.filterMatchedIds(expandedDatafeedIds);

// now read from the clusterstate
Set<String> clusterStateDatafeedIds = MlMetadata.getMlMetadata(clusterState).expandDatafeedIds(expression);
requiredMatches.filterMatchedIds(clusterStateDatafeedIds);

if (requiredMatches.hasUnmatchedIds()) {
listener.onFailure(ExceptionsHelper.missingDatafeedException(requiredMatches.unmatchedIdsString()));
} else {
Expand All @@ -104,7 +96,10 @@ public void expandDatafeedIds(String expression, boolean allowNoDatafeeds, Clust
public void expandDatafeedConfigs(String expression, boolean allowNoDatafeeds, ClusterState clusterState,
ActionListener<List<DatafeedConfig>> listener) {

Map<String, DatafeedConfig> clusterStateConfigs = expandClusterStateDatafeeds(expression, clusterState);

ExpandedIdsMatcher requiredMatches = new ExpandedIdsMatcher(expression, allowNoDatafeeds);
requiredMatches.filterMatchedIds(clusterStateConfigs.keySet());

datafeedConfigProvider.expandDatafeedConfigsWithoutMissingCheck(expression, ActionListener.wrap(
datafeedBuilders -> {
Expand All @@ -113,11 +108,9 @@ public void expandDatafeedConfigs(String expression, boolean allowNoDatafeeds, C
datafeedConfigs.add(builder.build());
}

Map<String, DatafeedConfig> clusterStateConfigs = expandClusterStateDatafeeds(expression, clusterState);

// Duplicate configs existing in both the clusterstate and index documents are ok
// this may occur during migration of configs.
// Prefer the index configs and filter duplicates from the clusterstate configs.
// Prefer the clusterstate configs and filter duplicates from the index
Set<String> indexConfigIds = datafeedConfigs.stream().map(DatafeedConfig::getId).collect(Collectors.toSet());
for (String clusterStateDatafeedId : clusterStateConfigs.keySet()) {
if (indexConfigIds.contains(clusterStateDatafeedId) == false) {
Expand All @@ -130,6 +123,7 @@ public void expandDatafeedConfigs(String expression, boolean allowNoDatafeeds, C
if (requiredMatches.hasUnmatchedIds()) {
listener.onFailure(ExceptionsHelper.missingDatafeedException(requiredMatches.unmatchedIdsString()));
} else {
datafeedConfigs.addAll(clusterStateConfigs.values());
Collections.sort(datafeedConfigs, Comparator.comparing(DatafeedConfig::getId));
listener.onResponse(datafeedConfigs);
}
Expand Down
Loading