[Rollup] Metric config parser must use builder so validation runs (#31159)

The parser for the Metric config was directly instantiating
the config object, rather than using the builder.  That means it was
bypassing the validation logic built into the builder, and would allow
users to create invalid metric configs (like using unsupported metrics).

The job would later blow up and abort due to bad configs, but this isn't
immediately obvious to the user since the PutJob API succeeded.

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

@@ -12,7 +12,7 @@ import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.mapper.NumberFieldMapper;
@@ -75,12 +75,11 @@ public class MetricConfig implements Writeable, ToXContentFragment {
         MAPPER_TYPES = types;
     }
 
-    public static final ConstructingObjectParser<MetricConfig, Void> PARSER = new ConstructingObjectParser<>(
-            NAME, a -> new MetricConfig((String)a[0], (List<String>) a[1]));
+    public static final ObjectParser<MetricConfig.Builder, Void> PARSER = new ObjectParser<>(NAME, MetricConfig.Builder::new);
 
     static {
-        PARSER.declareString(ConstructingObjectParser.constructorArg(), FIELD);
-        PARSER.declareStringArray(ConstructingObjectParser.constructorArg(), METRICS);
+        PARSER.declareString(MetricConfig.Builder::setField, FIELD);
+        PARSER.declareStringArray(MetricConfig.Builder::setMetrics, METRICS);
     }
 
     MetricConfig(String name, List<String> metrics) {
@@ -257,4 +256,4 @@ public class MetricConfig implements Writeable, ToXContentFragment {
             return new MetricConfig(field, metrics);
         }
     }
-}
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/rollup/job/RollupJobConfig.java

@@ -63,7 +63,7 @@ public class RollupJobConfig implements NamedWriteable, ToXContentObject {
     static {
         PARSER.declareString(RollupJobConfig.Builder::setId, RollupField.ID);
         PARSER.declareObject(RollupJobConfig.Builder::setGroupConfig, (p, c) -> GroupConfig.PARSER.apply(p,c).build(), GROUPS);
-        PARSER.declareObjectArray(RollupJobConfig.Builder::setMetricsConfig, MetricConfig.PARSER, METRICS);
+        PARSER.declareObjectArray(RollupJobConfig.Builder::setMetricsConfig, (p, c) -> MetricConfig.PARSER.apply(p, c).build(), METRICS);
         PARSER.declareString((params, val) ->
                 params.setTimeout(TimeValue.parseTimeValue(val, TIMEOUT.getPreferredName())), TIMEOUT);
         PARSER.declareString(RollupJobConfig.Builder::setIndexPattern, INDEX_PATTERN);

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/rollup/job/MetricsConfigSerializingTests.java

@@ -24,7 +24,7 @@ import static org.mockito.Mockito.when;
 public class MetricsConfigSerializingTests extends AbstractSerializingTestCase<MetricConfig> {
     @Override
     protected MetricConfig doParseInstance(XContentParser parser) throws IOException {
-        return MetricConfig.PARSER.apply(parser, null);
+        return MetricConfig.PARSER.apply(parser, null).build();
     }
 
     @Override
@@ -36,7 +36,7 @@ public class MetricsConfigSerializingTests extends AbstractSerializingTestCase<M
     protected MetricConfig createTestInstance() {
         return ConfigTestHelpers.getMetricConfig().build();
     }
-    
+
     public void testValidateNoMapping() throws IOException {
         ActionRequestValidationException e = new ActionRequestValidationException();
         Map<String, Map<String, FieldCapabilities>> responseMap = new HashMap<>();

x-pack/plugin/src/test/resources/rest-api-spec/test/rollup/put_job.yml

@@ -188,3 +188,33 @@ setup:
             ]
           }
 
+---
+"Unknown Metric":
+
+  - do:
+      catch: /Unsupported metric \[does_not_exist\]/
+      headers:
+        Authorization: "Basic eF9wYWNrX3Jlc3RfdXNlcjp4LXBhY2stdGVzdC1wYXNzd29yZA==" # run as x_pack_rest_user, i.e. the test setup superuser
+      xpack.rollup.put_job:
+        id: foo
+        body:  >
+          {
+            "index_pattern": "foo",
+            "rollup_index": "foo_rollup",
+            "cron": "*/30 * * * * ?",
+            "page_size" :10,
+            "groups" : {
+              "date_histogram": {
+                "field": "the_field",
+                "interval": "1h"
+              }
+            },
+            "metrics": [
+              {
+                "field": "value_field",
+                "metrics": ["min", "max", "sum", "does_not_exist"]
+              }
+            ]
+          }
+
+