periodically sample http sinks for metrics about queue size and inflight

Author: BrandonArp

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

@@ -15,7 +15,7 @@
  */
 package com.arpnetworking.tsdcore.sinks;
 
-import akka.actor.AbstractActor;
+import akka.actor.AbstractActorWithTimers;
 import akka.actor.Props;
 import akka.actor.Status;
 import akka.pattern.Patterns;
@@ -51,7 +51,7 @@
  *
  * @author Brandon Arp (brandon dot arp at inscopemetrics dot com)
  */
-public class HttpSinkActor extends AbstractActor {
+public class HttpSinkActor extends AbstractActorWithTimers {
     /**
      * Factory method to create a Props.
      *
@@ -120,6 +120,7 @@ public HttpSinkActor(
         _httpSinkAttemptsName = "sinks/http_post/" + _sink.getMetricSafeName() + "/attempts";
         _samplesSentName = "sinks/http_post/" + sink.getMetricSafeName() + "/samples_sent";
         _samplesDroppedName = "sinks/http_post/" + _sink.getMetricSafeName() + "/samples_dropped";
+        timers().startTimerAtFixedRate("metrics", SampleMetrics.INSTANCE, Duration.ofSeconds(1), Duration.ofSeconds(1));
     }
 
     @Override
@@ -244,6 +245,10 @@ public Receive createReceive() {
                             .setThrowable(message.cause())
                             .log();
                 })
+                .match(SampleMetrics.class, message -> {
+                    _periodicMetrics.recordCounter(_pendingRequestsQueueSizeName, _pendingRequests.size());
+                    _periodicMetrics.recordCounter(_inflightRequestsCountName, _inflightRequestsCount);
+                })
                 .matchAny(message -> {
                     LOGGER.error()
                             .setMessage("Unexpected message")
@@ -654,4 +659,8 @@ public void onThrowable(final Throwable throwable) {
 
         private final CompletableFuture<Response> _promise;
     }
+    private static final class SampleMetrics {
+        private SampleMetrics() { }
+        private static final SampleMetrics INSTANCE = new SampleMetrics();
+    }
 }