Compare commits

...
3 Commits
Author SHA1 Message Date
Bill Oley 3b93f4c743 Add age off iterator for meta table 2019-05-10 16:19:12 -04:00
Bill Oley 3d7d6d0efd Correct typos 2019-04-19 09:06:06 -04:00
Bill Oley dcd23d4fc8 fixes #181 Use netty-tcnative-boringssl-static 2019-04-18 11:56:54 -04:00
24 changed files with 625 additions and 259 deletions

No files matched your search

+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>timely-analytics</artifactId>
<!--
+3 -4
View File
@@ -27,15 +27,14 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
popd
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
#JVM_ARGS="${JVM_ARGS} -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
+3 -38
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>timely-balancer</artifactId>
<dependencies>
@@ -161,9 +161,8 @@
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>runtime</scope>
</dependency>
<dependency>
@@ -247,9 +246,8 @@
<artifactItems>
<artifactItem>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<type>jar</type>
<overWrite>false</overWrite>
<outputDirectory>${project.build.directory}/native</outputDirectory>
@@ -291,37 +289,4 @@
</extension>
</extensions>
</build>
<profiles>
<profile>
<id>native-epoll-linux</id>
<activation>
<os>
<family>!mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
<classifier>linux-x86_64</classifier>
</dependency>
</dependencies>
</profile>
<profile>
<id>native-epoll</id>
<activation>
<os>
<family>mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
+2 -2
View File
@@ -4,12 +4,12 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>timely-client</artifactId>
<properties>
<!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
<python.client.version>1.2.2.dev0</python.client.version>
<python.client.version>1.2.3</python.client.version>
</properties>
<dependencies>
<dependency>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>collectd-nsq-plugin</artifactId>
<dependencies>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>collectd-plugin-parent</artifactId>
<dependencies>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>collectd-timely-plugin</artifactId>
<dependencies>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>timely-common</artifactId>
<dependencies>
+3 -4
View File
@@ -27,15 +27,14 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
popd
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
JVM_ARGS="-Xmx1G -Xms1G -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
#JVM_ARGS="${JVM_ARGS} -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
+3 -38
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>grafana-auth</artifactId>
<dependencies>
@@ -165,9 +165,8 @@
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>runtime</scope>
</dependency>
<dependency>
@@ -251,9 +250,8 @@
<artifactItems>
<artifactItem>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<type>jar</type>
<overWrite>false</overWrite>
<outputDirectory>${project.build.directory}/native</outputDirectory>
@@ -295,37 +293,4 @@
</extension>
</extensions>
</build>
<profiles>
<profile>
<id>native-epoll-linux</id>
<activation>
<os>
<family>!mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
<classifier>linux-x86_64</classifier>
</dependency>
</dependencies>
</profile>
<profile>
<id>native-epoll</id>
<activation>
<os>
<family>mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>timely-app</artifactId>
<packaging>pom</packaging>
+2 -2
View File
@@ -23,8 +23,8 @@
{"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
{"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
],
"version": "1.2.2",
"updated": "2019-04-12"
"version": "1.2.3",
"updated": "2019-04-18"
},
"includes": [
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "timely-app",
"private": true,
"version": "1.2.2",
"version": "1.2.3",
"description": "",
"main": "index.js",
"scripts": {
+2 -2
View File
@@ -23,8 +23,8 @@
{"name": "Project site", "url": "https://github.com/NationalSecurityAgency/timely"},
{"name": "License & Terms", "url": "https://raw.githubusercontent.com/NationalSecurityAgency/timely/master/LICENSE"}
],
"version": "1.2.2",
"updated": "2019-04-12"
"version": "1.2.3",
"updated": "2019-04-18"
},
"includes": [
+11 -5
View File
@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
<packaging>pom</packaging>
<name>Timely</name>
<description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -75,8 +75,8 @@
<version.maven-rpm>2.1.4</version.maven-rpm>
<version.maven-shade>2.4.2</version.maven-shade>
<version.maven-surefire>2.19.1</version.maven-surefire>
<version.netty>4.1.33.Final</version.netty>
<version.netty-tcnative>2.0.24.Final</version.netty-tcnative>
<version.netty>4.1.35.Final</version.netty>
<version.netty-tcnative>2.0.25.Final</version.netty-tcnative>
<version.os-maven>1.5.0.Final</version.os-maven>
<version.powermock>1.6.4</version.powermock>
<version.slf4j>1.7.26</version.slf4j>
@@ -155,6 +155,12 @@
<groupId>org.apache.accumulo</groupId>
<artifactId>accumulo-core</artifactId>
<version>${version.accumulo}</version>
<exclusions>
<exclusion>
<groupId>io.netty</groupId>
<artifactId>netty</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.accumulo</groupId>
@@ -450,7 +456,7 @@
<artifactId>maven-failsafe-plugin</artifactId>
<version>${version.maven-surefire}</version>
<configuration>
<argLine>${failsafeArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty-tcnative.so -Dzookeeper.sasl.client=false</argLine>
<argLine>${failsafeArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty_tcnative_linux_x86_64.so -Dzookeeper.sasl.client=false</argLine>
<includes>
<include>**/*.java</include>
</includes>
@@ -493,7 +499,7 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>${version.maven-surefire}</version>
<configuration>
<argLine>${sureFireArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty-tcnative.so -Dzookeeper.sasl.client=false</argLine>
<argLine>${sureFireArgLine} -Djava.library.path=${project.build.directory}/native/META-INF/native/libnetty-tcnative_linux_x86_64.so -Dzookeeper.sasl.client=false</argLine>
<excludedGroups>timely.test.IntegrationTest</excludedGroups>
<redirectTestOutputToFile>true</redirectTestOutputToFile>
</configuration>
+4 -5
View File
@@ -27,15 +27,14 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty_tcnative.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-transport-native-epoll*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
popd
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}/libnetty-${TCNATIVE_SUFFIX}"
echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely"
$JAVA_HOME/bin/java ${JVM_ARGS} timely.Server --spring.config.name=timely
+6 -10
View File
@@ -2,17 +2,12 @@
if [[ `uname` == "Darwin" ]]; then
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
TCNATIVE_SUFFIX="tcnative.jnilib"
TCNATIVE_SUFFIX="jnilib"
else
THIS_SCRIPT=`readlink -f $0`
TCNATIVE_SUFFIX="tcnative.so"
TCNATIVE_SUFFIX="so"
fi
# netty tcnative file reference
#netty-tcnative-1.1.33.Fork18-linux-x86_64-fedora.jar -> libnetty-tcnative.so
#netty-tcnative-1.1.33.Fork18-linux-x86_64.jar -> libnetty-tcnative.so
#netty-tcnative-1.1.33.Fork18-osx-x86_64.jar -> libnetty-tcnative.jnilib
THIS_DIR="${THIS_SCRIPT%/*}"
NATIVE_DIR="${THIS_DIR}/META-INF/native"
BASE_DIR=${THIS_DIR}/..
@@ -32,11 +27,12 @@ fi
mkdir -p ${NATIVE_DIR}
pushd ${BASE_DIR}/bin
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative*.jar META-INF/native/libnetty-${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
$JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
popd
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS}"
export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
JVM_ARGS="-Xmx256m -Xms256m -Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2-spring.xml"
JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
# for debugging
+3 -38
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>1.2.2</version>
<version>1.2.3</version>
</parent>
<artifactId>timely-server</artifactId>
<dependencies>
@@ -149,9 +149,8 @@
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<scope>runtime</scope>
</dependency>
<dependency>
@@ -240,9 +239,8 @@
<artifactItems>
<artifactItem>
<groupId>io.netty</groupId>
<artifactId>netty-tcnative</artifactId>
<artifactId>netty-tcnative-boringssl-static</artifactId>
<version>${version.netty-tcnative}</version>
<classifier>${os.detected.classifier}</classifier>
<type>jar</type>
<overWrite>false</overWrite>
<outputDirectory>${project.build.directory}/native</outputDirectory>
@@ -284,37 +282,4 @@
</extension>
</extensions>
</build>
<profiles>
<profile>
<id>native-epoll-linux</id>
<activation>
<os>
<family>!mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
<classifier>linux-x86_64</classifier>
</dependency>
</dependencies>
</profile>
<profile>
<id>native-epoll</id>
<activation>
<os>
<family>mac</family>
</os>
</activation>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${version.netty}</version>
</dependency>
</dependencies>
</profile>
</profiles>
</project>
@@ -0,0 +1,53 @@
package timely.adapter.accumulo;
import org.apache.accumulo.core.client.lexicoder.LongLexicoder;
import org.apache.accumulo.core.client.lexicoder.PairLexicoder;
import org.apache.accumulo.core.client.lexicoder.StringLexicoder;
import org.apache.accumulo.core.data.Key;
import timely.api.model.Meta;
import timely.model.Metric;
import timely.model.parse.TagListParser;
import timely.model.parse.TagParser;
/**
* Creates Accumulo data structures from {@link Metric}
*/
public class MetaAdapter {
private static final PairLexicoder<String, Long> rowCoder = new PairLexicoder<>(new StringLexicoder(),
new LongLexicoder());
private static final PairLexicoder<Long, String> colQualCoder = new PairLexicoder<>(new LongLexicoder(),
new StringLexicoder());
private static final TagParser tagParser = new TagParser();
private static final TagListParser tagListParser = new TagListParser();
public static Key createMetricKey(String metricName, Long timestamp) {
return new Key(Meta.METRIC_PREFIX + metricName, "", "", timestamp);
}
public static Key createTagKey(String metricName, String tagName, Long timestamp) {
return new Key(Meta.TAG_PREFIX + metricName, tagName, "", timestamp);
}
public static Key createValueKey(String metricName, String tagName, String tagValue, Long timestamp) {
return new Key(Meta.VALUE_PREFIX + metricName, tagName, tagValue, timestamp);
}
public static Meta parse(Key key) {
String row = key.getRow().toString();
if (row.startsWith(Meta.METRIC_PREFIX)) {
return Meta.parse(key, null, Meta.METRIC_PREFIX);
} else if (row.startsWith(Meta.TAG_PREFIX)) {
return Meta.parse(key, null, Meta.TAG_PREFIX);
} else if (row.startsWith(Meta.VALUE_PREFIX)) {
return Meta.parse(key, null, Meta.VALUE_PREFIX);
}
throw new IllegalStateException("Invalid key in meta " + key.toString());
}
public static String decodeRowKey(Key key) {
return parse(key).getMetric();
}
}
@@ -55,8 +55,32 @@ public class Meta implements Comparable<Meta> {
return keys;
}
public static Meta parse(Key k, Value v) {
return Meta.parse(k, v, METRIC_PREFIX);
public static String prefix(Key key) {
String row = key.getRow().toString();
if (row.startsWith(Meta.METRIC_PREFIX)) {
return Meta.METRIC_PREFIX;
} else if (row.startsWith(Meta.TAG_PREFIX)) {
return Meta.TAG_PREFIX;
} else if (row.startsWith(Meta.VALUE_PREFIX)) {
return Meta.VALUE_PREFIX;
}
throw new IllegalStateException("Invalid key in meta " + key.toString());
}
public static Meta parse(Key key) {
return parse(key, null);
}
public static Meta parse(Key key, Value value) {
String row = key.getRow().toString();
if (row.startsWith(Meta.METRIC_PREFIX)) {
return Meta.parse(key, value, Meta.METRIC_PREFIX);
} else if (row.startsWith(Meta.TAG_PREFIX)) {
return Meta.parse(key, value, Meta.TAG_PREFIX);
} else if (row.startsWith(Meta.VALUE_PREFIX)) {
return Meta.parse(key, value, Meta.VALUE_PREFIX);
}
throw new IllegalStateException("Invalid key in meta " + key.toString());
}
public static Meta parse(Key k, Value v, String prefix) {
@@ -438,6 +438,7 @@ public class DataStoreImpl implements DataStore {
@Override
public SuggestResponse suggest(SuggestRequest request) throws TimelyException {
SuggestResponse result = new SuggestResponse();
Scanner scanner = null;
try {
if (request.getType().equals("metrics")) {
Range range;
@@ -457,7 +458,7 @@ public class DataStoreImpl implements DataStore {
end.append(lastBytes, 0, lastBytes.length);
range = new Range(start, end);
}
Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
scanner.setRange(range);
List<String> metrics = new ArrayList<>();
for (Entry<Key, Value> metric : scanner) {
@@ -472,6 +473,10 @@ public class DataStoreImpl implements DataStore {
LOG.error("Error during suggest: " + ex.getMessage(), ex);
throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
"Error during suggest: " + ex.getMessage(), ex.getMessage(), ex);
} finally {
if (scanner != null) {
scanner.close();
}
}
return result;
}
@@ -506,28 +511,34 @@ public class DataStoreImpl implements DataStore {
tagPatterns.put(k, Pattern.compile(v));
});
try {
List<Result> resultField = new ArrayList<>();
Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
Key start = new Key(Meta.VALUE_PREFIX + msg.getQuery());
Key end = start.followingKey(PartialKey.ROW);
Range range = new Range(start, end);
scanner.setRange(range);
tags.keySet().forEach(k -> scanner.fetchColumnFamily(new Text(k)));
int total = 0;
for (Entry<Key, Value> entry : scanner) {
Meta metaEntry = Meta.parse(entry.getKey(), entry.getValue());
if (matches(metaEntry.getTagKey(), metaEntry.getTagValue(), tagPatterns)) {
if (resultField.size() < msg.getLimit()) {
Result r = new Result();
r.putTag(metaEntry.getTagKey(), metaEntry.getTagValue());
resultField.add(r);
final Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
try {
List<Result> resultField = new ArrayList<>();
Key start = new Key(Meta.VALUE_PREFIX + msg.getQuery());
Key end = start.followingKey(PartialKey.ROW);
Range range = new Range(start, end);
scanner.setRange(range);
tags.keySet().forEach(k -> scanner.fetchColumnFamily(new Text(k)));
int total = 0;
for (Entry<Key, Value> entry : scanner) {
Meta metaEntry = Meta.parse(entry.getKey(), entry.getValue());
if (matches(metaEntry.getTagKey(), metaEntry.getTagValue(), tagPatterns)) {
if (resultField.size() < msg.getLimit()) {
Result r = new Result();
r.putTag(metaEntry.getTagKey(), metaEntry.getTagValue());
resultField.add(r);
}
total++;
}
total++;
}
result.setResults(resultField);
result.setTotalResults(total);
result.setTime((int) (System.currentTimeMillis() - startMillis));
} finally {
if (scanner != null) {
scanner.close();
}
}
result.setResults(resultField);
result.setTotalResults(total);
result.setTime((int) (System.currentTimeMillis() - startMillis));
} catch (Exception ex) {
LOG.error("Error during lookup: " + ex.getMessage(), ex);
throw new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
@@ -798,70 +809,77 @@ public class DataStoreImpl implements DataStore {
public Set<Tag> getColumnFamilies(String metric, Map<String, String> requestedTags) throws TableNotFoundException {
Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
LOG.trace("Looking for requested tags: {}", tags);
Scanner meta = connector.createScanner(metaTable, Authorizations.EMPTY);
Text start = new Text(Meta.VALUE_PREFIX + metric);
Text end = new Text(Meta.VALUE_PREFIX + metric + "\\x0000");
end.append(new byte[] { (byte) 0xff }, 0, 1);
meta.setRange(new Range(start, end));
// Only look for the meta entries that match our tags, if any
boolean onlyFirstRow = false;
Entry<String, String> first = null;
// Set the columns on the meta scanner based on the first tag
// in the set of tags passed in the query. If no tags are present
// then we are only going to return the first tag name present in the
// meta table.
Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
if (tagIter.hasNext()) {
first = tagIter.next();
if (isTagValueRegex(first.getValue())) {
meta.fetchColumnFamily(new Text(first.getKey()));
} else {
meta.fetchColumn(new Text(first.getKey()), new Text(first.getValue()));
}
} else {
// grab all of the values found for the first tag for the metric
onlyFirstRow = true;
}
final boolean ONLY_RETURN_FIRST_TAG = onlyFirstRow;
Iterator<Entry<Key, Value>> iter = meta.iterator();
Iterator<Pair<String, String>> knownKeyValues = new Iterator<Pair<String, String>>() {
Text firstTag = null;
Text tagName = null;
Text tagValue = null;
@Override
public boolean hasNext() {
if (iter.hasNext()) {
Entry<Key, Value> metaEntry = iter.next();
if (null == firstTag) {
firstTag = metaEntry.getKey().getColumnFamily();
}
tagName = metaEntry.getKey().getColumnFamily();
tagValue = metaEntry.getKey().getColumnQualifier();
LOG.trace("Found tag entry {}={}", tagName, tagValue);
if (ONLY_RETURN_FIRST_TAG && !tagName.equals(firstTag)) {
return false;
}
return true;
Scanner meta = null;
try {
Map<String, String> tags = (requestedTags == null) ? new LinkedHashMap<>() : requestedTags;
LOG.trace("Looking for requested tags: {}", tags);
meta = connector.createScanner(metaTable, Authorizations.EMPTY);
Text start = new Text(Meta.VALUE_PREFIX + metric);
Text end = new Text(Meta.VALUE_PREFIX + metric + "\\x0000");
end.append(new byte[] { (byte) 0xff }, 0, 1);
meta.setRange(new Range(start, end));
// Only look for the meta entries that match our tags, if any
boolean onlyFirstRow = false;
Entry<String, String> first = null;
// Set the columns on the meta scanner based on the first tag
// in the set of tags passed in the query. If no tags are present
// then we are only going to return the first tag name present in the
// meta table.
Iterator<Entry<String, String>> tagIter = tags.entrySet().iterator();
if (tagIter.hasNext()) {
first = tagIter.next();
if (isTagValueRegex(first.getValue())) {
meta.fetchColumnFamily(new Text(first.getKey()));
} else {
meta.fetchColumn(new Text(first.getKey()), new Text(first.getValue()));
}
return false;
} else {
// grab all of the values found for the first tag for the metric
onlyFirstRow = true;
}
final boolean ONLY_RETURN_FIRST_TAG = onlyFirstRow;
Iterator<Entry<Key, Value>> iter = meta.iterator();
Iterator<Pair<String, String>> knownKeyValues = new Iterator<Pair<String, String>>() {
@Override
public Pair<String, String> next() {
LOG.trace("Returning tag {}={}", tagName, tagValue);
return new Pair<>(tagName.toString(), tagValue.toString());
Text firstTag = null;
Text tagName = null;
Text tagValue = null;
@Override
public boolean hasNext() {
if (iter.hasNext()) {
Entry<Key, Value> metaEntry = iter.next();
if (null == firstTag) {
firstTag = metaEntry.getKey().getColumnFamily();
}
tagName = metaEntry.getKey().getColumnFamily();
tagValue = metaEntry.getKey().getColumnQualifier();
LOG.trace("Found tag entry {}={}", tagName, tagValue);
if (ONLY_RETURN_FIRST_TAG && !tagName.equals(firstTag)) {
return false;
}
return true;
}
return false;
}
@Override
public Pair<String, String> next() {
LOG.trace("Returning tag {}={}", tagName, tagValue);
return new Pair<>(tagName.toString(), tagValue.toString());
}
};
// Expand the list of tags in the meta table for this metric that
// matches
// the pattern of the first tag in the query. The resulting set of tags
// will be used to fetch specific columns from the metric table.
return expandTagValues(first, knownKeyValues);
} finally {
if (meta != null) {
meta.close();
}
};
// Expand the list of tags in the meta table for this metric that
// matches
// the pattern of the first tag in the query. The resulting set of tags
// will be used to fetch specific columns from the metric table.
return expandTagValues(first, knownKeyValues);
}
}
private void setQueryColumns(ScannerBase scanner, String metric, Map<String, String> tags, Set<Tag> colFamValues)
@@ -0,0 +1,116 @@
package timely.store;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.OptionDescriber;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.WrappingIterator;
import org.apache.commons.collections4.trie.PatriciaTrie;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.api.model.Meta;
public class MetaAgeOffIterator extends WrappingIterator implements OptionDescriber {
private static final Logger LOG = LoggerFactory.getLogger(MetaAgeOffIterator.class);
public static final String AGE_OFF_PREFIX = "ageoff.";
public static final String DEFAULT_AGEOFF_KEY = "default";
/* set in init */
private PatriciaTrie<Long> ageoffs = null;
private Long currentTime = 0L;
private Long defaultAgeOff = 0L;
private Long minAgeOff = Long.MAX_VALUE;
private Long maxAgeOff = Long.MIN_VALUE;
@Override
public SortedKeyValueIterator<Key, Value> deepCopy(IteratorEnvironment env) {
MetaAgeOffIterator iter = new MetaAgeOffIterator();
iter.ageoffs = this.ageoffs;
iter.defaultAgeOff = this.defaultAgeOff;
iter.currentTime = this.currentTime;
iter.minAgeOff = this.minAgeOff;
iter.maxAgeOff = this.maxAgeOff;
return iter;
}
@Override
public void init(SortedKeyValueIterator<Key, Value> source, Map<String, String> options, IteratorEnvironment env)
throws IOException {
super.init(source, options, env);
validateOptions(options);
ageoffs = new PatriciaTrie<>();
options.forEach((k, v) -> {
if (k.startsWith(AGE_OFF_PREFIX)) {
String name = k.substring(AGE_OFF_PREFIX.length());
LOG.trace("Adding {} to Trie with value {}", name, Long.parseLong(v));
long ageoff = Long.parseLong(v);
this.minAgeOff = Math.min(this.minAgeOff, ageoff);
this.maxAgeOff = Math.max(this.maxAgeOff, ageoff);
ageoffs.put(name, ageoff);
}
});
defaultAgeOff = ageoffs.get(DEFAULT_AGEOFF_KEY);
currentTime = System.currentTimeMillis();
}
@Override
public void next() throws IOException {
super.next();
skipExpiredData();
}
@Override
public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive) throws IOException {
super.seek(range, columnFamilies, inclusive);
skipExpiredData();
}
private void skipExpiredData() throws IOException {
boolean done = false;
while (!done && super.hasTop()) {
Key top = super.getTopKey();
long dataAge = this.currentTime - top.getTimestamp();
// If less than any configured ageoff, then we want to return this K,V
if (dataAge < this.minAgeOff) {
done = true;
} else {
String metricName = Meta.parse(top).getMetric();
long metricAgeOff = ageoffs.getOrDefault(metricName, defaultAgeOff);
if (dataAge > metricAgeOff) {
// data beyond ageOff, so skip
super.next();
} else {
done = true;
}
}
}
}
@Override
public IteratorOptions describeOptions() {
return new IteratorOptions("metric-age-off", "Iterator that ages off meta entries for Timely metrics",
Collections.singletonMap(AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY, "default age off days"),
Collections.singletonList("Additional metric age off properties where the metric is specified as "
+ AGE_OFF_PREFIX
+ " the metric name and the value is an integer representing the number of days to keep"));
}
@Override
public boolean validateOptions(Map<String, String> options) {
if (null == options.get(AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY)) {
throw new IllegalArgumentException(DEFAULT_AGEOFF_KEY + " must be configured for MetaAgeOffFilter");
}
return true;
}
}
@@ -5,8 +5,10 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeSet;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
@@ -61,42 +63,60 @@ public class MetaCacheImpl implements MetaCache {
final ClientConfiguration aConf = new ClientConfiguration(Collections.singletonList(apacheConf));
String metaTable = configuration.getMetaTable();
Map<String, Map<String, Long>> metricMap = new HashMap<>();
Scanner scanner = null;
try {
final Instance instance = new ZooKeeperInstance(aConf);
Connector connector = instance.getConnector(accumuloConf.getUsername(),
new PasswordToken(accumuloConf.getPassword()));
Scanner scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
Key begin = new Key(Meta.VALUE_PREFIX);
Key end = new Key("w" + '\0');
Range range = new Range(begin, true, end, false);
scanner.setRange(range);
scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
LOG.debug("Begin scanning " + metaTable);
Range metricRange = new Range(new Key(Meta.METRIC_PREFIX), true, new Key(Meta.METRIC_PREFIX + '\0'), false);
scanner.setRange(metricRange);
Set<String> allMetrics = new TreeSet<>();
for (Map.Entry<Key, Value> entry : scanner) {
Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.VALUE_PREFIX);
String metric = meta.getMetric();
String tagKey = meta.getTagKey();
Map<String, Long> tagMap = metricMap.get(metric);
if (tagMap == null) {
tagMap = new HashMap<>();
metricMap.put(metric, tagMap);
}
long numTagValues = tagMap.getOrDefault(tagKey, 0l);
if (entry.getKey().getTimestamp() > oldestTimestamp) {
tagMap.put(tagKey, ++numTagValues);
if (numTagValues <= configuration.getMetaCache().getMaxTagValues()) {
cache.put(meta, DUMMY);
Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.METRIC_PREFIX);
allMetrics.add(meta.getMetric());
}
for (String currMetric : allMetrics) {
Key begin = new Key(Meta.VALUE_PREFIX + currMetric);
Key end = new Key(Meta.VALUE_PREFIX + currMetric + '\0');
Range range = new Range(begin, true, end, false);
scanner.setRange(range);
Iterator<Map.Entry<Key, Value>> itr = scanner.iterator();
boolean maxedOutValues = false;
while (itr.hasNext() && !maxedOutValues) {
Map.Entry<Key, Value> entry = itr.next();
Meta meta = Meta.parse(entry.getKey(), entry.getValue(), Meta.VALUE_PREFIX);
String metric = meta.getMetric();
String tagKey = meta.getTagKey();
Map<String, Long> tagMap = metricMap.get(metric);
if (tagMap == null) {
tagMap = new HashMap<>();
metricMap.put(metric, tagMap);
}
long numTagValues = tagMap.getOrDefault(tagKey, 0l);
if (entry.getKey().getTimestamp() > oldestTimestamp) {
tagMap.put(tagKey, ++numTagValues);
if (numTagValues <= configuration.getMetaCache().getMaxTagValues()) {
cache.put(meta, DUMMY);
} else {
// found maxTagValues on this refresh, so remove this key
cache.invalidate(meta);
maxedOutValues = true;
}
} else {
// found maxTagValues on this refresh, so remove this key
// this entry is more than expirationMinutes old, so remove it
cache.invalidate(meta);
}
} else {
// this entry is more than expirationMinutes old, so remove it
cache.invalidate(meta);
}
}
LOG.debug("Finished scanning " + metaTable);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
} finally {
if (scanner != null) {
scanner.close();
}
}
}
@@ -0,0 +1,241 @@
package timely.store;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.SortedMapIterator;
import org.junit.Assert;
import org.junit.Test;
import timely.adapter.accumulo.MetaAdapter;
import timely.adapter.accumulo.MetricAdapter;
public class MetaAgeOffIteratorTest {
private static final Long TEST_TIME = System.currentTimeMillis() - 1000;
private static final Integer ONE_DAY = 86400000;
private static final Value EMPTY_VALUE = new Value(new byte[0]);
private static final Collection<ByteSequence> columnFamilies = new ArrayList<>();
@Test(expected = IllegalArgumentException.class)
public void testDefaultMissing() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
iter.init(source, options, null);
}
@Test
public void testDefault() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
seen++;
iter.next();
}
Assert.assertEquals(6, seen);
}
@Test
public void testMixed() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + 5), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
seen++;
iter.next();
}
Assert.assertEquals(12, seen);
}
@Test
public void testAgeoffMixedMetricKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.idle", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(
k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
seen++;
iter.next();
}
Assert.assertEquals(7, seen);
}
@Test
public void testAgeoffMixedTagKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag1", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag2", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag3", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag5", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.idle", "tag6", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag1", TEST_TIME - (3 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag2", TEST_TIME - (2 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag3", TEST_TIME - (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag5", TEST_TIME + (1 * ONE_DAY)), new Value(new byte[0]));
table.put(MetaAdapter.createTagKey("sys.cpu.user", "tag6", TEST_TIME + (2 * ONE_DAY)), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(
k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
seen++;
iter.next();
}
Assert.assertEquals(7, seen);
}
@Test
public void testAgeoffMixedValueKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag1", "value1", TEST_TIME - (3 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag2", "value2", TEST_TIME - (2 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag3", "value3", TEST_TIME - (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag4", "value4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag5", "value5", TEST_TIME + (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.idle", "tag6", "value6", TEST_TIME + (2 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag1", "value1", TEST_TIME - (3 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag2", "value2", TEST_TIME - (2 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag3", "value3", TEST_TIME - (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag4", "value4", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag5", "value5", TEST_TIME + (1 * ONE_DAY)),
new Value(new byte[0]));
table.put(MetaAdapter.createValueKey("sys.cpu.user", "tag6", "value6", TEST_TIME + (2 * ONE_DAY)),
new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1 * ONE_DAY));
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "sys.cpu.user", Integer.toString(2 * ONE_DAY));
iter.init(source, options, null);
iter.seek(new Range(), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(
k.getTimestamp() >= (TEST_TIME - (2 * ONE_DAY)) && k.getTimestamp() <= TEST_TIME + (2 * ONE_DAY));
seen++;
iter.next();
}
Assert.assertEquals(7, seen);
}
@Test
public void testSeekPastEndKey() throws Exception {
SortedMap<Key, Value> table = new TreeMap<>();
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 1), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 2), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 3), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 4), new Value(new byte[0]));
table.put(MetaAdapter.createMetricKey("sys.cpu.user", TEST_TIME + 5), new Value(new byte[0]));
SortedKeyValueIterator<Key, Value> source = new SortedMapIterator(table);
MetaAgeOffIterator iter = new MetaAgeOffIterator();
HashMap<String, String> options = new HashMap<>();
options.put(MetaAgeOffIterator.AGE_OFF_PREFIX + "default", Integer.toString(1));
iter.init(source, options, null);
iter.seek(new Range(new Key("sys.cpu.user"), true,
new Key(MetricAdapter.encodeRowKey("sys.cpu.user", TEST_TIME + 3), new byte[0], new byte[0],
new byte[0], TEST_TIME + 3),
true), columnFamilies, true);
int seen = 0;
while (iter.hasTop()) {
Key k = iter.getTopKey();
Assert.assertTrue(k.getTimestamp() >= TEST_TIME && k.getTimestamp() <= TEST_TIME + 5);
seen++;
iter.next();
}
Assert.assertEquals(0, seen);
}
}