Release version 2.1.5

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.3</version>
+    <version>2.1.5</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--

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,16 +46,19 @@ 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 ${THIS_DIR}/timely-balancer-*-exec.jar --instance=${INSTANCE}"
-    nohup "${JAVA_HOME}"/bin/java ${JVM_ARGS} -jar "${THIS_DIR}"/timely-balancer-*-exec.jar --instance="${INSTANCE}" >> "${LOG_DIR}/timely-balancer${INSTANCE}.out" 2>&1 &
+    echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-balancer-*-exec.jar --instance=${INSTANCE}"
+    nohup "${JAVA_HOME}"/bin/java ${JVM_ARGS} -jar "${BIN_DIR}"/timely-balancer-*-exec.jar --instance="${INSTANCE}" >> "${LOG_DIR}/timely-balancer${INSTANCE}.out" 2>&1 &
 else
     echo "timely-balancer --instance=${INSTANCE} already running with pid ${PID}"
 fi

balancer/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.3</version>
+    <version>2.1.5</version>
   </parent>
   <artifactId>timely-balancer</artifactId>
   <description>Timely Balancer</description>

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.3</version>
+    <version>2.1.5</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.3</python.client.version>
+    <python.client.version>2.1.5</python.client.version>
   </properties>
   <dependencies>
     <dependency>

collectd-nsq-plugin/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.3</version>
+    <version>2.1.5</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.3</version>
+    <version>2.1.5</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.3</version>
+    <version>2.1.5</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.3</version>
+    <version>2.1.5</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:

grafana/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.3</version>
+    <version>2.1.5</version>
   </parent>
   <artifactId>timely-app</artifactId>
   <packaging>pom</packaging>

grafana/timely-app/dist/plugin.json

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

grafana/timely-app/package.json

@@ -1,6 +1,6 @@
 {
   "name": "timely-datasource",
-  "version": "2.1.3",
+  "version": "2.1.5",
   "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.3</version>
+  <version>2.1.5</version>
   <packaging>pom</packaging>
   <name>Timely</name>
   <description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -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>1.4</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>3.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>3.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,16 +46,19 @@ 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 ${THIS_DIR}/timely-server*-exec.jar timelyserver --instance=${INSTANCE}"
-    nohup "${JAVA_HOME}"/bin/java ${JVM_ARGS} -jar "${THIS_DIR}"/timely-server-*-exec.jar timelyserver --instance="${INSTANCE}" >> "${LOG_DIR}/timely-server${INSTANCE}.out" 2>&1 &
+    echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server*-exec.jar timelyserver --instance=${INSTANCE}"
+    nohup "${JAVA_HOME}"/bin/java ${JVM_ARGS} -jar "${BIN_DIR}"/timely-server-*-exec.jar timelyserver --instance="${INSTANCE}" >> "${LOG_DIR}/timely-server${INSTANCE}.out" 2>&1 &
 else
     echo "timely-server --instance=${INSTANCE} already running with pid ${PID}"
 fi

server/bin/util/get-metrics.sh

@@ -1,25 +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
 
 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"
 
-echo "$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-server*-exec.jar metrics"
-$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-server-*-exec.jar metrics
+# 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,25 +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
 
 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 ${THIS_DIR}/timely-server*-exec.jar tablet --spring.profiles.active=timely,tablet"
-$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_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.3</version>
+    <version>2.1.5</version>
   </parent>
   <artifactId>timely-server</artifactId>
   <description>Timely Server</description>

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

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

test/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.3</version>
+    <version>2.1.5</version>
   </parent>
   <artifactId>timely-test</artifactId>
   <properties>