Add setup.py to .gitignore, add logging for metrics that fail to parse

Initial work

More work

cwImproved implementation of DownsampleIterator.MemoryEstimator

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

.gitignore

@@ -6,4 +6,5 @@ target/
 *.iml
 logs/
 server/pki/
+client/src/main/python/setup.py
 

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.7-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.7-SNAPSHOT</version>
     </parent>
+    <properties>
+        <!-- PEP 440 compliant version gets supstituted into src/main/python/template/setup.py -->
+        <python.client.version>0.0.7.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.7-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.7-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.7-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.7-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.7-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.7-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.7-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/adapter/accumulo/MetricAdapter.java

@@ -9,12 +9,14 @@ import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.security.ColumnVisibility;
 import org.apache.accumulo.core.util.ComparablePair;
 import org.apache.accumulo.core.util.Pair;
+import org.apache.hadoop.io.Text;
 import timely.model.Metric;
 import timely.model.Tag;
 import timely.model.parse.TagListParser;
 import timely.model.parse.TagParser;
 
 import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -28,12 +30,23 @@ public class MetricAdapter {
     private static final PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(),
             new LongLexicoder());
 
+    private static final PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(),
+            new StringLexicoder());
+
     private static final TagParser tagParser = new TagParser();
     private static final TagListParser tagListParser = new TagListParser();
 
     public static final ColumnVisibility EMPTY_VISIBILITY = new ColumnVisibility();
     public static final String VISIBILITY_TAG = "viz";
 
+    public static long roundTimestampToLastHour(long timestamp) {
+        return timestamp - (timestamp % 3600000);
+    }
+
+    public static long roundTimestampToNextHour(long timestamp) {
+        return timestamp - (timestamp % 3600000) + 3600000;
+    }
+
     public static Mutation toMutation(Metric metric) {
         final Mutation mutation = new Mutation(encodeRowKey(metric));
 
@@ -51,8 +64,9 @@ public class MetricAdapter {
                     .map(Tag::join)
                     .collect(Collectors.joining(","));
             // @formatter:on
-
-            mutation.put(cf, cq, extractVisibility(tags), metric.getValue().getTimestamp(), extractValue(metric));
+            byte[] cqBytes = encodeColQual(metric.getValue().getTimestamp(), cq);
+            mutation.put(new Text(cf.getBytes(Charset.forName("UTF-8"))), new Text(cqBytes), extractVisibility(tags),
+                    metric.getValue().getTimestamp(), extractValue(metric));
         }
         return mutation;
     }
@@ -87,10 +101,11 @@ public class MetricAdapter {
         // @formatter:off
         Metric.Builder builder = Metric.newBuilder()
                 .name(row.getFirst())
-                .value(row.getSecond(), ByteBuffer.wrap(v.get()).getDouble())
+                .value(k.getTimestamp(), ByteBuffer.wrap(v.get()).getDouble())
                 .tag(tagParser.parse(k.getColumnFamily().toString()));
         // @formatter:on
-        tagListParser.parse(k.getColumnQualifier().toString()).forEach(builder::tag);
+        ComparablePair<Long, String> cq = colQualCoder.decode(k.getColumnQualifier().getBytes());
+        tagListParser.parse(cq.getSecond()).forEach(builder::tag);
         if (includeVizTag && k.getColumnVisibility().getLength() > 0) {
             tagListParser.parse("viz=" + k.getColumnVisibility().toString()).forEach(builder::tag);
         }
@@ -106,7 +121,16 @@ public class MetricAdapter {
     }
 
     public static byte[] encodeRowKey(Metric metric) {
-        return encodeRowKey(metric.getName(), metric.getValue().getTimestamp());
+        // round timestamp to hour for scan efficiency and compression
+        return encodeRowKey(metric.getName(), roundTimestampToLastHour(metric.getValue().getTimestamp()));
+    }
+
+    public static byte[] encodeColQual(Long timestamp, String colQual) {
+        return colQualCoder.encode(new ComparablePair<>(timestamp, colQual));
+    }
+
+    public static Pair<Long, String> decodeColQual(byte[] colQual) {
+        return colQualCoder.decode(colQual);
     }
 
     public static Pair<String, Long> decodeRowKey(Key k) {

server/src/main/java/timely/api/request/MetricRequest.java

@@ -54,7 +54,11 @@ public class MetricRequest implements TcpRequest, HttpPostRequest, WebSocketRequ
     @Override
     public void parse(String line) {
         LOG.trace("Parsing Line: {}", line);
-        metric = metricParser.parse(line);
+        try {
+            metric = metricParser.parse(line);
+        } catch (Exception e) {
+            LOG.error("Error parsing metric: {}", line);
+        }
     }
 
     @Override

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.slf4j.LoggerFactory;
 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;
@@ -27,23 +26,36 @@ import timely.sample.DownsampleFactory;
 
 public class DownsampleIterator extends WrappingIterator {
 
+    private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(DownsampleIterator.class);
     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 DownsampleMemoryEstimator memoryEstimator = null;
 
     @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 DownsampleMemoryEstimator(maxDownsampleMemory, start, period);
+
         String aggClassname = options.get(AGGCLASS);
         Class<?> aggClass;
         try {
@@ -56,26 +68,45 @@ 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()) {
+            Key topKey = super.getTopKey();
+            Value topValue = super.getTopValue();
+            Metric metric = MetricAdapter.parse(topKey, topValue);
+            LOG.trace("entering hasTop() for metric=" + metric.toString());
+            while (super.hasTop()) {
+                topKey = super.getTopKey();
+                topValue = super.getTopValue();
+                metric = MetricAdapter.parse(topKey, topValue);
+                long timestamp = metric.getValue().getTimestamp();
+                if (memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, value)) {
+                    LOG.trace("returning current values - memory usage > " + memoryEstimator.maxDownsampleMemory
+                            + " for metric=" + metric.toString());
+                    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);
-            }
+            LOG.trace("exiting hasTop() for metric=" + metric.toString());
+            return last != null;
+        } else {
+            return false;
         }
-        return last != null;
     }
 
     @Override
@@ -89,6 +120,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 +131,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/sample/iterators/DownsampleMemoryEstimator.java

@@ -0,0 +1,101 @@
+package timely.sample.iterators;
+
+import timely.model.ObjectSizeOf;
+
+import java.util.LinkedList;
+
+public class DownsampleMemoryEstimator {
+
+    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.clear();
+        percentageChecks.add(1l);
+        percentageChecks.add(5l);
+        percentageChecks.add(25l);
+        percentageChecks.add(50l);
+        percentageChecks.add(75l);
+        percentageChecks.add(95l);
+    }
+
+    public DownsampleMemoryEstimator(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 < 100 || bucketsCompleted == 10 || 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;
+        }
+    }
+}

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

