Skip to content

Commit 59a1205

Browse files
davidkyledavidkyle
committed
[ML] Datafeed config CRUD operations (elastic#32854)
1 parent 27a81f7 commit 59a1205

23 files changed

+1047
-307
lines changed

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlMetaIndex.java

-2
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,6 @@ public final class MlMetaIndex {
2121
*/
2222
public static final String INDEX_NAME = ".ml-meta";
2323

24-
public static final String INCLUDE_TYPE_KEY = "include_type";
25-
2624
public static final String TYPE = "doc";
2725

2826
private MlMetaIndex() {}

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlMetadata.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,7 @@ private static <T extends Writeable> void writeMap(Map<String, T> map, StreamOut
167167
@Override
168168
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
169169
DelegatingMapParams extendedParams =
170-
new DelegatingMapParams(Collections.singletonMap(ToXContentParams.FOR_CLUSTER_STATE, "true"), params);
170+
new DelegatingMapParams(Collections.singletonMap(ToXContentParams.FOR_INTERNAL_STORAGE, "true"), params);
171171
mapValuesToXContent(JOBS_FIELD, jobs, builder, extendedParams);
172172
mapValuesToXContent(DATAFEEDS_FIELD, datafeeds, builder, extendedParams);
173173
return builder;

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

+1-3
Original file line numberDiff line numberDiff line change
@@ -138,9 +138,7 @@ public void writeTo(StreamOutput out) throws IOException {
138138

139139
@Override
140140
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
141-
builder.startObject();
142-
datafeed.doXContentBody(builder, params);
143-
builder.endObject();
141+
datafeed.toXContent(builder, params);
144142
return builder;
145143
}
146144

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/calendars/Calendar.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313
import org.elasticsearch.common.xcontent.ObjectParser;
1414
import org.elasticsearch.common.xcontent.ToXContentObject;
1515
import org.elasticsearch.common.xcontent.XContentBuilder;
16-
import org.elasticsearch.xpack.core.ml.MlMetaIndex;
16+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
1717

1818
import java.io.IOException;
1919
import java.util.Arrays;
@@ -111,7 +111,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
111111
if (description != null) {
112112
builder.field(DESCRIPTION.getPreferredName(), description);
113113
}
114-
if (params.paramAsBoolean(MlMetaIndex.INCLUDE_TYPE_KEY, false)) {
114+
if (params.paramAsBoolean(ToXContentParams.INCLUDE_TYPE, false)) {
115115
builder.field(TYPE.getPreferredName(), CALENDAR_TYPE);
116116
}
117117
builder.endObject();

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/calendars/ScheduledEvent.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,14 @@
1515
import org.elasticsearch.common.xcontent.ToXContentObject;
1616
import org.elasticsearch.common.xcontent.XContentBuilder;
1717
import org.elasticsearch.common.xcontent.XContentParser;
18-
import org.elasticsearch.xpack.core.ml.MlMetaIndex;
1918
import org.elasticsearch.xpack.core.ml.job.config.DetectionRule;
2019
import org.elasticsearch.xpack.core.ml.job.config.Operator;
2120
import org.elasticsearch.xpack.core.ml.job.config.RuleAction;
2221
import org.elasticsearch.xpack.core.ml.job.config.RuleCondition;
2322
import org.elasticsearch.xpack.core.ml.job.messages.Messages;
2423
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
2524
import org.elasticsearch.xpack.core.ml.utils.Intervals;
25+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
2626
import org.elasticsearch.xpack.core.ml.utils.time.TimeUtils;
2727

2828
import java.io.IOException;
@@ -170,7 +170,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
170170
if (eventId != null) {
171171
builder.field(EVENT_ID.getPreferredName(), eventId);
172172
}
173-
if (params.paramAsBoolean(MlMetaIndex.INCLUDE_TYPE_KEY, false)) {
173+
if (params.paramAsBoolean(ToXContentParams.INCLUDE_TYPE, false)) {
174174
builder.field(TYPE.getPreferredName(), SCHEDULED_EVENT_TYPE);
175175
}
176176
builder.endObject();

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

+25-7
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
7272
public static final String DOC_COUNT = "doc_count";
7373

7474
public static final ParseField ID = new ParseField("datafeed_id");
75+
public static final ParseField CONFIG_TYPE = new ParseField("config_type");
7576
public static final ParseField QUERY_DELAY = new ParseField("query_delay");
7677
public static final ParseField FREQUENCY = new ParseField("frequency");
7778
public static final ParseField INDEXES = new ParseField("indexes");
@@ -94,6 +95,7 @@ private static ObjectParser<Builder, Void> createParser(boolean ignoreUnknownFie
9495
ObjectParser<Builder, Void> parser = new ObjectParser<>("datafeed_config", ignoreUnknownFields, Builder::new);
9596

9697
parser.declareString(Builder::setId, ID);
98+
parser.declareString((c, s) -> {}, CONFIG_TYPE);
9799
parser.declareString(Builder::setJobId, Job.ID);
98100
parser.declareStringArray(Builder::setIndices, INDEXES);
99101
parser.declareStringArray(Builder::setIndices, INDICES);
@@ -199,6 +201,16 @@ public DatafeedConfig(StreamInput in) throws IOException {
199201
}
200202
}
201203

204+
/**
205+
* The name of datafeed configuration document name from the datafeed ID.
206+
*
207+
* @param datafeedId The datafeed ID
208+
* @return The ID of document the datafeed config is persisted in
209+
*/
210+
public static String documentId(String datafeedId) {
211+
return "datafeed-" + datafeedId;
212+
}
213+
202214
public String getId() {
203215
return id;
204216
}
@@ -207,6 +219,10 @@ public String getJobId() {
207219
return jobId;
208220
}
209221

222+
public String getConfigType() {
223+
return TYPE;
224+
}
225+
210226
public TimeValue getQueryDelay() {
211227
return queryDelay;
212228
}
@@ -297,14 +313,11 @@ public void writeTo(StreamOutput out) throws IOException {
297313
@Override
298314
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
299315
builder.startObject();
300-
doXContentBody(builder, params);
301-
builder.endObject();
302-
return builder;
303-
}
304-
305-
public XContentBuilder doXContentBody(XContentBuilder builder, Params params) throws IOException {
306316
builder.field(ID.getPreferredName(), id);
307317
builder.field(Job.ID.getPreferredName(), jobId);
318+
if (params.paramAsBoolean(ToXContentParams.INCLUDE_TYPE, false) == true) {
319+
builder.field(CONFIG_TYPE.getPreferredName(), TYPE);
320+
}
308321
builder.field(QUERY_DELAY.getPreferredName(), queryDelay.getStringRep());
309322
if (frequency != null) {
310323
builder.field(FREQUENCY.getPreferredName(), frequency.getStringRep());
@@ -326,9 +339,10 @@ public XContentBuilder doXContentBody(XContentBuilder builder, Params params) th
326339
if (chunkingConfig != null) {
327340
builder.field(CHUNKING_CONFIG.getPreferredName(), chunkingConfig);
328341
}
329-
if (headers.isEmpty() == false && params.paramAsBoolean(ToXContentParams.FOR_CLUSTER_STATE, false) == true) {
342+
if (headers.isEmpty() == false && params.paramAsBoolean(ToXContentParams.FOR_INTERNAL_STORAGE, false) == true) {
330343
builder.field(HEADERS.getPreferredName(), headers);
331344
}
345+
builder.endObject();
332346
return builder;
333347
}
334348

@@ -468,6 +482,10 @@ public void setId(String datafeedId) {
468482
id = ExceptionsHelper.requireNonNull(datafeedId, ID.getPreferredName());
469483
}
470484

485+
public String getId() {
486+
return id;
487+
}
488+
471489
public void setJobId(String jobId) {
472490
this.jobId = ExceptionsHelper.requireNonNull(jobId, Job.ID.getPreferredName());
473491
}

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -302,7 +302,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
302302
// negative means "unknown", which should only happen for a 5.4 job
303303
if (detectorIndex >= 0
304304
// no point writing this to cluster state, as the indexes will get reassigned on reload anyway
305-
&& params.paramAsBoolean(ToXContentParams.FOR_CLUSTER_STATE, false) == false) {
305+
&& params.paramAsBoolean(ToXContentParams.FOR_INTERNAL_STORAGE, false) == false) {
306306
builder.field(DETECTOR_INDEX.getPreferredName(), detectorIndex);
307307
}
308308
builder.endObject();

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

+2-2
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,10 @@
1414
import org.elasticsearch.common.xcontent.ObjectParser;
1515
import org.elasticsearch.common.xcontent.ToXContentObject;
1616
import org.elasticsearch.common.xcontent.XContentBuilder;
17-
import org.elasticsearch.xpack.core.ml.MlMetaIndex;
1817
import org.elasticsearch.xpack.core.ml.job.messages.Messages;
1918
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
2019
import org.elasticsearch.xpack.core.ml.utils.MlStrings;
20+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
2121

2222
import java.io.IOException;
2323
import java.util.Arrays;
@@ -101,7 +101,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
101101
builder.field(DESCRIPTION.getPreferredName(), description);
102102
}
103103
builder.field(ITEMS.getPreferredName(), items);
104-
if (params.paramAsBoolean(MlMetaIndex.INCLUDE_TYPE_KEY, false)) {
104+
if (params.paramAsBoolean(ToXContentParams.INCLUDE_TYPE, false)) {
105105
builder.field(TYPE.getPreferredName(), FILTER_TYPE);
106106
}
107107
builder.endObject();

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/utils/ToXContentParams.java

+10-2
Original file line numberDiff line numberDiff line change
@@ -12,9 +12,17 @@
1212
public final class ToXContentParams {
1313

1414
/**
15-
* Parameter to indicate whether we are serialising to X Content for cluster state output.
15+
* Parameter to indicate whether we are serialising to X Content for
16+
* internal storage. Certain fields need to be persisted but should
17+
* not be visible everywhere.
1618
*/
17-
public static final String FOR_CLUSTER_STATE = "for_cluster_state";
19+
public static final String FOR_INTERNAL_STORAGE = "for_internal_storage";
20+
21+
/**
22+
* When serialising POJOs to X Content this indicates whether the type field
23+
* should be included or not
24+
*/
25+
public static final String INCLUDE_TYPE = "include_type";
1826

1927
private ToXContentParams() {
2028
}

‎x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfigTests.java

+41-1
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,17 @@
88
import com.carrotsearch.randomizedtesting.generators.CodepointSetGenerator;
99

1010
import org.elasticsearch.ElasticsearchException;
11+
import org.elasticsearch.common.bytes.BytesReference;
1112
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
1213
import org.elasticsearch.common.io.stream.Writeable;
1314
import org.elasticsearch.common.settings.Settings;
1415
import org.elasticsearch.common.unit.TimeValue;
1516
import org.elasticsearch.common.xcontent.DeprecationHandler;
17+
import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
1618
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
19+
import org.elasticsearch.common.xcontent.ToXContent;
1720
import org.elasticsearch.common.xcontent.XContentFactory;
21+
import org.elasticsearch.common.xcontent.XContentHelper;
1822
import org.elasticsearch.common.xcontent.XContentParseException;
1923
import org.elasticsearch.common.xcontent.XContentParser;
2024
import org.elasticsearch.common.xcontent.XContentType;
@@ -36,17 +40,22 @@
3640
import org.elasticsearch.test.ESTestCase;
3741
import org.elasticsearch.xpack.core.ml.datafeed.ChunkingConfig.Mode;
3842
import org.elasticsearch.xpack.core.ml.job.messages.Messages;
43+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
3944
import org.joda.time.DateTimeZone;
4045

4146
import java.io.IOException;
4247
import java.util.ArrayList;
4348
import java.util.Collections;
49+
import java.util.HashMap;
4450
import java.util.List;
51+
import java.util.Map;
4552
import java.util.TimeZone;
4653

4754
import static org.hamcrest.Matchers.containsString;
4855
import static org.hamcrest.Matchers.equalTo;
4956
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
57+
import static org.hamcrest.Matchers.hasEntry;
58+
import static org.hamcrest.Matchers.hasSize;
5059
import static org.hamcrest.Matchers.is;
5160
import static org.hamcrest.Matchers.lessThan;
5261
import static org.hamcrest.Matchers.not;
@@ -63,6 +72,10 @@ public static DatafeedConfig createRandomizedDatafeedConfig(String jobId) {
6372
}
6473

6574
public static DatafeedConfig createRandomizedDatafeedConfig(String jobId, long bucketSpanMillis) {
75+
return createRandomizedDatafeedConfigBuilder(jobId, bucketSpanMillis).build();
76+
}
77+
78+
private static DatafeedConfig.Builder createRandomizedDatafeedConfigBuilder(String jobId, long bucketSpanMillis) {
6679
DatafeedConfig.Builder builder = new DatafeedConfig.Builder(randomValidDatafeedId(), jobId);
6780
builder.setIndices(randomStringList(1, 10));
6881
builder.setTypes(randomStringList(0, 10));
@@ -109,7 +122,7 @@ public static DatafeedConfig createRandomizedDatafeedConfig(String jobId, long b
109122
if (randomBoolean()) {
110123
builder.setChunkingConfig(ChunkingConfigTests.createRandomizedChunk());
111124
}
112-
return builder.build();
125+
return builder;
113126
}
114127

115128
@Override
@@ -167,6 +180,33 @@ public void testFutureMetadataParse() throws IOException {
167180
assertNotNull(DatafeedConfig.LENIENT_PARSER.apply(parser, null).build());
168181
}
169182

183+
public void testToXContentForInternalStorage() throws IOException {
184+
DatafeedConfig.Builder builder = createRandomizedDatafeedConfigBuilder("foo", 300);
185+
186+
// headers are only persisted to cluster state
187+
Map<String, String> headers = new HashMap<>();
188+
headers.put("header-name", "header-value");
189+
builder.setHeaders(headers);
190+
DatafeedConfig config = builder.build();
191+
192+
ToXContent.MapParams params = new ToXContent.MapParams(Collections.singletonMap(ToXContentParams.FOR_INTERNAL_STORAGE, "true"));
193+
194+
BytesReference forClusterstateXContent = XContentHelper.toXContent(config, XContentType.JSON, params, false);
195+
XContentParser parser = XContentFactory.xContent(XContentType.JSON)
196+
.createParser(xContentRegistry(), LoggingDeprecationHandler.INSTANCE, forClusterstateXContent.streamInput());
197+
198+
DatafeedConfig parsedConfig = DatafeedConfig.LENIENT_PARSER.apply(parser, null).build();
199+
assertThat(parsedConfig.getHeaders(), hasEntry("header-name", "header-value"));
200+
201+
// headers are not written without the FOR_INTERNAL_STORAGE param
202+
BytesReference nonClusterstateXContent = XContentHelper.toXContent(config, XContentType.JSON, ToXContent.EMPTY_PARAMS, false);
203+
parser = XContentFactory.xContent(XContentType.JSON)
204+
.createParser(xContentRegistry(), LoggingDeprecationHandler.INSTANCE, nonClusterstateXContent.streamInput());
205+
206+
parsedConfig = DatafeedConfig.LENIENT_PARSER.apply(parser, null).build();
207+
assertThat(parsedConfig.getHeaders().entrySet(), hasSize(0));
208+
}
209+
170210
public void testCopyConstructor() {
171211
for (int i = 0; i < NUMBER_OF_TEST_RUNS; i++) {
172212
DatafeedConfig datafeedConfig = createTestInstance();

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

+2-1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.elasticsearch.xpack.core.ml.calendars.Calendar;
2727
import org.elasticsearch.xpack.core.ml.calendars.ScheduledEvent;
2828
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
29+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
2930
import org.elasticsearch.xpack.ml.job.JobManager;
3031
import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider;
3132

@@ -67,7 +68,7 @@ protected void doExecute(Task task, PostCalendarEventsAction.Request request,
6768
IndexRequest indexRequest = new IndexRequest(MlMetaIndex.INDEX_NAME, MlMetaIndex.TYPE);
6869
try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
6970
indexRequest.source(event.toXContent(builder,
70-
new ToXContent.MapParams(Collections.singletonMap(MlMetaIndex.INCLUDE_TYPE_KEY,
71+
new ToXContent.MapParams(Collections.singletonMap(ToXContentParams.INCLUDE_TYPE,
7172
"true"))));
7273
} catch (IOException e) {
7374
throw new IllegalStateException("Failed to serialise event", e);

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

+2-1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.elasticsearch.xpack.core.ml.action.PutCalendarAction;
2727
import org.elasticsearch.xpack.core.ml.calendars.Calendar;
2828
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
29+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
2930

3031
import java.io.IOException;
3132
import java.util.Collections;
@@ -53,7 +54,7 @@ protected void doExecute(Task task, PutCalendarAction.Request request, ActionLis
5354
IndexRequest indexRequest = new IndexRequest(MlMetaIndex.INDEX_NAME, MlMetaIndex.TYPE, calendar.documentId());
5455
try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
5556
indexRequest.source(calendar.toXContent(builder,
56-
new ToXContent.MapParams(Collections.singletonMap(MlMetaIndex.INCLUDE_TYPE_KEY, "true"))));
57+
new ToXContent.MapParams(Collections.singletonMap(ToXContentParams.INCLUDE_TYPE, "true"))));
5758
} catch (IOException e) {
5859
throw new IllegalStateException("Failed to serialise calendar with id [" + calendar.getId() + "]", e);
5960
}

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

+2-1
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.elasticsearch.xpack.core.ml.action.PutFilterAction;
2828
import org.elasticsearch.xpack.core.ml.job.config.MlFilter;
2929
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
30+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
3031

3132
import java.io.IOException;
3233
import java.util.Collections;
@@ -53,7 +54,7 @@ protected void doExecute(Task task, PutFilterAction.Request request, ActionListe
5354
indexRequest.opType(DocWriteRequest.OpType.CREATE);
5455
indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
5556
try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
56-
ToXContent.MapParams params = new ToXContent.MapParams(Collections.singletonMap(MlMetaIndex.INCLUDE_TYPE_KEY, "true"));
57+
ToXContent.MapParams params = new ToXContent.MapParams(Collections.singletonMap(ToXContentParams.INCLUDE_TYPE, "true"));
5758
indexRequest.source(filter.toXContent(builder, params));
5859
} catch (IOException e) {
5960
throw new IllegalStateException("Failed to serialise filter with id [" + filter.getId() + "]", e);

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

+2-1
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.elasticsearch.xpack.core.ml.job.config.MlFilter;
3737
import org.elasticsearch.xpack.core.ml.job.messages.Messages;
3838
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
39+
import org.elasticsearch.xpack.core.ml.utils.ToXContentParams;
3940
import org.elasticsearch.xpack.ml.job.JobManager;
4041

4142
import java.io.IOException;
@@ -105,7 +106,7 @@ private void indexUpdatedFilter(MlFilter filter, long version, UpdateFilterActio
105106
indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
106107

107108
try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
108-
ToXContent.MapParams params = new ToXContent.MapParams(Collections.singletonMap(MlMetaIndex.INCLUDE_TYPE_KEY, "true"));
109+
ToXContent.MapParams params = new ToXContent.MapParams(Collections.singletonMap(ToXContentParams.INCLUDE_TYPE, "true"));
109110
indexRequest.source(filter.toXContent(builder, params));
110111
} catch (IOException e) {
111112
throw new IllegalStateException("Failed to serialise filter with id [" + filter.getId() + "]", e);

0 commit comments

Comments
 (0)