Add start/stop/restart metrics to bucket aggregator actor. (#230)

Author: vjkoskela

src/main/java/com/arpnetworking/metrics/mad/Aggregator.java

@@ -26,6 +26,7 @@
 import com.arpnetworking.commons.observer.Observable;
 import com.arpnetworking.commons.observer.Observer;
 import com.arpnetworking.logback.annotations.LogValue;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
 import com.arpnetworking.metrics.mad.model.Record;
 import com.arpnetworking.metrics.mad.model.statistics.Statistic;
 import com.arpnetworking.steno.LogValueMapFactory;
@@ -167,7 +168,7 @@ private List<ActorRef> createActors(final Key key) {
                     .setDependentStatistics(_cachedDependentStatistics)
                     .setPeriod(period)
                     .setSink(_sink);
-            periodWorkerList.add(_actorSystem.actorOf(Props.create(PeriodWorker.class, period, bucketBuilder)));
+            periodWorkerList.add(_actorSystem.actorOf(Props.create(PeriodWorker.class, period, bucketBuilder, _periodicMetrics)));
         }
         LOGGER.debug()
                 .setMessage("Created period worker actors")
@@ -187,6 +188,7 @@ private ImmutableSet<Statistic> computeDependentStatistics(final ImmutableSet<St
 
     private Aggregator(final Builder builder) {
         _actorSystem = builder._actorSystem;
+        _periodicMetrics = builder._periodicMetrics;
         _periods = ImmutableSet.copyOf(builder._periods);
         _sink = builder._sink;
         _specifiedCounterStatistics = ImmutableSet.copyOf(builder._counterStatistics);
@@ -235,6 +237,7 @@ public Optional<ImmutableSet<Statistic>> load(final String metric) throws Except
     }
 
     private final ActorSystem _actorSystem;
+    private final PeriodicMetrics _periodicMetrics;
     private final ImmutableSet<Duration> _periods;
     private final Sink _sink;
     private final ImmutableSet<Statistic> _specifiedTimerStatistics;
@@ -275,6 +278,17 @@ public Builder setActorSystem(final ActorSystem value) {
             return this;
         }
 
+        /**
+         * Set the {@link PeriodicMetrics}. Cannot be null.
+         *
+         * @param value The {@link PeriodicMetrics}.
+         * @return This {@link Builder} instance.
+         */
+        public Builder setPeriodicMetrics(final PeriodicMetrics value) {
+            _periodicMetrics = value;
+            return this;
+        }
+
         /**
          * Set the sink. Cannot be null or empty.
          *
@@ -345,6 +359,8 @@ public Builder setStatistics(final Map<String, Set<Statistic>> value) {
         @NotNull
         private ActorSystem _actorSystem;
         @NotNull
+        private PeriodicMetrics _periodicMetrics;
+        @NotNull
         private Sink _sink;
         @NotNull
         private Set<Duration> _periods;

src/main/java/com/arpnetworking/metrics/mad/Main.java

@@ -630,7 +630,8 @@ private PeriodicMetrics providePeriodicMetrics(
                     .build();
             final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
                     r -> new Thread(r, "PeriodicMetricsCloser"));
-            executor.scheduleAtFixedRate(periodicMetrics, 500, 500, TimeUnit.MILLISECONDS);
+            final long offsetMillis = 1250 - (System.currentTimeMillis() % 1000);;
+            executor.scheduleAtFixedRate(periodicMetrics, offsetMillis, 1000, TimeUnit.MILLISECONDS);
 
             // Register to shutdown the executor when the Guice stack is shutdown.
             lifecycle.registerShutdown(() -> {

src/main/java/com/arpnetworking/metrics/mad/PeriodWorker.java

@@ -18,6 +18,7 @@
 import akka.actor.AbstractActor;
 import akka.actor.Cancellable;
 import com.arpnetworking.logback.annotations.LogValue;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
 import com.arpnetworking.metrics.mad.model.Record;
 import com.arpnetworking.steno.LogValueMapFactory;
 import com.arpnetworking.steno.Logger;
@@ -46,9 +47,13 @@
      * Public constructor. Since this is an {@code Actor} this method should not be
      * called directly, but instead you should use {@code Props}.
      */
-    PeriodWorker(final Duration period, final Bucket.Builder bucketBuilder) {
+    PeriodWorker(
+            final Duration period,
+            final Bucket.Builder bucketBuilder,
+            final PeriodicMetrics periodicMetrics) {
         _period = period;
         _bucketBuilder = bucketBuilder;
+        _periodicMetrics = periodicMetrics;
         _hasRotateScheduled = false;
     }
 
@@ -60,6 +65,7 @@ public void preStart() {
         // creating a race condition. Note that period worker clean-up did not exist
         // in the previous implementation either (not that it shouldn't; just one
         // problem at a time).
+        _periodicMetrics.recordCounter("actors/period_worker/started", 1);
     }
 
     @Override
@@ -73,6 +79,12 @@ public void postStop() {
                     .addData("cancelResult", cancelResult)
                     .log();
         }
+        _periodicMetrics.recordCounter("actors/period_worker/stopped", 1);
+    }
+
+    @Override
+    public void preRestart(final Throwable reason, final Optional<Object> message) throws Exception {
+        _periodicMetrics.recordCounter("actors/period_worker/restarted", 1);
     }
 
     @Override
@@ -259,6 +271,7 @@ private void processRecord(final Record record) {
     private final Bucket.Builder _bucketBuilder;
     private final NavigableMap<ZonedDateTime, Bucket> _bucketsByStart = new TreeMap<>();
     private final NavigableMap<ZonedDateTime, List<Bucket>> _bucketsByExpiration = new TreeMap<>();
+    private final PeriodicMetrics _periodicMetrics;
 
     private static final Logger LOGGER = LoggerFactory.getLogger(PeriodWorker.class);
     private static final Duration MINIMUM_ROTATION_CHECK_INTERVAL = Duration.ofMillis(100);

src/main/java/com/arpnetworking/metrics/mad/Pipeline.java

@@ -66,6 +66,7 @@ public synchronized void launch() {
 
         final Aggregator aggregator = new Aggregator.Builder()
                 .setActorSystem(_pipelineConfiguration.getActorSystem())
+                .setPeriodicMetrics(_pipelineConfiguration.getPeriodicMetrics())
                 .setPeriods(_pipelineConfiguration.getPeriods())
                 .setTimerStatistics(_pipelineConfiguration.getTimerStatistics())
                 .setCounterStatistics(_pipelineConfiguration.getCounterStatistics())

src/main/java/com/arpnetworking/metrics/mad/configuration/PipelineConfiguration.java

@@ -21,6 +21,7 @@
 import com.arpnetworking.logback.annotations.Loggable;
 import com.arpnetworking.metrics.common.kafka.ConsumerDeserializer;
 import com.arpnetworking.metrics.common.sources.Source;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
 import com.arpnetworking.metrics.mad.model.statistics.Statistic;
 import com.arpnetworking.metrics.mad.model.statistics.StatisticDeserializer;
 import com.arpnetworking.metrics.mad.model.statistics.StatisticFactory;
@@ -86,6 +87,10 @@ public ActorSystem getActorSystem() {
         return _actorSystem;
     }
 
+    public PeriodicMetrics getPeriodicMetrics() {
+        return _periodicMetrics;
+    }
+
     public String getName() {
         return _name;
     }
@@ -123,6 +128,7 @@ public String toString() {
         return MoreObjects.toStringHelper(this)
                 .add("id", Integer.toHexString(System.identityHashCode(this)))
                 .add("ActorSystem", _actorSystem)
+                .add("PeriodicMetrics", _periodicMetrics)
                 .add("Name", _name)
                 .add("Sources", _sources)
                 .add("Sinks", _sinks)
@@ -135,6 +141,7 @@ public String toString() {
 
     private PipelineConfiguration(final Builder builder) {
         _actorSystem = builder._actorSystem;
+        _periodicMetrics = builder._periodicMetrics;
         _name = builder._name;
         _sources = ImmutableList.copyOf(builder._sources);
         _sinks = ImmutableList.copyOf(builder._sinks);
@@ -146,6 +153,7 @@ private PipelineConfiguration(final Builder builder) {
     }
 
     private final ActorSystem _actorSystem;
+    private final PeriodicMetrics _periodicMetrics;
     private final String _name;
     private final ImmutableList<Source> _sources;
     private final ImmutableList<Sink> _sinks;
@@ -183,6 +191,17 @@ public Builder setActorSystem(final ActorSystem value) {
             return this;
         }
 
+        /**
+         * The {@link PeriodicMetrics}. Cannot be null.
+         *
+         * @param value The {@link PeriodicMetrics}.
+         * @return This instance of {@link Builder}.
+         */
+        public Builder setPeriodicMetrics(final PeriodicMetrics value) {
+            _periodicMetrics = value;
+            return this;
+        }
+
         /**
          * The name of the pipeline. Cannot be null or empty.
          *
@@ -280,6 +299,9 @@ public Builder setStatistics(final Map<String, Set<Statistic>> value) {
         @JacksonInject
         private ActorSystem _actorSystem;
         @NotNull
+        @JacksonInject
+        private PeriodicMetrics _periodicMetrics;
+        @NotNull
         @NotEmpty
         private String _name;
         @NotNull

src/test/java/com/arpnetworking/metrics/mad/AggregatorTest.java

@@ -18,6 +18,7 @@
 import akka.actor.ActorSystem;
 import com.arpnetworking.commons.observer.Observable;
 import com.arpnetworking.commons.observer.Observer;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
 import com.arpnetworking.metrics.mad.model.AggregatedData;
 import com.arpnetworking.metrics.mad.model.DefaultMetric;
 import com.arpnetworking.metrics.mad.model.DefaultQuantity;
@@ -68,6 +69,7 @@ public void setUp() {
         _actorSystem = ActorSystem.create();
        _aggregator = new Aggregator.Builder()
                 .setActorSystem(_actorSystem)
+                .setPeriodicMetrics(_periodicMetrics)
                 .setSink(_sink)
                 .setCounterStatistics(Collections.singleton(MAX_STATISTIC))
                 .setTimerStatistics(Collections.singleton(MAX_STATISTIC))
@@ -404,6 +406,8 @@ private List<AggregatedData> getCapturedData(
     private ArgumentCaptor<PeriodicData> _periodicDataCaptor;
     @Mock
     private Sink _sink;
+    @Mock
+    private PeriodicMetrics _periodicMetrics;
 
     private static final StatisticFactory STATISTIC_FACTORY = new StatisticFactory();
     private static final Statistic MAX_STATISTIC = STATISTIC_FACTORY.getStatistic("max");