Skip to content

Commit

Permalink
[ML] Send autodetect config updates as JSON (elastic#67721)
Browse files Browse the repository at this point in the history
Send configuration updates to model plot, scheduled events, detector
rules and filters as JSON formatted strings.

Remove any now unneeded supporting code.
  • Loading branch information
edsavage committed Jan 20, 2021
1 parent 1d08d2f commit 920d34d
Show file tree
Hide file tree
Showing 11 changed files with 68 additions and 389 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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.
Expand Down Expand Up @@ -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
*
Expand Down Expand Up @@ -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 {
Expand All @@ -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());
}
Expand Down

This file was deleted.

This file was deleted.

This file was deleted.

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

Expand All @@ -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);
}

Expand All @@ -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);
}

Expand All @@ -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);
}

Expand All @@ -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);
}

Expand Down
Loading

0 comments on commit 920d34d

Please sign in to comment.