Remove the book-keeper since it does not work for non-reaggregation case and the messages are no longer java serializable. Also improve logging in the streaming aggregator. Finally, support configuration of the streaming aggregator timeout. (#94)

Author: vjkoskela

pom.xml

@@ -627,11 +627,6 @@
       <artifactId>akka-cluster_${scala.version}</artifactId>
       <version>${akka.version}</version>
     </dependency>
-    <dependency>
-      <groupId>com.typesafe.akka</groupId>
-      <artifactId>akka-cluster-tools_${scala.version}</artifactId>
-      <version>${akka.version}</version>
-    </dependency>
     <dependency>
       <groupId>com.typesafe.akka</groupId>
       <artifactId>akka-remote_${scala.version}</artifactId>

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

@@ -22,10 +22,6 @@
 import akka.cluster.Cluster;
 import akka.cluster.sharding.ClusterSharding;
 import akka.cluster.sharding.ClusterShardingSettings;
-import akka.cluster.singleton.ClusterSingletonManager;
-import akka.cluster.singleton.ClusterSingletonManagerSettings;
-import akka.cluster.singleton.ClusterSingletonProxy;
-import akka.cluster.singleton.ClusterSingletonProxySettings;
 import akka.http.javadsl.ConnectHttp;
 import akka.http.javadsl.Http;
 import akka.http.javadsl.IncomingConnection;
@@ -36,8 +32,6 @@
 import akka.stream.Materializer;
 import com.arpnetworking.clusteraggregator.aggregation.AggMessageExtractor;
 import com.arpnetworking.clusteraggregator.aggregation.AggregationRouter;
-import com.arpnetworking.clusteraggregator.aggregation.Bookkeeper;
-import com.arpnetworking.clusteraggregator.bookkeeper.persistence.InMemoryBookkeeper;
 import com.arpnetworking.clusteraggregator.client.AggClientServer;
 import com.arpnetworking.clusteraggregator.client.AggClientSupervisor;
 import com.arpnetworking.clusteraggregator.client.HttpSourceActor;
@@ -79,6 +73,7 @@
 import com.typesafe.config.ConfigParseOptions;
 import com.typesafe.config.ConfigSyntax;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import org.joda.time.Period;
 
 import java.io.File;
 import java.io.IOException;
@@ -210,35 +205,17 @@ private ActorRef launchEmitter(final Injector injector, final ActorSystem system
         return emitterConfigurationProxy;
     }
 
-    @Provides
-    @Singleton
-    @Named("bookkeeper-proxy")
-    @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
-    private ActorRef provideBookkeeperProxy(final ActorSystem system) {
-        system.actorOf(
-                ClusterSingletonManager.props(
-                        Bookkeeper.props(new InMemoryBookkeeper()),
-                        PoisonPill.getInstance(),
-                        ClusterSingletonManagerSettings.create(system)),
-                "bookkeeper");
-
-        return system.actorOf(
-                ClusterSingletonProxy.props("/user/bookkeeper", ClusterSingletonProxySettings.create(system)),
-                "bookkeeperProxy");
-    }
-
     @Provides
     @Singleton
     @Named("status-cache")
     @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
     private ActorRef provideStatusCache(
             final ActorSystem system,
-            @Named("bookkeeper-proxy") final ActorRef bookkeeperProxy,
             @Named("periodic-statistics") final ActorRef periodicStats,
             final MetricsFactory metricsFactory) {
         final Cluster cluster = Cluster.get(system);
         final ActorRef clusterStatusCache = system.actorOf(ClusterStatusCache.props(cluster, metricsFactory), "cluster-status");
-        return system.actorOf(Status.props(bookkeeperProxy, cluster, clusterStatusCache, periodicStats), "status");
+        return system.actorOf(Status.props(cluster, clusterStatusCache, periodicStats), "status");
     }
 
     @Provides
@@ -249,18 +226,6 @@ private ActorRef provideTcpServer(final Injector injector, final ActorSystem sys
         return system.actorOf(GuiceActorCreator.props(injector, AggClientServer.class), "tcp-server");
     }
 
-    @Provides
-    @Singleton
-    @Named("aggregator-lifecycle")
-    @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
-    private ActorRef provideAggregatorLifecycleTracker(
-            final ActorSystem system,
-            @Named("bookkeeper-proxy") final ActorRef bookkeeperProxy) {
-        final ActorRef aggLifecycle = system.actorOf(AggregatorLifecycle.props(), "agg-lifecycle");
-        aggLifecycle.tell(new AggregatorLifecycle.Subscribe(bookkeeperProxy), bookkeeperProxy);
-        return aggLifecycle;
-    }
-
     @Provides
     @Singleton
     @Named("http-server")
@@ -364,6 +329,13 @@ private boolean provideReaggregationInjectClusterAsHost(final ClusterAggregatorC
         return config.getReaggregationInjectClusterAsHost();
     }
 
+    @Provides
+    @Named("reaggregation-timeout")
+    @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
+    private Period provideReaggregationTimeout(final ClusterAggregatorConfiguration config) {
+        return config.getReaggregationTimeout();
+    }
+
     @Provides
     @Named("circonus-partition-set")
     @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice

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

@@ -172,12 +172,6 @@ private void launchActors(final Injector injector) {
         injector.getInstance(Key.get(ActorRef.class, Names.named("host-emitter")));
         injector.getInstance(Key.get(ActorRef.class, Names.named("cluster-emitter")));
 
-        LOGGER.info()
-                .setMessage("Launching bookkeeper singleton and proxy")
-                .log();
-        injector.getInstance(Key.get(ActorRef.class, Names.named("bookkeeper-proxy")));
-
-        injector.getInstance(Key.get(ActorRef.class, Names.named("aggregator-lifecycle")));
         injector.getInstance(Key.get(ActorRef.class, Names.named("periodic-statistics")));
 
         LOGGER.info()
@@ -231,9 +225,13 @@ private void shutdownAkka() {
         LOGGER.info()
                 .setMessage("Stopping Akka")
                 .log();
-        _shutdownActor.tell(GracefulShutdownActor.Shutdown.instance(), ActorRef.noSender());
+        if (_shutdownActor != null) {
+            _shutdownActor.tell(GracefulShutdownActor.Shutdown.instance(), ActorRef.noSender());
+        }
         try {
-            Await.result(_system.whenTerminated(), SHUTDOWN_TIMEOUT);
+            if (_system != null) {
+                Await.result(_system.whenTerminated(), SHUTDOWN_TIMEOUT);
+            }
             // CHECKSTYLE.OFF: IllegalCatch - Prevent program shutdown
         } catch (final Exception e) {
             // CHECKSTYLE.ON: IllegalCatch

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

@@ -23,7 +23,6 @@
 import akka.cluster.MemberStatus;
 import akka.pattern.PatternsCS;
 import akka.remote.AssociationErrorEvent;
-import com.arpnetworking.clusteraggregator.models.BookkeeperData;
 import com.arpnetworking.clusteraggregator.models.MetricsRequest;
 import com.arpnetworking.clusteraggregator.models.PeriodMetrics;
 import com.arpnetworking.clusteraggregator.models.StatusResponse;
@@ -53,18 +52,15 @@ public class Status extends AbstractActor {
     /**
      * Public constructor.
      *
-     * @param metricsBookkeeper Where to get the status metrics from.
      * @param cluster The instance of the Clustering extension.
      * @param clusterStatusCache The actor holding the cached cluster status.
      * @param localMetrics The actor holding the local node metrics.
      */
     public Status(
-            final ActorRef metricsBookkeeper,
             final Cluster cluster,
             final ActorRef clusterStatusCache,
             final ActorRef localMetrics) {
 
-        _metricsBookkeeper = metricsBookkeeper;
         _cluster = cluster;
         _clusterStatusCache = clusterStatusCache;
         _localMetrics = localMetrics;
@@ -74,19 +70,17 @@ public Status(
     /**
      * Creates a <code>Props</code> for use in Akka.
      *
-     * @param bookkeeper Where to get the status metrics from.
      * @param cluster The instance of the Clustering extension.
      * @param clusterStatusCache The actor holding the cached cluster status.
      * @param localMetrics The actor holding the local node metrics.
      * @return A new <code>Props</code>.
      */
     public static Props props(
-            final ActorRef bookkeeper,
             final Cluster cluster,
             final ActorRef clusterStatusCache,
             final ActorRef localMetrics) {
 
-        return Props.create(Status.class, bookkeeper, cluster, clusterStatusCache, localMetrics);
+        return Props.create(Status.class, cluster, clusterStatusCache, localMetrics);
     }
 
     @Override
@@ -116,13 +110,6 @@ public Receive createReceive() {
 
     private void processStatusRequest() {
         // Call the bookkeeper
-        final CompletableFuture<BookkeeperData> bookkeeperFuture = PatternsCS.ask(
-                _metricsBookkeeper,
-                new MetricsRequest(),
-                Duration.ofSeconds(3))
-                .<BookkeeperData>thenApply(new CastMapper<>())
-                .exceptionally(new AsNullRecovery<>())
-                .toCompletableFuture();
         final CompletableFuture<ClusterStatusCache.StatusResponse> clusterStateFuture =
                 PatternsCS.ask(
                         _clusterStatusCache,
@@ -143,11 +130,9 @@ private void processStatusRequest() {
 
         PatternsCS.pipe(
                 CompletableFuture.allOf(
-                        bookkeeperFuture.toCompletableFuture(),
                         clusterStateFuture.toCompletableFuture(),
                         localMetricsFuture.toCompletableFuture())
                         .thenApply(aVoid -> new StatusResponse.Builder()
-                                .setClusterMetrics(bookkeeperFuture.getNow(null))
                                 .setClusterState(clusterStateFuture.getNow(null))
                                 .setLocalMetrics(localMetricsFuture.getNow(null))
                                 .setLocalAddress(_cluster.selfAddress())
@@ -158,7 +143,6 @@ private void processStatusRequest() {
 
     private boolean _quarantined = false;
 
-    private final ActorRef _metricsBookkeeper;
     private final Cluster _cluster;
     private final ActorRef _clusterStatusCache;
     private final ActorRef _localMetrics;

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

@@ -26,6 +26,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 import com.google.inject.name.Named;
+import org.joda.time.Period;
 import scala.concurrent.duration.FiniteDuration;
 
 import java.io.Serializable;
@@ -42,57 +43,57 @@ public class AggregationRouter extends AbstractActor {
     /**
      * Creates a <code>Props</code> for use in Akka.
      *
-     * @param lifecycleTracker Where to register the liveliness of this aggregator.
      * @param metricsListener Where to send metrics about aggregation computations.
      * @param emitter Where to send the metrics data.
      * @param clusterHostSuffix The suffix to append to the hostname for cluster aggregations.
      * @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.
      * @return A new <code>Props</code>.
      */
     public static Props props(
-            final ActorRef lifecycleTracker,
             final ActorRef metricsListener,
             final ActorRef emitter,
             final String clusterHostSuffix,
             final ImmutableSet<String> reaggregationDimensions,
-            final boolean injectClusterAsHost) {
+            final boolean injectClusterAsHost,
+            final Period aggregatorTimeout) {
         return Props.create(
                 AggregationRouter.class,
-                lifecycleTracker,
                 metricsListener,
                 emitter,
                 clusterHostSuffix,
                 reaggregationDimensions,
-                injectClusterAsHost);
+                injectClusterAsHost,
+                aggregatorTimeout);
     }
 
     /**
      * Public constructor.
      *
-     * @param lifecycleTracker Where to register the liveliness of this aggregator.
      * @param periodicStatistics Where to send metrics about aggregation computations.
      * @param emitter Where to send the metrics data.
      * @param clusterHostSuffix The suffix to append to the hostname for cluster aggregations.
      * @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.
      */
     @Inject
     public AggregationRouter(
-            @Named("bookkeeper-proxy") final ActorRef lifecycleTracker,
             @Named("periodic-statistics") final ActorRef periodicStatistics,
             @Named("cluster-emitter") final ActorRef emitter,
             @Named("cluster-host-suffix") final String clusterHostSuffix,
             @Named("reaggregation-dimensions") final ImmutableSet<String> reaggregationDimensions,
-            @Named("reaggregation-cluster-as-host") final boolean injectClusterAsHost) {
+            @Named("reaggregation-cluster-as-host") final boolean injectClusterAsHost,
+            @Named("reaggregation-timeout") final Period aggregatorTimeout) {
         _streamingChild = context().actorOf(
                 StreamingAggregator.props(
-                        lifecycleTracker,
                         periodicStatistics,
                         emitter,
                         clusterHostSuffix,
                         reaggregationDimensions,
-                        injectClusterAsHost),
+                        injectClusterAsHost,
+                        aggregatorTimeout),
                 "streaming");
         context().setReceiveTimeout(FiniteDuration.apply(30, TimeUnit.MINUTES));
     }