closes #164 - Limit memory consumed by DownsampleIterator and return iteratively on time bucket boundaries

Docker & Grafana updates for 0.0.6-SNAPSHOT

analytics/pom.xml

@@ -23,7 +23,7 @@ under the License.
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.5</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <properties>
         <flink.version>1.1.4</flink.version>

client/pom.xml

@@ -5,8 +5,12 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.5</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
+    <properties>
+        <!-- PEP 440 compliant version gets supstituted into src/main/python/template/setup.py -->
+        <python.client.version>0.0.6.dev0</python.client.version>
+    </properties>
     <dependencies>
         <dependency>
             <groupId>timely</groupId>
@@ -99,6 +103,62 @@
                 <groupId>org.codehaus.mojo</groupId>
                 <artifactId>buildnumber-maven-plugin</artifactId>
             </plugin>
+            <plugin>
+                <artifactId>maven-resources-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>filter-setup</id>
+                        <goals>
+                            <goal>copy-resources</goal>
+                        </goals>
+                        <phase>process-resources</phase>
+                        <configuration>
+                            <outputDirectory>${project.basedir}/src/main/python</outputDirectory>
+                            <resources>
+                                <resource>
+                                    <directory>${project.basedir}/src/main/python/template</directory>
+                                    <includes>
+                                        <include>setup.py</include>
+                                    </includes>
+                                    <filtering>true</filtering>
+                                </resource>
+                            </resources>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+<!--
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>exec-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>python-dist</id>
+                        <goals>
+                            <goal>exec</goal>
+                        </goals>
+                        <phase>package</phase>
+                        <configuration>
+                            <executable>python</executable>
+                            <workingDirectory>${project.basedir}/src/main/python</workingDirectory>
+                            <commandlineArgs>setup.py sdist - -dist-dir ${project.build.directory}</commandlineArgs>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>python-clean</id>
+                        <goals>
+                            <goal>exec</goal>
+                        </goals>
+                        <phase>package</phase>
+                        <configuration>
+                            <executable>rm</executable>
+                            <workingDirectory>${project.basedir}/src/main/python</workingDirectory>
+                            <commandlineArgs>-rf timely.egg-info setup.py</commandlineArgs>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+-->
         </plugins>
     </build>
     <reporting>

client/src/main/java/timely/model/ObjectSizeOf.java

@@ -0,0 +1,200 @@
+package timely.model;
+
+import java.lang.reflect.Array;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.Stack;
+import java.util.logging.Logger;
+
+/**
+ * A simple interface that objects can implement to return the object size.
+ */
+public interface ObjectSizeOf {
+
+    /**
+     * The (approximate) size of the object
+     */
+    long sizeInBytes();
+
+    public static class ObjectInstance {
+
+        private Object o;
+
+        public ObjectInstance(Object _o) {
+            this.o = _o;
+        }
+
+        public Object getObject() {
+            return o;
+        }
+
+        @Override
+        public int hashCode() {
+            return System.identityHashCode(o);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (!(obj instanceof ObjectInstance)) {
+                return false;
+            }
+            return ((ObjectInstance) obj).o == this.o;
+        }
+    }
+
+    public static class Sizer {
+
+        private static final Logger log = Logger.getLogger(Sizer.class.getCanonicalName());
+        public static final short OBJECT_OVERHEAD = 8;
+        public static final short ARRAY_OVERHEAD = 12;
+        public static final short REFERENCE = 4;
+        // The size of the basic Number constructs (and Boolean and Character)
+        // is 16: roundUp(8 + primitiveSize)
+        public static final short NUMBER_SIZE = 16;
+
+        /**
+         * Get the size of an object. Note that we want something relatively
+         * fast that gives us an order of magnitude here. The java
+         * Instrumentation agent mechanism is a little too costly for general
+         * use here. This will look for the ObjectSizeOf interface and if
+         * implemented on the object will use that. Otherwise it will do a
+         * simple navigation of the fields using reflection.
+         *
+         * @param o
+         * @return an approximation of the object size
+         */
+        public static long getObjectSize(Object o, boolean useSizeInBytesMethod, boolean debug) {
+            long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(),
+                    useSizeInBytesMethod, debug);
+            return estSize;
+        }
+
+        public static long getObjectSize(Object o) {
+            long estSize = getObjectSize(o, new HashSet<ObjectInstance>(), new Stack<ObjectInstance>(), true, false);
+            return estSize;
+        }
+
+        public static long getObjectSize(Object o, Set<ObjectInstance> visited, Stack<ObjectInstance> stack,
+                boolean useSizeInBytesMethod, boolean debug) {
+            long totalSize = 0;
+            stack.add(new ObjectInstance(o));
+            while (!stack.isEmpty()) {
+                long size = 0;
+                ObjectInstance oi = stack.pop();
+                o = oi.getObject();
+                if (o != null && !visited.contains(oi)) {
+                    visited.add(oi);
+                    try {
+                        if (useSizeInBytesMethod) {
+                            try {
+                                if (o instanceof ObjectSizeOf) {
+                                    size = ((ObjectSizeOf) o).sizeInBytes();
+                                } else {
+                                    Method sizeInBytes = o.getClass().getMethod("sizeInBytes", (Class<?>[]) null);
+                                    size = (Long) sizeInBytes.invoke(o);
+                                }
+                            } catch (Throwable t) {
+                                // ok, lets do this the hard way...
+                            }
+                        }
+                        if (size == 0) {
+                            // the hard way...
+                            // do not include Class related objects or
+                            // reflection objects
+                            if ((o instanceof Class)
+                                    || (o instanceof ClassLoader)
+                                    || (o.getClass().getPackage() != null && o.getClass().getPackage().getName()
+                                            .startsWith("java.lang.reflect"))) {
+                                size = 0;
+                            } else if (o instanceof Number || o instanceof Boolean || o instanceof Character) {
+                                size = NUMBER_SIZE;
+                            } else {
+                                // lets do a simple sizing
+                                Class<?> c = o.getClass();
+                                if (c.isArray()) {
+                                    size = ARRAY_OVERHEAD;
+                                    int length = Array.getLength(o);
+                                    if (c.getComponentType().isPrimitive()) {
+                                        size += length * getPrimitiveObjectSize(c.getComponentType());
+                                    } else {
+                                        size += length * REFERENCE;
+                                        for (int i = 0; i < length; i++) {
+                                            Object element = Array.get(o, i);
+                                            if (element != null) {
+                                                stack.add(new ObjectInstance(element));
+                                            }
+                                        }
+                                    }
+                                } else {
+                                    size += OBJECT_OVERHEAD;
+                                    while (c != null) {
+                                        for (Field field : c.getDeclaredFields()) {
+                                            if (Modifier.isStatic(field.getModifiers())) {
+                                                continue;
+                                            }
+                                            if (field.getType().isPrimitive()) {
+                                                size += getPrimitiveObjectSize(field.getType());
+                                            } else {
+                                                size += REFERENCE;
+                                                boolean accessible = field.isAccessible();
+                                                field.setAccessible(true);
+                                                try {
+                                                    Object fieldObject = field.get(o);
+                                                    if (fieldObject != null) {
+                                                        stack.push(new ObjectInstance(fieldObject));
+                                                    }
+                                                } catch (Exception e) {
+                                                    // cannot get to field, so
+                                                    // ignore it in this size
+                                                    // calculation
+                                                    e.printStackTrace();
+                                                }
+                                                field.setAccessible(accessible);
+                                            }
+                                        }
+                                        c = c.getSuperclass();
+                                    }
+                                }
+                                size = roundUp(size);
+                            }
+                        }
+                    } catch (Throwable t) {
+                        log.warning("Unable to determine object size for " + o);
+                    }
+                    if (debug) {
+                        System.out.println(o.getClass().getCanonicalName() + "=" + size);
+                    }
+                }
+                totalSize += size;
+            }
+
+            return totalSize;
+        }
+
+        public static long roundUp(long size) {
+            long extra = size % 8;
+            if (extra > 0) {
+                size = size + 8 - extra;
+            }
+            return size;
+        }
+
+        public static short getPrimitiveObjectSize(Class<?> primitiveType) {
+            if (primitiveType.equals(int.class) || primitiveType.equals(float.class)) {
+                return 4;
+            } else if (primitiveType.equals(boolean.class) || primitiveType.equals(byte.class)) {
+                return 1;
+            } else if (primitiveType.equals(char.class) || primitiveType.equals(short.class)) {
+                return 2;
+            } else if (primitiveType.equals(long.class) || primitiveType.equals(double.class)) {
+                return 8;
+            } else { // if (primitiveType.equals(void.class)) {
+                return 0;
+            }
+        }
+    }
+
+}

client/src/main/python/template/setup.py

@@ -0,0 +1,12 @@
+from setuptools import setup
+
+setup(
+    name='timely',
+    version='${python.client.version}',
+    packages=['timely'],
+    install_requires=[
+        'seaborn',
+	'plotly'
+    ],
+    license='Apache License, Version 2.0'
+)

client/src/main/python/timely/DataOperations.py

@@ -289,15 +289,25 @@ def getTitle(timelyMetric, config, separator='\n'):
             alertTitle += " " + condition + " "
         alertTitle += "(y-avg > " + str(config.average_max_threshold) + ")"
 
-    if config.alert_percentage is not None:
+    if config.min_threshold_percentage is not None:
         if alertTitle != "":
             alertTitle += " " + condition + " "
