add a configuration flag to skip cluster aggregations (#70)

Author: BrandonArp

src/main/java/com/arpnetworking/clusteraggregator/client/AggClientConnection.java

@@ -55,10 +55,17 @@ public class AggClientConnection extends AbstractActor {
      * @param connection Reference to the client connection actor.
      * @param remote The address of the client socket.
      * @param maxConnectionAge The maximum duration to keep a connection open before cycling it.
+     * @param calculateAggregates True to compute cluster aggregations, false to only publish host aggregations
      * @return A new <code>Props</code>.
      */
-    public static Props props(final ActorRef connection, final InetSocketAddress remote, final FiniteDuration maxConnectionAge) {
-        return Props.create(AggClientConnection.class, connection, remote, maxConnectionAge);
+    public static Props props(
+            final ActorRef connection,
+            final InetSocketAddress remote,
+            final FiniteDuration maxConnectionAge,
+            final boolean calculateAggregates) {
+        return Props.create(
+                AggClientConnection.class,
+                () -> new AggClientConnection(connection, remote, maxConnectionAge, calculateAggregates));
     }
 
     /**
@@ -67,13 +74,16 @@ public static Props props(final ActorRef connection, final InetSocketAddress rem
      * @param connection Reference to the client connection actor.
      * @param remote The address of the client socket.
      * @param maxConnectionAge The maximum duration to keep a connection open before cycling it.
+     * @param calculateAggregates True to compute cluster aggregations, false to only publish host aggregations
      */
     public AggClientConnection(
             final ActorRef connection,
             final InetSocketAddress remote,
-            final FiniteDuration maxConnectionAge) {
+            final FiniteDuration maxConnectionAge,
+            final boolean calculateAggregates) {
         _connection = connection;
         _remoteAddress = remote;
+        _calculateAggregates = calculateAggregates;
 
         getContext().watch(connection);
 
@@ -150,7 +160,11 @@ private void processMessages() {
                         .addData("aggregation", setRecord)
                         .addContext("actor", self())
                         .log();
-                getContext().parent().tell(setRecord, getSelf());
+                // StatisticSetRecords get forwarded to the parent, who then forwards them to the shard for cluster aggregating
+                // If we aren't doing shard aggregating, don't forward it
+                if (_calculateAggregates) {
+                    getContext().parent().tell(setRecord, getSelf());
+                }
                 if (setRecord.getStatisticsCount() > 0) {
                     final Optional<PeriodicData> periodicData = buildPeriodicData(setRecord);
                     if (periodicData.isPresent()) {
@@ -268,6 +282,7 @@ private Optional<PeriodicData> buildPeriodicData(final Messages.StatisticSetReco
     private ByteString _buffer = ByteString.empty();
     private final ActorRef _connection;
     private final InetSocketAddress _remoteAddress;
+    private final boolean _calculateAggregates;
     private static final Logger LOGGER = LoggerFactory.getLogger(AggClientConnection.class);
     private static final Logger INCOMPLETE_RECORD_LOGGER = LoggerFactory.getRateLimitLogger(
             AggClientConnection.class,

src/main/java/com/arpnetworking/clusteraggregator/client/AggClientSupervisor.java

@@ -61,6 +61,7 @@ public AggClientSupervisor(
         _emitter = emitter;
         _minConnectionTimeout = configuration.getMinConnectionTimeout();
         _maxConnectionTimeout = configuration.getMaxConnectionTimeout();
+        _calculateClusterAggregates = configuration.getCalculateClusterAggregations();
     }
 
     @Override
@@ -75,7 +76,11 @@ public Receive createReceive() {
                     final ActorRef connection = getSender();
 
                     final ActorRef handler = getContext().actorOf(
-                            AggClientConnection.props(connection, connected.remoteAddress(), getRandomConnectionTime()),
+                            AggClientConnection.props(
+                                    connection,
+                                    connected.remoteAddress(),
+                                    getRandomConnectionTime(),
+                                    _calculateClusterAggregates),
                             "dataHandler");
                     connection.tell(TcpMessage.register(handler, true, true), getSelf());
                     getContext().watch(handler);
@@ -124,6 +129,7 @@ public SupervisorStrategy supervisorStrategy() {
     private final ActorRef _emitter;
     private final Period _minConnectionTimeout;
     private final Period _maxConnectionTimeout;
+    private final Boolean _calculateClusterAggregates;
     private final Random _random = new Random();
     private static final Logger LOGGER = LoggerFactory.getLogger(AggClientSupervisor.class);
 }

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

@@ -114,6 +114,10 @@ public String getClusterHostSuffix() {
         return _clusterHostSuffix;
     }
 
+    public boolean getCalculateClusterAggregations() {
+        return _calculateClusterAggregations;
+    }
+
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)
@@ -155,6 +159,7 @@ private ClusterAggregatorConfiguration(final Builder builder) {
         _jvmMetricsCollectionInterval = builder._jvmMetricsCollectionInterval;
         _rebalanceConfiguration = builder._rebalanceConfiguration;
         _clusterHostSuffix = builder._clusterHostSuffix;
+        _calculateClusterAggregations = builder._calculateClusterAggregations;
         _databaseConfigurations = Maps.newHashMap(builder._databaseConfigurations);
     }
 
@@ -174,6 +179,7 @@ private ClusterAggregatorConfiguration(final Builder builder) {
     private final Period _jvmMetricsCollectionInterval;
     private final RebalanceConfiguration _rebalanceConfiguration;
     private final String _clusterHostSuffix;
+    private final boolean _calculateClusterAggregations;
     private final Map<String, DatabaseConfiguration> _databaseConfigurations;
 
     private static final InterfaceDatabase INTERFACE_DATABASE = ReflectionsDatabase.newInstance();
@@ -387,6 +393,18 @@ public Builder setDatabaseConfigurations(final Map<String, DatabaseConfiguration
             return this;
         }
 
+        /**
+         * Whether or not to perform cluster-level aggregations. When using a datasource that supports
+         * native histograms, turning this off will reduce cpu cost. Optional. Defaults to true.
+         *
+         * @param value true to perform cluster aggregations, false to just forward host data.
+         * @return This instance of <code>Builder</code>.
+         */
+        public Builder setCalculateClusterAggregations(final Boolean value) {
+            _calculateClusterAggregations = value;
+            return this;
+        }
+
         @NotNull
         @NotEmpty
         private String _monitoringCluster;
@@ -426,6 +444,8 @@ public Builder setDatabaseConfigurations(final Map<String, DatabaseConfiguration
         private RebalanceConfiguration _rebalanceConfiguration;
         @NotNull
         private String _clusterHostSuffix = "";
+        @NotNull
+        private Boolean _calculateClusterAggregations = true;
         private Map<String, DatabaseConfiguration> _databaseConfigurations;
     }
 }