Track the minimum request timestamp aggregated into a MAD bucket, and push that to CAGG during flush (#188)

Co-authored-by: William Ehlhardt <[email protected]>
Co-authored-by: Ville Koskela <[email protected]>

Author: orborde

src/main/java/com/arpnetworking/metrics/mad/Bucket.java

@@ -76,7 +76,8 @@ public void close() {
                     b -> b.setData(data.build())
                             .setDimensions(_key)
                             .setPeriod(_period)
-                            .setStart(_start));
+                            .setStart(_start)
+                            .setMinRequestTime(_minRequestTime.orElse(null)));
             _sink.recordAggregateData(periodicData);
         } else {
             LOGGER.warn()
@@ -107,6 +108,7 @@ public void add(final Record record) {
         }
 
 
+
         for (final Map.Entry<String, ? extends Metric> entry : record.getMetrics().entrySet()) {
             final String name = entry.getKey();
             final Metric metric = entry.getValue();
@@ -178,6 +180,19 @@ public void add(final Record record) {
             }
             addMetric(metric, calculators);
         }
+
+        updateMinRequestTime(record);
+    }
+
+    private void updateMinRequestTime(final Record record) {
+        if (record.getRequestTime().isPresent()) {
+            if (!_minRequestTime.isPresent()) {
+                _minRequestTime = record.getRequestTime();
+            }
+            if (record.getRequestTime().get().isBefore(_minRequestTime.get())) {
+                _minRequestTime = record.getRequestTime();
+            }
+        }
     }
 
     public ZonedDateTime getStart() {
@@ -349,6 +364,7 @@ private Collection<Calculator<?>> getOrCreateCalculators(
     }
 
     private boolean _isOpen = true;
+    private Optional<ZonedDateTime> _minRequestTime = Optional.empty();
 
     private final Map<String, Collection<Calculator<?>>> _counterMetricCalculators = Maps.newHashMap();
     private final Map<String, Collection<Calculator<?>>> _gaugeMetricCalculators = Maps.newHashMap();

src/main/java/com/arpnetworking/metrics/mad/model/DefaultRecord.java

@@ -24,6 +24,9 @@
 import net.sf.oval.constraint.NotNull;
 
 import java.time.ZonedDateTime;
+import java.util.Optional;
+import javax.annotation.Nullable;
+
 
 /**
  * Default implementation of the <code>Record</code> interface.
@@ -49,6 +52,11 @@ public ZonedDateTime getTime() {
         return _time;
     }
 
+    @Override
+    public Optional<ZonedDateTime> getRequestTime() {
+        return _requestTime;
+    }
+
     @Override
     public ImmutableMap<String, ? extends Metric> getMetrics() {
         return _metrics;
@@ -101,13 +109,15 @@ private DefaultRecord(final Builder builder) {
         _metrics = builder._metrics;
         _id = builder._id;
         _time = builder._time;
+        _requestTime = Optional.ofNullable(builder._requestTime);
         _annotations = builder._annotations;
         _dimensions = builder._dimensions;
     }
 
     private final ImmutableMap<String, ? extends Metric> _metrics;
     private final String _id;
     private final ZonedDateTime _time;
+    private final Optional<ZonedDateTime> _requestTime;
     private final ImmutableMap<String, String> _annotations;
     private final ImmutableMap<String, String> _dimensions;
 
@@ -158,6 +168,17 @@ public Builder setTime(final ZonedDateTime value) {
             return this;
         }
 
+        /**
+         * The timestamp at which the record was received. Can be null.
+         *
+         * @param value The timestamp.
+         * @return This instance of <code>Builder</code>.
+         */
+        public Builder setRequestTime(final ZonedDateTime value) {
+            _requestTime = value;
+            return this;
+        }
+
         /**
          * The annotations <code>ImmutableMap</code>. Optional. Default is an empty
          * <code>ImmutableMap</code>. Cannot be null.
@@ -189,6 +210,7 @@ protected void reset() {
             _time = null;
             _annotations = ImmutableMap.of();
             _dimensions = ImmutableMap.of();
+            _requestTime = null;
         }
 
         @NotNull
@@ -198,6 +220,8 @@ protected void reset() {
         private String _id;
         @NotNull
         private ZonedDateTime _time;
+        @Nullable
+        private ZonedDateTime _requestTime;
         @NotNull
         private ImmutableMap<String, String> _annotations = ImmutableMap.of();
         @NotNull

src/main/java/com/arpnetworking/metrics/mad/model/Record.java

@@ -18,6 +18,7 @@
 import com.google.common.collect.ImmutableMap;
 
 import java.time.ZonedDateTime;
+import java.util.Optional;
 
 /**
  * The interface to a record. Records consistent of a timestamp, any number of
@@ -43,6 +44,13 @@ public interface Record {
      */
     ZonedDateTime getTime();
 
+    /**
+     * Gets the "received at" time stamp of the record.
+     *
+     * @return the time stamp
+     */
+    Optional<ZonedDateTime> getRequestTime();
+
     /**
      * Gets metrics.
      *

src/main/java/com/arpnetworking/metrics/mad/parsers/ProtobufV3ToRecordParser.java

@@ -68,10 +68,16 @@ public List<Record> parse(final HttpRequest data) throws ParsingException {
                 final long low = byteBuffer.getLong();
                 records.add(ThreadLocalBuilder.build(
                         DefaultRecord.Builder.class,
-                        builder -> builder.setId(new UUID(high, low).toString())
-                                .setTime(ZonedDateTime.ofInstant(Instant.ofEpochMilli(record.getEndMillisSinceEpoch()), ZoneOffset.UTC))
-                                .setDimensions(buildDimensions(record))
-                                .setMetrics(buildMetrics(record))));
+                        builder -> {
+                            builder.setId(new UUID(high, low).toString())
+                                    .setTime(ZonedDateTime.ofInstant(Instant.ofEpochMilli(record.getEndMillisSinceEpoch()), ZoneOffset.UTC))
+                                    .setDimensions(buildDimensions(record))
+                                    .setMetrics(buildMetrics(record));
+                            if (record.getRequestMillisSinceEpoch() != 0) {
+                                builder.setRequestTime(
+                                        ZonedDateTime.ofInstant(Instant.ofEpochMilli(record.getRequestMillisSinceEpoch()), ZoneOffset.UTC));
+                            }
+                        }));
             }
             return records;
         } catch (final InvalidProtocolBufferException e) {

src/main/java/com/arpnetworking/tsdcore/model/PeriodicData.java

@@ -25,6 +25,8 @@
 
 import java.time.Duration;
 import java.time.ZonedDateTime;
+import java.util.Optional;
+import javax.annotation.Nullable;
 
 /**
  * Contains the data for a specific period in time.
@@ -42,6 +44,10 @@ public ZonedDateTime getStart() {
         return _start;
     }
 
+    public Optional<ZonedDateTime> getMinRequestTime() {
+        return _minRequestTime;
+    }
+
     public Key getDimensions() {
         return _dimensions;
     }
@@ -64,7 +70,8 @@ public boolean equals(final Object object) {
         return Objects.equal(_data, other._data)
                 && Objects.equal(_dimensions, other._dimensions)
                 && Objects.equal(_period, other._period)
-                && Objects.equal(_start, other._start);
+                && Objects.equal(_start, other._start)
+                && Objects.equal(_minRequestTime, other._minRequestTime);
     }
 
     @Override
@@ -73,7 +80,8 @@ public int hashCode() {
                 _data,
                 _dimensions,
                 _period,
-                _start);
+                _start,
+                _minRequestTime);
     }
 
     @Override
@@ -84,18 +92,21 @@ public String toString() {
                 .add("Start", _start)
                 .add("Dimensions", _dimensions)
                 .add("Data", _data)
+                .add("MinRequestTime", _minRequestTime)
                 .toString();
     }
 
     private PeriodicData(final Builder builder) {
         _period = builder._period;
         _start = builder._start;
+        _minRequestTime = Optional.ofNullable(builder._minRequestTime);
         _dimensions = builder._dimensions;
         _data = builder._data;
     }
 
     private final Duration _period;
     private final ZonedDateTime _start;
+    private final Optional<ZonedDateTime> _minRequestTime;
     private final Key _dimensions;
     private final ImmutableMultimap<String, AggregatedData> _data;
 
@@ -133,6 +144,18 @@ public Builder setStart(final ZonedDateTime value) {
             return this;
         }
 
+        /**
+         * Set the earliest time at which data in this bucket was recorded. May be null.
+         *
+         * @param value The timestamp
+         * @return This <code>Builder</code> instance.
+         */
+        public Builder setMinRequestTime(@Nullable final ZonedDateTime value) {
+            _minRequestTime = value;
+            return this;
+        }
+
+
         /**
          * Set the dimensions. Required. Cannot be null.
          *
@@ -159,6 +182,7 @@ public Builder setData(final ImmutableMultimap<String, AggregatedData> value) {
         protected void reset() {
             _period = null;
             _start = null;
+            _minRequestTime = null;
             _dimensions = null;
             _data = ImmutableMultimap.of();
         }
@@ -167,6 +191,9 @@ protected void reset() {
         private Duration _period;
         @NotNull
         private ZonedDateTime _start;
+        @Nullable
+        private ZonedDateTime _minRequestTime;
+
         @NotNull
         private Key _dimensions;
         @NotNull

src/main/java/com/arpnetworking/tsdcore/sinks/MetricsDataSerializer.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright 2020 Dropbox Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.arpnetworking.tsdcore.sinks;
+
+import com.arpnetworking.metrics.aggregation.protocol.Messages;
+import com.arpnetworking.metrics.mad.model.AggregatedData;
+import com.arpnetworking.metrics.mad.model.statistics.HistogramStatistic;
+import com.arpnetworking.metrics.mad.model.statistics.Statistic;
+import com.arpnetworking.metrics.mad.model.statistics.StatisticFactory;
+import com.arpnetworking.tsdcore.model.AggregationMessage;
+import com.arpnetworking.tsdcore.model.PeriodicData;
+import com.google.protobuf.ByteString;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Utility class to serialize to metrics-aggregator-protocol protobuf messages.
+ *
+ * @author William Ehlhardt (whale at dropbox dot com)
+ */
+public final class MetricsDataSerializer {
+    private static final StatisticFactory STATISTIC_FACTORY = new StatisticFactory();
+    private static final Statistic EXPRESSION_STATISTIC = STATISTIC_FACTORY.getStatistic("expression");
+
+    /**
+     * Serialize a metric's data to a StatisticSetRecord.
+     *
+     * @param periodicData Originating PeriodicData
+     * @param metricName Name of metric being serialized.
+     * @param data Recorded metric data to serialize.
+     * @return StatisticSetRecord protobuf corresponding to the above.
+     */
+    public static Messages.StatisticSetRecord serializeMetricData(
+            final PeriodicData periodicData,
+            final String metricName,
+            final Collection<AggregatedData> data) {
+
+        // Create a statistic record set
+        final Messages.StatisticSetRecord.Builder builder = Messages.StatisticSetRecord.newBuilder()
+                .setMetric(metricName)
+                .setPeriod(periodicData.getPeriod().toString())
+                .setPeriodStart(periodicData.getStart().toString())
+                .setClientMinimumRequestTime(periodicData.getMinRequestTime().map(t -> t.toString()).orElse(""))
+                .putAllDimensions(periodicData.getDimensions().getParameters())
+                .setCluster(periodicData.getDimensions().getCluster())
+                .setService(periodicData.getDimensions().getService());
+
+        for (final AggregatedData datum : data) {
+            if (Objects.equals(EXPRESSION_STATISTIC, datum.getStatistic())) {
+                continue;
+            }
+
+            final String unit;
+            if (datum.getValue().getUnit().isPresent()) {
+                // TODO(ville): The protocol needs to support compound units.
+                unit = datum.getValue().getUnit().get().toString();
+            } else {
+                unit = "";
+            }
+
+            final Messages.StatisticRecord.Builder entryBuilder = builder.addStatisticsBuilder()
+                    .setStatistic(datum.getStatistic().getName())
+                    .setValue(datum.getValue().getValue())
+                    .setUnit(unit)
+                    .setUserSpecified(datum.getIsSpecified());
+
+            final ByteString supportingData = serializeSupportingData(datum);
+            if (supportingData != null) {
+                entryBuilder.setSupportingData(supportingData);
+            }
+            entryBuilder.build();
+        }
+
+        return builder.build();
+    }
+
+    private static ByteString serializeSupportingData(final AggregatedData datum) {
+        final Object data = datum.getSupportingData();
+        final ByteString byteString;
+        if (data instanceof HistogramStatistic.HistogramSupportingData) {
+            final HistogramStatistic.HistogramSupportingData histogramSupportingData = (HistogramStatistic.HistogramSupportingData) data;
+            final Messages.SparseHistogramSupportingData.Builder builder = Messages.SparseHistogramSupportingData.newBuilder();
+            final HistogramStatistic.HistogramSnapshot histogram = histogramSupportingData.getHistogramSnapshot();
+            final String unit;
+            if (histogramSupportingData.getUnit().isPresent()) {
+                // TODO(ville): The protocol needs to support compound units.
+                unit = histogramSupportingData.getUnit().get().toString();
+            } else {
+                unit = "";
+            }
+            builder.setUnit(unit);
+
+            for (final Map.Entry<Double, Long> entry : histogram.getValues()) {
+                builder.addEntriesBuilder()
+                        .setBucket(entry.getKey())
+                        .setCount(entry.getValue())
+                        .build();
+            }
+            byteString = ByteString.copyFrom(
+                    AggregationMessage.create(builder.build()).serializeToByteString().toArray());
+        } else {
+            return null;
+        }
+        return byteString;
+    }
+
+    private MetricsDataSerializer() {
+        throw new AssertionError("utility class should not be instantiated");
+    }
+}