-        if config.alert_percentage > 0:
-            alertTitle += "(y > " + str(config.alert_percentage) + "% above " + str(
-                config.rolling_average_period) + " sample average)"
+        if config.min_threshold_percentage >= 0:
+            descriptivePercentage = 100 + config.min_threshold_percentage
         else:
-            alertTitle += "(y < " + str(-config.alert_percentage) + "% below " + str(
-                config.rolling_average_period) + " sample average)"
+            descriptivePercentage = abs(config.min_threshold_percentage)
+        alertTitle += "(y < " + str(descriptivePercentage) + "% of " + str(
+            config.rolling_average_period) + " sample average)"
+
+    if config.max_threshold_percentage is not None:
+        if alertTitle != "":
+            alertTitle += " " + condition + " "
+        if config.max_threshold_percentage >= 0:
+            descriptivePercentage = 100 + config.max_threshold_percentage
+        else:
+            descriptivePercentage = abs(config.max_threshold_percentage)
+        alertTitle += "(y > " + str(descriptivePercentage) + "% of " + str(
+            config.rolling_average_period) + " sample average)"
 
     if config.system_name is None:
         title = metricTitle
@@ -309,7 +319,7 @@ def getTitle(timelyMetric, config, separator='\n'):
 
     if alertTitle != "":
         if config.min_alert_period is not None:
-            alertTitle = '(' + alertTitle + ') for greater than ' + config.min_alert_period
+            alertTitle = '(' + alertTitle + ') for at least ' + config.min_alert_period
         if config.last_alert is not None:
             alertTitle = alertTitle + ' in last ' + config.last_alert
         title += separator + alertTitle

client/src/main/python/timely/TimelyAnalytic.py

@@ -5,7 +5,7 @@ from TimelyAnalyticConfiguration import TimelyAnalyticConfiguration
 import DataOperations
 import pandas
 import re
-from pandas.tslib import Timestamp
+from datetime import datetime
 from datetime import timedelta
 
 from TimelyAlert import TimelyAlert
@@ -49,7 +49,7 @@ def keepConsecutiveAlerts(dataFrame, exceptions, minimumSpan):
                 currentLast = index
                 if currentFirst is None:
                     currentFirst = index
-                if int((currentLast - currentFirst).total_seconds() / 60) > int(minimumSpan):
+                if int((currentLast - currentFirst).total_seconds() / 60) >= int(minimumSpan):
                     result_span_values['bool'] = result_span_values['bool'] | current_result_span_values['bool']
             else:
                 current_result_span_values['bool'] = np.zeros(exceptions.shape, bool)
@@ -60,8 +60,22 @@ def keepConsecutiveAlerts(dataFrame, exceptions, minimumSpan):
     else:
         return exceptions
 
+def convertCounterToRate(timelyMetric, analyticConfig):
+    # Adjust counter metric into a rate metric
+    df = timelyMetric.getDataFrame()
+    columnNames = df[analyticConfig.groupByColumn].unique()
+    df = DataOperations.pivot(df, timelyMetric.metric, analyticConfig.groupByColumn)
+    df = DataOperations.resample(df, analyticConfig.sample, how=analyticConfig.how)
+    for c in columnNames:
+        df[c] = df[c].diff(1) / (analyticConfig.sample_minutes * 60 * 1000)
+    df = DataOperations.unpivot(df, timelyMetric.metric, analyticConfig.groupByColumn)
+    df = df.dropna()
+    timelyMetric.setDataFrame(df)
+
 def find_alerts(timelyMetric, analyticConfig, notebook=False):
 
+    if (analyticConfig.counter == True):
+        convertCounterToRate(timelyMetric, analyticConfig)
 
     df = timelyMetric.getDataFrame()
 
@@ -79,8 +93,6 @@ def find_alerts(timelyMetric, analyticConfig, notebook=False):
 
     combined = pandas.DataFrame()
 
-    now = Timestamp.now()
-
     seriesConfig = {}
     for i in graphDF_avg.columns:
         col = str(i)
@@ -107,18 +119,23 @@ def find_alerts(timelyMetric, analyticConfig, notebook=False):
 
         if analyticConfig.rolling_average_samples is not None:
             graphDF_avg = TimelyMetric.rolling_average(graphDF_avg, col, rolling_average=analyticConfig.rolling_average_samples)
-            if analyticConfig.alert_percentage is not None:
-                if analyticConfig.alert_percentage > 0:
-                    multiple = 1.0 + (float(abs(analyticConfig.alert_percentage)) / float(100))
-                    currCondition = graphDF[col].astype(float) > (graphDF_avg[col].astype(float) * multiple)
-                    result_values = addCondition(analyticConfig.orCondition, result_values, currCondition)
-                    any_conditions_values = True
-                if analyticConfig.alert_percentage < 0:
-                    multiple = 1.0 - (float(abs(analyticConfig.alert_percentage)) / float(100))
-                    if multiple > 0:
-                        currCondition = graphDF[col].astype(float) < (graphDF_avg[col].astype(float) * multiple)
-                        result_values = addCondition(analyticConfig.orCondition, result_values, currCondition)
-                        any_conditions_values = True
+            if analyticConfig.min_threshold_percentage is not None:
+                if analyticConfig.min_threshold_percentage >= 0:
+                    multiple = 1.0 + (float(abs(analyticConfig.min_threshold_percentage)) / float(100))
+                else:
+                    multiple = 1.0 - (float(abs(analyticConfig.min_threshold_percentage)) / float(100))
+                currCondition = graphDF[col].astype(float) < (graphDF_avg[col].astype(float) * multiple)
+                result_values = addCondition(analyticConfig.orCondition, result_values, currCondition)
+                any_conditions_values = True
+
+            if analyticConfig.max_threshold_percentage is not None:
+                if analyticConfig.max_threshold_percentage >= 0:
+                    multiple = 1.0 + (float(abs(analyticConfig.max_threshold_percentage)) / float(100))
+                else:
+                    multiple = 1.0 - (float(abs(analyticConfig.max_threshold_percentage)) / float(100))
+                currCondition = graphDF[col].astype(float) > (graphDF_avg[col].astype(float) * multiple)
+                result_values = addCondition(analyticConfig.orCondition, result_values, currCondition)
+                any_conditions_values = True
 
             if analyticConfig.average_min_threshold is not None:
                 currCondition = graphDF_avg[col].astype(float) < analyticConfig.average_min_threshold
@@ -139,16 +156,17 @@ def find_alerts(timelyMetric, analyticConfig, notebook=False):
             result_average = np.zeros(graphDF_avg[col].shape, bool)
         exceptional_average = graphDF_avg.loc[result_average, col]
 
-        # only evaluate the last analyticConfig.last_alert_minutes if set
-        if analyticConfig.last_alert_minutes is not None:
-            recentEnoughBegin = now - timedelta(minutes=analyticConfig.last_alert_minutes)
-            exceptional_values = exceptional_values.ix[recentEnoughBegin:now]
-            exceptional_average = exceptional_average.ix[recentEnoughBegin:now]
-
         # only keep alerts that are in consecutive periods of length analyticConfig.min_alert_minutes
         exceptional_values = keepConsecutiveAlerts(graphDF, exceptional_values, analyticConfig.min_alert_minutes)
         exceptional_average = keepConsecutiveAlerts(graphDF_avg, exceptional_average, analyticConfig.min_alert_minutes)
 
+        # only evaluate the last analyticConfig.last_alert_minutes if set
+        if analyticConfig.last_alert_minutes is not None:
+            end = datetime.fromtimestamp(timelyMetric.timeDateRange.getEndMs() / 1000.00, DataOperations.utc)
+            recentEnoughBegin = end - timedelta(minutes=analyticConfig.last_alert_minutes)
+            exceptional_values = exceptional_values.ix[recentEnoughBegin:end]
+            exceptional_average = exceptional_average.ix[recentEnoughBegin:end]
+
         anyValueExceptions = exceptional_values.size > 0
         anyAverageExceptions = exceptional_average.size > 0
 
@@ -156,8 +174,13 @@ def find_alerts(timelyMetric, analyticConfig, notebook=False):
             combined[col] = graphDF[col]
 
         if analyticConfig.rolling_average_samples is not None:
-            if (analyticConfig.display.lower() == "all") or (analyticConfig.display.lower() == "alerts" and (anyAverageExceptions or analyticConfig.alert_percentage is not None)):
-                combined[col+'_avg'] = graphDF_avg[col]
+            if (analyticConfig.display.lower() == "all"):
+                combined[col + '_avg'] = graphDF_avg[col]
+            else:
+                if (anyAverageExceptions):
+                    combined[col + '_avg'] = graphDF_avg[col]
+                if (anyValueExceptions and (analyticConfig.min_threshold_percentage is not None or analyticConfig.max_threshold_percentage is not None)):
+                    combined[col + '_avg'] = graphDF_avg[col]
 
         if ((analyticConfig.display.lower() == "all") or (analyticConfig.display.lower() == "alerts" and anyValueExceptions)):
             combined[col + '_warn'] = exceptional_values.dropna()
@@ -189,7 +212,7 @@ def find_alerts(timelyMetric, analyticConfig, notebook=False):
         combined = TimelyMetric.unpivot(combined, timelyMetric.metric, groupByColumn=alertAnalyticConfig.groupByColumn)
         combined = combined.sort_index()
         combined['date'] = combined.index.values
