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

Merged
merged 4 commits into from
Nov 28, 2018
Merged

[ML] Jindex: Prefer index config documents to cluster state config #35940

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
417624f
Prefer index config documents to cluster state config
davidkyle Nov 26, 2018
a7c78ee
Fix delete job
davidkyle Nov 27, 2018
833e1e6
Simplify detecting duplicate configs
davidkyle Nov 27, 2018
925fae4
Handle index not found errors
davidkyle Nov 27, 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: 15 additions & 9 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,15 +131,21 @@ private void deleteDatafeedConfig(DeleteDatafeedAction.Request request, ClusterS
return;
}

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
));
}

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);
}
));
}

private void deleteDatafeedFromMetadata(DeleteDatafeedAction.Request request, ActionListener<AcknowledgedResponse> listener) {
Expand Down
81 changes: 44 additions & 37 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,6 +11,7 @@
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 @@ -692,47 +693,53 @@ public void onTimeout(TimeValue timeout) {

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

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();
}
JobUpdate update = new JobUpdate.Builder(jobId).setClearFinishTime(true).build();

@Override
public void onFailure(String source, Exception e) {
logger.error("[" + jobId + "] Failed to clear finished_time; source [" + source + "]", e);
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
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 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));
}
});
}

private void cancelJobStart(PersistentTasksCustomMetaData.PersistentTask<OpenJobAction.JobParams> persistentTask, Exception exception,
Expand Down
32 changes: 18 additions & 14 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,6 +5,7 @@
*/
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 @@ -69,19 +70,6 @@ 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 @@ -91,14 +79,30 @@ private void updateDatafeedInIndex(UpdateDatafeedAction.Request request, Cluster
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)),
listener::onFailure
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);
}
));
};

Expand Down
69 changes: 34 additions & 35 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,6 +5,7 @@
*/
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 @@ -50,17 +51,22 @@ public DatafeedConfigReader(DatafeedConfigProvider datafeedConfigProvider) {
* @param listener DatafeedConfig listener
*/
public void datafeedConfig(String datafeedId, ClusterState state, ActionListener<DatafeedConfig> listener) {
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
));
}

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);
}
));
}

/**
Expand All @@ -70,23 +76,16 @@ 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 -> {
// Check for duplicate Ids
expandedDatafeedIds.forEach(id -> {
if (clusterStateDatafeedIds.contains(id)) {
listener.onFailure(new IllegalStateException("Datafeed [" + id + "] configuration " +
"exists in both clusterstate and index"));
return;
}
});

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 @@ -105,33 +104,33 @@ 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 -> {
// Check for duplicate Ids
datafeedBuilders.forEach(datafeedBuilder -> {
if (clusterStateConfigs.containsKey(datafeedBuilder.getId())) {
listener.onFailure(new IllegalStateException("Datafeed [" + datafeedBuilder.getId() + "] configuration " +
"exists in both clusterstate and index"));
return;
}
});

List<DatafeedConfig> datafeedConfigs = new ArrayList<>();
for (DatafeedConfig.Builder builder : datafeedBuilders) {
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.
for (String clusterStateDatafeedId : clusterStateConfigs.keySet()) {
boolean isDuplicate = datafeedConfigs.stream()
.anyMatch(datafeed -> datafeed.getId().equals(clusterStateDatafeedId));
if (isDuplicate == false) {
datafeedConfigs.add(clusterStateConfigs.get(clusterStateDatafeedId));
}
}
Copy link
Member

@benwtrent benwtrent Nov 27, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

One thing is sort of nagging me with this code:

It seems to me that if we are checking Collection Membership all the time, that it should be a Set, though these collections may be so small, that this sort of optimization is not necessary and may result in worse performance. So, this is a suggestion you are free to ignore if these are adequately small collections.

Set<String> indexConfigIds = datafeedConfigs.stream().map(DatafeedConfig::getId).collect(Collectors.toSet());
...
if(indexConfigIds.contains(clusterStateDatafeedId) == false)

Copy link
Member Author

Choose a reason for hiding this comment

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

I agonised over this for ages with the same nagging feeling. The code looks too complicated but I was erring on the side that the collection would be small making any optimisation premature. I'll make your change as it's more readable

Copy link
Member

Choose a reason for hiding this comment

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

:D, glad to see we agonized over the same thing


requiredMatches.filterMatchedIds(datafeedConfigs.stream().map(DatafeedConfig::getId).collect(Collectors.toList()));

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