Skip to content

Commit

Permalink
[ML] Send autodetect config updates as JSON (#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.
edsavage authored Jan 20, 2021

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature. The key has expired.
1 parent d8de8f2 commit 1e14a01
Showing 11 changed files with 68 additions and 389 deletions.
Original file line number Diff line number Diff line change
@@ -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));
Original file line number Diff line number Diff line change
@@ -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";
Original file line number Diff line number Diff line change
@@ -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());
}

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

This file was deleted.

This file was deleted.

This file was deleted.

0 comments on commit 1e14a01

Please sign in to comment.