@@ -2,9 +2,12 @@ package timely.store;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.accumulo.core.client.Scanner;
 
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Range;
 import timely.model.Metric;
 import timely.api.request.timeseries.QueryRequest;
 import timely.api.request.timeseries.SearchLookupRequest;
@@ -13,6 +16,7 @@ import timely.api.response.TimelyException;
 import timely.api.response.timeseries.QueryResponse;
 import timely.api.response.timeseries.SearchLookupResponse;
 import timely.api.response.timeseries.SuggestResponse;
+import timely.model.Tag;
 
 public interface DataStore {
 
@@ -26,6 +30,12 @@ public interface DataStore {
 
     void flush() throws TimelyException;
 
+    long getAgeOffForMetric(String metricName);
+
+    List<Range> getQueryRanges(String metric, long start, long end, Set<Tag> colFamValues);
+
+    Set<Tag> getColumnFamilies(String metric, Map<String, String> tags) throws TableNotFoundException;
+
     Scanner createScannerForMetric(String sessionId, String metric, Map<String, String> tags, long startTime,
             long endTime, int lag, int scannerBatchSize, int scannerReadAhead) throws TimelyException;
 

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

@@ -5,25 +5,9 @@ import static org.apache.accumulo.core.conf.AccumuloConfiguration.getTimeInMilli
 import io.netty.handler.codec.http.HttpResponseStatus;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import java.nio.charset.Charset;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.TreeMap;
-import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -45,11 +29,7 @@ import org.apache.accumulo.core.client.TableExistsException;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.client.ZooKeeperInstance;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.accumulo.core.data.PartialKey;
-import org.apache.accumulo.core.data.Range;
-import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.data.*;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
 import org.apache.accumulo.core.iterators.user.RegExFilter;
 import org.apache.accumulo.core.security.Authorizations;
@@ -125,6 +105,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 +130,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();
@@ -312,7 +294,6 @@ public class DataStoreImpl implements DataStore {
             metaCache.addAll(toCache);
         }
         try {
-
             batchWriter.get().addMutation(MetricAdapter.toMutation(metric));
             internalMetrics.incrementMetricKeysInserted(metric.getTags().size());
         } catch (MutationsRejectedException e) {
@@ -464,13 +445,20 @@ public class DataStoreImpl implements DataStore {
     @Override
     public List<QueryResponse> query(QueryRequest msg) throws TimelyException {
         List<QueryResponse> result = new ArrayList<>();
+        LOG.debug("Query request {}", msg);
         long startTs = msg.getStart();
         long endTs = msg.getEnd();
+        StringBuilder metricList = new StringBuilder();
         try {
+            long numResults = 0;
             long now = System.currentTimeMillis();
             for (SubQuery query : msg.getQueries()) {
                 Map<Set<Tag>, List<Aggregation>> allSeries = new HashMap<>();
                 String metric = query.getMetric();
+                if (metricList.length() > 0) {
+                    metricList.append(",");
+                }
+                metricList.append(metric);
                 BatchScanner scanner = connector.createBatchScanner(metricsTable, getSessionAuthorizations(msg),
                         scannerThreads);
                 try {
@@ -484,10 +472,12 @@ public class DataStoreImpl implements DataStore {
                     long endOfLastPeriod = (endDistanceFromDownSample > 0 ? endTs + downsample
                             - endDistanceFromDownSample : endTs);
 
-                    setQueryRange(scanner, metric, startOfFirstPeriod, endOfLastPeriod);
-                    List<String> tagOrder = prioritizeTags(query);
+                    List<String> tagOrder = prioritizeTags(query.getMetric(), query.getTags());
                     Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
-                    setQueryColumns(scanner, metric, orderedTags);
+                    Set<Tag> colFamValues = getColumnFamilies(metric, orderedTags);
+                    List<Range> ranges = getQueryRanges(metric, startOfFirstPeriod, endOfLastPeriod, colFamValues);
+                    scanner.setRanges(ranges);
+                    setQueryColumns(scanner, metric, orderedTags, colFamValues);
 
                     if (query.isRate()) {
                         LOG.trace("Adding rate iterator");
@@ -506,7 +496,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);
                     }
 
@@ -527,6 +517,7 @@ public class DataStoreImpl implements DataStore {
                         // Aggregation
                         Map<Set<Tag>, Aggregation> samples = AggregationIterator.decodeValue(encoded.getValue());
                         for (Entry<Set<Tag>, Aggregation> entry : samples.entrySet()) {
+                            numResults++;
                             Set<Tag> key = new HashSet<>();
                             for (Tag tag : entry.getKey()) {
                                 if (query.getTags().keySet().contains(tag.getKey())) {
@@ -546,10 +537,13 @@ public class DataStoreImpl implements DataStore {
                 // TODO groupby here?
                 long tsDivisor = msg.isMsResolution() ? 1 : 1000;
                 for (Entry<Set<Tag>, List<Aggregation>> entry : allSeries.entrySet()) {
+                    numResults += entry.getValue().size();
                     result.add(convertToQueryResponse(query, entry.getKey(), entry.getValue(), tsDivisor));
                 }
             }
-            LOG.debug("Query time: {}", (System.currentTimeMillis() - now));
+            LOG.debug("Query time: {} duration: {} metrics: {} results: {}", (System.currentTimeMillis() - now),
+                    (endTs - startTs), metricList.toString(), numResults);
+            internalMetrics.addQueryResponse(result.size(), (System.currentTimeMillis() - now));
             return result;
         } catch (ClassNotFoundException | IOException | TableNotFoundException ex) {
             LOG.error("Error during query: " + ex.getMessage(), ex);
@@ -576,10 +570,9 @@ public class DataStoreImpl implements DataStore {
      * @param query
      * @return ordered list of most specific to least specific tags in the query
      */
-    private List<String> prioritizeTags(SubQuery query) {
+    private List<String> prioritizeTags(String metric, Map<String, String> tags) {
         // trivial cases
-        Map<String, String> tags = query.getTags();
-        if (tags.isEmpty()) {
+        if (tags == null || tags.isEmpty()) {
             return Collections.emptyList();
         }
         if (tags.size() == 1) {
@@ -587,7 +580,6 @@ public class DataStoreImpl implements DataStore {
         }
         // favor tags with fewer values
         Map<String, Integer> priority = new HashMap<>();
-        String metric = query.getMetric();
         // Count matching tags
         updateMetricCounts();
         for (Entry<String, String> entry : tags.entrySet()) {
@@ -643,8 +635,9 @@ public class DataStoreImpl implements DataStore {
         return !REGEX_TEST.matcher(value).matches();
     }
 
-    private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags)
-            throws TableNotFoundException, TimelyException {
+    public Set<Tag> getColumnFamilies(String metric, Map<String, String> requestedTags) throws TableNotFoundException {
+
+        Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
         LOG.trace("Looking for requested tags: {}", tags);
         Scanner meta = connector.createScanner(metaTable, Authorizations.EMPTY);
         Text start = new Text(Meta.VALUE_PREFIX + metric);
@@ -707,24 +700,35 @@ public class DataStoreImpl implements DataStore {
         // matches
         // the pattern of the first tag in the query. The resulting set of tags
         // will be used to fetch specific columns from the metric table.
-        Set<Tag> concrete = expandTagValues(first, knownKeyValues);
-        if (concrete.size() == 0) {
+        return expandTagValues(first, knownKeyValues);
+    }
+
+    private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags, Set<Tag> colFamValues)
+            throws TimelyException {
+
+        if (colFamValues.size() == 0) {
             throw new TimelyException(HttpResponseStatus.BAD_REQUEST.code(), "No matching tags", "No tags were found "
                     + " that matched the submitted tags. Please fix and retry");
         }
-        LOG.trace("Found matching tags: {}", concrete);
-        for (Tag tag : concrete) {
+        LOG.trace("Found matching tags: {}", colFamValues);
+        for (Tag tag : colFamValues) {
             Text colf = new Text(tag.getKey() + "=" + tag.getValue());
             scanner.fetchColumnFamily(colf);
             LOG.trace("Fetching metric table column family: {}", colf);
         }
         // Add the regular expression to filter the other tags
         int priority = 100;
+        Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
+        // skip over first tag which was already expanded into colFamValues
+        if (tagIter.hasNext()) {
+            tagIter.next();
+        }
+        Entry<String, String> tag = null;
         while (tagIter.hasNext()) {
-            Entry<String, String> tag = tagIter.next();
+            tag = tagIter.next();
             LOG.trace("Adding regex filter for tag {}", tag);
             StringBuffer pattern = new StringBuffer();
-            pattern.append("(^|.*,)");
+            pattern.append("(^.*\\x00|.*,)");
             pattern.append(tag.getKey());
             pattern.append("=");
             pattern.append(tag.getValue());
@@ -767,14 +771,55 @@ public class DataStoreImpl implements DataStore {
         return result;
     }
 
-    private void setQueryRange(BatchScanner scanner, String metric, long start, long end) {
-        final byte[] start_row = MetricAdapter.encodeRowKey(metric, start);
-        LOG.trace("Start key for metric {} and time {} is {}", metric, start, start_row);
-        final byte[] end_row = MetricAdapter.encodeRowKey(metric, end);
-        LOG.trace("End key for metric {} and time {} is {}", metric, end, end_row);
-        Range range = new Range(new Text(start_row), new Text(end_row));
-        LOG.trace("Set query range to {}", range);
-        scanner.setRanges(Collections.singletonList(range));
+    public List<Range> getQueryRanges(String metric, long start, long end, Set<Tag> colFamValues) {
+        List<Range> ranges = new ArrayList<>();
+        long beginRangeRounded = MetricAdapter.roundTimestampToLastHour(start);
+        if (colFamValues.isEmpty()) {
+            final byte[] start_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+            Key startKey = new Key(new Text(start_row));
+            LOG.trace("Start key for metric {} and time {} is {}", metric, beginRangeRounded, startKey.toStringNoTime());
+            final byte[] end_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+            Key endKey = new Key(new Text(end_row));
+            LOG.trace("End key for metric {} and time {} is {}", metric, MetricAdapter.roundTimestampToNextHour(end),
+                    endKey.toStringNoTime());
+            Range range = new Range(startKey, true, endKey, false);
+            LOG.trace("Set query range to {}", range);
+            ranges.add(range);
+        } else {
+            try {
+                long lastBeginRangeRounded = MetricAdapter.roundTimestampToLastHour(end);
+                while (beginRangeRounded <= lastBeginRangeRounded) {
+                    // use end timestamp of begin + 1 hour and one msec
+                    // except the last range where we use end + 1 msec
+                    long endRangeTimestamp = (beginRangeRounded == lastBeginRangeRounded) ? end + 1 : beginRangeRounded
+                            + (1000 * 60 * 60) + 1;
+                    long beginRangeTimestamp = (beginRangeRounded == MetricAdapter.roundTimestampToLastHour(start)) ? start
+                            : beginRangeRounded;
+                    for (Tag t : colFamValues) {
+                        final byte[] start_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+                        Key startKey = new Key(new Text(start_row), new Text(t.join()
+                                .getBytes(Charset.forName("UTF-8"))), new Text(MetricAdapter.encodeColQual(
+                                beginRangeTimestamp, "")), new Text(new byte[0]), beginRangeTimestamp);
+                        LOG.trace("Start key for metric {} and time {} is {}", metric, beginRangeTimestamp,
+                                startKey.toStringNoTime());
+                        final byte[] end_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+                        Key endKey = new Key(new Text(end_row), new Text(t.join().getBytes(Charset.forName("UTF-8"))),
+                                new Text(MetricAdapter.encodeColQual(endRangeTimestamp, "")), new Text(new byte[0]),
+                                endRangeTimestamp);
+                        LOG.trace("End key for metric {} and time {} is {}", metric, beginRangeTimestamp,
+                                endKey.toStringNoTime());
+                        Range range = new Range(startKey, true, endKey, false);
+                        LOG.trace("Set query range to {}", range);
+                        ranges.add(range);
+                    }
+                    beginRangeRounded += (1000 * 60 * 60); // add an hour
+                }
+            } catch (Exception e) {
+                LOG.error("Error creating query ranges", e);
+            }
+        }
+
+        return ranges;
     }
 
     private Class<? extends Aggregator> getAggregator(SubQuery query) {
@@ -835,7 +880,7 @@ public class DataStoreImpl implements DataStore {
         }
     }
 
-    private long getAgeOffForMetric(String metricName) {
+    public long getAgeOffForMetric(String metricName) {
         String age = this.ageOff.get(MetricAgeOffIterator.AGE_OFF_PREFIX + metricName);
         if (null == age) {
             return this.defaultAgeOff;
@@ -865,23 +910,13 @@ public class DataStoreImpl implements DataStore {
             if (null == metric) {
                 throw new IllegalArgumentException("metric name must be specified");
             }
-            if (0 == startTime) {
-                startTime = (System.currentTimeMillis() - getAgeOffForMetric(metric) - 1000);
-                LOG.debug("Overriding zero start time to {} due to age off configuration", startTime);
+            if (tags == null) {
+                tags = new LinkedHashMap<>();
             }
-            byte[] start = MetricAdapter.encodeRowKey(metric, startTime);
-            long endTimeStamp = (endTime == 0) ? (System.currentTimeMillis() - (lag * 1000)) : endTime;
-            byte[] end = MetricAdapter.encodeRowKey(metric, endTimeStamp);
-            s.setRange(new Range(new Text(start), true, new Text(end), false));
-            SubQuery query = new SubQuery();
-            query.setMetric(metric);
-            if (null == tags) {
-                tags = Collections.emptyMap();
-            }
-            query.setTags(tags);
-            List<String> tagOrder = prioritizeTags(query);
-            Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
-            setQueryColumns(s, metric, orderedTags);
+            List<String> tagOrder = prioritizeTags(metric, tags);
+            Map<String, String> orderedTags = orderTags(tagOrder, tags);
+            Set<Tag> colFamValues = getColumnFamilies(metric, orderedTags);
+            setQueryColumns(s, metric, orderedTags, colFamValues);
             s.setBatchSize(scannerBatchSize);
             s.setReadaheadThreshold(scannerReadAhead);
             return s;

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/main/java/timely/store/iterators/TimeSeriesGroupingIterator.java

@@ -231,7 +231,7 @@ public class TimeSeriesGroupingIterator extends WrappingIterator {
         while (super.hasTop()) {
             Key k = super.getTopKey();
 
-            Long newTime = MetricAdapter.decodeRowKey(k).getSecond();
+            Long newTime = k.getTimestamp();
             if (time == null) {
                 time = newTime;
             }

server/src/main/java/timely/subscription/MetricScanner.java

@@ -2,16 +2,20 @@ package timely.subscription;
 
 import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.netty.util.concurrent.ScheduledFuture;
 
 import java.lang.Thread.UncaughtExceptionHandler;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
@@ -25,6 +29,7 @@ import timely.api.response.MetricResponse;
 import timely.api.response.MetricResponses;
 import timely.api.response.TimelyException;
 import timely.model.Metric;
+import timely.model.Tag;
 import timely.store.DataStore;
 import timely.util.JsonUtil;
 
@@ -45,11 +50,16 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     private final int lag;
     private final String subscriptionId;
     private final String metric;
-    private final long endTime;
+    private long startTime;
+    private long endTime;
     private final Subscription subscription;
     private MetricResponses responses = new MetricResponses();
     private final ScheduledFuture<?> flusher;
     private final int subscriptionBatchSize;
+    private List<Range> ranges = null;
+    private Iterator<Range> rangeItr = null;
+    private DataStore store = null;
+    private Set<Tag> colFamValues = null;
 
     public MetricScanner(Subscription sub, String subscriptionId, String sessionId, DataStore store, String metric,
             Map<String, String> tags, long startTime, long endTime, long delay, int lag, ChannelHandlerContext ctx,
@@ -57,13 +67,34 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
             throws TimelyException {
         this.setDaemon(true);
         this.setUncaughtExceptionHandler(this);
+        this.store = store;
         this.subscription = sub;
         this.ctx = ctx;
         this.lag = lag;
         this.metric = metric;
+        this.startTime = startTime;
         this.endTime = endTime;
-        this.scanner = store.createScannerForMetric(sessionId, metric, tags, startTime, endTime, lag, scannerBatchSize,
-                scannerReadAhead);
+        this.scanner = this.store.createScannerForMetric(sessionId, metric, tags, startTime, endTime, lag,
+                scannerBatchSize, scannerReadAhead);
+
+        if (0 == startTime) {
+            this.startTime = (System.currentTimeMillis() - this.store.getAgeOffForMetric(metric) - 1000);
+            LOG.debug("Overriding zero start time to {} due to age off configuration", startTime);
+        }
+        long endTimeStamp = (endTime == 0) ? (System.currentTimeMillis() - (lag * 1000)) : endTime;
+
+        try {
+            this.colFamValues = this.store.getColumnFamilies(metric, tags);
+            ranges = this.store.getQueryRanges(metric, this.startTime, endTimeStamp, colFamValues);
+            rangeItr = ranges.iterator();
+            if (rangeItr.hasNext()) {
+                Range r = rangeItr.next();
+                this.scanner.setRange(r);
+            }
+        } catch (TableNotFoundException e) {
+            throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error in MetricScanner",
+                    e.getMessage(), e);
+        }
         this.iter = scanner.iterator();
         this.delay = delay;
         this.subscriptionId = subscriptionId;
@@ -86,6 +117,7 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     }
 
     public synchronized void flush() {
+        LOG.info("Flush called");
         synchronized (responses) {
             if (responses.size() > 0) {
                 try {
@@ -103,7 +135,8 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
     public void run() {
         Metric m = null;
         try {
-            while (!closed) {
+            boolean done = false;
+            while (!done && !closed) {
 
                 if (this.iter.hasNext()) {
                     Entry<Key, Value> e = this.iter.next();
@@ -112,39 +145,55 @@ public class MetricScanner extends Thread implements UncaughtExceptionHandler {
                         flush();
                     }
                     this.responses.addResponse(MetricResponse.fromMetric(m, this.subscriptionId));
+                } else if (rangeItr.hasNext()) {
+                    // set next range on the scanner
+                    Range r = rangeItr.next();
+                    this.scanner.close();
+                    this.scanner.setRange(r);
+                    this.iter = scanner.iterator();
                 } else if (this.endTime == 0) {
                     flush();
                     long endTimeStamp = (System.currentTimeMillis() - (lag * 1000));
-                    byte[] end = MetricAdapter.encodeRowKey(this.metric, endTimeStamp);
-                    Text endRow = new Text(end);
                     this.scanner.close();
-                    Range prevRange = this.scanner.getRange();
                     if (null == m) {
                         LOG.debug("No results found, waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
-                        this.scanner.setRange(new Range(prevRange.getStartKey().getRow(), prevRange
-                                .isStartKeyInclusive(), endRow, false));
-                        this.iter = this.scanner.iterator();
+                        ranges = this.store.getQueryRanges(metric, startTime, endTimeStamp, colFamValues);
+                        rangeItr = ranges.iterator();
+                        if (rangeItr.hasNext()) {
+                            Range r = rangeItr.next();
+                            this.scanner.setRange(r);
+                            this.iter = scanner.iterator();
+                        }
                     } else {
                         // Reset the starting range to the last key returned
                         LOG.debug("Exhausted scanner, last metric returned was {}", m);
                         LOG.debug("Waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
-                        this.scanner.setRange(new Range(new Text(MetricAdapter.encodeRowKey(m)), false, endRow,
-                                prevRange.isEndKeyInclusive()));
-                        this.iter = this.scanner.iterator();
+                        ranges = this.store.getQueryRanges(metric, m.getValue().getTimestamp() + 1, endTimeStamp,
+                                colFamValues);
+                        rangeItr = ranges.iterator();
+                        if (rangeItr.hasNext()) {
+                            Range r = rangeItr.next();
+                            this.scanner.setRange(r);
+                            this.iter = this.scanner.iterator();
+                        } else {
+                            done = true;
+                        }
                     }
                 } else {
+                    done = true;
+                }
+                if (done) {
                     LOG.debug("Exhausted scanner, sending completed message for subscription {}", this.subscriptionId);
                     final MetricResponse last = new MetricResponse();
                     last.setSubscriptionId(this.subscriptionId);
                     last.setComplete(true);
                     this.responses.addResponse(last);
                     flush();
-                    break;
                 }
             }
-        } catch (Exception e) {
+        } catch (Throwable e) {
             LOG.error("Error in metric scanner, closing.", e);
         } finally {
             close();

server/src/main/java/timely/util/TimelyMetricsFormatter.java

@@ -15,6 +15,8 @@ public class TimelyMetricsFormatter extends DefaultFormatter {
 
     private static final PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(),
             new LongLexicoder());
+    private static final PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(),
+            new StringLexicoder());
 
     @Override
     public String next() {
@@ -25,7 +27,8 @@ public class TimelyMetricsFormatter extends DefaultFormatter {
         b.append(" ");
         b.append(e.getKey().getColumnFamily().toString());
         b.append(":");
-        b.append(e.getKey().getColumnQualifier().toString());
+        ComparablePair<Long, String> cq = colQualCoder.decode(e.getKey().getColumnQualifier().copyBytes());
+        b.append(cq.getFirst()).append("\\x00").append(cq.getSecond());
         b.append(" ");
         b.append(e.getKey().getColumnVisibility().toString());
         if (super.isDoTimestamps()) {

server/src/test/java/timely/api/model/MetricAdapterTest.java

@@ -42,28 +42,33 @@ public class MetricAdapterTest {
         Mutation mut = MetricAdapter.toMutation(m);
 
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(ts)));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
         Assert.assertEquals(rowCoder.decode(row), rowCoder.decode(mut.getRow()));
         Assert.assertEquals(3, mut.getUpdates().size());
         ColumnUpdate up = mut.getUpdates().get(0);
-        Assert.assertTrue(new String(up.getColumnFamily()).equals("tag1=value1"));
-        Assert.assertTrue(new String(up.getColumnQualifier()).equals("tag2=value2,tag3=value3"));
+        Assert.assertEquals("tag1=value1", new String(up.getColumnFamily()));
+        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag2=value2,tag3=value3"))),
+                new String(up.getColumnQualifier()));
+
         Assert.assertEquals(ts, up.getTimestamp());
-        Assert.assertTrue(new String(up.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(new String(up.getColumnVisibility())));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up2 = mut.getUpdates().get(1);
-        Assert.assertTrue(new String(up2.getColumnFamily()).equals("tag2=value2"));
-        Assert.assertTrue(new String(up2.getColumnQualifier()).equals("tag1=value1,tag3=value3"));
+        Assert.assertEquals("tag2=value2", new String(up2.getColumnFamily()));
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag1=value1,tag3=value3"))),
+                new String(up2.getColumnQualifier()));
         Assert.assertEquals(ts, up2.getTimestamp());
-        Assert.assertTrue(new String(up2.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(up2.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up3 = mut.getUpdates().get(2);
-        Assert.assertTrue(new String(up3.getColumnFamily()).equals("tag3=value3"));
-        Assert.assertTrue(new String(up3.getColumnQualifier()).equals("tag1=value1,tag2=value2"));
+        Assert.assertEquals("tag3=value3", new String(up3.getColumnFamily()));
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag1=value1,tag2=value2"))),
+                new String(up3.getColumnQualifier()));
         Assert.assertEquals(ts, up3.getTimestamp());
-        Assert.assertTrue(new String(up3.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(up3.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
     }
 
@@ -78,14 +83,16 @@ public class MetricAdapterTest {
         Mutation mut = MetricAdapter.toMutation(m);
 
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(ts)));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
         Assert.assertEquals(rowCoder.decode(row), rowCoder.decode(mut.getRow()));
         Assert.assertEquals(1, mut.getUpdates().size());
         ColumnUpdate up = mut.getUpdates().get(0);
         Assert.assertEquals("tag1=value1", new String(up.getColumnFamily()));
-        Assert.assertEquals("", new String(up.getColumnQualifier()));
+        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, ""))),
+                new String(up.getColumnQualifier()));
         Assert.assertEquals(ts, up.getTimestamp());
         Assert.assertEquals("(a&b)|(c&d)", new String(up.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
@@ -110,8 +117,9 @@ public class MetricAdapterTest {
         byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
-        Key k = new Key(row, "tag1=value1".getBytes(), "tag2=value2,tag3=value3".getBytes(), "(a&b)|(c&d)".getBytes(),
-                1000);
+        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
+        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000),
+                "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), 1000);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v);
         Assert.assertEquals("sys.cpu.user", m.getName());
@@ -130,8 +138,9 @@ public class MetricAdapterTest {
         byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
-        Key k = new Key(row, "tag1=value1".getBytes(), "tag2=value2,tag3=value3".getBytes(), "(a&b)|(c&d)".getBytes(),
-                1000);
+        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
+        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000),
+                "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), 1000);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v, true);
         Assert.assertEquals("sys.cpu.user", m.getName());

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;
@@ -57,6 +54,7 @@ public class DownsampleIteratorTest {
             for (ColumnUpdate cu : mutation.getUpdates()) {
                 Key key = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(),
                         cu.getColumnVisibility(), cu.getTimestamp());
+                System.out.println(key.toString());
                 testData1.put(key, new Value(cu.getValue()));
             }
         }
@@ -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,142 @@ 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();
+        DownsampleMemoryEstimator memoryEstimator = new DownsampleMemoryEstimator(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();
+        DownsampleMemoryEstimator memoryEstimator = new DownsampleMemoryEstimator(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();
+            System.out.println(key.toString());
+        } while (iter.hasTop());
+
         assertEquals(testData.lastKey(), key);
-        Map<Set<Tag>, Downsample> samples = DownsampleIterator.decodeValue(iter.getTopValue());
         return samples;
     }
 

server/src/test/java/timely/store/iterators/RateIteratorTest.java

@@ -22,6 +22,7 @@ import timely.adapter.accumulo.MetricAdapter;
 import timely.api.request.timeseries.QueryRequest;
 import timely.model.Metric;
 import timely.model.Tag;
+import timely.store.DataStoreImpl;
 
 public class RateIteratorTest extends IteratorTestBase {
 
@@ -38,7 +39,8 @@ public class RateIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -68,7 +70,8 @@ public class RateIteratorTest extends IteratorTestBase {
             ts += 1000 + r.nextInt(100);
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -113,7 +116,8 @@ public class RateIteratorTest extends IteratorTestBase {
                 ts += 1000;
                 Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
                 byte[] row = MetricAdapter.encodeRowKey(m);
-                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                        ts, ""), new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }
@@ -147,7 +151,8 @@ public class RateIteratorTest extends IteratorTestBase {
                 ts += 1000;
                 Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
                 byte[] row = MetricAdapter.encodeRowKey(m);
-                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                        ts, ""), new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }
@@ -181,7 +186,8 @@ public class RateIteratorTest extends IteratorTestBase {
                 ts += 1000;
                 Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
                 byte[] row = MetricAdapter.encodeRowKey(m);
-                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+                Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                        ts, ""), new byte[0], ts);
                 Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
                 table.put(k, v);
             }

server/src/test/java/timely/store/iterators/TimeSeriesGroupingIteratorTest.java

@@ -6,10 +6,7 @@ import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.TreeMap;
+import java.util.*;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.Key;
@@ -43,7 +40,8 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
         }
@@ -92,12 +90,14 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             Metric m2 = new Metric("sys.cpu.user", ts, i * 2.0D, tags2);
             byte[] row2 = MetricAdapter.encodeRowKey(m2);
-            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                    ts, ""), new byte[0], ts);
             Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
             table.put(k2, v2);
         }
@@ -108,8 +108,12 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         iter.init(source, settings.getOptions(), SCAN_IE);
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
 
+        // this section changed when the key structure changed so that identical
+        // colFam values sorted consecutively within an given time period
         for (int i = 4; i < 100; i++) {
             checkNextResult(iter, new double[] { i - 4, i - 3, i - 2, i - 1, i });
+        }
+        for (int i = 4; i < 100; i++) {
             checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
         }
         assertFalse(iter.hasTop());
@@ -128,19 +132,21 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             if (i < 50) {
                 // only populate this series 50 times
                 Metric m2 = new Metric("sys.cpu.user", ts, i * 2.0D, tags2);
                 byte[] row2 = MetricAdapter.encodeRowKey(m2);
-                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0],
-                        ts);
+                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8),
+                        MetricAdapter.encodeColQual(ts, ""), new byte[0], ts);
                 Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
                 table.put(k2, v2);
             }
         }
+
         SortedMapIterator source = new SortedMapIterator(table);
         TimeSeriesGroupingIterator iter = new TimeSeriesGroupingIterator();
         IteratorSetting settings = new IteratorSetting(100, TimeSeriesGroupingIterator.class);
@@ -148,12 +154,15 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         iter.init(source, settings.getOptions(), SCAN_IE);
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
 
+        // this section changed when the key structure changed so that identical
+        // colFam values sorted consecutively within an given time period
         for (int i = 4; i < 100; i++) {
             System.out.println(i);
             checkNextResult(iter, new double[] { i - 4, i - 3, i - 2, i - 1, i });
-            if (i < 50) {
-                checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
-            }
+        }
+        for (int i = 4; i < 50; i++) {
+            System.out.println(i);
+            checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
         }
         assertFalse(iter.hasTop());
     }
@@ -170,15 +179,16 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             if (i > 50) {
                 // only populate this series 50 times
                 Metric m2 = new Metric("sys.cpu.user", ts, i * 2.0D, tags2);
                 byte[] row2 = MetricAdapter.encodeRowKey(m2);
-                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0],
-                        ts);
+                Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8),
+                        MetricAdapter.encodeColQual(ts, ""), new byte[0], ts);
                 Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
                 table.put(k2, v2);
             }
@@ -190,12 +200,15 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         iter.init(source, settings.getOptions(), SCAN_IE);
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
 
+        // this section changed when the key structure changed so that identical
+        // colFam values sorted consecutively within an given time period
         for (int i = 4; i < 100; i++) {
             checkNextResult(iter, new double[] { i - 4, i - 3, i - 2, i - 1, i });
-            if (i > 54) {
-                checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
-            }
         }
+        for (int i = 55; i < 100; i++) {
+            checkNextResult(iter, new double[] { (i - 4) * 2, (i - 3) * 2, (i - 2) * 2, (i - 1) * 2, i * 2 });
+        }
+
         assertFalse(iter.hasTop());
 
     }
@@ -214,22 +227,25 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
             ts += 1000;
             Metric m = new Metric("sys.cpu.user", ts, i * 1.0D, tags1);
             byte[] row = MetricAdapter.encodeRowKey(m);
-            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k = new Key(row, tags1.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(ts,
+                    ""), new byte[0], ts);
             Value v = new Value(MetricAdapter.encodeValue(m.getValue().getMeasure()));
             table.put(k, v);
             // jitter the time on the second time series
             Metric m2 = new Metric("sys.cpu.user", ts + 50, i * 2.0D, tags2);
             byte[] row2 = MetricAdapter.encodeRowKey(m2);
-            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0],
-                    ts + 50);
+            Key k2 = new Key(row2, tags2.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                    ts, ""), new byte[0], ts + 50);
             Value v2 = new Value(MetricAdapter.encodeValue(m2.getValue().getMeasure()));
             table.put(k2, v2);
             Metric m3 = new Metric("sys.cpu.user", ts, i * 3.0D, tags3);
             byte[] row3 = MetricAdapter.encodeRowKey(m3);
-            Key k3 = new Key(row3, tags3.get(0).join().getBytes(StandardCharsets.UTF_8), new byte[0], new byte[0], ts);
+            Key k3 = new Key(row3, tags3.get(0).join().getBytes(StandardCharsets.UTF_8), MetricAdapter.encodeColQual(
+                    ts, ""), new byte[0], ts);
             Value v3 = new Value(MetricAdapter.encodeValue(m3.getValue().getMeasure()));
             table.put(k3, v3);
         }
+
         SortedMapIterator source = new SortedMapIterator(table);
         TimeSeriesGroupingIterator iter = new TimeSeriesGroupingIterator();
         IteratorSetting settings = new IteratorSetting(100, TimeSeriesGroupingIterator.class);
@@ -256,23 +272,20 @@ public class TimeSeriesGroupingIteratorTest extends IteratorTestBase {
         third.add(9D);
         third.add(12D);
 
-        for (int i = 4; i < 196; i++) {
+        // this section changed when the key structure changed so that identical
+        // colFam values sorted consecutively within an given time period
+        for (int i = 4; i < 100; i++) {
+            checkNextResult(iter, first);
+            shiftAndAdd(first, 1);
+        }
+        for (int i = 4; i < 100; i++) {
             System.out.println(i);
-            if (i == 4) {
-                checkNextResult(iter, first);
-                checkNextResult(iter, third);
-            } else if (i == 5) {
-                checkNextResult(iter, second);
-            } else if (i % 2 == 0) {
-                shiftAndAdd(first, 1);
-                checkNextResult(iter, first);
-
-                shiftAndAdd(third, 3);
-                checkNextResult(iter, third);
-            } else if (i % 2 != 0) {
-                shiftAndAdd(second, 2);
-                checkNextResult(iter, second);
-            }
+            checkNextResult(iter, second);
+            shiftAndAdd(second, 2);
+        }
+        for (int i = 4; i < 100; i++) {
+            checkNextResult(iter, third);
+            shiftAndAdd(third, 3);
         }
         assertFalse(iter.hasTop());
     }

server/src/test/java/timely/test/integration/DataStoreIT.java

@@ -49,6 +49,7 @@ public class DataStoreIT extends OneWaySSLBase {
             subQuery.setDownsample(Optional.of("1s-max"));
             request.addQuery(subQuery);
             List<QueryResponse> response = query("https://127.0.0.1:54322/api/query", request);
+
             assertEquals(1, response.size());
             Map<String, String> tags = response.get(0).getTags();
             assertEquals(1, tags.size());

server/src/test/java/timely/test/integration/http/HttpApiIT.java

@@ -572,6 +572,11 @@ public class HttpApiIT extends OneWaySSLBase {
         final Server s = new Server(conf);
         s.run();
         try {
+            System.out.println("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1");
+            System.out.println("sys.cpu.user " + (TEST_TIME + 1) + " 1.0 tag3=value3 rack=r2");
+            System.out.println("sys.cpu.idle " + (TEST_TIME + 2) + " 1.0 tag3=value3 tag4=value4 rack=r1");
+            System.out.println("sys.cpu.idle " + (TEST_TIME + 1000) + " 3.0 tag3=value3 tag4=value4 rack=r2");
+
             put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2 rack=r1", "sys.cpu.user " + (TEST_TIME + 1)
                     + " 1.0 tag3=value3 rack=r2", "sys.cpu.idle " + (TEST_TIME + 2)
                     + " 1.0 tag3=value3 tag4=value4 rack=r1", "sys.cpu.idle " + (TEST_TIME + 1000)
@@ -590,6 +595,11 @@ public class HttpApiIT extends OneWaySSLBase {
             subQuery.setDownsample(Optional.of("1s-max"));
             request.addQuery(subQuery);
             List<QueryResponse> response = query("https://127.0.0.1:54322/api/query", request);
+
+            for (QueryResponse r : response) {
+                System.out.println(r.toString());
+            }
+
             assertEquals(2, response.size());
             QueryResponse response1 = response.get(0);
             Map<String, String> tags = response1.getTags();

server/src/test/java/timely/test/integration/websocket/WebSocketIT.java

@@ -346,6 +346,15 @@ public class WebSocketIT extends OneWaySSLBase {
             second = Metric.newBuilder().name("sys.cpu.user").value(TEST_TIME + 500, 7.0D).tag(new Tag("rack", "r2"))
                     .tag(new Tag("tag3", "value3")).build();
 
+            int x = 1;
+            for (String r : response) {
+                responses = JsonUtil.getObjectMapper().readValue(r, MetricResponses.class);
+                for (MetricResponse r1 : responses.getResponses()) {
+                    System.out.println(x + " -- " + r1.toString());
+                }
+                x++;
+            }
+
             Assert.assertEquals(1, response.size());
             responses = JsonUtil.getObjectMapper().readValue(response.get(0), MetricResponses.class);
             Assert.assertEquals(2, responses.size());