Record samplesDropped and samplesSent metrics in HttpSink (#133)

* record samplesSent and samplesDropped metrics in KDB sink

Author: joeyjackson

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

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2020 Dropbox
+ *
+ * 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.model;
+
+import com.arpnetworking.commons.builder.OvalBuilder;
+import net.sf.oval.constraint.NotNull;
+import org.asynchttpclient.Request;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/**
+ * Contains the info for a http request.
+ *
+ * @author Qinyan Li (lqy520s at hotmail dot com)
+ */
+public final class RequestEntry {
+    public Request getRequest() {
+        return _request;
+    }
+
+    public Instant getEnterTime() {
+        return _enterTime;
+    }
+
+    public Optional<Long> getPopulationSize() {
+        return _populationSize;
+    }
+
+    private RequestEntry(final Builder builder) {
+        _request = builder._request;
+        _enterTime = builder._enterTime;
+        _populationSize = builder._populationSize;
+    }
+
+    private final Request _request;
+    private final Instant _enterTime;
+    private final Optional<Long> _populationSize;
+
+    /**
+     * {@link com.arpnetworking.commons.builder.Builder} implementation for
+     * {@link RequestEntry}.
+     *
+     * TODO(ville): Convert RequestEntry.Builder would be a ThreadLocalBuilder
+     * See comments in HttpPostSink:createRequests
+     */
+    public static final class Builder extends OvalBuilder<RequestEntry> {
+
+        /**
+         * Public constructor.
+         */
+        public Builder() {
+            super(RequestEntry::new);
+        }
+
+        /**
+         * Set the request. Required. Cannot be null.
+         *
+         * @param value The request.
+         * @return This {@link Builder} instance.
+         */
+        public Builder setRequest(final Request value) {
+            _request = value;
+            return this;
+        }
+
+        /**
+         * Set the time when the request enter the pending request queue. Required. Cannot be null.
+         *
+         * @param value The enter time.
+         * @return This {@link Builder} instance.
+         */
+        public Builder setEnterTime(final Instant value) {
+            _enterTime = value;
+            return this;
+        }
+
+        /**
+         * Set the population size of the request. Optional. Cannot be null.
+         *
+         * @param value The population size.
+         * @return This {@link Builder} instance.
+         */
+        public Builder setPopulationSize(final Optional<Long> value) {
+            _populationSize = value;
+            return this;
+        }
+
+        @NotNull
+        private Request _request;
+        @NotNull
+        private Instant _enterTime;
+        @NotNull
+        private Optional<Long> _populationSize = Optional.empty();
+    }
+}
+

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

@@ -37,6 +37,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 
 /**
  * Publishes aggregations to Data Dog. This class is thread safe.
@@ -63,7 +64,7 @@ public Object toLogValue() {
     }
 
     @Override
-    protected Collection<byte[]> serialize(final PeriodicData periodicData) {
+    protected Collection<SerializedDatum> serialize(final PeriodicData periodicData) {
         final String period = periodicData.getPeriod().toString();
         final long timestamp = (periodicData.getStart().toInstant().toEpochMilli()
                 + periodicData.getPeriod().toMillis()) / 1000;
@@ -94,7 +95,9 @@ protected Collection<byte[]> serialize(final PeriodicData periodicData) {
                     .log();
             return Collections.emptyList();
         }
-        return Collections.singletonList(dataDogDataAsJson.getBytes(Charsets.UTF_8));
+        return Collections.singletonList(new SerializedDatum(
+                dataDogDataAsJson.getBytes(Charsets.UTF_8),
+                Optional.empty()));
     }
 
     private static List<String> createTags(final PeriodicData periodicData, final AggregatedData datum) {

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

@@ -26,6 +26,7 @@
 import com.arpnetworking.steno.Logger;
 import com.arpnetworking.steno.LoggerFactory;
 import com.arpnetworking.tsdcore.model.PeriodicData;
+import com.arpnetworking.tsdcore.model.RequestEntry;
 import com.fasterxml.jackson.annotation.JacksonInject;
 import com.google.common.collect.Lists;
 import net.sf.oval.constraint.CheckWith;
@@ -43,6 +44,7 @@
 import java.net.URI;
 import java.time.Duration;
 import java.util.Collection;
+import java.util.Optional;
 import java.util.function.Function;
 
 /**
@@ -67,11 +69,6 @@ public void close() {
         _sinkActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
     }
 
-    /**
-     * Generate a Steno log compatible representation.
-     *
-     * @return Steno log compatible representation.
-     */
     @LogValue
     @Override
     public Object toLogValue() {
@@ -107,15 +104,21 @@ protected Request createRequest(final AsyncHttpClient client, final byte[] seria
      * @param periodicData The {@link PeriodicData} to be serialized.
      * @return The {@link Request} instance to execute.
      */
-    protected Collection<Request> createRequests(
+    protected Collection<RequestEntry.Builder> createRequests(
             final AsyncHttpClient client,
             final PeriodicData periodicData) {
-        final Collection<byte[]> serializedData = serialize(periodicData);
-        final Collection<Request> requests = Lists.newArrayListWithExpectedSize(serializedData.size());
-        for (final byte[] serializedDatum : serializedData) {
-            requests.add(createRequest(client, serializedDatum));
+        final Collection<SerializedDatum> serializedData = serialize(periodicData);
+        final Collection<RequestEntry.Builder> requestEntryBuilders = Lists.newArrayListWithExpectedSize(serializedData.size());
+        for (final SerializedDatum serializedDatum : serializedData) {
+            // TODO(ville): Convert RequestEntry.Builder would be a ThreadLocalBuilder
+            // Unfortunately, the split builder logic across HttpPostSink and
+            // HttpSinkActor does not permit this as-is. The logic would need
+            // to be refactored to permit the use of a TLB.
+            requestEntryBuilders.add(new RequestEntry.Builder()
+                    .setRequest(createRequest(client, serializedDatum.getDatum()))
+                    .setPopulationSize(serializedDatum.getPopulationSize()));
         }
-        return requests;
+        return requestEntryBuilders;
     }
 
     /**
@@ -169,7 +172,7 @@ protected Duration getRetryMaximumDelay() {
      * @param periodicData The {@link PeriodicData} to be serialized.
      * @return The serialized representation of {@link PeriodicData}.
      */
-    protected abstract Collection<byte[]> serialize(PeriodicData periodicData);
+    protected abstract Collection<SerializedDatum> serialize(PeriodicData periodicData);
 
     /**
      * Protected constructor.
@@ -376,4 +379,22 @@ public boolean isSatisfied(final Object validatedObject, final Object value) {
             }
         }
     }
+
+    static final class SerializedDatum {
+        SerializedDatum(final byte[] datum, final Optional<Long> populationSize) {
+            _datum = datum;
+            _populationSize = populationSize;
+        }
+
+        public byte[] getDatum() {
+            return _datum;
+        }
+
+        public Optional<Long> getPopulationSize() {
+            return _populationSize;
+        }
+
+        private final byte[] _datum;
+        private final Optional<Long> _populationSize;
+    }
 }