[ML] Resolve 7.0.0 TODOs in ML code (#36842)

This change cleans up a number of ugly BWC
workarounds in the ML code.

7.0 cannot run in a mixed version cluster with
versions prior to 6.7, so code that deals with
these old versions is no longer required.

Closes #29963

Author: David Roberts

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

@@ -39,7 +39,6 @@ public class OpenJobAction extends Action<AcknowledgedResponse> {
     public static final OpenJobAction INSTANCE = new OpenJobAction();
     public static final String NAME = "cluster:admin/xpack/ml/job/open";
 
-
     private OpenJobAction() {
         super(NAME);
     }
@@ -132,15 +131,12 @@ public String toString() {
 
     public static class JobParams implements XPackPlugin.XPackPersistentTaskParams {
 
-        /** TODO Remove in 7.0.0 */
-        public static final ParseField IGNORE_DOWNTIME = new ParseField("ignore_downtime");
         public static final ParseField TIMEOUT = new ParseField("timeout");
         public static final ParseField JOB = new ParseField("job");
 
         public static ObjectParser<JobParams, Void> PARSER = new ObjectParser<>(MlTasks.JOB_TASK_NAME, true, JobParams::new);
         static {
             PARSER.declareString(JobParams::setJobId, Job.ID);
-            PARSER.declareBoolean((p, v) -> {}, IGNORE_DOWNTIME);
             PARSER.declareString((params, val) ->
                     params.setTimeout(TimeValue.parseTimeValue(val, TIMEOUT.getPreferredName())), TIMEOUT);
             PARSER.declareObject(JobParams::setJob, (p, c) -> Job.LENIENT_PARSER.apply(p, c).build(), JOB);

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

@@ -130,7 +130,6 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
     public static final ParseField AGGREGATIONS = new ParseField("aggregations");
     public static final ParseField AGGS = new ParseField("aggs");
     public static final ParseField SCRIPT_FIELDS = new ParseField("script_fields");
-    public static final ParseField SOURCE = new ParseField("_source");
     public static final ParseField CHUNKING_CONFIG = new ParseField("chunking_config");
     public static final ParseField HEADERS = new ParseField("headers");
     public static final ParseField DELAYED_DATA_CHECK_CONFIG = new ParseField("delayed_data_check_config");
@@ -185,9 +184,6 @@ private static ObjectParser<Builder, Void> createParser(boolean ignoreUnknownFie
             return parsedScriptFields;
         }, SCRIPT_FIELDS);
         parser.declareInt(Builder::setScrollSize, SCROLL_SIZE);
-        // TODO this is to read former _source field. Remove in v7.0.0
-        parser.declareBoolean((builder, value) -> {
-        }, SOURCE);
         parser.declareObject(Builder::setChunkingConfig, ignoreUnknownFields ? ChunkingConfig.LENIENT_PARSER : ChunkingConfig.STRICT_PARSER,
             CHUNKING_CONFIG);
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/job/config/AnalysisConfig.java

@@ -136,30 +136,7 @@ public AnalysisConfig(StreamInput in) throws IOException {
         detectors = Collections.unmodifiableList(in.readList(Detector::new));
         influencers = Collections.unmodifiableList(in.readList(StreamInput::readString));
 
-        // BWC for result_finalization_window and overlapping_buckets
-        // TODO Remove in 7.0.0
-        if (in.getVersion().before(Version.V_6_6_0)) {
-            in.readOptionalBoolean();
-            in.readOptionalLong();
-        }
         multivariateByFields = in.readOptionalBoolean();
-
-        // BWC for removed multiple_bucket_spans
-        // TODO Remove in 7.0.0
-        if (in.getVersion().before(Version.V_6_5_0)) {
-            if (in.readBoolean()) {
-                final int arraySize = in.readVInt();
-                for (int i = 0; i < arraySize; i++) {
-                    in.readTimeValue();
-                }
-            }
-        }
-
-        // BWC for removed per-partition normalization
-        // TODO Remove in 7.0.0
-        if (in.getVersion().before(Version.V_6_5_0)) {
-            in.readBoolean();
-        }
     }
 
     @Override
@@ -180,25 +157,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeList(detectors);
         out.writeStringList(influencers);
 
-        // BWC for result_finalization_window and overlapping_buckets
-        // TODO Remove in 7.0.0
-        if (out.getVersion().before(Version.V_6_6_0)) {
-            out.writeOptionalBoolean(null);
-            out.writeOptionalLong(null);
-        }
         out.writeOptionalBoolean(multivariateByFields);
-
-        // BWC for removed multiple_bucket_spans
-        // TODO Remove in 7.0.0
-        if (out.getVersion().before(Version.V_6_5_0)) {
-            out.writeBoolean(false);
-        }
-
-        // BWC for removed per-partition normalization
-        // TODO Remove in 7.0.0
-        if (out.getVersion().before(Version.V_6_5_0)) {
-            out.writeBoolean(false);
-        }
     }
 
     /**

x-pack/plugin/ml/qa/basic-multi-node/src/test/java/org/elasticsearch/xpack/ml/integration/MlBasicMultiNodeIT.java

@@ -266,7 +266,6 @@ private Response createDatafeed(String datafeedId, String jobId) throws Exceptio
         xContentBuilder.field("job_id", jobId);
         xContentBuilder.array("indexes", "airline-data");
         xContentBuilder.array("types", "_doc");
-        xContentBuilder.field("_source", true);
         xContentBuilder.endObject();
         Request request = new Request("PUT", MachineLearning.BASE_PATH + "datafeeds/" + datafeedId);
         request.setJsonEntity(Strings.toString(xContentBuilder));

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

@@ -17,6 +17,7 @@
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.IndexTemplateMetaData;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -259,7 +260,8 @@ public class MachineLearning extends Plugin implements ActionPlugin, AnalysisPlu
 
     public static final Setting<Boolean> ML_ENABLED =
             Setting.boolSetting("node.ml", XPackSettings.MACHINE_LEARNING_ENABLED, Property.NodeScope);
-    public static final String ML_ENABLED_NODE_ATTR = "ml.enabled";
+    // This is not used in v7 and higher, but users are still prevented from setting it directly to avoid confusion
+    private static final String PRE_V7_ML_ENABLED_NODE_ATTR = "ml.enabled";
     public static final String MAX_OPEN_JOBS_NODE_ATTR = "ml.max_open_jobs";
     public static final String MACHINE_MEMORY_NODE_ATTR = "ml.machine_memory";
     public static final Setting<Integer> CONCURRENT_JOB_ALLOCATIONS =
@@ -289,6 +291,14 @@ public MachineLearning(Settings settings, Path configPath) {
 
     protected XPackLicenseState getLicenseState() { return XPackPlugin.getSharedLicenseState(); }
 
+    public static boolean isMlNode(DiscoveryNode node) {
+        Map<String, String> nodeAttributes = node.getAttributes();
+        try {
+            return Integer.parseInt(nodeAttributes.get(MAX_OPEN_JOBS_NODE_ATTR)) > 0;
+        } catch (NumberFormatException e) {
+            return false;
+        }
+    }
 
     public List<Setting<?>> getSettings() {
         return Collections.unmodifiableList(
@@ -299,16 +309,14 @@ public List<Setting<?>> getSettings() {
                         MAX_LAZY_ML_NODES,
                         MAX_MACHINE_MEMORY_PERCENT,
                         AutodetectBuilder.DONT_PERSIST_MODEL_STATE_SETTING,
-                        AutodetectBuilder.MAX_ANOMALY_RECORDS_SETTING,
                         AutodetectBuilder.MAX_ANOMALY_RECORDS_SETTING_DYNAMIC,
-                        AutodetectProcessManager.MAX_RUNNING_JOBS_PER_NODE,
                         AutodetectProcessManager.MAX_OPEN_JOBS_PER_NODE,
                         AutodetectProcessManager.MIN_DISK_SPACE_OFF_HEAP,
                         MlConfigMigrationEligibilityCheck.ENABLE_CONFIG_MIGRATION));
     }
 
     public Settings additionalSettings() {
-        String mlEnabledNodeAttrName = "node.attr." + ML_ENABLED_NODE_ATTR;
+        String mlEnabledNodeAttrName = "node.attr." + PRE_V7_ML_ENABLED_NODE_ATTR;
         String maxOpenJobsPerNodeNodeAttrName = "node.attr." + MAX_OPEN_JOBS_NODE_ATTR;
         String machineMemoryAttrName = "node.attr." + MACHINE_MEMORY_NODE_ATTR;
 
@@ -320,12 +328,12 @@ public Settings additionalSettings() {
         Settings.Builder additionalSettings = Settings.builder();
         Boolean allocationEnabled = ML_ENABLED.get(settings);
         if (allocationEnabled != null && allocationEnabled) {
-            // TODO: the simple true/false flag will not be required once all supported versions have the number - consider removing in 7.0
-            addMlNodeAttribute(additionalSettings, mlEnabledNodeAttrName, "true");
             addMlNodeAttribute(additionalSettings, maxOpenJobsPerNodeNodeAttrName,
                     String.valueOf(AutodetectProcessManager.MAX_OPEN_JOBS_PER_NODE.get(settings)));
             addMlNodeAttribute(additionalSettings, machineMemoryAttrName,
                     Long.toString(machineMemoryFromStats(OsProbe.getInstance().osStats())));
+            // This is not used in v7 and higher, but users are still prevented from setting it directly to avoid confusion
+            disallowMlNodeAttributes(mlEnabledNodeAttrName);
         } else {
             disallowMlNodeAttributes(mlEnabledNodeAttrName, maxOpenJobsPerNodeNodeAttrName, machineMemoryAttrName);
         }

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

@@ -143,8 +143,7 @@ private int mlNodeCount(final ClusterState clusterState) {
 
         int mlNodeCount = 0;
         for (DiscoveryNode node : clusterState.getNodes()) {
-            String enabled = node.getAttributes().get(MachineLearning.ML_ENABLED_NODE_ATTR);
-            if (Boolean.parseBoolean(enabled)) {
+            if (MachineLearning.isMlNode(node)) {
                 ++mlNodeCount;
             }
         }

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

@@ -399,9 +399,7 @@ private void deleteQuantiles(ParentTaskAssigningClient parentTaskClient, String
         // The quantiles type and doc ID changed in v5.5 so delete both the old and new format
         DeleteByQueryRequest request = new DeleteByQueryRequest(AnomalyDetectorsIndex.jobStateIndexName());
         // Just use ID here, not type, as trying to delete different types spams the logs with an exception stack trace
-        IdsQueryBuilder query = new IdsQueryBuilder().addIds(Quantiles.documentId(jobId),
-                // TODO: remove in 7.0
-                Quantiles.v54DocumentId(jobId));
+        IdsQueryBuilder query = new IdsQueryBuilder().addIds(Quantiles.documentId(jobId));
         request.setQuery(query);
         request.setIndicesOptions(MlIndicesUtils.addIgnoreUnavailable(IndicesOptions.lenientExpandOpen()));
         request.setAbortOnVersionConflict(false);
@@ -436,9 +434,7 @@ private void deleteCategorizerState(ParentTaskAssigningClient parentTaskClient,
         // The categorizer state type and doc ID changed in v5.5 so delete both the old and new format
         DeleteByQueryRequest request = new DeleteByQueryRequest(AnomalyDetectorsIndex.jobStateIndexName());
         // Just use ID here, not type, as trying to delete different types spams the logs with an exception stack trace
-        IdsQueryBuilder query = new IdsQueryBuilder().addIds(CategorizerState.documentId(jobId, docNum),
-                // TODO: remove in 7.0
-                CategorizerState.v54DocumentId(jobId, docNum));
+        IdsQueryBuilder query = new IdsQueryBuilder().addIds(CategorizerState.documentId(jobId, docNum));
         request.setQuery(query);
         request.setIndicesOptions(MlIndicesUtils.addIgnoreUnavailable(IndicesOptions.lenientExpandOpen()));
         request.setAbortOnVersionConflict(false);