2.2.1

Fixed the metric age off iterator to ensure forward progress when seeking past keys for ageoff

analytics/pom.xml

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

balancer/pom.xml

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

client/pom.xml

@@ -4,13 +4,13 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.2.0</version>
+    <version>2.2.1</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.2.0</python.client.version>
+    <python.client.version>2.2.1.dev0</python.client.version>
   </properties>
   <dependencies>
     <dependency>

collectd/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.2.0</version>
+    <version>2.2.1</version>
   </parent>
   <artifactId>timely-collectd</artifactId>
   <dependencies>

common/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.2.0</version>
+    <version>2.2.1</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>

grafana/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.2.0</version>
+    <version>2.2.1</version>
   </parent>
   <artifactId>timely-datasource</artifactId>
   <packaging>pom</packaging>

pom.xml

@@ -3,7 +3,7 @@
   <modelVersion>4.0.0</modelVersion>
   <groupId>timely</groupId>
   <artifactId>timely</artifactId>
-  <version>2.2.0</version>
+  <version>2.2.1</version>
   <packaging>pom</packaging>
   <name>Timely</name>
   <description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -33,7 +33,8 @@
   </issueManagement>
   <properties>
     <assembly.tarLongFileMode>posix</assembly.tarLongFileMode>
-    <collectd-api-jar>/usr/share/collectd/java/collectd-api.jar</collectd-api-jar>
+    <!--collectd-api-jar>/usr/share/collectd/java/collectd-api.jar</collectd-api-jar-->
+    <collectd-api-jar>/home/inbella/Downloads/temp/collectd-5.7.1/bindings/java/.libs/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>

server/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.2.0</version>
+    <version>2.2.1</version>
   </parent>
   <artifactId>timely-server</artifactId>
   <description>Timely Server</description>

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

@@ -16,6 +16,7 @@ import org.apache.accumulo.core.util.Pair;
 import org.apache.commons.collections4.trie.PatriciaTrie;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableComparator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -83,7 +84,7 @@ public class MetricAgeOffIterator extends WrappingIterator implements OptionDesc
                 log.trace("Current key is older than max age off, seeking to start of valid data");
                 String metricName = MetricAdapter.decodeRowKey(top).getFirst();
                 handleNewMetricName(metricName);
-                seekPastAgedOffMetricData(metricName, this.maxAgeOff);
+                seekPastAgedOffMetricData(top.getRow().getBytes(), metricName, this.maxAgeOff);
                 return;
             }
             if (isNextMetricTheSame(top.getRow())) {
@@ -93,7 +94,7 @@ public class MetricAgeOffIterator extends WrappingIterator implements OptionDesc
                     // process
                     String metricName = new String(prevMetricBytes.copyBytes(), UTF_8);
                     log.trace("Current metric is older than age off for metric {}, seeking to start of valid data", metricName);
-                    seekPastAgedOffMetricData(metricName, prevAgeOff);
+                    seekPastAgedOffMetricData(top.getRow().getBytes(), metricName, prevAgeOff);
                 }
             } else {
                 // Metric name is different or prev information is not set
@@ -101,7 +102,7 @@ public class MetricAgeOffIterator extends WrappingIterator implements OptionDesc
                 handleNewMetricName(metricName);
                 if (currentTime - top.getTimestamp() > prevAgeOff) {
                     log.trace("New metric found, but older than age off for metric {}, seeking to start of valid data", metricName);
-                    seekPastAgedOffMetricData(metricName, prevAgeOff);
+                    seekPastAgedOffMetricData(top.getRow().getBytes(), metricName, prevAgeOff);
                 }
             }
         }
@@ -127,16 +128,34 @@ public class MetricAgeOffIterator extends WrappingIterator implements OptionDesc
         }
     }
 
