Release version 1.0.0

closes #166 : Gracefully handle tag values with delimiters and handle bad data

.gitignore

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

analytics/pom.xml

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

client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>1.0.0</version>
     </parent>
     <properties>
         <!-- PEP 440 compliant version gets supstituted into src/main/python/template/setup.py -->

client/src/main/java/timely/model/parse/TagListParser.java

@@ -1,6 +1,5 @@
 package timely.model.parse;
 
-import com.google.common.base.Splitter;
 import timely.model.Tag;
 
 import java.util.ArrayList;
@@ -15,13 +14,22 @@ import java.util.Map;
  */
 public class TagListParser implements Parser<List<Tag>>, Combiner<Collection<Tag>> {
 
-    private static final Splitter commaSplitter = Splitter.on(",").omitEmptyStrings();
     private static final TagParser tagParser = new TagParser();
 
     @Override
     public List<Tag> parse(String t) {
         ArrayList<Tag> tags = new ArrayList<>();
-        commaSplitter.splitToList(t).forEach(p -> tags.add(tagParser.parse(p)));
+        // only split on commas that are not preceded with a backslash
+        String[] parts = t.split("(?<!\\\\),");
+        for (String p : parts) {
+            if (!p.isEmpty()) {
+                Tag tag = tagParser.parse(p);
+                // unescape commas in the key and value of the tag
+                tag.setKey(tag.getKey().replaceAll("\\\\,", ","));
+                tag.setValue(tag.getValue().replaceAll("\\\\,", ","));
+                tags.add(tag);
+            }
+        }
         return tags;
     }
 

client/src/main/java/timely/model/parse/TagParser.java

@@ -12,7 +12,7 @@ import java.util.List;
  */
 public class TagParser implements Parser<Tag>, Combiner<Tag> {
 
-    private static final Splitter equalSplitter = Splitter.on("=").trimResults();
+    private static final Splitter equalSplitter = Splitter.on("=").limit(2).trimResults();
     private static final Joiner equalsJoiner = Joiner.on("=");
 
     @Override

client/src/test/java/timely/model/parse/MetricParserTest.java

@@ -0,0 +1,40 @@
+package timely.model.parse;
+
+import org.junit.Assert;
+import org.junit.Test;
+import timely.model.Metric;
+import timely.model.Tag;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class MetricParserTest {
+
+    @Test
+    public void testParseWithEscapedCharacters() {
+
+        MetricParser parser = new MetricParser();
+        Metric m = parser.parse("put mymetric 12341234 5.0 tag1=value1,value1 tag2=value2=value2");
+
+        Assert.assertEquals("mymetric", m.getName());
+        Assert.assertEquals(12341234, (long) m.getValue().getTimestamp());
+        Assert.assertEquals(5.0, (double) m.getValue().getMeasure(), 0);
+        List<Tag> expected = new ArrayList<>();
+        expected.add(new Tag("tag1", "value1,value1"));
+        expected.add(new Tag("tag2", "value2=value2"));
+        Assert.assertEquals(expected, m.getTags());
+    }
+
+    @Test
+    public void testParseMalformatted() {
+
+        MetricParser parser = new MetricParser();
+        try {
+            // parser should throw an exception
+            parser.parse("put mymetric 12341234 5.0 tag1 tag2=value2");
+            Assert.fail();
+        } catch (IllegalArgumentException e) {
+
+        }
+    }
+}

client/src/test/java/timely/model/parse/TagListParserTest.java

@@ -1,7 +1,8 @@
-package timely.model;
+package timely.model.parse;
 
 import org.junit.Assert;
 import org.junit.Test;
+import timely.model.Tag;
 import timely.model.parse.TagListParser;
 
 import java.util.ArrayList;
@@ -37,4 +38,15 @@ public class TagListParserTest {
         String combined = new TagListParser().combine(map);
         Assert.assertEquals("tag1=value1,tag2=value2", combined);
     }
+
+    @Test
+    public void testParseTagsWithCommas() {
+
+        try {
+            String s = "tag1=value1,tag2=3.4.3_(default\\,_Date\\,_Time)_";
+            new TagListParser().parse(s);
+        } catch (Exception e) {
+            Assert.fail(e.getMessage());
+        }
+    }
 }

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>timely</groupId>
         <artifactId>timely</artifactId>
-        <version>0.0.7-SNAPSHOT</version>
+        <version>1.0.0</version>
     </parent>
     <artifactId>collectd-nsq-plugin</artifactId>
     <dependencies>

collectd-plugin-parent/pom.xml

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

collectd-timely-plugin/pom.xml

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

common/pom.xml

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

grafana/pom.xml

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

pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>0.0.7-SNAPSHOT</version>
+    <version>1.0.0</version>
     <packaging>pom</packaging>
     <name>Timely</name>
     <description>A secure and scalable time series database based on Apache Accumulo</description>

server/pom.xml

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

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

@@ -9,12 +9,15 @@ 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.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -28,16 +31,43 @@ public class MetricAdapter {
     private static final PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(),
             new LongLexicoder());
 
+    private static final PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(),
+            new StringLexicoder());
+
     private static final TagParser tagParser = new TagParser();
     private static final TagListParser tagListParser = new TagListParser();
 
     public static final ColumnVisibility EMPTY_VISIBILITY = new ColumnVisibility();
     public static final String VISIBILITY_TAG = "viz";
 
