make StreamingAggregator actors shutdown when no data is received

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 subsequent checks.
+         *
+         * This control is useful for culling aggregator instances for very infrequently occurring
+         * dimension sets, especially if the application 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 from period start to wait for all data to arrive.
+         * @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,11 @@ public Builder setCalculateClusterAggregations(final Boolean value) {
             return this;
         }
 
+        @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 +655,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());

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

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2015 Groupon.com
+ *
+ * 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.aggregation;
+
+import akka.actor.ActorRef;
+import akka.actor.ReceiveTimeout;
+import akka.actor.Terminated;
+import akka.cluster.sharding.ShardRegion;
+import akka.testkit.TestActorRef;
+import akka.testkit.TestProbe;
+import com.arpnetworking.metrics.aggregation.protocol.Messages;
+import com.arpnetworking.utility.BaseActorTest;
+import com.google.common.collect.ImmutableSet;
+import org.junit.Assert;
+import org.junit.Test;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Tests for the {@link StreamingAggregator} actor.
+ *
+ * @author Christian Briones (cbriones at dropbox dot com)
+ */
+public class StreamingAggregatorTest extends BaseActorTest {
+
+    @Test
+    public void passivatesProperlyWhenLivelinessTimeoutIsExceeded() {
+        final TestProbe probe = TestProbe.apply(getSystem());
+        final ActorRef aggregator = createAggregator(probe);
+        probe.watch(aggregator);
+
+        // We expect a passivate message after at least LIVELINESS_TIMEOUT_SECS
+        final ShardRegion.Passivate passivate = probe.expectMsgClass(PROBE_TIMEOUT, ShardRegion.Passivate.class);
+        aggregator.tell(passivate.stopMessage(), aggregator);
+        final Terminated terminated = probe.expectMsgClass(Terminated.class);
+        Assert.assertEquals(aggregator, terminated.getActor());
+    }
+
+    @Test
+    public void dataKeepsActorLive() {
+        final TestProbe probe = TestProbe.apply(getSystem());
+        final ActorRef aggregator = createAggregator(probe);
+        probe.watch(aggregator);
+
+        final Messages.StatisticSetRecord record = Messages.StatisticSetRecord.newBuilder()
+                .setMetric("my_metric")
+                .setPeriod("PT1M")
+                .setPeriodStart("2020-12-10T19:00:00Z")
+                .build();
+
+        aggregator.tell(record, probe.ref());
+        probe.expectNoMessage(FiniteDuration.apply(LIVELINESS_TIMEOUT_SEC, TimeUnit.SECONDS));
+
+        aggregator.tell(ReceiveTimeout.getInstance(), aggregator);
+        final ShardRegion.Passivate passivate = probe.expectMsgClass(PROBE_TIMEOUT, ShardRegion.Passivate.class);
+        aggregator.tell(passivate.stopMessage(), aggregator);
+        final Terminated terminated = probe.expectMsgClass(Terminated.class);
+        Assert.assertEquals(aggregator, terminated.getActor());
+    }
+
+    public ActorRef createAggregator(final TestProbe probe) {
+        final TestProbe ignored = TestProbe.apply(getSystem());
+        return TestActorRef.apply(
+                StreamingAggregator.props(
+                        ignored.ref(),
+                        ignored.ref(),
+                        "",
+                        ImmutableSet.of(),
+                        true,
+                        Duration.ofSeconds(5),
+                        Duration.ofSeconds(LIVELINESS_TIMEOUT_SEC)),
+        probe.ref(),
+                "agg",
+                getSystem());
+    }
+
+    private static final int LIVELINESS_TIMEOUT_SEC = 5;
+    private static final FiniteDuration PROBE_TIMEOUT = FiniteDuration.apply(10, TimeUnit.SECONDS);
+}