Make WrappedGorillaCompressor Serializable

.gitignore

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

pom.xml

@@ -134,6 +134,11 @@
                 <artifactId>commons-beanutils</artifactId>
                 <version>LATEST</version>
             </dependency>
+            <dependency>
+                <groupId>fi.iki.yak</groupId>
+                <artifactId>compression-gorilla</artifactId>
+                <version>2.0.4-SNAPSHOT</version>
+            </dependency>
             <dependency>
 	        <groupId>io.github.lukehutch</groupId>
 	        <artifactId>fast-classpath-scanner</artifactId>

server/conf/timely.yml

@@ -3,6 +3,10 @@ timely:
   meta-table: timely.meta
   metric-age-off-days:
     default: 7
+  cache:
+    enabled:
+    age-off-hours:
+      default: 30
   metrics-report-ignored-tags:
   accumulo:
     instance-name:

server/pom.xml

@@ -54,6 +54,10 @@
             <artifactId>disruptor</artifactId>
             <scope>runtime</scope>
         </dependency>
+        <dependency>
+            <groupId>fi.iki.yak</groupId>
+            <artifactId>compression-gorilla</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.github.lukehutch</groupId>
             <artifactId>fast-classpath-scanner</artifactId>

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

@@ -4,6 +4,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import javax.validation.Valid;
@@ -15,6 +16,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 import org.springframework.stereotype.Component;
 
+import timely.store.cache.DataStoreCache;
 import timely.validator.NotEmptyIfFieldSet;
 
 import com.google.common.collect.Lists;
@@ -45,6 +47,9 @@ public class Configuration {
     private MetaCache metaCache = new MetaCache();
     @Valid
     @NestedConfigurationProperty
+    private Cache cache = new Cache();
+    @Valid
+    @NestedConfigurationProperty
     private VisibilityCache visibilityCache = new VisibilityCache();
     @Valid
     @NestedConfigurationProperty
@@ -109,6 +114,10 @@ public class Configuration {
         return metaCache;
     }
 
+    public Cache getCache() {
+        return cache;
+    }
+
     public VisibilityCache getVisibilityCache() {
         return visibilityCache;
     }
@@ -652,6 +661,34 @@ public class Configuration {
         }
     }
 
