Add age off iterator for meta table

balancer/bin/timely-balancer.sh

@@ -32,7 +32,7 @@ $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transpo
 popd
 
 export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
-JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}" -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml
+JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
 

grafana-auth/bin/grafana-auth.sh

@@ -32,7 +32,7 @@ $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transpo
 popd
 
 export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
-JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
+JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
 

pom.xml

@@ -456,7 +456,7 @@
           <artifactId>maven-failsafe-plugin</artifactId>
           <version>${version.maven-surefire}</version>
           <configuration>
-            <argLine>${failsafeArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty-tcnative_linux_x86_64.so -Dzookeeper.sasl.client=false</argLine>
+            <argLine>${failsafeArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty_tcnative_linux_x86_64.so -Dzookeeper.sasl.client=false</argLine>
             <includes>
               <include>**/*.java</include>
             </includes>

server/bin/timely-server.sh

@@ -32,7 +32,7 @@ $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transpo
 popd
 
 export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
-JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}" -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml
+JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
 

server/bin/timely-standalone.sh

@@ -32,7 +32,7 @@ $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transpo
 popd
 
 export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
-JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}" -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml
+JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
 JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
 JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
 # for debugging

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

@@ -0,0 +1,53 @@
+package timely.adapter.accumulo;
+
+import org.apache.accumulo.core.client.lexicoder.LongLexicoder;
+import org.apache.accumulo.core.client.lexicoder.PairLexicoder;
+import org.apache.accumulo.core.client.lexicoder.StringLexicoder;
+import org.apache.accumulo.core.data.Key;
+import timely.api.model.Meta;
+import timely.model.Metric;
+import timely.model.parse.TagListParser;
+import timely.model.parse.TagParser;
+
+/**
+ * Creates Accumulo data structures from {@link Metric}
+ */
+public class MetaAdapter {
+
+    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 Key createMetricKey(String metricName, Long timestamp) {
+        return new Key(Meta.METRIC_PREFIX + metricName, "", "", timestamp);
+    }
+
+    public static Key createTagKey(String metricName, String tagName, Long timestamp) {
+        return new Key(Meta.TAG_PREFIX + metricName, tagName, "", timestamp);
+    }
+
+    public static Key createValueKey(String metricName, String tagName, String tagValue, Long timestamp) {
+        return new Key(Meta.VALUE_PREFIX + metricName, tagName, tagValue, timestamp);
+    }
+
+    public static Meta parse(Key key) {
+        String row = key.getRow().toString();
+        if (row.startsWith(Meta.METRIC_PREFIX)) {
+            return Meta.parse(key, null, Meta.METRIC_PREFIX);
+        } else if (row.startsWith(Meta.TAG_PREFIX)) {
+            return Meta.parse(key, null, Meta.TAG_PREFIX);
+        } else if (row.startsWith(Meta.VALUE_PREFIX)) {
+            return Meta.parse(key, null, Meta.VALUE_PREFIX);
+        }
+        throw new IllegalStateException("Invalid key in meta " + key.toString());
+    }
+
+    public static String decodeRowKey(Key key) {
+        return parse(key).getMetric();
+    }
+}

server/src/main/java/timely/api/model/Meta.java

@@ -55,8 +55,32 @@ public class Meta implements Comparable<Meta> {
         return keys;
     }
 
-    public static Meta parse(Key k, Value v) {
+    public static String prefix(Key key) {
-        return Meta.parse(k, v, METRIC_PREFIX);
+        String row = key.getRow().toString();
+        if (row.startsWith(Meta.METRIC_PREFIX)) {
+            return Meta.METRIC_PREFIX;
+        } else if (row.startsWith(Meta.TAG_PREFIX)) {
+            return Meta.TAG_PREFIX;
+        } else if (row.startsWith(Meta.VALUE_PREFIX)) {
+            return Meta.VALUE_PREFIX;
+        }
+        throw new IllegalStateException("Invalid key in meta " + key.toString());
+    }
+
+    public static Meta parse(Key key) {
+        return parse(key, null);
+    }
+
+    public static Meta parse(Key key, Value value) {
+        String row = key.getRow().toString();
+        if (row.startsWith(Meta.METRIC_PREFIX)) {
+            return Meta.parse(key, value, Meta.METRIC_PREFIX);
+        } else if (row.startsWith(Meta.TAG_PREFIX)) {
+            return Meta.parse(key, value, Meta.TAG_PREFIX);
+        } else if (row.startsWith(Meta.VALUE_PREFIX)) {
+            return Meta.parse(key, value, Meta.VALUE_PREFIX);
+        }
+        throw new IllegalStateException("Invalid key in meta " + key.toString());
     }
 
     public static Meta parse(Key k, Value v, String prefix) {

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

@@ -438,6 +438,7 @@ public class DataStoreImpl implements DataStore {
     @Override
     public SuggestResponse suggest(SuggestRequest request) throws TimelyException {
         SuggestResponse result = new SuggestResponse();
+        Scanner scanner = null;
         try {
             if (request.getType().equals("metrics")) {
                 Range range;
@@ -457,7 +458,7 @@ public class DataStoreImpl implements DataStore {
                     end.append(lastBytes, 0, lastBytes.length);
                     range = new Range(start, end);
                 }
-                Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
+                scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
                 scanner.setRange(range);
                 List<String> metrics = new ArrayList<>();
                 for (Entry<Key, Value> metric : scanner) {
@@ -472,6 +473,10 @@ public class DataStoreImpl implements DataStore {
             LOG.error("Error during suggest: " + ex.getMessage(), ex);
             throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
                     "Error during suggest: " + ex.getMessage(), ex.getMessage(), ex);
+        } finally {
+            if (scanner != null) {
+                scanner.close();
+            }
         }
         return result;
     }
@@ -506,28 +511,34 @@ public class DataStoreImpl implements DataStore {
             tagPatterns.put(k, Pattern.compile(v));
         });
         try {
-            List<Result> resultField = new ArrayList<>();
+            final Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
-            Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
+            try {
-            Key start = new Key(Meta.VALUE_PREFIX + msg.getQuery());
+                List<Result> resultField = new ArrayList<>();
-            Key end = start.followingKey(PartialKey.ROW);
+                Key start = new Key(Meta.VALUE_PREFIX + msg.getQuery());
-            Range range = new Range(start, end);
+                Key end = start.followingKey(PartialKey.ROW);
-            scanner.setRange(range);
+                Range range = new Range(start, end);
-            tags.keySet().forEach(k -> scanner.fetchColumnFamily(new Text(k)));
+                scanner.setRange(range);
-            int total = 0;
+                tags.keySet().forEach(k -> scanner.fetchColumnFamily(new Text(k)));
-            for (Entry<Key, Value> entry : scanner) {
+                int total = 0;
-                Meta metaEntry = Meta.parse(entry.getKey(), entry.getValue());
+                for (Entry<Key, Value> entry : scanner) {
-                if (matches(metaEntry.getTagKey(), metaEntry.getTagValue(), tagPatterns)) {
+                    Meta metaEntry = Meta.parse(entry.getKey(), entry.getValue());
-                    if (resultField.size() < msg.getLimit()) {
+                    if (matches(metaEntry.getTagKey(), metaEntry.getTagValue(), tagPatterns)) {
-                        Result r = new Result();
+                        if (resultField.size() < msg.getLimit()) {
-                        r.putTag(metaEntry.getTagKey(), metaEntry.getTagValue());
+                            Result r = new Result();
-                        resultField.add(r);
+                            r.putTag(metaEntry.getTagKey(), metaEntry.getTagValue());
+                            resultField.add(r);
+                        }
+                        total++;
                     }
-                    total++;
+                }
+                result.setResults(resultField);
+                result.setTotalResults(total);
+                result.setTime((int) (System.currentTimeMillis() - startMillis));
+            } finally {
+                if (scanner != null) {
+                    scanner.close();
                 }
             }
-            result.setResults(resultField);
-            result.setTotalResults(total);
-            result.setTime((int) (System.currentTimeMillis() - startMillis));
         } catch (Exception ex) {
             LOG.error("Error during lookup: " + ex.getMessage(), ex);
             throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
@@ -798,70 +809,77 @@ public class DataStoreImpl implements DataStore {
 
     public Set<Tag> getColumnFamilies(String metric, Map<String, String> requestedTags) throws TableNotFoundException {
 
-        Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
+        Scanner meta = null;
-        LOG.trace("Looking for requested tags: {}", tags);
+        try {
-        Scanner meta = connector.createScanner(metaTable, Authorizations.EMPTY);
+            Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
-        Text start = new Text(Meta.VALUE_PREFIX + metric);
+            LOG.trace("Looking for requested tags: {}", tags);
-        Text end = new Text(Meta.VALUE_PREFIX + metric + "\\x0000");
+            meta = connector.createScanner(metaTable, Authorizations.EMPTY);
-        end.append(new byte[] { (byte) 0xff }, 0, 1);
+            Text start = new Text(Meta.VALUE_PREFIX + metric);
-        meta.setRange(new Range(start, end));
+            Text end = new Text(Meta.VALUE_PREFIX + metric + "\\x0000");
-        // Only look for the meta entries that match our tags, if any
+            end.append(new byte[] { (byte) 0xff }, 0, 1);
-        boolean onlyFirstRow = false;
+            meta.setRange(new Range(start, end));
-        Entry<String, String> first = null;
+            // Only look for the meta entries that match our tags, if any
-        // Set the columns on the meta scanner based on the first tag
+            boolean onlyFirstRow = false;
-        // in the set of tags passed in the query. If no tags are present
+            Entry<String, String> first = null;
-        // then we are only going to return the first tag name present in the
+            // Set the columns on the meta scanner based on the first tag
-        // meta table.
+            // in the set of tags passed in the query. If no tags are present
-        Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
+            // then we are only going to return the first tag name present in the
-        if (tagIter.hasNext()) {
+            // meta table.
-            first = tagIter.next();
+            Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
-            if (isTagValueRegex(first.getValue())) {
+            if (tagIter.hasNext()) {
-                meta.fetchColumnFamily(new Text(first.getKey()));
+                first = tagIter.next();
-            } else {
+                if (isTagValueRegex(first.getValue())) {
-                meta.fetchColumn(new Text(first.getKey()), new Text(first.getValue()));
+                    meta.fetchColumnFamily(new Text(first.getKey()));
-            }
+                } else {
-        } else {
+                    meta.fetchColumn(new Text(first.getKey()), new Text(first.getValue()));
-            // grab all of the values found for the first tag for the metric
-            onlyFirstRow = true;
-        }
-        final boolean ONLY_RETURN_FIRST_TAG = onlyFirstRow;
-        Iterator<Entry<Key, Value>> iter = meta.iterator();
-        Iterator<Pair<String, String>> knownKeyValues = new Iterator<Pair<String, String>>() {
-
-            Text firstTag = null;
-            Text tagName = null;
-            Text tagValue = null;
-
-            @Override
-            public boolean hasNext() {
-                if (iter.hasNext()) {
-                    Entry<Key, Value> metaEntry = iter.next();
-                    if (null == firstTag) {
-                        firstTag = metaEntry.getKey().getColumnFamily();
-                    }
-                    tagName = metaEntry.getKey().getColumnFamily();
-                    tagValue = metaEntry.getKey().getColumnQualifier();
-                    LOG.trace("Found tag entry {}={}", tagName, tagValue);
-
-                    if (ONLY_RETURN_FIRST_TAG && !tagName.equals(firstTag)) {
-                        return false;
-                    }
-                    return true;
                 }
-                return false;
+            } else {
+                // grab all of the values found for the first tag for the metric
+                onlyFirstRow = true;
             }
+            final boolean ONLY_RETURN_FIRST_TAG = onlyFirstRow;
+            Iterator<Entry<Key, Value>> iter = meta.iterator();
+            Iterator<Pair<String, String>> knownKeyValues = new Iterator<Pair<String, String>>() {
 
-            @Override
+                Text firstTag = null;
-            public Pair<String, String> next() {
+                Text tagName = null;
-                LOG.trace("Returning tag {}={}", tagName, tagValue);
+                Text tagValue = null;
-                return new Pair<>(tagName.toString(), tagValue.toString());
+
+                @Override
+                public boolean hasNext() {
+                    if (iter.hasNext()) {
+                        Entry<Key, Value> metaEntry = iter.next();
+                        if (null == firstTag) {
+                            firstTag = metaEntry.getKey().getColumnFamily();
+                        }
+                        tagName = metaEntry.getKey().getColumnFamily();
+                        tagValue = metaEntry.getKey().getColumnQualifier();
+                        LOG.trace("Found tag entry {}={}", tagName, tagValue);
+
+                        if (ONLY_RETURN_FIRST_TAG && !tagName.equals(firstTag)) {
+                            return false;
+                        }
+                        return true;
+                    }
+                    return false;
+                }
+
+                @Override
+                public Pair<String, String> next() {
+                    LOG.trace("Returning tag {}={}", tagName, tagValue);
+                    return new Pair<>(tagName.toString(), tagValue.toString());
+                }
+            };
+            // Expand the list of tags in the meta table for this metric that
+            // 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.
+            return expandTagValues(first, knownKeyValues);
+        } finally {
+            if (meta != null) {
+                meta.close();
             }
-        };
+        }
-        // Expand the list of tags in the meta table for this metric that
-        // 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.
-        return expandTagValues(first, knownKeyValues);
     }
 
     private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags, Set<Tag> colFamValues)

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

@@ -0,0 +1,116 @@
+package timely.store;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.accumulo.core.data.ByteSequence;
+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.IteratorEnvironment;
+import org.apache.accumulo.core.iterators.OptionDescriber;
+import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
+import org.apache.accumulo.core.iterators.WrappingIterator;
+import org.apache.commons.collections4.trie.PatriciaTrie;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import timely.api.model.Meta;
+
+public class MetaAgeOffIterator extends WrappingIterator implements OptionDescriber {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MetaAgeOffIterator.class);
+    public static final String AGE_OFF_PREFIX = "ageoff.";
+    public static final String DEFAULT_AGEOFF_KEY = "default";
+
+    /* set in init */
+    private PatriciaTrie<Long> ageoffs = null;
+    private Long currentTime = 0L;
+    private Long defaultAgeOff = 0L;
+    private Long minAgeOff = Long.MAX_VALUE;
+    private Long maxAgeOff = Long.MIN_VALUE;
+
+    @Override
+    public SortedKeyValueIterator<Key, Value> deepCopy(IteratorEnvironment env) {
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        iter.ageoffs = this.ageoffs;
+        iter.defaultAgeOff = this.defaultAgeOff;
+        iter.currentTime = this.currentTime;
+        iter.minAgeOff = this.minAgeOff;
+        iter.maxAgeOff = this.maxAgeOff;
+        return iter;
+    }
+
+    @Override
+    public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options, IteratorEnvironment env)
+            throws IOException {
+        super.init(source, options, env);
+        validateOptions(options);
+        ageoffs = new PatriciaTrie<>();
+        options.forEach((k, v) -> {
+            if (k.startsWith(AGE_OFF_PREFIX)) {
+                String name = k.substring(AGE_OFF_PREFIX.length());
+                LOG.trace("Adding {} to Trie with value {}", name, Long.parseLong(v));
+                long ageoff = Long.parseLong(v);
+                this.minAgeOff = Math.min(this.minAgeOff, ageoff);
+                this.maxAgeOff = Math.max(this.maxAgeOff, ageoff);
+                ageoffs.put(name, ageoff);
+            }
+        });
+        defaultAgeOff = ageoffs.get(DEFAULT_AGEOFF_KEY);
+        currentTime = System.currentTimeMillis();
+    }
+
+    @Override
+    public void next() throws IOException {
+        super.next();
+        skipExpiredData();
+    }
+
+    @Override
+    public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive) throws IOException {
+        super.seek(range, columnFamilies, inclusive);
+        skipExpiredData();
+    }
+
+    private void skipExpiredData() throws IOException {
+
+        boolean done = false;
+        while (!done && super.hasTop()) {
+            Key top = super.getTopKey();
+            long dataAge = this.currentTime - top.getTimestamp();
+            // If less than any configured ageoff, then we want to return this K,V
+            if (dataAge < this.minAgeOff) {
+                done = true;
+            } else {
+                String metricName = Meta.parse(top).getMetric();
+                long metricAgeOff = ageoffs.getOrDefault(metricName, defaultAgeOff);
+                if (dataAge > metricAgeOff) {
+                    // data beyond ageOff, so skip
+                    super.next();
+                } else {
+                    done = true;
+                }
+            }
+        }
+    }
+
+    @Override
+    public IteratorOptions describeOptions() {
+        return new IteratorOptions("metric-age-off", "Iterator that ages off meta entries for Timely metrics",
+                Collections.singletonMap(AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY, "default age off days"),
+                Collections.singletonList("Additional metric age off properties where the metric is specified as "
+                        + AGE_OFF_PREFIX
+                        + " the metric name and the value is an integer representing the number of days to keep"));
+    }
+
+    @Override
+    public boolean validateOptions(Map<String, String> options) {
+        if (null == options.get(AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY)) {
+            throw new IllegalArgumentException(DEFAULT_AGEOFF_KEY + " must be configured for MetaAgeOffFilter");
+        }
+        return true;
+    }
+
+}

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

@@ -5,8 +5,10 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.TreeSet;
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
@@ -61,42 +63,60 @@ public class MetaCacheImpl implements MetaCache {
         final ClientConfiguration aConf = new ClientConfiguration(Collections.singletonList(apacheConf));
         String metaTable = configuration.getMetaTable();
         Map<String, Map<String, Long>> metricMap = new HashMap<>();
+        Scanner scanner = null;
         try {
             final Instance instance = new ZooKeeperInstance(aConf);
             Connector connector = instance.getConnector(accumuloConf.getUsername(),
                     new PasswordToken(accumuloConf.getPassword()));
-            Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
+            scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
-            Key begin = new Key(Meta.VALUE_PREFIX);
-            Key end = new Key("w" + '\0');
-            Range range = new Range(begin, true, end, false);
-            scanner.setRange(range);
             LOG.debug("Begin scanning " + metaTable);
+            Range metricRange = new Range(new Key(Meta.METRIC_PREFIX), true, new Key(Meta.METRIC_PREFIX + '\0'), false);
+            scanner.setRange(metricRange);
+            Set<String> allMetrics = new TreeSet<>();
             for (Map.Entry<Key, Value> entry : scanner) {
-                Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.VALUE_PREFIX);
+                Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.METRIC_PREFIX);
-                String metric = meta.getMetric();
+                allMetrics.add(meta.getMetric());
-                String tagKey = meta.getTagKey();
+            }
-                Map<String, Long> tagMap = metricMap.get(metric);
+            for (String currMetric : allMetrics) {
-                if (tagMap == null) {
+                Key begin = new Key(Meta.VALUE_PREFIX + currMetric);
-                    tagMap = new HashMap<>();
+                Key end = new Key(Meta.VALUE_PREFIX + currMetric + '\0');
-                    metricMap.put(metric, tagMap);
+                Range range = new Range(begin, true, end, false);
-                }
+                scanner.setRange(range);
-                long numTagValues = tagMap.getOrDefault(tagKey, 0l);
+                Iterator<Map.Entry<Key, Value>> itr = scanner.iterator();
-                if (entry.getKey().getTimestamp() > oldestTimestamp) {
+                boolean maxedOutValues = false;
-                    tagMap.put(tagKey, ++numTagValues);
+                while (itr.hasNext() && !maxedOutValues) {
-                    if (numTagValues <= configuration.getMetaCache().getMaxTagValues()) {
+                    Map.Entry<Key, Value> entry = itr.next();
-                        cache.put(meta, DUMMY);
+                    Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.VALUE_PREFIX);
+                    String metric = meta.getMetric();
+                    String tagKey = meta.getTagKey();
+                    Map<String, Long> tagMap = metricMap.get(metric);
+                    if (tagMap == null) {
+                        tagMap = new HashMap<>();
+                        metricMap.put(metric, tagMap);
+                    }
+                    long numTagValues = tagMap.getOrDefault(tagKey, 0l);
+                    if (entry.getKey().getTimestamp() > oldestTimestamp) {
+                        tagMap.put(tagKey, ++numTagValues);
+                        if (numTagValues <= configuration.getMetaCache().getMaxTagValues()) {
+                            cache.put(meta, DUMMY);
+                        } else {
+                            // found maxTagValues on this refresh, so remove this key
+                            cache.invalidate(meta);
+                            maxedOutValues = true;
+                        }
                     } else {
-                        // found maxTagValues on this refresh, so remove this key
+                        // this entry is more than expirationMinutes old, so remove it
                         cache.invalidate(meta);
                     }
-                } else {
-                    // this entry is more than expirationMinutes old, so remove it
-                    cache.invalidate(meta);
                 }
             }
             LOG.debug("Finished scanning " + metaTable);
         } catch (Exception e) {
             LOG.error(e.getMessage(), e);
+        } finally {
+            if (scanner != null) {
+                scanner.close();
+            }
         }
     }
 

server/src/test/java/timely/store/MetaAgeOffIteratorTest.java

@@ -0,0 +1,241 @@
+package timely.store;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import org.apache.accumulo.core.data.ByteSequence;
+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.iterators.SortedMapIterator;
+import org.junit.Assert;
+import org.junit.Test;
+import timely.adapter.accumulo.MetaAdapter;
+import timely.adapter.accumulo.MetricAdapter;
+
+public class MetaAgeOffIteratorTest {
+
+    private static final Long TEST_TIME = System.currentTimeMillis() - 1000;
+    private static final Integer ONE_DAY = 86400000;
+    private static final Value EMPTY_VALUE = new Value(new byte[0]);
+    private static final Collection<ByteSequence> columnFamilies = new ArrayList<>();
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testDefaultMissing() throws Exception {
+        SortedMap<Key, Value> table = new TreeMap<>();
+        SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        HashMap<String, String> options = new HashMap<>();
+        iter.init(source, options, null);
+    }
+
+    @Test
+    public void testDefault() throws Exception {
+        SortedMap<Key, Value> table = new TreeMap<>();
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
+
+        SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        HashMap<String, String> options = new HashMap<>();
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
+        iter.init(source, options, null);
+        iter.seek(new Range(), columnFamilies, true);
+        int seen = 0;
+        while (iter.hasTop()) {
+            Key k = iter.getTopKey();
+            Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
+            seen++;
+            iter.next();
+        }
+        Assert.assertEquals(6, seen);
+    }
+
+    @Test
+    public void testMixed() throws Exception {
+        SortedMap<Key, Value> table = new TreeMap<>();
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 1), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 2), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 3), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 4), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 5), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
+
+        SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        HashMap<String, String> options = new HashMap<>();
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
+        iter.init(source, options, null);
+        iter.seek(new Range(), columnFamilies, true);
+        int seen = 0;
+        while (iter.hasTop()) {
+            Key k = iter.getTopKey();
+            Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
+            seen++;
+            iter.next();
+        }
+        Assert.assertEquals(12, seen);
+    }
+
+    @Test
+    public void testAgeoffMixedMetricKey() throws Exception {
+        SortedMap<Key, Value> table = new TreeMap<>();
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
+
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
+
+        SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        HashMap<String, String> options = new HashMap<>();
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
+        iter.init(source, options, null);
+        iter.seek(new Range(), columnFamilies, true);
+        int seen = 0;
+        while (iter.hasTop()) {
+            Key k = iter.getTopKey();
+            Assert.assertTrue(
+                    k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
+            seen++;
+            iter.next();
+        }
+        Assert.assertEquals(7, seen);
+
+    }
+
+    @Test
+    public void testAgeoffMixedTagKey() throws Exception {
+        SortedMap<Key, Value> table = new TreeMap<>();
+        table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag1", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag2", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag3", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag4", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag5", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag6", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
+
+        table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag1", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag2", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag3", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag4", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag5", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
+        table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag6", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
+
+        SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        HashMap<String, String> options = new HashMap<>();
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
+        iter.init(source, options, null);
+        iter.seek(new Range(), columnFamilies, true);
+        int seen = 0;
+        while (iter.hasTop()) {
+            Key k = iter.getTopKey();
+            Assert.assertTrue(
+                    k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
+            seen++;
+            iter.next();
+        }
+        Assert.assertEquals(7, seen);
+
+    }
+
+    @Test
+    public void testAgeoffMixedValueKey() throws Exception {
+        SortedMap<Key, Value> table = new TreeMap<>();
+        table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag1", "value1", TEST_TIME - (3 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag2", "value2", TEST_TIME - (2 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag3", "value3", TEST_TIME - (1 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag4", "value4", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag5", "value5", TEST_TIME + (1 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag6", "value6", TEST_TIME + (2 * ONE_DAY)),
+                new Value(new byte[0]));
+
+        table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag1", "value1", TEST_TIME - (3 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag2", "value2", TEST_TIME - (2 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag3", "value3", TEST_TIME - (1 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag4", "value4", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag5", "value5", TEST_TIME + (1 * ONE_DAY)),
+                new Value(new byte[0]));
+        table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag6", "value6", TEST_TIME + (2 * ONE_DAY)),
+                new Value(new byte[0]));
+
+        SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        HashMap<String, String> options = new HashMap<>();
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
+        iter.init(source, options, null);
+        iter.seek(new Range(), columnFamilies, true);
+        int seen = 0;
+        while (iter.hasTop()) {
+            Key k = iter.getTopKey();
+            Assert.assertTrue(
+                    k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
+            seen++;
+            iter.next();
+        }
+        Assert.assertEquals(7, seen);
+
+    }
+
+    @Test
+    public void testSeekPastEndKey() throws Exception {
+        SortedMap<Key, Value> table = new TreeMap<>();
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
+        table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
+
+        SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
+        MetaAgeOffIterator iter = new MetaAgeOffIterator();
+        HashMap<String, String> options = new HashMap<>();
+        options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1));
+        iter.init(source, options, null);
+        iter.seek(new Range(new Key("sys.cpu.user"), true,
+                new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME + 3), new byte[0], new byte[0],
+                        new byte[0], TEST_TIME + 3),
+                true), columnFamilies, true);
+        int seen = 0;
+        while (iter.hasTop()) {
+            Key k = iter.getTopKey();
+            Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
+            seen++;
+            iter.next();
+        }
+        Assert.assertEquals(0, seen);
+    }
+
+}