Actor reap (#238)

* Reap period worker actors after idle period.

* Convert PeriodWorker to timers().

* Add super start/stop.

* Call parent start/stop methods.

* Add comment about idle worker period.

Author: vjkoskela

config/pipelines/pipeline.conf

@@ -9,6 +9,10 @@ name=example_pipeline
 #    "PT1M"
 #]
 
+# Idle Timeout
+# ~~~~
+#idleTimeout="PT5M"
+
 # Statistics
 # ~~~~
 #timerStatistics=[

pom.xml

@@ -71,7 +71,7 @@
 
   <properties>
     <!--Dependency versions-->
-    <akka.version>2.5.16</akka.version>
+    <akka.version>2.5.32</akka.version>
     <akka.http.version>10.1.5</akka.http.version>
     <apache.httpclient.version>4.5.6</apache.httpclient.version>
     <apache.httpcore.version>4.4.10</apache.httpcore.version>
@@ -99,6 +99,7 @@
     <oval.version>1.90</oval.version>
     <protobuf.version>3.8.0</protobuf.version>
     <scala.version>2.11</scala.version>
+    <scala.java8.compat.version>0.7.0</scala.java8.compat.version>
     <scala.library.version>2.11.12</scala.library.version>
     <slf4j.version>1.7.25</slf4j.version>
     <snappy.version>1.1.7.2</snappy.version>
@@ -758,6 +759,11 @@
       <artifactId>scala-library</artifactId>
       <version>${scala.library.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.scala-lang.modules</groupId>
+      <artifactId>scala-java8-compat_${scala.version}</artifactId>
+      <version>${scala.java8.compat.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpclient</artifactId>

src/main/java/com/arpnetworking/http/Routes.java

@@ -59,6 +59,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.Resources;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import scala.compat.java8.FutureConverters;
 import scala.concurrent.duration.FiniteDuration;
 
 import java.util.Objects;
@@ -251,8 +252,9 @@ private CompletionStage<HttpResponse> process(final HttpRequest request) {
     }
 
     private CompletionStage<HttpResponse> dispatchHttpRequest(final HttpRequest request, final String actorName) {
-        final CompletionStage<ActorRef> refFuture = _actorSystem.actorSelection(actorName)
-                .resolveOneCS(FiniteDuration.create(1, TimeUnit.SECONDS));
+        final CompletionStage<ActorRef> refFuture = FutureConverters.toJava(
+                _actorSystem.actorSelection(actorName)
+                        .resolveOne(FiniteDuration.create(1, TimeUnit.SECONDS)));
         return refFuture.thenCompose(
                 ref -> {
                     final CompletableFuture<HttpResponse> response = new CompletableFuture<>();

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

@@ -42,9 +42,14 @@
 import com.arpnetworking.http.RequestReply;
 import com.arpnetworking.metrics.common.parsers.Parser;
 import com.arpnetworking.metrics.common.parsers.exceptions.ParsingException;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
+import com.arpnetworking.metrics.mad.model.Metric;
 import com.arpnetworking.metrics.mad.model.Record;
+import com.arpnetworking.metrics.mad.model.statistics.StatisticFactory;
 import com.arpnetworking.steno.Logger;
 import com.arpnetworking.steno.LoggerFactory;
+import com.arpnetworking.tsdcore.model.CalculatedValue;
+import com.fasterxml.jackson.annotation.JacksonInject;
 import com.google.common.collect.ImmutableMultimap;
 import net.sf.oval.constraint.NotNull;
 
@@ -74,8 +79,10 @@ protected Props createProps() {
     protected HttpSource(final Builder<?, ? extends HttpSource> builder) {
         super(builder);
         _parser = builder._parser;
+        _periodicMetrics = builder._periodicMetrics;
     }
 
+    private final PeriodicMetrics _periodicMetrics;
     private final Parser<List<Record>, com.arpnetworking.metrics.mad.model.HttpRequest> _parser;
 
     /**
@@ -98,6 +105,7 @@ public Receive createReceive() {
                     .match(RequestReply.class, requestReply -> {
                         // TODO(barp): Fix the ugly HttpRequest cast here due to java vs scala dsl
                         akka.stream.javadsl.Source.single(requestReply.getRequest())
+                                .log("http source stream failure")
                                 .via(_processGraph)
                                 .toMat(_sink, Keep.right())
                                 .run(_materializer)
@@ -127,6 +135,8 @@ public Receive createReceive() {
          * @param source The {@link HttpSource} to send notifications through.
          */
         /* package private */ Actor(final HttpSource source) {
+            _periodicMetrics = source._periodicMetrics;
+            _metricSafeName = source.getMetricSafeName();
             _parser = source._parser;
             _sink = Sink.foreach(source::notify);
             _materializer = ActorMaterializer.create(
@@ -193,14 +203,36 @@ private static com.arpnetworking.metrics.mad.model.HttpRequest mapModel(
 
         private List<Record> parseRecords(final com.arpnetworking.metrics.mad.model.HttpRequest request)
                 throws ParsingException {
-            return _parser.parse(request);
+            final List<Record> records = _parser.parse(request);
+            long samples = 0;
+            for (final Record record : records) {
+                for (final Metric metric : record.getMetrics().values()) {
+                    samples += metric.getValues().size();
+                    final List<CalculatedValue<?>> countStatistic =
+                            metric.getStatistics().get(STATISTIC_FACTORY.getStatistic("count"));
+                    if (countStatistic != null) {
+                        samples += countStatistic.stream()
+                                .map(s -> s.getValue().getValue())
+                                .reduce(Double::sum)
+                                .orElse(0.0d);
+                    }
+                }
+            }
+            _periodicMetrics.recordGauge(
+                    String.format("sources/http/%s/metric_samples", _metricSafeName),
+                    samples);
+
+            return records;
         }
 
+        private final PeriodicMetrics _periodicMetrics;
+        private final String _metricSafeName;
         private final Sink<Record, CompletionStage<Done>> _sink;
         private final Parser<List<Record>, com.arpnetworking.metrics.mad.model.HttpRequest> _parser;
         private final Materializer _materializer;
         private final Graph<FlowShape<HttpRequest, Record>, NotUsed> _processGraph;
 
+        private static final StatisticFactory STATISTIC_FACTORY = new StatisticFactory();
         private static final Logger BAD_REQUEST_LOGGER =
                 LoggerFactory.getRateLimitLogger(HttpSource.class, Duration.ofSeconds(30));
     }
@@ -235,7 +267,22 @@ public B setParser(final Parser<List<Record>, com.arpnetworking.metrics.mad.mode
             return self();
         }
 
+        /**
+         * Sets the periodic metrics instance.
+         *
+         * @param value The periodic metrics.
+         * @return This instance of {@link Builder}
+         */
+        public final B setPeriodicMetrics(final PeriodicMetrics value) {
+            _periodicMetrics = value;
+            return self();
+        }
+
         @NotNull
         private Parser<List<Record>, com.arpnetworking.metrics.mad.model.HttpRequest> _parser;
+
+        @NotNull
+        @JacksonInject
+        private PeriodicMetrics _periodicMetrics;
     }
 }