+    public class Cache {
+
+        private boolean enabled = false;
+        private HashMap<String, Integer> metricAgeOffHours = new HashMap<>();
+
+        public HashMap<String, Integer> getMetricAgeOffHours() {
+            return metricAgeOffHours;
+        }
+
+        public Configuration setMetricAgeOffHours(HashMap<String, Integer> metricAgeOffHours) {
+            this.metricAgeOffHours = metricAgeOffHours;
+            return Configuration.this;
+        }
+
+        public void setDefaultAgeOffHours(int defaultAgeOffHours) {
+            this.metricAgeOffHours.put(DataStoreCache.DEFAULT_AGEOFF_KEY, defaultAgeOffHours);
+        }
+
+        public boolean isEnabled() {
+            return enabled;
+        }
+
+        public Configuration setEnabled(boolean enabled) {
+            this.enabled = enabled;
+            return Configuration.this;
+        }
+    }
+
     public class VisibilityCache {
 
         private long expirationMinutes = 60;

server/src/main/java/timely/Server.java

@@ -60,6 +60,7 @@ import org.springframework.context.ConfigurableApplicationContext;
 import timely.api.response.TimelyException;
 import timely.auth.AuthCache;
 import timely.auth.VisibilityCache;
+import timely.netty.http.HttpCacheRequestHandler;
 import timely.netty.http.HttpMetricPutHandler;
 import timely.netty.http.HttpStaticFileServerHandler;
 import timely.netty.http.HttpVersionRequestHandler;
@@ -96,6 +97,7 @@ import timely.netty.websocket.timeseries.WSSuggestRequestHandler;
 import timely.store.DataStore;
 import timely.store.DataStoreFactory;
 import timely.store.MetaCacheFactory;
+import timely.store.cache.DataStoreCache;
 
 public class Server {
 
@@ -124,6 +126,7 @@ public class Server {
     protected Channel wsChannelHandle = null;
     protected Channel udpChannelHandle = null;
     protected DataStore dataStore = null;
+    protected DataStoreCache dataStoreCache = null;
     protected volatile boolean shutdown = false;
 
     private static boolean useEpoll() {
@@ -322,6 +325,10 @@ public class Server {
         int nettyThreads = Math.max(1,
                 SystemPropertyUtil.getInt("io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
         dataStore = DataStoreFactory.create(config, nettyThreads);
+        if (config.getCache().isEnabled()) {
+            dataStoreCache = new DataStoreCache(config);
+            dataStore.setCache(dataStoreCache);
+        }
         // Initialize the MetaCache
         MetaCacheFactory.getCache(config);
         // initialize the auth cache
@@ -509,6 +516,7 @@ public class Server {
                 ch.pipeline().addLast("search", new HttpSearchLookupRequestHandler(dataStore));
                 ch.pipeline().addLast("suggest", new HttpSuggestRequestHandler(dataStore));
                 ch.pipeline().addLast("version", new HttpVersionRequestHandler());
+                ch.pipeline().addLast("cache", new HttpCacheRequestHandler(dataStoreCache));
                 ch.pipeline().addLast("put", new HttpMetricPutHandler(dataStore));
                 ch.pipeline().addLast("error", new TimelyExceptionHandler());
             }

server/src/main/java/timely/adapter/accumulo/MetricAdapter.java

@@ -1,5 +1,6 @@
 package timely.adapter.accumulo;
 
+import com.google.common.base.Joiner;
 import org.apache.accumulo.core.client.lexicoder.LongLexicoder;
 import org.apache.accumulo.core.client.lexicoder.PairLexicoder;
 import org.apache.accumulo.core.client.lexicoder.StringLexicoder;
@@ -9,14 +10,17 @@ 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.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
@@ -28,11 +32,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 final Joiner equalsJoiner = Joiner.on("=");
+
+    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,12 +67,36 @@ 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;
     }
 
+    public static Key toKey(String metric, Map<String, String> tags, long timestamp) {
+        byte[] row = encodeRowKey(metric, timestamp);
+
+        StringBuilder colQualSb = new StringBuilder();
+        String cf = null;
+        for (Map.Entry<String, String> entry : tags.entrySet()) {
+            if (entry.getKey().equals(VISIBILITY_TAG))
+                continue;
+
+            if (cf == null) {
+                cf = equalsJoiner.join(entry.getKey(), entry.getValue());
+                continue;
+            }
+            if (colQualSb.length() > 0) {
+                colQualSb.append(",");
+            }
+            colQualSb.append(equalsJoiner.join(entry.getKey(), entry.getValue()));
+        }
+        byte[] cqBytes = encodeColQual(timestamp, colQualSb.toString());
+        ColumnVisibility colVis = extractVisibility(tags);
+        return new Key(new Text(row), new Text(cf), new Text(cqBytes), colVis, timestamp);
+    }
+
     private static Value extractValue(Metric metric) {
         return new Value(encodeValue(metric.getValue().getMeasure()));
     }
@@ -73,7 +113,7 @@ public class MetricAdapter {
         return bb.getDouble();
     }
 
-    private static ColumnVisibility extractVisibility(List<Tag> tags) {
+    public static ColumnVisibility extractVisibility(List<Tag> tags) {
         // @formatter:off
         Optional<Tag> visTag = tags.stream()
                 .filter(t -> t.getKey().equals(VISIBILITY_TAG))
@@ -82,15 +122,24 @@ public class MetricAdapter {
         // @formatter:on
     }
 
+    public static ColumnVisibility extractVisibility(Map<String, String> tags) {
+        if (tags.containsKey(VISIBILITY_TAG)) {
+            return new ColumnVisibility(tags.get(VISIBILITY_TAG));
+        } else {
+            return EMPTY_VISIBILITY;
+        }
+    }
+
     public static Metric parse(Key k, Value v, boolean includeVizTag) {
         ComparablePair<String, Long> row = rowCoder.decode(k.getRow().getBytes());
         // @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 +155,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/CacheRequest.java

@@ -0,0 +1,20 @@
+package timely.api.request;
+
+import io.netty.handler.codec.http.QueryStringDecoder;
+import timely.api.annotation.Http;
+import timely.api.annotation.WebSocket;
+
+@Http(path = "/api/cache")
+@WebSocket(operation = "cache")
+public class CacheRequest extends AuthenticatedRequest implements HttpGetRequest, HttpPostRequest, WebSocketRequest {
+
+    @Override
+    public HttpGetRequest parseQueryParameters(QueryStringDecoder decoder) throws Exception {
+        return new CacheRequest();
+    }
+
+    @Override
+    public HttpPostRequest parseBody(String content) throws Exception {
+        return new CacheRequest();
+    }
+}

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

@@ -54,7 +54,12 @@ 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);
+            throw e;
+        }
     }
 
     @Override

server/src/main/java/timely/api/response/CacheResponse.java

@@ -0,0 +1,46 @@
+package timely.api.response;
+
+import org.apache.commons.lang3.builder.ToStringBuilder;
+
+import java.util.List;
+
+public class CacheResponse {
+
+    private Long oldestTimestamp;
+    private Long newestTimestamp;
+    private List<String> metrics;
+
+    public Long getOldestTimestamp() {
+        return oldestTimestamp;
+    }
+
+    public void setOldestTimestamp(Long oldestTimestamp) {
+        this.oldestTimestamp = oldestTimestamp;
+    }
+
+    public Long getNewestTimestamp() {
+        return newestTimestamp;
+    }
+
+    public void setNewestTimestamp(Long newestTimestamp) {
+        this.newestTimestamp = newestTimestamp;
+    }
+
+    public List<String> getMetrics() {
+        return metrics;
+    }
+
+    public void setMetrics(List<String> metrics) {
+        this.metrics = metrics;
+    }
+
+    @Override
+    public String toString() {
+        ToStringBuilder tsb = new ToStringBuilder(this);
+        tsb.append("oldestTimestamp", this.oldestTimestamp);
+        tsb.append("newestTimestamp", this.newestTimestamp);
+        tsb.append("metrics", this.metrics);
+        return tsb.toString();
+    }
+
+}

server/src/main/java/timely/api/response/timeseries/QueryResponse.java

@@ -1,10 +1,6 @@
 package timely.api.response.timeseries;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeMap;
+import java.util.*;
 
 import org.apache.commons.lang3.builder.ToStringBuilder;
 
@@ -13,7 +9,7 @@ public class QueryResponse {
     private String metric;
     private Map<String, String> tags = new HashMap<>();
     private List<String> aggregatedTags = new ArrayList<>();
-    private Map<String, Object> dps = new TreeMap<>();
+    private Map<String, Object> dps = new LinkedHashMap<>();
 
     public String getMetric() {
         return metric;

server/src/main/java/timely/netty/http/HttpCacheRequestHandler.java

@@ -0,0 +1,43 @@
+package timely.netty.http;
+
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaders.Names;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import timely.api.request.CacheRequest;
+import timely.api.response.TimelyException;
+import timely.netty.Constants;
+import timely.store.cache.DataStoreCache;
+import timely.util.JsonUtil;
+
+public class HttpCacheRequestHandler extends SimpleChannelInboundHandler<CacheRequest> implements TimelyHttpHandler {
+
+    private DataStoreCache cache = null;
+
+    public HttpCacheRequestHandler(DataStoreCache cache) {
+        this.cache = cache;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, CacheRequest v) throws Exception {
+        byte[] buf;
+        try {
+            buf = JsonUtil.getObjectMapper().writeValueAsBytes(cache.getCacheStatus());
+        } catch (Exception e) {
+            LOG.error(e.getMessage(), e);
+            this.sendHttpError(ctx, new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                    "Error getting cache status", "Error getting cache status", e));
+            return;
+        }
+        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
+                Unpooled.copiedBuffer(buf));
+        response.headers().set(Names.CONTENT_TYPE, Constants.JSON_TYPE);
+        response.headers().set(Names.CONTENT_LENGTH, response.content().readableBytes());
+        sendResponse(ctx, response);
+    }
+
+}

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

@@ -5,7 +5,11 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
-import java.util.*;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.data.Key;
@@ -14,8 +18,11 @@ import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.iterators.WrappingIterator;
 
+import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import timely.adapter.accumulo.MetricAdapter;
+import timely.api.request.timeseries.QueryRequest;
 import timely.model.Metric;
 import timely.model.ObjectSizeOf;
 import timely.model.Tag;
@@ -23,22 +30,29 @@ import timely.api.response.TimelyException;
 import timely.sample.Aggregator;
 import timely.sample.Downsample;
 import timely.sample.DownsampleFactory;
+import timely.sample.aggregators.Avg;
+
+import static org.apache.accumulo.core.conf.AccumuloConfiguration.getTimeInMillis;
 
 public class DownsampleIterator extends WrappingIterator {
 
-    private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(DownsampleIterator.class);
+    private static final 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 static final long DEFAULT_DOWNSAMPLE_MS = 1;
+    private static final String DEFAULT_DOWNSAMPLE_AGGREGATOR = Avg.class.getSimpleName().toLowerCase();
+
     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")
@@ -75,6 +89,9 @@ public class DownsampleIterator extends WrappingIterator {
             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
@@ -148,4 +165,30 @@ public class DownsampleIterator extends WrappingIterator {
         ObjectInputStream ois = new ObjectInputStream(bis);
         return (Map<Set<Tag>, Downsample>) ois.readObject();
     }
+
+    public static long getDownsamplePeriod(QueryRequest.SubQuery query) {
+        // disabling the downsampling OR setting the aggregation to none are
+        // both considered to be disabling
+        if (!query.getDownsample().isPresent() || query.getDownsample().get().endsWith("-none")) {
+            return DEFAULT_DOWNSAMPLE_MS;
+        }
+        String parts[] = query.getDownsample().get().split("-");
+        return getTimeInMillis(parts[0]);
+    }
+
+    public static Class<? extends Aggregator> getDownsampleAggregator(QueryRequest.SubQuery query) {
+        String aggregatorName = Aggregator.NONE;
+        if (query.getDownsample().isPresent()) {
+            String parts[] = query.getDownsample().get().split("-");
+            aggregatorName = parts[1];
+        }
+        // disabling the downsampling OR setting the aggregation to none are
+        // both considered to be disabling
+        if (aggregatorName.equals(Aggregator.NONE)) {
+            // we need a downsampling iterator, so default to max to ensure we
+            // return something
+            aggregatorName = DEFAULT_DOWNSAMPLE_AGGREGATOR;
+        }
+        return Aggregator.getAggregator(aggregatorName);
+    }
 }

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;
@@ -14,6 +17,10 @@ import timely.api.response.timeseries.QueryResponse;
 import timely.api.response.timeseries.SearchLookupResponse;
 import timely.api.response.timeseries.SuggestResponse;
 
+import timely.model.Tag;
+
+import timely.store.cache.DataStoreCache;
+
 public interface DataStore {
 
     void store(Metric metric) throws TimelyException;
@@ -26,7 +33,16 @@ 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;
 
+    void setCache(DataStoreCache cache);
+
+    DataStoreCache getCache();
 }

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

@@ -5,6 +5,7 @@ import static org.apache.accumulo.core.conf.AccumuloConfiguration.getTimeInMilli
 import io.netty.handler.codec.http.HttpResponseStatus;
 
 import java.io.IOException;
+import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -45,11 +46,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;
@@ -81,10 +78,10 @@ import timely.model.Tag;
 import timely.sample.Aggregation;
 import timely.sample.Aggregator;
 import timely.sample.Sample;
-import timely.sample.aggregators.Avg;
 import timely.sample.iterators.AggregationIterator;
 import timely.sample.iterators.DownsampleIterator;
 import timely.store.iterators.RateIterator;
+import timely.store.cache.DataStoreCache;
 import timely.util.MetaKeySet;
 
 public class DataStoreImpl implements DataStore {
@@ -92,8 +89,6 @@ public class DataStoreImpl implements DataStore {
     private static final Logger LOG = LoggerFactory.getLogger(DataStoreImpl.class);
 
     private static final long METRICS_PERIOD = 30000;
-    private static final long DEFAULT_DOWNSAMPLE_MS = 1;
-    private static final String DEFAULT_DOWNSAMPLE_AGGREGATOR = Avg.class.getSimpleName().toLowerCase();
     private static final Pattern REGEX_TEST = Pattern.compile("^\\w+$");
 
     /*
@@ -132,7 +127,8 @@ public class DataStoreImpl implements DataStore {
     private final ThreadLocal<BatchWriter> batchWriter = new ThreadLocal<>();
     private boolean anonAccessAllowed = false;
     private final Map<String, String> ageOff;
-    private final long defaultAgeOff;
+    private final long defaultAgeOffMilliSec;
+    private DataStoreCache cache = null;
 
     public DataStoreImpl(Configuration conf, int numWriteThreads) throws TimelyException {
 
@@ -167,7 +163,7 @@ public class DataStoreImpl implements DataStore {
                 }
             }
             ageOff = getAgeOff(conf);
-            defaultAgeOff = this.getAgeOffForMetric(MetricAgeOffIterator.DEFAULT_AGEOFF_KEY);
+            defaultAgeOffMilliSec = this.getAgeOffForMetric(MetricAgeOffIterator.DEFAULT_AGEOFF_KEY);
 
             final Map<String, String> tableIdMap = connector.tableOperations().tableIdMap();
             if (!tableIdMap.containsKey(metricsTable)) {
@@ -201,6 +197,7 @@ public class DataStoreImpl implements DataStore {
                 }
 
             }, METRICS_PERIOD, METRICS_PERIOD);
+
             this.metaCache = MetaCacheFactory.getCache(conf);
         } catch (Exception e) {
             throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error creating DataStoreImpl",
@@ -208,6 +205,16 @@ public class DataStoreImpl implements DataStore {
         }
     }
 
+    @Override
+    public void setCache(DataStoreCache cache) {
+        this.cache = cache;
+    }
+
+    @Override
+    public DataStoreCache getCache() {
+        return cache;
+    }
+
     private static final EnumSet<IteratorScope> AGEOFF_SCOPES = EnumSet.allOf(IteratorScope.class);
 
     private void removeAgeOffIterators(Connector con, String tableName) throws Exception {
@@ -242,6 +249,11 @@ public class DataStoreImpl implements DataStore {
     @Override
     public void store(Metric metric) {
         LOG.trace("Received Store Request for: {}", metric);
+
+        if (cache != null) {
+            cache.store(metric);
+        }
+
         if (null == metaWriter.get()) {
             try {
                 BatchWriter w = connector.createBatchWriter(metaTable, bwConfig);
@@ -314,7 +326,6 @@ public class DataStoreImpl implements DataStore {
             metaCache.addAll(toCache);
         }
         try {
-
             batchWriter.get().addMutation(MetricAdapter.toMutation(metric));
             internalMetrics.incrementMetricKeysInserted(metric.getTags().size());
         } catch (MutationsRejectedException e) {
@@ -466,93 +477,131 @@ 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();
+        long requestedStartTs = msg.getStart();
+        long requestedEndTs = 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();
-                BatchScanner scanner = connector.createBatchScanner(metricsTable, getSessionAuthorizations(msg),
-                        scannerThreads);
-                try {
+                if (metricList.length() > 0) {
+                    metricList.append(",");
+                }
+                metricList.append(metric);
+                Map<Set<Tag>, List<Aggregation>> cachedMetrics = new HashMap<>();
+                long oldestTimestampFromCache = 0;
+
+                if (cache != null) {
+                    long oldestCacheTimestamp = cache.getOldestTimestamp(query.getMetric());
+                    if (requestedEndTs >= oldestCacheTimestamp) {
+                        cachedMetrics = cache.subquery(msg, query);
+                        allSeries.putAll(cachedMetrics);
+                    }
+                    oldestTimestampFromCache = Math.max(oldestCacheTimestamp,
+                            System.currentTimeMillis() - cache.getAgeOffForMetric(query.getMetric()) + 1);
+                }
+
+                if (cachedMetrics.isEmpty() || requestedStartTs < oldestTimestampFromCache) {
+                    // we have already searched from oldestTimestampFromCache to
+                    // requestedEndTs
+                    long endTs = (oldestTimestampFromCache == 0) ? requestedEndTs : oldestTimestampFromCache - 1;
 
                     // Reset the start timestamp for the query to the
                     // beginning of the downsample period based on the epoch
-                    long downsample = getDownsamplePeriod(query);
+                    long downsample = DownsampleIterator.getDownsamplePeriod(query);
                     LOG.trace("Downsample period {}", downsample);
-                    long startOfFirstPeriod = startTs - (startTs % downsample);
+                    long startOfFirstPeriod = requestedStartTs - (requestedStartTs % downsample);
                     long endDistanceFromDownSample = endTs % downsample;
                     long endOfLastPeriod = (endDistanceFromDownSample > 0 ? endTs + downsample
                             - endDistanceFromDownSample : endTs);
 
-                    setQueryRange(scanner, metric, startOfFirstPeriod, endOfLastPeriod);
-                    List<String> tagOrder = prioritizeTags(query);
-                    Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
-                    setQueryColumns(scanner, metric, orderedTags);
+                    if (endOfLastPeriod > startOfFirstPeriod) {
+                        BatchScanner scanner = null;
+                        try {
+                            scanner = connector.createBatchScanner(metricsTable, getSessionAuthorizations(msg),
+                                    scannerThreads);
+                            List<String> tagOrder = prioritizeTags(query.getMetric(), query.getTags());
+                            Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
+                            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");
-                        IteratorSetting rate = new IteratorSetting(499, RateIterator.class);
-                        RateIterator.setRateOptions(rate, query.getRateOptions());
-                        scanner.addScanIterator(rate);
-                    }
+                            if (query.isRate()) {
+                                LOG.trace("Adding rate iterator");
+                                IteratorSetting rate = new IteratorSetting(499, RateIterator.class);
+                                RateIterator.setRateOptions(rate, query.getRateOptions());
+                                scanner.addScanIterator(rate);
+                            }
 
-                    Class<? extends Aggregator> daggClass = getDownsampleAggregator(query);
-                    if (daggClass == null) {
-                        // we should always have a downsample iterator in the
-                        // stack.
-                        throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
-                                "Error during query: programming error", "daggClass == null");
-                    } else {
-                        LOG.trace("Downsample Aggregator type {}", daggClass.getSimpleName());
-                        IteratorSetting is = new IteratorSetting(500, DownsampleIterator.class);
-                        DownsampleIterator.setDownsampleOptions(is, startOfFirstPeriod, endOfLastPeriod, downsample,
-                                maxDownsampleMemory, daggClass.getName());
-                        scanner.addScanIterator(is);
-                    }
+                            Class<? extends Aggregator> daggClass = DownsampleIterator.getDownsampleAggregator(query);
+                            if (daggClass == null) {
+                                // we should always have a downsample iterator
+                                // in the stack.
+                                throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                                        "Error during query: programming error", "daggClass == null");
+                            } else {
+                                LOG.trace("Downsample Aggregator type {}", daggClass.getSimpleName());
+                                IteratorSetting is = new IteratorSetting(500, DownsampleIterator.class);
+                                DownsampleIterator.setDownsampleOptions(is, startOfFirstPeriod, endOfLastPeriod,
+                                        downsample, maxDownsampleMemory, daggClass.getName());
+                                scanner.addScanIterator(is);
+                            }
 
-                    Class<? extends Aggregator> aggClass = getAggregator(query);
-                    // the aggregation iterator is optional
-                    if (aggClass != null) {
-                        LOG.trace("Aggregator type {}", aggClass.getSimpleName());
-                        IteratorSetting is = new IteratorSetting(501, AggregationIterator.class);
-                        AggregationIterator.setAggregationOptions(is, query.getTags(), aggClass.getName());
-                        scanner.addScanIterator(is);
-                    }
+                            Class<? extends Aggregator> aggClass = getAggregator(query);
+                            // the aggregation iterator is optional
+                            if (aggClass != null) {
+                                LOG.trace("Aggregator type {}", aggClass.getSimpleName());
+                                IteratorSetting is = new IteratorSetting(501, AggregationIterator.class);
+                                AggregationIterator.setAggregationOptions(is, query.getTags(), aggClass.getName());
+                                scanner.addScanIterator(is);
+                            }
 
-                    // tag -> array of results by period starting at start
-                    for (Entry<Key, Value> encoded : scanner) {
-                        // we can decode the value as a Map<Set<Tag>,
-                        // Aggregation> even if an AggregationIterator
-                        // is not used because Downsample is a subclass of
-                        // Aggregation
-                        Map<Set<Tag>, Aggregation> samples = AggregationIterator.decodeValue(encoded.getValue());
-                        for (Entry<Set<Tag>, Aggregation> entry : samples.entrySet()) {
-                            Set<Tag> key = new HashSet<>();
-                            for (Tag tag : entry.getKey()) {
-                                if (query.getTags().keySet().contains(tag.getKey())) {
-                                    key.add(tag);
+                            // tag -> array of results by period starting at
+                            // start
+                            for (Entry<Key, Value> encoded : scanner) {
+                                // we can decode the value as a Map<Set<Tag>,
+                                // Aggregation> even if an AggregationIterator
+                                // is not used because Downsample is a subclass
+                                // of Aggregation
+                                Map<Set<Tag>, Aggregation> samples = AggregationIterator
+                                        .decodeValue(encoded.getValue());
+                                for (Entry<Set<Tag>, Aggregation> entry : samples.entrySet()) {
+                                    Set<Tag> key = new HashSet<>();
+                                    for (Tag tag : entry.getKey()) {
+                                        if (query.getTags().keySet().contains(tag.getKey())) {
+                                            key.add(tag);
+                                        }
+                                    }
+                                    List<Aggregation> aggregations = allSeries.getOrDefault(key, new ArrayList<>());
+                                    aggregations.add(entry.getValue());
+                                    allSeries.put(key, aggregations);
                                 }
                             }
-                            List<Aggregation> aggregations = allSeries.getOrDefault(key, new ArrayList<>());
-                            aggregations.add(entry.getValue());
-                            allSeries.put(key, aggregations);
+                            LOG.trace("allSeries: {}", allSeries);
+                        } finally {
+                            if (scanner != null) {
+                                scanner.close();
+                            }
                         }
                     }
-                    LOG.trace("allSeries: {}", allSeries);
-                } finally {
-                    scanner.close();
                 }
 
                 // 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),
+                    (requestedEndTs - requestedStartTs), metricList.toString(), numResults);
             internalMetrics.addQueryResponse(result.size(), (System.currentTimeMillis() - now));
             return result;
         } catch (ClassNotFoundException | IOException | TableNotFoundException ex) {
@@ -580,10 +629,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) {
@@ -591,7 +639,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()) {
@@ -647,8 +694,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);
@@ -711,24 +759,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());
@@ -771,46 +830,61 @@ 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) {
         return Aggregator.getAggregator(query.getAggregator());
     }
 
-    private Class<? extends Aggregator> getDownsampleAggregator(SubQuery query) {
-        String aggregatorName = Aggregator.NONE;
-        if (query.getDownsample().isPresent()) {
-            String parts[] = query.getDownsample().get().split("-");
-            aggregatorName = parts[1];
-        }
-        // disabling the downsampling OR setting the aggregation to none are
-        // both considered to be disabling
-        if (aggregatorName.equals(Aggregator.NONE)) {
-            // we need a downsampling iterator, so default to max to ensure we
-            // return something
-            aggregatorName = DEFAULT_DOWNSAMPLE_AGGREGATOR;
-        }
-        return Aggregator.getAggregator(aggregatorName);
-    }
-
-    private long getDownsamplePeriod(SubQuery query) {
-        // disabling the downsampling OR setting the aggregation to none are
-        // both considered to be disabling
-        if (!query.getDownsample().isPresent() || query.getDownsample().get().endsWith("-none")) {
-            return DEFAULT_DOWNSAMPLE_MS;
-        }
-        String parts[] = query.getDownsample().get().split("-");
-        return getTimeInMillis(parts[0]);
-    }
-
     private Authorizations getSessionAuthorizations(AuthenticatedRequest request) {
         return getSessionAuthorizations(request.getSessionId());
     }
@@ -839,10 +913,10 @@ 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;
+            return this.defaultAgeOffMilliSec;
         } else {
             return Long.parseLong(age);
         }
@@ -869,23 +943,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/cache/DataStoreCache.java

@@ -0,0 +1,401 @@
+package timely.store.cache;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.commons.collections.map.LRUMap;
+import org.apache.commons.lang.time.DateUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.Configuration;
+import timely.api.request.AuthenticatedRequest;
+import timely.api.request.timeseries.QueryRequest;
+import timely.api.response.CacheResponse;
+import timely.api.response.TimelyException;
+import timely.api.response.timeseries.QueryResponse;
+import timely.auth.AuthCache;
+import timely.model.Metric;
+import timely.model.Tag;
+import timely.sample.Aggregation;
+import timely.sample.Aggregator;
+import timely.sample.Sample;
+import timely.sample.iterators.AggregationIterator;
+import timely.sample.iterators.DownsampleIterator;
+import timely.store.MetricAgeOffIterator;
+import timely.store.iterators.RateIterator;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+
+public class DataStoreCache {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DataStoreCache.class);
+    public static final String DEFAULT_AGEOFF_KEY = "default";
+
+    private Map<String, Map<TaggedMetric, GorillaStore>> gorillaMap = new HashMap<>();
+    private boolean anonAccessAllowed;
+    private Map<String, Long> minimumAgeOff;
+    private Map<String, String> minimumAgeOffForIterator;
+
+    private Timer maintenanceTimer = new Timer();
+
+    public DataStoreCache(Configuration conf) {
+        anonAccessAllowed = conf.getSecurity().isAllowAnonymousAccess();
+        Map<String, Integer> cacheAgeOff = conf.getCache().getMetricAgeOffHours();
+        Map<String, Integer> accumuloAgeOff = conf.getMetricAgeOffDays();
+        minimumAgeOff = getMinimumAgeOffs(accumuloAgeOff, cacheAgeOff);
+        minimumAgeOffForIterator = getAgeOffForIterator(minimumAgeOff);
+        Date firstExecution = DateUtils.truncate(new Date(), Calendar.HOUR);
+        firstExecution = DateUtils.addHours(firstExecution, 1);
+        maintenanceTimer.schedule(new TimerTask() {
+
+            @Override
+            public void run() {
+                ageOffGorillaStores();
+                archiveGorillaStoreCurrentCompressors();
+            }
+        }, firstExecution, 3600000);
+    }
+
+    private void ageOffGorillaStores() {
+        for (Map.Entry<String, Map<TaggedMetric, GorillaStore>> entry1 : gorillaMap.entrySet()) {
+            long maxAge = getAgeOffForMetric(entry1.getKey());
+            for (GorillaStore store : entry1.getValue().values()) {
+                store.ageOffArchivedCompressors(maxAge);
+            }
+        }
+    }
+
+    private void archiveGorillaStoreCurrentCompressors() {
+        for (Map.Entry<String, Map<TaggedMetric, GorillaStore>> entry : gorillaMap.entrySet()) {
+            for (GorillaStore store : entry.getValue().values()) {
+                store.archiveCurrentCompressor();
+            }
+        }
+    }
+
+    public long getAgeOffForMetric(String metricName) {
+        if (this.minimumAgeOff.containsKey(metricName)) {
+            return this.minimumAgeOff.get(metricName);
+        } else {
+            return this.minimumAgeOff.get(DEFAULT_AGEOFF_KEY);
+        }
+    }
+
+    public void setDefaultAgeOffMilliSec(long defaultAgeOffMilliSec) {
+        this.minimumAgeOff.put(DEFAULT_AGEOFF_KEY, defaultAgeOffMilliSec);
+        this.minimumAgeOffForIterator.put(
+                MetricAgeOffIterator.AGE_OFF_PREFIX + MetricAgeOffIterator.DEFAULT_AGEOFF_KEY,
+                Long.toString(defaultAgeOffMilliSec));
+    }
+
+    private Map<String, Long> getMinimumAgeOffs(Map<String, Integer> accumuloAgeOffDays,
+            Map<String, Integer> cacheAgeOffHours) {
+        Map<String, Long> minimumAgeOffs = new HashMap<>();
+        Set<String> keys = new HashSet<>();
+        keys.addAll(accumuloAgeOffDays.keySet());
+        keys.addAll(cacheAgeOffHours.keySet());
+        for (String name : keys) {
+            Long accumuloAgeOffValue = (accumuloAgeOffDays.containsKey(name)) ? accumuloAgeOffDays.get(name) * 86400000L
+                    : Long.MAX_VALUE;
+            Long cacheAgeOffValue = (cacheAgeOffHours.containsKey(name)) ? cacheAgeOffHours.get(name) * 3600000L
+                    : Long.MAX_VALUE;
+            minimumAgeOffs.put(name, Math.min(accumuloAgeOffValue, cacheAgeOffValue));
+        }
+        return minimumAgeOffs;
+    }
+
+    private Map<String, String> getAgeOffForIterator(Map<String, Long> minimumAgeOff) {
+        Map<String, String> ageOffOptions = new HashMap<>();
+        minimumAgeOff.forEach((k, v) -> {
+            String ageoff = Long.toString(v);
+            LOG.trace("Adding age off for metric: {} of {} milliseconds", k, v);
+            ageOffOptions.put(MetricAgeOffIterator.AGE_OFF_PREFIX + k, ageoff);
+        });
+        ageOffOptions
+                .put(MetricAgeOffIterator.DEFAULT_AGEOFF_KEY, Long.toString(minimumAgeOff.get(DEFAULT_AGEOFF_KEY)));
+        return ageOffOptions;
+    }
+
+    public Map<TaggedMetric, GorillaStore> getGorillaStores(String metric) {
+        Map<TaggedMetric, GorillaStore> metricMap = gorillaMap.get(metric);
+        if (metricMap == null) {
+            metricMap = new HashMap<>();
+            gorillaMap.put(metric, metricMap);
+        }
+        return metricMap;
+    }
+
+    public GorillaStore getGorillaStore(TaggedMetric taggedMetric) {
+        Map<TaggedMetric, GorillaStore> metricMap = gorillaMap.get(taggedMetric.getMetric());
+        if (metricMap == null) {
+            metricMap = new HashMap<>();
+            gorillaMap.put(taggedMetric.getMetric(), metricMap);
+        }
+        GorillaStore gStore = metricMap.get(taggedMetric);
+        if (gStore == null) {
+            gStore = new GorillaStore();
+            metricMap.put(taggedMetric, gStore);
+        }
+        return gStore;
+    }
+
+    public void store(Metric metric) {
+        TaggedMetric taggedMetric = new TaggedMetric(metric.getName(), metric.getTags());
+        GorillaStore gs = getGorillaStore(taggedMetric);
+        gs.addValue(metric.getValue().getTimestamp(), metric.getValue().getMeasure());
+    }
+
+    public List<QueryResponse> query(QueryRequest msg) throws TimelyException {
+
+        List<QueryResponse> result = new ArrayList<>();
+        try {
+            Collection<QueryRequest.SubQuery> subQueries = msg.getQueries();
+            for (QueryRequest.SubQuery query : subQueries) {
+                Map<Set<Tag>, List<Aggregation>> aggregations = subquery(msg, query);
+                for (Map.Entry<Set<Tag>, List<Aggregation>> entry : aggregations.entrySet()) {
+                    long tsDivisor = msg.isMsResolution() ? 1 : 1000;
+                    result.add(convertToQueryResponse(query, entry.getKey(), entry.getValue(), tsDivisor));
+                }
+            }
+            return result;
+        } catch (Exception e) {
+            LOG.error("Error during query: " + e.getMessage(), e);
+            throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error during query: "
+                    + e.getMessage(), e.getMessage(), e);
+        }
+    }
+
+    public Map<Set<Tag>, List<Aggregation>> subquery(QueryRequest msg, QueryRequest.SubQuery query)
+            throws TimelyException {
+
+        Map<Set<Tag>, List<Aggregation>> aggregationList = new HashMap<>();
+        long start = System.currentTimeMillis();
+        try {
+            SortedKeyValueIterator<org.apache.accumulo.core.data.Key, org.apache.accumulo.core.data.Value> itr = null;
+            itr = setupIterator(msg, query, getSessionAuthorizations(msg), getAgeOffForMetric(query.getMetric()));
+            LRUMap tagMap = new LRUMap(500);
+            while (itr.hasTop()) {
+                Map<Set<Tag>, Aggregation> samples = AggregationIterator.decodeValue(itr.getTopValue());
+                for (Map.Entry<Set<Tag>, Aggregation> entry : samples.entrySet()) {
+                    Set<Tag> allMatchingTags = (Set<Tag>) tagMap.get(entry.getKey());
+                    if (allMatchingTags == null) {
+                        allMatchingTags = new HashSet<>();
+                        for (Tag tag : entry.getKey()) {
+                            if (query.getTags().keySet().contains(tag.getKey())) {
+                                allMatchingTags.add(tag);
+                            }
+                        }
+                        tagMap.put(entry.getKey(), allMatchingTags);
+                    }
+                    List<Aggregation> aggregations = aggregationList.getOrDefault(allMatchingTags, new ArrayList<>());
+                    aggregations.add(entry.getValue());
+                    aggregationList.put(allMatchingTags, aggregations);
+                }
+                itr.next();
+            }
+            return aggregationList;
+        } catch (Exception e) {
+            LOG.error("Error during query: " + e.getMessage(), e);
+            throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), "Error during query: "
+                    + e.getMessage(), e.getMessage(), e);
+        } finally {
+            LOG.info("Time for cache subquery for {} - {}ms", query.toString(), System.currentTimeMillis() - start);
+        }
+    }
+
+    protected SortedKeyValueIterator<Key, Value> setupIterator(QueryRequest query, QueryRequest.SubQuery subQuery,
+            Authorizations authorizations, long ageOffForMetric) throws TimelyException {
+
+        SortedKeyValueIterator<org.apache.accumulo.core.data.Key, org.apache.accumulo.core.data.Value> itr = null;
+
+        long downsamplePeriod = DownsampleIterator.getDownsamplePeriod(subQuery);
+        long startTs = query.getStart();
+        long endTs = query.getEnd();
+        long ageOffTs = System.currentTimeMillis() - ageOffForMetric;
+        if (startTs <= ageOffTs) {
+            startTs = ageOffTs + 1;
+        }
+
+        long startOfFirstPeriod = startTs - (startTs % downsamplePeriod);
+        long endDistanceFromDownSample = endTs % downsamplePeriod;
+        long endOfLastPeriod = (endDistanceFromDownSample > 0 ? endTs + downsamplePeriod - endDistanceFromDownSample
+                : endTs);
+
+        try {
+            // create DataStoreCacheIterator which is the base iterator of
+            // the stack
+            VisibilityFilter visFilter = new VisibilityFilter(authorizations);
+
+            itr = new DataStoreCacheIterator(this, visFilter, subQuery, startOfFirstPeriod, endOfLastPeriod);
+
+            // IteratorSetting ageOffIteratorSettings = new IteratorSetting(100,
+            // "ageoff", MetricAgeOffIterator.class,
+            // this.minimumAgeOffForIterator);
+            // MetricAgeOffIterator ageOff = new MetricAgeOffIterator();
+            // ageOff.init(itr, ageOffIteratorSettings.getOptions(), null);
+            // itr = ageOff;
+
+            // create RateIterator if necessary
+            if (subQuery.isRate()) {
+                LOG.trace("Adding rate iterator");
+                IteratorSetting rate = new IteratorSetting(499, RateIterator.class);
+                RateIterator.setRateOptions(rate, subQuery.getRateOptions());
+                RateIterator rateIterator = new RateIterator();
+                rateIterator.init(itr, rate.getOptions(), null);
+                itr = rateIterator;
+            }
+
+            // create DownsampleIterator - we should always have a downsample
+            // iterator
+            // in the stack even if only using the default downsample settings
+            Class<? extends Aggregator> daggClass = DownsampleIterator.getDownsampleAggregator(subQuery);
+            if (daggClass == null) {
+                throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                        "Error during query: programming error", "daggClass == null");
+            } else {
+                LOG.trace("Downsample Aggregator type {}", daggClass.getSimpleName());
+                IteratorSetting downsample = new IteratorSetting(500, DownsampleIterator.class);
+                DownsampleIterator.setDownsampleOptions(downsample, startOfFirstPeriod, endOfLastPeriod,
+                        DownsampleIterator.getDownsamplePeriod(subQuery), -1, daggClass.getName());
+                DownsampleIterator downsampleIterator = new DownsampleIterator();
+                downsampleIterator.init(itr, downsample.getOptions(), null);
+                itr = downsampleIterator;
+            }
+
+            // create AggregatingIterator if necessary
+            Class<? extends Aggregator> aggClass = Aggregator.getAggregator(subQuery.getAggregator());
+            // the aggregation iterator is optional
+            if (aggClass != null) {
+                LOG.trace("Aggregator type {}", aggClass.getSimpleName());
+                IteratorSetting aggregation = new IteratorSetting(501, AggregationIterator.class);
+                AggregationIterator.setAggregationOptions(aggregation, subQuery.getTags(), aggClass.getName());
+                AggregationIterator aggregationIterator = new AggregationIterator();
+                aggregationIterator.init(itr, aggregation.getOptions(), null);
+                itr = aggregationIterator;
+            }
+
+            itr.seek(new Range(subQuery.getMetric()), null, true);
+
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return itr;
+    }
+
+    private Authorizations getSessionAuthorizations(AuthenticatedRequest request) {
+        return getSessionAuthorizations(request.getSessionId());
+    }
+
+    private Authorizations getSessionAuthorizations(String sessionId) {
+        if (anonAccessAllowed) {
+            if (StringUtils.isEmpty(sessionId)) {
+                return Authorizations.EMPTY;
+            } else {
+                Authorizations auths = AuthCache.getAuthorizations(sessionId);
+                if (null == auths) {
+                    auths = Authorizations.EMPTY;
+                }
+                return auths;
+            }
+        } else {
+            if (StringUtils.isEmpty(sessionId)) {
+                throw new IllegalArgumentException("session id cannot be null");
+            } else {
+                Authorizations auths = AuthCache.getAuthorizations(sessionId);
+                if (null == auths) {
+                    throw new IllegalStateException("No auths found for sessionId: " + sessionId);
+                }
+                return auths;
+            }
+        }
+    }
+
+    private QueryResponse convertToQueryResponse(QueryRequest.SubQuery query, Set<Tag> tags,
+            Collection<Aggregation> values, long tsDivisor) {
+        QueryResponse response = new QueryResponse();
+        response.setMetric(query.getMetric());
+        for (Tag tag : tags) {
+            response.putTag(tag.getKey(), tag.getValue());
+        }
+        QueryRequest.RateOption rateOptions = query.getRateOptions();
+        Aggregation combined = Aggregation.combineAggregation(values, rateOptions);
+        for (Sample entry : combined) {
+            long ts = entry.timestamp / tsDivisor;
+            response.putDps(Long.toString(ts), entry.value);
+        }
+        LOG.trace("Created query response {}", response);
+        return response;
+    }
+
+    public long getNewestTimestamp(String metric) {
+
+        long newest = 0;
+        Map<TaggedMetric, GorillaStore> gorillaStoreMap = gorillaMap.get(metric);
+        for (Map.Entry<TaggedMetric, GorillaStore> entry : gorillaStoreMap.entrySet()) {
+            if (entry.getValue().getNewestTimestamp() > newest) {
+                newest = entry.getValue().getNewestTimestamp();
+            }
+        }
+        return newest;
+    }
+
+    public long getOldestTimestamp(String metric) {
+        long oldest = Long.MAX_VALUE;
+        Map<TaggedMetric, GorillaStore> gorillaStoreMap = gorillaMap.get(metric);
+        if (gorillaStoreMap != null) {
+            for (Map.Entry<TaggedMetric, GorillaStore> entry : gorillaStoreMap.entrySet()) {
+                if (entry.getValue().getOldestTimestamp() < oldest) {
+                    oldest = entry.getValue().getOldestTimestamp();
+                }
+            }
+        }
+        return oldest;
+    }
+
+    public long getNewestTimestamp() {
+
+        long newest = 0;
+        for (String metric : gorillaMap.keySet()) {
+            Long newestForMetric = getNewestTimestamp(metric);
+            if (newestForMetric > newest) {
+                newest = newestForMetric;
+            }
+        }
+        return newest;
+    }
+
+    public long getOldestTimestamp() {
+        long oldest = Long.MAX_VALUE;
+        for (String metric : gorillaMap.keySet()) {
+            Long oldestForMetric = getOldestTimestamp(metric);
+            if (oldestForMetric < oldest) {
+                oldest = oldestForMetric;
+            }
+        }
+        return oldest;
+    }
+
+    public CacheResponse getCacheStatus() {
+        CacheResponse response = new CacheResponse();
+        response.setOldestTimestamp(getOldestTimestamp());
+        response.setNewestTimestamp(getNewestTimestamp());
+        response.setMetrics(new ArrayList<>(gorillaMap.keySet()));
+        return response;
+    }
+}

server/src/main/java/timely/store/cache/DataStoreCacheIterator.java

@@ -0,0 +1,146 @@
+package timely.store.cache;
+
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.KeyValue;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.iterators.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.adapter.accumulo.MetricAdapter;
+import timely.api.request.timeseries.QueryRequest;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.TreeMap;
+
+public class DataStoreCacheIterator implements SortedKeyValueIterator<Key, Value> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DataStoreCacheIterator.class);
+    private DataStoreCache store;
+    private VisibilityFilter visibilityFilter;
+    private QueryRequest.SubQuery query;
+    private long startTs;
+    private long endTs;
+
+    private Iterator<Map.Entry<TaggedMetric, GorillaStore>> storeItr = null;
+    private WrappedGorillaDecompressorIterator decompressors = null;
+    private KeyValue currentKeyValue = null;
+    private Queue<KeyValue> kvQueue = new LinkedList<>();
+
+    public DataStoreCacheIterator(DataStoreCache store, VisibilityFilter visibilityFilter, QueryRequest.SubQuery query,
+            long startTs, long endTs) {
+
+        this.store = store;
+        this.visibilityFilter = visibilityFilter;
+        this.query = query;
+        this.startTs = startTs;
+        this.endTs = endTs;
+        this.storeItr = this.store.getGorillaStores(query.getMetric()).entrySet().iterator();
+        this.decompressors = getNextDecompressorIterable();
+
+        long start = System.currentTimeMillis();
+        for (Map.Entry<Key, Value> entry : getEntries().entrySet()) {
+            kvQueue.add(new KeyValue(entry.getKey(), entry.getValue()));
+        }
+        LOG.info("Time to initialize cache iterator for {} - {}ms", query.toString(), System.currentTimeMillis()
+                - start);
+    }
+
+    private WrappedGorillaDecompressorIterator getNextDecompressorIterable() {
+        Map<String, String> requestedTags = query.getTags();
+        WrappedGorillaDecompressorIterator nextPair = null;
+        while (nextPair == null && storeItr.hasNext()) {
+            Map.Entry<TaggedMetric, GorillaStore> entry = storeItr.next();
+            if (entry.getKey().matches(requestedTags) && entry.getKey().isVisible(visibilityFilter)) {
+                List<WrappedGorillaDecompressor> listDecompressors = entry.getValue().getDecompressors(startTs, endTs);
+                nextPair = new WrappedGorillaDecompressorIterator(entry.getKey(), listDecompressors);
+            }
+        }
+        return nextPair;
+    }
+
+    private Map<Key, Value> getEntries() {
+
+        // sort all retrieved entries by key order, consistent with the accumulo
+        // version
+        Map<Key, Value> entries = new TreeMap<>();
+        if (decompressors != null) {
+            TaggedMetric tm = decompressors.getTaggedMetric();
+            WrappedGorillaDecompressor decompressor = decompressors.getDecompressorWrapper();
+
+            fi.iki.yak.ts.compression.gorilla.Pair gPair = null;
+            while (decompressor != null && decompressors != null) {
+                gPair = decompressor.readPair();
+                if (gPair == null) {
+                    if (decompressors.hasNext()) {
+                        decompressor = decompressors.next();
+                    } else {
+                        if ((decompressors = getNextDecompressorIterable()) != null) {
+                            tm = decompressors.getTaggedMetric();
+                            decompressor = decompressors.getDecompressorWrapper();
+                        }
+                    }
+                } else {
+                    if (gPair.getTimestamp() >= startTs && gPair.getTimestamp() <= endTs) {
+                        entries.put(MetricAdapter.toKey(tm.getMetric(), tm.getTags(), gPair.getTimestamp()), new Value(
+                                MetricAdapter.encodeValue(gPair.getDoubleValue())));
+                    }
+                }
+            }
+        }
+        return entries;
+    }
+
+    @Override
+    public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options, IteratorEnvironment env)
+            throws IOException {
+
+    }
+
+    @Override
+    public boolean hasTop() {
+        return currentKeyValue != null;
+    }
+
+    @Override
+    public void next() throws IOException {
+        currentKeyValue = kvQueue.poll();
+    }
+
+    @Override
+    public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive) throws IOException {
+        currentKeyValue = kvQueue.poll();
+    }
+
+    @Override
+    public Key getTopKey() {
+        if (currentKeyValue != null) {
+            return currentKeyValue.getKey();
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public Value getTopValue() {
+        if (currentKeyValue != null) {
+            return currentKeyValue.getValue();
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public SortedKeyValueIterator<Key, Value> deepCopy(IteratorEnvironment env) {
+        return null;
+    }
+
+}

server/src/main/java/timely/store/cache/GorillaStore.java

@@ -0,0 +1,112 @@
+package timely.store.cache;
+
+import fi.iki.yak.ts.compression.gorilla.GorillaDecompressor;
+import fi.iki.yak.ts.compression.gorilla.LongArrayInput;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+
+public class GorillaStore {
+
+    private Queue<WrappedGorillaCompressor> archivedCompressors = new LinkedList<WrappedGorillaCompressor>();
+
+    transient private WrappedGorillaCompressor current = null;
+
+    private long oldestTimestamp = -1;
+    private long newestTimestamp = -1;
+
+    private WrappedGorillaCompressor getCompressor(long timestamp) {
+        if (current == null) {
+            if (oldestTimestamp == -1) {
+                oldestTimestamp = timestamp;
+            }
+            if (newestTimestamp == -1) {
+                newestTimestamp = timestamp;
+            }
+            synchronized (this) {
+                current = new WrappedGorillaCompressor(timestamp);
+            }
+        }
+        return current;
+    }
+
+    public int ageOffArchivedCompressors(long maxAge) {
+        int numRemoved = 0;
+        long oldestRemainingTimestamp = Long.MAX_VALUE;
+        synchronized (archivedCompressors) {
+            Iterator<WrappedGorillaCompressor> itr = archivedCompressors.iterator();
+            long now = System.currentTimeMillis();
+            while (itr.hasNext()) {
+                WrappedGorillaCompressor c = itr.next();
+                if (now - c.getNewestTimestamp() > maxAge) {
+                    itr.remove();
+                    numRemoved++;
+                } else {
+                    if (c.getOldestTimestamp() < oldestRemainingTimestamp) {
+                        oldestRemainingTimestamp = c.getOldestTimestamp();
+                    }
+                }
+            }
+            oldestTimestamp = oldestRemainingTimestamp;
+        }
+        return numRemoved;
+    }
+
+    public void archiveCurrentCompressor() {
+        if (current != null) {
+            synchronized (this) {
+                current.close();
+                archivedCompressors.add(current);
+                current = null;
+            }
+        }
+    }
+
+    public List<WrappedGorillaDecompressor> getDecompressors(long begin, long end) {
+
+        List<WrappedGorillaDecompressor> decompressors = new ArrayList<>();
+
+        synchronized (archivedCompressors) {
+            for (WrappedGorillaCompressor r : archivedCompressors) {
+                if (r.inRange(begin, end)) {
+                    GorillaDecompressor d = new GorillaDecompressor(new LongArrayInput(r.getCompressorOutput()));
+                    // use -1 length since this compressor is closed
+                    decompressors.add(new WrappedGorillaDecompressor(d, -1));
+                }
+            }
+        }
+
+        // as long as begin is inRange, we should use the data in the current
+        // compressor as well
+        synchronized (this) {
+            if (current != null && current.inRange(begin, end)) {
+                LongArrayInput decompressorByteBufferInput = new LongArrayInput(current.getCompressorOutput());
+                GorillaDecompressor d = new GorillaDecompressor(decompressorByteBufferInput);
+                decompressors.add(new WrappedGorillaDecompressor(d, current.getNumEntries()));
+            }
+        }
+        return decompressors;
+    }
+
+    public void addValue(long timestamp, double value) {
+
+        // values must be inserted in order
+        if (timestamp >= newestTimestamp) {
+            newestTimestamp = timestamp;
+            synchronized (this) {
+                getCompressor(timestamp).addValue(timestamp, value);
+            }
+        }
+    }
+
+    public long getNewestTimestamp() {
+        return newestTimestamp;
+    }
+
+    public long getOldestTimestamp() {
+        return oldestTimestamp;
+    }
+}

server/src/main/java/timely/store/cache/OrderedTags.java

@@ -0,0 +1,89 @@
+package timely.store.cache;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.regex.Pattern;
+
+public class OrderedTags {
+
+    private Set<Pair<String, String>> tagSet = new TreeSet<>();
+    private static final Pattern REGEX_TEST = Pattern.compile("^\\w+$");
+
+    public OrderedTags(Map<String, String> tags) {
+        for (Map.Entry<String, String> entry : tags.entrySet()) {
+            tagSet.add(Pair.of(entry.getKey(), entry.getValue()));
+        }
+    }
+
+    @Override
+    public String toString() {
+        return tagSet.toString();
+    }
+
+    @Override
+    public int hashCode() {
+        return this.toString().hashCode();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+        if (null == o || getClass() != o.getClass())
+            return false;
+        OrderedTags other = (OrderedTags) o;
+        return this.tagSet.equals(other.tagSet);
+    }
+
+    public Map<String, String> getTags() {
+        Map<String, String> t = new LinkedHashMap();
+        for (Pair<String, String> p : tagSet) {
+            t.put(p.getLeft(), p.getRight());
+        }
+        return t;
+    }
+
+    public boolean matches(Map<String, String> tags) {
+        // default to false in case no tags match
+        boolean matches = false;
+        if (tags.isEmpty()) {
+            matches = true;
+        } else {
+            done: for (Map.Entry<String, String> entry : tags.entrySet()) {
+                for (Pair<String, String> p : tagSet) {
+                    if (entry.getKey().equals(p.getLeft())) {
+                        boolean regex = isTagValueRegex(entry.getValue());
+                        if (regex) {
+                            if (p.getRight().matches(entry.getValue())) {
+                                // at least one match
+                                matches = true;
+                            } else {
+                                // failure to match - return false
+                                matches = false;
+                                break done;
+                            }
+                        } else {
+                            if (p.getRight().equals(entry.getValue())) {
+                                // at least one match
+                                matches = true;
+                            } else {
+                                // failure to match - return false
+                                matches = false;
+                                break done;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        return matches;
+    }
+
+    private boolean isTagValueRegex(String value) {
+        return !REGEX_TEST.matcher(value).matches();
+    }
+}

server/src/main/java/timely/store/cache/TaggedMetric.java

@@ -0,0 +1,91 @@
+package timely.store.cache;
+
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.commons.lang.builder.CompareToBuilder;
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+import timely.model.Tag;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+public class TaggedMetric implements Comparable<TaggedMetric> {
+
+    public static final ColumnVisibility EMPTY_VISIBILITY = new ColumnVisibility();
+    public static final String VISIBILITY_TAG = "viz";
+    private String metric = null;
+    private ColumnVisibility columnVisibility = null;
+    private OrderedTags orderedTags = null;
+
+    public TaggedMetric(String metric, List<Tag> tags) {
+        this.metric = metric;
+        Map<String, String> tagMap = new LinkedHashMap<>();
+        for (Tag t : tags) {
+            tagMap.put(t.getKey(), t.getValue());
+        }
+        this.columnVisibility = extractVisibility(tagMap);
+        this.orderedTags = new OrderedTags(tagMap);
+    }
+
+    public String getMetric() {
+        return metric;
+    }
+
+    public Map<String, String> getTags() {
+        return orderedTags.getTags();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null || !(obj instanceof TaggedMetric)) {
+            return false;
+        }
+        TaggedMetric o = (TaggedMetric) obj;
+        EqualsBuilder eb = new EqualsBuilder();
+        eb.append(this.metric, o.metric);
+        eb.append(this.columnVisibility.toString(), o.columnVisibility.toString());
+        eb.append(this.orderedTags, o.orderedTags);
+        return eb.isEquals();
+    }
+
+    @Override
+    public int hashCode() {
+        HashCodeBuilder hcb = new HashCodeBuilder();
+        hcb.append(metric);
+        hcb.append(columnVisibility);
+        hcb.append(orderedTags);
+        return hcb.toHashCode();
+    }
+
+    @Override
+    public int compareTo(TaggedMetric o) {
+        CompareToBuilder ctb = new CompareToBuilder();
+        ctb.append(this.metric, o.metric);
+        ctb.append(this.columnVisibility, o.columnVisibility);
+        ctb.append(this.orderedTags, o.orderedTags);
+        return ctb.toComparison();
+    }
+
+    public boolean matches(Map<String, String> tags) {
+        return orderedTags.matches(tags);
+    }
+
+    public boolean isVisible(VisibilityFilter visibilityFilter) {
+
+        if (columnVisibility == null) {
+            return true;
+        } else {
+            return visibilityFilter.accept(columnVisibility);
+        }
+    }
+
+    private static ColumnVisibility extractVisibility(Map<String, String> tags) {
+
+        if (tags.containsKey(VISIBILITY_TAG)) {
+            return new ColumnVisibility(tags.get(VISIBILITY_TAG));
+        } else {
+            return EMPTY_VISIBILITY;
+        }
+    }
+}

server/src/main/java/timely/store/cache/VisibilityFilter.java

@@ -0,0 +1,40 @@
+package timely.store.cache;
+
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.accumulo.core.security.VisibilityEvaluator;
+import org.apache.accumulo.core.security.VisibilityParseException;
+import org.apache.accumulo.core.util.BadArgumentException;
+import org.apache.commons.collections.map.LRUMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class VisibilityFilter {
+
+    protected VisibilityEvaluator ve;
+    protected LRUMap cache;
+
+    private static final Logger log = LoggerFactory
+            .getLogger(org.apache.accumulo.core.iterators.system.VisibilityFilter.class);
+
+    public VisibilityFilter(Authorizations authorizations) {
+        this.ve = new VisibilityEvaluator(authorizations);
+        this.cache = new LRUMap(1000);
+    }
+
+    protected boolean accept(ColumnVisibility visibility) {
+
+        Boolean b = (Boolean) cache.get(visibility);
+        if (b != null)
+            return b;
+
+        try {
+            Boolean bb = ve.evaluate(visibility);
+            cache.put(visibility, bb);
+            return bb;
+        } catch (VisibilityParseException | BadArgumentException e) {
+            log.error("Parse Error", e);
+            return false;
+        }
+    }
+}

server/src/main/java/timely/store/cache/WrappedGorillaCompressor.java

@@ -0,0 +1,100 @@
+package timely.store.cache;
+
+import fi.iki.yak.ts.compression.gorilla.GorillaCompressor;
+import fi.iki.yak.ts.compression.gorilla.LongArrayOutput;
+import org.apache.commons.lang3.Range;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
+public class WrappedGorillaCompressor implements Serializable {
+
+    private boolean closed = false;
+    private long numEntries = 0;
+    private long oldestTimestamp;
+    private long newestTimestamp;
+    private LongArrayOutput compressorOutput = null;
+    private long[] backingArray = null;
+    private GorillaCompressor compressor;
+
+    public WrappedGorillaCompressor() {
+
+    }
+
+    public WrappedGorillaCompressor(long timestamp) {
+        this.compressorOutput = new LongArrayOutput(480);
+        this.compressor = new GorillaCompressor(timestamp, this.compressorOutput);
+        this.oldestTimestamp = timestamp;
+        this.newestTimestamp = timestamp;
+    }
+
+    public long[] getCompressorOutput() {
+        if (closed) {
+            return backingArray;
+        } else {
+            return compressorOutput.getLongArray();
+        }
+    }
+
+    public long getOldestTimestamp() {
+        return oldestTimestamp;
+    }
+
+    public long getNewestTimestamp() {
+        return newestTimestamp;
+    }
+
+    public boolean inRange(long begin, long end) {
+        Range<Long> requestedRange = Range.between(begin, end);
+        Range<Long> compressorRange = Range.between(oldestTimestamp, newestTimestamp);
+        return compressorRange.isOverlappedBy(requestedRange);
+    }
+
+    public long getNumEntries() {
+        return numEntries;
+    }
+
+    public void addValue(long timestamp, double value) {
+        if (closed) {
+            throw new IllegalStateException("Compressor is closed");
+        }
+        numEntries++;
+        newestTimestamp = timestamp;
+        compressor.addValue(timestamp, value);
+    }
+
+    public void close() {
+        compressor.close();
+        backingArray = compressorOutput.getLongArray();
+        compressorOutput = null;
+        closed = true;
+    }
+
+    private void writeObject(ObjectOutputStream out) throws IOException {
+        if (!closed) {
+            throw new IllegalStateException("Can not serialize before closed");
+        }
+        out.writeLong(numEntries);
+        out.writeLong(oldestTimestamp);
+        out.writeLong(newestTimestamp);
+        int length = backingArray.length;
+        out.writeInt(length);
+        for (int x=0; x < length; x++) {
+            out.writeLong(backingArray[x]);
+        }
+    }
+
+    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+        closed = true;
+        numEntries = in.readLong();
+        oldestTimestamp = in.readLong();
+        newestTimestamp = in.readLong();
+        int length = in.readInt();
+        backingArray = new long[length];
+        for (int x=0; x < length; x++) {
+            backingArray[x] = in.readLong();
+        }
+    }
+}

server/src/main/java/timely/store/cache/WrappedGorillaDecompressor.java

@@ -0,0 +1,29 @@
+package timely.store.cache;
+
+import fi.iki.yak.ts.compression.gorilla.*;
+
+public class WrappedGorillaDecompressor {
+
+    private GorillaDecompressor decompressor;
+    private long length;
+    private long current = 1;
+
+    public WrappedGorillaDecompressor(GorillaDecompressor decompressor, long length) {
+
+        this.decompressor = decompressor;
+        this.length = length;
+    }
+
+    public void setDecompressor(GorillaDecompressor decompressor) {
+        this.decompressor = decompressor;
+    }
+
+    public Pair readPair() {
+        if (length == -1 || current <= length) {
+            current++;
+            return decompressor.readPair();
+        } else {
+            return null;
+        }
+    }
+}

server/src/main/java/timely/store/cache/WrappedGorillaDecompressorIterator.java

@@ -0,0 +1,42 @@
+package timely.store.cache;
+
+import java.util.Iterator;
+import java.util.List;
+
+public class WrappedGorillaDecompressorIterator implements Iterator<WrappedGorillaDecompressor> {
+
+    private TaggedMetric taggedMetric;
+    private List<WrappedGorillaDecompressor> decompressors;
+    private Iterator<WrappedGorillaDecompressor> iterator;
+    private WrappedGorillaDecompressor current = null;
+
+    public WrappedGorillaDecompressorIterator(TaggedMetric taggedMetric, List<WrappedGorillaDecompressor> decompressors) {
+        this.taggedMetric = taggedMetric;
+        this.decompressors = decompressors;
+        this.iterator = this.decompressors.iterator();
+    }
+
+    public WrappedGorillaDecompressor getDecompressorWrapper() {
+        if (current == null) {
+            if (iterator.hasNext()) {
+                current = iterator.next();
+            }
+        }
+        return current;
+    }
+
+    @Override
+    public boolean hasNext() {
+        return iterator.hasNext();
+    }
+
+    @Override
+    public WrappedGorillaDecompressor next() {
+        current = iterator.next();
+        return current;
+    }
+
+    public TaggedMetric getTaggedMetric() {
+        return taggedMetric;
+    }
+}

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/InsertTestData.java

@@ -23,8 +23,8 @@ public class InsertTestData {
         METRICS.add("sys.cpu.idle");
         METRICS.add("sys.cpu.system");
         METRICS.add("sys.cpu.wait");
-        METRICS.add("sys.memory.free");
-        METRICS.add("sys.memory.used");
+        METRICS.add("sys.cache.free");
+        METRICS.add("sys.cache.used");
         METRICS.add("sys.eth.rx-errors");
         METRICS.add("sys.eth.rx-dropped");
         METRICS.add("sys.eth.rx-packets");

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

@@ -54,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()));
             }
         }
@@ -69,7 +70,6 @@ public class DownsampleIteratorTest {
     }
 
     private void createTestData2() {
-        List<List<Tag>> listOfTagVariations = new ArrayList<>();
         List<Tag> tags = Collections.singletonList(new Tag("host", "host1"));
         List<Tag> tags2 = Collections.singletonList(new Tag("host", "host2"));
 
@@ -279,6 +279,7 @@ public class DownsampleIteratorTest {
                 }
             }
             key = iter.getTopKey();
+            System.out.println(key.toString());
         } while (iter.hasTop());
 
         assertEquals(testData.lastKey(), key);

server/src/test/java/timely/store/cache/TestDataStoreCacheIterator.java

@@ -0,0 +1,215 @@
+package timely.store.cache;
+
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.security.Authorizations;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.Configuration;
+import timely.api.request.timeseries.QueryRequest;
+import timely.api.response.TimelyException;
+import timely.model.Metric;
+import timely.model.Tag;
+import timely.model.Value;
+import timely.sample.Aggregation;
+import timely.sample.Sample;
+import timely.sample.iterators.AggregationIterator;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
+
+public class TestDataStoreCacheIterator {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TestDataStoreCacheIterator.class);
+    private static Configuration configuration = null;
+
+    @BeforeClass
+    public static void setup() {
+        configuration = new Configuration();
+        configuration.getSecurity().setAllowAnonymousAccess(true);
+        HashMap<String, Integer> ageoffs = new HashMap<>();
+        ageoffs.put(DataStoreCache.DEFAULT_AGEOFF_KEY, Integer.MAX_VALUE);
+        configuration.getCache().setMetricAgeOffHours(ageoffs);
+    }
+
+    private DataStoreCache getMetricMemoryStore1(long baseTime) throws TimelyException {
+
+        DataStoreCache mmStore = new DataStoreCache(configuration);
+
+        Map<String, String> tags = new HashMap<>();
+        tags.put("part", "webservice");
+        tags.put("instance", "sample");
+
+        Random r = new Random();
+        int value = 0;
+        for (int x = 0; x <= 60 * 24; x++) {
+            for (int y = 0; y < 2; y++) {
+                tags.put("host", (y % 2 == 0) ? "r01n01" : "r02n01");
+                tags.put("rack", (y % 2 == 0) ? "r01" : "r02");
+                value = r.nextInt(1000);
+                mmStore.store(createMetric("metric.number.1", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.2", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.3", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.4", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.5", tags, value, baseTime + (x * 1000)));
+            }
+        }
+        return mmStore;
+    }
+
+    private DataStoreCache getMetricMemoryStore2(long baseTime) throws TimelyException {
+
+        DataStoreCache mmStore = new DataStoreCache(configuration);
+
+        int increment = 10;
+        Map<String, String> tags = new HashMap<>();
+        tags.put("part", "webservice");
+        tags.put("instance", "sample");
+
+        int value = 0;
+        for (int x = 0; x <= 60 * 24; x++) {
+            for (int y = 0; y < 2; y++) {
+                tags.put("host", (y % 2 == 0) ? "r01n01" : "r02n01");
+                tags.put("rack", (y % 2 == 0) ? "r01" : "r02");
+                if (y % 2 == 0) {
+                    value += increment;
+                } else {
+                    value += 2 * increment;
+                }
+                mmStore.store(createMetric("metric.number.1", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.2", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.3", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.4", tags, value, baseTime + (x * 1000)));
+                mmStore.store(createMetric("metric.number.5", tags, value, baseTime + (x * 1000)));
+            }
+        }
+        return mmStore;
+    }
+
+    @Test
+    public void testDownsampleIterator() throws TimelyException {
+
+        long BASETIME = System.currentTimeMillis();
+        // align basetime to a downsample period
+        BASETIME = BASETIME - (BASETIME % (1000 * 60));
+        DataStoreCache mmStore = getMetricMemoryStore1(BASETIME);
+
+        QueryRequest query = new QueryRequest();
+        query.setStart(BASETIME);
+        query.setEnd(BASETIME + 1440000);
+        query.setMsResolution(true);
+        QueryRequest.SubQuery subQuery = new QueryRequest.SubQuery();
+        subQuery.setDownsample(Optional.of("1m-avg"));
+        subQuery.setMetric("metric.number.1");
+        subQuery.addTag("host", ".*");
+        query.setQueries(Collections.singleton(subQuery));
+
+        SortedKeyValueIterator<org.apache.accumulo.core.data.Key, org.apache.accumulo.core.data.Value> itr = null;
+        try {
+            long firstTimestamp = -1;
+            long lastTimestamp = -1;
+            int numSamples = 0;
+            itr = mmStore.setupIterator(query, subQuery, new Authorizations(), Long.MAX_VALUE);
+            while (itr.hasTop()) {
+                itr.next();
+                Map<Set<Tag>, Aggregation> aggregations = AggregationIterator.decodeValue(itr.getTopValue());
+                for (Map.Entry<Set<Tag>, Aggregation> entry : aggregations.entrySet()) {
+                    for (Sample s : entry.getValue()) {
+                        numSamples++;
+                        if (firstTimestamp == -1) {
+                            firstTimestamp = s.timestamp;
+                        }
+                        lastTimestamp = s.timestamp;
+                    }
+                }
+            }
+            Assert.assertEquals("First timestamp incorrect", BASETIME, firstTimestamp);
+            Assert.assertEquals("Last timestamp incorrect", BASETIME + 1440000, lastTimestamp);
+            Assert.assertEquals("Number of samples incorrect", 50, numSamples);
+        } catch (IOException | ClassNotFoundException e) {
+            LOG.error("exception in test", e);
+        }
+    }
+
+    @Test
+    public void testRateIterator() throws TimelyException {
+
+        long BASETIME = System.currentTimeMillis();
+        // align basetime to a downsample period
+        BASETIME = BASETIME - (BASETIME % 1000);
+        DataStoreCache mmStore = getMetricMemoryStore2(BASETIME);
+
+        QueryRequest query = new QueryRequest();
+        query.setStart(BASETIME);
+        query.setEnd(BASETIME + 1440000);
+        query.setMsResolution(true);
+        QueryRequest.SubQuery subQuery = new QueryRequest.SubQuery();
+        subQuery.setDownsample(Optional.of("1ms-avg"));
+        subQuery.setMetric("metric.number.1");
+        subQuery.addTag("host", ".*");
+        QueryRequest.RateOption rateOption = new QueryRequest.RateOption();
+        rateOption.setCounter(false);
+        subQuery.setRate(true);
+        subQuery.setRateOptions(rateOption);
+        query.setQueries(Collections.singleton(subQuery));
+
+        int x = 0;
+        SortedKeyValueIterator<org.apache.accumulo.core.data.Key, org.apache.accumulo.core.data.Value> itr = null;
+        try {
+            // long firstTimestamp = Long.MAX_VALUE;
+            long firstTimestamp = -1;
+            long lastTimestamp = -1;
+            int numSamples = 0;
+            itr = mmStore.setupIterator(query, subQuery, new Authorizations(), Long.MAX_VALUE);
+            while (itr.hasTop()) {
+                itr.next();
+                Map<Set<Tag>, Aggregation> aggregations = AggregationIterator.decodeValue(itr.getTopValue());
+                for (Map.Entry<Set<Tag>, Aggregation> entry : aggregations.entrySet()) {
+                    for (Sample s : entry.getValue()) {
+                        numSamples++;
+                        if (firstTimestamp == -1) {
+                            firstTimestamp = s.timestamp;
+                        }
+                        lastTimestamp = s.timestamp;
+                        // if (s.timestamp < firstTimestamp) {
+                        // firstTimestamp = s.timestamp;
+                        // }
+                        // if (s.timestamp > lastTimestamp) {
+                        // lastTimestamp = s.timestamp;
+                        // }
+                    }
+                }
+            }
+            Assert.assertEquals("First timestamp incorrect", BASETIME + 1000, firstTimestamp);
+            Assert.assertEquals("Last timestamp incorrect", BASETIME + 1440000, lastTimestamp);
+            Assert.assertEquals("Number of samples incorrect", 2880, numSamples);
+        } catch (IOException | ClassNotFoundException e) {
+            LOG.error("exception in test", e);
+        }
+    }
+
+    private Metric createMetric(String metric, Map<String, String> tags, double value, long timestamp) {
+        Metric m = new Metric();
+        m.setName(metric);
+        List<Tag> tagList = new ArrayList<>();
+        for (Map.Entry<String, String> entry : tags.entrySet()) {
+            tagList.add(new Tag(entry.getKey(), entry.getValue()));
+        }
+        m.setTags(tagList);
+        Value metricValue = new Value();
+        metricValue.setMeasure(value);
+        metricValue.setTimestamp(timestamp);
+        m.setValue(metricValue);
+        return m;
+    }
+
+}

server/src/test/java/timely/store/cache/TestGorillaStore.java

@@ -0,0 +1,121 @@
+package timely.store.cache;
+
+import fi.iki.yak.ts.compression.gorilla.Pair;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import timely.Configuration;
+import timely.api.request.timeseries.QueryRequest;
+import timely.api.response.TimelyException;
+import timely.api.response.timeseries.QueryResponse;
+import timely.model.Metric;
+import timely.model.Tag;
+import timely.model.Value;
+
+import java.util.*;
+
+public class TestGorillaStore {
+
+    static private Configuration configuration = null;
+
+    @BeforeClass
+    public static void setup() {
+        configuration = new Configuration();
+        configuration.getSecurity().setAllowAnonymousAccess(true);
+    }
+
+    @Test
+    public void testOne() {
+
+        GorillaStore gStore = new GorillaStore();
+
+        long now = System.currentTimeMillis();
+        gStore.addValue(now += 100, 1.123);
+        gStore.addValue(now += 100, 2.314);
+        gStore.addValue(now += 100, 3.856);
+        gStore.addValue(now += 100, 4.7678);
+        gStore.addValue(now += 100, 5.8966);
+        gStore.addValue(now += 100, 6.0976);
+        gStore.addValue(now += 100, 1.2345);
+
+        List<WrappedGorillaDecompressor> decompressorList = gStore.getDecompressors(0, Long.MAX_VALUE);
+        Pair pair = null;
+        for (WrappedGorillaDecompressor w : decompressorList) {
+            while ((pair = w.readPair()) != null) {
+                System.out.println(pair.getTimestamp() + " --> " + pair.getDoubleValue());
+            }
+        }
+
+        System.out.println("---------------");
+
+        gStore.addValue(now += 100, 2.3456);
+        gStore.addValue(now += 100, 3.4567);
+
+        decompressorList = gStore.getDecompressors(0, Long.MAX_VALUE);
+        pair = null;
+        for (WrappedGorillaDecompressor w : decompressorList) {
+            while ((pair = w.readPair()) != null) {
+                System.out.println(pair.getTimestamp() + " --> " + pair.getDoubleValue());
+            }
+        }
+    }
+
+    private Metric createMetric(String metric, Map<String, String> tags, double value, long timestamp) {
+        Metric m = new Metric();
+        m.setName(metric);
+        List<Tag> tagList = new ArrayList<>();
+        for (Map.Entry<String, String> entry : tags.entrySet()) {
+            tagList.add(new Tag(entry.getKey(), entry.getValue()));
+        }
+        m.setTags(tagList);
+        Value metricValue = new Value();
+        metricValue.setMeasure(value);
+        metricValue.setTimestamp(timestamp);
+        m.setValue(metricValue);
+        return m;
+    }
+
+    @Test
+    public void TestExtentOfStorage() {
+
+        GorillaStore gStore = new GorillaStore();
+
+        HashMap<String, String> tags = new HashMap<>();
+        tags.put("host", "localhost");
+
+        long start = System.currentTimeMillis();
+        long timestamp = start;
+
+        for (int x = 1; x <= 100; x++) {
+
+            System.out.println("adding value x:" + x);
+            gStore.addValue(timestamp, 2.0);
+            timestamp = timestamp + 1000;
+
+            if (x % 10 == 0) {
+                gStore.archiveCurrentCompressor();
+            }
+            if (x < 50) {
+                continue;
+            }
+
+            System.out.println("fetching values x:" + x);
+            long totalObservations = 0;
+
+            List<WrappedGorillaDecompressor> decompressorList = gStore.getDecompressors(start, timestamp);
+            Pair pair = null;
+            for (WrappedGorillaDecompressor w : decompressorList) {
+                while ((pair = w.readPair()) != null) {
+                    totalObservations++;
+                    // System.out.println(pair.getTimestamp() + " --> " +
+                    // pair.getDoubleValue());
+                }
+            }
+
+            Assert.assertEquals("Unexpected number of total observations", x, totalObservations);
+
+        }
+
+    }
+
+}

server/src/test/java/timely/store/cache/TestMemoryDataStore.java

@@ -0,0 +1,194 @@
+package timely.store.cache;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import timely.Configuration;
+import timely.api.request.timeseries.QueryRequest;
+import timely.api.response.TimelyException;
+import timely.api.response.timeseries.QueryResponse;
+import timely.model.Metric;
+import timely.model.ObjectSizeOf;
+import timely.model.Tag;
+import timely.model.Value;
+
+import java.util.*;
+
+public class TestMemoryDataStore {
+
+    static private Configuration configuration = null;
+
+    @BeforeClass
+    public static void setup() {
+        configuration = new Configuration();
+        configuration.getSecurity().setAllowAnonymousAccess(true);
+        configuration.getCache().setDefaultAgeOffHours(24);
+    }
+
+    private DataStoreCache getMetricMemoryStore1(long baseTimestamp) throws TimelyException {
+
+        DataStoreCache mmStore = new DataStoreCache(configuration);
+
+        long timestamp = baseTimestamp + 10000;
+        Map<String, String> tags = new HashMap<>();
+
+        tags.put("part", "webservice");
+        tags.put("host", "host1");
+        double[] values = new double[] { 2.1, 2.3, 2.0, 2.8, 4.2, 2.1, 7.2, 4.1, 10.2 };
+        for (double d : values) {
+            mmStore.store(createMetric("mymetric", tags, d, timestamp));
+            timestamp += 1000;
+        }
+
+        tags.put("host", "host2");
+        values = new double[] { 2.9, 2.4, 2.2, 2.7, 4.2, 2.1, 7.2, 4.1, 10.2 };
+        for (double d : values) {
+            mmStore.store(createMetric("mymetric", tags, d, timestamp));
+            timestamp += 1000;
+        }
+
+        tags.put("part", "ingest");
+        tags.put("host", "host3");
+        values = new double[] { 2.9, 2.4, 2.2, 2.7, 4.2, 2.1, 7.2, 4.1, 10.2 };
+        for (double d : values) {
+            mmStore.store(createMetric("mymetric", tags, d, timestamp));
+            timestamp += 1000;
+        }
+        return mmStore;
+
+    }
+
+    private DataStoreCache getMetricMemoryStore2(long baseTimestamp) throws TimelyException {
+
+        DataStoreCache mmStore = new DataStoreCache(configuration);
+
+        Map<String, String> tags = new HashMap<>();
+        tags.put("part", "webservice");
+        tags.put("host", "r01n01");
+        tags.put("rack", "r01n01");
+        tags.put("instance", "sample");
+
+        boolean debug = false;
+        System.out.println("bytesUsed=" + ObjectSizeOf.Sizer.getObjectSize(mmStore, true, debug));
+
+        Random r = new Random();
+        long timestamp = baseTimestamp;
+        for (int x = 0; x <= 60 * 24; x++) {
+            mmStore.store(createMetric("metric.number.1", tags, r.nextInt(1000), timestamp + (x * 1000)));
+            if (x == 0) {
+                System.out.println("bytesUsed=" + ObjectSizeOf.Sizer.getObjectSize(mmStore, true, debug));
+            }
+        }
+        System.out.println("bytesUsed=" + ObjectSizeOf.Sizer.getObjectSize(mmStore, true, debug));
+        return mmStore;
+    }
+
+    @Test
+    public void testOne() throws TimelyException {
+
+        long now = System.currentTimeMillis();
+        DataStoreCache mmStore = getMetricMemoryStore1(now);
+
+        QueryRequest query = new QueryRequest();
+        query.setStart(now);
+        query.setEnd(now + 100000);
+        query.setMsResolution(true);
+        QueryRequest.SubQuery subQuery = new QueryRequest.SubQuery();
+        subQuery.setDownsample(Optional.of("5s-avg"));
+        subQuery.setMetric("mymetric");
+        subQuery.addTag("host", ".*");
+        query.setQueries(Collections.singleton(subQuery));
+
+        try {
+            List<QueryResponse> responseList = mmStore.query(query);
+            for (QueryResponse response : responseList) {
+                System.out.println(response.toString());
+            }
+        } catch (TimelyException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Test
+    public void testStorage() throws TimelyException {
+
+        long now = System.currentTimeMillis();
+        DataStoreCache mmStore = getMetricMemoryStore2(now);
+
+        QueryRequest query = new QueryRequest();
+        query.setStart(now);
+        query.setEnd(now + 86400000);
+        query.setMsResolution(true);
+        QueryRequest.SubQuery subQuery = new QueryRequest.SubQuery();
+        subQuery.setDownsample(Optional.of("5m-avg"));
+        subQuery.setMetric("metric.number.1");
+        subQuery.addTag("host", ".*");
+        query.setQueries(Collections.singleton(subQuery));
+
+        try {
+            List<QueryResponse> responseList = mmStore.query(query);
+            for (QueryResponse response : responseList) {
+                System.out.println(response.toString());
+            }
+        } catch (TimelyException e) {
+            e.printStackTrace();
+        }
+    }
+
+    private Metric createMetric(String metric, Map<String, String> tags, double value, long timestamp) {
+        Metric m = new Metric();
+        m.setName(metric);
+        List<Tag> tagList = new ArrayList<>();
+        for (Map.Entry<String, String> entry : tags.entrySet()) {
+            tagList.add(new Tag(entry.getKey(), entry.getValue()));
+        }
+        m.setTags(tagList);
+        Value metricValue = new Value();
+        metricValue.setMeasure(value);
+        metricValue.setTimestamp(timestamp);
+        m.setValue(metricValue);
+        return m;
+    }
+
+    @Test
+    public void TestExtentOfStorage() {
+        DataStoreCache mmStore = new DataStoreCache(configuration);
+
+        HashMap<String, String> tags = new HashMap<>();
+        tags.put("host", "localhost");
+
+        long start = System.currentTimeMillis();
+        long timestamp = start;
+
+        for (int x = 1; x <= 100; x++) {
+
+            Metric m = createMetric("test.metric", tags, 2.0, timestamp);
+            mmStore.store(m);
+            timestamp = timestamp + 60000;
+
+            QueryRequest query = new QueryRequest();
+            query.setStart(start);
+            query.setEnd(start + 86400000);
+            query.setMsResolution(true);
+            QueryRequest.SubQuery subQuery = new QueryRequest.SubQuery();
+            // subQuery.setDownsample(Optional.of("5m-avg"));
+            subQuery.setMetric("test.metric");
+            query.setQueries(Collections.singleton(subQuery));
+
+            try {
+                List<QueryResponse> responseList = mmStore.query(query);
+                long totalObservations = 0;
+                for (QueryResponse r : responseList) {
+                    totalObservations += r.getDps().size();
+                }
+                Assert.assertEquals("Unexpected number of total observations", x, totalObservations);
+
+            } catch (TimelyException e) {
+                e.printStackTrace();
+            }
+
+        }
+
+    }
+
+}

server/src/test/java/timely/store/cache/TestWrappedGorillaCompressor.java

@@ -0,0 +1,51 @@
+package timely.store.cache;
+
+import fi.iki.yak.ts.compression.gorilla.GorillaDecompressor;
+import fi.iki.yak.ts.compression.gorilla.LongArrayInput;
+import fi.iki.yak.ts.compression.gorilla.Pair;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.LinkedList;
+
+public class TestWrappedGorillaCompressor {
+
+    @Test
+    public void testSerialization() throws IOException, ClassNotFoundException {
+
+        long start = System.currentTimeMillis();
+        WrappedGorillaCompressor originalComprressor =  new WrappedGorillaCompressor(start);
+        long t = start;
+
+        for (int x=1; x <= 10; x++) {
+            originalComprressor.addValue(t, 10);
+            t = t + 1000;
+        }
+        originalComprressor.close();
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        ObjectOutputStream oos = new ObjectOutputStream(outputStream);
+        oos.writeObject(originalComprressor);
+        oos.close();
+
+        ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
+        ObjectInputStream ois = new ObjectInputStream(inputStream);
+        WrappedGorillaCompressor copyCompressor = (WrappedGorillaCompressor)ois.readObject();
+
+        GorillaDecompressor d = new GorillaDecompressor(new LongArrayInput(copyCompressor.getCompressorOutput()));
+
+        LinkedList<Pair> q = new LinkedList<>();
+        Pair p = null;
+        while ((p = d.readPair()) != null) {
+            q.add(p);
+        }
+
+        Assert.assertEquals(10, q.size());
+        Assert.assertEquals(start, q.peekFirst().getTimestamp());
+        Assert.assertEquals(start + 9000, q.peekLast().getTimestamp());
+    }
+}

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

@@ -12,11 +12,13 @@ import java.util.concurrent.TimeUnit;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 
+import timely.Configuration;
 import timely.Server;
 import timely.api.request.timeseries.QueryRequest;
 import timely.api.request.timeseries.QueryRequest.SubQuery;
 import timely.api.response.timeseries.QueryResponse;
 import timely.store.MetricAgeOffIterator;
+import timely.store.cache.DataStoreCache;
 import timely.test.IntegrationTest;
 
 @Category(IntegrationTest.class)
@@ -27,7 +29,32 @@ public class DataStoreIT extends OneWaySSLBase {
     private static final long TEST_TIME = System.currentTimeMillis() - ONE_DAY;
 
     @Test
-    public void testDefaultAgeOff() throws Exception {
+    public void testDefaultAgeOffWithoutCache() throws Exception {
+        conf.getCache().setEnabled(false);
+        testDefaultAgeOff(conf);
+    }
+
+    @Test
+    public void testDefaultAgeOffWithCache() throws Exception {
+        conf.getCache().setEnabled(true);
+        testDefaultAgeOff(conf);
+    }
+
+    @Test
+    public void testDefaultAgeOffWithPartialCache() throws Exception {
+        conf.getCache().setEnabled(true);
+        // This test has three values in relation to TEST_TIME = 24 hours ago
+        // TEST_TIME, TEST_TIME + 1 hour, TEST_TIME + 2 hours
+        // TEST_TIME should be aged off
+        // TEST_TIME + 1 should be retrieved from Accumulo
+        // TEST_TIME + 2 should be retrieved from the cache
+        HashMap<String, Integer> ageOffHours = new HashMap<>();
+        ageOffHours.put(DataStoreCache.DEFAULT_AGEOFF_KEY, 23);
+        conf.getCache().setMetricAgeOffHours(ageOffHours);
+        testDefaultAgeOff(conf);
+    }
+
+    public void testDefaultAgeOff(Configuration conf) throws Exception {
         HashMap<String, Integer> ageOffSettings = new HashMap<>();
         ageOffSettings.put(MetricAgeOffIterator.DEFAULT_AGEOFF_KEY, 1);
         conf.setMetricAgeOffDays(ageOffSettings);
@@ -49,6 +76,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());
@@ -61,7 +89,18 @@ public class DataStoreIT extends OneWaySSLBase {
     }
 
     @Test
-    public void testMultipleAgeOff() throws Exception {
+    public void testMultipleAgeOffWithoutCache() throws Exception {
+        conf.getCache().setEnabled(false);
+        testMultipleAgeOff(conf);
+    }
+
+    @Test
+    public void testMultipleAgeOffWithCache() throws Exception {
+        conf.getCache().setEnabled(true);
+        testMultipleAgeOff(conf);
+    }
+
+    public void testMultipleAgeOff(Configuration conf) throws Exception {
         HashMap<String, Integer> ageOffSettings = new HashMap<>();
         ageOffSettings.put(MetricAgeOffIterator.DEFAULT_AGEOFF_KEY, 1);
         ageOffSettings.put("sys.cpu.user", 1);

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

@@ -42,6 +42,7 @@ import timely.api.request.timeseries.QueryRequest;
 import timely.api.response.timeseries.QueryResponse;
 import timely.auth.AuthCache;
 import timely.netty.Constants;
+import timely.store.cache.DataStoreCache;
 import timely.test.IntegrationTest;
 import timely.test.TestConfiguration;
 import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
@@ -177,16 +178,27 @@ public class TwoWaySSLIT extends QueryBase {
     }
 
     @Test
-    public void testQueryWithVisibility() throws Exception {
+    public void testQueryWithVisibilityWithoutCache() throws Exception {
+        conf.getCache().setEnabled(false);
+        testQueryWithVisibility(conf);
+    }
+
+    @Test
+    public void testQueryWithVisibilityWithCache() throws Exception {
+        conf.getCache().setEnabled(true);
+        testQueryWithVisibility(conf);
+    }
+
+    public void testQueryWithVisibility(Configuration conf) throws Exception {
         final Server s = new Server(conf);
         s.run();
         try {
             // @formatter:off
             put("sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2",
-                "sys.cpu.user " + (TEST_TIME + 1000) + " 3.0 tag1=value1 tag2=value2",
-                "sys.cpu.user " + (TEST_TIME + 2000) + " 2.0 tag1=value1 tag3=value3 viz=A",
-                "sys.cpu.user " + (TEST_TIME + 3000) + " 2.0 tag1=value1 tag3=value3 viz=D",
-                "sys.cpu.user " + (TEST_TIME + 3000) + " 2.0 tag1=value1 tag3=value3 viz=G");
+                "sys.cpu.user " + (TEST_TIME + 1000) + " 2.0 tag1=value1 tag2=value2",
+                "sys.cpu.user " + (TEST_TIME + 2000) + " 3.0 tag1=value1 tag3=value3 viz=A",
+                "sys.cpu.user " + (TEST_TIME + 3000) + " 4.0 tag1=value1 tag3=value3 viz=D",
+                "sys.cpu.user " + (TEST_TIME + 3000) + " 5.0 tag1=value1 tag3=value3 viz=G");
             sleepUninterruptibly(TestConfiguration.WAIT_SECONDS, TimeUnit.SECONDS);
             QueryRequest request = new QueryRequest();
                 request.setStart(TEST_TIME);

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());