[ML] Send autodetect config updates as JSON

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/DetectionRulesIT.java

@@ -122,6 +122,7 @@ public void testCondition() throws Exception {
         assertThat(records.get(0).getByFieldValue(), equalTo("low"));
     }
 
+    @AwaitsFix(bugUrl = "https://github.com/elastic/ml-cpp/issues/1253")
     public void testScope() throws Exception {
         MlFilter safeIps = MlFilter.builder("safe_ips").setItems("111.111.111.111", "222.222.222.222").build();
         assertThat(putMlFilter(safeIps).getFilter(), equalTo(safeIps));

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/ScheduledEventsIT.java

@@ -195,6 +195,7 @@ public void testScheduledEventWithInterimResults() throws IOException {
     /**
      * Test an open job picks up changes to scheduled events/calendars
      */
+    @AwaitsFix(bugUrl = "https://github.com/elastic/ml-cpp/issues/1253")
     public void testAddEventsToOpenJob() throws Exception {
         TimeValue bucketSpan = TimeValue.timeValueMinutes(30);
         Job.Builder job = createJob("scheduled-events-add-events-to-open-job", bucketSpan);
@@ -263,6 +264,7 @@ public void testAddEventsToOpenJob() throws Exception {
     /**
      * An open job that later gets added to a calendar, should take the scheduled events into account
      */
+    @AwaitsFix(bugUrl = "https://github.com/elastic/ml-cpp/issues/1253")
     public void testAddOpenedJobToGroupWithCalendar() throws Exception {
         TimeValue bucketSpan = TimeValue.timeValueMinutes(30);
         String groupName = "opened-calendar-job-group";

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/writer/AutodetectControlMsgWriter.java

@@ -7,7 +7,6 @@
 
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
@@ -23,9 +22,9 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.io.StringWriter;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 
 /**
  * A writer for sending control messages to the C++ autodetect process.
@@ -84,6 +83,21 @@ public class AutodetectControlMsgWriter extends AbstractControlMsgWriter {
      */
     private static AtomicLong ms_FlushNumber = new AtomicLong(1);
 
+    /**
+     * This field name must match that in the api::CAnomalyJobConfig C++ class.
+     */
+    private static final String DETECTOR_INDEX = "detector_index";
+
+    /**
+     * This field name must match that in the api::CAnomalyJobConfig C++ class.
+     */
+    private static final String CUSTOM_RULES = "custom_rules";
+
+    /**
+     * This field name must match that in the api::CAnomalyJobConfig C++ class.
+     */
+    private static final String DETECTOR_RULES = "detector_rules";
+
     /**
      * Construct the control message writer with a LengthEncodedWriter
      *
@@ -192,10 +206,16 @@ private void writeControlCodeFollowedByTimeRange(String code, String start, Stri
     }
 
     public void writeUpdateModelPlotMessage(ModelPlotConfig modelPlotConfig) throws IOException {
-        StringWriter configWriter = new StringWriter();
-        configWriter.append(UPDATE_MESSAGE_CODE).append("[modelPlotConfig]\n");
-        new ModelPlotConfigWriter(modelPlotConfig, configWriter).write();
-        writeMessage(configWriter.toString());
+        StringBuilder stringBuilder = new StringBuilder();
+        stringBuilder.append(UPDATE_MESSAGE_CODE);
+        try (XContentBuilder jsonBuilder = JsonXContent.contentBuilder()) {
+            jsonBuilder.startObject();
+            jsonBuilder.field(ModelPlotConfig.TYPE_FIELD.getPreferredName(), modelPlotConfig);
+            jsonBuilder.endObject();
+            String msg = Strings.toString(jsonBuilder);
+            stringBuilder.append(msg);
+        }
+        writeMessage(stringBuilder.toString());
     }
 
     public void writeCategorizationStopOnWarnMessage(boolean isStopOnWarn) throws IOException {
@@ -204,37 +224,44 @@ public void writeCategorizationStopOnWarnMessage(boolean isStopOnWarn) throws IO
 
     public void writeUpdateDetectorRulesMessage(int detectorIndex, List<DetectionRule> rules) throws IOException {
         StringBuilder stringBuilder = new StringBuilder();
-        stringBuilder.append(UPDATE_MESSAGE_CODE).append("[detectorRules]\n");
-        stringBuilder.append("detectorIndex=").append(Integer.toString(detectorIndex)).append("\n");
-
-        stringBuilder.append("rulesJson=");
+        stringBuilder.append(UPDATE_MESSAGE_CODE);
 
         try (XContentBuilder builder = JsonXContent.contentBuilder()) {
-            builder.startArray();
-            for (DetectionRule rule : rules) {
-                rule.toXContent(builder, ToXContent.EMPTY_PARAMS);
-            }
-            builder.endArray();
+            builder.startObject();
+            builder.field(DETECTOR_RULES).startObject();
+            builder.field(DETECTOR_INDEX, detectorIndex);
+            builder.field(CUSTOM_RULES, rules);
+            builder.endObject();
+            builder.endObject();
             stringBuilder.append(Strings.toString(builder));
         }
-
         writeMessage(stringBuilder.toString());
     }
 
     public void writeUpdateFiltersMessage(List<MlFilter> filters) throws IOException {
+
         StringBuilder stringBuilder = new StringBuilder();
-        stringBuilder.append(UPDATE_MESSAGE_CODE).append("[filters]\n");
-        new MlFilterWriter(filters, stringBuilder).write();
+        stringBuilder.append(UPDATE_MESSAGE_CODE);
+        try (XContentBuilder jsonBuilder = JsonXContent.contentBuilder()) {
+            jsonBuilder.startObject().field(MlFilter.RESULTS_FIELD.getPreferredName(), filters).endObject();
+            String msg = Strings.toString(jsonBuilder);
+            stringBuilder.append(msg);
+        }
         writeMessage(stringBuilder.toString());
     }
 
     public void writeUpdateScheduledEventsMessage(List<ScheduledEvent> events, TimeValue bucketSpan) throws IOException {
         StringBuilder stringBuilder = new StringBuilder();
-        stringBuilder.append(UPDATE_MESSAGE_CODE).append("[scheduledEvents]\n");
-        if (events.isEmpty()) {
-            stringBuilder.append("clear = true\n");
-        } else {
-            new ScheduledEventsWriter(events, bucketSpan, stringBuilder).write();
+        stringBuilder.append(UPDATE_MESSAGE_CODE);
+
+        List<ScheduledEventToRuleWriter> scheduledEventToRuleWriters = events.stream()
+            .map(x -> new ScheduledEventToRuleWriter(x.getDescription(), x.toDetectionRule(bucketSpan)))
+            .collect(Collectors.toList());
+
+        try (XContentBuilder jsonBuilder = JsonXContent.contentBuilder()) {
+            jsonBuilder.startObject().field(ScheduledEvent.RESULTS_FIELD.getPreferredName(), scheduledEventToRuleWriters).endObject();
+            String msg = Strings.toString(jsonBuilder);
+            stringBuilder.append(msg);
         }
         writeMessage(stringBuilder.toString());
     }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/writer/AutodetectControlMsgWriterTests.java

@@ -187,7 +187,7 @@ public void testWriteUpdateModelPlotMessage() throws IOException {
         inOrder.verify(lengthEncodedWriter).writeNumFields(4);
         inOrder.verify(lengthEncodedWriter, times(3)).writeField("");
         inOrder.verify(lengthEncodedWriter)
-            .writeField("u[modelPlotConfig]\nboundspercentile = 95.0\nterms = foo,bar\nannotations_enabled = false\n");
+            .writeField("u{\"model_plot_config\":{\"enabled\":true,\"terms\":\"foo,bar\",\"annotations_enabled\":false}}");
         verifyNoMoreInteractions(lengthEncodedWriter);
     }
 
@@ -201,11 +201,10 @@ public void testWriteUpdateDetectorRulesMessage() throws IOException {
         InOrder inOrder = inOrder(lengthEncodedWriter);
         inOrder.verify(lengthEncodedWriter).writeNumFields(4);
         inOrder.verify(lengthEncodedWriter, times(3)).writeField("");
-        inOrder.verify(lengthEncodedWriter).writeField("u[detectorRules]\ndetectorIndex=2\n" +
-                "rulesJson=[{\"actions\":[\"skip_result\"],\"conditions\":" +
-                "[{\"applies_to\":\"actual\",\"operator\":\"gt\",\"value\":5.0}]}," +
-                "{\"actions\":[\"skip_result\"],\"conditions\":[" +
-                "{\"applies_to\":\"actual\",\"operator\":\"gt\",\"value\":5.0}]}]");
+        inOrder.verify(lengthEncodedWriter).writeField("u{\"detector_rules\":{\"detector_index\":2," +
+            "\"custom_rules\":[{\"actions\":[\"skip_result\"]," +
+            "\"conditions\":[{\"applies_to\":\"actual\",\"operator\":\"gt\",\"value\":5.0}]}," +
+            "{\"actions\":[\"skip_result\"],\"conditions\":[{\"applies_to\":\"actual\",\"operator\":\"gt\",\"value\":5.0}]}]}}");
         verifyNoMoreInteractions(lengthEncodedWriter);
     }
 
@@ -220,7 +219,8 @@ public void testWriteUpdateFiltersMessage() throws IOException {
         InOrder inOrder = inOrder(lengthEncodedWriter);
         inOrder.verify(lengthEncodedWriter).writeNumFields(2);
         inOrder.verify(lengthEncodedWriter, times(1)).writeField("");
-        inOrder.verify(lengthEncodedWriter).writeField("u[filters]\nfilter.filter_1 = [\"a\"]\nfilter.filter_2 = [\"b\",\"c\"]\n");
+        inOrder.verify(lengthEncodedWriter).writeField("u{\"filters\":[{\"filter_id\":\"filter_1\",\"items\":[\"a\"]}," +
+            "{\"filter_id\":\"filter_2\",\"items\":[\"b\",\"c\"]}]}");
         verifyNoMoreInteractions(lengthEncodedWriter);
     }
 
@@ -246,15 +246,13 @@ public void testWriteUpdateScheduledEventsMessage() throws IOException {
         inOrder.verify(lengthEncodedWriter, times(1)).writeField("");
         ArgumentCaptor<String> capturedMessage = ArgumentCaptor.forClass(String.class);
         inOrder.verify(lengthEncodedWriter).writeField(capturedMessage.capture());
-        assertThat(capturedMessage.getValue(), equalTo("u[scheduledEvents]\n"
-                + "scheduledevent.0.description = new year\n"
-                + "scheduledevent.0.rules = [{\"actions\":[\"skip_result\",\"skip_model_update\"],"
-                +     "\"conditions\":[{\"applies_to\":\"time\",\"operator\":\"gte\",\"value\":1.5147648E9},"
-                +     "{\"applies_to\":\"time\",\"operator\":\"lt\",\"value\":1.5148512E9}]}]\n"
-                + "scheduledevent.1.description = Jan maintenance day\n"
-                + "scheduledevent.1.rules = [{\"actions\":[\"skip_result\",\"skip_model_update\"],"
-                +     "\"conditions\":[{\"applies_to\":\"time\",\"operator\":\"gte\",\"value\":1.5151968E9},"
-                +     "{\"applies_to\":\"time\",\"operator\":\"lt\",\"value\":1.5152832E9}]}]\n"));
+        assertThat(capturedMessage.getValue(), equalTo("u{\"events\":[{\"description\":\"new year\"," +
+            "\"rules\":[{\"actions\":[\"skip_result\",\"skip_model_update\"]," +
+            "\"conditions\":[{\"applies_to\":\"time\",\"operator\":\"gte\",\"value\":1.5147648E9}," +
+            "{\"applies_to\":\"time\",\"operator\":\"lt\",\"value\":1.5148512E9}]}]}," +
+            "{\"description\":\"Jan maintenance day\",\"rules\":[{\"actions\":[\"skip_result\",\"skip_model_update\"]," +
+            "\"conditions\":[{\"applies_to\":\"time\",\"operator\":\"gte\",\"value\":1.5151968E9}," +
+            "{\"applies_to\":\"time\",\"operator\":\"lt\",\"value\":1.5152832E9}]}]}]}"));
         verifyNoMoreInteractions(lengthEncodedWriter);
     }
 
@@ -266,7 +264,7 @@ public void testWriteUpdateScheduledEventsMessage_GivenEmpty() throws IOExceptio
         InOrder inOrder = inOrder(lengthEncodedWriter);
         inOrder.verify(lengthEncodedWriter).writeNumFields(2);
         inOrder.verify(lengthEncodedWriter, times(1)).writeField("");
-        inOrder.verify(lengthEncodedWriter).writeField("u[scheduledEvents]\nclear = true\n");
+        inOrder.verify(lengthEncodedWriter).writeField("u{\"events\":[]}");
         verifyNoMoreInteractions(lengthEncodedWriter);
     }