Create actor pools to parallelize ingest of metrics from a source (#262)

* add a pool argument for sources

* add inflight and queue size metrics to http post

Author: BrandonArp

src/main/java/com/arpnetworking/metrics/common/sources/ActorSource.java

@@ -20,9 +20,11 @@
 import akka.actor.PoisonPill;
 import akka.actor.Props;
 import akka.pattern.Patterns;
+import akka.routing.RoundRobinPool;
 import com.arpnetworking.steno.Logger;
 import com.arpnetworking.steno.LoggerFactory;
 import com.fasterxml.jackson.annotation.JacksonInject;
+import net.sf.oval.constraint.Min;
 import net.sf.oval.constraint.NotEmpty;
 import net.sf.oval.constraint.NotNull;
 
@@ -41,7 +43,7 @@ public abstract class ActorSource extends BaseSource {
     @Override
     public void start() {
         if (_actor == null) {
-            _actor = _actorSystem.actorOf(createProps(), _actorName);
+            _actor = _actorSystem.actorOf(new RoundRobinPool(_poolSize).props(createProps()), _actorName);
         }
     }
 
@@ -110,12 +112,14 @@ protected ActorSource(final Builder<?, ? extends ActorSource> builder) {
         super(builder);
         _actorName = builder._actorName;
         _actorSystem = builder._actorSystem;
+        _poolSize = builder._poolSize;
     }
 
     private ActorRef _actor = null;
 
     private final String _actorName;
     private final ActorSystem _actorSystem;
+    private final int _poolSize;
 
     private static final Duration SHUTDOWN_TIMEOUT = Duration.ofSeconds(1);
     private static final Logger LOGGER = LoggerFactory.getLogger(ActorSource.class);
@@ -159,9 +163,22 @@ public final B setActorSystem(final ActorSystem value) {
             return self();
         }
 
+        /**
+         * Sets the actor pool size.
+         * @param value Number of actors in the pool
+         * @return This instance of {@link Builder}
+         */
+        public final B setPoolSize(final Integer value) {
+            _poolSize = value;
+            return self();
+        }
+
         @NotNull
         @NotEmpty
         private String _actorName;
+        @NotNull
+        @Min(1)
+        private Integer _poolSize = 1;
         @JacksonInject
         private ActorSystem _actorSystem;
     }

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

@@ -106,6 +106,8 @@ public HttpPostSinkActor(
         _evictedRequestsName = "sinks/http_post/" + sink.getMetricSafeName() + "/evicted_requests";
         _requestLatencyName = "sinks/http_post/" + sink.getMetricSafeName() + "/request_latency";
         _inQueueLatencyName = "sinks/http_post/" + sink.getMetricSafeName() + "/queue_time";
+        _pendingRequestsQueueSizeName = "sinks/http_post/" + sink.getMetricSafeName() + "/queue_size";
+        _inflightRequestsCountName = "sinks/http_post/" + sink.getMetricSafeName() + "/inflight_count";
         _requestSuccessName = "sinks/http_post/" + sink.getMetricSafeName() + "/success";
         _responseStatusName = "sinks/http_post/" + sink.getMetricSafeName() + "/status";
         _samplesDroppedName = "sinks/http_post/" + sink.getMetricSafeName() + "/samples_dropped";
@@ -240,6 +242,7 @@ private void processEmitAggregation(final EmitAggregation emitMessage) {
                 .addData("dataSize", periodicData.getData().size())
                 .addContext("actor", self())
                 .log();
+        _periodicMetrics.recordGauge(_inflightRequestsCountName, _inflightRequestsCount);
 
         if (!periodicData.getData().isEmpty()) {
             final Collection<RequestEntry.Builder> requestEntryBuilders = _sink.createRequests(_client, periodicData);
@@ -262,6 +265,8 @@ private void processEmitAggregation(final EmitAggregation emitMessage) {
                         .log();
             }
 
+            _periodicMetrics.recordGauge(_pendingRequestsQueueSizeName, _pendingRequests.size());
+
             if (_spreadingDelayMillis > 0) {
                 // If we don't currently have anything in-flight, we'll need to wait the spreading duration.
                 if (!_waiting && pendingWasEmpty) {
@@ -362,6 +367,8 @@ private void fireNextRequest() {
     private final String _evictedRequestsName;
     private final String _requestLatencyName;
     private final String _inQueueLatencyName;
+    private final String _pendingRequestsQueueSizeName;
+    private final String _inflightRequestsCountName;
     private final String _requestSuccessName;
     private final String _responseStatusName;
     private final String _samplesDroppedName;