-        combined = combined.sort_values(['date', analyticConfig.groupByColumn])
+        combined = combined.sort_values(['date', alertAnalyticConfig.groupByColumn])
         combined = combined.drop(['date'], 1)
         combined = combined.dropna()
         combined = DataOperations.ensureMinSeriesLength(combined, alertAnalyticConfig.groupByColumn)

client/src/main/python/timely/TimelyAnalyticConfiguration.py

@@ -10,6 +10,7 @@ class TimelyAnalyticConfiguration():
             self.groupByColumn = analyticConfig.get('groupByColumn', None)
             self.includeColRegex = analyticConfig.get('includeColRegex', None)
             self.excludeColRegex = analyticConfig.get('excludeColRegex', None)
+            self.counter = analyticConfig.get('counter', False)
             self.sample_period = analyticConfig.get('sample', None)
             self.how = analyticConfig.get('how', 'mean')
             self.rolling_average_period = analyticConfig.get('rolling_average_period', None)
@@ -17,7 +18,8 @@ class TimelyAnalyticConfiguration():
             self.average_min_threshold = analyticConfig.get('average_min_threshold', None)
             self.max_threshold = analyticConfig.get('max_threshold', None)
             self.average_max_threshold = analyticConfig.get('average_max_threshold', None)
-            self.alert_percentage = analyticConfig.get('alert_percentage', None)
+            self.min_threshold_percentage = analyticConfig.get('min_threshold_percentage', None)
+            self.max_threshold_percentage = analyticConfig.get('max_threshold_percentage', None)
             self.min_alert_period = analyticConfig.get('min_alert_period', None)
             boolean = analyticConfig.get('boolean', 'or')
             self.orCondition = boolean == 'or' or boolean == 'OR'
@@ -56,6 +58,7 @@ class TimelyAnalyticConfiguration():
             self.groupByColumn = analyticConfig.groupByColumn
             self.includeColRegex = analyticConfig.includeColRegex
             self.excludeColRegex = analyticConfig.excludeColRegex
+            self.counter = analyticConfig.counter
             self.sample_period = analyticConfig.sample_period
             self.sample_minutes = analyticConfig.sample_minutes
             self.sample = analyticConfig.sample
@@ -67,7 +70,8 @@ class TimelyAnalyticConfiguration():
             self.average_min_threshold = analyticConfig.average_min_threshold
             self.max_threshold = analyticConfig.max_threshold
             self.average_max_threshold = analyticConfig.average_max_threshold
-            self.alert_percentage = analyticConfig.alert_percentage
+            self.min_threshold_percentage = analyticConfig.min_threshold_percentage
+            self.max_threshold_percentage = analyticConfig.max_threshold_percentage
             self.min_alert_period = analyticConfig.min_alert_period
             self.min_alert_minutes = analyticConfig.min_alert_minutes
             self.orCondition = analyticConfig.orCondition

client/src/main/python/timely/TimelyMetric.py

@@ -81,10 +81,9 @@ class TimelyMetric(TimelyWebSocketClient):
         self.debug = False
         self.dataFrame = None
         self.data = []
+        self.timeDateRange = TimeDateRange(beginStr, endStr, period)
 
-        timeDateRange = TimeDateRange(beginStr, endStr, period)
-
-        TimelyWebSocketClient.__init__(self, hostport, metric, tags, timeDateRange.getBeginMs(), timeDateRange.getEndMs())
+        TimelyWebSocketClient.__init__(self, hostport, metric, tags, self.timeDateRange.getBeginMs(), self.timeDateRange.getEndMs())
 
     def debugOn(self):
         self.debug = True

client/src/main/python/timely/sampleAnalytic.py

@@ -9,6 +9,7 @@ timelyMetric = TimelyMetric.TimelyMetric(hostport, '', None, None, None, 48, Non
 
 analyticConfig = TimelyAnalyticConfiguration({
     'system_name' : 'TestSystem',
+    'counter' : False,
     'includeColRegex' : None,
     'excludeColRegex' : None,
     'groupByColumn' : 'instance',
@@ -19,7 +20,8 @@ analyticConfig = TimelyAnalyticConfiguration({
     'average_min_threshold' : None,
     'max_threshold' : None,
     'average_max_threshold' : None,
-    'alert_percentage' : 25,
+    'min_threshold_percentage' : -50,
+    'max_threshold_percentage' : 50,
     'boolean' : 'and',
     'min_alert_period' : '5 minutes',
     'last_alert' : '1 hour',

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.5</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <artifactId>collectd-nsq-plugin</artifactId>
     <dependencies>
@@ -16,6 +16,10 @@
             <groupId>org.apache.httpcomponents</groupId>
             <artifactId>httpclient</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-pool2</artifactId>
+        </dependency>
         <dependency>
             <groupId>timely</groupId>
             <artifactId>collectd-plugin-parent</artifactId>

collectd-nsq-plugin/src/main/java/timely/collectd/plugin/PooledCloseableHttpClientFactory.java

@@ -0,0 +1,97 @@
+package timely.collectd.plugin;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.PooledObjectFactory;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.collectd.api.OConfigItem;
+
+import java.util.concurrent.TimeUnit;
+
+public class PooledCloseableHttpClientFactory implements PooledObjectFactory {
+
+    private int connectionRequestTimeout = 10000; // 10 seconds in milliseconds
+    private int connectTimeout = 30000; // 30 seconds in milliseconds
+    private int socketTimeout = 60000; // 60 seconds in milliseconds
+    private long connectionTimeToLive = 300000; // 300 seconds in milliseconds
+
+    public int config(OConfigItem config) {
+        for (OConfigItem child : config.getChildren()) {
+            switch (child.getKey()) {
+                case "connectionRequestTimeout":
+                case "ConnectionRequestTimeout":
+                    connectionRequestTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "connectTimeout":
+                case "ConnectTimeout":
+                    connectTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "socketTimeout":
+                case "SocketTimeout":
+                    socketTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "connectionTimeToLive":
+                case "ConnectionTimeToLive":
+                    connectionTimeToLive = Long.parseLong(child.getValues().get(0).getString());
+                    break;
+                default:
+            }
+        }
+        return 0;
+    }
+
+    @Override
+    public PooledObject makeObject() throws Exception {
+        RequestConfig requestConfig = RequestConfig.custom().setConnectionRequestTimeout(connectionRequestTimeout)
+                .setConnectTimeout(connectTimeout).setSocketTimeout(socketTimeout).build();
+        HttpClientBuilder builder = HttpClientBuilder.create();
+        builder.setDefaultRequestConfig(requestConfig);
+        builder.setConnectionTimeToLive(connectionTimeToLive, TimeUnit.MILLISECONDS);
+        return new DefaultPooledObject(builder.build());
+    }
+
+    @Override
+    public void destroyObject(PooledObject pooledObject) throws Exception {
+        try {
+            ((CloseableHttpClient) pooledObject.getObject()).close();
+        } catch (Exception e) {
+            // do nothing
+        } finally {
+            pooledObject.invalidate();
+            ;
+        }
+    }
+
+    @Override
+    public boolean validateObject(PooledObject pooledObject) {
+        return pooledObject.getObject() != null;
+    }
+
+    @Override
+    public void activateObject(PooledObject pooledObject) throws Exception {
+        // do nothing
+    }
+
+    @Override
+    public void passivateObject(PooledObject pooledObject) throws Exception {
+        // do nothing
+    }
+
+    public void setConnectionRequestTimeout(int connectionRequestTimeout) {
+        this.connectionRequestTimeout = connectionRequestTimeout;
+    }
+
+    public void setConnectTimeout(int connectTimeout) {
+        this.connectTimeout = connectTimeout;
+    }
+
+    public void setSocketTimeout(int socketTimeout) {
+        this.socketTimeout = socketTimeout;
+    }
+
+    public void setConnectionTimeToLive(long connectionTimeToLive) {
+        this.connectionTimeToLive = connectionTimeToLive;
+    }
+}

collectd-nsq-plugin/src/main/java/timely/collectd/plugin/WriteNSQPlugin.java

@@ -3,8 +3,11 @@ package timely.collectd.plugin;
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.util.Random;
 
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.entity.EntityBuilder;
 import org.apache.http.client.methods.CloseableHttpResponse;
@@ -33,10 +36,11 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
     private int port = 0;
     private String topic = "metrics#ephemeral";
     private String endpoint = null;
-    private ByteArrayOutputStream baos = null;
-    private DataOutputStream out = null;
+    private GenericObjectPool<CloseableHttpClient> clientPool = null;
 
-    private CloseableHttpClient client = HttpClients.createDefault();
+    // pool should be limited by the number of WriteThreads configured in
+    // collectd
+    final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
 
     public WriteNSQPlugin() {
         Collectd.registerConfig(WriteNSQPlugin.class.getName(), this);
@@ -44,16 +48,18 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
         Collectd.registerWrite(WriteNSQPlugin.class.getName(), this);
     }
 
-    public synchronized int write(ValueList vl) {
-        baos = new ByteArrayOutputStream();
-        out = new DataOutputStream(baos);
-        super.process(vl);
+    public int write(ValueList vl) {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream out = new DataOutputStream(baos);
+        super.process(vl, out);
+        CloseableHttpClient client = null;
         try {
             out.flush();
             HttpPost post = new HttpPost(endpoint);
             EntityBuilder request = EntityBuilder.create();
             request.setBinary(baos.toByteArray());
             post.setEntity(request.build());
+            client = clientPool.borrowObject();
             CloseableHttpResponse response = client.execute(post);
             try {
                 int code = response.getStatusLine().getStatusCode();
@@ -66,13 +72,25 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
             } finally {
                 response.close();
             }
-        } catch (IOException e) {
+        } catch (Exception e) {
             Collectd.logError("Error sending metrics to NSQ: " + e.getMessage());
+            if (client != null) {
+                try {
+                    client.close();
+                } catch (IOException e1) {
+                }
+                try {
+                    clientPool.invalidateObject(client);
+                } catch (Exception e2) {
+                }
+            }
         } finally {
+            if (client != null) {
+                clientPool.returnObject(client);
+            }
             if (null != out) {
                 try {
                     out.close();
-                    out = null;
                 } catch (IOException e) {
                 }
             }
@@ -81,9 +99,9 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
     }
 
     @Override
-    public synchronized void write(String metric) {
+    public void write(String metric, OutputStream out) {
         try {
-            out.writeBytes(metric);
+            ((DataOutputStream) out).writeBytes(metric);
         } catch (IOException e) {
             Collectd.logError("Error writing metric: " + e.getMessage());
             try {
@@ -93,21 +111,23 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
         }
     }
 
-    public void flush() {
-    }
-
-    public synchronized int shutdown() {
+    public int shutdown() {
         Collectd.logInfo("Shutting down connection to NSQ at " + host + ":" + port);
-        try {
-            client.close();
-        } catch (IOException e) {
-            Collectd.logError("Error closing HttpClient: " + e.getMessage());
-        }
+        clientPool.close();
         return 0;
     }
 
-    public synchronized int config(OConfigItem config) {
+    public int config(OConfigItem config) {
         super.config(config);
+        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
+        // use max size for maxTotal and maxIdle
+        // no need to activate or passivate
+        poolConfig.setMaxTotal(POOL_MAX_SIZE);
+        poolConfig.setMaxIdle(POOL_MAX_SIZE);
+        poolConfig.setTestOnReturn(true);
+        PooledCloseableHttpClientFactory pooledCloseableHttpClientFactory = new PooledCloseableHttpClientFactory();
+        pooledCloseableHttpClientFactory.config(config);
+
         for (OConfigItem child : config.getChildren()) {
             switch (child.getKey()) {
                 case "host":
@@ -133,17 +153,13 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
                 default:
             }
         }
+        if (host == null || port == 0 || topic == null) {
+            Collectd.logError("NSQ host, port, and topic must be configured");
+            return -1;
+        }
+        this.clientPool = new GenericObjectPool(pooledCloseableHttpClientFactory, poolConfig);
         this.endpoint = "http://" + host + ":" + port + "/mpub?topic=" + topic;
-        if (null != host) {
-            return 0;
-        }
-        if (host == null) {
-            Collectd.logError("NSQ host must be configured");
-        }
-        if (port == 0) {
-            Collectd.logError("NSQ port must be configured");
-        }
-        return -1;
+        return 0;
     }
 
 }

collectd-plugin-parent/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.5</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <artifactId>collectd-plugin-parent</artifactId>
     <dependencies>

collectd-plugin-parent/src/main/java/timely/collectd/plugin/CollectDPluginParent.java

@@ -1,5 +1,6 @@
 package timely.collectd.plugin;
 
+import java.io.OutputStream;
 import java.text.MessageFormat;
 import java.util.HashSet;
 import java.util.Set;
@@ -54,9 +55,9 @@ public abstract class CollectDPluginParent {
         return 0;
     }
 
-    public abstract void write(String metric);
+    public abstract void write(String metric, OutputStream out);
 
-    public void process(ValueList vl) {
+    public void process(ValueList vl, OutputStream out) {
         debugValueList(vl);
 
         StringBuilder metric = new StringBuilder();
@@ -214,7 +215,7 @@ public abstract class CollectDPluginParent {
             String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(),
                     tagsWithSample.toString());
             Collectd.logDebug("Writing: " + datapoint);
-            write(datapoint);
+            write(datapoint, out);
         }
     }
 

collectd-plugin-parent/src/test/java/timely/collectd/plugin/PluginParentTest.java

@@ -2,6 +2,7 @@ package timely.collectd.plugin;
 
 import static org.junit.Assert.*;
 
+import java.io.OutputStream;
 import java.util.Collections;
 
 import net.jcip.annotations.NotThreadSafe;
@@ -29,7 +30,7 @@ public class PluginParentTest {
         }
 
         @Override
-        public void write(String metric) {
+        public void write(String metric, OutputStream out) {
             result = metric;
         }
 
@@ -58,7 +59,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put statsd.dfs.BlocksRead 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=DataNode sample=value sampleType=GAUGE\n",
                 result);
@@ -78,7 +79,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals("put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n",
                 result);
     }
@@ -97,7 +98,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=bar sample=value sampleType=GAUGE\n",
                 result);
@@ -117,7 +118,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.hddtemp.temperature 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n",
                 result);
@@ -137,7 +138,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.smart.smart_badsectors 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n",
                 result);
@@ -158,7 +159,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.smart.raw-read-error-rate 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n",
                 result);
@@ -179,7 +180,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.smart.Total_LBAs_Read 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo code=242 instance=sda sample=value sampleType=GAUGE\n",
                 result);
@@ -200,7 +201,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.snmp.if_octets 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=Ethernet1 sample=value sampleType=GAUGE\n",
                 result);
@@ -222,7 +223,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.sensors.temperature.coretemp-isa-0000 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=1 sample=value sampleType=GAUGE\n",
                 result);
@@ -242,7 +243,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.haproxy.run_queue 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n",
                 result);
@@ -263,7 +264,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.haproxy.queue_current 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo proxy_name=proxy1 service_name=server1 sample=value sampleType=GAUGE\n",
                 result);
