Update Accumulo 4 dependencies

This change assumes that the timestamp is sent in seconds,
and multiplies it by 1000 to create ms, if the timestamp
is older than a year from the start of the Timely JVM. This
means that loading older data into Timely will require it
to be sent in seconds instead of milliseconds.

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.4-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--
@@ -71,6 +71,8 @@
           <artifactId>maven-compiler-plugin</artifactId>
           <configuration>
             <compilerId>jdt</compilerId>
+            <source>11</source>
+            <target>11</target>
           </configuration>
           <dependencies>
             <dependency>

balancer/bin/timely-balancer.sh

@@ -1,18 +1,17 @@
 #!/bin/bash
 
 if [[ $(uname) == "Darwin" ]]; then
-        THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
+    THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
 else
-        THIS_SCRIPT=$(readlink -f "$0")
-        TCNATIVE_SUFFIX="so"
-        HW=`uname -m`
-        OS_PATH=linux-${HW}
+    THIS_SCRIPT=$(readlink -f "$0")
+    HW=`uname -m`
+    OS_PATH=linux-${HW}
 fi
 
 if [ -z "$1" ]; then
-  export INSTANCE="1"
+    export INSTANCE="1"
 else
-  export INSTANCE="$1"
+    export INSTANCE="$1"
 fi
 
 PID=$(jps -m | grep -E 'timely-balancer-.*-exec.jar' | grep "instance=${INSTANCE}" | awk '{print $1}')
