Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ML] Addressing bug streaming DatafeedConfig aggs from (<= 6.5.4) -> 6.7.0 #40610

Merged
merged 3 commits into from Mar 29, 2019
Merged

[ML] Addressing bug streaming DatafeedConfig aggs from (<= 6.5.4) -> 6.7.0 #40610

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0300b46
Addressing stream failure and adding tests to catch such in the future
benwtrent Mar 28, 2019
2cdf5a6
Add aggs to full cluster restart tests
davidkyle Mar 29, 2019
aa804e6
Test BWC for datafeeds with and without aggs
droberts195 Mar 29, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 4 additions & 2 deletions x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/AggProvider.java
View file
Open in desktop
Expand Up @@ -76,7 +76,8 @@ static AggProvider fromStream(StreamInput in) throws IOException {
} else if (in.getVersion().onOrAfter(Version.V_6_6_0)) { // Has the bug, but supports lazy objects
return new AggProvider(in.readMap(), null, null);
} else { // only supports eagerly parsed objects
return AggProvider.fromParsedAggs(in.readOptionalWriteable(AggregatorFactories.Builder::new));
// Upstream, we have read the bool already and know for sure that we have parsed aggs in the stream
return AggProvider.fromParsedAggs(new AggregatorFactories.Builder(in));
}
}

Expand Down Expand Up @@ -114,7 +115,8 @@ public void writeTo(StreamOutput out) throws IOException {
// actually are aggregations defined
throw new ElasticsearchException("Unsupported operation: parsed aggregations are null");
}
out.writeOptionalWriteable(parsedAggs);
// Upstream we already verified that this calling object is not null, no need to write a second boolean to the stream
parsedAggs.writeTo(out);
}
}

Expand Down
2 changes: 2 additions & 0 deletions ...ck/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfig.java
View file
Open in desktop
Expand Up @@ -219,6 +219,7 @@ public DatafeedConfig(StreamInput in) throws IOException {
}
// each of these writables are version aware
this.queryProvider = QueryProvider.fromStream(in);
// This reads a boolean from the stream, if true, it sends the stream to the `fromStream` method
this.aggProvider = in.readOptionalWriteable(AggProvider::fromStream);

if (in.readBoolean()) {
Expand Down Expand Up @@ -445,6 +446,7 @@ public void writeTo(StreamOutput out) throws IOException {

// Each of these writables are version aware
queryProvider.writeTo(out); // never null
// This writes a boolean to the stream, if true, it sends the stream to the `writeTo` method
out.writeOptionalWriteable(aggProvider);

if (scriptFields != null) {
Expand Down
32 changes: 31 additions & 1 deletion ...lling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/40_ml_datafeed_crud.yml
View file
Open in desktop
Expand Up @@ -7,6 +7,8 @@
- length: { datafeeds.0.indices: 1 }
- length: { datafeeds.0.types: 1 }
- gte: { datafeeds.0.scroll_size: 2000 }
- match: { datafeeds.0.aggregations.buckets.date_histogram.field: time }
- match: { datafeeds.0.aggregations.buckets.aggregations.time.max.field: time }

- do:
xpack.ml.get_datafeed_stats:
Expand All @@ -25,6 +27,7 @@
"description":"Cluster upgrade",
"analysis_config" : {
"bucket_span": "60s",
"summary_count_field_name": "doc_count",
"detectors" :[{"function":"count"}]
},
"analysis_limits" : {
Expand All @@ -44,7 +47,34 @@
"job_id":"mixed-cluster-datafeed-job",
"indices":["airline-data"],
"types":["response"],
"scroll_size": 2000
"scroll_size": 2000,
"aggregations": {
"buckets": {
"date_histogram": {
"field": "time",
"interval": "30s",
"time_zone": "UTC"
},
"aggregations": {
"time": {
"max": {"field": "time"}
},
"airline": {
"terms": {
"field": "airline",
"size": 100
},
"aggregations": {
"responsetime": {
"avg": {
"field": "responsetime"
}
}
}
}
}
}
}
}

- do:
Expand Down
30 changes: 29 additions & 1 deletion ...rolling-upgrade/src/test/resources/rest-api-spec/test/old_cluster/40_ml_datafeed_crud.yml
View file
Open in desktop
Expand Up @@ -9,6 +9,7 @@
"description":"Cluster upgrade",
"analysis_config" : {
"bucket_span": "60s",
"summary_count_field_name": "doc_count",
"detectors" :[{"function":"count"}]
},
"analysis_limits" : {
Expand All @@ -29,7 +30,34 @@
"job_id":"old-cluster-datafeed-job",
"indices":["airline-data"],
"types":["response"],
"scroll_size": 2000
"scroll_size": 2000,
"aggregations": {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since the wire transport is quite different for null and non-null aggregations I think it would be good to have a second datafeed that doesn't use aggregations. I will push another test to this PR soon.

"buckets": {
"date_histogram": {
"field": "time",
"interval": "30s",
"time_zone": "UTC"
},
"aggregations": {
"time": {
"max": {"field": "time"}
},
"airline": {
"terms": {
"field": "airline",
"size": 100
},
"aggregations": {
"responsetime": {
"avg": {
"field": "responsetime"
}
}
}
}
}
}
}
}

- do:
Expand Down
4 changes: 4 additions & 0 deletions ...ng-upgrade/src/test/resources/rest-api-spec/test/upgraded_cluster/40_ml_datafeed_crud.yml
View file
Open in desktop
Expand Up @@ -25,6 +25,8 @@ setup:
- length: { datafeeds.0.indices: 1 }
- length: { datafeeds.0.types: 1 }
- gte: { datafeeds.0.scroll_size: 2000 }
- match: { datafeeds.0.aggregations.buckets.date_histogram.field: time }
- match: { datafeeds.0.aggregations.buckets.aggregations.time.max.field: time }

- do:
xpack.ml.get_datafeed_stats:
Expand All @@ -39,6 +41,8 @@ setup:
- length: { datafeeds.0.indices: 1 }
- length: { datafeeds.0.types: 1 }
- gte: { datafeeds.0.scroll_size: 2000 }
- match: { datafeeds.0.aggregations.buckets.date_histogram.field: time }
- match: { datafeeds.0.aggregations.buckets.aggregations.time.max.field: time }

- do:
xpack.ml.get_datafeed_stats:
Expand Down