-    private void seekPastAgedOffMetricData(String metricName, long ageOffTime) {
+    /**
+     * Ensure that we have a row that is following that of the current top row. If the newStart is not after the topRow, then return a row that is just after
+     * topRow.
+     *
+     * @param topRow
+     * @param newStart
+     * @return the new start
+     */
+    private byte[] ensureFollowing(byte[] topRow, byte[] newStart) {
+        if (WritableComparator.compareBytes(topRow, 0, topRow.length, newStart, 0, newStart.length) >= 0) {
+            byte[] followingRow = new byte[topRow.length + 1];
+            System.arraycopy(topRow, 0, followingRow, 0, topRow.length);
+            followingRow[topRow.length] = (byte) 0x00;
+            return followingRow;
+        }
+        return newStart;
+    }
+
+    private void seekPastAgedOffMetricData(byte[] topRow, String metricName, long ageOffTime) {
         long timeTarget = currentTime - ageOffTime;
         Range newRange = null;
         if (null == this.range) {
-            byte[] newStartRow = MetricAdapter.encodeRowKey(metricName, timeTarget);
+            byte[] newStartRow = ensureFollowing(topRow, MetricAdapter.encodeRowKey(metricName, timeTarget));
             newRange = new Range(new Key(new Text(newStartRow), timeTarget), null);
         } else {
             Key startKey = new Key(new Text(MetricAdapter.encodeRowKey(metricName, timeTarget)), timeTarget);
             Pair<String,Long> start = MetricAdapter.decodeRowKey(startKey);
-            byte[] newStartRow = MetricAdapter.encodeRowKey(start.getFirst(), timeTarget);
+            byte[] newStartRow = ensureFollowing(topRow, MetricAdapter.encodeRowKey(start.getFirst(), timeTarget));
             // @formatter:off
             Key newStartKey = new Key(newStartRow,
                     null != startKey.getColumnFamily() ? startKey.getColumnFamilyData().getBackingArray() : null,

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

@@ -0,0 +1,39 @@
+package timely.server.store;
+
+import org.apache.accumulo.core.data.ColumnUpdate;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.junit.Assert;
+import org.junit.Test;
+
+import timely.accumulo.MetricAdapter;
+import timely.model.Metric;
+import timely.server.test.TestMetrics;
+
+public class MutationsTest {
+
+    private final TestMetrics metrics = new TestMetrics();
+
+    @Test
+    public void testMutations() {
+        long iterations = 65536;
+        long delta = Long.MAX_VALUE / iterations * 2;
+        long ts = Long.MIN_VALUE;
+
+        for (int i = 0; i < iterations; i++) {
+            Metric m = metrics.randomMetric(ts);
+            Mutation mutation = MetricAdapter.toMutation(m);
+            for (ColumnUpdate update : mutation.getUpdates()) {
+                Key key = new Key(mutation.getRow(), update.getColumnFamily(), update.getColumnQualifier(), update.getColumnVisibility(),
+                                update.getTimestamp());
+                long time = MetricAdapter.decodeRowKey(key).getSecond();
+                Assert.assertEquals("timestamps " + update.getTimestamp() + " expected to be " + ts, ts, update.getTimestamp());
+                Assert.assertEquals("time " + time + " expected to be " + MetricAdapter.roundTimestampToLastHour(ts),
+                                MetricAdapter.roundTimestampToLastHour(ts), time);
+                long diff = Math.abs(time - ts);
+                Assert.assertTrue("encoded time " + time + " not close to " + ts, diff < 3600000);
+            }
+            ts += delta;
+        }
+    }
+}

server/src/test/java/timely/server/store/ageoff/MetricAgeOffIteratorTest.java

@@ -1,12 +1,10 @@
 package timely.server.store.ageoff;
 
-import java.util.ArrayList;
-import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.SortedMap;
 import java.util.TreeMap;
 
-import org.apache.accumulo.core.data.ByteSequence;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
@@ -23,7 +21,6 @@ public class MetricAgeOffIteratorTest {
     private static final Long TEST_TIME = System.currentTimeMillis() - 1000;
     private static final Integer ONE_DAY = 86400000;
     private static final Value EMPTY_VALUE = new Value(new byte[0]);
-    private static final Collection<ByteSequence> columnFamilies = new ArrayList<>();
 
     @Test(expected = IllegalArgumentException.class)
     public void testDefaultMissing() throws Exception {
@@ -49,7 +46,7 @@ public class MetricAgeOffIteratorTest {
         HashMap<String,String> options = new HashMap<>();
         options.put(MetricAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
         iter.init(source, options, null);
-        iter.seek(new Range(), columnFamilies, true);
+        iter.seek(new Range(), Collections.emptySet(), false);
         int seen = 0;
         while (iter.hasTop()) {
             Key k = iter.getTopKey();
@@ -81,7 +78,7 @@ public class MetricAgeOffIteratorTest {
         HashMap<String,String> options = new HashMap<>();
         options.put(MetricAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
         iter.init(source, options, null);
-        iter.seek(new Range(), columnFamilies, true);
+        iter.seek(new Range(), Collections.emptySet(), false);
         int seen = 0;
         while (iter.hasTop()) {
             Key k = iter.getTopKey();
@@ -124,7 +121,7 @@ public class MetricAgeOffIteratorTest {
         options.put(MetricAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
         options.put(MetricAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
         iter.init(source, options, null);
-        iter.seek(new Range(), columnFamilies, true);
+        iter.seek(new Range(), Collections.emptySet(), false);
         int seen = 0;
         while (iter.hasTop()) {
             Key k = iter.getTopKey();
@@ -153,7 +150,7 @@ public class MetricAgeOffIteratorTest {
         iter.init(source, options, null);
         iter.seek(new Range(new Key("sys.cpu.user"), true,
                         new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME + 3), new byte[0], new byte[0], new byte[0], TEST_TIME + 3), true),
-                        columnFamilies, true);
+                        Collections.emptySet(), false);
         int seen = 0;
         while (iter.hasTop()) {
             Key k = iter.getTopKey();
@@ -164,4 +161,54 @@ public class MetricAgeOffIteratorTest {
         Assert.assertEquals(0, seen);
     }
 
+    @Test
+    public void testKeysInOrder() throws Exception {
+        SortedMap<Key,Value> table = new TreeMap<>();
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.idle", TEST_TIME - (3 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME - (3 * ONE_DAY)), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.idle", TEST_TIME - (2 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME - (2 * ONE_DAY)), EMPTY_VALUE);
+        // creating a key that previously would cause the age off iterator to seek backwards
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.idle", TEST_TIME + (1000 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME - (1 * ONE_DAY)), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.idle", TEST_TIME), new byte[0], new byte[0], new byte[0], TEST_TIME), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.idle", TEST_TIME + ONE_DAY), new byte[0], new byte[0], new byte[0], TEST_TIME + ONE_DAY),
+                        EMPTY_VALUE);
+        // creating a key that previously would cause the age off iterator to seek backwards
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.idle", TEST_TIME + (2000 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME + (2 * ONE_DAY)), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME - (3 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME - (3 * ONE_DAY)), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME - (2 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME - (2 * ONE_DAY)), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME - (1 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME - (1 * ONE_DAY)), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME), new byte[0], new byte[0], new byte[0], TEST_TIME), EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME + ONE_DAY), new byte[0], new byte[0], new byte[0], TEST_TIME + ONE_DAY),
+                        EMPTY_VALUE);
+        table.put(new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME + (2 * ONE_DAY)), new byte[0], new byte[0], new byte[0],
+                        TEST_TIME + (2 * ONE_DAY)), EMPTY_VALUE);
+
+        SortedKeyValueIterator<Key,Value> source = new SortedMapIterator(table);
+        MetricAgeOffIterator iter = new MetricAgeOffIterator();
+        HashMap<String,String> options = new HashMap<>();
+        options.put(MetricAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
+        options.put(MetricAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
+        iter.init(source, options, null);
+        iter.seek(new Range(), Collections.emptySet(), false);
+        int seen = 0;
+        Key last = null;
+        while (iter.hasTop()) {
+            Key k = iter.getTopKey();
+            Assert.assertTrue(k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
+            if (last != null) {
+                Assert.assertTrue(k.compareTo(last) > 0);
+            }
+            last = k;
+            seen++;
+            iter.next();
+        }
+        Assert.assertEquals(7, seen);
+    }
+
 }

test/pom.xml

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