+    public static long roundTimestampToLastHour(long timestamp) {
+        return timestamp - (timestamp % 3600000);
+    }
+
+    public static long roundTimestampToNextHour(long timestamp) {
+        return timestamp - (timestamp % 3600000) + 3600000;
+    }
+
+    private static List<Tag> escapeDelimiters(List<Tag> tags) {
+        List<Tag> newTags = new ArrayList<>();
+        for (Tag t : tags) {
+            // escape all commas and equals in the tag key and value since they
+            // are used as a delimiter
+            String k = t.getKey();
+            String v = t.getValue();
+            k = k.replaceAll("=", "\\=");
+            k = k.replaceAll(",", "\\,");
+            v = v.replaceAll("=", "\\=");
+            v = v.replaceAll(",", "\\,");
+            newTags.add(new Tag(k, v));
+        }
+        return newTags;
+    }
+
     public static Mutation toMutation(Metric metric) {
         final Mutation mutation = new Mutation(encodeRowKey(metric));
-
         List<Tag> tags = metric.getTags();
+        tags = escapeDelimiters(tags);
         Collections.sort(tags);
 
         for (final Tag entry : tags) {
@@ -51,8 +81,9 @@ public class MetricAdapter {
                     .map(Tag::join)
                     .collect(Collectors.joining(","));
             // @formatter:on
-
-            mutation.put(cf, cq, extractVisibility(tags), metric.getValue().getTimestamp(), extractValue(metric));
+            byte[] cqBytes = encodeColQual(metric.getValue().getTimestamp(), cq);
+            mutation.put(new Text(cf.getBytes(Charset.forName("UTF-8"))), new Text(cqBytes), extractVisibility(tags),
+                    metric.getValue().getTimestamp(), extractValue(metric));
         }
         return mutation;
     }
@@ -87,10 +118,11 @@ public class MetricAdapter {
         // @formatter:off
         Metric.Builder builder = Metric.newBuilder()
                 .name(row.getFirst())
-                .value(row.getSecond(), ByteBuffer.wrap(v.get()).getDouble())
+                .value(k.getTimestamp(), ByteBuffer.wrap(v.get()).getDouble())
                 .tag(tagParser.parse(k.getColumnFamily().toString()));
         // @formatter:on
-        tagListParser.parse(k.getColumnQualifier().toString()).forEach(builder::tag);
+        ComparablePair<Long, String> cq = colQualCoder.decode(k.getColumnQualifier().getBytes());
+        tagListParser.parse(cq.getSecond()).forEach(builder::tag);
         if (includeVizTag && k.getColumnVisibility().getLength() > 0) {
             tagListParser.parse("viz=" + k.getColumnVisibility().toString()).forEach(builder::tag);
         }
@@ -106,7 +138,16 @@ public class MetricAdapter {
     }
 
     public static byte[] encodeRowKey(Metric metric) {
-        return encodeRowKey(metric.getName(), metric.getValue().getTimestamp());
+        // round timestamp to hour for scan efficiency and compression
+        return encodeRowKey(metric.getName(), roundTimestampToLastHour(metric.getValue().getTimestamp()));
+    }
+
+    public static byte[] encodeColQual(Long timestamp, String colQual) {
+        return colQualCoder.encode(new ComparablePair<>(timestamp, colQual));
+    }
+
+    public static Pair<Long, String> decodeColQual(byte[] colQual) {
+        return colQualCoder.decode(colQual);
     }
 
     public static Pair<String, Long> decodeRowKey(Key k) {

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

@@ -54,7 +54,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/timeseries/QueryResponse.java

@@ -2,9 +2,9 @@ package timely.api.response.timeseries;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.TreeMap;
 
 import org.apache.commons.lang3.builder.ToStringBuilder;
 
@@ -13,7 +13,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/sample/iterators/DownsampleIterator.java

@@ -72,27 +72,39 @@ public class DownsampleIterator extends WrappingIterator {
         if (super.hasTop()) {
             Key topKey = super.getTopKey();
             Value topValue = super.getTopValue();
-            Metric metric = MetricAdapter.parse(topKey, topValue);
-            LOG.trace("entering hasTop() for metric=" + metric.toString());
+            Metric metric = null;
+            try {
+                metric = MetricAdapter.parse(topKey, topValue);
+                LOG.trace("entering hasTop() for metric=" + metric.toString());
+            } catch (Exception e) {
+                LOG.error("Error: {} parsing metric at key: {}", e.getMessage(), topKey.toString());
+            }
             while (super.hasTop()) {
-                long timestamp = metric.getValue().getTimestamp();
-                if (memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, value)) {
-                    LOG.trace("returning current values - memory usage > " + memoryEstimator.maxDownsampleMemory
-                            + " for metric=" + metric.toString());
-                    break;
-                }
-                last = topKey;
-
-                Set<Tag> tags = new HashSet<Tag>(metric.getTags());
-                Downsample sample = value.get(tags);
-                if (sample == null) {
-                    try {
-                        value.put(tags, sample = factory.create());
-                    } catch (TimelyException e) {
-                        throw new RuntimeException(e);
+                topKey = super.getTopKey();
+                topValue = super.getTopValue();
+                try {
+                    metric = MetricAdapter.parse(topKey, topValue);
+                    long timestamp = metric.getValue().getTimestamp();
+                    if (memoryEstimator.shouldReturnBasedOnMemoryUsage(timestamp, value)) {
+                        LOG.trace("returning current values - memory usage > " + memoryEstimator.maxDownsampleMemory
+                                + " for metric=" + metric.toString());
+                        break;
                     }
+                    last = topKey;
+
+                    Set<Tag> tags = new HashSet<Tag>(metric.getTags());
+                    Downsample sample = value.get(tags);
+                    if (sample == null) {
+                        try {
+                            value.put(tags, sample = factory.create());
+                        } catch (TimelyException e) {
+                            throw new RuntimeException(e);
+                        }
+                    }
+                    sample.add(metric.getValue().getTimestamp(), metric.getValue().getMeasure());
+                } catch (Exception e) {
+                    LOG.error("Error: {} parsing metric at key: {}", e.getMessage(), topKey.toString());
                 }
-                sample.add(metric.getValue().getTimestamp(), metric.getValue().getMeasure());
                 try {
                     super.next();
                 } catch (IOException e) {

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

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

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

@@ -5,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;
@@ -314,7 +311,6 @@ public class DataStoreImpl implements DataStore {
             metaCache.addAll(toCache);
         }
         try {
-
             batchWriter.get().addMutation(MetricAdapter.toMutation(metric));
             internalMetrics.incrementMetricKeysInserted(metric.getTags().size());
         } catch (MutationsRejectedException e) {
@@ -469,11 +465,17 @@ public class DataStoreImpl implements DataStore {
         LOG.debug("Query request {}", msg);
         long startTs = msg.getStart();
         long endTs = msg.getEnd();
+        StringBuilder metricList = new StringBuilder();
         try {
+            long numResults = 0;
             long now = System.currentTimeMillis();
             for (SubQuery query : msg.getQueries()) {
                 Map<Set<Tag>, List<Aggregation>> allSeries = new HashMap<>();
                 String metric = query.getMetric();
+                if (metricList.length() > 0) {
+                    metricList.append(",");
+                }
+                metricList.append(metric);
                 BatchScanner scanner = connector.createBatchScanner(metricsTable, getSessionAuthorizations(msg),
                         scannerThreads);
                 try {
@@ -487,10 +489,12 @@ public class DataStoreImpl implements DataStore {
                     long endOfLastPeriod = (endDistanceFromDownSample > 0 ? endTs + downsample
                             - endDistanceFromDownSample : endTs);
 
-                    setQueryRange(scanner, metric, startOfFirstPeriod, endOfLastPeriod);
-                    List<String> tagOrder = prioritizeTags(query);
+                    List<String> tagOrder = prioritizeTags(query.getMetric(), query.getTags());
                     Map<String, String> orderedTags = orderTags(tagOrder, query.getTags());
-                    setQueryColumns(scanner, metric, orderedTags);
+                    Set<Tag> colFamValues = getColumnFamilies(metric, orderedTags);
+                    List<Range> ranges = getQueryRanges(metric, startOfFirstPeriod, endOfLastPeriod, colFamValues);
+                    scanner.setRanges(ranges);
+                    setQueryColumns(scanner, metric, orderedTags, colFamValues);
 
                     if (query.isRate()) {
                         LOG.trace("Adding rate iterator");
@@ -530,6 +534,7 @@ public class DataStoreImpl implements DataStore {
                         // Aggregation
                         Map<Set<Tag>, Aggregation> samples = AggregationIterator.decodeValue(encoded.getValue());
                         for (Entry<Set<Tag>, Aggregation> entry : samples.entrySet()) {
+                            numResults++;
                             Set<Tag> key = new HashSet<>();
                             for (Tag tag : entry.getKey()) {
                                 if (query.getTags().keySet().contains(tag.getKey())) {
@@ -549,10 +554,12 @@ public class DataStoreImpl implements DataStore {
                 // TODO groupby here?
                 long tsDivisor = msg.isMsResolution() ? 1 : 1000;
                 for (Entry<Set<Tag>, List<Aggregation>> entry : allSeries.entrySet()) {
+                    numResults += entry.getValue().size();
                     result.add(convertToQueryResponse(query, entry.getKey(), entry.getValue(), tsDivisor));
                 }
             }
-            LOG.debug("Query time: {}", (System.currentTimeMillis() - now));
+            LOG.debug("Query time: {} duration: {} metrics: {} results: {}", (System.currentTimeMillis() - now),
+                    (endTs - startTs), metricList.toString(), numResults);
             internalMetrics.addQueryResponse(result.size(), (System.currentTimeMillis() - now));
             return result;
         } catch (ClassNotFoundException | IOException | TableNotFoundException ex) {
@@ -580,10 +587,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 +597,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 +652,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 +717,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,14 +788,55 @@ public class DataStoreImpl implements DataStore {
         return result;
     }
 
-    private void setQueryRange(BatchScanner scanner, String metric, long start, long end) {
-        final byte[] start_row = MetricAdapter.encodeRowKey(metric, start);
-        LOG.trace("Start key for metric {} and time {} is {}", metric, start, start_row);
-        final byte[] end_row = MetricAdapter.encodeRowKey(metric, end);
-        LOG.trace("End key for metric {} and time {} is {}", metric, end, end_row);
-        Range range = new Range(new Text(start_row), new Text(end_row));
-        LOG.trace("Set query range to {}", range);
-        scanner.setRanges(Collections.singletonList(range));
+    public List<Range> getQueryRanges(String metric, long start, long end, Set<Tag> colFamValues) {
+        List<Range> ranges = new ArrayList<>();
+        long beginRangeRounded = MetricAdapter.roundTimestampToLastHour(start);
+        if (colFamValues.isEmpty()) {
+            final byte[] start_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+            Key startKey = new Key(new Text(start_row));
+            LOG.trace("Start key for metric {} and time {} is {}", metric, beginRangeRounded, startKey.toStringNoTime());
+            final byte[] end_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+            Key endKey = new Key(new Text(end_row));
+            LOG.trace("End key for metric {} and time {} is {}", metric, MetricAdapter.roundTimestampToNextHour(end),
+                    endKey.toStringNoTime());
+            Range range = new Range(startKey, true, endKey, false);
+            LOG.trace("Set query range to {}", range);
+            ranges.add(range);
+        } else {
+            try {
+                long lastBeginRangeRounded = MetricAdapter.roundTimestampToLastHour(end);
+                while (beginRangeRounded <= lastBeginRangeRounded) {
+                    // use end timestamp of begin + 1 hour and one msec
+                    // except the last range where we use end + 1 msec
+                    long endRangeTimestamp = (beginRangeRounded == lastBeginRangeRounded) ? end + 1 : beginRangeRounded
+                            + (1000 * 60 * 60) + 1;
+                    long beginRangeTimestamp = (beginRangeRounded == MetricAdapter.roundTimestampToLastHour(start)) ? start
+                            : beginRangeRounded;
+                    for (Tag t : colFamValues) {
+                        final byte[] start_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+                        Key startKey = new Key(new Text(start_row), new Text(t.join()
+                                .getBytes(Charset.forName("UTF-8"))), new Text(MetricAdapter.encodeColQual(
+                                beginRangeTimestamp, "")), new Text(new byte[0]), beginRangeTimestamp);
+                        LOG.trace("Start key for metric {} and time {} is {}", metric, beginRangeTimestamp,
+                                startKey.toStringNoTime());
+                        final byte[] end_row = MetricAdapter.encodeRowKey(metric, beginRangeRounded);
+                        Key endKey = new Key(new Text(end_row), new Text(t.join().getBytes(Charset.forName("UTF-8"))),
+                                new Text(MetricAdapter.encodeColQual(endRangeTimestamp, "")), new Text(new byte[0]),
+                                endRangeTimestamp);
+                        LOG.trace("End key for metric {} and time {} is {}", metric, beginRangeTimestamp,
+                                endKey.toStringNoTime());
+                        Range range = new Range(startKey, true, endKey, false);
+                        LOG.trace("Set query range to {}", range);
+                        ranges.add(range);
+                    }
+                    beginRangeRounded += (1000 * 60 * 60); // add an hour
+                }
+            } catch (Exception e) {
+                LOG.error("Error creating query ranges", e);
+            }
+        }
+
+        return ranges;
     }
 
     private Class<? extends Aggregator> getAggregator(SubQuery query) {
@@ -839,7 +897,7 @@ public class DataStoreImpl implements DataStore {
         }
     }
 
-    private long getAgeOffForMetric(String metricName) {
+    public long getAgeOffForMetric(String metricName) {
         String age = this.ageOff.get(MetricAgeOffIterator.AGE_OFF_PREFIX + metricName);
         if (null == age) {
             return this.defaultAgeOff;
@@ -869,23 +927,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/iterators/TimeSeriesGroupingIterator.java

@@ -231,29 +231,33 @@ 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;
             }
             if (time.equals(newTime)) {
 
-                Metric m = MetricAdapter.parse(k, super.getTopValue());
-                timely.model.Value v = m.getValue();
-                m.setValue(null);
+                try {
+                    Metric m = MetricAdapter.parse(k, super.getTopValue());
+                    timely.model.Value v = m.getValue();
+                    m.setValue(null);
 
-                TimeSeries values = series.get(m);
-                if (null == values) {
-                    LOG.trace("Creating new time series {}", m);
-                    values = new TimeSeries(this, filters.length);
+                    TimeSeries values = series.get(m);
+                    if (null == values) {
+                        LOG.trace("Creating new time series {}", m);
+                        values = new TimeSeries(this, filters.length);
+                    }
+
+                    LOG.trace("Adding value {} to series {}", v.getMeasure(), m);
+                    values.add(k, v.getMeasure());
+
+                    // always re-put the metric back into the TimeSeriesGroup as
+                    // it
+                    // maintains a list of prepared answers
+                    series.put(m, values);
+                } catch (Exception e) {
+                    LOG.error("Error: {} parsing metric at key: {}", e.getMessage(), k.toString());
                 }
-
-                LOG.trace("Adding value {} to series {}", v.getMeasure(), m);
-                values.add(k, v.getMeasure());
-
-                // always re-put the metric back into the TimeSeriesGroup as it
-                // maintains a list of prepared answers
-                series.put(m, values);
-
                 super.next();
             } else {
                 break;

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,48 +135,69 @@ 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();
-                    m = MetricAdapter.parse(e.getKey(), e.getValue(), true);
-                    if (responses.size() >= this.subscriptionBatchSize) {
-                        flush();
+                    try {
+                        m = MetricAdapter.parse(e.getKey(), e.getValue(), true);
+                        if (responses.size() >= this.subscriptionBatchSize) {
+                            flush();
+                        }
+                        this.responses.addResponse(MetricResponse.fromMetric(m, this.subscriptionId));
+                    } catch (Exception e1) {
+                        LOG.error("Error {} parsing metric at key: {}", e1.getMessage(), e.getKey().toString());
                     }
-                    this.responses.addResponse(MetricResponse.fromMetric(m, this.subscriptionId));
+                } else if (rangeItr.hasNext()) {
+                    // set next range on the scanner
+                    Range r = rangeItr.next();
+                    this.scanner.close();
+                    this.scanner.setRange(r);
+                    this.iter = scanner.iterator();
                 } else if (this.endTime == 0) {
                     flush();
                     long endTimeStamp = (System.currentTimeMillis() - (lag * 1000));
-                    byte[] end = MetricAdapter.encodeRowKey(this.metric, endTimeStamp);
-                    Text endRow = new Text(end);
                     this.scanner.close();
-                    Range prevRange = this.scanner.getRange();
                     if (null == m) {
                         LOG.debug("No results found, waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
-                        this.scanner.setRange(new Range(prevRange.getStartKey().getRow(), prevRange
-                                .isStartKeyInclusive(), endRow, false));
-                        this.iter = this.scanner.iterator();
+                        ranges = this.store.getQueryRanges(metric, startTime, endTimeStamp, colFamValues);
+                        rangeItr = ranges.iterator();
+                        if (rangeItr.hasNext()) {
+                            Range r = rangeItr.next();
+                            this.scanner.setRange(r);
+                            this.iter = scanner.iterator();
+                        }
                     } else {
                         // Reset the starting range to the last key returned
                         LOG.debug("Exhausted scanner, last metric returned was {}", m);
                         LOG.debug("Waiting {}ms to retry with new end time {}.", delay, endTimeStamp);
                         sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
-                        this.scanner.setRange(new Range(new Text(MetricAdapter.encodeRowKey(m)), false, endRow,
-                                prevRange.isEndKeyInclusive()));
-                        this.iter = this.scanner.iterator();
+                        ranges = this.store.getQueryRanges(metric, m.getValue().getTimestamp() + 1, endTimeStamp,
+                                colFamValues);
+                        rangeItr = ranges.iterator();
+                        if (rangeItr.hasNext()) {
+                            Range r = rangeItr.next();
+                            this.scanner.setRange(r);
+                            this.iter = this.scanner.iterator();
+                        } else {
+                            done = true;
+                        }
                     }
                 } else {
+                    done = true;
+                }
+                if (done) {
                     LOG.debug("Exhausted scanner, sending completed message for subscription {}", this.subscriptionId);
                     final MetricResponse last = new MetricResponse();
                     last.setSubscriptionId(this.subscriptionId);
                     last.setComplete(true);
                     this.responses.addResponse(last);
                     flush();
-                    break;
                 }
             }
-        } catch (Exception e) {
+        } catch (Throwable e) {
             LOG.error("Error in metric scanner, closing.", e);
         } finally {
             close();

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

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

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

@@ -42,28 +42,33 @@ public class MetricAdapterTest {
         Mutation mut = MetricAdapter.toMutation(m);
 
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(ts)));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
         Assert.assertEquals(rowCoder.decode(row), rowCoder.decode(mut.getRow()));
         Assert.assertEquals(3, mut.getUpdates().size());
         ColumnUpdate up = mut.getUpdates().get(0);
-        Assert.assertTrue(new String(up.getColumnFamily()).equals("tag1=value1"));
-        Assert.assertTrue(new String(up.getColumnQualifier()).equals("tag2=value2,tag3=value3"));
+        Assert.assertEquals("tag1=value1", new String(up.getColumnFamily()));
+        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag2=value2,tag3=value3"))),
+                new String(up.getColumnQualifier()));
+
         Assert.assertEquals(ts, up.getTimestamp());
-        Assert.assertTrue(new String(up.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(new String(up.getColumnVisibility())));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up2 = mut.getUpdates().get(1);
-        Assert.assertTrue(new String(up2.getColumnFamily()).equals("tag2=value2"));
-        Assert.assertTrue(new String(up2.getColumnQualifier()).equals("tag1=value1,tag3=value3"));
+        Assert.assertEquals("tag2=value2", new String(up2.getColumnFamily()));
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag1=value1,tag3=value3"))),
+                new String(up2.getColumnQualifier()));
         Assert.assertEquals(ts, up2.getTimestamp());
-        Assert.assertTrue(new String(up2.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(up2.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
         ColumnUpdate up3 = mut.getUpdates().get(2);
-        Assert.assertTrue(new String(up3.getColumnFamily()).equals("tag3=value3"));
-        Assert.assertTrue(new String(up3.getColumnQualifier()).equals("tag1=value1,tag2=value2"));
+        Assert.assertEquals("tag3=value3", new String(up3.getColumnFamily()));
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, "tag1=value1,tag2=value2"))),
+                new String(up3.getColumnQualifier()));
         Assert.assertEquals(ts, up3.getTimestamp());
-        Assert.assertTrue(new String(up3.getColumnVisibility()).equals(""));
+        Assert.assertEquals("", new String(up3.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
     }
 
@@ -78,14 +83,16 @@ public class MetricAdapterTest {
         Mutation mut = MetricAdapter.toMutation(m);
 
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", MetricAdapter.roundTimestampToLastHour(ts)));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
         Assert.assertEquals(rowCoder.decode(row), rowCoder.decode(mut.getRow()));
         Assert.assertEquals(1, mut.getUpdates().size());
         ColumnUpdate up = mut.getUpdates().get(0);
         Assert.assertEquals("tag1=value1", new String(up.getColumnFamily()));
-        Assert.assertEquals("", new String(up.getColumnQualifier()));
+        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
+        Assert.assertEquals(new String(colQualCoder.encode(new ComparablePair<>(ts, ""))),
+                new String(up.getColumnQualifier()));
         Assert.assertEquals(ts, up.getTimestamp());
         Assert.assertEquals("(a&b)|(c&d)", new String(up.getColumnVisibility()));
         Assert.assertArrayEquals(value, up.getValue());
@@ -110,8 +117,9 @@ public class MetricAdapterTest {
         byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
         byte[] value = new byte[Double.BYTES];
         ByteBuffer.wrap(value).putDouble(2.0D);
-        Key k = new Key(row, "tag1=value1".getBytes(), "tag2=value2,tag3=value3".getBytes(), "(a&b)|(c&d)".getBytes(),
-                1000);
+        PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
+        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000),
+                "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), 1000);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v);
         Assert.assertEquals("sys.cpu.user", m.getName());
@@ -124,14 +132,36 @@ public class MetricAdapterTest {
         Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
     }
 
+    @Test
+    public void testParseWithEscapes() throws Exception {
+        PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
+        byte[] value = new byte[Double.BYTES];
+        ByteBuffer.wrap(value).putDouble(2.0D);
+        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=test\\,value2,tag3=val\\=ue3")), "(a&b)|(c&d)".getBytes(), 1000);
+        Value v = new Value(value);
+        Metric m = MetricAdapter.parse(k, v);
+        Assert.assertEquals("sys.cpu.user", m.getName());
+        List<Tag> tags = new ArrayList<>();
+        tags.add(new Tag("tag1=value1"));
+        tags.add(new Tag("tag2=test,value2"));
+        tags.add(new Tag("tag3=val\\=ue3"));
+        Assert.assertEquals(tags, m.getTags());
+        Assert.assertEquals(new Long(1000), m.getValue().getTimestamp());
+        Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
+    }
+
     @Test
     public void testParseWithViz() throws Exception {
         PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
         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/iterators/RateIteratorTest.java

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

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

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

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

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

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

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

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

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