skip prometheus histograms (#266)

* skip prometheus histograms

Author: BrandonArp

src/main/docker/kafka/Dockerfile

@@ -15,7 +15,7 @@
 # Original: https://github.com/spotify/docker-kafka
 
 # Kafka and Zookeeper
-FROM openjdk:8-jre-buster
+FROM openjdk:17-slim-bullseye
 
 ENV SCALA_VERSION 2.13
 ENV KAFKA_VERSION 2.7.1

src/main/docker/mad/Dockerfile

@@ -32,6 +32,8 @@ ENV JAVA_OPTS=""
 # Build
 RUN apt update && \
     apt install -y gosu openssl && \
+    rm -rf /var/lib/apt/lists/* && \
+    apt-get clean && \
     mkdir -p /opt/mad/lib/ext && \
     mkdir -p /opt/mad/logs && \
     mkdir -p /opt/mad/config/pipelines && \

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

@@ -114,32 +114,27 @@ ParseResult parseNameAndUnit(final String name) {
     @Override
     public List<Record> parse(final HttpRequest data) throws ParsingException {
         final List<Record> records = Lists.newArrayList();
-        final byte[] uncompressed;
-        try {
-            final byte[] input = data.getBody().toArray();
-            if (_outputDebugInfo) {
-                final int outputFile = _outputFileNumber.incrementAndGet();
-                if (outputFile < 10) {
-                    Files.write(Paths.get("prometheus_debug_" + outputFile), input);
-                }
-            }
-            uncompressed = Snappy.uncompress(input);
-        } catch (final IOException e) {
-            throw new ParsingException("Failed to decompress snappy stream", data.getBody().toArray(), e);
-        }
+        final byte[] uncompressed = decompress(data);
         try {
             final Remote.WriteRequest writeRequest = Remote.WriteRequest.parseFrom(uncompressed);
             for (final TimeSeries timeSeries : writeRequest.getTimeseriesList()) {
+                boolean skipSeries = false;
                 Optional<String> nameOpt = Optional.empty();
                 final ImmutableMap.Builder<String, String> dimensionsBuilder = ImmutableMap.builder();
                 for (final Types.Label label : timeSeries.getLabelsList()) {
                     if ("__name__".equals(label.getName())) {
                         final String value = label.getValue();
                         nameOpt = Optional.ofNullable(value);
+                    } else if ("le".equals(label.getName())) {
+                        skipSeries = true;
                     } else {
                         dimensionsBuilder.put(label.getName(), label.getValue());
                     }
                 }
+
+                if (skipSeries) {
+                    continue;
+                }
                 final ParseResult result = parseNameAndUnit(nameOpt.orElse("").trim());
                 final String metricName = result.getName();
                 if (metricName.isEmpty()) {
@@ -172,6 +167,23 @@ public List<Record> parse(final HttpRequest data) throws ParsingException {
         return records;
     }
 
+    private byte[] decompress(final HttpRequest data) throws ParsingException {
+        final byte[] uncompressed;
+        try {
+            final byte[] input = data.getBody().toArray();
+            if (_outputDebugInfo) {
+                final int outputFile = _outputFileNumber.incrementAndGet();
+                if (outputFile < 10) {
+                    Files.write(Paths.get("prometheus_debug_" + outputFile), input);
+                }
+            }
+            uncompressed = Snappy.uncompress(input);
+        } catch (final IOException e) {
+            throw new ParsingException("Failed to decompress snappy stream", data.getBody().toArray(), e);
+        }
+        return uncompressed;
+    }
+
     private ImmutableMap<String, ? extends Metric> createMetric(final String name, final Types.Sample sample, final Optional<Unit> unit) {
         final Metric metric = ThreadLocalBuilder.build(
                 DefaultMetric.Builder.class,