[Rollup] Add `value_count` metric (elastic/x-pack-elasticsearch#4315)

Adds `value_count` as one of the accepted metrics.  The caveat is that
it only accepts numeric values for two reasons:

- Job validation at creation makes sure all metrics are numeric fields.
Changing this would require new syntax (or disallowing anything but
value_count on mixed fields)
- when `toBuilders()` is called, we have to supply a ValueSource to
the ValueCountBuilder, and we don't know what the field type is at that
time.

These are both fixable, but relatively more involved.  I think numeric-only
is a reasonable limitation to start with

Original commit: elastic/x-pack-elasticsearch@270f24c8bf

docs/en/rest-api/rollup/rollup-job-config.asciidoc

@@ -211,7 +211,7 @@ The `metrics` configuration accepts an array of objects, where each object has t
   The field to collect metrics for.  This must be a numeric of some kind
 
 `metrics` (required)::
-  An array of metrics to collect for the field.  At least one metric must be configured.  Acceptable metrics are min/max/sum/avg.
+  An array of metrics to collect for the field.  At least one metric must be configured.  Acceptable metrics are min/max/sum/avg/value_count.
 
 
 

plugin/core/src/main/java/org/elasticsearch/xpack/core/rollup/RollupField.java

@@ -6,8 +6,16 @@
 package org.elasticsearch.xpack.core.rollup;
 
 import org.elasticsearch.common.ParseField;
+import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.min.MinAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.valuecount.ValueCountAggregationBuilder;
 import org.elasticsearch.search.aggregations.support.ValuesSourceAggregationBuilder;
 
+import java.util.Arrays;
+import java.util.List;
+
 public class RollupField {
     // Fields that are used both in core Rollup actions and Rollup plugin
     public static final ParseField ID = new ParseField("id");
@@ -22,6 +30,8 @@ public class RollupField {
     public static final String NAME = "rollup";
     public static final String AGG = "agg";
     public static final String ROLLUP_MISSING = "ROLLUP_MISSING_40710B25931745D4B0B8B310F6912A69";
+    public static final List<String> SUPPORTED_METRICS = Arrays.asList(MaxAggregationBuilder.NAME, MinAggregationBuilder.NAME,
+            SumAggregationBuilder.NAME, AvgAggregationBuilder.NAME, ValueCountAggregationBuilder.NAME);
 
     /**
      * Format to the appropriate Rollup field name convention

plugin/core/src/main/java/org/elasticsearch/xpack/core/rollup/job/MetricConfig.java

@@ -65,8 +65,8 @@ public class MetricConfig implements Writeable, ToXContentFragment {
     private static final ParseField MAX = new ParseField("max");
     private static final ParseField SUM = new ParseField("sum");
     private static final ParseField AVG = new ParseField("avg");
+    private static final ParseField VALUE_COUNT = new ParseField("value_count");
 
-    private static List<String> METRIC_WHITELIST = Arrays.asList("min", "max", "sum", "avg");
     private static final List<String> MAPPER_TYPES = Stream.of(NumberFieldMapper.NumberType.values())
             .map(NumberFieldMapper.NumberType::typeName)
             .collect(Collectors.toList());
@@ -123,6 +123,11 @@ public class MetricConfig implements Writeable, ToXContentFragment {
                 aggs.add(countBuilder);
             } else if (metric.equals(SUM.getPreferredName())) {
                 newBuilder = new SumAggregationBuilder(RollupField.formatFieldName(field, SumAggregationBuilder.NAME, RollupField.VALUE));
+            } else if (metric.equals(VALUE_COUNT.getPreferredName())) {
+                // TODO allow non-numeric value_counts.
+                // Hardcoding this is fine for now since the job validation guarantees that all metric fields are numerics
+                newBuilder = new ValueCountAggregationBuilder(
+                        RollupField.formatFieldName(field, ValueCountAggregationBuilder.NAME, RollupField.VALUE), ValueType.NUMERIC);
             } else {
                 throw new IllegalArgumentException("Unsupported metric type [" + metric + "]");
             }
@@ -240,9 +245,9 @@ public class MetricConfig implements Writeable, ToXContentFragment {
                         + "] must be a non-null, non-empty array of strings.");
             }
             metrics.forEach(m -> {
-                if (METRIC_WHITELIST.contains(m) == false) {
+                if (RollupField.SUPPORTED_METRICS.contains(m) == false) {
                     throw new IllegalArgumentException("Unsupported metric [" + m + "].  " +
-                            "Supported metrics include: " + METRIC_WHITELIST);
+                            "Supported metrics include: " + RollupField.SUPPORTED_METRICS);
                 }
             });
             return new MetricConfig(field, metrics);

plugin/core/src/test/java/org/elasticsearch/xpack/core/rollup/ConfigTestHelpers.java

@@ -69,6 +69,9 @@ public class ConfigTestHelpers {
         if (ESTestCase.randomBoolean()) {
             metrics.add("avg");
         }
+        if (ESTestCase.randomBoolean()) {
+            metrics.add("value_count");
+        }
         if (metrics.size() == 0) {
             metrics.add("min");
         }

plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/Rollup.java

@@ -22,17 +22,13 @@ import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.persistent.PersistentTasksExecutor;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.PersistentTaskPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
 import org.elasticsearch.script.ScriptService;
-import org.elasticsearch.persistent.PersistentTasksExecutor;
-import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
-import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder;
-import org.elasticsearch.search.aggregations.metrics.min.MinAggregationBuilder;
-import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
 import org.elasticsearch.threadpool.ExecutorBuilder;
 import org.elasticsearch.threadpool.FixedExecutorBuilder;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -98,9 +94,6 @@ public class Rollup extends Plugin implements ActionPlugin, PersistentTaskPlugin
     public static final Set<String> HEADER_FILTERS =
             new HashSet<>(Arrays.asList("es-security-runas-user", "_xpack_security_authentication"));
 
-    public static final List<String> SUPPORTED_METRICS = Arrays.asList(MaxAggregationBuilder.NAME, MinAggregationBuilder.NAME,
-            SumAggregationBuilder.NAME, AvgAggregationBuilder.NAME);
-
 
     private final Settings settings;
     private final boolean enabled;

plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupJobIdentifierUtils.java

@@ -71,7 +71,7 @@ public class RollupJobIdentifierUtils {
             checkDateHisto((DateHistogramAggregationBuilder) source, jobCaps, bestCaps);
         } else if (source.getWriteableName().equals(HistogramAggregationBuilder.NAME)) {
             checkHisto((HistogramAggregationBuilder) source, jobCaps, bestCaps);
-        } else if (Rollup.SUPPORTED_METRICS.contains(source.getWriteableName())) {
+        } else if (RollupField.SUPPORTED_METRICS.contains(source.getWriteableName())) {
             checkVSLeaf((ValuesSourceAggregationBuilder.LeafOnly) source, jobCaps, bestCaps);
         } else if (source.getWriteableName().equals(TermsAggregationBuilder.NAME)) {
             checkTerms((TermsAggregationBuilder)source, jobCaps, bestCaps);

plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/RollupRequestTranslator.java

@@ -131,7 +131,7 @@ public class RollupRequestTranslator {
             return translateDateHistogram((DateHistogramAggregationBuilder) source, filterConditions, registry);
         } else if (source.getWriteableName().equals(HistogramAggregationBuilder.NAME)) {
             return translateHistogram((HistogramAggregationBuilder) source, filterConditions, registry);
-        } else if (Rollup.SUPPORTED_METRICS.contains(source.getWriteableName())) {
+        } else if (RollupField.SUPPORTED_METRICS.contains(source.getWriteableName())) {
             return translateVSLeaf((ValuesSourceAggregationBuilder.LeafOnly)source, registry);
         } else if (source.getWriteableName().equals(TermsAggregationBuilder.NAME)) {
             return translateTerms((TermsAggregationBuilder)source, filterConditions, registry);

plugin/rollup/src/test/java/org/elasticsearch/xpack/rollup/config/ConfigTests.java

@@ -217,6 +217,6 @@ public class ConfigTests extends ESTestCase {
         MetricConfig.Builder config = ConfigTestHelpers.getMetricConfig();
         config.setMetrics(Arrays.asList("max","foo"));
         Exception e = expectThrows(IllegalArgumentException.class, config::build);
-        assertThat(e.getMessage(), equalTo("Unsupported metric [foo].  Supported metrics include: [min, max, sum, avg]"));
+        assertThat(e.getMessage(), equalTo("Unsupported metric [foo].  Supported metrics include: [max, min, sum, avg, value_count]"));
     }
 }