@@ -20,7 +19,6 @@ PID=$(jps -m | grep -E 'timely-balancer-.*-exec.jar' | grep "instance=${INSTANCE
 if [ "$PID" == "" ]; then
     THIS_DIR="${THIS_SCRIPT%/*}"
     BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
-    TMP_DIR="${BASE_DIR}/tmp"
     LIB_DIR="${BASE_DIR}/lib"
     BIN_DIR="${BASE_DIR}/bin"
     export CONF_DIR="${BASE_DIR}/conf"
@@ -33,18 +31,12 @@ if [ "$PID" == "" ]; then
     # use either a value from timely-balancer-env.sh or the default
     export LOG_DIR="${TIMELY_LOG_DIR:-${BASE_DIR}/logs}"
 
-    if [[ -e ${TMP_DIR} ]]; then
-      rm -rf "${TMP_DIR}"
+    if [[ ! -e ${NATIVE_DIR} ]]; then
+        mkdir -p "${NATIVE_DIR}"
     fi
-    mkdir "${TMP_DIR}"
-
-    if [[ -e ${NATIVE_DIR} ]]; then
-      rm -rf "${NATIVE_DIR}"
-    fi
-    mkdir -p "${NATIVE_DIR}"
 
     if [[ ! -e ${LOG_DIR} ]]; then
-      mkdir "${LOG_DIR}"
+        mkdir "${LOG_DIR}"
     fi
 
     JVM_ARGS="-Xmx2G -Xms2G -XX:NewSize=1G -XX:MaxNewSize=1G"
@@ -54,12 +46,15 @@ if [ "$PID" == "" ]; then
     JVM_ARGS="${JVM_ARGS} -Djava.net.preferIPv4Stack=true"
     JVM_ARGS="${JVM_ARGS} -XX:+UseNUMA"
 
+    # for linux varieties of OS, extract native libraries if not already present
     if [[ -n "$OS_PATH" ]]; then
-      pushd "${BASE_DIR}/bin" || exit
-      "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
-      "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
-      popd || exit
-      JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+        if [[ ! -d ${NATIVE_DIR} || -z "$(ls -A ${NATIVE_DIR})" ]]; then
+            pushd "${BASE_DIR}/bin" || exit
+            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
+            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
+            popd || exit
+        fi
+        JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
     fi
 
     echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-balancer-*-exec.jar --instance=${INSTANCE}"

balancer/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.4-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>timely-balancer</artifactId>
   <description>Timely Balancer</description>
@@ -44,6 +44,10 @@
       <groupId>io.github.classgraph</groupId>
       <artifactId>classgraph</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-classes-epoll</artifactId>
+    </dependency>
     <dependency>
       <groupId>net.sourceforge.javacsv</groupId>
       <artifactId>javacsv</artifactId>

balancer/src/main/java/timely/balancer/MetricResolver.java

@@ -91,17 +91,21 @@ public class MetricResolver {
     final protected BalancerProperties balancerProperties;
     final protected CacheProperties cacheProperties;
     final protected HealthChecker healthChecker;
+    final protected Path assignmentFile;
+    final protected FileSystem fs;
 
     private enum BalanceType {
         HIGH_LOW, HIGH_AVG, AVG_LOW
     }
 
     public MetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         this.curatorFramework = curatorFramework;
         this.balancerProperties = balancerProperties;
         this.cacheProperties = cacheProperties;
         this.healthChecker = healthChecker;
+        this.assignmentFile = new Path(balancerProperties.getAssignmentFile());
+        this.fs = getFileSystem(this.assignmentFile);
     }
 
     public void start() {
@@ -916,10 +920,8 @@ public class MetricResolver {
 
     protected void createAssignmentFile() {
         try {
-            log.info("Creating assignment file: " + balancerProperties.getAssignmentFile());
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            FileSystem fs = getFileSystem(assignmentFile);
-            fs.create(assignmentFile);
+            log.info("Creating assignment file: " + this.assignmentFile);
+            this.fs.create(this.assignmentFile);
         } catch (Exception e) {
             log.error(e.getMessage(), e);
         }
@@ -927,9 +929,7 @@ public class MetricResolver {
 
     protected long getLastWrittenToHdfsTimestamp() {
         try {
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            FileSystem fs = getFileSystem(assignmentFile);
-            FileStatus fileStatus = fs.getFileStatus(assignmentFile);
+            FileStatus fileStatus = this.fs.getFileStatus(this.assignmentFile);
             return fileStatus.getModificationTime();
         } catch (FileNotFoundException e) {
             createAssignmentFile();
@@ -960,16 +960,15 @@ public class MetricResolver {
         // proceed with reading from HDFS
 
         Map<String,TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
+        CsvReader reader = null;
         try {
             boolean acquired = false;
             while (!acquired) {
                 acquired = assignmentsIPRWLock.readLock().acquire(60, TimeUnit.SECONDS);
             }
             balancerLock.writeLock().lock();
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            FileSystem fs = getFileSystem(assignmentFile);
-            FSDataInputStream iStream = fs.open(assignmentFile);
-            CsvReader reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
+            FSDataInputStream iStream = this.fs.open(this.assignmentFile);
+            reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
             reader.setUseTextQualifier(false);
 
             // skip the headers
@@ -1031,6 +1030,9 @@ public class MetricResolver {
         } catch (Exception e) {
             log.error(e.getMessage(), e);
         } finally {
+            if (reader != null) {
+                reader.close();
+            }
             balancerLock.writeLock().unlock();
             try {
                 assignmentsIPRWLock.readLock().release();
@@ -1052,12 +1054,10 @@ public class MetricResolver {
             nonCachedMetricsLocalLock.readLock().lock();
             try {
                 if (!metricToHostMap.isEmpty()) {
-                    Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-                    FileSystem fs = getFileSystem(assignmentFile);
-                    if (!fs.exists(assignmentFile.getParent())) {
-                        fs.mkdirs(assignmentFile.getParent());
+                    if (!this.fs.exists(this.assignmentFile.getParent())) {
+                        this.fs.mkdirs(this.assignmentFile.getParent());
                     }
-                    FSDataOutputStream oStream = fs.create(assignmentFile, true);
+                    FSDataOutputStream oStream = this.fs.create(this.assignmentFile, true);
                     writer = new CsvWriter(oStream, ',', Charset.forName("UTF-8"));
                     writer.setUseTextQualifier(false);
                     writer.write("metric");
@@ -1080,7 +1080,7 @@ public class MetricResolver {
                     if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
                         assignmentsLastUpdatedInHdfs.forceSet(now);
                     }
-                    log.debug("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
+                    log.info("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
                                     new Date(assignmentsLastUpdatedLocal.get()));
                     // remove metric from metricMap (ArrivalRate) if not being cached
                     metricMap.entrySet().removeIf(e -> !metricToHostMap.containsKey(e.getKey()));

balancer/src/main/java/timely/balancer/configuration/BalancerConfiguration.java

@@ -50,7 +50,7 @@ public class BalancerConfiguration {
     @Bean
     @ConditionalOnMissingBean
     public MetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         MetricResolver metricResolver = new MetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
         metricResolver.start();
         return metricResolver;

balancer/src/test/java/timely/balancer/configuration/TestBalancerConfiguration.java

@@ -48,7 +48,7 @@ public class TestBalancerConfiguration {
 
     @Bean
     public TestMetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         TestMetricResolver metricResolver = new TestMetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
         metricResolver.start();
         return metricResolver;

balancer/src/test/java/timely/balancer/test/TestMetricResolver.java

@@ -20,7 +20,7 @@ public class TestMetricResolver extends MetricResolver {
     private static final Logger log = LoggerFactory.getLogger(TestMetricResolver.class);
 
     public TestMetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         super(curatorFramework, balancerProperties, cacheProperties, healthChecker);
     }
 

client/pom.xml

@@ -4,13 +4,13 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.4-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>timely-client</artifactId>
   <properties>
     <!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
     <!-- use .dev0 instead of -SNAPSHOT -->
-    <python.client.version>2.1.4.dev0</python.client.version>
+    <python.client.version>2.1.6.dev0</python.client.version>
   </properties>
   <dependencies>
     <dependency>

client/src/test/java/timely/model/MetricTest.java

@@ -45,18 +45,21 @@ public class MetricTest {
     public void testJson() throws IOException {
         ObjectMapper mapper = new ObjectMapper();
 
-        String expectedJson = "{\"name\":\"m1\",\"timestamp\":1,\"measure\":1.0,\"tags\":[{\"k1\":\"v1\"}]}";
+        Long now = System.currentTimeMillis();
+        String ts = Long.toString(now);
 
-        Metric m1 = Metric.newBuilder().name("m1").tag("k1", "v1").value(1, 1.0).build();
+        String expectedJson = "{\"name\":\"m1\",\"timestamp\":" + ts + ",\"measure\":1.0,\"tags\":[{\"k1\":\"v1\"}]}";
+
+        Metric m1 = Metric.newBuilder().name("m1").tag("k1", "v1").value(now, 1.0).build();
 
         Metric expectedMetric = mapper.readValue(expectedJson, Metric.class);
 
         assertTrue(m1.equals(expectedMetric));
 
-        expectedJson = "{\"name\":\"m1\",\"tags\":[{\"k1\":\"v1\"}],\"timestamp\":5,\"measure\":5.0}";
+        expectedJson = "{\"name\":\"m1\",\"tags\":[{\"k1\":\"v1\"}],\"timestamp\":" + ts + ",\"measure\":5.0}";
         expectedMetric = mapper.readValue(expectedJson, Metric.class);
 
-        assertEquals((long) expectedMetric.getValue().getTimestamp(), 5L);
+        assertEquals(expectedMetric.getValue().getTimestamp(), now);
         assertEquals(expectedMetric.getValue().getMeasure(), 5.0D, 0.0);
 
     }

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

@@ -14,11 +14,32 @@ public class MetricParserTest {
     @Test
     public void testParseWithEscapedCharacters() {
 
+        Long now = System.currentTimeMillis();
+        String ts = Long.toString(now);
+
         MetricParser parser = new MetricParser();
-        Metric m = parser.parse("put mymetric 12341234 5.0 tag1=value1,value1 tag2=value2=value2");
+        Metric m = parser.parse("put mymetric " + ts + " 5.0 tag1=value1,value1 tag2=value2=value2");
 
         Assert.assertEquals("mymetric", m.getName());
-        Assert.assertEquals(12341234, (long) m.getValue().getTimestamp());
+        Assert.assertEquals(now, 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 testParseWithSecondsTimestamp() {
+
+        Long now = System.currentTimeMillis() / 1000;
+        String ts = Long.toString(now);
+
+        MetricParser parser = new MetricParser();
+        Metric m = parser.parse("put mymetric " + ts + " 5.0 tag1=value1,value1 tag2=value2=value2");
+
+        Assert.assertEquals("mymetric", m.getName());
+        Assert.assertEquals((Long) (now * 1000L), m.getValue().getTimestamp());
         Assert.assertEquals(5.0, (double) m.getValue().getMeasure(), 0);
         List<Tag> expected = new ArrayList<>();
         expected.add(new Tag("tag1", "value1,value1"));

collectd-nsq-plugin/pom.xml

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

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

@@ -79,6 +79,7 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
                 } catch (IOException e1) {}
                 try {
                     clientPool.invalidateObject(client);
+                    client = null;
                 } catch (Exception e2) {}
             }
         } finally {

collectd-plugin-parent/pom.xml

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

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

@@ -3,6 +3,7 @@ package timely.collectd.plugin;
 import java.io.OutputStream;
 import java.text.MessageFormat;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.regex.Matcher;
@@ -25,6 +26,9 @@ public abstract class CollectDPluginParent {
     private static final String GAUGE = "GAUGE";
     private static final String DERIVE = "DERIVE";
     private static final String ABSOLUTE = "ABSOLUTE";
+    private static final Collection<String> HADOOP_CONTEXTS = Arrays.asList("NameNode", "DataNode", "JobManager", "NodeManager", "JobHistoryServer",
+                    "ResourceManager", "MRAppMaster");
+    private static final Pattern HADOOP_PATTERN_4_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
     private static final Pattern STATSD_PATTERN_3_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
     private static final Pattern STATSD_PATTERN_4_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
     private static final Pattern STATSD_PATTERN_6_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_#]+)\\.([\\w-_]+)");
@@ -64,6 +68,8 @@ public abstract class CollectDPluginParent {
 
     protected void addTag(Map<String,String> m, String k, String v) {
         if (k != null && !k.isBlank() && v != null && !v.isBlank()) {
+            k = k.replaceAll("=", ":");
+            v = v.replaceAll("=", ":");
             m.put(k.trim(), v.trim());
         }
     }
@@ -71,8 +77,8 @@ public abstract class CollectDPluginParent {
     protected void addTag(Map<String,String> m, String kv) {
         if (kv != null && !kv.isBlank()) {
             String[] parts = kv.split("=");
-            if (parts.length == 2 && !parts[0].isBlank() && !parts[1].isBlank()) {
-                m.put(parts[0].trim(), parts[1].trim());
+            if (parts.length == 2) {
+                addTag(m, parts[0], parts[1]);
             }
         }
     }
@@ -99,9 +105,10 @@ public abstract class CollectDPluginParent {
         } else {
             addTag(tagMap, "host", host);
         }
-        int nIdx = host.indexOf('n');
+        String[] hostSplit = host.split("\\.");
+        int nIdx = hostSplit[0].lastIndexOf('n');
         if (-1 != nIdx) {
-            addTag(tagMap, "rack", host.substring(0, nIdx));
+            addTag(tagMap, "rack", hostSplit[0].substring(0, nIdx));
         }
         tagMap.putAll(addlTags);
 
@@ -112,13 +119,48 @@ public abstract class CollectDPluginParent {
 
         switch (plugin) {
             case STATSD_PREFIX:
+                Matcher hadoop_4_groups = HADOOP_PATTERN_4_GROUPS.matcher(typeInstance);
                 Matcher statsd_3_groups = STATSD_PATTERN_3_GROUPS.matcher(typeInstance);
                 Matcher statsd_4_groups = STATSD_PATTERN_4_GROUPS.matcher(typeInstance);
                 Matcher statsd_6_groups = STATSD_PATTERN_6_GROUPS.matcher(typeInstance);
                 String instance = null;
                 if (!typeInstance.startsWith("nsq")) {
                     String[] parts = typeInstance.split("\\.");
-                    if (parts.length % 2 == 1) {
+                    if (parts.length >= 4 && (HADOOP_CONTEXTS.contains(parts[0]) || hadoop_4_groups.matches())) {
+                        // Here we are processing the statsd metrics coming from the Hadoop Metrics2 StatsDSink without the host name.
+                        // The format of metric is: serviceName.contextName.recordName.metricName
+                        // The serviceName is used as the instance.
+                        instance = parts[0];
+                        String contextName = parts[1];
+                        int firstPeriod = typeInstance.indexOf(".");
+                        int lastPeriod = typeInstance.lastIndexOf(".");
+                        String metricName = typeInstance.substring(lastPeriod + 1);
+                        String recordName = typeInstance.substring(firstPeriod + 1, lastPeriod);
+                        // the recordName may contain useful tags such as op=<operation> and user<user>
+                        // in addition to a possibly duplicative recordName
+                        String[] recordNameSplits = recordName.split("\\.");
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(contextName).append(PERIOD);
+                        for (String split : recordNameSplits) {
+                            // avoid duplication with either the contextName or instance
+                            if (!split.equalsIgnoreCase(contextName) && !split.equalsIgnoreCase(instance)) {
+                                if (split.startsWith("NNTopUserOpCounts") && split.contains("windowMs=")) {
+                                    metric.append("NNTopUserOpCounts").append(PERIOD);
+                                    addTag(tagMap, split.substring(split.indexOf("windowMs=")));
+                                } else if (split.contains("=")) {
+                                    addTag(tagMap, split);
+                                } else {
+                                    addTag(tagMap, "record", split);
+                                }
+                            }
+                        }
+                        metric.append(metricName);
+                    } else if (parts.length >= 2 && parts[1].equals("dwquery")) {
+                        // These come from Datawave's QueryIterator metrics - the first split part is the queryId
+                        // Use statsd plus the rest of the typeInstance after the first period as the metric name
+                        int x = typeInstance.indexOf(".");
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(typeInstance.substring(x + 1));
+                        addTag(tagMap, "queryId", parts[0]);
+                    } else if (parts.length % 2 == 1) {
                         // EtsyStatsD format -- metric.(tagName.tagValue)*
                         metric.append(STATSD_PREFIX).append(PERIOD).append(parts[0]);
                         for (int x = 1; x < parts.length; x += 2) {
@@ -213,6 +255,12 @@ public abstract class CollectDPluginParent {
                 metric.append("sys.").append(plugin).append(PERIOD).append(type);
                 addTag(tagMap, INSTANCE, typeInstance.replaceAll(" ", "_"));
                 break;
+            case "load":
+                metric.append("sys.load.").append(plugin).append(PERIOD).append(type);
+                // plugin returns 1 min, 5 min, 15 min averages, but Timely can
+                // downsample on its own so we only use the first (1 min) value
+                vl.setValues(Arrays.asList(vl.getValues().get(0)));
+                break;
             case "GenericJMX":
                 metric.append("sys.").append(plugin).append(PERIOD).append(type).append(PERIOD).append(typeInstance);
                 String[] pluginInstanceSplit = pluginInstance.split("-");
@@ -222,6 +270,7 @@ public abstract class CollectDPluginParent {
                 if (pluginInstanceSplit.length > 1) {
                     addTag(tagMap, NAME, pluginInstanceSplit[1].replaceAll(" ", "_"));
                 }
+                break;
             default:
                 if (notEmpty(type) && notEmpty(typeInstance) && notEmpty(plugin) && notEmpty(pluginInstance)) {
                     metric.append("sys.").append(plugin).append(PERIOD).append(type).append(PERIOD).append(typeInstance);
@@ -248,9 +297,15 @@ public abstract class CollectDPluginParent {
             }
             Double value = vl.getValues().get(i).doubleValue();
             String tagString = tagMap.entrySet().stream().map(e -> e.getKey() + "=" + e.getValue()).collect(Collectors.joining(" "));
-            String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(), tagString);
-            logDebug(String.format("Writing: %s", datapoint));
-            write(datapoint + "\n", out);
+
+            if (metricName.isBlank() || timestamp == null || value == null) {
+                logDebug(String.format("Not writing unhandled metric: plugin:%s pluginInstance:%s type:%s typeInstance:%s", plugin, pluginInstance, type,
+                                typeInstance));
+            } else {
+                String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(), tagString);
+                logDebug(String.format("Writing: %s", datapoint));
+                write(datapoint + "\n", out);
+            }
         }
     }
 
@@ -301,5 +356,4 @@ public abstract class CollectDPluginParent {
     private boolean notEmpty(String arg) {
         return (null != arg) && !(arg.equals(""));
     }
-
 }

collectd-timely-plugin/pom.xml

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

common/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.4-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>

common/src/main/java/timely/api/request/timeseries/QueryRequest.java

@@ -32,6 +32,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
         private boolean counter = false;
         private long counterMax = 0;
         private long resetValue = 0;
+        private String interval = null;
 
         public RateOption() {}
 
@@ -84,12 +85,21 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
             this.resetValue = resetValue;
         }
 
+        public String getInterval() {
+            return interval;
+        }
+
+        public void setInterval(String interval) {
+            this.interval = interval;
+        }
+
         @Override
         public String toString() {
             ToStringBuilder tsb = new ToStringBuilder(this);
             tsb.append("counter", counter);
             tsb.append("counterMax", counterMax);
             tsb.append("resetValue", resetValue);
+            tsb.append("interval", interval);
             return tsb.toString();
         }
 
@@ -99,6 +109,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
             hcb.append(counter);
             hcb.append(counterMax);
             hcb.append(resetValue);
+            hcb.append(interval);
             return hcb.toHashCode();
         }
 
@@ -210,7 +221,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
         private String aggregator;
         private String metric;
         private boolean rate = false;
-        private RateOption rateOptions = null;
+        private RateOption rateOptions = new RateOption();
         private Optional<String> downsample = Optional.empty();
         private Map<String,String> tags = new LinkedHashMap<>();
         private Collection<Filter> filters = new ArrayList<>();
@@ -536,9 +547,12 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
                                         options.setCounter(rateOptions[x].endsWith("counter"));
                                         break;
                                     case 1:
-                                        options.setCounterMax(Long.parseLong(rateOptions[x]));
+                                        options.setInterval(rateOptions[x]);
                                         break;
                                     case 2:
+                                        options.setCounterMax(Long.parseLong(rateOptions[x]));
+                                        break;
+                                    case 3:
                                         options.setResetValue(Long.parseLong(rateOptions[x]));
                                         break;
                                     default:

common/src/main/java/timely/model/Metric.java

@@ -1,6 +1,7 @@
 package timely.model;
 
 import java.io.Serializable;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.TreeSet;
@@ -24,6 +25,12 @@ import timely.model.parse.TagParser;
 @JsonPropertyOrder({"name"})
 public class Metric implements Serializable {
 
+    // Capture the time that this class is loaded by the JVM for the purposes of
+    // comparing the timestamp. Some systems, like Telegraf's OpenTSDB output
+    // plugin, send the timestamp in seconds.
+    private static final long TIME_LOADED = System.currentTimeMillis();
+    private static final long TS_THRESHOLD = Duration.ofMillis(TIME_LOADED).minusDays(365).toMillis();
+
     private static final long serialVersionUID = 1L;
 
     private String name;
@@ -176,6 +183,9 @@ public class Metric implements Serializable {
         }
 
         public Builder value(long timestamp, double measure) {
+            if (timestamp <= TS_THRESHOLD) {
+                timestamp = timestamp * 1000;
+            }
             metric.setValue(new Value(timestamp, measure));
             return this;
         }

common/src/test/java/timely/accumulo/MetricAdapterTest.java

@@ -84,13 +84,13 @@ public class MetricAdapterTest {
 
     @Test
     public void testParse() throws Exception {
+        Long ts = System.currentTimeMillis();
         PairLexicoder<String,Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
         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=value2,tag3=value3")),
-                        "(a&b)|(c&d)".getBytes(), 1000);
+        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(ts, "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), ts);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v);
         Assert.assertEquals("sys.cpu.user", m.getName());
@@ -99,19 +99,20 @@ public class MetricAdapterTest {
         tags.add(new Tag("tag2=value2"));
         tags.add(new Tag("tag3=value3"));
         Assert.assertEquals(tags, m.getTags());
-        Assert.assertEquals(new Long(1000), m.getValue().getTimestamp());
+        Assert.assertEquals(ts, m.getValue().getTimestamp());
         Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
     }
 
     @Test
     public void testParseWithEscapes() {
+        Long ts = System.currentTimeMillis();
         PairLexicoder<String,Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
         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);
+        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(ts, "tag2=test\\,value2,tag3=val\\=ue3")),
+                        "(a&b)|(c&d)".getBytes(), ts);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v);
         Assert.assertEquals("sys.cpu.user", m.getName());
@@ -120,19 +121,19 @@ public class MetricAdapterTest {
         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(ts, m.getValue().getTimestamp());
         Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
     }
 
     @Test
     public void testParseWithViz() throws Exception {
+        Long ts = System.currentTimeMillis();
         PairLexicoder<String,Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
-        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
+        byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
         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=value2,tag3=value3")),
-                        "(a&b)|(c&d)".getBytes(), 1000);
+        Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(ts, "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), ts);
         Value v = new Value(value);
         Metric m = MetricAdapter.parse(k, v, true);
         Assert.assertEquals("sys.cpu.user", m.getName());
@@ -142,7 +143,7 @@ public class MetricAdapterTest {
         tags.add(new Tag("tag3=value3"));
         tags.add(new Tag("viz=(a&b)|(c&d)"));
         Assert.assertEquals(tags, m.getTags());
-        Assert.assertEquals(new Long(1000), m.getValue().getTimestamp());
+        Assert.assertEquals(ts, m.getValue().getTimestamp());
         Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
     }
 

grafana/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.4-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>timely-app</artifactId>
   <packaging>pom</packaging>

grafana/timely-app/dist/plugin.json

@@ -28,8 +28,8 @@
       }
     ],
     "screenshots": [],
-    "version": "2.1.4-SNAPSHOT",
-    "updated": "2024-03-13"
+    "version": "2.1.6-SNAPSHOT",
+    "updated": "2024-04-10"
   },
   "dependencies": {
     "classnames": "2.2.6",

grafana/timely-app/package.json

@@ -1,6 +1,6 @@
 {
   "name": "timely-datasource",
-  "version": "2.1.4-SNAPSHOT",
+  "version": "2.1.6-SNAPSHOT",
   "description": "Timely",
   "backend": true,
   "alerting": true,

pom.xml

@@ -3,7 +3,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>timely</groupId>
   <artifactId>timely</artifactId>
-  <version>2.1.4-SNAPSHOT</version>
+  <version>3.0.0-SNAPSHOT</version>
   <packaging>pom</packaging>
   <name>Timely</name>
   <description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -35,8 +35,8 @@
     <assembly.tarLongFileMode>posix</assembly.tarLongFileMode>
     <collectd-api-jar>/usr/share/collectd/java/collectd-api.jar</collectd-api-jar>
     <docker.image.prefix>timely/</docker.image.prefix>
-    <maven.compiler.source>11</maven.compiler.source>
-    <maven.compiler.target>11</maven.compiler.target>
+    <maven.compiler.source>17</maven.compiler.source>
+    <maven.compiler.target>17</maven.compiler.target>
     <!-- Configure the os-maven-plugin extension to expand the classifier on -->
     <!-- Fedora-"like" systems. -->
     <os.detection.classifierWithLikes>fedora</os.detection.classifierWithLikes>
@@ -44,7 +44,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <spotbugs.excludes.file>${project.basedir}/src/main/spotbugs/excludes.xml</spotbugs.excludes.file>
     <surefire.forkCount>1C</surefire.forkCount>
-    <version.accumulo>2.1.2</version.accumulo>
+    <version.accumulo>4.0.0-SNAPSHOT</version.accumulo>
     <version.caffeine>3.1.8</version.caffeine>
     <version.classgraph>4.8.165</version.classgraph>
     <version.collectd>5.7.1</version.collectd>
@@ -56,7 +56,7 @@
     <version.commons-text>1.11.0</version.commons-text>
     <version.compression.gorilla>2.1.1</version.compression.gorilla>
     <version.curator>5.5.0</version.curator>
-    <version.datawave-in-memory-accumulo>3.0.1</version.datawave-in-memory-accumulo>
+    <version.datawave-in-memory-accumulo>8.0.0-SNAPSHOT</version.datawave-in-memory-accumulo>
     <version.disruptor>3.4.4</version.disruptor>
     <version.easymock>4.0.2</version.easymock>
     <version.eclipse-lifecycle>1.0.0</version.eclipse-lifecycle>
@@ -79,7 +79,7 @@
     <version.maven-assembly-plugin>3.6.0</version.maven-assembly-plugin>
     <version.maven-build-helper-plugin>3.5.0</version.maven-build-helper-plugin>
     <version.maven-buildnumber-plugin>1.4</version.maven-buildnumber-plugin>
-    <version.maven-compiler-plugin>3.12.1</version.maven-compiler-plugin>
+    <version.maven-compiler-plugin>3.14.1</version.maven-compiler-plugin>
     <version.maven-dependency-plugin>3.6.1</version.maven-dependency-plugin>
     <version.maven-docker-plugin>1.2.1</version.maven-docker-plugin>
     <version.maven-failsafe-plugin>3.2.5</version.maven-failsafe-plugin>
@@ -89,7 +89,7 @@
     <version.maven-os-plugin>1.7.1</version.maven-os-plugin>
     <version.maven-rpm-plugin>2.3.0</version.maven-rpm-plugin>
     <version.maven-shade-plugin>3.5.1</version.maven-shade-plugin>
-    <version.maven-spotbugs-plugin>4.7.3.0</version.maven-spotbugs-plugin>
+    <version.maven-spotbugs-plugin>4.9.8.2</version.maven-spotbugs-plugin>
     <version.maven-surefire-plugin>3.2.5</version.maven-surefire-plugin>
     <version.micrometer>1.12.2</version.micrometer>
     <version.netty>4.1.106.Final</version.netty>
@@ -845,6 +845,45 @@
           <groupId>org.apache.maven.plugins</groupId>
           <artifactId>maven-dependency-plugin</artifactId>
           <version>${version.maven-dependency-plugin}</version>
+          <dependencies>
+            <dependency>
+              <groupId>org.apache.commons</groupId>
+              <artifactId>commons-lang3</artifactId>
+              <version>${version.commons-lang3}</version>
+            </dependency>
+            <dependency>
+              <groupId>org.apache.logging.log4j</groupId>
+              <artifactId>log4j-1.2-api</artifactId>
+              <version>${version.log4j2}</version>
+            </dependency>
+            <dependency>
+              <groupId>org.apache.logging.log4j</groupId>
+              <artifactId>log4j-core</artifactId>
+              <version>${version.log4j2}</version>
+            </dependency>
+            <dependency>
+              <groupId>org.apache.maven.doxia</groupId>
+              <artifactId>doxia-site-renderer</artifactId>
+              <version>2.0.0</version>
+              <exclusions>
+                <exclusion>
+                  <groupId>log4j</groupId>
+                  <artifactId>log4j</artifactId>
+                </exclusion>
+              </exclusions>
+            </dependency>
+            <dependency>
+              <groupId>org.apache.maven.reporting</groupId>
+              <artifactId>maven-reporting-impl</artifactId>
+              <version>4.0.0</version>
+              <exclusions>
+                <exclusion>
+                  <groupId>log4j</groupId>
+                  <artifactId>log4j</artifactId>
+                </exclusion>
+              </exclusions>
+            </dependency>
+          </dependencies>
         </plugin>
         <plugin>
           <groupId>org.apache.maven.plugins</groupId>
@@ -969,6 +1008,29 @@
             <threshold>Medium</threshold>
             <xmlOutput>true</xmlOutput>
           </configuration>
+          <dependencies>
+            <dependency>
+              <groupId>org.apache.logging.log4j</groupId>
+              <artifactId>log4j-1.2-api</artifactId>
+              <version>${version.log4j2}</version>
+            </dependency>
+            <dependency>
+              <groupId>org.apache.logging.log4j</groupId>
+              <artifactId>log4j-core</artifactId>
+              <version>${version.log4j2}</version>
+            </dependency>
+            <dependency>
+              <groupId>org.apache.maven.reporting</groupId>
+              <artifactId>maven-reporting-impl</artifactId>
+              <version>4.0.0</version>
+              <exclusions>
+                <exclusion>
+                  <groupId>log4j</groupId>
+                  <artifactId>log4j</artifactId>
+                </exclusion>
+              </exclusions>
+            </dependency>
+          </dependencies>
         </plugin>
         <plugin>
           <groupId>org.jacoco</groupId>

server/bin/timely-server.sh

@@ -1,18 +1,17 @@
 #!/bin/bash
 
 if [[ $(uname) == "Darwin" ]]; then
-        THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
+    THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
 else
-        THIS_SCRIPT=$(readlink -f "$0")
-        TCNATIVE_SUFFIX="so"
-        HW=`uname -m`
-        OS_PATH=linux-${HW}
+    THIS_SCRIPT=$(readlink -f "$0")
+    HW=`uname -m`
+    OS_PATH=linux-${HW}
 fi
 
 if [ -z "$1" ]; then
-  export INSTANCE="1"
+    export INSTANCE="1"
 else
-  export INSTANCE="$1"
+    export INSTANCE="$1"
 fi
 
 PID=$(jps -m | grep -E 'timely-server-.*-exec.jar timelyserver' | grep "instance=${INSTANCE}" | awk '{print $1}')
@@ -20,7 +19,6 @@ PID=$(jps -m | grep -E 'timely-server-.*-exec.jar timelyserver' | grep "instance
 if [ "$PID" == "" ]; then
     THIS_DIR="${THIS_SCRIPT%/*}"
     BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
-    TMP_DIR="${BASE_DIR}/tmp"
     LIB_DIR="${BASE_DIR}/lib"
     BIN_DIR="${BASE_DIR}/bin"
     export CONF_DIR="${BASE_DIR}/conf"
@@ -33,18 +31,12 @@ if [ "$PID" == "" ]; then
     # use either a value from timely-server-env.sh or the default
     export LOG_DIR="${TIMELY_LOG_DIR:-${BASE_DIR}/logs}"
 
-    if [[ -e ${TMP_DIR} ]]; then
-      rm -rf "${TMP_DIR}"
+    if [[ ! -e ${NATIVE_DIR} ]]; then
+        mkdir -p "${NATIVE_DIR}"
     fi
-    mkdir "${TMP_DIR}"
-
-    if [[ -e ${NATIVE_DIR} ]]; then
-      rm -rf "${NATIVE_DIR}"
-    fi
-    mkdir -p "${NATIVE_DIR}"
 
     if [[ ! -e ${LOG_DIR} ]]; then
-      mkdir "${LOG_DIR}"
+        mkdir "${LOG_DIR}"
     fi
 
     JVM_ARGS="-Xmx4G -Xms4G -XX:NewSize=1G -XX:MaxNewSize=1G"
@@ -54,12 +46,15 @@ if [ "$PID" == "" ]; then
     JVM_ARGS="${JVM_ARGS} -Djava.net.preferIPv4Stack=true"
     JVM_ARGS="${JVM_ARGS} -XX:+UseNUMA"
 
+    # for linux varieties of OS, extract native libraries if not already present
     if [[ -n "$OS_PATH" ]]; then
-      pushd "${BASE_DIR}/bin" || exit
-      "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
-      "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
-      popd || exit
-      JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+        if [[ ! -d ${NATIVE_DIR} || -z "$(ls -A ${NATIVE_DIR})" ]]; then
+            pushd "${BASE_DIR}/bin" || exit
+            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
+            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
+            popd || exit
+        fi
+        JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
     fi
 
     echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server*-exec.jar timelyserver --instance=${INSTANCE}"

server/bin/util/get-metrics.sh

@@ -1,27 +1,52 @@
 #!/bin/bash
 
-if [[ `uname` == "Darwin" ]]; then
-        THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
+if [[ $(uname) == "Darwin" ]]; then
+    THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
 else
-        THIS_SCRIPT=`readlink -f $0`
+    THIS_SCRIPT=$(readlink -f "$0")
+    HW=`uname -m`
+    OS_PATH=linux-${HW}
 fi
 
-export INSTANCE="1"
-
 THIS_DIR="${THIS_SCRIPT%/*}"
-BASE_DIR=$(cd $THIS_DIR/../.. && pwd)
+BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
 LIB_DIR="${BASE_DIR}/lib"
 BIN_DIR="${BASE_DIR}/bin"
 export CONF_DIR="${BASE_DIR}/conf"
-export LOG_DIR="${BASE_DIR}/logs"
+NATIVE_DIR="${BIN_DIR}/META-INF/native"
 
 set -a
-. ${BIN_DIR}/timely-server-env.sh
+. "${BIN_DIR}/timely-server-env.sh"
 set +a
 
+# use either a value from timely-server-env.sh or the default
+export LOG_DIR="${TIMELY_LOG_DIR:-${BASE_DIR}/logs}"
+
+if [[ ! -e ${NATIVE_DIR} ]]; then
+    mkdir -p "${NATIVE_DIR}"
+fi
+
+if [[ ! -e ${LOG_DIR} ]]; then
+    mkdir "${LOG_DIR}"
+fi
+
 JVM_ARGS="-Xmx256m -Xms128m"
 JVM_ARGS="${JVM_ARGS} -Dlogging.config=${CONF_DIR}/log4j2-console.yml"
 JVM_ARGS="${JVM_ARGS} -Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
+JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:+UseStringDeduplication"
+JVM_ARGS="${JVM_ARGS} -Djava.net.preferIPv4Stack=true"
+JVM_ARGS="${JVM_ARGS} -XX:+UseNUMA"
+
+# for linux varieties of OS, extract native libraries if not already present
+if [[ -n "$OS_PATH" ]]; then
+    if [[ ! -d ${NATIVE_DIR} || -z "$(ls -A ${NATIVE_DIR})" ]]; then
+        pushd "${BASE_DIR}/bin" || exit
+        "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
+        "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
+        popd || exit
+    fi
+    JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+fi
 
 echo "$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server*-exec.jar metrics"
 $JAVA_HOME/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server-*-exec.jar metrics

server/bin/util/get-tablet-summary.sh

@@ -1,27 +1,52 @@
 #!/bin/bash
 
-if [[ `uname` == "Darwin" ]]; then
-        THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
+if [[ $(uname) == "Darwin" ]]; then
+    THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
 else
-        THIS_SCRIPT=`readlink -f $0`
+    THIS_SCRIPT=$(readlink -f "$0")
+    HW=`uname -m`
+    OS_PATH=linux-${HW}
 fi
 
-export INSTANCE="1"
-
 THIS_DIR="${THIS_SCRIPT%/*}"
-BASE_DIR=$(cd $THIS_DIR/../.. && pwd)
+BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
 LIB_DIR="${BASE_DIR}/lib"
 BIN_DIR="${BASE_DIR}/bin"
 export CONF_DIR="${BASE_DIR}/conf"
-export LOG_DIR="${BASE_DIR}/logs"
+NATIVE_DIR="${BIN_DIR}/META-INF/native"
 
 set -a
-. ${BIN_DIR}/timely-server-env.sh
+. "${BIN_DIR}/timely-server-env.sh"
 set +a
 
+# use either a value from timely-server-env.sh or the default
+export LOG_DIR="${TIMELY_LOG_DIR:-${BASE_DIR}/logs}"
+
+if [[ ! -e ${NATIVE_DIR} ]]; then
+    mkdir -p "${NATIVE_DIR}"
+fi
+
+if [[ ! -e ${LOG_DIR} ]]; then
+    mkdir "${LOG_DIR}"
+fi
+
 JVM_ARGS="-Xmx1024m -Xms128m"
 JVM_ARGS="${JVM_ARGS} -Dlogging.config=${CONF_DIR}/log4j2-console.yml"
 JVM_ARGS="${JVM_ARGS} -Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
+JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:+UseStringDeduplication"
+JVM_ARGS="${JVM_ARGS} -Djava.net.preferIPv4Stack=true"
+JVM_ARGS="${JVM_ARGS} -XX:+UseNUMA"
 
-echo "$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server*-exec.jar tablet --spring.profiles.active=timely,tablet"
-$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server-*-exec.jar tablet --spring.profiles.active=timely,tablet
+# for linux varieties of OS, extract native libraries if not already present
+if [[ -n "$OS_PATH" ]]; then
+    if [[ ! -d ${NATIVE_DIR} || -z "$(ls -A ${NATIVE_DIR})" ]]; then
+        pushd "${BASE_DIR}/bin" || exit
+        "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
+        "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
+        popd || exit
+    fi
+    JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+fi
+
+echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server*-exec.jar tablet --spring.profiles.active=timely,tablet"
+"${JAVA_HOME}"/bin/java ${JVM_ARGS} -jar "${BIN_DIR}"/timely-server-*-exec.jar tablet --spring.profiles.active=timely,tablet

server/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.4-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>timely-server</artifactId>
   <description>Timely Server</description>
@@ -68,6 +68,10 @@
       <groupId>io.micrometer</groupId>
       <artifactId>micrometer-core</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-classes-epoll</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.projectreactor.netty</groupId>
       <artifactId>reactor-netty</artifactId>
@@ -224,6 +228,12 @@
       <artifactId>log4j-core</artifactId>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcprov-ext-jdk18on</artifactId>
+      <version>1.73</version>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>org.javassist</groupId>
       <artifactId>javassist</artifactId>
@@ -370,8 +380,8 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-compiler-plugin</artifactId>
         <configuration>
-          <source>11</source>
-          <target>11</target>
+          <source>17</source>
+          <target>17</target>
         </configuration>
       </plugin>
     </plugins>

server/src/main/java/timely/application/testquery/MetricQuery.java

@@ -53,11 +53,9 @@ public class MetricQuery {
     }
 
     public void setCounterOptions(long max, long resetValue) {
-        QueryRequest.RateOption rateOption = new QueryRequest.RateOption();
-        rateOption.setCounter(true);
-        rateOption.setCounterMax(max);
-        rateOption.setResetValue(resetValue);
-        this.subQuery.setRateOptions(rateOption);
+        this.subQuery.getRateOptions().setCounter(true);
+        this.subQuery.getRateOptions().setCounterMax(max);
+        this.subQuery.getRateOptions().setResetValue(resetValue);
     }
 
     public void setTags(Map<String,String> tags) {

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

@@ -1,5 +1,7 @@
 package timely.server.sample.iterators;
 
+import static org.apache.accumulo.core.conf.ConfigurationTypeHelper.getTimeInMillis;
+
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -12,6 +14,7 @@ import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.util.Pair;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -20,12 +23,14 @@ import timely.api.request.timeseries.QueryRequest;
 public class RateIterator extends TimeSeriesGroupingIterator {
 
     private static final Logger log = LoggerFactory.getLogger(RateIterator.class);
-    private static final String COUNTER_MAX = "rate.counter.max";
-    private static final String RATE_RESET_VALUE = "rate.reset.value";
+    public static final String COUNTER_MAX = "rate.counter.max";
+    public static final String RATE_RESET_VALUE = "rate.reset.value";
+    public static final String RATE_INTERVAL = "rate.interval";
 
     private boolean isCounter = false;
     private long maxCounter = 0;
     private long resetValue = 0;
+    private long interval = 1000;
 
     public static void setRateOptions(IteratorSetting is, QueryRequest.RateOption options) {
         if (options != null && options.isCounter()) {
@@ -33,6 +38,11 @@ public class RateIterator extends TimeSeriesGroupingIterator {
             is.addOption(COUNTER_MAX, Long.toString(options.getCounterMax()));
             is.addOption(RATE_RESET_VALUE, Long.toString(options.getResetValue()));
         }
+        if (options != null && StringUtils.isNotBlank(options.getInterval())) {
+            is.addOption(RATE_INTERVAL, Long.toString(getTimeInMillis(options.getInterval())));
+        } else {
+            is.addOption(RATE_INTERVAL, "1000");
+        }
     }
 
     @Override
@@ -41,8 +51,11 @@ public class RateIterator extends TimeSeriesGroupingIterator {
             this.isCounter = true;
             this.maxCounter = Long.parseLong(options.get(COUNTER_MAX));
             this.resetValue = Long.parseLong(options.get(RATE_RESET_VALUE));
-            log.trace("Setting rate counter options cm:{}, rv:{}", this.maxCounter, this.resetValue);
         }
+        if (options.containsKey(RATE_INTERVAL)) {
+            this.interval = Long.parseLong(options.get(RATE_INTERVAL));
+        }
+        log.trace("Setting rate options interval:{} isCounter:{} cm:{}, rv:{}", this.interval, this.isCounter, this.maxCounter, this.resetValue);
         Map<String,String> opts = new HashMap<>(options);
         opts.put(FILTER, "-1,1");
         super.init(source, opts, env);
@@ -64,9 +77,12 @@ public class RateIterator extends TimeSeriesGroupingIterator {
 
         if (isCounter && (secondVal < firstVal)) {
             if (maxCounter > 0) {
+                // assume that the data has reset so add maxCounter to the lower value to calculate the difference
                 secondVal += maxCounter;
                 log.trace("second counter reset based on max:{} to ts:{}, value:{}", maxCounter, secondTs, secondVal);
             } else {
+                // assume that the data has reset but no maxCounter set, so add the higher value
+                // (approximate maxCounter) to the lower value to calculate the difference
                 secondVal += firstVal;
                 log.trace("second counter reset based on first:{} to ts:{}, value:{}", firstVal, secondTs, secondVal);
             }
@@ -76,7 +92,9 @@ public class RateIterator extends TimeSeriesGroupingIterator {
         if (timeDiff == 0) {
             return 0.0D;
         }
-        Double result = ((secondVal - firstVal) / timeDiff);
+        // (secondVal - firstVal) / timeDiff will be value per millisecond, so multiply by interval to get
+        // value per interval. Example 60000 milliseconds in 1m, so multiply by 60000 milliseconds
+        Double result = ((secondVal - firstVal) / timeDiff * interval);
         log.trace("compute - result: {}", result);
 
         if (isCounter && resetValue > 0 && result > resetValue) {

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

@@ -700,6 +700,12 @@ public class DataStore {
                             if (query.isRate()) {
                                 log.trace("Adding rate iterator");
                                 IteratorSetting rate = new IteratorSetting(499, RateIterator.class);
+                                // if there is no rate interval set, then use the downsample value
+                                // so that the result is the change per downsample period
+                                RateOption rateOptions = query.getRateOptions();
+                                if (StringUtils.isBlank(rateOptions.getInterval())) {
+                                    rateOptions.setInterval(downsample + "ms");
+                                }
                                 RateIterator.setRateOptions(rate, query.getRateOptions());
                                 scanner.addScanIterator(rate);
                             }
@@ -946,7 +952,7 @@ public class DataStore {
             pattern.append(tag.getValue());
             pattern.append("(,.*|$)");
 
-            IteratorSetting setting = new IteratorSetting(priority++, tag.getKey() + " tag filter", RegExFilter.class);
+            IteratorSetting setting = new IteratorSetting(priority++, tag.getKey().replaceAll("\\.", "_") + " tag filter", RegExFilter.class);
             log.trace("Using {} additional filter on tag: {}", pattern, tag.getKey());
             RegExFilter.setRegexs(setting, null, null, pattern.toString(), null, false, true);
             scanner.addScanIterator(setting);

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

@@ -4,21 +4,17 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.ClientConfiguration;
-import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.Scanner;
 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.Range;
 import org.apache.accumulo.core.data.Value;
@@ -61,18 +57,14 @@ public final class MetaCache implements Iterable<Meta> {
 
     private void refreshCache(long expirationMinutes) {
         long oldestTimestamp = System.currentTimeMillis() - (expirationMinutes * 60 * 1000);
-        Map<String,String> properties = new HashMap<>();
-        properties.put("instance.name", accumuloProperties.getInstanceName());
-        properties.put("instance.zookeeper.host", zookeeperProperties.getServers());
-        final ClientConfiguration aConf = ClientConfiguration.fromMap(properties);
+        final Properties p = new Properties();
+        p.put("instance.name", accumuloProperties.getInstanceName());
+        p.put("instance.zookeeper.host", zookeeperProperties.getServers());
+
         String metaTable = timelyProperties.getMetaTable();
         Map<String,Map<String,Long>> metricMap = new HashMap<>();
-        Scanner scanner = null;
-        try {
+        try (AccumuloClient client = Accumulo.newClient().from(p).build(); Scanner scanner = client.createScanner(metaTable, Authorizations.EMPTY);) {
             Map<Meta,Object> newCache = new HashMap<>();
-            final Instance instance = new ZooKeeperInstance(aConf);
-            Connector connector = instance.getConnector(accumuloProperties.getUsername(), new PasswordToken(accumuloProperties.getPassword()));
-            scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
             log.debug("Begin scanning " + metaTable);
             Key metricPrefixBeginKey = new Key(Meta.METRIC_PREFIX);
             int firstChar = Meta.METRIC_PREFIX.charAt(0);
@@ -118,12 +110,8 @@ public final class MetaCache implements Iterable<Meta> {
                 cache.putAll(newCache);
             }
             log.debug("Finished scanning " + metaTable);
-        } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
+        } catch (TableNotFoundException e) {
             log.error(e.getMessage(), e);
-        } finally {
-            if (scanner != null) {
-                scanner.close();
-            }
         }
     }
 

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

@@ -28,6 +28,7 @@ import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.security.Authorizations;
 import org.apache.commons.lang3.SerializationUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.recipes.atomic.DistributedAtomicValue;
 import org.apache.curator.framework.recipes.cache.TreeCache;
@@ -765,7 +766,13 @@ public class DataStoreCache {
             if (subQuery.isRate()) {
                 log.trace("Adding rate iterator");
                 IteratorSetting rate = new IteratorSetting(499, RateIterator.class);
-                RateIterator.setRateOptions(rate, subQuery.getRateOptions());
+                // if there is no rate interval set, then use the downsample value
+                // so that the result is the change per downsample period
+                QueryRequest.RateOption rateOptions = subQuery.getRateOptions();
+                if (StringUtils.isBlank(rateOptions.getInterval())) {
+                    rateOptions.setInterval(downsamplePeriod + "ms");
+                }
+                RateIterator.setRateOptions(rate, rateOptions);
                 RateIterator rateIterator = new RateIterator();
                 rateIterator.init(itr, rate.getOptions(), null);
                 itr = rateIterator;

server/src/main/java/timely/server/store/compaction/MetricCompactionStrategy.java

@@ -1,187 +0,0 @@
-package timely.server.store.compaction;
-
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Optional;
-import java.util.OptionalLong;
-
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.data.TabletId;
-import org.apache.accumulo.core.metadata.StoredTabletFile;
-import org.apache.accumulo.tserver.compaction.CompactionPlan;
-import org.apache.accumulo.tserver.compaction.CompactionStrategy;
-import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
-import org.apache.commons.collections4.trie.PatriciaTrie;
-import org.apache.commons.configuration2.Configuration;
-import org.apache.commons.configuration2.MapConfiguration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import timely.server.store.MetricAgeOffIterator;
-
-public class MetricCompactionStrategy extends CompactionStrategy {
-
-    private static final Logger log = LoggerFactory.getLogger(MetricCompactionStrategy.class);
-
-    private boolean hasMinAgeOff;
-    private MetricAgeOffConfiguration minAgeOffConfig;
-    private boolean logOnly;
-    private String filterPrefix;
-    private String majcIteratorName;
-    private long explicitTimeMillis;
-    private boolean useExplicitTime;
-
-    public static final String LOG_ONLY_KEY = "logonly";
-    public static final String MAJC_ITERATOR_NAME_KEY = "majc.name";
-    public static final String DEFAULT_MAJC_ITERATOR_NAME = "ageoffmetrics";
-    public static final String AGE_OFF_PREFIX = MetricAgeOffIterator.AGE_OFF_PREFIX;
-    public static final String FILTER_PREFIX_KEY = AGE_OFF_PREFIX + "filter";
-    public static final String MIN_AGEOFF_KEY = AGE_OFF_PREFIX + "min";
-    public static final String DEFAULT_AGEOFF_KEY_SUFFIX = MetricAgeOffIterator.DEFAULT_AGEOFF_KEY;
-    public static final String DEFAULT_AGEOFF_KEY = AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY_SUFFIX;
-
-    @Override
-    public void init(Map<String,String> options) {
-        // options:
-        // logonly - disables compaction and will only log result
-        // majciterator - the iterator name to gather age-off values
-        // filter - limit the compaction to just a subset of prefixes
-        // min - minimum age-off which may override the metric age-off;
-        // allows more data to accumulate in tablets without being removed
-
-        logOnly = options.containsKey(LOG_ONLY_KEY) && Boolean.parseBoolean(options.get(LOG_ONLY_KEY));
-        majcIteratorName = options.getOrDefault(MAJC_ITERATOR_NAME_KEY, DEFAULT_MAJC_ITERATOR_NAME);
-        filterPrefix = options.get(FILTER_PREFIX_KEY);
-        if (options.containsKey(MIN_AGEOFF_KEY)) {
-            long minAgeOff = Long.parseLong(options.get(MIN_AGEOFF_KEY));
-            hasMinAgeOff = true;
-            minAgeOffConfig = MetricAgeOffConfiguration.newFromDefaultingMinimum(minAgeOff);
-        }
-    }
-
-    @Override
-    public boolean shouldCompact(MajorCompactionRequest request) {
-        TabletId tabletId = request.getTabletId();
-
-        if (null == tabletId.getEndRow() || null == tabletId.getPrevEndRow()) {
-            return false;
-        } else if (request.getFiles().isEmpty()) {
-            return false;
-        }
-
-        // check if prev/end key are both at age-off intervals
-        // skip tablet if there is no prev-end key, or previous age-off is not at max;
-        // other age-off will compact
-        Optional<String> endName = TabletRowAdapter.decodeRowPrefix(tabletId.getEndRow());
-        OptionalLong endOffset = TabletRowAdapter.decodeRowOffset(tabletId.getEndRow());
-        Optional<String> prevEndName = TabletRowAdapter.decodeRowPrefix(tabletId.getPrevEndRow());
-
-        if (!endName.isPresent() || !prevEndName.isPresent() || !endOffset.isPresent()) {
-            return false;
-        }
-
-        // compute age-off from majc iterator
-        // then maximize on the min-default-ageoff so that the
-        // configuration with largest age-off is used
-        MetricAgeOffConfiguration ageOffConfig = MetricAgeOffConfiguration.newFromRequest(request, majcIteratorName);
-        if (hasMinAgeOff) {
-            ageOffConfig = MetricAgeOffConfiguration.maximizeDefaultAgeOff(ageOffConfig, minAgeOffConfig);
-        }
-
-        long currentTime = useExplicitTime ? explicitTimeMillis : System.currentTimeMillis();
-        long ageOffComputed = currentTime - ageOffConfig.computeAgeOff(endName.get());
-        boolean shouldCompact = (ageOffComputed > endOffset.getAsLong() && endName.equals(prevEndName))
-                        && (null == filterPrefix || endName.get().startsWith(filterPrefix));
-
-        if (log.isDebugEnabled() && logOnly && shouldCompact) {
-            log.debug("Tablet will be metric age-off compacted {threshold: {}, endKey: {}, offset: {}, prevKey: {}}", ageOffComputed, endName,
-                            endOffset.getAsLong(), prevEndName);
-        } else if (log.isTraceEnabled()) {
-            log.trace("Tablet check metric compaction {threshold: {}, endKey: {}, offset: {}, prevKey: {}, result: {}}", ageOffComputed, endName,
-                            endOffset.getAsLong(), prevEndName, shouldCompact);
-        }
-
-        return shouldCompact && !logOnly;
-    }
-
-    @Override
-    public CompactionPlan getCompactionPlan(MajorCompactionRequest request) {
-        CompactionPlan plan = null;
-        if (shouldCompact(request)) {
-            plan = new CompactionPlan();
-            for (StoredTabletFile f : request.getFiles().keySet()) {
-                plan.inputFiles.add(f);
-            }
-
-            if (log.isDebugEnabled()) {
-                log.debug("Tablet selected for age-off: " + TabletRowAdapter.toDebugOutput(request.getTabletId()));
-            }
-        }
-
-        return plan;
-    }
-
-    public void setExplicitTime(long timeMillis) {
-        explicitTimeMillis = timeMillis;
-        useExplicitTime = true;
-    }
-
-    private static class MetricAgeOffConfiguration {
-
-        private final PatriciaTrie<Long> ageoffs;
-        private final long defaultAgeOff;
-
-        private MetricAgeOffConfiguration(PatriciaTrie<Long> ageoffs, long defaultAgeOff) {
-            this.ageoffs = ageoffs;
-            this.defaultAgeOff = defaultAgeOff;
-        }
-
-        public static MetricAgeOffConfiguration maximizeDefaultAgeOff(MetricAgeOffConfiguration x, MetricAgeOffConfiguration y) {
-            return (x.defaultAgeOff > y.defaultAgeOff) ? x : y;
-        }
-
-        public static MetricAgeOffConfiguration newFromDefaultingMinimum(long minAgeOff) {
-            return new MetricAgeOffConfiguration(new PatriciaTrie<>(), minAgeOff);
-        }
-
-        public static MetricAgeOffConfiguration newFromRequest(MajorCompactionRequest request, String majcIteratorName) {
-            Configuration config = new MapConfiguration(request.getTableProperties());
-            String majcIteratorKey = Property.TABLE_ITERATOR_MAJC_PREFIX.getKey() + majcIteratorName;
-
-            if (log.isTraceEnabled()) {
-                log.trace("Using key lookup for iterator: {}", majcIteratorKey);
-            }
-
-            Configuration configAgeOff = config.subset((majcIteratorKey + ".opt"));
-            if (null == configAgeOff.getString(DEFAULT_AGEOFF_KEY)) {
-                throw new IllegalArgumentException(DEFAULT_AGEOFF_KEY_SUFFIX + " must be configured for  " + majcIteratorKey);
-            }
-
-            PatriciaTrie<Long> ageoffs = new PatriciaTrie<>();
-            long configureMinAgeOff = Long.MAX_VALUE;
-            long configureMaxAgeOff = Long.MIN_VALUE;
-            @SuppressWarnings("unchecked")
-            Iterator<String> keys = configAgeOff.getKeys();
-            while (keys.hasNext()) {
-                String k = keys.next();
-                String v = configAgeOff.getString(k);
-                if (k.startsWith((AGE_OFF_PREFIX))) {
-                    String name = k.substring(AGE_OFF_PREFIX.length());
-                    if (log.isTraceEnabled()) {
-                        log.trace("Adding {} to Trie with value {}", name, Long.parseLong(v));
-                    }
-                    long ageoff = Long.parseLong(v);
-                    configureMinAgeOff = Math.min(configureMinAgeOff, ageoff);
-                    configureMaxAgeOff = Math.max(configureMaxAgeOff, ageoff);
-                    ageoffs.put(name, ageoff);
-                }
-            }
-            long defaultAgeOff = ageoffs.get(DEFAULT_AGEOFF_KEY_SUFFIX);
-            return new MetricAgeOffConfiguration(ageoffs, defaultAgeOff);
-        }
-
-        public long computeAgeOff(String metricName) {
-            return ageoffs.getOrDefault(metricName, defaultAgeOff);
-        }
-    }
-}

server/src/main/java/timely/server/store/compaction/TieredCompactionStrategy.java

@@ -1,156 +0,0 @@
-package timely.server.store.compaction;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
-import org.apache.accumulo.tserver.compaction.CompactionPlan;
-import org.apache.accumulo.tserver.compaction.CompactionStrategy;
-import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TieredCompactionStrategy extends CompactionStrategy {
-
-    public static final String TIERED_PREFIX = "tier.";
-    public static final String TIERED_CLASS = "class";
-    public static final String TIERED_OPTS = "opts";
-
-    protected final Collection<TieredCompactorSupplier> compactors;
-
-    public TieredCompactionStrategy() {
-        compactors = new ArrayList<>();
-    }
-
-    public Collection<TieredCompactorSupplier> getCompactors() {
-        return Collections.unmodifiableCollection(compactors);
-    }
-
-    @Override
-    public void init(Map<String,String> config) {
-        // example:
-        // <table-compaction-prefix>.tiered.0.class=<majcompaction-class>
-        // <table-compaction-prefix>.tiered.0.opts.key=value
-        // <table-compaction-prefix>.tiered.1.class=<majcompaction-class>
-        // <table-compaction-prefix>.tiered.1.opts.key_a=value
-        // <table-compaction-prefix>.tiered.1.opts.key_b=value
-        //
-        // the key=value pairs will be sent to the specific tiered compaction
-        // strategy as arguments
-
-        compactors.clear();
-        createSuppliers(config);
-    }
-
-    @Override
-    public void gatherInformation(MajorCompactionRequest request) throws IOException {
-        for (TieredCompactorSupplier compactor : compactors) {
-            CompactionStrategy strategy = compactor.get(request.getTableProperties());
-            strategy.gatherInformation(request);
-        }
-    }
-
-    @Override
-    public boolean shouldCompact(MajorCompactionRequest request) throws IOException {
-        boolean shouldCompact = false;
-        for (TieredCompactorSupplier supplier : compactors) {
-            CompactionStrategy strategy = supplier.get(request.getTableProperties());
-            shouldCompact = strategy.shouldCompact(request);
-            if (shouldCompact) {
-                break;
-            }
-        }
-
-        return shouldCompact;
-    }
-
-    @Override
-    public CompactionPlan getCompactionPlan(MajorCompactionRequest request) throws IOException {
-        CompactionPlan plan = null;
-        Iterator<TieredCompactorSupplier> iter = compactors.iterator();
-        while ((null == plan || plan.inputFiles.isEmpty()) && iter.hasNext()) {
-            CompactionStrategy strategy = iter.next().get(request.getTableProperties());
-            plan = strategy.getCompactionPlan(request);
-        }
-
-        return plan;
-    }
-
-    protected void createSuppliers(Map<String,String> config) {
-        int tier = 0;
-        boolean continueLoop = true;
-        while (continueLoop) {
-            String clazzKey = TIERED_PREFIX + tier + "." + TIERED_CLASS;
-            String clazzName = config.get(clazzKey);
-            if (null != clazzName) {
-                compactors.add(new TieredConfiguredSupplier(clazzName, tier, config));
-                tier++;
-            } else {
-                continueLoop = false;
-            }
-        }
-    }
-
-    static class TieredConfiguredSupplier implements TieredCompactorSupplier {
-
-        private static final Logger log = LoggerFactory.getLogger(TieredConfiguredSupplier.class);
-
-        private final String clazzName;
-        private final int tier;
-        private final Map<String,String> initConfig;
-
-        private CompactionStrategy strategy;
-        private Map<String,String> options;
-
-        TieredConfiguredSupplier(String clazzName, int tier, Map<String,String> initConfig) {
-            this.strategy = null;
-            this.clazzName = clazzName;
-            this.tier = tier;
-            this.initConfig = initConfig;
-        }
-
-        public Map<String,String> options() {
-            if (null == options) {
-                options = new HashMap<>();
-                String tierCompare = Integer.toString(tier);
-                int tierIdx = TIERED_PREFIX.length();
-                int optsIdx = TIERED_PREFIX.length() + tierCompare.length() + 1;
-                for (Map.Entry<String,String> entry : initConfig.entrySet()) {
-                    String key = entry.getKey();
-                    if (key.regionMatches(tierIdx, tierCompare, 0, tierCompare.length()) && key.regionMatches(optsIdx, TIERED_OPTS, 0, TIERED_OPTS.length())) {
-                        options.put(key.substring(optsIdx + TIERED_OPTS.length() + 1), entry.getValue());
-                    }
-                }
-            }
-
-            return Collections.unmodifiableMap(options);
-        }
-
-        public CompactionStrategy get(Map<String,String> tableProperties) {
-            if (null == strategy) {
-                try {
-                    String tableContext = tableProperties.get(Property.TABLE_CLASSPATH.getKey());
-                    Class<? extends CompactionStrategy> clazz;
-                    if (null != tableContext && !tableContext.equals("")) {
-                        clazz = AccumuloVFSClassLoader.getContextClassLoader(tableContext).loadClass(clazzName).asSubclass(CompactionStrategy.class);
-                    } else {
-                        clazz = AccumuloVFSClassLoader.getClassLoader().loadClass(clazzName).asSubclass(CompactionStrategy.class);
-                    }
-                    strategy = clazz.newInstance();
-                    strategy.init(options());
-                } catch (Exception e) {
-                    log.debug("Failed to load class for tiered compaction: " + e.toString());
-                    throw new RuntimeException(e);
-                }
-            }
-
-            return strategy;
-        }
-    }
-}

server/src/main/java/timely/server/store/compaction/TieredCompactorSupplier.java

@@ -1,12 +0,0 @@
-package timely.server.store.compaction;
-
-import java.util.Map;
-
-import org.apache.accumulo.tserver.compaction.CompactionStrategy;
-
-public interface TieredCompactorSupplier {
-
-    Map<String,String> options();
-
-    CompactionStrategy get(Map<String,String> tableProperties);
-}

server/src/main/java/timely/server/store/compaction/util/TabletMetadataQuery.java

@@ -10,7 +10,7 @@ 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;
-import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.SystemTables;
 
 import timely.common.configuration.TimelyProperties;
 import timely.server.configuration.TabletMetadataProperties;
@@ -32,10 +32,11 @@ public class TabletMetadataQuery {
         Map<String,String> tableMap = accumuloClient.tableOperations().tableIdMap();
         String tableId = tableMap.get(timelyProperties.getMetricsTable());
         if (null == tableId) {
-            throw new IllegalStateException("Unable to find " + MetadataTable.NAME);
+            throw new IllegalStateException("Unable to find " + SystemTables.METADATA.tableName());
         }
 
-        try (Scanner s = accumuloClient.createScanner(MetadataTable.NAME, accumuloClient.securityOperations().getUserAuthorizations(accumuloClient.whoami()))) {
+        try (Scanner s = accumuloClient.createScanner(SystemTables.METADATA.tableName(),
+                        accumuloClient.securityOperations().getUserAuthorizations(accumuloClient.whoami()))) {
             s.setRange(Range.prefix(tableId));
             for (Map.Entry<Key,Value> e : s) {
                 tabletMetadataView.addEntry(e);

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

@@ -189,10 +189,8 @@ public class DataStoreCacheIteratorIT extends ITBase {
         subQuery.setDownsample(Optional.of("1ms-avg"));
         subQuery.setMetric("metric.number.1");
         subQuery.addTag("host", ".*");
-        QueryRequest.RateOption rateOption = new QueryRequest.RateOption();
-        rateOption.setCounter(false);
         subQuery.setRate(true);
-        subQuery.setRateOptions(rateOption);
+        subQuery.getRateOptions().setCounter(false);
         query.setQueries(Collections.singleton(subQuery));
 
         SortedKeyValueIterator<org.apache.accumulo.core.data.Key,org.apache.accumulo.core.data.Value> itr;

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

@@ -702,25 +702,20 @@ public class HttpApiIT extends OneWaySSLBase {
         assertEquals(1.0, entry2.getValue());
     }
 
+    private void addRateData(String metric, String tagString, long testStartTime, long timeIncrement, long testStopTime, double startValue, double maxValue,
+                    double valueIncrement) throws Exception {
+        long ts = testStartTime;
+        double value = startValue;
+        while (ts <= testStopTime) {
+            put(metric + " " + ts + " " + value + " " + tagString);
+            ts += timeIncrement;
+            value = (value >= maxValue) ? startValue : (value + valueIncrement);
+        }
+    }
+
     @Test
     public void testRateQuery() throws Exception {
-        // @formatter:off
-        put("sys.cpu.user " + (TEST_TIME) + " 1.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 1) + " 2.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 2) + " 3.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 3) + " 4.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 4) + " 5.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 5) + " 6.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 6) + " 7.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 7) + " 8.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 8) + " 9.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 9) + " 10.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 10) + " 11.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 11) + " 12.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 12) + " 13.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 13) + " 14.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME + 14) + " 15.0 tag1=value1 tag2=value2 rack=r1");
-        // @formatter:on
+        addRateData("sys.cpu.user", "tag1=value1 tag2=value2 rack=r1", TEST_TIME, 1, TEST_TIME + 14, 1.0, Double.MAX_VALUE, 1.0);
         dataStore.flush();
         dataStoreCache.flushCaches(-1);
         // Latency in TestConfiguration is 2s, wait for it
@@ -751,23 +746,7 @@ public class HttpApiIT extends OneWaySSLBase {
 
     @Test
     public void testRateCounterQuery() throws Exception {
-      // @formatter:off
-        put("sys.cpu.user " + (TEST_TIME) + " 1.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+1) + " 2.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+2) + " 3.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+3) + " 4.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+4) + " 5.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+5) + " 6.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+6) + " 7.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+7) + " 8.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+8) + " 9.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+9) + " 10.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+10) + " 1.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+11) + " 2.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+12) + " 3.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+13) + " 4.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+14) + " 5.0 tag1=value1 tag2=value2 rack=r1");
-        // @formatter:on
+        addRateData("sys.cpu.user", "tag1=value1 tag2=value2 rack=r1", TEST_TIME, 1, TEST_TIME + 14, 1.0, 10.0, 1.0);
         dataStore.flush();
         dataStoreCache.flushCaches(-1);
         // Latency in TestConfiguration is 2s, wait for it
@@ -782,7 +761,6 @@ public class HttpApiIT extends OneWaySSLBase {
         subQuery.setTags(t);
         subQuery.setDownsample(Optional.of("1ms-max"));
         subQuery.setRate(true);
-        subQuery.setRateOptions(new QueryRequest.RateOption());
         subQuery.getRateOptions().setCounter(true);
         request.addQuery(subQuery);
         List<QueryResponse> response = query(baseUrl + "/api/query", request);
@@ -799,24 +777,8 @@ public class HttpApiIT extends OneWaySSLBase {
     }
 
     @Test
-    public void testRateCounterWithOptions() throws Exception {
-      // @formatter:off
-        put("sys.cpu.user " + (TEST_TIME) + " 1.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+1) + " 2.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+2) + " 3.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+3) + " 4.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+4) + " 5.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+5) + " 6.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+6) + " 7.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+7) + " 8.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+8) + " 9.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+9) + " 10.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+10) + " 0.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+11) + " 1.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+12) + " 2.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+13) + " 3.0 tag1=value1 tag2=value2 rack=r1",
-                "sys.cpu.user " + (TEST_TIME+14) + " 4.0 tag1=value1 tag2=value2 rack=r1");
-        // @formatter:on
+    public void testRateCounterWithCounterMaxOption() throws Exception {
+        addRateData("sys.cpu.user", "tag1=value1 tag2=value2 rack=r1", TEST_TIME, 1, TEST_TIME + 14, 1.0, 10.0, 1.0);
         dataStore.flush();
         dataStoreCache.flushCaches(-1);
         // Latency in TestConfiguration is 2s, wait for it
@@ -831,9 +793,8 @@ public class HttpApiIT extends OneWaySSLBase {
         subQuery.setTags(t);
         subQuery.setDownsample(Optional.of("1ms-max"));
         subQuery.setRate(true);
-        subQuery.setRateOptions(new QueryRequest.RateOption());
         subQuery.getRateOptions().setCounter(true);
-        subQuery.getRateOptions().setCounterMax(Long.MAX_VALUE);
+        subQuery.getRateOptions().setCounterMax(10);
         subQuery.getRateOptions().setResetValue(2);
         request.addQuery(subQuery);
         List<QueryResponse> response = query(baseUrl + "/api/query", request);
@@ -844,7 +805,54 @@ public class HttpApiIT extends OneWaySSLBase {
         int i = 1;
         for (Entry<String,Object> e : dps.entrySet()) {
             assertEquals(Long.toString(TEST_TIME + i), e.getKey());
-            assertEquals((i == 10 ? 0.0 : 1.0), e.getValue());
+            assertEquals(1.0, e.getValue());
+            i++;
+        }
+    }
+
+    @Test
+    public void testRateCounterWithRateIntervalOption() throws Exception {
+        Long TEST_TIME = (System.currentTimeMillis() / 60000) * 60000;
+        addRateData("sys.cpu.user", "tag1=value1 tag2=value2 rack=r1", TEST_TIME, 60000, TEST_TIME + 600000, 1.0, Double.MAX_VALUE, 5);
+        dataStore.flush();
+        dataStoreCache.flushCaches(-1);
+        // Latency in TestConfiguration is 2s, wait for it
+        sleepUninterruptibly(TestConfiguration.WAIT_SECONDS, TimeUnit.SECONDS);
+        QueryRequest request = new QueryRequest();
+        request.setStart(TEST_TIME);
+        request.setEnd(TEST_TIME + 600000);
+        request.setMsResolution(true);
+        SubQuery subQuery = new SubQuery();
+        subQuery.setMetric("sys.cpu.user");
+        subQuery.setDownsample(Optional.of("1m-max"));
+        subQuery.setRate(true);
+        subQuery.getRateOptions().setInterval("2m");
+        request.addQuery(subQuery);
+
+        // test interval (2m) that is different from the downsample period (1m)
+        List<QueryResponse> response = query(baseUrl + "/api/query", request);
+        assertEquals(1, response.size());
+        QueryResponse response1 = response.get(0);
+        Map<String,Object> dps = response1.getDps();
+        assertEquals(10, dps.size());
+        int i = 1;
+        for (Entry<String,Object> e : dps.entrySet()) {
+            assertEquals(Long.toString(TEST_TIME + (60000 * i)), e.getKey());
+            assertEquals(10.0, e.getValue());
+            i++;
+        }
+
+        // test that interval defaults to downsample period
+        subQuery.getRateOptions().setInterval(null);
+        response = query(baseUrl + "/api/query", request);
+        assertEquals(1, response.size());
+        response1 = response.get(0);
+        dps = response1.getDps();
+        assertEquals(10, dps.size());
+        i = 1;
+        for (Entry<String,Object> e : dps.entrySet()) {
+            assertEquals(Long.toString(TEST_TIME + (60000 * i)), e.getKey());
+            assertEquals(5.0, e.getValue());
             i++;
         }
     }

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

@@ -499,7 +499,7 @@ public class HttpRequestDecoderIT extends ITBase {
     public void testQueryURIAllAnonAccess() throws Exception {
         decoder = new HttpRequestDecoder(authenticationService, anonymousSecurity, httpProperties);
         DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET,
-                        "/api/query?start=1356998400&end=1356998460&m=sum:rate{false,100,0}:sys.cpu.user{host=*}{rack=r1|r2}&tsuid=sum:000001000002000042,000001000002000043");
+                        "/api/query?start=1356998400&end=1356998460&m=sum:rate{false,1s,100,0}:sys.cpu.user{host=*}{rack=r1|r2}&tsuid=sum:000001000002000042,000001000002000043");
         decoder.decode(null, request, results);
         Assert.assertEquals(1, results.size());
         Assert.assertEquals(QueryRequest.class, results.iterator().next().getClass());
@@ -515,6 +515,7 @@ public class HttpRequestDecoderIT extends ITBase {
         Assert.assertEquals("sys.cpu.user", first.getMetric());
         Assert.assertEquals(true, first.isRate());
         RateOption firstRateOption = first.getRateOptions();
+        Assert.assertEquals("1s", firstRateOption.getInterval());
         Assert.assertEquals(false, firstRateOption.isCounter());
         Assert.assertEquals(100, firstRateOption.getCounterMax());
         Assert.assertEquals(0, firstRateOption.getResetValue());
@@ -545,7 +546,7 @@ public class HttpRequestDecoderIT extends ITBase {
     public void testQueryURIAllWithSession() throws Exception {
         decoder = new HttpRequestDecoder(authenticationService, requireUserSecurity, httpProperties);
         DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET,
-                        "/api/query?start=1356998400&end=1356998460&m=sum:rate{false,100,0}:sys.cpu.user{host=*}{rack=r1|r2}&tsuid=sum:000001000002000042,000001000002000043");
+                        "/api/query?start=1356998400&end=1356998460&m=sum:rate{false,1s,100,0}:sys.cpu.user{host=*}{rack=r1|r2}&tsuid=sum:000001000002000042,000001000002000043");
         addCookie(request);
         decoder.decode(null, request, results);
         Assert.assertEquals(1, results.size());
@@ -562,6 +563,7 @@ public class HttpRequestDecoderIT extends ITBase {
         Assert.assertEquals("sys.cpu.user", first.getMetric());
         Assert.assertEquals(true, first.isRate());
         RateOption firstRateOption = first.getRateOptions();
+        Assert.assertEquals("1s", firstRateOption.getInterval());
         Assert.assertEquals(false, firstRateOption.isCounter());
         Assert.assertEquals(100, firstRateOption.getCounterMax());
         Assert.assertEquals(0, firstRateOption.getResetValue());

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

@@ -185,7 +185,7 @@ public class ITBase {
         TestDataStore.StoreCallback storeCallback = () -> PUT_REQUESTS.countDown();
         try {
             dataStore.addStoreCallback(storeCallback);
-            StringBuffer format = new StringBuffer();
+            StringBuilder format = new StringBuilder();
             for (String line : lines) {
                 format.append("put ");
                 format.append(line);

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

@@ -1,250 +0,0 @@
-package timely.server.integration;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.util.Collection;
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import org.apache.accumulo.core.client.AccumuloClient;
-import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.iterators.IteratorUtil;
-import org.apache.accumulo.tserver.compaction.CompactionStrategy;
-import org.apache.accumulo.tserver.compaction.DefaultCompactionStrategy;
-import org.apache.hadoop.io.Text;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.ActiveProfiles;
-import org.springframework.test.context.junit4.SpringRunner;
-
-import com.google.common.collect.Sets;
-
-import io.jsonwebtoken.lang.Assert;
-import timely.server.store.MetricAgeOffIterator;
-import timely.server.store.compaction.MetricCompactionStrategy;
-import timely.server.store.compaction.TabletRowAdapter;
-import timely.server.store.compaction.util.TabletMetadataQuery;
-import timely.server.store.compaction.util.TabletMetadataView;
-import timely.server.store.compaction.util.TabletStatistic;
-import timely.server.store.compaction.util.TabletSummary;
-import timely.server.test.TestMetricWriter;
-import timely.server.test.TestMetrics;
-import timely.test.IntegrationTest;
-import timely.test.TimelyTestRule;
-
-@Category(IntegrationTest.class)
-@RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
-@ActiveProfiles({"miniAccumulo"})
-public class MetricCompactionIT extends ITBase {
-
-    private static final String AGE_OFF_ITERATOR_NAME = "ageoffmetrics";
-    private static final EnumSet<IteratorUtil.IteratorScope> AGEOFF_SCOPES = EnumSet.allOf(IteratorUtil.IteratorScope.class);
-
-    @Autowired
-    @Rule
-    public TimelyTestRule testRule;
-
-    private MetricConfigurationAdapter adapter;
-
-    @Before
-    public void setup() {
-        super.setup();
-        try {
-            accumuloClient.instanceOperations().setProperty(Property.TSERV_MAJC_DELAY.getKey(), "3s");
-            adapter = new MetricConfigurationAdapter(accumuloClient, timelyProperties.getMetricsTable());
-            accumuloClient.tableOperations().setProperty(timelyProperties.getMetricsTable(), Property.TABLE_SPLIT_THRESHOLD.getKey(), "50K");
-            adapter.resetState();
-        } catch (Exception e) {
-            log.error(e.getMessage(), e);
-        }
-    }
-
-    @After
-    public void cleanup() {
-        super.cleanup();
-        deleteTable(timelyProperties.getMetricsTable());
-        deleteTable(timelyProperties.getMetaTable());
-    }
-
-    @After
-    public void teardown() throws Exception {
-        accumuloClient.instanceOperations().setProperty(Property.TSERV_MAJC_DELAY.getKey(), "30s");
-        adapter.resetState();
-    }
-
-    @Test
-    public void tableMetadataEnumeratesTablets() throws Exception {
-        String metricsTable = timelyProperties.getMetricsTable();
-
-        TestMetricWriter writer = new TestMetricWriter(metricsTable, accumuloClient, new TestMetrics());
-        writer.ingestRandomDuration(60, TimeUnit.SECONDS, System.currentTimeMillis(), 250);
-
-        Collection<Text> splits = accumuloClient.tableOperations().listSplits(timelyProperties.getMetricsTable());
-        TabletMetadataQuery query = new TabletMetadataQuery(accumuloClient, timelyProperties, tabletMetadataProperties);
-        TabletMetadataView view = query.run();
-
-        // @formatter:off
-        TabletSummary summary = view.computeSummary()
-                .disableTabletRowCheckFilter()
-                .build();
-
-        Set<String> splitsPrefixed = splits.stream()
-                .map(row -> TabletRowAdapter.decodeRowPrefix(row).orElse("unknown"))
-                .collect(Collectors.toSet());
-
-        Set<String> tabletsPrefixed = summary.getSummary().stream()
-                .map(TabletStatistic::getKeyName)
-                .collect(Collectors.toSet());
-
-        // difference both sides
-        // should be empty
-        // @formatter:on
-        Set<String> i1 = Sets.difference(tabletsPrefixed, splitsPrefixed);
-        Set<String> i2 = Sets.difference(splitsPrefixed, tabletsPrefixed);
-        assertTrue("tablets-diff" + i1, i1.isEmpty());
-        assertTrue("splits-diff" + i2, i2.isEmpty());
-
-        // dump data to screen as summary
-        Assert.notNull(view.toText());
-    }
-
-    @Test
-    public void tabletsCompactedUntilAgeOff() throws Exception {
-        // test will replicate problem in time-series tablets where data is
-        // split automatically and the resulting tablets will not be cleaned up
-        // by the default compaction strategy
-        //
-        // metric compaction strategy will age-off tablets by date on tablets
-        // but ignore differing end/prev rows
-        //
-        // the mini-accumulo cluster should not have any age-off
-        // iterators on timely.metrics
-
-        String metricsTable = timelyProperties.getMetricsTable();
-
-        // check to make sure the table has two iterators
-        Map<String,EnumSet<IteratorUtil.IteratorScope>> itrs = accumuloClient.tableOperations().listIterators(metricsTable);
-        assertEquals(1, itrs.size());
-        assertTrue(itrs.containsKey("vers"));
-
-        long timestampMax = System.currentTimeMillis();
-        TestMetricWriter writer = new TestMetricWriter(timelyProperties.getMetricsTable(), accumuloClient, new TestMetrics());
-        long timestampMin = writer.ingestRandomDuration(30, TimeUnit.SECONDS, timestampMax, 250);
-
-        // invoke compact operation to flush everything through
-        // without this call the compaction strategy later on
-        // was not compacting all of the tablets
-        adapter.runCompaction();
-
-        // figure out what should be targeted based on an ageoff date
-        // cut the delta by 1/3
-        long ageoff = (timestampMax - timestampMin) / 3;
-
-        // run check to verify there are pre-existing splits for a prefix
-        // @formatter:off
-        TabletMetadataQuery query = new TabletMetadataQuery(accumuloClient, timelyProperties, tabletMetadataProperties);
-        TabletSummary summary1 = query.run()
-                .computeSummary()
-                .build();
-
-        Set<String> tabletsBefore = summary1.getSummary().stream()
-                .filter(t -> t.maxAge() > ageoff)
-                .map(TabletStatistic::getKeyName)
-                .collect(Collectors.toSet());
-
-        // @formatter:on
-        assertFalse("Tablets were empty before compaction, expected test data", tabletsBefore.isEmpty());
-
-        long compactBeforeMillis = System.currentTimeMillis();
-
-        // invoke the age-off compactor
-        // after this runs it should have gotten rid o fdata
-        // have pre-age-off tablets
-        adapter.applyCompactionConfiguration(MetricCompactionStrategy.class, ageoff);
-        adapter.runCompaction();
-
-        long compactDeltaMillis = System.currentTimeMillis() - compactBeforeMillis;
-
-        // re-run check on tablet metadata
-        // check that there aren't new tablets past the predicted age-off
-        // any tablets with differing end/prev rows are filtered by the summary
-        // @formatter:off
-        TabletSummary summary2 = query.run()
-                .computeSummary()
-                .build();
-
-        List<TabletStatistic> tabletsFilterAfter = summary2.getSummary().stream()
-                .filter(t -> ((t.maxAge() - compactDeltaMillis) > ageoff) && t.getKeyCount() > 1)
-                .collect(Collectors.toList());
-
-        // @formatter:on
-        if (!tabletsFilterAfter.isEmpty()) {
-            log.debug("tablets-before: {}", tabletsBefore.toString());
-            log.debug("tablets-after: {}", tabletsFilterAfter.toString());
-        }
-
-        // should not have found tablets past the age-off and should also
-        // have found existing tablets still in system (i.e. did not compact everything)
-        assertTrue("Tablets still had unexpected after compaction", tabletsFilterAfter.isEmpty());
-        assertTrue("Some tablets should exist, but did not find any", summary2.totalTablets() > 0);
-    }
-
-    private static class MetricConfigurationAdapter {
-
-        private final AccumuloClient accumuloClient;
-        private final String tableName;
-
-        private static String[] CLEAR_PROPERTY_KEYS = {Property.TABLE_COMPACTION_STRATEGY.getKey() + MetricCompactionStrategy.MIN_AGEOFF_KEY};
-
-        public MetricConfigurationAdapter(AccumuloClient accumuloClient, String tableName) {
-            this.accumuloClient = accumuloClient;
-            this.tableName = tableName;
-        }
-
-        @SuppressWarnings("removal")
-        public void applyCompactionConfiguration(Class<? extends CompactionStrategy> clazz, long ageOff) throws Exception {
-            Map<String,String> ageOffs = new HashMap<>();
-            ageOffs.put("ageoff.default", Long.toString(ageOff));
-            IteratorSetting ageOffIteratorSettings = new IteratorSetting(100, AGE_OFF_ITERATOR_NAME, MetricAgeOffIterator.class, ageOffs);
-            accumuloClient.tableOperations().removeIterator(tableName, AGE_OFF_ITERATOR_NAME, AGEOFF_SCOPES);
-            accumuloClient.tableOperations().attachIterator(tableName, ageOffIteratorSettings, AGEOFF_SCOPES);
-            accumuloClient.tableOperations().setProperty(tableName, Property.TABLE_COMPACTION_STRATEGY.getKey(), clazz.getName());
-        }
-
-        public void runCompaction() throws Exception {
-            accumuloClient.tableOperations().compact(tableName, null, null, true, true);
-        }
-
-        public void resetState() throws Exception {
-            Map<String,EnumSet<IteratorUtil.IteratorScope>> iters = accumuloClient.tableOperations().listIterators(tableName);
-            for (String name : iters.keySet()) {
-                if (name.startsWith("ageoff")) {
-                    accumuloClient.tableOperations().removeIterator(tableName, name, AGEOFF_SCOPES);
-                }
-            }
-
-            // reset compaction strategy to default
-            accumuloClient.tableOperations().setProperty(tableName, Property.TABLE_COMPACTION_STRATEGY.getKey(), DefaultCompactionStrategy.class.getName());
-
-            // reset any properties
-            for (String s : CLEAR_PROPERTY_KEYS) {
-                accumuloClient.tableOperations().removeProperty(tableName, s);
-            }
-        }
-    }
-}

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

@@ -31,6 +31,9 @@ public class DownsampleIteratorTest {
     final private SortedMap<Key,Value> testData1 = new TreeMap<>();
     final private SortedMap<Key,Value> testData2 = new TreeMap<>();
 
+    private long now = System.currentTimeMillis();
+    private long startingTimestamp = now - (now % 100);
+
     @Before
     public void createTestData() {
         createTestData1();
@@ -41,7 +44,7 @@ public class DownsampleIteratorTest {
         List<Tag> tags = Collections.singletonList(new Tag("host", "host1"));
 
         for (long i = 0; i < 1000; i += 100) {
-            Metric m = new Metric("sys.loadAvg", i, .2, tags);
+            Metric m = new Metric("sys.loadAvg", startingTimestamp + i, .2, tags);
             Mutation mutation = MetricAdapter.toMutation(m);
             for (ColumnUpdate cu : mutation.getUpdates()) {
                 Key key = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(), cu.getColumnVisibility(), cu.getTimestamp());
@@ -63,8 +66,8 @@ public class DownsampleIteratorTest {
         List<Tag> tags2 = Collections.singletonList(new Tag("host", "host2"));
 
         for (long i = 0; i < 1000; i += 100) {
-            put(testData2, new Metric("sys.loadAvg", i, .2, tags));
-            put(testData2, new Metric("sys.loadAvg", i + 50, .5, tags2));
+            put(testData2, new Metric("sys.loadAvg", startingTimestamp + i, .2, tags));
+            put(testData2, new Metric("sys.loadAvg", startingTimestamp + i + 50, .5, tags2));
         }
     }
 
@@ -78,13 +81,14 @@ public class DownsampleIteratorTest {
             Set<Tag> tags = entry.getKey();
             assertEquals(1, tags.size());
             assertEquals(Collections.singleton(new Tag("host", "host1")), tags);
-            long ts = 0;
+
+            long ts = startingTimestamp - (startingTimestamp % 100);
             for (Sample sample : entry.getValue()) {
                 assertEquals(ts, sample.getTimestamp());
                 ts += 100;
                 assertEquals(0.2, sample.getValue(), 0.0001);
             }
-            assertEquals(1000, ts);
+            assertEquals(startingTimestamp - (startingTimestamp % 100) + 1000, ts);
         }
     }
 
@@ -96,7 +100,7 @@ public class DownsampleIteratorTest {
         for (Tag tag : new Tag[] {new Tag("host", "host1"), new Tag("host", "host2")}) {
             Downsample dsample = samples.get(Collections.singleton(tag));
             assertNotNull(dsample);
-            long ts = 0;
+            long ts = startingTimestamp - (startingTimestamp % 100);
             double value = .2;
             if (tag.getValue().equals("host2")) {
                 value = .5;
@@ -120,7 +124,7 @@ public class DownsampleIteratorTest {
         for (Tag tag : new Tag[] {new Tag("host", "host1"), new Tag("host", "host2")}) {
             Downsample dsample = samples.get(Collections.singleton(tag));
             assertNotNull(dsample);
-            long ts = 0;
+            long ts = startingTimestamp - (startingTimestamp % 200);
             double value = .2;
             if (tag.getValue().equals("host2")) {
                 value = .5;
@@ -216,7 +220,7 @@ public class DownsampleIteratorTest {
 
         for (long i = 0; i < elapsedTime; i += skipInterval) {
             for (List<Tag> tags : listOfTagVariations) {
-                put(testData3, new Metric("sys.loadAvg", i, i * 2, tags));
+                put(testData3, new Metric("sys.loadAvg", startingTimestamp + i, i * 2, tags));
             }
         }
         return testData3;

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

@@ -1,22 +1,17 @@
 package timely.server.sample.iterators;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 
+import org.apache.accumulo.core.client.PluginEnvironment;
 import org.apache.accumulo.core.client.sample.SamplerConfiguration;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.DefaultConfiguration;
 import org.apache.accumulo.core.data.ByteSequence;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
-import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
-import org.apache.accumulo.core.iteratorsImpl.system.MapFileIterator;
 import org.apache.accumulo.core.security.Authorizations;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
 
 public class IteratorTestBase {
 
@@ -32,18 +27,6 @@ public class IteratorTestBase {
             this.conf = DefaultConfiguration.getInstance();
         }
 
-        @Override
-        public SortedKeyValueIterator<Key,Value> reserveMapFileReader(String mapFileName) throws IOException {
-            Configuration conf = new Configuration();
-            FileSystem fs = FileSystem.get(conf);
-            return new MapFileIterator(fs, mapFileName, conf);
-        }
-
-        @Override
-        public AccumuloConfiguration getConfig() {
-            return conf;
-        }
-
         @Override
         public IteratorScope getIteratorScope() {
             throw new UnsupportedOperationException();
@@ -54,11 +37,6 @@ public class IteratorTestBase {
             throw new UnsupportedOperationException();
         }
 
-        @Override
-        public void registerSideChannel(SortedKeyValueIterator<Key,Value> iter) {
-            throw new UnsupportedOperationException();
-        }
-
         @Override
         public Authorizations getAuthorizations() {
             throw new UnsupportedOperationException();
@@ -78,6 +56,26 @@ public class IteratorTestBase {
         public boolean isSamplingEnabled() {
             return false;
         }
+
+        @Override
+        public boolean isUserCompaction() {
+            return false;
+        }
+
+        @Override
+        public PluginEnvironment getPluginEnv() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public TableId getTableId() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean isRunningLowOnMemory() {
+            return false;
+        }
     }
 
     protected static class CombinerIteratorEnvironment extends DefaultIteratorEnvironment {

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

@@ -46,14 +46,26 @@ public class RateIteratorTest extends IteratorTestBase {
 
     @Test
     public void testConstantTimeRate() throws Exception {
+        testInterval(null, 1.0D);
+        testInterval("1ms", 0.001D);
+        testInterval("1s", 1.0D);
+        testInterval("60s", 60.0D);
+    }
+
+    private void testInterval(String interval, double result) throws Exception {
         SortedMapIterator source = new SortedMapIterator(table);
         RateIterator iter = new RateIterator();
         IteratorSetting settings = new IteratorSetting(100, RateIterator.class);
+        QueryRequest.RateOption rateOption = new QueryRequest.RateOption();
+        if (interval != null) {
+            rateOption.setInterval(interval);
+        }
+        RateIterator.setRateOptions(settings, rateOption);
         iter.init(source, settings.getOptions(), SCAN_IE);
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
         for (int i = 0; i < 99; i++) {
             assertTrue(iter.hasTop());
-            assertEquals(0.001D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
+            assertEquals(result, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
             iter.next();
         }
         assertFalse(iter.hasTop());
@@ -83,7 +95,7 @@ public class RateIteratorTest extends IteratorTestBase {
             Value v = source.getTopValue();
             if (prevTs != -1L) {
                 Double thisValue = MetricAdapter.decodeValue(v.get());
-                expected.add((thisValue + (prevValue * -1)) / (k.getTimestamp() - prevTs));
+                expected.add((thisValue + (prevValue * -1)) / (k.getTimestamp() - prevTs) * 1000);
             }
             prevTs = k.getTimestamp();
             prevValue = MetricAdapter.decodeValue(v.get());
@@ -132,7 +144,7 @@ public class RateIteratorTest extends IteratorTestBase {
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
         for (int i = 0; i < 99; i++) {
             assertTrue(iter.hasTop());
-            assertEquals(0.001D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
+            assertEquals(1.0D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
             iter.next();
         }
         assertFalse(iter.hasTop());
@@ -166,7 +178,7 @@ public class RateIteratorTest extends IteratorTestBase {
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
         for (int i = 0; i < 99; i++) {
             assertTrue(iter.hasTop());
-            assertEquals(0.001D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
+            assertEquals(1.0D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
             iter.next();
         }
         assertFalse(iter.hasTop());
@@ -201,7 +213,7 @@ public class RateIteratorTest extends IteratorTestBase {
         iter.seek(new Range(), EMPTY_COL_FAMS, true);
         for (int i = 0; i < 99; i++) {
             assertTrue(iter.hasTop());
-            assertEquals(((i + 1) % 10 == 0 ? 0.0D : 0.001D), MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
+            assertEquals(((i + 1) % 10 == 0 ? 0.0D : 1.0D), MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
             iter.next();
         }
         assertFalse(iter.hasTop());

server/src/test/java/timely/server/store/compaction/MetricCompactionStrategyTest.java

@@ -1,288 +0,0 @@
-package timely.server.store.compaction;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-import java.io.IOException;
-import java.util.Calendar;
-import java.util.GregorianCalendar;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.TreeMap;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.tserver.compaction.CompactionPlan;
-import org.apache.accumulo.tserver.compaction.CompactionStrategy;
-import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
-import org.junit.Test;
-
-import timely.server.test.CompactionRequestBuilder;
-
-public class MetricCompactionStrategyTest {
-
-    private static final long TIME_MILLIS = new GregorianCalendar(2019, Calendar.JANUARY, 1).getTimeInMillis();
-
-    @Test(expected = IllegalArgumentException.class)
-    public void initThrowsWhenNoDefaultAgeOff() {
-        Map<String,String> config = new HashMap<>();
-        MetricCompactionStrategy strategy = new MetricCompactionStrategy();
-        strategy.init(config);
-        CompactionRequestBuilder builder = new CompactionRequestBuilder().endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
-                        .prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12)).file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                        .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        MajorCompactionRequest request = builder.build();
-        strategy.shouldCompact(request);
-    }
-
-    @Test
-    public void minAgeOffOverridesMajcDefaultWillUseMaximum() throws IOException {
-        Map<String,String> config = new HashMap<>();
-        config.put(MetricCompactionStrategy.MIN_AGEOFF_KEY, Long.toString(TimeUnit.DAYS.toMillis(10)));
-        CompactionStrategy strategy = newStrategy(config);
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(13)
-                .endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void minAgeOffOverridesMajcDefaultNotCompact() throws IOException {
-        Map<String,String> config = new HashMap<>();
-        config.put(MetricCompactionStrategy.MIN_AGEOFF_KEY, Long.toString(TimeUnit.DAYS.toMillis(13)));
-        CompactionStrategy strategy = newStrategy(config);
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void compactFilesIfKeyIsPastAgeOff() throws IOException {
-        String f1 = "hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf";
-        String f2 = "hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf";
-        String f3 = "hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao6.rf";
-        CompactionStrategy strategy = newStrategy();
-
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys", TimeUnit.DAYS.toMillis(12))
-                .file(f1, 12, 1)
-                .file(f2, 11, 1)
-                .file(f3, 16, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertTrue(shouldCompact);
-        assertNotNull(plan);
-
-        assertEquals(3, plan.inputFiles.size());
-        assertEquals(f1, plan.inputFiles.get(0).getPathStr());
-        assertEquals(f2, plan.inputFiles.get(1).getPathStr());
-        assertEquals(f3, plan.inputFiles.get(2).getPathStr());
-    }
-
-    @Test
-    public void noCompactionIfPrevKeyIsDifferent() throws IOException {
-        CompactionStrategy strategy = newStrategy();
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys.mem", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void noCompactionIfPrevKeyIsNull() throws IOException {
-        CompactionStrategy strategy = newStrategy();
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys", TimeUnit.DAYS.toMillis(11))
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void noCompactionWhenExceedsButDoesNotHaveFiles() throws IOException {
-        CompactionStrategy strategy = newStrategy();
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys", TimeUnit.DAYS.toMillis(12));
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void noCompactionIfKeyIsNotOffset() throws IOException {
-        CompactionStrategy strategy = newStrategy();
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetricNonOffset("sys.mem", null)
-                .prevEndKeyMetricNonOffset("sys.cpu", null)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void noCompactionIfKeyIsWrongOffset() throws IOException {
-        CompactionStrategy strategy = newStrategy();
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetricNonOffset("sys.cpu", "test2")
-                .prevEndKeyMetricNonOffset("sys.cpu", "test1")
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void compactWithLogOnlyDoesNotCompact() throws IOException {
-        Map<String,String> defaults = new HashMap<>();
-        defaults.put(MetricCompactionStrategy.LOG_ONLY_KEY, "true");
-        CompactionStrategy strategy = newStrategy(defaults);
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    @Test
-    public void compactWithFilterPrefixWillCompact() throws IOException {
-        Map<String,String> defaults = new HashMap<>();
-        defaults.put(MetricCompactionStrategy.FILTER_PREFIX_KEY, "sys.cp");
-        CompactionStrategy strategy = newStrategy(defaults);
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertTrue(shouldCompact);
-        assertNotNull(plan);
-
-        assertEquals(2, plan.inputFiles.size());
-    }
-
-    @Test
-    public void compactWithoutFilterPrefixWillNotCompact() throws IOException {
-        Map<String,String> defaults = new HashMap<>();
-        defaults.put(MetricCompactionStrategy.FILTER_PREFIX_KEY, "sys.mem");
-        CompactionStrategy strategy = newStrategy(defaults);
-        // @formatter:off
-        CompactionRequestBuilder builder = newRequestBuilder(10)
-                .endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
-                .prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
-                .file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
-
-        // @formatter:on
-        MajorCompactionRequest request = builder.build();
-        boolean shouldCompact = strategy.shouldCompact(request);
-        CompactionPlan plan = strategy.getCompactionPlan(request);
-
-        assertFalse(shouldCompact);
-        assertNull(plan);
-    }
-
-    private static CompactionRequestBuilder newRequestBuilder(int defaultAgeOffDays) {
-        return new CompactionRequestBuilder(TIME_MILLIS).tableProperties(Property.TABLE_ITERATOR_MAJC_PREFIX.getKey() + "ageoffmetrics.opt.ageoff.default",
-                        Long.toString(TimeUnit.DAYS.toMillis(defaultAgeOffDays)));
-    }
-
-    private static MetricCompactionStrategy newStrategy() {
-        return newStrategy(new TreeMap<>());
-    }
-
-    private static MetricCompactionStrategy newStrategy(Map<String,String> config) {
-        if (!config.containsKey(MetricCompactionStrategy.LOG_ONLY_KEY)) {
-            config.put(MetricCompactionStrategy.LOG_ONLY_KEY, "false");
-        }
-        MetricCompactionStrategy strategy = new MetricCompactionStrategy();
-        strategy.setExplicitTime(TIME_MILLIS);
-        strategy.init(config);
-        return strategy;
-    }
-}

server/src/test/java/timely/server/store/compaction/TieredCompactionStrategyTest.java

@@ -1,238 +0,0 @@
-package timely.server.store.compaction;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.accumulo.core.metadata.StoredTabletFile;
-import org.apache.accumulo.tserver.compaction.CompactionPlan;
-import org.apache.accumulo.tserver.compaction.CompactionStrategy;
-import org.apache.accumulo.tserver.compaction.DefaultCompactionStrategy;
-import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
-import org.easymock.EasyMock;
-import org.junit.Test;
-
-import com.google.common.collect.Lists;
-
-import timely.server.test.CompactionRequestBuilder;
-
-public class TieredCompactionStrategyTest {
-
-    @Test
-    public void initConfiguresWithMultipleTiers() {
-        Map<String,String> config = new HashMap<>();
-        config.put("tier.0.class", DefaultCompactionStrategy.class.getName());
-        config.put("tier.0.opts.arg1", "foo");
-        config.put("tier.0.opts.arg2", "sys");
-        config.put("tier.1.class", DefaultCompactionStrategy.class.getName());
-        config.put("tier.1.opts.arg1", "bar");
-        config.put("tier.2.class", DefaultCompactionStrategy.class.getName());
-
-        TieredCompactionStrategy strategy = new TieredCompactionStrategy();
-        strategy.init(config);
-        List<TieredCompactorSupplier> suppliers = Lists.newArrayList(strategy.getCompactors());
-        assertEquals(3, suppliers.size());
-        assertEquals(2, suppliers.get(0).options().size());
-        assertEquals(1, suppliers.get(1).options().size());
-        assertEquals(0, suppliers.get(2).options().size());
-        assertEquals("foo", suppliers.get(0).options().get("arg1"));
-        assertEquals("sys", suppliers.get(0).options().get("arg2"));
-        assertEquals("bar", suppliers.get(1).options().get("arg1"));
-    }
-
-    @Test
-    public void initConfiguresWithSingleTierNoArguments() {
-        Map<String,String> config = new HashMap<>();
-        config.put("tier.0.class", DefaultCompactionStrategy.class.getName());
-
-        TieredCompactionStrategy strategy = new TieredCompactionStrategy();
-        strategy.init(config);
-        List<TieredCompactorSupplier> suppliers = Lists.newArrayList(strategy.getCompactors());
-        assertEquals(1, suppliers.size());
-        assertEquals(0, suppliers.get(0).options().size());
-    }
-
-    @Test
-    public void gatherInformationAcrossTiers() throws IOException {
-        CompactionStrategy c1 = EasyMock.createMock(CompactionStrategy.class);
-        CompactionStrategy c2 = EasyMock.createMock(CompactionStrategy.class);
-
-        c1.gatherInformation(EasyMock.anyObject());
-        EasyMock.expectLastCall();
-
-        c2.gatherInformation(EasyMock.anyObject());
-        EasyMock.expectLastCall();
-
-        EasyMock.replay(c1, c2);
-
-        TieredCompactionStrategy strategy = newStrategy(c1, c2);
-        CompactionRequestBuilder builder = new CompactionRequestBuilder();
-        MajorCompactionRequest request = builder.build();
-
-        strategy.gatherInformation(request);
-
-        EasyMock.verify(c1, c2);
-    }
-
-    @Test
-    public void shouldCompactStopsAtTier() throws IOException {
-        CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
-        CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
-        CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
-
-        EasyMock.expect(c1.shouldCompact(EasyMock.anyObject())).andReturn(false);
-        EasyMock.expect(c2.shouldCompact(EasyMock.anyObject())).andReturn(true);
-
-        EasyMock.replay(c1, c2, c3);
-
-        TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
-        CompactionRequestBuilder builder = new CompactionRequestBuilder();
-        MajorCompactionRequest request = builder.build();
-
-        boolean result = strategy.shouldCompact(request);
-
-        assertTrue(result);
-        EasyMock.verify(c1, c2, c3);
-    }
-
-    @Test
-    public void shouldCompactWithSingleTier() throws Exception {
-        CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
-
-        EasyMock.expect(c1.shouldCompact(EasyMock.anyObject())).andReturn(false);
-        EasyMock.replay(c1);
-
-        TieredCompactionStrategy strategy = newStrategy(c1);
-        CompactionRequestBuilder builder = new CompactionRequestBuilder();
-        MajorCompactionRequest request = builder.build();
-
-        boolean result = strategy.shouldCompact(request);
-
-        assertFalse(result);
-        EasyMock.verify(c1);
-    }
-
-    @Test
-    public void getCompactionPlanSkipsAndStopsAtTier() throws IOException {
-        CompactionPlan p1 = new CompactionPlan();
-        p1.inputFiles.add(new StoredTabletFile("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf"));
-
-        CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
-        CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
-        CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
-
-        EasyMock.expect(c1.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
-        EasyMock.expect(c2.getCompactionPlan(EasyMock.anyObject())).andReturn(p1);
-
-        EasyMock.replay(c1, c2, c3);
-
-        TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
-        CompactionRequestBuilder builder = new CompactionRequestBuilder();
-        MajorCompactionRequest request = builder.build();
-
-        CompactionPlan actualPlan = strategy.getCompactionPlan(request);
-
-        assertNotNull(actualPlan);
-        assertSame(p1, actualPlan);
-        EasyMock.verify(c1, c2, c3);
-    }
-
-    @Test
-    public void getCompactionPlanSkipsWhenEmptyInputAndStopsAtTier() throws IOException {
-        CompactionPlan p1 = new CompactionPlan();
-        p1.inputFiles.add(new StoredTabletFile("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf"));
-
-        CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
-        CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
-        CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
-
-        EasyMock.expect(c1.getCompactionPlan(EasyMock.anyObject())).andReturn(new CompactionPlan());
-        EasyMock.expect(c2.getCompactionPlan(EasyMock.anyObject())).andReturn(p1);
-
-        EasyMock.replay(c1, c2, c3);
-
-        TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
-        CompactionRequestBuilder builder = new CompactionRequestBuilder();
-        MajorCompactionRequest request = builder.build();
-
-        CompactionPlan actualPlan = strategy.getCompactionPlan(request);
-
-        assertNotNull(actualPlan);
-        assertSame(p1, actualPlan);
-        EasyMock.verify(c1, c2, c3);
-    }
-
-    @Test
-    public void getCompactionPlanWillReturnNullIfNonCompact() throws IOException {
-        CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
-        CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
-        CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
-
-        EasyMock.expect(c1.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
-        EasyMock.expect(c2.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
-        EasyMock.expect(c3.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
-
-        EasyMock.replay(c1, c2, c3);
-
-        TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
-        CompactionRequestBuilder builder = new CompactionRequestBuilder();
-        MajorCompactionRequest request = builder.build();
-
-        CompactionPlan actualPlan = strategy.getCompactionPlan(request);
-
-        assertNull(actualPlan);
-        EasyMock.verify(c1, c2, c3);
-    }
-
-    private static TieredCompactionStrategy newStrategy(CompactionStrategy... strategies) {
-        Collection<TieredCompactorSupplier> mockSuppliers = new ArrayList<>();
-        for (CompactionStrategy s : strategies) {
-            mockSuppliers.add(new MockSupplier(s));
-        }
-        // @formatter:off
-        TieredCompactionStrategy strategy = EasyMock.partialMockBuilder(TieredCompactionStrategy.class)
-                .withConstructor()
-                .addMockedMethod("createSuppliers")
-                .createMock();
-
-        // @formatter:on
-        strategy.createSuppliers(EasyMock.anyObject());
-        EasyMock.expectLastCall().andAnswer(() -> strategy.compactors.addAll(mockSuppliers));
-
-        EasyMock.replay(strategy);
-
-        strategy.init(Collections.emptyMap());
-
-        return strategy;
-    }
-
-    static class MockSupplier implements TieredCompactorSupplier {
-
-        private final CompactionStrategy strategy;
-
-        MockSupplier(CompactionStrategy strategy) {
-            this.strategy = strategy;
-        }
-
-        @Override
-        public Map<String,String> options() {
-            return Collections.emptyMap();
-        }
-
-        @Override
-        public CompactionStrategy get(Map<String,String> tableProperites) {
-            return strategy;
-        }
-    }
-}

server/src/test/java/timely/server/test/CompactionRequestBuilder.java

@@ -1,119 +0,0 @@
-package timely.server.test;
-
-import java.util.Calendar;
-import java.util.GregorianCalendar;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.TreeMap;
-
-import org.apache.accumulo.core.client.lexicoder.Lexicoder;
-import org.apache.accumulo.core.client.lexicoder.PairLexicoder;
-import org.apache.accumulo.core.client.lexicoder.StringLexicoder;
-import org.apache.accumulo.core.conf.AccumuloConfiguration;
-import org.apache.accumulo.core.conf.DefaultConfiguration;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.TableId;
-import org.apache.accumulo.core.data.TabletId;
-import org.apache.accumulo.core.dataImpl.KeyExtent;
-import org.apache.accumulo.core.dataImpl.TabletIdImpl;
-import org.apache.accumulo.core.metadata.StoredTabletFile;
-import org.apache.accumulo.core.metadata.schema.DataFileValue;
-import org.apache.accumulo.core.util.ComparablePair;
-import org.apache.accumulo.server.ServerContext;
-import org.apache.accumulo.tserver.compaction.MajorCompactionReason;
-import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
-import org.apache.hadoop.io.Text;
-import org.easymock.EasyMock;
-
-import timely.accumulo.MetricAdapter;
-
-public class CompactionRequestBuilder {
-
-    private static final long DEFAULT_TIME_MILLIS = new GregorianCalendar(2019, Calendar.JANUARY, 1).getTimeInMillis();
-    private static final Lexicoder<String> stringEncoder = new StringLexicoder();
-    private static final PairLexicoder<String,String> badPairEncoder = new PairLexicoder<>(stringEncoder, stringEncoder);
-
-    private final Map<StoredTabletFile,DataFileValue> refs = new LinkedHashMap<>();
-    private final Map<String,String> tableProperties;
-    private final long timeMillis;
-    private Key endKeyMetric;
-    private Key prevKeyMetric;
-
-    public CompactionRequestBuilder() {
-        this(DEFAULT_TIME_MILLIS);
-    }
-
-    public CompactionRequestBuilder(long timeMillis) {
-        this.timeMillis = timeMillis;
-        this.tableProperties = new TreeMap<>();
-    }
-
-    public CompactionRequestBuilder endKeyMetricNonOffset(String name, String offset) {
-        byte[] rowKey = offset != null ? badPairEncoder.encode(new ComparablePair<>(name, offset)) : stringEncoder.encode(name);
-        endKeyMetric = new Key(rowKey);
-        return this;
-    }
-
-    public CompactionRequestBuilder endKeyMetric(String name, long offset) {
-        long ts = timeMillis - offset;
-        byte[] rowKey = MetricAdapter.encodeRowKey(name, ts);
-        endKeyMetric = new Key(rowKey);
-        return this;
-    }
-
-    public CompactionRequestBuilder prevEndKeyMetric(String name, long offset) {
-        long ts = timeMillis - offset;
-        byte[] rowKey = MetricAdapter.encodeRowKey(name, ts);
-        prevKeyMetric = new Key(rowKey);
-        return this;
-    }
-
-    public CompactionRequestBuilder prevEndKeyMetricNonOffset(String name, String offset) {
-        byte[] rowKey = offset != null ? badPairEncoder.encode(new ComparablePair<>(name, offset)) : stringEncoder.encode(name);
-        prevKeyMetric = new Key(rowKey);
-        return this;
-    }
-
-    public CompactionRequestBuilder file(String path, long size, long entries) {
-        refs.put(new StoredTabletFile(path), new DataFileValue(size, entries));
-        return this;
-    }
-
-    public CompactionRequestBuilder tableProperties(String key, String value) {
-        tableProperties.put(key, value);
-        return this;
-    }
-
-    public MajorCompactionRequest build() {
-        TableId tableId = TableId.of("1");
-        Text endRow = null;
-        if (null != endKeyMetric) {
-            endRow = endKeyMetric.getRow();
-        }
-
-        Text prevEndRow = null;
-        if (null != prevKeyMetric) {
-            prevEndRow = prevKeyMetric.getRow();
-        }
-        KeyExtent ke = new KeyExtent(tableId, endRow, prevEndRow);
-        TabletId tid = new TabletIdImpl(ke);
-
-        // @formatter:off
-        MajorCompactionRequest req = EasyMock.partialMockBuilder(MajorCompactionRequest.class)
-                .withConstructor(KeyExtent.class, MajorCompactionReason.class,
-                        AccumuloConfiguration.class, ServerContext.class)
-                .withArgs(ke, MajorCompactionReason.NORMAL, DefaultConfiguration.getInstance(), null)
-                .addMockedMethod("getTabletId")
-                .addMockedMethod("getTableProperties")
-                .createMock();
-
-        // @formatter:on
-        req.setFiles(refs);
-        EasyMock.expect(req.getTabletId()).andStubReturn(tid);
-        EasyMock.expect(req.getTableProperties()).andStubReturn(tableProperties);
-
-        EasyMock.replay(req);
-
-        return req;
-    }
-}

test/pom.xml

@@ -4,12 +4,12 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.4-SNAPSHOT</version>
+    <version>3.0.0-SNAPSHOT</version>
   </parent>
   <artifactId>timely-test</artifactId>
   <properties>
-    <maven.compiler.source>11</maven.compiler.source>
-    <maven.compiler.target>11</maven.compiler.target>
+    <maven.compiler.source>17</maven.compiler.source>
+    <maven.compiler.target>17</maven.compiler.target>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
   </properties>
   <dependencies>

test/src/main/java/timely/test/configuration/InMemoryAccumuloConfiguration.java

@@ -9,8 +9,8 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
 
+import datawave.accumulo.inmemory.InMemoryAccumulo;
 import datawave.accumulo.inmemory.InMemoryAccumuloClient;
-import datawave.accumulo.inmemory.InMemoryInstance;
 import timely.common.configuration.AccumuloProperties;
 
 @Configuration
@@ -20,15 +20,15 @@ public class InMemoryAccumuloConfiguration {
 
     protected static final String ROOT_USER = "root";
 
-    @Bean(name = "InMemoryInstance")
+    @Bean(name = "InMemoryAccumulo")
     @Singleton
-    public InMemoryInstance inMemoryInstance() throws Exception {
-        return new InMemoryInstance();
+    public InMemoryAccumulo inMemoryAccumulo() throws Exception {
+        return new InMemoryAccumulo();
     }
 
     @Bean
     @Primary
-    public AccumuloClient inMemoryAccumuloClient(InMemoryInstance instance) throws Exception {
+    public AccumuloClient inMemoryAccumuloClient(InMemoryAccumulo instance) throws Exception {
         return new InMemoryAccumuloClient(ROOT_USER, instance);
     }
 }

test/src/main/java/timely/test/configuration/MiniAccumuloConfiguration.java

@@ -69,7 +69,6 @@ public class MiniAccumuloConfiguration {
         }
 
         macConfig.setZooKeeperPort(zookeeperPort);
-        macConfig.setNumTservers(1);
         macConfig.setMemory(ServerType.TABLET_SERVER, 1, MemoryUnit.GIGABYTE);
         try {
             miniAccumuloCluster = new MiniAccumuloCluster(macConfig);