Merge pull request #136 from ArpNetworking/actor_liveliness

CAGG will constantly start up StreamingAggregator actors throughout the cluster as it receives new metric dimension sets. If these dimension sets are infrequent, the actor will aggregate that single occurrence and live for the remaining lifetime of the application. This can lead to a slowly growing pool of actors not doing any meaningful work.

To fix this, we add a new configuration option, **aggregatorLivelinessTimeout**, which controls how often a StreamingAggregator actor will check if it should shutdown due to lack of data. If the actor has not received data between consecutive checks, it will request shutdown. Note that in the event an actor is shutdown for a dimension set that will occur again, the cluster sharding manager will just recreate the actor.

Author: cwbriones

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

@@ -358,6 +358,13 @@ private Duration provideReaggregationTimeout(final ClusterAggregatorConfiguratio
         return config.getReaggregationTimeout();
     }
 
+    @Provides
+    @Named("aggregator-liveliness-timeout")
+    @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice
+    private Duration provideLivelinessTimeout(final ClusterAggregatorConfiguration config) {
+        return config.getAggregatorLivelinessTimeout();
+    }
+
     @Provides
     @Named("circonus-partition-set")
     @SuppressFBWarnings("UPM_UNCALLED_PRIVATE_METHOD") // Invoked reflectively by Guice

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

@@ -49,6 +49,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 livelinessTimeout How often to check that a child is still receiving data.
      * @return A new {@link Props}.
      */
     public static Props props(
@@ -57,15 +58,17 @@ public static Props props(
             final String clusterHostSuffix,
             final ImmutableSet<String> reaggregationDimensions,
             final boolean injectClusterAsHost,
-            final Duration aggregatorTimeout) {
+            final Duration aggregatorTimeout,
+            final Duration livelinessTimeout) {
         return Props.create(
                 AggregationRouter.class,
                 metricsListener,
                 emitter,
                 clusterHostSuffix,
                 reaggregationDimensions,
                 injectClusterAsHost,
-                aggregatorTimeout);
+                aggregatorTimeout,
+                livelinessTimeout);
     }
 
     /**
@@ -77,6 +80,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 livelinessTimeout How often to check that a child is still receiving data.
      */
     @Inject
     public AggregationRouter(
@@ -85,15 +89,18 @@ 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) {
-        _streamingChild = context().actorOf(
+            @Named("reaggregation-timeout") final Duration aggregatorTimeout,
+            @Named("aggregator-liveliness-timeout") final Duration livelinessTimeout) {
+            _streamingChild = context().actorOf(
                 StreamingAggregator.props(
                         periodicStatistics,
                         emitter,
                         clusterHostSuffix,
                         reaggregationDimensions,
                         injectClusterAsHost,
-                        aggregatorTimeout),
+                        aggregatorTimeout,
+                        livelinessTimeout
+                ),
                 "streaming");
         context().setReceiveTimeout(FiniteDuration.apply(30, TimeUnit.MINUTES));
     }

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

