Improve rolling up metrics (#124739)

Remove unneeded Metric abstraction in MetricFieldProducer, which speeds-up rolling up gauges/counters.

Author: martijnvg

docs/changelog/124739.yaml

@@ -0,0 +1,5 @@
+pr: 124739
+summary: Improve rolling up metrics
+area: Downsampling
+type: enhancement
+issues: []

x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/AggregateMetricFieldSerializer.java

@@ -12,7 +12,7 @@
 import java.io.IOException;
 import java.util.Collection;
 
-public class AggregateMetricFieldSerializer implements DownsampleFieldSerializer {
+final class AggregateMetricFieldSerializer implements DownsampleFieldSerializer {
     private final Collection<AbstractDownsampleFieldProducer> producers;
     private final String name;
 
@@ -22,7 +22,7 @@ public class AggregateMetricFieldSerializer implements DownsampleFieldSerializer
      * @param producers a collection of {@link AbstractDownsampleFieldProducer} instances with the subfields
      *                  of the aggregate_metric_double field.
      */
-    public AggregateMetricFieldSerializer(String name, Collection<AbstractDownsampleFieldProducer> producers) {
+    AggregateMetricFieldSerializer(String name, Collection<AbstractDownsampleFieldProducer> producers) {
         this.name = name;
         this.producers = producers;
     }
@@ -38,8 +38,22 @@ public void write(XContentBuilder builder) throws IOException {
             assert name.equals(fieldProducer.name()) : "producer has a different name";
             if (fieldProducer.isEmpty() == false) {
                 if (fieldProducer instanceof MetricFieldProducer metricFieldProducer) {
-                    for (MetricFieldProducer.Metric metric : metricFieldProducer.metrics()) {
-                        builder.field(metric.name(), metric.get());
+                    if (metricFieldProducer instanceof MetricFieldProducer.GaugeMetricFieldProducer gaugeProducer) {
+                        builder.field("max", gaugeProducer.max);
+                        builder.field("min", gaugeProducer.min);
+                        builder.field("sum", gaugeProducer.sum.value());
+                        builder.field("value_count", gaugeProducer.count);
+                    } else if (metricFieldProducer instanceof MetricFieldProducer.CounterMetricFieldProducer counterProducer) {
+                        builder.field("last_value", counterProducer.lastValue);
+                    } else if (metricFieldProducer instanceof MetricFieldProducer.AggregatedGaugeMetricFieldProducer producer) {
+                        switch (producer.metric) {
+                            case max -> builder.field("max", producer.max);
+                            case min -> builder.field("min", producer.min);
+                            case sum -> builder.field("sum", producer.sum.value());
+                            case value_count -> builder.field("value_count", producer.count);
+                        }
+                    } else {
+                        throw new IllegalStateException();
                     }
                 } else if (fieldProducer instanceof LabelFieldProducer labelFieldProducer) {
                     LabelFieldProducer.Label label = labelFieldProducer.label();

x-pack/plugin/downsample/src/main/java/org/elasticsearch/xpack/downsample/AggregateMetricFieldValueFetcher.java

@@ -47,14 +47,7 @@ private AbstractDownsampleFieldProducer createFieldProducer() {
         if (aggMetricFieldType.getMetricType() != null) {
             // If the field is an aggregate_metric_double field, we should use the correct subfields
             // for each aggregation. This is a downsample-of-downsample case
-            MetricFieldProducer.Metric metricOperation = switch (metric) {
-                case max -> new MetricFieldProducer.Max();
-                case min -> new MetricFieldProducer.Min();
-                case sum -> new MetricFieldProducer.Sum();
-                // To compute value_count summary, we must sum all field values
-                case value_count -> new MetricFieldProducer.Sum(AggregateMetricDoubleFieldMapper.Metric.value_count.name());
-            };
-            return new MetricFieldProducer.GaugeMetricFieldProducer(aggMetricFieldType.name(), metricOperation);
+            return new MetricFieldProducer.AggregatedGaugeMetricFieldProducer(aggMetricFieldType.name(), metric);
         } else {
             // If field is not a metric, we downsample it as a label
             return new LabelFieldProducer.AggregateMetricFieldProducer.AggregateMetricFieldProducer(aggMetricFieldType.name(), metric);