@@ -284,7 +285,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.ethstat.tx_comp_queue_full 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=eth0 sample=value sampleType=GAUGE\n",
                 result);
@@ -306,7 +307,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.ethstat.rx_bytes 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo queue=1 instance=eth0 sample=value sampleType=GAUGE\n",
                 result);
@@ -327,7 +328,7 @@ public class PluginParentTest {
         vl.setDataSet(ds);
 
         TestPlugin test = new TestPlugin();
-        test.process(vl);
+        test.process(vl, null);
         assertEquals(
                 "put sys.ipmi.temperature 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=LAN_NIC_Temp_system_board_(3.2) sample=value sampleType=GAUGE\n",
                 result);

collectd-timely-plugin/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.5</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <artifactId>collectd-timely-plugin</artifactId>
     <dependencies>
@@ -12,6 +12,10 @@
             <groupId>collectd</groupId>
             <artifactId>collectd-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-pool2</artifactId>
+        </dependency>
         <dependency>
             <groupId>net.jcip</groupId>
             <artifactId>jcip-annotations</artifactId>

collectd-timely-plugin/src/main/java/timely/collectd/plugin/PooledSocketFactory.java

@@ -0,0 +1,125 @@
+package timely.collectd.plugin;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.PooledObjectFactory;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.collectd.api.Collectd;
+import org.collectd.api.OConfigItem;
+
+import java.net.Socket;
+
+public class PooledSocketFactory implements PooledObjectFactory {
+
+    private String host = null;
+    private int port = 0;
+    private static final long initialBackoff = 2000;
+    private int socketTimeout = 60000; // 60 seconds in milliseconds
+    private long connectionTimeToLive = 300000; // 300 seconds in milliseconds
+
+    public int config(OConfigItem config) {
+        for (OConfigItem child : config.getChildren()) {
+            switch (child.getKey()) {
+                case "host":
+                case "hostname":
+                case "Host":
+                case "HostName":
+                    host = child.getValues().get(0).getString();
+                    break;
+                case "Port":
+                case "port":
+                    port = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "socketTimeout":
+                case "SocketTimeout":
+                    socketTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "connectionTimeToLive":
+                case "ConnectionTimeToLive":
+                    connectionTimeToLive = Long.parseLong(child.getValues().get(0).getString());
+                    break;
+                default:
+
+            }
+        }
+        if (null != host) {
+            return 0;
+        }
+        if (host == null) {
+            Collectd.logError("Timely host must be configured");
+        }
+        if (port == 0) {
+            Collectd.logError("Timely port must be configured");
+        }
+        return -1;
+    }
+
+    private Socket connect() {
+
+        Socket socket = null;
+        long currentBackoff = initialBackoff;
+        while (socket == null) {
+            try {
+                socket = new Socket(host, port);
+                if (socketTimeout > -1) {
+                    socket.setSoTimeout(socketTimeout);
+                }
+                Collectd.logInfo("Connected to Timely at " + host + ":" + port + " from local port:"
+                        + socket.getLocalPort());
+            } catch (Exception e) {
+                Collectd.logError("Error connecting to Timely at " + host + ":" + port + ". Error: " + e.getMessage()
+                        + ".  Will retry connection in " + currentBackoff + " ms.");
+                try {
+                    Thread.sleep(currentBackoff);
+                } catch (InterruptedException e1) {
+
+                }
+                // max out reconnect period at one minute
+                currentBackoff = (currentBackoff * 2 > 60000) ? 60000 : currentBackoff * 2;
+            }
+        }
+        return socket;
+    }
+
+    @Override
+    public PooledObject makeObject() throws Exception {
+        return new DefaultPooledObject(connect());
+    }
+
+    @Override
+    public void destroyObject(PooledObject pooledObject) throws Exception {
+        try {
+            Collectd.logDebug("Shutting down connection to Timely at " + host + ":" + port);
+            Socket socket = (Socket) pooledObject.getObject();
+            if (socket != null) {
+                socket.close();
+            }
+        } catch (Exception e) {
+            Collectd.logError("Error closing connection to Timely at " + host + ":" + port + ". Error: "
+                    + e.getMessage());
+        } finally {
+            pooledObject.invalidate();
+            ;
+        }
+    }
+
+    @Override
+    public boolean validateObject(PooledObject pooledObject) {
+        if (connectionTimeToLive > -1) {
+            if (pooledObject.getCreateTime() + connectionTimeToLive > System.currentTimeMillis()) {
+                return false;
+            }
+        }
+        Socket socket = (Socket) pooledObject.getObject();
+        return socket != null && !socket.isClosed();
+    }
+
+    @Override
+    public void activateObject(PooledObject pooledObject) throws Exception {
+        // do nothing
+    }
+
+    @Override
+    public void passivateObject(PooledObject pooledObject) throws Exception {
+        // do nothing
+    }
+}

collectd-timely-plugin/src/main/java/timely/collectd/plugin/WriteTimelyPlugin.java

@@ -1,9 +1,12 @@
 package timely.collectd.plugin;
 
 import java.io.IOException;
+import java.io.OutputStream;
 import java.io.PrintWriter;
 import java.net.Socket;
 
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.collectd.api.Collectd;
 import org.collectd.api.CollectdConfigInterface;
 import org.collectd.api.CollectdShutdownInterface;
@@ -19,12 +22,11 @@ import org.collectd.api.ValueList;
 public class WriteTimelyPlugin extends CollectDPluginParent implements CollectdConfigInterface,
         CollectdShutdownInterface, CollectdWriteInterface {
 
-    private String host = null;
-    private int port = 0;
-    private Socket sock = null;
-    private PrintWriter out = null;
-    private long connectTime = 0L;
-    private long backoff = 2000;
+    private GenericObjectPool<Socket> socketPool = null;
+
+    // pool should be limited by the number of WriteThreads configured in
+    // collectd
+    final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
 
     public WriteTimelyPlugin() {
         Collectd.registerConfig(WriteTimelyPlugin.class.getName(), this);
@@ -32,98 +34,63 @@ public class WriteTimelyPlugin extends CollectDPluginParent implements CollectdC
         Collectd.registerWrite(WriteTimelyPlugin.class.getName(), this);
     }
 
-    private synchronized int connect() {
-        if (null == sock || !sock.isConnected() || out.checkError()) {
-            if (System.currentTimeMillis() > (connectTime + backoff)) {
-                try {
-                    connectTime = System.currentTimeMillis();
-                    sock = new Socket(host, port);
-                    out = new PrintWriter(sock.getOutputStream(), false);
-                    backoff = 2000;
-                    Collectd.logInfo("Connected to Timely at " + host + ":" + port);
-                } catch (IOException e) {
-                    Collectd.logError("Error connecting to Timely at " + host + ":" + port + ". Error: "
-                            + e.getMessage());
-                    backoff = backoff * 2;
-                    sock = null;
-                    out = null;
-                    Collectd.logWarning("Will retry connection in " + backoff + " ms.");
-                    return -1;
-                }
-            } else {
-                Collectd.logWarning("Not writing to Timely, waiting to reconnect");
+    public int write(ValueList vl) {
+
+        Socket socket = null;
+        try {
+            socket = socketPool.borrowObject();
+            if (socket == null) {
                 return -1;
             }
+            super.process(vl, socket.getOutputStream());
+        } catch (Exception e) {
+            Collectd.logWarning(e.getMessage());
+            try {
+                // close socket so that the object pool will discard it and
+                // reconnect
+                socket.close();
+            } catch (IOException e1) {
+                Collectd.logError(e1.getMessage());
+            }
+            return -1;
+        } finally {
+            if (socket != null) {
+                socketPool.returnObject(socket);
+            }
         }
         return 0;
     }
 
-    public synchronized int write(ValueList vl) {
-
-        int c = connect();
-        if (c != 0) {
-            return c;
-        }
-        super.process(vl);
-        return 0;
-    }
-
     @Override
-    public synchronized void write(String metric) {
-        out.write(metric);
-    }
-
-    public synchronized void flush() {
-        if (null != out) {
-            out.flush();
+    public void write(String metric, OutputStream out) {
+        PrintWriter printWriter = new PrintWriter(out, false);
+        printWriter.write(metric);
+        printWriter.flush();
+        if (printWriter.checkError()) {
+            throw new RuntimeException("Error writing to Timely");
         }
     }
 
-    public synchronized int shutdown() {
-        Collectd.logInfo("Shutting down connection to Timely at " + host + ":" + port);
-        if (null != sock) {
-            try {
-                if (null != out) {
-                    out.close();
-                }
-                sock.close();
-            } catch (IOException e) {
-                Collectd.logError("Error closing connection to Timely at " + host + ":" + port + ". Error: "
-                        + e.getMessage());
-                return -1;
-            }
-        }
+    public int shutdown() {
+        socketPool.close();
         return 0;
     }
 
-    public synchronized int config(OConfigItem config) {
+    public int config(OConfigItem config) {
         super.config(config);
-        for (OConfigItem child : config.getChildren()) {
-            switch (child.getKey()) {
-                case "host":
-                case "hostname":
-                case "Host":
-                case "HostName":
-                    host = child.getValues().get(0).getString();
-                    break;
-                case "Port":
-                case "port":
-                    port = Integer.parseInt(child.getValues().get(0).getString());
-                    break;
-                default:
-
-            }
+        int retval = 0;
+        PooledSocketFactory socketFactory = new PooledSocketFactory();
+        retval = socketFactory.config(config);
+        if (retval == 0) {
+            GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
+            // use max size for maxTotal and maxIdle
+            // no need to activate and passivate
+            poolConfig.setMaxTotal(POOL_MAX_SIZE);
+            poolConfig.setMaxIdle(POOL_MAX_SIZE);
+            poolConfig.setTestOnReturn(true);
+            socketPool = new GenericObjectPool(socketFactory, poolConfig);
         }
-        if (null != host) {
-            return 0;
-        }
-        if (host == null) {
-            Collectd.logError("Timely host must be configured");
-        }
-        if (port == 0) {
-            Collectd.logError("Timely port must be configured");
-        }
-        return -1;
+        return retval;
     }
 
 }

collectd-timely-plugin/src/test/java/timely/collectd/plugin/WriteTimelyPluginTest.java

@@ -19,12 +19,14 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.powermock.api.easymock.PowerMock;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
 @RunWith(PowerMockRunner.class)
 @PrepareForTest({ Collectd.class })
 @NotThreadSafe
+@PowerMockIgnore("javax.management.*")
 public class WriteTimelyPluginTest {
 
     private WriteTimelyPlugin plugin = null;
@@ -63,7 +65,6 @@ public class WriteTimelyPluginTest {
         Assert.assertEquals(0, plugin.config(config));
         // Seed with one metric to create the connection
         Assert.assertEquals(0, plugin.write(createMetric()));
-        plugin.flush();
         Thread.sleep(100);
         Assert.assertTrue(server.messageReceived());
     }
@@ -77,7 +78,6 @@ public class WriteTimelyPluginTest {
             Thread.sleep(1000);
         }
         Assert.assertEquals(0, plugin.write(createMetric()));
-        plugin.flush();
         Thread.sleep(100);
         Assert.assertTrue(server.messageReceived());
         plugin.shutdown();
@@ -94,7 +94,6 @@ public class WriteTimelyPluginTest {
             Thread.sleep(1000);
         }
         Assert.assertEquals(0, plugin.write(createMetric()));
-        plugin.flush();
         Thread.sleep(100);
         Assert.assertTrue(server.messageReceived());
         server.shutdown();
@@ -104,16 +103,18 @@ public class WriteTimelyPluginTest {
         server.create();
         Thread t2 = new Thread(server);
         t2.start();
+        // Need to call this again because the server is not guaranteed to be
+        // listening on the same local port as the first time that it was
+        // started
+        setupPlugin();
         while (!server.ready()) {
             Thread.sleep(1000);
             // Keep sending metrics to plugin to force reconnect
             int result = plugin.write(createMetric());
             System.out.println("Wrote to client, result: " + result);
             Assert.assertEquals(0, result);
-            plugin.flush();
         }
         Assert.assertEquals(0, plugin.write(createMetric()));
-        plugin.flush();
         Thread.sleep(1000);
         Assert.assertTrue(server.messageReceived());
         plugin.shutdown();
@@ -139,6 +140,7 @@ public class WriteTimelyPluginTest {
             server.setReceiveBufferSize(8192);
             host = server.getInetAddress().getHostAddress();
             port = server.getLocalPort();
+            System.out.println("Local port " + port);
         }
 
         public String getHost() {

common/pom.xml

@@ -3,7 +3,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>0.0.5</version>
+    <version>0.0.6-SNAPSHOT</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>

grafana/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.5</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <build>
         <plugins>

grafana/timely-app/dist/plugin.json

@@ -23,8 +23,8 @@
       {"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
       {"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
     ],
-    "version": "0.0.5-SNAPSHOT",
-    "updated": "2017-07-21"
+    "version": "0.0.6-SNAPSHOT",
+    "updated": "2017-08-11"
   },
 
   "includes": [

grafana/timely-app/plugin.json

@@ -23,8 +23,8 @@
       {"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
       {"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
     ],
-    "version": "0.0.5-SNAPSHOT",
-    "updated": "2017-07-21"
+    "version": "0.0.6-SNAPSHOT",
+    "updated": "2017-08-11"
   },
 
   "includes": [

pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>0.0.5</version>
+    <version>0.0.6-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>Timely</name>
     <description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -44,6 +44,7 @@
         <version.collectd>5.5.1</version.collectd>
         <version.commons-collections4>4.1</version.commons-collections4>
         <version.commons-lang3>3.5</version.commons-lang3>
+        <version.commons-pool2>2.4.2</version.commons-pool2>
         <version.disruptor>3.3.6</version.disruptor>
         <version.fast-classpath-scanner>2.0.5</version.fast-classpath-scanner>
         <version.google-flatbuffers>1.4.0</version.google-flatbuffers>
@@ -199,6 +200,11 @@
                 <artifactId>commons-lang3</artifactId>
                 <version>${version.commons-lang3}</version>
             </dependency>
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-pool2</artifactId>
+                <version>${version.commons-pool2}</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.hadoop</groupId>
                 <artifactId>hadoop-common</artifactId>

server/Dockerfile

@@ -1,13 +1,8 @@
 FROM centos:centos7
 
-MAINTAINER Jim Klucar "[email protected]"
-
 VOLUME /timely-server-src
 VOLUME /timely
 
-
-
-
 RUN yum -y update && \
   yum -y install java-1.8.0-openjdk-devel maven openssl apr wget epel-release && \
   yum -y install bouncycastle && \
@@ -20,5 +15,5 @@ ENV JAVA_HOME /usr/lib/jvm/java-1.8.0-openjdk
 
 EXPOSE 9804 54321 54322
 
-ENV TIMELY_VERSION 0.0.3-SNAPSHOT
+ENV TIMELY_VERSION 0.0.6-SNAPSHOT
 

server/conf/timely-docker.yml

@@ -15,6 +15,7 @@ timely:
       bufferSize: 52428800
     scan:
       threads: 4
+      max-downsample-memory: 10000000
   security:
     session-max-age: 86400
     allow-anonymous-access: true

server/conf/timely-standalone.yml

@@ -15,6 +15,7 @@ timely:
       bufferSize: 52428800
     scan:
       threads: 4
+      max-downsample-memory: 10000000
   security:
     session-max-age: 86400
     allow-anonymous-access: true

server/conf/timely.yml

@@ -15,6 +15,7 @@ timely:
       bufferSize: 52428800
     scan:
       threads: 4
+      max-downsample-memory: 10000000
   security:
     session-max-age: 86400
     allow-anonymous-access: false

server/docker-compose.yml

@@ -11,7 +11,7 @@ services:
      - ${PWD}/:/timely-server-src
     command: /timely-server-src/bin/docker/deploy-startup-standalone.sh
     environment:
-     - TIMELY_VERSION=0.0.3-SNAPSHOT
+     - TIMELY_VERSION=0.0.6-SNAPSHOT
   ingest:
     image: timely
     entrypoint: /timely-server-src/bin/docker/deploy-wait-and-insert.sh
@@ -22,9 +22,9 @@ services:
     environment:
      - TIMELY_HOST=timely
      - TIMELY_PORT=54321
-     - TIMELY_VERSION=0.0.3-SNAPSHOT
+     - TIMELY_VERSION=0.0.6-SNAPSHOT
   grafana:
-    image: grafana/grafana
+    image: grafana/grafana:4.4.3
     ports:
      - "3000:3000"
     volumes:

server/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.5</version>
+        <version>0.0.6-SNAPSHOT</version>
     </parent>
     <dependencies>
         <dependency>

server/src/main/java/timely/Configuration.java

@@ -218,6 +218,7 @@ public class Configuration {
     public class Scan {
 
         private int threads = 4;
+        private long maxDownsampleMemory = -1;
 
         public int getThreads() {
             return threads;
@@ -227,6 +228,15 @@ public class Configuration {
             this.threads = threads;
             return Configuration.this;
         }
+
+        public long getMaxDownsampleMemory() {
+            return maxDownsampleMemory;
+        }
+
+        public Configuration setMaxDownsampleMemory(long maxDownsampleMemory) {
+            this.maxDownsampleMemory = maxDownsampleMemory;
+            return Configuration.this;
+        }
     }
 
     public class Security {

server/src/main/java/timely/netty/websocket/WebSocketRequestDecoder.java

@@ -5,6 +5,7 @@ import io.netty.handler.codec.MessageToMessageDecoder;
 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
+import io.netty.handler.ssl.SslCompletionEvent;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 
@@ -99,6 +100,11 @@ public class WebSocketRequestDecoder extends MessageToMessageDecoder<WebSocketFr
                     LOG.warn("Channel idle, but no subscription id found on context. Unable to close subscriptions");
                 }
             }
+        } else if (evt instanceof SslCompletionEvent) {
+            SslCompletionEvent ssl = (SslCompletionEvent) evt;
+            if (!ssl.isSuccess()) {
+                LOG.error("SSL error: {}", ssl.getClass().getSimpleName(), ssl.cause());
+            }
         } else {
             LOG.warn("Received unhandled user event {}", evt);
         }

server/src/main/java/timely/sample/Aggregation.java

@@ -3,6 +3,7 @@ package timely.sample;
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import timely.api.request.timeseries.QueryRequest;
+import timely.model.ObjectSizeOf;
 
 import java.io.IOException;
 import java.io.ObjectInputStream;
@@ -140,4 +141,7 @@ public class Aggregation implements Iterable<Sample>, Serializable {
         return result;
     }
 
+    public int getNumBuckets() {
+        return buckets.size();
+    }
 }

server/src/main/java/timely/sample/Downsample.java

@@ -3,6 +3,7 @@ package timely.sample;
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import timely.api.request.timeseries.QueryRequest.RateOption;
+import timely.model.ObjectSizeOf;
 
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -64,5 +65,4 @@ public class Downsample extends Aggregation implements Iterable<Sample>, Seriali
         }
         return result;
     }
-
 }

server/src/main/java/timely/sample/iterators/DownsampleIterator.java

@@ -5,10 +5,7 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.Key;
@@ -17,8 +14,10 @@ import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.iterators.WrappingIterator;
 
+import org.apache.log4j.Logger;
 import timely.adapter.accumulo.MetricAdapter;
 import timely.model.Metric;
+import timely.model.ObjectSizeOf;
 import timely.model.Tag;
 import timely.api.response.TimelyException;
 import timely.sample.Aggregator;
@@ -30,20 +29,125 @@ public class DownsampleIterator extends WrappingIterator {
     private static final String START = "downsample.start";
     private static final String END = "downsample.end";
     private static final String PERIOD = "downsample.period";
+    private static final String MAX_DOWNSAMPLE_MEMORY = "downsample.maxDownsampleMemory";
     private static final String AGGCLASS = "downsample.aggclass";
 
     private DownsampleFactory factory;
     private final Map<Set<Tag>, Downsample> value = new HashMap<>();
+    private long start;
+    private long end;
+    private long period;
     private Key last;
+    private MemoryEstimator memoryEstimator = null;
+
+    static public class MemoryEstimator {
+
+        private boolean newBucket = false;
+        private long start;
+        private long period;
+        private long startOfCurrentBucket;
+        private long bucketsCompleted = 0;
+        private long bytesPerBucket = 0;
+        boolean highVolumeBuckets = false;
+        long maxDownsampleMemory = 0; // max aggregation memory (bytes) before
+                                      // current batch is returned (after
+                                      // bucket is complete)
+        LinkedList<Long> percentageChecks = new LinkedList<>();
+
+        public void reset() {
+            newBucket = false;
+            bucketsCompleted = 0;
+            bytesPerBucket = 0;
+            percentageChecks.add(5l);
+            percentageChecks.add(25l);
+            percentageChecks.add(50l);
+            percentageChecks.add(75l);
+        }
+
+        public MemoryEstimator(long maxDownsampleMemory, long start, long period) {
+            this.maxDownsampleMemory = maxDownsampleMemory;
+            this.start = start;
+            this.period = period;
+            this.startOfCurrentBucket = this.start;
+            reset();
+        }
+
+        public double getMemoryUsedPercentage() {
+            return (bucketsCompleted * bytesPerBucket) / (double) maxDownsampleMemory * 100;
+        }
+
+        public long getBucketsCompleted() {
+            return bucketsCompleted;
+        }
+
+        public long getBytesPerBucket() {
+            return bytesPerBucket;
+        }
+
+        public boolean isHighVolumeBuckets() {
+            return highVolumeBuckets;
+        }
+
+        public boolean shouldReturnBasedOnMemoryUsage(long timestamp, Object value) {
+
+            boolean shouldReturn = false;
+            if (maxDownsampleMemory >= 0) {
+                sample(timestamp);
+                if (isNewBucket()) {
+                    bucketsCompleted++;
+                    double memoryUsedPercentage = getMemoryUsedPercentage();
+                    if (memoryUsedPercentage >= 100) {
+                        shouldReturn = true;
+                    } else {
+                        boolean checkMemoryNow = false;
+                        Long check = percentageChecks.peek();
+                        if (check != null && memoryUsedPercentage >= check) {
+                            checkMemoryNow = true;
+                            percentageChecks.removeFirst();
+                        }
+                        // recalculate bytesPerBucket
+                        if (bytesPerBucket == 0 || highVolumeBuckets || checkMemoryNow) {
+                            long memoryUsed = ObjectSizeOf.Sizer.getObjectSize(value);
+                            bytesPerBucket = memoryUsed / bucketsCompleted;
+                        }
+                        // bucket average greater than 10% of max
+                        highVolumeBuckets = (bytesPerBucket / (double) maxDownsampleMemory) >= 0.1;
+                        shouldReturn = false;
+                    }
+                }
+            }
+            return shouldReturn;
+        }
+
+        public boolean isNewBucket() {
+            return newBucket;
+        }
+
+        private void sample(long timestamp) {
+            if (timestamp >= (startOfCurrentBucket + period)) {
+                newBucket = true;
+                startOfCurrentBucket = timestamp - ((timestamp - start) % period);
+            } else {
+                newBucket = false;
+            }
+        }
+    }
 
     @SuppressWarnings("unchecked")
     @Override
     public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options, IteratorEnvironment env)
             throws IOException {
         super.init(source, options, env);
-        long start = Long.parseLong(options.get(START));
-        long end = Long.parseLong(options.get(END));
-        long period = Long.parseLong(options.get(PERIOD));
+        start = Long.parseLong(options.get(START));
+        end = Long.parseLong(options.get(END));
+        period = Long.parseLong(options.get(PERIOD));
+        // default = 100 MB
+        long maxDownsampleMemory = -1;
+        if (options.containsKey(MAX_DOWNSAMPLE_MEMORY)) {
+            maxDownsampleMemory = Long.parseLong(options.get(MAX_DOWNSAMPLE_MEMORY));
+        }
+        memoryEstimator = new MemoryEstimator(maxDownsampleMemory, start, period);
+
         String aggClassname = options.get(AGGCLASS);
         Class<?> aggClass;
         try {
@@ -56,26 +160,38 @@ public class DownsampleIterator extends WrappingIterator {
 
     @Override
     public boolean hasTop() {
-        while (super.hasTop()) {
-            last = super.getTopKey();
-            Metric metric = MetricAdapter.parse(super.getTopKey(), super.getTopValue());
-            Set<Tag> tags = new HashSet<Tag>(metric.getTags());
-            Downsample sample = value.get(tags);
-            if (sample == null) {
+
+        if (super.hasTop()) {
+            while (super.hasTop()) {
+                Key topKey = super.getTopKey();
+                Value topValue = super.getTopValue();
+                Metric metric = MetricAdapter.parse(topKey, topValue);
+                long timestamp = metric.getValue().getTimestamp();
+                if (memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, value)) {
+                    break;
+                }
+                last = topKey;
+
+                Set<Tag> tags = new HashSet<Tag>(metric.getTags());
+                Downsample sample = value.get(tags);
+                if (sample == null) {
+                    try {
+                        value.put(tags, sample = factory.create());
+                    } catch (TimelyException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+                sample.add(metric.getValue().getTimestamp(), metric.getValue().getMeasure());
                 try {
-                    value.put(tags, sample = factory.create());
-                } catch (TimelyException e) {
-                    throw new RuntimeException(e);
+                    super.next();
+                } catch (IOException e) {
+                    throw new RuntimeException("Downstream next() failed", e);
                 }
             }
-            sample.add(metric.getValue().getTimestamp(), metric.getValue().getMeasure());
-            try {
-                super.next();
-            } catch (IOException e) {
-                throw new RuntimeException("Downstream next() failed", e);
-            }
+            return last != null;
+        } else {
+            return false;
         }
-        return last != null;
     }
 
     @Override
@@ -89,6 +205,9 @@ public class DownsampleIterator extends WrappingIterator {
                 ObjectOutputStream out = new ObjectOutputStream(bos)) {
             out.writeObject(value);
             out.flush();
+            // empty for next batch of downsamples
+            value.clear();
+            memoryEstimator.reset();
             return new Value(bos.toByteArray());
         } catch (IOException e) {
             throw new RuntimeException(e);
@@ -97,13 +216,17 @@ public class DownsampleIterator extends WrappingIterator {
 
     @Override
     public void next() throws IOException {
-        last = null;
+        if (!super.hasTop()) {
+            last = null;
+        }
     }
 
-    public static void setDownsampleOptions(IteratorSetting is, long start, long end, long period, String classname) {
+    public static void setDownsampleOptions(IteratorSetting is, long start, long end, long period,
+            long maxDownsampleMemory, String classname) {
         is.addOption(START, "" + start);
         is.addOption(END, "" + end);
         is.addOption(PERIOD, "" + period);
+        is.addOption(MAX_DOWNSAMPLE_MEMORY, "" + maxDownsampleMemory);
         is.addOption(AGGCLASS, classname);
     }
 

server/src/main/java/timely/store/DataStoreImpl.java

@@ -125,6 +125,7 @@ public class DataStoreImpl implements DataStore {
     private final InternalMetrics internalMetrics = new InternalMetrics();
     private final Timer internalMetricsTimer = new Timer(true);
     private final int scannerThreads;
+    private final long maxDownsampleMemory;
     private final BatchWriterConfig bwConfig;
     private final List<BatchWriter> writers = new ArrayList<>();
     private final ThreadLocal<BatchWriter> metaWriter = new ThreadLocal<>();
@@ -149,6 +150,7 @@ public class DataStoreImpl implements DataStore {
             bwConfig.setMaxMemory(getMemoryInBytes(accumuloConf.getWrite().getBufferSize()) / numWriteThreads);
             bwConfig.setMaxWriteThreads(accumuloConf.getWrite().getThreads());
             scannerThreads = accumuloConf.getScan().getThreads();
+            maxDownsampleMemory = accumuloConf.getScan().getMaxDownsampleMemory();
             anonAccessAllowed = conf.getSecurity().isAllowAnonymousAccess();
 
             metricsTable = conf.getMetricsTable();
@@ -506,7 +508,7 @@ public class DataStoreImpl implements DataStore {
                         LOG.trace("Downsample Aggregator type {}", daggClass.getSimpleName());
                         IteratorSetting is = new IteratorSetting(500, DownsampleIterator.class);
                         DownsampleIterator.setDownsampleOptions(is, startOfFirstPeriod, endOfLastPeriod, downsample,
-                                daggClass.getName());
+                                maxDownsampleMemory, daggClass.getName());
                         scanner.addScanIterator(is);
                     }
 
@@ -550,6 +552,7 @@ public class DataStoreImpl implements DataStore {
                 }
             }
             LOG.debug("Query time: {}", (System.currentTimeMillis() - now));
+            internalMetrics.addQueryResponse(result.size(), (System.currentTimeMillis() - now));
             return result;
         } catch (ClassNotFoundException | IOException | TableNotFoundException ex) {
             LOG.error("Error during query: " + ex.getMessage(), ex);

server/src/main/java/timely/store/InternalMetrics.java

@@ -20,12 +20,21 @@ public class InternalMetrics {
     private static final String METRICS_RECEIVED_METRIC = "timely.metrics.received";
     private static final String META_KEYS_METRIC = "timely.keys.meta.inserted";
     private static final String METRIC_KEYS_METRIC = "timely.keys.metric.inserted";
+    private static final String QUERIES_COMPLETED = "timely.query.num.completed";
+    private static final String QUERY_RETURN_TIME = "timely.query.return.time";
+    private static final String METRICS_RETURNED = "timely.query.metrics.returned";
+    private static final String METRIC_RETURN_RATE = "timely.query.metrics.returned.rate";
+
     private static final String HOSTNAME_TAG = "host";
 
     private String hostName = "localhost";
     private AtomicDouble numMetricsReceived = new AtomicDouble(0);
     private AtomicDouble numMetaKeysInserted = new AtomicDouble(0);
     private AtomicDouble numMetricKeysInserted = new AtomicDouble(0);
+    private AtomicDouble numQueriesCompleted = new AtomicDouble(0);
+    private AtomicDouble numMetricsReturned = new AtomicDouble(0);
+    private AtomicDouble elapsedQueryTime = new AtomicDouble(0);
+
     private List<Tag> tags = new ArrayList<Tag>();
 
     public InternalMetrics() {
@@ -50,6 +59,12 @@ public class InternalMetrics {
         numMetricKeysInserted.addAndGet(num);
     }
 
+    public void addQueryResponse(long numMetrics, long elapsedTime) {
+        numQueriesCompleted.addAndGet(1);
+        numMetricsReturned.addAndGet(numMetrics);
+        elapsedQueryTime.addAndGet(elapsedTime);
+    }
+
     public List<Metric> getMetricsAndReset() {
         List<Metric> metrics = new ArrayList<Metric>();
         long ts = System.currentTimeMillis();
@@ -59,6 +74,25 @@ public class InternalMetrics {
         numMetaKeysInserted.set(0);
         metrics.add(new Metric(METRIC_KEYS_METRIC, ts, numMetricKeysInserted.get(), tags));
         numMetricKeysInserted.set(0);
+
+        metrics.add(new Metric(QUERIES_COMPLETED, ts, numQueriesCompleted.get(), tags));
+        metrics.add(new Metric(METRICS_RETURNED, ts, numMetricsReturned.get(), tags));
+
+        if (elapsedQueryTime.get() > 0) {
+            // average response time per query
+            double averageQueryTime = (numQueriesCompleted.get() / elapsedQueryTime.get());
+            metrics.add(new Metric(QUERY_RETURN_TIME, ts, averageQueryTime, tags));
+
+            // metrics returned per minute
+            double rate = (numMetricsReturned.get() / (elapsedQueryTime.get() / (1000 * 60)));
+            metrics.add(new Metric(METRIC_RETURN_RATE, ts, rate, tags));
+        } else {
+            // if no elapsed time, then rate is zero
+            metrics.add(new Metric(METRIC_RETURN_RATE, ts, 0, tags));
+        }
+        numQueriesCompleted.set(0);
+        numMetricsReturned.set(0);
+        elapsedQueryTime.set(0);
         return metrics;
     }
 

server/src/test/java/timely/sample/iterators/DownsampleIteratorTest.java

@@ -4,13 +4,8 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.ColumnUpdate;
@@ -26,8 +21,10 @@ import org.junit.Test;
 import timely.Configuration;
 import timely.adapter.accumulo.MetricAdapter;
 import timely.model.Metric;
+import timely.model.ObjectSizeOf;
 import timely.model.Tag;
 import timely.auth.VisibilityCache;
+import timely.model.parse.TagListParser;
 import timely.sample.Downsample;
 import timely.sample.Sample;
 import timely.sample.aggregators.Avg;
@@ -72,6 +69,7 @@ public class DownsampleIteratorTest {
     }
 
     private void createTestData2() {
+        List<List<Tag>> listOfTagVariations = new ArrayList<>();
         List<Tag> tags = Collections.singletonList(new Tag("host", "host1"));
         List<Tag> tags2 = Collections.singletonList(new Tag("host", "host2"));
 
@@ -85,7 +83,7 @@ public class DownsampleIteratorTest {
     public void simpleGetOneSample() throws Exception {
         // check that data gets pulled out
         DownsampleIterator iter = new DownsampleIterator();
-        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData1, 100);
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData1, 100, -1);
         assertEquals(1, samples.size());
         for (Entry<Set<Tag>, Downsample> entry : samples.entrySet()) {
             Set<Tag> tags = entry.getKey();
@@ -104,7 +102,7 @@ public class DownsampleIteratorTest {
     @Test
     public void simpleGetTwoSamples() throws Exception {
         DownsampleIterator iter = new DownsampleIterator();
-        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 100);
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 100, -1);
         assertEquals(2, samples.size());
         for (Tag tag : new Tag[] { new Tag("host", "host1"), new Tag("host", "host2") }) {
             Downsample dsample = samples.get(Collections.singleton(tag));
@@ -128,7 +126,7 @@ public class DownsampleIteratorTest {
     @Test
     public void simpleTestDownsampling() throws Exception {
         DownsampleIterator iter = new DownsampleIterator();
-        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 200);
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData2, 200, -1);
         assertEquals(2, samples.size());
         for (Tag tag : new Tag[] { new Tag("host", "host1"), new Tag("host", "host2") }) {
             Downsample dsample = samples.get(Collections.singleton(tag));
@@ -149,17 +147,143 @@ public class DownsampleIteratorTest {
         }
     }
 
+    static public class SampleObject implements ObjectSizeOf {
+
+        private long sizeInBytes = 0;
+
+        public void setSizeInBytes(long sizeInBytes) {
+            this.sizeInBytes = sizeInBytes;
+        }
+
+        @Override
+        public long sizeInBytes() {
+            return sizeInBytes;
+        }
+    }
+
+    @Test
+    public void memoryEstimatorTestSmallObjects() {
+        long maxMemory = 1000;
+        long start = System.currentTimeMillis();
+        long period = 500l;
+        long sizeOfObjects = 20;
+        SampleObject o = new SampleObject();
+        DownsampleIterator.MemoryEstimator memoryEstimator = new DownsampleIterator.MemoryEstimator(maxMemory, start,
+                period);
+        boolean shouldReturn = false;
+        for (long x = 100; x <= 5000; x += 100) {
+            long timestamp = start + x;
+            o.setSizeInBytes(o.sizeInBytes() + sizeOfObjects);
+            shouldReturn = memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, o);
+            if (memoryEstimator.isNewBucket()) {
+                long memoryPercentageUsedCalculated = Math.round((double) o.sizeInBytes() / maxMemory * 100);
+                long memoryPercentageUsedEstimate = Math.round(memoryEstimator.getMemoryUsedPercentage());
+                long percentError = Math.round(Math.abs(memoryPercentageUsedCalculated - memoryPercentageUsedEstimate)
+                        / memoryPercentageUsedCalculated * 100);
+                assertTrue(percentError == 0);
+            }
+
+            if (shouldReturn) {
+                o.setSizeInBytes(0);
+                memoryEstimator.reset();
+            }
+        }
+        assertTrue(shouldReturn);
+    }
+
+    @Test
+    public void memoryEstimatorTestLargeObjects() {
+        long maxMemory = 10000;
+        long start = System.currentTimeMillis();
+        long period = 500l;
+        long sizeOfObjects = 200;
+        SampleObject o = new SampleObject();
+        DownsampleIterator.MemoryEstimator memoryEstimator = new DownsampleIterator.MemoryEstimator(maxMemory, start,
+                period);
+        boolean shouldReturn = false;
+        for (long x = 100; x <= 5000; x += 100) {
+            long timestamp = start + x;
+            o.setSizeInBytes(o.sizeInBytes() + sizeOfObjects);
+            shouldReturn = memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, o);
+            if (memoryEstimator.isNewBucket()) {
+                long memoryPercentageUsedCalculated = Math.round((double) o.sizeInBytes() / maxMemory * 100);
+                long memoryPercentageUsedEstimate = Math.round(memoryEstimator.getMemoryUsedPercentage());
+                long percentError = Math.round(Math.abs(memoryPercentageUsedCalculated - memoryPercentageUsedEstimate)
+                        / memoryPercentageUsedCalculated * 100);
+                assertTrue(percentError == 0);
+                assertTrue(memoryEstimator.isHighVolumeBuckets());
+            }
+
+            if (shouldReturn) {
+                o.setSizeInBytes(0);
+                memoryEstimator.reset();
+            }
+        }
+        assertTrue(shouldReturn);
+    }
+
+    private SortedMap<Key, Value> createTestData3(int elapsedTime, int skipInterval, int numTagVariations) {
+        SortedMap<Key, Value> testData3 = new TreeMap<>();
+        List<List<Tag>> listOfTagVariations = new ArrayList<>();
+        for (int x = 1; x <= numTagVariations; x++) {
+            listOfTagVariations.add(Collections.singletonList(new Tag("instance", Integer.toString(x))));
+        }
+
+        for (long i = 0; i < elapsedTime; i += skipInterval) {
+            for (List<Tag> tags : listOfTagVariations) {
+                put(testData3, new Metric("sys.loadAvg", i, i * 2, tags));
+            }
+        }
+        return testData3;
+    }
+
+    @Test
+    public void testDownsampleCombining() throws Exception {
+
+        int numTagVariations = 2;
+        int sampleInterval = 50;
+        int elapsedTime = 100;
+        int skipInterval = 10;
+        SortedMap<Key, Value> testData3 = createTestData3(elapsedTime, skipInterval, numTagVariations);
+        DownsampleIterator iter = new DownsampleIterator();
+        Map<Set<Tag>, Downsample> samples = runQuery(iter, testData3, sampleInterval, 1000);
+        assertEquals(numTagVariations, samples.size());
+        long totalBuckets = 0;
+        for (Entry<Set<Tag>, Downsample> entry : samples.entrySet()) {
+            totalBuckets = totalBuckets + entry.getValue().getNumBuckets();
+        }
+        assertEquals((elapsedTime / sampleInterval) * numTagVariations, totalBuckets);
+    }
+
     private Map<Set<Tag>, Downsample> runQuery(SortedKeyValueIterator<Key, Value> iter, SortedMap<Key, Value> testData,
-            long period) throws Exception {
+            long period, long maxDownsampleMemory) throws Exception {
         IteratorSetting is = new IteratorSetting(100, DownsampleIterator.class);
-        DownsampleIterator.setDownsampleOptions(is, 0, 1000, period, Avg.class.getName());
+        DownsampleIterator.setDownsampleOptions(is, 0, 1000, period, maxDownsampleMemory, Avg.class.getName());
         SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(testData);
         iter.init(source, is.getOptions(), null);
         iter.seek(new Range(), Collections.emptyList(), true);
-        assertTrue(iter.hasTop());
-        Key key = iter.getTopKey();
+        boolean hasTop = iter.hasTop();
+        assertTrue(hasTop);
+        Key key = null;
+        Map<Set<Tag>, Downsample> samples = new HashMap<>();
+        do {
+            Map<Set<Tag>, Downsample> currentSamples = DownsampleIterator.decodeValue(iter.getTopValue());
+            List<Downsample> downsampleArray = new ArrayList<>();
+            for (Entry<Set<Tag>, Downsample> entry : currentSamples.entrySet()) {
+                Downsample downsample = samples.get(entry.getKey());
+                if (downsample == null) {
+                    samples.put(entry.getKey(), entry.getValue());
+                } else {
+                    downsampleArray.clear();
+                    downsampleArray.add(downsample);
+                    downsampleArray.add(entry.getValue());
+                    samples.put(entry.getKey(), Downsample.combineDownsample(downsampleArray, null));
+                }
+            }
+            key = iter.getTopKey();
+        } while (iter.hasTop());
+
         assertEquals(testData.lastKey(), key);
-        Map<Set<Tag>, Downsample> samples = DownsampleIterator.decodeValue(iter.getTopValue());
         return samples;
     }