Add start/stop/restart metrics to streaming aggregator actor. (#141)

* Add start/stop/restart metrics to streaming aggregator actor.

* Removed double semi-colon.

* Increase test timeout.

Author: vjkoskela

pom.xml

@@ -133,6 +133,7 @@
     <logback.steno.version>1.18.0</logback.steno.version>
     <log4j.over.slf4j.version>1.7.22</log4j.over.slf4j.version>
     <metrics.client.version>0.11.3</metrics.client.version>
+    <metrics.client.incubator.version>0.11.2</metrics.client.incubator.version>
     <metrics.jvm.extra.version>0.11.2</metrics.jvm.extra.version>
     <metrics.http.extra.version>0.11.2</metrics.http.extra.version>
     <metrics.aggregator.protocol.version>1.0.10</metrics.aggregator.protocol.version>
@@ -583,6 +584,18 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>com.arpnetworking.metrics.extras</groupId>
+      <artifactId>incubator-extra</artifactId>
+      <version>${metrics.client.incubator.version}</version>
+      <exclusions>
+        <!-- TODO(ville): Remove suppression once findbugs to spotbugs migration is complete -->
+        <exclusion>
+          <groupId>com.google.code.findbugs</groupId>
+          <artifactId>findbugs-annotations</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
     <dependency>
       <groupId>com.arpnetworking.metrics.extras</groupId>
       <artifactId>jvm-extra</artifactId>

src/main/java/com/arpnetworking/clusteraggregator/AppShutdown.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2017 Inscope Metrics, Inc
+ *
+ * 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.clusteraggregator;
+
+import com.arpnetworking.steno.Logger;
+import com.arpnetworking.steno.LoggerFactory;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
+
+/**
+ * Assists Guice-instantiated classes in shutting down.
+ *
+ * @author Brandon Arp (brandon dot arp at inscopemetrics dot io)
+ */
+public class AppShutdown implements LifecycleRegistration {
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void registerShutdown(final Supplier<CompletionStage<Void>> callback) {
+        _shutdownCallbacks.offer(callback);
+    }
+
+    /**
+     * Executes all of the registered shutdown hooks and waits for them to complete.
+     */
+    public void shutdown() {
+        final CompletableFuture<?>[] completionStages = _shutdownCallbacks.stream()
+                .map(Supplier::get)
+                .map(CompletionStage::toCompletableFuture)
+                .toArray(CompletableFuture[]::new);
+
+        try {
+            CompletableFuture.allOf(completionStages).get();
+        } catch (final InterruptedException | ExecutionException e) {
+            LOGGER.error()
+                    .setMessage("Exception in application shutdown")
+                    .setThrowable(e)
+                    .log();
+        }
+    }
+
+    private final ConcurrentLinkedQueue<Supplier<CompletionStage<Void>>> _shutdownCallbacks = new ConcurrentLinkedQueue<>();
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(AppShutdown.class);
+}

src/main/java/com/arpnetworking/clusteraggregator/GuiceModule.java

@@ -54,6 +54,8 @@
 import com.arpnetworking.metrics.Sink;
 import com.arpnetworking.metrics.impl.ApacheHttpSink;
 import com.arpnetworking.metrics.impl.TsdMetricsFactory;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
+import com.arpnetworking.metrics.incubator.impl.TsdPeriodicMetrics;
 import com.arpnetworking.utility.ActorConfigurator;
 import com.arpnetworking.utility.ConfiguredLaunchableFactory;
 import com.arpnetworking.utility.Database;
@@ -86,7 +88,11 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
 
 /**
@@ -100,14 +106,17 @@ public class GuiceModule extends AbstractModule {
      * Public constructor.
      *
      * @param configuration The configuration.
+     * @param shutdown The shutdown hook.
      */
-    public GuiceModule(final ClusterAggregatorConfiguration configuration) {
+    public GuiceModule(final ClusterAggregatorConfiguration configuration, final LifecycleRegistration shutdown) {
         _configuration = configuration;
+        _shutdown = shutdown;
     }
 
     @Override
     protected void configure() {
         bind(ClusterAggregatorConfiguration.class).toInstance(_configuration);
+        bind(LifecycleRegistration.class).toInstance(_shutdown);
 
         for (final Map.Entry<String, DatabaseConfiguration> entry : _configuration.getDatabaseConfigurations().entrySet()) {
             bind(Database.class)
@@ -314,6 +323,24 @@ private ActorRef provideJvmMetricsCollector(final ActorSystem system, final Metr
         return system.actorOf(JvmMetricsCollector.props(_configuration.getJvmMetricsCollectionInterval(), metricsFactory));
     }
 
+    @Provides
+    @Singleton
+    @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
+    private PeriodicMetrics providePeriodicMetrics(final MetricsFactory metricsFactory, final LifecycleRegistration lifecycle) {
+        final TsdPeriodicMetrics periodicMetrics = new TsdPeriodicMetrics.Builder()
+                .setMetricsFactory(metricsFactory)
+                .build();
+        final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
+                r -> new Thread(r, "PeriodicMetricsCloser"));
+        final long offsetMillis = 1250 - (System.currentTimeMillis() % 1000);
+        executor.scheduleAtFixedRate(periodicMetrics, offsetMillis, 1000, TimeUnit.MILLISECONDS);
+        lifecycle.registerShutdown(() -> {
+            executor.shutdown();
+            return CompletableFuture.completedFuture(null);
+        });
+        return periodicMetrics;
+    }
+
     @Provides
     @Singleton
     @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
@@ -414,6 +441,7 @@ static List<Sink> createSinks(final ImmutableList<JsonNode> monitoringSinks) {
     }
 
     private final ClusterAggregatorConfiguration _configuration;
+    private final LifecycleRegistration _shutdown;
 
     private static final String HOCON_FILE_EXTENSION = ".conf";
     private static final ObjectMapper OBJECT_MAPPER = ObjectMapperFactory.getInstance();

src/main/java/com/arpnetworking/clusteraggregator/LifecycleRegistration.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2017 Inscope Metrics, Inc
+ *
+ * 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.clusteraggregator;
+
+import java.util.concurrent.CompletionStage;
+import java.util.function.Supplier;
+
+/**
+ * Provides a way to register for lifecycle events, ensuring that Guice-instantiated classes can be notified.
+ *
+ * @author Brandon Arp (brandon dot arp at inscopemetrics dot io)
+ */
+public interface LifecycleRegistration {
+    /**
+     * Registers a method to be called when the application is shutting down.  The shutdown
+     * will wait for all CompletionStages to complete before shutting down.
+     *
+     * @param callback The callback to register
+     */
+    void registerShutdown(Supplier<CompletionStage<Void>> callback);
+}

src/main/java/com/arpnetworking/clusteraggregator/Main.java

@@ -149,10 +149,12 @@ public synchronized void launch() {
     public synchronized void shutdown() {
         shutdownAkka();
         shutdownDatabases();
+        shutdownGuice();
     }
 
     private Injector launchGuice() {
-        return Guice.createInjector(new GuiceModule(_configuration));
+        _guiceAppShutdown = new AppShutdown();
+        return Guice.createInjector(new GuiceModule(_configuration, _guiceAppShutdown));
     }
 
     private void launchDatabases(final Injector injector) {
@@ -211,6 +213,13 @@ private void launchAkka(final Injector injector) {
         _system = injector.getInstance(ActorSystem.class);
     }
 
+    private void shutdownGuice() {
+        LOGGER.info().setMessage("Stopping guice").log();
+        if (_guiceAppShutdown != null) {
+            _guiceAppShutdown.shutdown();
+        }
+    }
+
     private void shutdownDatabases() {
         for (final Database database : _databases) {
             LOGGER.info()
@@ -259,6 +268,7 @@ private static Builder<? extends JsonNodeSource> getFileSourceBuilder(
 
     private volatile ActorSystem _system;
     private volatile ActorRef _shutdownActor;
+    private volatile AppShutdown _guiceAppShutdown;
     private volatile List<Database> _databases;
 
     private static final Logger LOGGER = com.arpnetworking.steno.LoggerFactory.getLogger(Main.class);

src/main/java/com/arpnetworking/clusteraggregator/aggregation/AggregationRouter.java

@@ -21,6 +21,7 @@
 import akka.actor.ReceiveTimeout;
 import akka.cluster.sharding.ShardRegion;
 import com.arpnetworking.metrics.aggregation.protocol.Messages;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
 import com.arpnetworking.steno.Logger;
 import com.arpnetworking.steno.LoggerFactory;
 import com.google.common.collect.ImmutableSet;
@@ -49,6 +50,7 @@ public class AggregationRouter extends AbstractActor {
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
+     * @param periodicMetrics The {@link PeriodicMetrics} instance.
      * @return A new {@link Props}.
      */
     public static Props props(
@@ -57,15 +59,17 @@ public static Props props(
             final String clusterHostSuffix,
             final ImmutableSet<String> reaggregationDimensions,
             final boolean injectClusterAsHost,
-            final Duration aggregatorTimeout) {
+            final Duration aggregatorTimeout,
+            final PeriodicMetrics periodicMetrics) {
         return Props.create(
                 AggregationRouter.class,
                 metricsListener,
                 emitter,
                 clusterHostSuffix,
                 reaggregationDimensions,
                 injectClusterAsHost,
-                aggregatorTimeout);
+                aggregatorTimeout,
+                periodicMetrics);
     }
 
     /**
@@ -77,6 +81,7 @@ public static Props props(
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
+     * @param periodicMetrics The {@link PeriodicMetrics} instance.
      */
     @Inject
     public AggregationRouter(
@@ -85,15 +90,17 @@ public AggregationRouter(
             @Named("cluster-host-suffix") final String clusterHostSuffix,
             @Named("reaggregation-dimensions") final ImmutableSet<String> reaggregationDimensions,
             @Named("reaggregation-cluster-as-host") final boolean injectClusterAsHost,
-            @Named("reaggregation-timeout") final Duration aggregatorTimeout) {
+            @Named("reaggregation-timeout") final Duration aggregatorTimeout,
+            final PeriodicMetrics periodicMetrics) {
         _streamingChild = context().actorOf(
                 StreamingAggregator.props(
                         periodicStatistics,
                         emitter,
                         clusterHostSuffix,
                         reaggregationDimensions,
                         injectClusterAsHost,
-                        aggregatorTimeout),
+                        aggregatorTimeout,
+                        periodicMetrics),
                 "streaming");
         context().setReceiveTimeout(FiniteDuration.apply(30, TimeUnit.MINUTES));
     }

src/main/java/com/arpnetworking/clusteraggregator/aggregation/StreamingAggregator.java

@@ -22,6 +22,7 @@
 import akka.cluster.sharding.ShardRegion;
 import com.arpnetworking.clusteraggregator.models.CombinedMetricData;
 import com.arpnetworking.metrics.aggregation.protocol.Messages;
+import com.arpnetworking.metrics.incubator.PeriodicMetrics;
 import com.arpnetworking.steno.Logger;
 import com.arpnetworking.steno.LoggerFactory;
 import com.arpnetworking.tsdcore.model.AggregatedData;
@@ -68,6 +69,7 @@ public class StreamingAggregator extends AbstractActorWithTimers {
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
+     * @param periodicMetrics The {@link PeriodicMetrics} instance.
      * @return A new {@link Props}.
      */
     public static Props props(
@@ -76,15 +78,17 @@ public static Props props(
             final String clusterHostSuffix,
             final ImmutableSet<String> reaggregationDimensions,
             final boolean injectClusterAsHost,
-            final Duration aggregatorTimeout) {
+            final Duration aggregatorTimeout,
+            final PeriodicMetrics periodicMetrics) {
         return Props.create(
                 StreamingAggregator.class,
                 metricsListener,
                 emitter,
                 clusterHostSuffix,
                 reaggregationDimensions,
                 injectClusterAsHost,
-                aggregatorTimeout);
+                aggregatorTimeout,
+                periodicMetrics);
     }
 
     /**
@@ -96,6 +100,7 @@ public static Props props(
      * @param reaggregationDimensions The dimensions to reaggregate over.
      * @param injectClusterAsHost Whether to inject a host dimension based on cluster.
      * @param aggregatorTimeout The time to wait from the start of the period for all data.
+     * @param periodicMetrics The {@link PeriodicMetrics} instance.
      */
     @Inject
     public StreamingAggregator(
@@ -104,12 +109,14 @@ public StreamingAggregator(
             @Named("cluster-host-suffix") final String clusterHostSuffix,
             @Named("reaggregation-dimensions") final ImmutableSet<String> reaggregationDimensions,
             @Named("reaggregation-cluster-as-host") final boolean injectClusterAsHost,
-            @Named("reaggregation-timeout") final Duration aggregatorTimeout) {
+            @Named("reaggregation-timeout") final Duration aggregatorTimeout,
+            final PeriodicMetrics periodicMetrics) {
         _periodicStatistics = periodicStatistics;
         _clusterHostSuffix = clusterHostSuffix;
         _reaggregationDimensions = reaggregationDimensions;
         _injectClusterAsHost = injectClusterAsHost;
         _aggregatorTimeout = aggregatorTimeout;
+        _periodicMetrics = periodicMetrics;
         context().setReceiveTimeout(FiniteDuration.apply(30, TimeUnit.MINUTES));
 
         timers().startPeriodicTimer(BUCKET_CHECK_TIMER_KEY, BucketCheck.getInstance(), FiniteDuration.apply(5, TimeUnit.SECONDS));
@@ -189,8 +196,19 @@ public Receive createReceive() {
                 .build();
     }
 
+    @Override
+    public void preStart() {
+        _periodicMetrics.recordCounter("actors/streaming_aggregator/started", 1);
+    }
+
+    @Override
+    public void postStop() {
+        _periodicMetrics.recordCounter("actors/streaming_aggregator/stopped", 1);
+    }
+
     @Override
     public void preRestart(final Throwable reason, final Optional<Object> message) throws Exception {
+        _periodicMetrics.recordCounter("actors/streaming_aggregator/restarted", 1);
         LOGGER.error()
                 .setMessage("Aggregator crashing")
                 .setThrowable(reason)
@@ -355,6 +373,7 @@ private String createHost() {
     private final boolean _injectClusterAsHost;
     private final Set<Statistic> _statistics = Sets.newHashSet();
     private final Duration _aggregatorTimeout;
+    private final PeriodicMetrics _periodicMetrics;
     private boolean _initialized = false;
     private Duration _period;
     private String _cluster;