Compare commits

..
1 Commits
Author SHA1 Message Date
Bill Oley 1597ec61ae scripts, interval, collectd, try to build grafana plugin 2024-03-28 16:42:59 -04:00
56 changed files with 1913 additions and 1136 deletions

No files matched your search

+1 -3
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>timely-analytics</artifactId>
<!--
@@ -71,8 +71,6 @@
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<compilerId>jdt</compilerId>
<source>11</source>
<target>11</target>
</configuration>
<dependencies>
<dependency>
+21 -17
View File
@@ -1,17 +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")
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}')
@@ -19,6 +19,7 @@ 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"
@@ -31,12 +32,18 @@ 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 ${NATIVE_DIR} ]]; then
mkdir -p "${NATIVE_DIR}"
if [[ -e ${TMP_DIR} ]]; then
rm -rf "${TMP_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"
@@ -46,15 +53,12 @@ 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
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}"
pushd "${BASE_DIR}/bin" || exit
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_tcnative_linux_"${HW}".so
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_transport_native_epoll_"${HW}".so
popd || exit
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
fi
echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-balancer-*-exec.jar --instance=${INSTANCE}"
+1 -5
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>timely-balancer</artifactId>
<description>Timely Balancer</description>
@@ -44,10 +44,6 @@
<groupId>io.github.classgraph</groupId>
<artifactId>classgraph</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-classes-epoll</artifactId>
</dependency>
<dependency>
<groupId>net.sourceforge.javacsv</groupId>
<artifactId>javacsv</artifactId>
@@ -91,21 +91,17 @@ 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) throws Exception {
HealthChecker healthChecker) {
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() {
@@ -920,8 +916,10 @@ public class MetricResolver {
protected void createAssignmentFile() {
try {
log.info("Creating assignment file: " + this.assignmentFile);
this.fs.create(this.assignmentFile);
log.info("Creating assignment file: " + balancerProperties.getAssignmentFile());
Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
FileSystem fs = getFileSystem(assignmentFile);
fs.create(assignmentFile);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
@@ -929,7 +927,9 @@ public class MetricResolver {
protected long getLastWrittenToHdfsTimestamp() {
try {
FileStatus fileStatus = this.fs.getFileStatus(this.assignmentFile);
Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
FileSystem fs = getFileSystem(assignmentFile);
FileStatus fileStatus = fs.getFileStatus(assignmentFile);
return fileStatus.getModificationTime();
} catch (FileNotFoundException e) {
createAssignmentFile();
@@ -960,15 +960,16 @@ 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();
FSDataInputStream iStream = this.fs.open(this.assignmentFile);
reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
FileSystem fs = getFileSystem(assignmentFile);
FSDataInputStream iStream = fs.open(assignmentFile);
CsvReader reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
reader.setUseTextQualifier(false);
// skip the headers
@@ -1030,9 +1031,6 @@ public class MetricResolver {
} catch (Exception e) {
log.error(e.getMessage(), e);
} finally {
if (reader != null) {
reader.close();
}
balancerLock.writeLock().unlock();
try {
assignmentsIPRWLock.readLock().release();
@@ -1054,10 +1052,12 @@ public class MetricResolver {
nonCachedMetricsLocalLock.readLock().lock();
try {
if (!metricToHostMap.isEmpty()) {
if (!this.fs.exists(this.assignmentFile.getParent())) {
this.fs.mkdirs(this.assignmentFile.getParent());
Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
FileSystem fs = getFileSystem(assignmentFile);
if (!fs.exists(assignmentFile.getParent())) {
fs.mkdirs(assignmentFile.getParent());
}
FSDataOutputStream oStream = this.fs.create(this.assignmentFile, true);
FSDataOutputStream oStream = fs.create(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.info("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
log.debug("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()));
@@ -50,7 +50,7 @@ public class BalancerConfiguration {
@Bean
@ConditionalOnMissingBean
public MetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
HealthChecker healthChecker) throws Exception {
HealthChecker healthChecker) {
MetricResolver metricResolver = new MetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
metricResolver.start();
return metricResolver;
@@ -48,7 +48,7 @@ public class TestBalancerConfiguration {
@Bean
public TestMetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
HealthChecker healthChecker) throws Exception {
HealthChecker healthChecker) {
TestMetricResolver metricResolver = new TestMetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
metricResolver.start();
return metricResolver;
@@ -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) throws Exception {
HealthChecker healthChecker) {
super(curatorFramework, balancerProperties, cacheProperties, healthChecker);
}
+2 -2
View File
@@ -4,13 +4,13 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>timely-client</artifactId>
<properties>
<!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
<!-- use .dev0 instead of -SNAPSHOT -->
<python.client.version>2.1.6.dev0</python.client.version>
<python.client.version>2.1.4.dev0</python.client.version>
</properties>
<dependencies>
<dependency>
@@ -45,21 +45,18 @@ public class MetricTest {
public void testJson() throws IOException {
ObjectMapper mapper = new ObjectMapper();
Long now = System.currentTimeMillis();
String ts = Long.toString(now);
String expectedJson = "{\"name\":\"m1\",\"timestamp\":1,\"measure\":1.0,\"tags\":[{\"k1\":\"v1\"}]}";
String expectedJson = "{\"name\":\"m1\",\"timestamp\":" + ts + ",\"measure\":1.0,\"tags\":[{\"k1\":\"v1\"}]}";
Metric m1 = Metric.newBuilder().name("m1").tag("k1", "v1").value(now, 1.0).build();
Metric m1 = Metric.newBuilder().name("m1").tag("k1", "v1").value(1, 1.0).build();
Metric expectedMetric = mapper.readValue(expectedJson, Metric.class);
assertTrue(m1.equals(expectedMetric));
expectedJson = "{\"name\":\"m1\",\"tags\":[{\"k1\":\"v1\"}],\"timestamp\":" + ts + ",\"measure\":5.0}";
expectedJson = "{\"name\":\"m1\",\"tags\":[{\"k1\":\"v1\"}],\"timestamp\":5,\"measure\":5.0}";
expectedMetric = mapper.readValue(expectedJson, Metric.class);
assertEquals(expectedMetric.getValue().getTimestamp(), now);
assertEquals((long) expectedMetric.getValue().getTimestamp(), 5L);
assertEquals(expectedMetric.getValue().getMeasure(), 5.0D, 0.0);
}
@@ -14,32 +14,11 @@ public class MetricParserTest {
@Test
public void testParseWithEscapedCharacters() {
Long now = System.currentTimeMillis();
String ts = Long.toString(now);
MetricParser parser = new MetricParser();
Metric m = parser.parse("put mymetric " + ts + " 5.0 tag1=value1,value1 tag2=value2=value2");
Metric m = parser.parse("put mymetric 12341234 5.0 tag1=value1,value1 tag2=value2=value2");
Assert.assertEquals("mymetric", m.getName());
Assert.assertEquals(now, m.getValue().getTimestamp());
Assert.assertEquals(5.0, (double) m.getValue().getMeasure(), 0);
List<Tag> expected = new ArrayList<>();
expected.add(new Tag("tag1", "value1,value1"));
expected.add(new Tag("tag2", "value2=value2"));
Assert.assertEquals(expected, m.getTags());
}
@Test
public void testParseWithSecondsTimestamp() {
Long now = System.currentTimeMillis() / 1000;
String ts = Long.toString(now);
MetricParser parser = new MetricParser();
Metric m = parser.parse("put mymetric " + ts + " 5.0 tag1=value1,value1 tag2=value2=value2");
Assert.assertEquals("mymetric", m.getName());
Assert.assertEquals((Long) (now * 1000L), m.getValue().getTimestamp());
Assert.assertEquals(12341234, (long) m.getValue().getTimestamp());
Assert.assertEquals(5.0, (double) m.getValue().getMeasure(), 0);
List<Tag> expected = new ArrayList<>();
expected.add(new Tag("tag1", "value1,value1"));
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>collectd-nsq-plugin</artifactId>
<dependencies>
@@ -79,7 +79,6 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
} catch (IOException e1) {}
try {
clientPool.invalidateObject(client);
client = null;
} catch (Exception e2) {}
}
} finally {
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>collectd-plugin-parent</artifactId>
<dependencies>
@@ -3,7 +3,6 @@ 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;
@@ -26,9 +25,6 @@ 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-_]+)");
@@ -68,8 +64,6 @@ 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());
}
}
@@ -77,8 +71,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) {
addTag(m, parts[0], parts[1]);
if (parts.length == 2 && !parts[0].isBlank() && !parts[1].isBlank()) {
m.put(parts[0].trim(), parts[1].trim());
}
}
}
@@ -105,10 +99,9 @@ public abstract class CollectDPluginParent {
} else {
addTag(tagMap, "host", host);
}
String[] hostSplit = host.split("\\.");
int nIdx = hostSplit[0].lastIndexOf('n');
int nIdx = host.indexOf('n');
if (-1 != nIdx) {
addTag(tagMap, "rack", hostSplit[0].substring(0, nIdx));
addTag(tagMap, "rack", host.substring(0, nIdx));
}
tagMap.putAll(addlTags);
@@ -119,53 +112,34 @@ 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 >= 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);
if (parts.length % 2 == 1 && parts.length >= 3) {
switch(parts[0]) {
case "dwingest":
if (parts.length >= 4) {
metric.append(STATSD_PREFIX).append(PERIOD).append(parts[1]).append(PERIOD).append(parts[3]);
instance = parts[0];
}
break;
case "dwquery":
if (parts.length >= 4) {
metric.append(STATSD_PREFIX).append(PERIOD).append(parts[1]).append(PERIOD).append(parts[3]);
addTag(tagMap, "queryId", parts[0]);
}
break;
default:
// EtsyStatsD format -- metric.(tagName.tagValue)*
metric.append(STATSD_PREFIX).append(PERIOD).append(parts[0]);
for (int x = 1; x < parts.length; x += 2) {
addTag(tagMap, parts[x], parts[x + 1]);
}
}
}
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) {
addTag(tagMap, parts[x], parts[x + 1]);
}
} else if (statsd_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 recordName is typically duplicative
@@ -255,12 +229,6 @@ 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("-");
@@ -270,7 +238,6 @@ 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);
@@ -297,15 +264,9 @@ 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(" "));
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);
}
String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(), tagString);
logDebug(String.format("Writing: %s", datapoint));
write(datapoint + "\n", out);
}
}
@@ -356,4 +317,5 @@ public abstract class CollectDPluginParent {
private boolean notEmpty(String arg) {
return (null != arg) && !(arg.equals(""));
}
}
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>collectd-timely-plugin</artifactId>
<dependencies>
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>timely-common</artifactId>
<dependencies>
@@ -32,7 +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;
private Optional<String> interval = Optional.empty();
public RateOption() {}
@@ -85,11 +85,11 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
this.resetValue = resetValue;
}
public String getInterval() {
public Optional<String> getInterval() {
return interval;
}
public void setInterval(String interval) {
public void setInterval(Optional<String> interval) {
this.interval = interval;
}
@@ -221,7 +221,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
private String aggregator;
private String metric;
private boolean rate = false;
private RateOption rateOptions = new RateOption();
private RateOption rateOptions = null;
private Optional<String> downsample = Optional.empty();
private Map<String,String> tags = new LinkedHashMap<>();
private Collection<Filter> filters = new ArrayList<>();
@@ -547,7 +547,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
options.setCounter(rateOptions[x].endsWith("counter"));
break;
case 1:
options.setInterval(rateOptions[x]);
options.setInterval(Optional.of(rateOptions[x]));
break;
case 2:
options.setCounterMax(Long.parseLong(rateOptions[x]));
@@ -1,7 +1,6 @@
package timely.model;
import java.io.Serializable;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.TreeSet;
@@ -25,12 +24,6 @@ import timely.model.parse.TagParser;
@JsonPropertyOrder({"name"})
public class Metric implements Serializable {
// Capture the time that this class is loaded by the JVM for the purposes of
// comparing the timestamp. Some systems, like Telegraf's OpenTSDB output
// plugin, send the timestamp in seconds.
private static final long TIME_LOADED = System.currentTimeMillis();
private static final long TS_THRESHOLD = Duration.ofMillis(TIME_LOADED).minusDays(365).toMillis();
private static final long serialVersionUID = 1L;
private String name;
@@ -183,9 +176,6 @@ public class Metric implements Serializable {
}
public Builder value(long timestamp, double measure) {
if (timestamp <= TS_THRESHOLD) {
timestamp = timestamp * 1000;
}
metric.setValue(new Value(timestamp, measure));
return this;
}
@@ -84,13 +84,13 @@ public class MetricAdapterTest {
@Test
public void testParse() throws Exception {
Long ts = System.currentTimeMillis();
PairLexicoder<String,Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
byte[] value = new byte[Double.BYTES];
ByteBuffer.wrap(value).putDouble(2.0D);
PairLexicoder<Long,String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(ts, "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), ts);
Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000), "tag2=value2,tag3=value3")),
"(a&b)|(c&d)".getBytes(), 1000);
Value v = new Value(value);
Metric m = MetricAdapter.parse(k, v);
Assert.assertEquals("sys.cpu.user", m.getName());
@@ -99,20 +99,19 @@ public class MetricAdapterTest {
tags.add(new Tag("tag2=value2"));
tags.add(new Tag("tag3=value3"));
Assert.assertEquals(tags, m.getTags());
Assert.assertEquals(ts, m.getValue().getTimestamp());
Assert.assertEquals(new Long(1000), m.getValue().getTimestamp());
Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
}
@Test
public void testParseWithEscapes() {
Long ts = System.currentTimeMillis();
PairLexicoder<String,Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
byte[] value = new byte[Double.BYTES];
ByteBuffer.wrap(value).putDouble(2.0D);
PairLexicoder<Long,String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(ts, "tag2=test\\,value2,tag3=val\\=ue3")),
"(a&b)|(c&d)".getBytes(), ts);
Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000), "tag2=test\\,value2,tag3=val\\=ue3")),
"(a&b)|(c&d)".getBytes(), 1000);
Value v = new Value(value);
Metric m = MetricAdapter.parse(k, v);
Assert.assertEquals("sys.cpu.user", m.getName());
@@ -121,19 +120,19 @@ public class MetricAdapterTest {
tags.add(new Tag("tag2=test,value2"));
tags.add(new Tag("tag3=val\\=ue3"));
Assert.assertEquals(tags, m.getTags());
Assert.assertEquals(ts, m.getValue().getTimestamp());
Assert.assertEquals(new Long(1000), m.getValue().getTimestamp());
Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
}
@Test
public void testParseWithViz() throws Exception {
Long ts = System.currentTimeMillis();
PairLexicoder<String,Long> rowCoder = new PairLexicoder<>(new StringLexicoder(), new LongLexicoder());
byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", ts));
byte[] row = rowCoder.encode(new ComparablePair<>("sys.cpu.user", 1000L));
byte[] value = new byte[Double.BYTES];
ByteBuffer.wrap(value).putDouble(2.0D);
PairLexicoder<Long,String> colQualCoder = new PairLexicoder<>(new LongLexicoder(), new StringLexicoder());
Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(ts, "tag2=value2,tag3=value3")), "(a&b)|(c&d)".getBytes(), ts);
Key k = new Key(row, "tag1=value1".getBytes(), colQualCoder.encode(new ComparablePair<>(new Long(1000), "tag2=value2,tag3=value3")),
"(a&b)|(c&d)".getBytes(), 1000);
Value v = new Value(value);
Metric m = MetricAdapter.parse(k, v, true);
Assert.assertEquals("sys.cpu.user", m.getName());
@@ -143,7 +142,7 @@ public class MetricAdapterTest {
tags.add(new Tag("tag3=value3"));
tags.add(new Tag("viz=(a&b)|(c&d)"));
Assert.assertEquals(tags, m.getTags());
Assert.assertEquals(ts, m.getValue().getTimestamp());
Assert.assertEquals(new Long(1000), m.getValue().getTimestamp());
Assert.assertEquals(2.0D, m.getValue().getMeasure(), 0.0D);
}
+1 -1
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>timely-app</artifactId>
<packaging>pom</packaging>
+2 -2
View File
@@ -28,8 +28,8 @@
}
],
"screenshots": [],
"version": "2.1.6-SNAPSHOT",
"updated": "2024-04-10"
"version": "2.1.4-SNAPSHOT",
"updated": "2024-03-13"
},
"dependencies": {
"classnames": "2.2.6",
+91 -6
View File
@@ -1,11 +1,96 @@
module github.com/NationalSecurityAgency/timely-datasource
go 1.14
go 1.21
toolchain go1.22.1
require (
github.com/grafana/grafana-plugin-sdk-go v0.111.0
github.com/hashicorp/go-hclog v0.16.1
github.com/mattn/go-colorable v0.1.6 // indirect
github.com/prometheus/common v0.29.0 // indirect
golang.org/x/net v0.17.0 // indirect
github.com/grafana/grafana-plugin-sdk-go v0.218.0
github.com/hashicorp/go-hclog v1.6.2
)
require (
github.com/BurntSushi/toml v1.3.2 // indirect
github.com/apache/arrow/go/v15 v15.0.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cenkalti/backoff/v4 v4.2.1 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/cheekybits/genny v1.0.0 // indirect
github.com/chromedp/cdproto v0.0.0-20220208224320-6efb837e6bc2 // indirect
github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
github.com/elazarl/goproxy v0.0.0-20230731152917-f99041a5c027 // indirect
github.com/fatih/color v1.15.0 // indirect
github.com/fsnotify/fsnotify v1.4.9 // indirect
github.com/getkin/kin-openapi v0.120.0 // indirect
github.com/go-logr/logr v1.4.1 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/go-openapi/jsonpointer v0.19.6 // indirect
github.com/go-openapi/swag v0.22.4 // indirect
github.com/goccy/go-json v0.10.2 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/protobuf v1.5.3 // indirect
github.com/google/flatbuffers v23.5.26+incompatible // indirect
github.com/google/go-cmp v0.6.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/mux v1.8.0 // indirect
github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.0 // indirect
github.com/hashicorp/go-plugin v1.6.0 // indirect
github.com/hashicorp/yamux v0.1.1 // indirect
github.com/invopop/yaml v0.2.0 // indirect
github.com/josharian/intern v1.0.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/compress v1.16.7 // indirect
github.com/klauspost/cpuid/v2 v2.2.5 // indirect
github.com/magefile/mage v1.15.0 // indirect
github.com/mailru/easyjson v0.7.7 // indirect
github.com/mattetti/filebuffer v1.0.1 // indirect
github.com/mattn/go-colorable v0.1.13 // indirect
github.com/mattn/go-isatty v0.0.19 // indirect
github.com/mattn/go-runewidth v0.0.9 // indirect
github.com/mitchellh/go-testing-interface v1.14.1 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect
github.com/oklog/run v1.1.0 // indirect
github.com/olekukonko/tablewriter v0.0.5 // indirect
github.com/perimeterx/marshmallow v1.1.5 // indirect
github.com/pierrec/lz4/v4 v4.1.18 // indirect
github.com/prometheus/client_golang v1.18.0 // indirect
github.com/prometheus/client_model v0.5.0 // indirect
github.com/prometheus/common v0.46.0 // indirect
github.com/prometheus/procfs v0.12.0 // indirect
github.com/russross/blackfriday/v2 v2.1.0 // indirect
github.com/smartystreets/goconvey v1.6.4 // indirect
github.com/unknwon/bra v0.0.0-20200517080246-1e3013ecaff8 // indirect
github.com/unknwon/com v1.0.1 // indirect
github.com/unknwon/log v0.0.0-20150304194804-e617c87089d3 // indirect
github.com/urfave/cli v1.22.14 // indirect
github.com/zeebo/xxh3 v1.0.2 // indirect
go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.49.0 // indirect
go.opentelemetry.io/contrib/propagators/jaeger v1.22.0 // indirect
go.opentelemetry.io/contrib/samplers/jaegerremote v0.18.0 // indirect
go.opentelemetry.io/otel v1.24.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.24.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.24.0 // indirect
go.opentelemetry.io/otel/metric v1.24.0 // indirect
go.opentelemetry.io/otel/sdk v1.24.0 // indirect
go.opentelemetry.io/otel/trace v1.24.0 // indirect
go.opentelemetry.io/proto/otlp v1.1.0 // indirect
golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
golang.org/x/mod v0.13.0 // indirect
golang.org/x/net v0.22.0 // indirect
golang.org/x/sys v0.18.0 // indirect
golang.org/x/text v0.14.0 // indirect
golang.org/x/tools v0.14.0 // indirect
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20240123012728-ef4313101c80 // indirect
google.golang.org/grpc v1.62.1 // indirect
google.golang.org/protobuf v1.33.0 // indirect
gopkg.in/fsnotify/fsnotify.v1 v1.4.7 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
+229 -642
View File
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "timely-datasource",
"version": "2.1.6-SNAPSHOT",
"version": "2.1.4-SNAPSHOT",
"description": "Timely",
"backend": true,
"alerting": true,
+7
View File
@@ -398,6 +398,13 @@ func (td *TimelyDatasource) convertTimelyQuery(timelyQueryForm *TimelyQueryForm)
timelyQuery.RateOptions.Counter = timelyQueryForm.IsCounter
}
if len(timelyQueryForm.RateInterval) > 0 {
rateInterval, error := strconv.ParseInt(timelyQueryForm.RateInterval, 10, 32)
if error == nil {
timelyQuery.RateOptions.Interval = int32(rateInterval)
}
}
if len(timelyQueryForm.CounterMax) > 0 {
counterMax, error := strconv.ParseInt(timelyQueryForm.CounterMax, 10, 32)
if error == nil {
+4 -1
View File
@@ -11,6 +11,7 @@ type TimelyQueryForm struct {
Tags map[string]string `json:"tags"`
Filters []Filter `json:"filters"`
ShouldComputeRate bool `json:"shouldComputeRate"`
RateInterval string `json:"rateInterval"`
IsCounter bool `json:"isCounter"`
CounterMax string `json:"counterMax"`
CounterResetValue string `json:"counterResetValue"`
@@ -34,6 +35,7 @@ type TimelyQueryServer struct {
Tags map[string]string `json:"tags"`
Filters []Filter `json:"filters"`
ShouldComputeRate bool `json:"shouldComputeRate"`
RateInterval string `json:"rateInterval"`
IsCounter bool `json:"isCounter"`
CounterMax string `json:"counterMax"`
CounterResetValue string `json:"counterResetValue"`
@@ -50,6 +52,7 @@ type Filter struct {
}
type RateOptions struct {
Interval int32 `json:"interval"`
Counter bool `json:"counter"`
CounterMax int32 `json:"counterMax"`
ResetValue int32 `json:"resetValue"`
@@ -96,4 +99,4 @@ type TimelyErrorResponse struct {
ResponseCode int32 `json:responseCode`
Message string `json:message`
DetailMessage string `json:detailMessage`
}
}
+19 -2
View File
@@ -37,6 +37,7 @@ export class QueryEditor extends Component<Props, State> {
downsampleInterval: '1m',
downsampleAggregator: 'avg',
shouldComputeRate: false,
rateInterval: '1m',
isCounter: false,
tags: {},
};
@@ -123,6 +124,15 @@ export class QueryEditor extends Component<Props, State> {
});
};
onRateIntervalChange = (option: React.FormEvent<HTMLInputElement>) => {
const value = e.target.value;
this.setState((state, props) => {
var query = Object.assign(state.query, { rateInterval: value });
this.updateQuery(query);
return { query: query };
});
};
onCounterChange = (option: React.FormEvent<HTMLInputElement>) => {
const checked = option.currentTarget.checked;
this.setState((state, props) => {
@@ -247,7 +257,7 @@ export class QueryEditor extends Component<Props, State> {
const { metric, aggregator, alias } = this.state.query;
const { downsampleInterval, downsampleAggregator, disableDownsampling } = this.state.query;
const { tags } = this.state.query;
const { shouldComputeRate, isCounter, counterResetValue, counterMax } = this.state.query;
const { shouldComputeRate, rateInterval, isCounter, counterResetValue, counterMax } = this.state.query;
const aggregatorOptions: Array<{}> = _.map(this.state.aggregatorTypes, (value: string) => ({
label: value,
value: value,
@@ -345,7 +355,7 @@ export class QueryEditor extends Component<Props, State> {
spellCheck={false}
value={downsampleInterval || ''}
onChange={this.onDownsampleChange}
placeholder={'interval'}
placeholder={'downsample interval'}
/>
</div>
<div className="gf-form timely">
@@ -390,6 +400,13 @@ export class QueryEditor extends Component<Props, State> {
onChange={this.onRateChange}
/>
<div hidden={!shouldComputeRate}>
<Input
className={'min-width-6 flex-grow-1'}
spellCheck={false}
value={rateInterval || ''}
onChange={this.onRateIntervalChange}
placeholder={'rate interval'}
/>
<Switch
label={'Counter'}
labelClass="width-10"
+1
View File
@@ -9,6 +9,7 @@ export interface TimelyQueryForm extends DataQuery {
downsampleAggregator?: string;
tags: { [key: string]: string };
shouldComputeRate?: boolean;
rateInterval?: string;
isCounter?: boolean;
counterMax?: string;
counterResetValue?: string;
+7 -69
View File
@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Timely</name>
<description>A secure and scalable time series database based on Apache Accumulo</description>
@@ -35,8 +35,8 @@
<assembly.tarLongFileMode>posix</assembly.tarLongFileMode>
<collectd-api-jar>/usr/share/collectd/java/collectd-api.jar</collectd-api-jar>
<docker.image.prefix>timely/</docker.image.prefix>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<maven.compiler.source>11</maven.compiler.source>
<maven.compiler.target>11</maven.compiler.target>
<!-- Configure the os-maven-plugin extension to expand the classifier on -->
<!-- Fedora-"like" systems. -->
<os.detection.classifierWithLikes>fedora</os.detection.classifierWithLikes>
@@ -44,7 +44,7 @@
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<spotbugs.excludes.file>${project.basedir}/src/main/spotbugs/excludes.xml</spotbugs.excludes.file>
<surefire.forkCount>1C</surefire.forkCount>
<version.accumulo>4.0.0-SNAPSHOT</version.accumulo>
<version.accumulo>2.1.2</version.accumulo>
<version.caffeine>3.1.8</version.caffeine>
<version.classgraph>4.8.165</version.classgraph>
<version.collectd>5.7.1</version.collectd>
@@ -56,7 +56,7 @@
<version.commons-text>1.11.0</version.commons-text>
<version.compression.gorilla>2.1.1</version.compression.gorilla>
<version.curator>5.5.0</version.curator>
<version.datawave-in-memory-accumulo>8.0.0-SNAPSHOT</version.datawave-in-memory-accumulo>
<version.datawave-in-memory-accumulo>3.0.1</version.datawave-in-memory-accumulo>
<version.disruptor>3.4.4</version.disruptor>
<version.easymock>4.0.2</version.easymock>
<version.eclipse-lifecycle>1.0.0</version.eclipse-lifecycle>
@@ -79,7 +79,7 @@
<version.maven-assembly-plugin>3.6.0</version.maven-assembly-plugin>
<version.maven-build-helper-plugin>3.5.0</version.maven-build-helper-plugin>
<version.maven-buildnumber-plugin>1.4</version.maven-buildnumber-plugin>
<version.maven-compiler-plugin>3.14.1</version.maven-compiler-plugin>
<version.maven-compiler-plugin>3.12.1</version.maven-compiler-plugin>
<version.maven-dependency-plugin>3.6.1</version.maven-dependency-plugin>
<version.maven-docker-plugin>1.2.1</version.maven-docker-plugin>
<version.maven-failsafe-plugin>3.2.5</version.maven-failsafe-plugin>
@@ -89,7 +89,7 @@
<version.maven-os-plugin>1.7.1</version.maven-os-plugin>
<version.maven-rpm-plugin>2.3.0</version.maven-rpm-plugin>
<version.maven-shade-plugin>3.5.1</version.maven-shade-plugin>
<version.maven-spotbugs-plugin>4.9.8.2</version.maven-spotbugs-plugin>
<version.maven-spotbugs-plugin>4.7.3.0</version.maven-spotbugs-plugin>
<version.maven-surefire-plugin>3.2.5</version.maven-surefire-plugin>
<version.micrometer>1.12.2</version.micrometer>
<version.netty>4.1.106.Final</version.netty>
@@ -845,45 +845,6 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<version>${version.maven-dependency-plugin}</version>
<dependencies>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>${version.commons-lang3}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-1.2-api</artifactId>
<version>${version.log4j2}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>${version.log4j2}</version>
</dependency>
<dependency>
<groupId>org.apache.maven.doxia</groupId>
<artifactId>doxia-site-renderer</artifactId>
<version>2.0.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.maven.reporting</groupId>
<artifactId>maven-reporting-impl</artifactId>
<version>4.0.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -1008,29 +969,6 @@
<threshold>Medium</threshold>
<xmlOutput>true</xmlOutput>
</configuration>
<dependencies>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-1.2-api</artifactId>
<version>${version.log4j2}</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>${version.log4j2}</version>
</dependency>
<dependency>
<groupId>org.apache.maven.reporting</groupId>
<artifactId>maven-reporting-impl</artifactId>
<version>4.0.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
</plugin>
<plugin>
<groupId>org.jacoco</groupId>
+21 -17
View File
@@ -1,17 +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")
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}')
@@ -19,6 +19,7 @@ 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"
@@ -31,12 +32,18 @@ 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 ${NATIVE_DIR} ]]; then
mkdir -p "${NATIVE_DIR}"
if [[ -e ${TMP_DIR} ]]; then
rm -rf "${TMP_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"
@@ -46,15 +53,12 @@ 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
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}"
pushd "${BASE_DIR}/bin" || exit
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_tcnative_linux_"${HW}".so
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_transport_native_epoll_"${HW}".so
popd || exit
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
fi
echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-server*-exec.jar timelyserver --instance=${INSTANCE}"
+12 -27
View File
@@ -1,50 +1,35 @@
#!/bin/bash
if [[ $(uname) == "Darwin" ]]; then
THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
if [[ `uname` == "Darwin" ]]; then
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
else
THIS_SCRIPT=$(readlink -f "$0")
THIS_SCRIPT=`readlink -f $0`
HW=`uname -m`
OS_PATH=linux-${HW}
fi
export INSTANCE="1"
THIS_DIR="${THIS_SCRIPT%/*}"
BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
BASE_DIR=$(cd $THIS_DIR/../.. && pwd)
LIB_DIR="${BASE_DIR}/lib"
BIN_DIR="${BASE_DIR}/bin"
export CONF_DIR="${BASE_DIR}/conf"
NATIVE_DIR="${BIN_DIR}/META-INF/native"
export LOG_DIR="${BASE_DIR}/logs"
set -a
. "${BIN_DIR}/timely-server-env.sh"
. ${BIN_DIR}/timely-server-env.sh
set +a
# use either a value from timely-server-env.sh or the default
export LOG_DIR="${TIMELY_LOG_DIR:-${BASE_DIR}/logs}"
if [[ ! -e ${NATIVE_DIR} ]]; then
mkdir -p "${NATIVE_DIR}"
fi
if [[ ! -e ${LOG_DIR} ]]; then
mkdir "${LOG_DIR}"
fi
JVM_ARGS="-Xmx256m -Xms128m"
JVM_ARGS="${JVM_ARGS} -Dlogging.config=${CONF_DIR}/log4j2-console.yml"
JVM_ARGS="${JVM_ARGS} -Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:+UseStringDeduplication"
JVM_ARGS="${JVM_ARGS} -Djava.net.preferIPv4Stack=true"
JVM_ARGS="${JVM_ARGS} -XX:+UseNUMA"
# for linux varieties of OS, extract native libraries if not already present
if [[ -n "$OS_PATH" ]]; then
if [[ ! -d ${NATIVE_DIR} || -z "$(ls -A ${NATIVE_DIR})" ]]; then
pushd "${BASE_DIR}/bin" || exit
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
popd || exit
fi
pushd "${BASE_DIR}/bin" || exit
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_tcnative_linux_"${HW}".so
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_transport_native_epoll_"${HW}".so
popd || exit
JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
fi
+14 -29
View File
@@ -1,52 +1,37 @@
#!/bin/bash
if [[ $(uname) == "Darwin" ]]; then
THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
if [[ `uname` == "Darwin" ]]; then
THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
else
THIS_SCRIPT=$(readlink -f "$0")
THIS_SCRIPT=`readlink -f $0`
HW=`uname -m`
OS_PATH=linux-${HW}
fi
export INSTANCE="1"
THIS_DIR="${THIS_SCRIPT%/*}"
BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
BASE_DIR=$(cd $THIS_DIR/../.. && pwd)
LIB_DIR="${BASE_DIR}/lib"
BIN_DIR="${BASE_DIR}/bin"
export CONF_DIR="${BASE_DIR}/conf"
NATIVE_DIR="${BIN_DIR}/META-INF/native"
export LOG_DIR="${BASE_DIR}/logs"
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"
# 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
pushd "${BASE_DIR}/bin" || exit
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_tcnative_linux_"${HW}".so
"${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar ${NATIVE_DIR}/libnetty_transport_native_epoll_"${HW}".so
popd || exit
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
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
+3 -13
View File
@@ -4,7 +4,7 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>timely-server</artifactId>
<description>Timely Server</description>
@@ -68,10 +68,6 @@
<groupId>io.micrometer</groupId>
<artifactId>micrometer-core</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-classes-epoll</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor.netty</groupId>
<artifactId>reactor-netty</artifactId>
@@ -228,12 +224,6 @@
<artifactId>log4j-core</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcprov-ext-jdk18on</artifactId>
<version>1.73</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.javassist</groupId>
<artifactId>javassist</artifactId>
@@ -380,8 +370,8 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>17</source>
<target>17</target>
<source>11</source>
<target>11</target>
</configuration>
</plugin>
</plugins>
@@ -53,9 +53,11 @@ public class MetricQuery {
}
public void setCounterOptions(long max, long resetValue) {
this.subQuery.getRateOptions().setCounter(true);
this.subQuery.getRateOptions().setCounterMax(max);
this.subQuery.getRateOptions().setResetValue(resetValue);
QueryRequest.RateOption rateOption = new QueryRequest.RateOption();
rateOption.setCounter(true);
rateOption.setCounterMax(max);
rateOption.setResetValue(resetValue);
this.subQuery.setRateOptions(rateOption);
}
public void setTags(Map<String,String> tags) {
@@ -14,7 +14,6 @@ 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;
@@ -30,7 +29,7 @@ public class RateIterator extends TimeSeriesGroupingIterator {
private boolean isCounter = false;
private long maxCounter = 0;
private long resetValue = 0;
private long interval = 1000;
private long interval = 1;
public static void setRateOptions(IteratorSetting is, QueryRequest.RateOption options) {
if (options != null && options.isCounter()) {
@@ -38,10 +37,10 @@ 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())));
if (options != null && options.getInterval().isPresent()) {
is.addOption(RATE_INTERVAL, Long.toString(getTimeInMillis(options.getInterval().get())));
} else {
is.addOption(RATE_INTERVAL, "1000");
is.addOption(RATE_INTERVAL, "1");
}
}
@@ -77,12 +76,9 @@ 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);
}
@@ -700,11 +700,11 @@ public class DataStore {
if (query.isRate()) {
log.trace("Adding rate iterator");
IteratorSetting rate = new IteratorSetting(499, RateIterator.class);
RateOption rateOptions = query.getRateOptions();
// 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");
if (rateOptions != null && !rateOptions.getInterval().isPresent()) {
rateOptions.setInterval(Optional.of(downsample + "ms"));
}
RateIterator.setRateOptions(rate, query.getRateOptions());
scanner.addScanIterator(rate);
@@ -952,7 +952,7 @@ public class DataStore {
pattern.append(tag.getValue());
pattern.append("(,.*|$)");
IteratorSetting setting = new IteratorSetting(priority++, tag.getKey().replaceAll("\\.", "_") + " tag filter", RegExFilter.class);
IteratorSetting setting = new IteratorSetting(priority++, tag.getKey() + " tag filter", RegExFilter.class);
log.trace("Using {} additional filter on tag: {}", pattern, tag.getKey());
RegExFilter.setRegexs(setting, null, null, pattern.toString(), null, false, true);
scanner.addScanIterator(setting);
@@ -4,17 +4,21 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.client.Accumulo;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.ClientConfiguration;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
@@ -57,14 +61,18 @@ public final class MetaCache implements Iterable<Meta> {
private void refreshCache(long expirationMinutes) {
long oldestTimestamp = System.currentTimeMillis() - (expirationMinutes * 60 * 1000);
final Properties p = new Properties();
p.put("instance.name", accumuloProperties.getInstanceName());
p.put("instance.zookeeper.host", zookeeperProperties.getServers());
Map<String,String> properties = new HashMap<>();
properties.put("instance.name", accumuloProperties.getInstanceName());
properties.put("instance.zookeeper.host", zookeeperProperties.getServers());
final ClientConfiguration aConf = ClientConfiguration.fromMap(properties);
String metaTable = timelyProperties.getMetaTable();
Map<String,Map<String,Long>> metricMap = new HashMap<>();
try (AccumuloClient client = Accumulo.newClient().from(p).build(); Scanner scanner = client.createScanner(metaTable, Authorizations.EMPTY);) {
Scanner scanner = null;
try {
Map<Meta,Object> newCache = new HashMap<>();
final Instance instance = new ZooKeeperInstance(aConf);
Connector connector = instance.getConnector(accumuloProperties.getUsername(), new PasswordToken(accumuloProperties.getPassword()));
scanner = connector.createScanner(metaTable, Authorizations.EMPTY);
log.debug("Begin scanning " + metaTable);
Key metricPrefixBeginKey = new Key(Meta.METRIC_PREFIX);
int firstChar = Meta.METRIC_PREFIX.charAt(0);
@@ -110,8 +118,12 @@ public final class MetaCache implements Iterable<Meta> {
cache.putAll(newCache);
}
log.debug("Finished scanning " + metaTable);
} catch (TableNotFoundException e) {
} catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
log.error(e.getMessage(), e);
} finally {
if (scanner != null) {
scanner.close();
}
}
}
@@ -14,6 +14,7 @@ import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Executors;
@@ -28,7 +29,6 @@ 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;
@@ -766,11 +766,11 @@ public class DataStoreCache {
if (subQuery.isRate()) {
log.trace("Adding rate iterator");
IteratorSetting rate = new IteratorSetting(499, RateIterator.class);
QueryRequest.RateOption rateOptions = 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");
if (rateOptions != null && !rateOptions.getInterval().isPresent()) {
rateOptions.setInterval(Optional.of(downsamplePeriod + "ms"));
}
RateIterator.setRateOptions(rate, rateOptions);
RateIterator rateIterator = new RateIterator();
@@ -0,0 +1,187 @@
package timely.server.store.compaction;
import java.util.Iterator;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalLong;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.TabletId;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.tserver.compaction.CompactionPlan;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
import org.apache.commons.collections4.trie.PatriciaTrie;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.MapConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import timely.server.store.MetricAgeOffIterator;
public class MetricCompactionStrategy extends CompactionStrategy {
private static final Logger log = LoggerFactory.getLogger(MetricCompactionStrategy.class);
private boolean hasMinAgeOff;
private MetricAgeOffConfiguration minAgeOffConfig;
private boolean logOnly;
private String filterPrefix;
private String majcIteratorName;
private long explicitTimeMillis;
private boolean useExplicitTime;
public static final String LOG_ONLY_KEY = "logonly";
public static final String MAJC_ITERATOR_NAME_KEY = "majc.name";
public static final String DEFAULT_MAJC_ITERATOR_NAME = "ageoffmetrics";
public static final String AGE_OFF_PREFIX = MetricAgeOffIterator.AGE_OFF_PREFIX;
public static final String FILTER_PREFIX_KEY = AGE_OFF_PREFIX + "filter";
public static final String MIN_AGEOFF_KEY = AGE_OFF_PREFIX + "min";
public static final String DEFAULT_AGEOFF_KEY_SUFFIX = MetricAgeOffIterator.DEFAULT_AGEOFF_KEY;
public static final String DEFAULT_AGEOFF_KEY = AGE_OFF_PREFIX + DEFAULT_AGEOFF_KEY_SUFFIX;
@Override
public void init(Map<String,String> options) {
// options:
// logonly - disables compaction and will only log result
// majciterator - the iterator name to gather age-off values
// filter - limit the compaction to just a subset of prefixes
// min - minimum age-off which may override the metric age-off;
// allows more data to accumulate in tablets without being removed
logOnly = options.containsKey(LOG_ONLY_KEY) && Boolean.parseBoolean(options.get(LOG_ONLY_KEY));
majcIteratorName = options.getOrDefault(MAJC_ITERATOR_NAME_KEY, DEFAULT_MAJC_ITERATOR_NAME);
filterPrefix = options.get(FILTER_PREFIX_KEY);
if (options.containsKey(MIN_AGEOFF_KEY)) {
long minAgeOff = Long.parseLong(options.get(MIN_AGEOFF_KEY));
hasMinAgeOff = true;
minAgeOffConfig = MetricAgeOffConfiguration.newFromDefaultingMinimum(minAgeOff);
}
}
@Override
public boolean shouldCompact(MajorCompactionRequest request) {
TabletId tabletId = request.getTabletId();
if (null == tabletId.getEndRow() || null == tabletId.getPrevEndRow()) {
return false;
} else if (request.getFiles().isEmpty()) {
return false;
}
// check if prev/end key are both at age-off intervals
// skip tablet if there is no prev-end key, or previous age-off is not at max;
// other age-off will compact
Optional<String> endName = TabletRowAdapter.decodeRowPrefix(tabletId.getEndRow());
OptionalLong endOffset = TabletRowAdapter.decodeRowOffset(tabletId.getEndRow());
Optional<String> prevEndName = TabletRowAdapter.decodeRowPrefix(tabletId.getPrevEndRow());
if (!endName.isPresent() || !prevEndName.isPresent() || !endOffset.isPresent()) {
return false;
}
// compute age-off from majc iterator
// then maximize on the min-default-ageoff so that the
// configuration with largest age-off is used
MetricAgeOffConfiguration ageOffConfig = MetricAgeOffConfiguration.newFromRequest(request, majcIteratorName);
if (hasMinAgeOff) {
ageOffConfig = MetricAgeOffConfiguration.maximizeDefaultAgeOff(ageOffConfig, minAgeOffConfig);
}
long currentTime = useExplicitTime ? explicitTimeMillis : System.currentTimeMillis();
long ageOffComputed = currentTime - ageOffConfig.computeAgeOff(endName.get());
boolean shouldCompact = (ageOffComputed > endOffset.getAsLong() && endName.equals(prevEndName))
&& (null == filterPrefix || endName.get().startsWith(filterPrefix));
if (log.isDebugEnabled() && logOnly && shouldCompact) {
log.debug("Tablet will be metric age-off compacted {threshold: {}, endKey: {}, offset: {}, prevKey: {}}", ageOffComputed, endName,
endOffset.getAsLong(), prevEndName);
} else if (log.isTraceEnabled()) {
log.trace("Tablet check metric compaction {threshold: {}, endKey: {}, offset: {}, prevKey: {}, result: {}}", ageOffComputed, endName,
endOffset.getAsLong(), prevEndName, shouldCompact);
}
return shouldCompact && !logOnly;
}
@Override
public CompactionPlan getCompactionPlan(MajorCompactionRequest request) {
CompactionPlan plan = null;
if (shouldCompact(request)) {
plan = new CompactionPlan();
for (StoredTabletFile f : request.getFiles().keySet()) {
plan.inputFiles.add(f);
}
if (log.isDebugEnabled()) {
log.debug("Tablet selected for age-off: " + TabletRowAdapter.toDebugOutput(request.getTabletId()));
}
}
return plan;
}
public void setExplicitTime(long timeMillis) {
explicitTimeMillis = timeMillis;
useExplicitTime = true;
}
private static class MetricAgeOffConfiguration {
private final PatriciaTrie<Long> ageoffs;
private final long defaultAgeOff;
private MetricAgeOffConfiguration(PatriciaTrie<Long> ageoffs, long defaultAgeOff) {
this.ageoffs = ageoffs;
this.defaultAgeOff = defaultAgeOff;
}
public static MetricAgeOffConfiguration maximizeDefaultAgeOff(MetricAgeOffConfiguration x, MetricAgeOffConfiguration y) {
return (x.defaultAgeOff > y.defaultAgeOff) ? x : y;
}
public static MetricAgeOffConfiguration newFromDefaultingMinimum(long minAgeOff) {
return new MetricAgeOffConfiguration(new PatriciaTrie<>(), minAgeOff);
}
public static MetricAgeOffConfiguration newFromRequest(MajorCompactionRequest request, String majcIteratorName) {
Configuration config = new MapConfiguration(request.getTableProperties());
String majcIteratorKey = Property.TABLE_ITERATOR_MAJC_PREFIX.getKey() + majcIteratorName;
if (log.isTraceEnabled()) {
log.trace("Using key lookup for iterator: {}", majcIteratorKey);
}
Configuration configAgeOff = config.subset((majcIteratorKey + ".opt"));
if (null == configAgeOff.getString(DEFAULT_AGEOFF_KEY)) {
throw new IllegalArgumentException(DEFAULT_AGEOFF_KEY_SUFFIX + " must be configured for " + majcIteratorKey);
}
PatriciaTrie<Long> ageoffs = new PatriciaTrie<>();
long configureMinAgeOff = Long.MAX_VALUE;
long configureMaxAgeOff = Long.MIN_VALUE;
@SuppressWarnings("unchecked")
Iterator<String> keys = configAgeOff.getKeys();
while (keys.hasNext()) {
String k = keys.next();
String v = configAgeOff.getString(k);
if (k.startsWith((AGE_OFF_PREFIX))) {
String name = k.substring(AGE_OFF_PREFIX.length());
if (log.isTraceEnabled()) {
log.trace("Adding {} to Trie with value {}", name, Long.parseLong(v));
}
long ageoff = Long.parseLong(v);
configureMinAgeOff = Math.min(configureMinAgeOff, ageoff);
configureMaxAgeOff = Math.max(configureMaxAgeOff, ageoff);
ageoffs.put(name, ageoff);
}
}
long defaultAgeOff = ageoffs.get(DEFAULT_AGEOFF_KEY_SUFFIX);
return new MetricAgeOffConfiguration(ageoffs, defaultAgeOff);
}
public long computeAgeOff(String metricName) {
return ageoffs.getOrDefault(metricName, defaultAgeOff);
}
}
}
@@ -0,0 +1,156 @@
package timely.server.store.compaction;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
import org.apache.accumulo.tserver.compaction.CompactionPlan;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TieredCompactionStrategy extends CompactionStrategy {
public static final String TIERED_PREFIX = "tier.";
public static final String TIERED_CLASS = "class";
public static final String TIERED_OPTS = "opts";
protected final Collection<TieredCompactorSupplier> compactors;
public TieredCompactionStrategy() {
compactors = new ArrayList<>();
}
public Collection<TieredCompactorSupplier> getCompactors() {
return Collections.unmodifiableCollection(compactors);
}
@Override
public void init(Map<String,String> config) {
// example:
// <table-compaction-prefix>.tiered.0.class=<majcompaction-class>
// <table-compaction-prefix>.tiered.0.opts.key=value
// <table-compaction-prefix>.tiered.1.class=<majcompaction-class>
// <table-compaction-prefix>.tiered.1.opts.key_a=value
// <table-compaction-prefix>.tiered.1.opts.key_b=value
//
// the key=value pairs will be sent to the specific tiered compaction
// strategy as arguments
compactors.clear();
createSuppliers(config);
}
@Override
public void gatherInformation(MajorCompactionRequest request) throws IOException {
for (TieredCompactorSupplier compactor : compactors) {
CompactionStrategy strategy = compactor.get(request.getTableProperties());
strategy.gatherInformation(request);
}
}
@Override
public boolean shouldCompact(MajorCompactionRequest request) throws IOException {
boolean shouldCompact = false;
for (TieredCompactorSupplier supplier : compactors) {
CompactionStrategy strategy = supplier.get(request.getTableProperties());
shouldCompact = strategy.shouldCompact(request);
if (shouldCompact) {
break;
}
}
return shouldCompact;
}
@Override
public CompactionPlan getCompactionPlan(MajorCompactionRequest request) throws IOException {
CompactionPlan plan = null;
Iterator<TieredCompactorSupplier> iter = compactors.iterator();
while ((null == plan || plan.inputFiles.isEmpty()) && iter.hasNext()) {
CompactionStrategy strategy = iter.next().get(request.getTableProperties());
plan = strategy.getCompactionPlan(request);
}
return plan;
}
protected void createSuppliers(Map<String,String> config) {
int tier = 0;
boolean continueLoop = true;
while (continueLoop) {
String clazzKey = TIERED_PREFIX + tier + "." + TIERED_CLASS;
String clazzName = config.get(clazzKey);
if (null != clazzName) {
compactors.add(new TieredConfiguredSupplier(clazzName, tier, config));
tier++;
} else {
continueLoop = false;
}
}
}
static class TieredConfiguredSupplier implements TieredCompactorSupplier {
private static final Logger log = LoggerFactory.getLogger(TieredConfiguredSupplier.class);
private final String clazzName;
private final int tier;
private final Map<String,String> initConfig;
private CompactionStrategy strategy;
private Map<String,String> options;
TieredConfiguredSupplier(String clazzName, int tier, Map<String,String> initConfig) {
this.strategy = null;
this.clazzName = clazzName;
this.tier = tier;
this.initConfig = initConfig;
}
public Map<String,String> options() {
if (null == options) {
options = new HashMap<>();
String tierCompare = Integer.toString(tier);
int tierIdx = TIERED_PREFIX.length();
int optsIdx = TIERED_PREFIX.length() + tierCompare.length() + 1;
for (Map.Entry<String,String> entry : initConfig.entrySet()) {
String key = entry.getKey();
if (key.regionMatches(tierIdx, tierCompare, 0, tierCompare.length()) && key.regionMatches(optsIdx, TIERED_OPTS, 0, TIERED_OPTS.length())) {
options.put(key.substring(optsIdx + TIERED_OPTS.length() + 1), entry.getValue());
}
}
}
return Collections.unmodifiableMap(options);
}
public CompactionStrategy get(Map<String,String> tableProperties) {
if (null == strategy) {
try {
String tableContext = tableProperties.get(Property.TABLE_CLASSPATH.getKey());
Class<? extends CompactionStrategy> clazz;
if (null != tableContext && !tableContext.equals("")) {
clazz = AccumuloVFSClassLoader.getContextClassLoader(tableContext).loadClass(clazzName).asSubclass(CompactionStrategy.class);
} else {
clazz = AccumuloVFSClassLoader.getClassLoader().loadClass(clazzName).asSubclass(CompactionStrategy.class);
}
strategy = clazz.newInstance();
strategy.init(options());
} catch (Exception e) {
log.debug("Failed to load class for tiered compaction: " + e.toString());
throw new RuntimeException(e);
}
}
return strategy;
}
}
}
@@ -0,0 +1,12 @@
package timely.server.store.compaction;
import java.util.Map;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
public interface TieredCompactorSupplier {
Map<String,String> options();
CompactionStrategy get(Map<String,String> tableProperties);
}
@@ -10,7 +10,7 @@ import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.metadata.SystemTables;
import org.apache.accumulo.core.metadata.MetadataTable;
import timely.common.configuration.TimelyProperties;
import timely.server.configuration.TabletMetadataProperties;
@@ -32,11 +32,10 @@ public class TabletMetadataQuery {
Map<String,String> tableMap = accumuloClient.tableOperations().tableIdMap();
String tableId = tableMap.get(timelyProperties.getMetricsTable());
if (null == tableId) {
throw new IllegalStateException("Unable to find " + SystemTables.METADATA.tableName());
throw new IllegalStateException("Unable to find " + MetadataTable.NAME);
}
try (Scanner s = accumuloClient.createScanner(SystemTables.METADATA.tableName(),
accumuloClient.securityOperations().getUserAuthorizations(accumuloClient.whoami()))) {
try (Scanner s = accumuloClient.createScanner(MetadataTable.NAME, accumuloClient.securityOperations().getUserAuthorizations(accumuloClient.whoami()))) {
s.setRange(Range.prefix(tableId));
for (Map.Entry<Key,Value> e : s) {
tabletMetadataView.addEntry(e);
@@ -189,8 +189,10 @@ 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.getRateOptions().setCounter(false);
subQuery.setRateOptions(rateOption);
query.setQueries(Collections.singleton(subQuery));
SortedKeyValueIterator<org.apache.accumulo.core.data.Key,org.apache.accumulo.core.data.Value> itr;
@@ -702,20 +702,25 @@ 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 {
addRateData("sys.cpu.user", "tag1=value1 tag2=value2 rack=r1", TEST_TIME, 1, TEST_TIME + 14, 1.0, Double.MAX_VALUE, 1.0);
// @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
dataStore.flush();
dataStoreCache.flushCaches(-1);
// Latency in TestConfiguration is 2s, wait for it
@@ -746,7 +751,23 @@ public class HttpApiIT extends OneWaySSLBase {
@Test
public void testRateCounterQuery() throws Exception {
addRateData("sys.cpu.user", "tag1=value1 tag2=value2 rack=r1", TEST_TIME, 1, TEST_TIME + 14, 1.0, 10.0, 1.0);
// @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
dataStore.flush();
dataStoreCache.flushCaches(-1);
// Latency in TestConfiguration is 2s, wait for it
@@ -761,6 +782,7 @@ 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);
@@ -777,8 +799,24 @@ public class HttpApiIT extends OneWaySSLBase {
}
@Test
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);
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
dataStore.flush();
dataStoreCache.flushCaches(-1);
// Latency in TestConfiguration is 2s, wait for it
@@ -793,8 +831,9 @@ 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(10);
subQuery.getRateOptions().setCounterMax(Long.MAX_VALUE);
subQuery.getRateOptions().setResetValue(2);
request.addQuery(subQuery);
List<QueryResponse> response = query(baseUrl + "/api/query", request);
@@ -805,54 +844,7 @@ public class HttpApiIT extends OneWaySSLBase {
int i = 1;
for (Entry<String,Object> e : dps.entrySet()) {
assertEquals(Long.toString(TEST_TIME + i), e.getKey());
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());
assertEquals((i == 10 ? 0.0 : 1.0), e.getValue());
i++;
}
}
@@ -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,1s,100,0}:sys.cpu.user{host=*}{rack=r1|r2}&tsuid=sum:000001000002000042,000001000002000043");
"/api/query?start=1356998400&end=1356998460&m=sum:rate{false,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,7 +515,6 @@ 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());
@@ -546,7 +545,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,1s,100,0}:sys.cpu.user{host=*}{rack=r1|r2}&tsuid=sum:000001000002000042,000001000002000043");
"/api/query?start=1356998400&end=1356998460&m=sum:rate{false,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());
@@ -563,7 +562,6 @@ 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());
@@ -185,7 +185,7 @@ public class ITBase {
TestDataStore.StoreCallback storeCallback = () -> PUT_REQUESTS.countDown();
try {
dataStore.addStoreCallback(storeCallback);
StringBuilder format = new StringBuilder();
StringBuffer format = new StringBuffer();
for (String line : lines) {
format.append("put ");
format.append(line);
@@ -0,0 +1,250 @@
package timely.server.integration;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.util.Collection;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.iterators.IteratorUtil;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
import org.apache.accumulo.tserver.compaction.DefaultCompactionStrategy;
import org.apache.hadoop.io.Text;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.junit4.SpringRunner;
import com.google.common.collect.Sets;
import io.jsonwebtoken.lang.Assert;
import timely.server.store.MetricAgeOffIterator;
import timely.server.store.compaction.MetricCompactionStrategy;
import timely.server.store.compaction.TabletRowAdapter;
import timely.server.store.compaction.util.TabletMetadataQuery;
import timely.server.store.compaction.util.TabletMetadataView;
import timely.server.store.compaction.util.TabletStatistic;
import timely.server.store.compaction.util.TabletSummary;
import timely.server.test.TestMetricWriter;
import timely.server.test.TestMetrics;
import timely.test.IntegrationTest;
import timely.test.TimelyTestRule;
@Category(IntegrationTest.class)
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
@ActiveProfiles({"miniAccumulo"})
public class MetricCompactionIT extends ITBase {
private static final String AGE_OFF_ITERATOR_NAME = "ageoffmetrics";
private static final EnumSet<IteratorUtil.IteratorScope> AGEOFF_SCOPES = EnumSet.allOf(IteratorUtil.IteratorScope.class);
@Autowired
@Rule
public TimelyTestRule testRule;
private MetricConfigurationAdapter adapter;
@Before
public void setup() {
super.setup();
try {
accumuloClient.instanceOperations().setProperty(Property.TSERV_MAJC_DELAY.getKey(), "3s");
adapter = new MetricConfigurationAdapter(accumuloClient, timelyProperties.getMetricsTable());
accumuloClient.tableOperations().setProperty(timelyProperties.getMetricsTable(), Property.TABLE_SPLIT_THRESHOLD.getKey(), "50K");
adapter.resetState();
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
@After
public void cleanup() {
super.cleanup();
deleteTable(timelyProperties.getMetricsTable());
deleteTable(timelyProperties.getMetaTable());
}
@After
public void teardown() throws Exception {
accumuloClient.instanceOperations().setProperty(Property.TSERV_MAJC_DELAY.getKey(), "30s");
adapter.resetState();
}
@Test
public void tableMetadataEnumeratesTablets() throws Exception {
String metricsTable = timelyProperties.getMetricsTable();
TestMetricWriter writer = new TestMetricWriter(metricsTable, accumuloClient, new TestMetrics());
writer.ingestRandomDuration(60, TimeUnit.SECONDS, System.currentTimeMillis(), 250);
Collection<Text> splits = accumuloClient.tableOperations().listSplits(timelyProperties.getMetricsTable());
TabletMetadataQuery query = new TabletMetadataQuery(accumuloClient, timelyProperties, tabletMetadataProperties);
TabletMetadataView view = query.run();
// @formatter:off
TabletSummary summary = view.computeSummary()
.disableTabletRowCheckFilter()
.build();
Set<String> splitsPrefixed = splits.stream()
.map(row -> TabletRowAdapter.decodeRowPrefix(row).orElse("unknown"))
.collect(Collectors.toSet());
Set<String> tabletsPrefixed = summary.getSummary().stream()
.map(TabletStatistic::getKeyName)
.collect(Collectors.toSet());
// difference both sides
// should be empty
// @formatter:on
Set<String> i1 = Sets.difference(tabletsPrefixed, splitsPrefixed);
Set<String> i2 = Sets.difference(splitsPrefixed, tabletsPrefixed);
assertTrue("tablets-diff" + i1, i1.isEmpty());
assertTrue("splits-diff" + i2, i2.isEmpty());
// dump data to screen as summary
Assert.notNull(view.toText());
}
@Test
public void tabletsCompactedUntilAgeOff() throws Exception {
// test will replicate problem in time-series tablets where data is
// split automatically and the resulting tablets will not be cleaned up
// by the default compaction strategy
//
// metric compaction strategy will age-off tablets by date on tablets
// but ignore differing end/prev rows
//
// the mini-accumulo cluster should not have any age-off
// iterators on timely.metrics
String metricsTable = timelyProperties.getMetricsTable();
// check to make sure the table has two iterators
Map<String,EnumSet<IteratorUtil.IteratorScope>> itrs = accumuloClient.tableOperations().listIterators(metricsTable);
assertEquals(1, itrs.size());
assertTrue(itrs.containsKey("vers"));
long timestampMax = System.currentTimeMillis();
TestMetricWriter writer = new TestMetricWriter(timelyProperties.getMetricsTable(), accumuloClient, new TestMetrics());
long timestampMin = writer.ingestRandomDuration(30, TimeUnit.SECONDS, timestampMax, 250);
// invoke compact operation to flush everything through
// without this call the compaction strategy later on
// was not compacting all of the tablets
adapter.runCompaction();
// figure out what should be targeted based on an ageoff date
// cut the delta by 1/3
long ageoff = (timestampMax - timestampMin) / 3;
// run check to verify there are pre-existing splits for a prefix
// @formatter:off
TabletMetadataQuery query = new TabletMetadataQuery(accumuloClient, timelyProperties, tabletMetadataProperties);
TabletSummary summary1 = query.run()
.computeSummary()
.build();
Set<String> tabletsBefore = summary1.getSummary().stream()
.filter(t -> t.maxAge() > ageoff)
.map(TabletStatistic::getKeyName)
.collect(Collectors.toSet());
// @formatter:on
assertFalse("Tablets were empty before compaction, expected test data", tabletsBefore.isEmpty());
long compactBeforeMillis = System.currentTimeMillis();
// invoke the age-off compactor
// after this runs it should have gotten rid o fdata
// have pre-age-off tablets
adapter.applyCompactionConfiguration(MetricCompactionStrategy.class, ageoff);
adapter.runCompaction();
long compactDeltaMillis = System.currentTimeMillis() - compactBeforeMillis;
// re-run check on tablet metadata
// check that there aren't new tablets past the predicted age-off
// any tablets with differing end/prev rows are filtered by the summary
// @formatter:off
TabletSummary summary2 = query.run()
.computeSummary()
.build();
List<TabletStatistic> tabletsFilterAfter = summary2.getSummary().stream()
.filter(t -> ((t.maxAge() - compactDeltaMillis) > ageoff) && t.getKeyCount() > 1)
.collect(Collectors.toList());
// @formatter:on
if (!tabletsFilterAfter.isEmpty()) {
log.debug("tablets-before: {}", tabletsBefore.toString());
log.debug("tablets-after: {}", tabletsFilterAfter.toString());
}
// should not have found tablets past the age-off and should also
// have found existing tablets still in system (i.e. did not compact everything)
assertTrue("Tablets still had unexpected after compaction", tabletsFilterAfter.isEmpty());
assertTrue("Some tablets should exist, but did not find any", summary2.totalTablets() > 0);
}
private static class MetricConfigurationAdapter {
private final AccumuloClient accumuloClient;
private final String tableName;
private static String[] CLEAR_PROPERTY_KEYS = {Property.TABLE_COMPACTION_STRATEGY.getKey() + MetricCompactionStrategy.MIN_AGEOFF_KEY};
public MetricConfigurationAdapter(AccumuloClient accumuloClient, String tableName) {
this.accumuloClient = accumuloClient;
this.tableName = tableName;
}
@SuppressWarnings("removal")
public void applyCompactionConfiguration(Class<? extends CompactionStrategy> clazz, long ageOff) throws Exception {
Map<String,String> ageOffs = new HashMap<>();
ageOffs.put("ageoff.default", Long.toString(ageOff));
IteratorSetting ageOffIteratorSettings = new IteratorSetting(100, AGE_OFF_ITERATOR_NAME, MetricAgeOffIterator.class, ageOffs);
accumuloClient.tableOperations().removeIterator(tableName, AGE_OFF_ITERATOR_NAME, AGEOFF_SCOPES);
accumuloClient.tableOperations().attachIterator(tableName, ageOffIteratorSettings, AGEOFF_SCOPES);
accumuloClient.tableOperations().setProperty(tableName, Property.TABLE_COMPACTION_STRATEGY.getKey(), clazz.getName());
}
public void runCompaction() throws Exception {
accumuloClient.tableOperations().compact(tableName, null, null, true, true);
}
public void resetState() throws Exception {
Map<String,EnumSet<IteratorUtil.IteratorScope>> iters = accumuloClient.tableOperations().listIterators(tableName);
for (String name : iters.keySet()) {
if (name.startsWith("ageoff")) {
accumuloClient.tableOperations().removeIterator(tableName, name, AGEOFF_SCOPES);
}
}
// reset compaction strategy to default
accumuloClient.tableOperations().setProperty(tableName, Property.TABLE_COMPACTION_STRATEGY.getKey(), DefaultCompactionStrategy.class.getName());
// reset any properties
for (String s : CLEAR_PROPERTY_KEYS) {
accumuloClient.tableOperations().removeProperty(tableName, s);
}
}
}
}
@@ -31,9 +31,6 @@ public class DownsampleIteratorTest {
final private SortedMap<Key,Value> testData1 = new TreeMap<>();
final private SortedMap<Key,Value> testData2 = new TreeMap<>();
private long now = System.currentTimeMillis();
private long startingTimestamp = now - (now % 100);
@Before
public void createTestData() {
createTestData1();
@@ -44,7 +41,7 @@ public class DownsampleIteratorTest {
List<Tag> tags = Collections.singletonList(new Tag("host", "host1"));
for (long i = 0; i < 1000; i += 100) {
Metric m = new Metric("sys.loadAvg", startingTimestamp + i, .2, tags);
Metric m = new Metric("sys.loadAvg", i, .2, tags);
Mutation mutation = MetricAdapter.toMutation(m);
for (ColumnUpdate cu : mutation.getUpdates()) {
Key key = new Key(mutation.getRow(), cu.getColumnFamily(), cu.getColumnQualifier(), cu.getColumnVisibility(), cu.getTimestamp());
@@ -66,8 +63,8 @@ public class DownsampleIteratorTest {
List<Tag> tags2 = Collections.singletonList(new Tag("host", "host2"));
for (long i = 0; i < 1000; i += 100) {
put(testData2, new Metric("sys.loadAvg", startingTimestamp + i, .2, tags));
put(testData2, new Metric("sys.loadAvg", startingTimestamp + i + 50, .5, tags2));
put(testData2, new Metric("sys.loadAvg", i, .2, tags));
put(testData2, new Metric("sys.loadAvg", i + 50, .5, tags2));
}
}
@@ -81,14 +78,13 @@ public class DownsampleIteratorTest {
Set<Tag> tags = entry.getKey();
assertEquals(1, tags.size());
assertEquals(Collections.singleton(new Tag("host", "host1")), tags);
long ts = startingTimestamp - (startingTimestamp % 100);
long ts = 0;
for (Sample sample : entry.getValue()) {
assertEquals(ts, sample.getTimestamp());
ts += 100;
assertEquals(0.2, sample.getValue(), 0.0001);
}
assertEquals(startingTimestamp - (startingTimestamp % 100) + 1000, ts);
assertEquals(1000, ts);
}
}
@@ -100,7 +96,7 @@ public class DownsampleIteratorTest {
for (Tag tag : new Tag[] {new Tag("host", "host1"), new Tag("host", "host2")}) {
Downsample dsample = samples.get(Collections.singleton(tag));
assertNotNull(dsample);
long ts = startingTimestamp - (startingTimestamp % 100);
long ts = 0;
double value = .2;
if (tag.getValue().equals("host2")) {
value = .5;
@@ -124,7 +120,7 @@ public class DownsampleIteratorTest {
for (Tag tag : new Tag[] {new Tag("host", "host1"), new Tag("host", "host2")}) {
Downsample dsample = samples.get(Collections.singleton(tag));
assertNotNull(dsample);
long ts = startingTimestamp - (startingTimestamp % 200);
long ts = 0;
double value = .2;
if (tag.getValue().equals("host2")) {
value = .5;
@@ -220,7 +216,7 @@ public class DownsampleIteratorTest {
for (long i = 0; i < elapsedTime; i += skipInterval) {
for (List<Tag> tags : listOfTagVariations) {
put(testData3, new Metric("sys.loadAvg", startingTimestamp + i, i * 2, tags));
put(testData3, new Metric("sys.loadAvg", i, i * 2, tags));
}
}
return testData3;
@@ -1,17 +1,22 @@
package timely.server.sample.iterators;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import org.apache.accumulo.core.client.PluginEnvironment;
import org.apache.accumulo.core.client.sample.SamplerConfiguration;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.DefaultConfiguration;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.IteratorUtil.IteratorScope;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iteratorsImpl.system.MapFileIterator;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
public class IteratorTestBase {
@@ -27,6 +32,18 @@ public class IteratorTestBase {
this.conf = DefaultConfiguration.getInstance();
}
@Override
public SortedKeyValueIterator<Key,Value> reserveMapFileReader(String mapFileName) throws IOException {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
return new MapFileIterator(fs, mapFileName, conf);
}
@Override
public AccumuloConfiguration getConfig() {
return conf;
}
@Override
public IteratorScope getIteratorScope() {
throw new UnsupportedOperationException();
@@ -37,6 +54,11 @@ public class IteratorTestBase {
throw new UnsupportedOperationException();
}
@Override
public void registerSideChannel(SortedKeyValueIterator<Key,Value> iter) {
throw new UnsupportedOperationException();
}
@Override
public Authorizations getAuthorizations() {
throw new UnsupportedOperationException();
@@ -56,26 +78,6 @@ public class IteratorTestBase {
public boolean isSamplingEnabled() {
return false;
}
@Override
public boolean isUserCompaction() {
return false;
}
@Override
public PluginEnvironment getPluginEnv() {
throw new UnsupportedOperationException();
}
@Override
public TableId getTableId() {
throw new UnsupportedOperationException();
}
@Override
public boolean isRunningLowOnMemory() {
return false;
}
}
protected static class CombinerIteratorEnvironment extends DefaultIteratorEnvironment {
@@ -7,6 +7,7 @@ import static org.junit.Assert.assertTrue;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.TreeMap;
@@ -46,7 +47,7 @@ public class RateIteratorTest extends IteratorTestBase {
@Test
public void testConstantTimeRate() throws Exception {
testInterval(null, 1.0D);
testInterval(null, 0.001D);
testInterval("1ms", 0.001D);
testInterval("1s", 1.0D);
testInterval("60s", 60.0D);
@@ -58,7 +59,7 @@ public class RateIteratorTest extends IteratorTestBase {
IteratorSetting settings = new IteratorSetting(100, RateIterator.class);
QueryRequest.RateOption rateOption = new QueryRequest.RateOption();
if (interval != null) {
rateOption.setInterval(interval);
rateOption.setInterval(Optional.of(interval));
}
RateIterator.setRateOptions(settings, rateOption);
iter.init(source, settings.getOptions(), SCAN_IE);
@@ -95,7 +96,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) * 1000);
expected.add((thisValue + (prevValue * -1)) / (k.getTimestamp() - prevTs));
}
prevTs = k.getTimestamp();
prevValue = MetricAdapter.decodeValue(v.get());
@@ -144,7 +145,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(1.0D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
assertEquals(0.001D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
iter.next();
}
assertFalse(iter.hasTop());
@@ -178,7 +179,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(1.0D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
assertEquals(0.001D, MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
iter.next();
}
assertFalse(iter.hasTop());
@@ -213,7 +214,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 : 1.0D), MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
assertEquals(((i + 1) % 10 == 0 ? 0.0D : 0.001D), MetricAdapter.decodeValue(iter.getTopValue().get()), 0.0D);
iter.next();
}
assertFalse(iter.hasTop());
@@ -0,0 +1,288 @@
package timely.server.store.compaction;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.tserver.compaction.CompactionPlan;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
import org.junit.Test;
import timely.server.test.CompactionRequestBuilder;
public class MetricCompactionStrategyTest {
private static final long TIME_MILLIS = new GregorianCalendar(2019, Calendar.JANUARY, 1).getTimeInMillis();
@Test(expected = IllegalArgumentException.class)
public void initThrowsWhenNoDefaultAgeOff() {
Map<String,String> config = new HashMap<>();
MetricCompactionStrategy strategy = new MetricCompactionStrategy();
strategy.init(config);
CompactionRequestBuilder builder = new CompactionRequestBuilder().endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12)).file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
MajorCompactionRequest request = builder.build();
strategy.shouldCompact(request);
}
@Test
public void minAgeOffOverridesMajcDefaultWillUseMaximum() throws IOException {
Map<String,String> config = new HashMap<>();
config.put(MetricCompactionStrategy.MIN_AGEOFF_KEY, Long.toString(TimeUnit.DAYS.toMillis(10)));
CompactionStrategy strategy = newStrategy(config);
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(13)
.endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void minAgeOffOverridesMajcDefaultNotCompact() throws IOException {
Map<String,String> config = new HashMap<>();
config.put(MetricCompactionStrategy.MIN_AGEOFF_KEY, Long.toString(TimeUnit.DAYS.toMillis(13)));
CompactionStrategy strategy = newStrategy(config);
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void compactFilesIfKeyIsPastAgeOff() throws IOException {
String f1 = "hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf";
String f2 = "hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf";
String f3 = "hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao6.rf";
CompactionStrategy strategy = newStrategy();
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys", TimeUnit.DAYS.toMillis(12))
.file(f1, 12, 1)
.file(f2, 11, 1)
.file(f3, 16, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertTrue(shouldCompact);
assertNotNull(plan);
assertEquals(3, plan.inputFiles.size());
assertEquals(f1, plan.inputFiles.get(0).getPathStr());
assertEquals(f2, plan.inputFiles.get(1).getPathStr());
assertEquals(f3, plan.inputFiles.get(2).getPathStr());
}
@Test
public void noCompactionIfPrevKeyIsDifferent() throws IOException {
CompactionStrategy strategy = newStrategy();
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys.mem", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void noCompactionIfPrevKeyIsNull() throws IOException {
CompactionStrategy strategy = newStrategy();
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys", TimeUnit.DAYS.toMillis(11))
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void noCompactionWhenExceedsButDoesNotHaveFiles() throws IOException {
CompactionStrategy strategy = newStrategy();
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys", TimeUnit.DAYS.toMillis(12));
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void noCompactionIfKeyIsNotOffset() throws IOException {
CompactionStrategy strategy = newStrategy();
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetricNonOffset("sys.mem", null)
.prevEndKeyMetricNonOffset("sys.cpu", null)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void noCompactionIfKeyIsWrongOffset() throws IOException {
CompactionStrategy strategy = newStrategy();
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetricNonOffset("sys.cpu", "test2")
.prevEndKeyMetricNonOffset("sys.cpu", "test1")
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void compactWithLogOnlyDoesNotCompact() throws IOException {
Map<String,String> defaults = new HashMap<>();
defaults.put(MetricCompactionStrategy.LOG_ONLY_KEY, "true");
CompactionStrategy strategy = newStrategy(defaults);
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
@Test
public void compactWithFilterPrefixWillCompact() throws IOException {
Map<String,String> defaults = new HashMap<>();
defaults.put(MetricCompactionStrategy.FILTER_PREFIX_KEY, "sys.cp");
CompactionStrategy strategy = newStrategy(defaults);
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertTrue(shouldCompact);
assertNotNull(plan);
assertEquals(2, plan.inputFiles.size());
}
@Test
public void compactWithoutFilterPrefixWillNotCompact() throws IOException {
Map<String,String> defaults = new HashMap<>();
defaults.put(MetricCompactionStrategy.FILTER_PREFIX_KEY, "sys.mem");
CompactionStrategy strategy = newStrategy(defaults);
// @formatter:off
CompactionRequestBuilder builder = newRequestBuilder(10)
.endKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(11))
.prevEndKeyMetric("sys.cpu", TimeUnit.DAYS.toMillis(12))
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf", 1, 1)
.file("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao7.rf", 1, 1);
// @formatter:on
MajorCompactionRequest request = builder.build();
boolean shouldCompact = strategy.shouldCompact(request);
CompactionPlan plan = strategy.getCompactionPlan(request);
assertFalse(shouldCompact);
assertNull(plan);
}
private static CompactionRequestBuilder newRequestBuilder(int defaultAgeOffDays) {
return new CompactionRequestBuilder(TIME_MILLIS).tableProperties(Property.TABLE_ITERATOR_MAJC_PREFIX.getKey() + "ageoffmetrics.opt.ageoff.default",
Long.toString(TimeUnit.DAYS.toMillis(defaultAgeOffDays)));
}
private static MetricCompactionStrategy newStrategy() {
return newStrategy(new TreeMap<>());
}
private static MetricCompactionStrategy newStrategy(Map<String,String> config) {
if (!config.containsKey(MetricCompactionStrategy.LOG_ONLY_KEY)) {
config.put(MetricCompactionStrategy.LOG_ONLY_KEY, "false");
}
MetricCompactionStrategy strategy = new MetricCompactionStrategy();
strategy.setExplicitTime(TIME_MILLIS);
strategy.init(config);
return strategy;
}
}
@@ -0,0 +1,238 @@
package timely.server.store.compaction;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.tserver.compaction.CompactionPlan;
import org.apache.accumulo.tserver.compaction.CompactionStrategy;
import org.apache.accumulo.tserver.compaction.DefaultCompactionStrategy;
import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
import org.easymock.EasyMock;
import org.junit.Test;
import com.google.common.collect.Lists;
import timely.server.test.CompactionRequestBuilder;
public class TieredCompactionStrategyTest {
@Test
public void initConfiguresWithMultipleTiers() {
Map<String,String> config = new HashMap<>();
config.put("tier.0.class", DefaultCompactionStrategy.class.getName());
config.put("tier.0.opts.arg1", "foo");
config.put("tier.0.opts.arg2", "sys");
config.put("tier.1.class", DefaultCompactionStrategy.class.getName());
config.put("tier.1.opts.arg1", "bar");
config.put("tier.2.class", DefaultCompactionStrategy.class.getName());
TieredCompactionStrategy strategy = new TieredCompactionStrategy();
strategy.init(config);
List<TieredCompactorSupplier> suppliers = Lists.newArrayList(strategy.getCompactors());
assertEquals(3, suppliers.size());
assertEquals(2, suppliers.get(0).options().size());
assertEquals(1, suppliers.get(1).options().size());
assertEquals(0, suppliers.get(2).options().size());
assertEquals("foo", suppliers.get(0).options().get("arg1"));
assertEquals("sys", suppliers.get(0).options().get("arg2"));
assertEquals("bar", suppliers.get(1).options().get("arg1"));
}
@Test
public void initConfiguresWithSingleTierNoArguments() {
Map<String,String> config = new HashMap<>();
config.put("tier.0.class", DefaultCompactionStrategy.class.getName());
TieredCompactionStrategy strategy = new TieredCompactionStrategy();
strategy.init(config);
List<TieredCompactorSupplier> suppliers = Lists.newArrayList(strategy.getCompactors());
assertEquals(1, suppliers.size());
assertEquals(0, suppliers.get(0).options().size());
}
@Test
public void gatherInformationAcrossTiers() throws IOException {
CompactionStrategy c1 = EasyMock.createMock(CompactionStrategy.class);
CompactionStrategy c2 = EasyMock.createMock(CompactionStrategy.class);
c1.gatherInformation(EasyMock.anyObject());
EasyMock.expectLastCall();
c2.gatherInformation(EasyMock.anyObject());
EasyMock.expectLastCall();
EasyMock.replay(c1, c2);
TieredCompactionStrategy strategy = newStrategy(c1, c2);
CompactionRequestBuilder builder = new CompactionRequestBuilder();
MajorCompactionRequest request = builder.build();
strategy.gatherInformation(request);
EasyMock.verify(c1, c2);
}
@Test
public void shouldCompactStopsAtTier() throws IOException {
CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
EasyMock.expect(c1.shouldCompact(EasyMock.anyObject())).andReturn(false);
EasyMock.expect(c2.shouldCompact(EasyMock.anyObject())).andReturn(true);
EasyMock.replay(c1, c2, c3);
TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
CompactionRequestBuilder builder = new CompactionRequestBuilder();
MajorCompactionRequest request = builder.build();
boolean result = strategy.shouldCompact(request);
assertTrue(result);
EasyMock.verify(c1, c2, c3);
}
@Test
public void shouldCompactWithSingleTier() throws Exception {
CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
EasyMock.expect(c1.shouldCompact(EasyMock.anyObject())).andReturn(false);
EasyMock.replay(c1);
TieredCompactionStrategy strategy = newStrategy(c1);
CompactionRequestBuilder builder = new CompactionRequestBuilder();
MajorCompactionRequest request = builder.build();
boolean result = strategy.shouldCompact(request);
assertFalse(result);
EasyMock.verify(c1);
}
@Test
public void getCompactionPlanSkipsAndStopsAtTier() throws IOException {
CompactionPlan p1 = new CompactionPlan();
p1.inputFiles.add(new StoredTabletFile("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf"));
CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
EasyMock.expect(c1.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
EasyMock.expect(c2.getCompactionPlan(EasyMock.anyObject())).andReturn(p1);
EasyMock.replay(c1, c2, c3);
TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
CompactionRequestBuilder builder = new CompactionRequestBuilder();
MajorCompactionRequest request = builder.build();
CompactionPlan actualPlan = strategy.getCompactionPlan(request);
assertNotNull(actualPlan);
assertSame(p1, actualPlan);
EasyMock.verify(c1, c2, c3);
}
@Test
public void getCompactionPlanSkipsWhenEmptyInputAndStopsAtTier() throws IOException {
CompactionPlan p1 = new CompactionPlan();
p1.inputFiles.add(new StoredTabletFile("hdfs://hdfs-name/accumulo/tables/2/default_tablet/C0000ao8.rf"));
CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
EasyMock.expect(c1.getCompactionPlan(EasyMock.anyObject())).andReturn(new CompactionPlan());
EasyMock.expect(c2.getCompactionPlan(EasyMock.anyObject())).andReturn(p1);
EasyMock.replay(c1, c2, c3);
TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
CompactionRequestBuilder builder = new CompactionRequestBuilder();
MajorCompactionRequest request = builder.build();
CompactionPlan actualPlan = strategy.getCompactionPlan(request);
assertNotNull(actualPlan);
assertSame(p1, actualPlan);
EasyMock.verify(c1, c2, c3);
}
@Test
public void getCompactionPlanWillReturnNullIfNonCompact() throws IOException {
CompactionStrategy c1 = EasyMock.createMock("c1", CompactionStrategy.class);
CompactionStrategy c2 = EasyMock.createMock("c2", CompactionStrategy.class);
CompactionStrategy c3 = EasyMock.createMock("c3", CompactionStrategy.class);
EasyMock.expect(c1.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
EasyMock.expect(c2.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
EasyMock.expect(c3.getCompactionPlan(EasyMock.anyObject())).andReturn(null);
EasyMock.replay(c1, c2, c3);
TieredCompactionStrategy strategy = newStrategy(c1, c2, c3);
CompactionRequestBuilder builder = new CompactionRequestBuilder();
MajorCompactionRequest request = builder.build();
CompactionPlan actualPlan = strategy.getCompactionPlan(request);
assertNull(actualPlan);
EasyMock.verify(c1, c2, c3);
}
private static TieredCompactionStrategy newStrategy(CompactionStrategy... strategies) {
Collection<TieredCompactorSupplier> mockSuppliers = new ArrayList<>();
for (CompactionStrategy s : strategies) {
mockSuppliers.add(new MockSupplier(s));
}
// @formatter:off
TieredCompactionStrategy strategy = EasyMock.partialMockBuilder(TieredCompactionStrategy.class)
.withConstructor()
.addMockedMethod("createSuppliers")
.createMock();
// @formatter:on
strategy.createSuppliers(EasyMock.anyObject());
EasyMock.expectLastCall().andAnswer(() -> strategy.compactors.addAll(mockSuppliers));
EasyMock.replay(strategy);
strategy.init(Collections.emptyMap());
return strategy;
}
static class MockSupplier implements TieredCompactorSupplier {
private final CompactionStrategy strategy;
MockSupplier(CompactionStrategy strategy) {
this.strategy = strategy;
}
@Override
public Map<String,String> options() {
return Collections.emptyMap();
}
@Override
public CompactionStrategy get(Map<String,String> tableProperites) {
return strategy;
}
}
}
@@ -0,0 +1,119 @@
package timely.server.test;
import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.TreeMap;
import org.apache.accumulo.core.client.lexicoder.Lexicoder;
import org.apache.accumulo.core.client.lexicoder.PairLexicoder;
import org.apache.accumulo.core.client.lexicoder.StringLexicoder;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.DefaultConfiguration;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.data.TabletId;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.dataImpl.TabletIdImpl;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.util.ComparablePair;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.tserver.compaction.MajorCompactionReason;
import org.apache.accumulo.tserver.compaction.MajorCompactionRequest;
import org.apache.hadoop.io.Text;
import org.easymock.EasyMock;
import timely.accumulo.MetricAdapter;
public class CompactionRequestBuilder {
private static final long DEFAULT_TIME_MILLIS = new GregorianCalendar(2019, Calendar.JANUARY, 1).getTimeInMillis();
private static final Lexicoder<String> stringEncoder = new StringLexicoder();
private static final PairLexicoder<String,String> badPairEncoder = new PairLexicoder<>(stringEncoder, stringEncoder);
private final Map<StoredTabletFile,DataFileValue> refs = new LinkedHashMap<>();
private final Map<String,String> tableProperties;
private final long timeMillis;
private Key endKeyMetric;
private Key prevKeyMetric;
public CompactionRequestBuilder() {
this(DEFAULT_TIME_MILLIS);
}
public CompactionRequestBuilder(long timeMillis) {
this.timeMillis = timeMillis;
this.tableProperties = new TreeMap<>();
}
public CompactionRequestBuilder endKeyMetricNonOffset(String name, String offset) {
byte[] rowKey = offset != null ? badPairEncoder.encode(new ComparablePair<>(name, offset)) : stringEncoder.encode(name);
endKeyMetric = new Key(rowKey);
return this;
}
public CompactionRequestBuilder endKeyMetric(String name, long offset) {
long ts = timeMillis - offset;
byte[] rowKey = MetricAdapter.encodeRowKey(name, ts);
endKeyMetric = new Key(rowKey);
return this;
}
public CompactionRequestBuilder prevEndKeyMetric(String name, long offset) {
long ts = timeMillis - offset;
byte[] rowKey = MetricAdapter.encodeRowKey(name, ts);
prevKeyMetric = new Key(rowKey);
return this;
}
public CompactionRequestBuilder prevEndKeyMetricNonOffset(String name, String offset) {
byte[] rowKey = offset != null ? badPairEncoder.encode(new ComparablePair<>(name, offset)) : stringEncoder.encode(name);
prevKeyMetric = new Key(rowKey);
return this;
}
public CompactionRequestBuilder file(String path, long size, long entries) {
refs.put(new StoredTabletFile(path), new DataFileValue(size, entries));
return this;
}
public CompactionRequestBuilder tableProperties(String key, String value) {
tableProperties.put(key, value);
return this;
}
public MajorCompactionRequest build() {
TableId tableId = TableId.of("1");
Text endRow = null;
if (null != endKeyMetric) {
endRow = endKeyMetric.getRow();
}
Text prevEndRow = null;
if (null != prevKeyMetric) {
prevEndRow = prevKeyMetric.getRow();
}
KeyExtent ke = new KeyExtent(tableId, endRow, prevEndRow);
TabletId tid = new TabletIdImpl(ke);
// @formatter:off
MajorCompactionRequest req = EasyMock.partialMockBuilder(MajorCompactionRequest.class)
.withConstructor(KeyExtent.class, MajorCompactionReason.class,
AccumuloConfiguration.class, ServerContext.class)
.withArgs(ke, MajorCompactionReason.NORMAL, DefaultConfiguration.getInstance(), null)
.addMockedMethod("getTabletId")
.addMockedMethod("getTableProperties")
.createMock();
// @formatter:on
req.setFiles(refs);
EasyMock.expect(req.getTabletId()).andStubReturn(tid);
EasyMock.expect(req.getTableProperties()).andStubReturn(tableProperties);
EasyMock.replay(req);
return req;
}
}
+3 -3
View File
@@ -4,12 +4,12 @@
<parent>
<groupId>timely</groupId>
<artifactId>timely</artifactId>
<version>3.0.0-SNAPSHOT</version>
<version>2.1.4-SNAPSHOT</version>
</parent>
<artifactId>timely-test</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<maven.compiler.source>11</maven.compiler.source>
<maven.compiler.target>11</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
@@ -9,8 +9,8 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import datawave.accumulo.inmemory.InMemoryAccumulo;
import datawave.accumulo.inmemory.InMemoryAccumuloClient;
import datawave.accumulo.inmemory.InMemoryInstance;
import timely.common.configuration.AccumuloProperties;
@Configuration
@@ -20,15 +20,15 @@ public class InMemoryAccumuloConfiguration {
protected static final String ROOT_USER = "root";
@Bean(name = "InMemoryAccumulo")
@Bean(name = "InMemoryInstance")
@Singleton
public InMemoryAccumulo inMemoryAccumulo() throws Exception {
return new InMemoryAccumulo();
public InMemoryInstance inMemoryInstance() throws Exception {
return new InMemoryInstance();
}
@Bean
@Primary
public AccumuloClient inMemoryAccumuloClient(InMemoryAccumulo instance) throws Exception {
public AccumuloClient inMemoryAccumuloClient(InMemoryInstance instance) throws Exception {
return new InMemoryAccumuloClient(ROOT_USER, instance);
}
}
@@ -69,6 +69,7 @@ public class MiniAccumuloConfiguration {
}
macConfig.setZooKeeperPort(zookeeperPort);
macConfig.setNumTservers(1);
macConfig.setMemory(ServerType.TABLET_SERVER, 1, MemoryUnit.GIGABYTE);
try {
miniAccumuloCluster = new MiniAccumuloCluster(macConfig);