Prefer index config documents to cluster state config

Author: davidkyle

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

@@ -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) {

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

@@ -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;
@@ -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,

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

@@ -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;
@@ -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) {
@@ -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);
+                            }
                     ));
         };
 

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedConfigReader.java

@@ -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;
@@ -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);
+                }
+        ));
     }
 
     /**
@@ -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 {
@@ -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));
+                        }
+                    }
+
                     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);
                     }