@@ -68,6 +68,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 livelinessTimeout How often to check that this actor is still receiving data.
      * @return A new {@link Props}.
      */
     public static Props props(
@@ -76,15 +77,17 @@ public static Props props(
             final String clusterHostSuffix,
             final ImmutableSet<String> reaggregationDimensions,
             final boolean injectClusterAsHost,
-            final Duration aggregatorTimeout) {
+            final Duration aggregatorTimeout,
+            final Duration livelinessTimeout) {
         return Props.create(
                 StreamingAggregator.class,
                 metricsListener,
                 emitter,
                 clusterHostSuffix,
                 reaggregationDimensions,
                 injectClusterAsHost,
-                aggregatorTimeout);
+                aggregatorTimeout,
+                livelinessTimeout);
     }
 
     /**
@@ -96,6 +99,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 livelinessTimeout How often to check that this actor is still receiving data.
      */
     @Inject
     public StreamingAggregator(
@@ -104,7 +108,8 @@ 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,
+            @Named("aggregator-liveliness-timeout") final Duration livelinessTimeout) {
         _periodicStatistics = periodicStatistics;
         _clusterHostSuffix = clusterHostSuffix;
         _reaggregationDimensions = reaggregationDimensions;
@@ -113,6 +118,7 @@ public StreamingAggregator(
         context().setReceiveTimeout(FiniteDuration.apply(30, TimeUnit.MINUTES));
 
         timers().startPeriodicTimer(BUCKET_CHECK_TIMER_KEY, BucketCheck.getInstance(), FiniteDuration.apply(5, TimeUnit.SECONDS));
+        timers().startPeriodicTimer(LIVELINESS_CHECK_TIMER, LIVELINESS_CHECK_MSG, livelinessTimeout);
 
         _emitter = emitter;
     }
@@ -121,6 +127,8 @@ public StreamingAggregator(
     public Receive createReceive() {
         return receiveBuilder()
                 .match(Messages.StatisticSetRecord.class, record -> {
+                    // Mark this actor as live since we're still receiving data.
+                    _live = true;
                     LOGGER.debug()
                             .setMessage("Processing a StatisticSetRecord")
                             .addData("workItem", record)
@@ -182,10 +190,25 @@ public Receive createReceive() {
                         }
                     }
                 })
-                .match(ShutdownAggregator.class, message -> context().stop(self()))
-                .match(ReceiveTimeout.class, message -> {
-                    getContext().parent().tell(new ShardRegion.Passivate(ShutdownAggregator.getInstance()), getSelf());
+                .matchEquals(LIVELINESS_CHECK_MSG, msg -> {
+                    // If we've received data since our last check, reset and wait until another round.
+                    // otherwise shutdown.
+                    if (_live) {
+                        LOGGER.debug()
+                                .setMessage("aggregator is still live, continuing.")
+                                .addContext("actor", self())
+                                .log();
+                        _live = false;
+                        return;
+                    }
+                    LOGGER.debug()
+                            .setMessage("aggregator is stale, requesting shutdown.")
+                            .addContext("actor", self())
+                            .log();
+                    requestShutdownFromParent();
                 })
+                .match(ShutdownAggregator.class, message -> context().stop(self()))
+                .match(ReceiveTimeout.class, message -> requestShutdownFromParent())
                 .build();
     }
 
@@ -309,6 +332,10 @@ private void initialize(final Messages.StatisticSetRecord data, final CombinedMe
         }
     }
 
+    private void requestShutdownFromParent() {
+        getContext().parent().tell(new ShardRegion.Passivate(ShutdownAggregator.getInstance()), getSelf());
+    }
+
     private ImmutableMap<String, String> dimensionsToMap(final Messages.StatisticSetRecord statisticSetRecord) {
         // Build a map of dimension key-value pairs dropping any that are to be reaggregated over
         final ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
@@ -356,6 +383,9 @@ private String createHost() {
     private final Set<Statistic> _statistics = Sets.newHashSet();
     private final Duration _aggregatorTimeout;
     private boolean _initialized = false;
+    // This actor is _live if it's received data since the last LIVELINESS_CHECK_MSG.
+    // If this is ever false during a check, the actor will shutdown.
+    private boolean _live = false;
     private Duration _period;
     private String _cluster;
     private String _metric;
@@ -367,6 +397,9 @@ private String createHost() {
                     || entry.getKey().equals(CombinedMetricData.HOST_KEY)
                     || entry.getKey().equals(CombinedMetricData.SERVICE_KEY));
     private static final String BUCKET_CHECK_TIMER_KEY = "bucketcheck";
+    private static final String LIVELINESS_CHECK_TIMER = "livelinesscheck";
+
+    private static final String LIVELINESS_CHECK_MSG = "LIVELINESS_CHECK_MSG";
 
     private static final Logger LOGGER = LoggerFactory.getLogger(StreamingAggregator.class);
     private static final Logger WORK_TOO_OLD_LOGGER = LoggerFactory.getRateLimitLogger(

src/main/java/com/arpnetworking/clusteraggregator/configuration/ClusterAggregatorConfiguration.java

@@ -25,9 +25,11 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import net.sf.oval.constraint.NotEmpty;
 import net.sf.oval.constraint.NotNull;
 import net.sf.oval.constraint.Range;
+import net.sf.oval.constraint.ValidateWithMethod;
 
 import java.io.File;
 import java.time.Duration;
@@ -132,6 +134,10 @@ public Duration getReaggregationTimeout() {
         return _reaggregationTimeout;
     }
 
+    public Duration getAggregatorLivelinessTimeout() {
+        return _aggregatorLivelinessTimeout;
+    }
+
     public RebalanceConfiguration getRebalanceConfiguration() {
         return _rebalanceConfiguration;
     }
@@ -180,6 +186,7 @@ public String toString() {
                 .add("ReaggregationDimensions", _reaggregationDimensions)
                 .add("ReaggregationInjectClusterAsHost", _reaggregationInjectClusterAsHost)
                 .add("ReaggregationTimeout", _reaggregationTimeout)
+                .add("AggregatorLivelinessTimeout", _aggregatorLivelinessTimeout)
                 .add("MinConnectionTimeout", _minConnectionTimeout)
                 .add("MaxConnectionTimeout", _maxConnectionTimeout)
                 .add("RebalanceConfiguration", _rebalanceConfiguration)
@@ -208,6 +215,7 @@ private ClusterAggregatorConfiguration(final Builder builder) {
         _reaggregationDimensions = builder._reaggregationDimensions;
         _reaggregationInjectClusterAsHost = builder._reaggregationInjectClusterAsHost;
         _reaggregationTimeout = builder._reaggregationTimeout;
+        _aggregatorLivelinessTimeout = builder._aggregatorLivelinessTimeout;
         _minConnectionTimeout = builder._minConnectionTimeout;
         _maxConnectionTimeout = builder._maxConnectionTimeout;
         _jvmMetricsCollectionInterval = builder._jvmMetricsCollectionInterval;
@@ -236,6 +244,7 @@ private ClusterAggregatorConfiguration(final Builder builder) {
     private final ImmutableSet<String> _reaggregationDimensions;
     private final boolean _reaggregationInjectClusterAsHost;
     private final Duration _reaggregationTimeout;
+    private final Duration _aggregatorLivelinessTimeout;
     private final Duration _minConnectionTimeout;
     private final Duration _maxConnectionTimeout;
     private final Duration _jvmMetricsCollectionInterval;
@@ -540,6 +549,26 @@ public Builder setReaggregationTimeout(final Duration value) {
             return this;
         }
 
+        /**
+         * How often an aggregator actor should check for liveliness. An actor is considered live
+         * if any data is received between consecutive checks.
+         *
+         * This control is useful for culling instances which aggregate very infrequent dimension
+         * sets, especially if the application itself is long-lived.
+         *
+         * This must be greater than the reaggregation timeout, otherwise an actor could be
+         * incorrectly marked as stale before flushing its data.
+         *
+         * Optional. Defaults to twice the reaggregation timeout. Cannot be null.
+         *
+         * @param value Timeout between consecutive liveliness checks.
+         * @return This instance of {@link Builder}.
+         */
+        public Builder setAggregatorLivelinessTimeout(final Duration value) {
+            _aggregatorLivelinessTimeout = value;
+            return this;
+        }
+
         /**
          * Configuration for the shard rebalance settings.
          *
@@ -574,6 +603,17 @@ public Builder setCalculateClusterAggregations(final Boolean value) {
             return this;
         }
 
+        /**
+         * Validate that the aggregator liveliness timeout is greater than the reaggregation timeout.
+         *
+         * @param aggregatorLivelinessTimeout the configured liveliness timeout
+         * @return true if the given value is valid
+         */
+        @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", justification = "invoked reflectively by @ValidateWithMethod")
+        public boolean validateAggregatorLivelinessTimeout(final Duration aggregatorLivelinessTimeout) {
+            return aggregatorLivelinessTimeout.compareTo(_reaggregationTimeout) > 0;
+        }
+
         @NotNull
         @NotEmpty
         private String _monitoringCluster;
@@ -621,6 +661,9 @@ public Builder setCalculateClusterAggregations(final Boolean value) {
         @NotNull
         private Duration _reaggregationTimeout = Duration.ofMinutes(1);
         @NotNull
+        @ValidateWithMethod(methodName = "validateAggregatorLivelinessTimeout", parameterType = Duration.class)
+        private Duration _aggregatorLivelinessTimeout = _reaggregationTimeout.multipliedBy(2);
+        @NotNull
         private File _hostPipelineConfiguration;
         @NotNull
         private Map<String, ?> _akkaConfiguration;

src/test/java/com/arpnetworking/clusteraggregator/aggregation/AggregatorTest.java

@@ -59,7 +59,8 @@ public ActorRef createAggregator(final TestProbe probe) {
                         "",
                         ImmutableSet.of(),
                         true,
-                        Duration.ofMinutes(1)),
+                        Duration.ofMinutes(1),
+                        Duration.ofMinutes(2)),
                 probe.ref(),
                 "agg",
                 getSystem());