diff --git a/.gitignore b/.gitignore index 133071f206..a66d70433b 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,8 @@ pom.xml */target/* *.orig *.log +build-aux/fetchdep.sh +opentsdb.spec #for Intellij \.idea @@ -50,3 +52,4 @@ tools/docker/opentsdb.conf fat-jar-pom.xml src-resources/ test-resources/ +third_party/**/*.jar \ No newline at end of file diff --git a/Makefile.am b/Makefile.am index d3ce9287e7..becc811b12 100644 --- a/Makefile.am +++ b/Makefile.am @@ -78,6 +78,7 @@ tsdb_SRC := \ src/core/RequestBuilder.java \ src/core/RowKey.java \ src/core/RowSeq.java \ + src/core/RpcResponder.java \ src/core/iRowSeq.java \ src/core/SaltScanner.java \ src/core/SeekableView.java \ @@ -120,6 +121,7 @@ tsdb_SRC := \ src/query/expression/ExpressionReader.java \ src/query/expression/Expressions.java \ src/query/expression/ExpressionTree.java \ + src/query/expression/FirstDifference.java \ src/query/expression/HighestCurrent.java \ src/query/expression/HighestMax.java \ src/query/expression/IntersectionIterator.java \ @@ -178,7 +180,9 @@ tsdb_SRC := \ src/tools/TSDMain.java \ src/tools/TextImporter.java \ src/tools/TreeSync.java \ + src/tools/UidGarbageCollector.java \ src/tools/UidManager.java \ + src/tools/Uids.java \ src/tools/ArgValueValidator.java \ src/tools/ConfigArgP.java \ src/tools/ConfigMetaType.java \ @@ -316,6 +320,7 @@ test_SRC := \ test/core/TestRateSpan.java \ test/core/TestRowKey.java \ test/core/TestRowSeq.java \ + test/core/TestRpcResponsder.java \ test/core/TestSaltScanner.java \ test/core/TestSpan.java \ test/core/TestSpanGroup.java \ diff --git a/configure.ac b/configure.ac index 659a5cd976..fe726e760a 100644 --- a/configure.ac +++ b/configure.ac @@ -14,7 +14,7 @@ # along with this library. If not, see . # Semantic Versioning (see http://semver.org/). -AC_INIT([opentsdb], [2.4.0RC2], [opentsdb@googlegroups.com]) +AC_INIT([opentsdb], [2.5.0-SNAPSHOT], [opentsdb@googlegroups.com]) AC_CONFIG_AUX_DIR([build-aux]) AM_INIT_AUTOMAKE([foreign]) diff --git a/src/core/RpcResponder.java b/src/core/RpcResponder.java new file mode 100644 index 0000000000..97e7b22bb8 --- /dev/null +++ b/src/core/RpcResponder.java @@ -0,0 +1,110 @@ +// This file is part of OpenTSDB. +// Copyright (C) 2010-2017 The OpenTSDB Authors. +// +// This program is free software: you can redistribute it and/or modify it +// under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 2.1 of the License, or (at your +// option) any later version. This program is distributed in the hope that it +// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty +// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser +// General Public License for more details. You should have received a copy +// of the GNU Lesser General Public License along with this program. If not, +// see . +package net.opentsdb.core; + +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import net.opentsdb.utils.Config; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * This class is responsible for building result of requests and + * respond to clients asynchronously. + * + * It can reduce requests that stacking in AsyncHBase, especially put requests. + * When a HBase's RPC has completed, the "AsyncHBase I/O worker" just decodes + * the response, and then do callback by this class asynchronously. We should + * take up workers as short as possible time so that workers can remove RPCs + * from in-flight state more quickly. + * + */ +public class RpcResponder { + + private static final Logger LOG = LoggerFactory.getLogger(RpcResponder.class); + + public static final String TSD_RESPONSE_ASYNC_KEY = "tsd.core.response.async"; + public static final boolean TSD_RESPONSE_ASYNC_DEFAULT = true; + + public static final String TSD_RESPONSE_WORKER_NUM_KEY = + "tsd.core.response.worker.num"; + public static final int TSD_RESPONSE_WORKER_NUM_DEFAULT = 10; + + private final boolean async; + private ExecutorService responders; + private volatile boolean running = true; + + RpcResponder(final Config config) { + async = config.getBoolean(TSD_RESPONSE_ASYNC_KEY, + TSD_RESPONSE_ASYNC_DEFAULT); + + if (async) { + int threads = config.getInt(TSD_RESPONSE_WORKER_NUM_KEY, + TSD_RESPONSE_WORKER_NUM_DEFAULT); + responders = Executors.newFixedThreadPool(threads, + new ThreadFactoryBuilder() + .setNameFormat("OpenTSDB Responder #%d") + .setDaemon(true) + .setUncaughtExceptionHandler(new ExceptionHandler()) + .build()); + } + + LOG.info("RpcResponder mode: {}", async ? "async" : "sync"); + } + + public void response(Runnable run) { + if (async) { + if (running) { + responders.execute(run); + } else { + throw new IllegalStateException("RpcResponder is closing or closed."); + } + } else { + run.run(); + } + } + + public void close() { + if (running) { + running = false; + responders.shutdown(); + } + + boolean completed; + try { + completed = responders.awaitTermination(5, TimeUnit.MINUTES); + } catch (InterruptedException e) { + completed = false; + } + + if (!completed) { + LOG.warn( + "There are still some results that are not returned to the clients."); + } + } + + public boolean isAsync() { + return async; + } + + private class ExceptionHandler implements Thread.UncaughtExceptionHandler { + @Override + public void uncaughtException(Thread t, Throwable e) { + LOG.error("Run into an uncaught exception in thread: " + t.getName(), e); + } + } +} diff --git a/src/core/TSDB.java b/src/core/TSDB.java index 119c207a24..26f2e0f71f 100644 --- a/src/core/TSDB.java +++ b/src/core/TSDB.java @@ -134,6 +134,9 @@ public enum OperationMode { /** Timer used for various tasks such as idle timeouts or query timeouts */ private final HashedWheelTimer timer; + /** RpcResponder for doing response asynchronously*/ + private final RpcResponder rpcResponder; + /** * Row keys that need to be compacted. * Whenever we write a new data point to a row, we add the row key to this @@ -343,7 +346,10 @@ public TSDB(final HBaseClient client, final Config config) { // set any extra tags from the config for stats StatsCollector.setGlobalTags(config); - + + + rpcResponder = new RpcResponder(config); + LOG.debug(config.dumpConfiguration()); } @@ -1657,20 +1663,43 @@ public String toString() { } } + final class RpcResponsderShutdown implements Callback { + @Override + public Object call(Object arg) throws Exception { + try { + TSDB.this.rpcResponder.close(); + } catch (Exception e) { + LOG.error( + "Run into unknown exception while closing RpcResponder.", e); + } finally { + return arg; + } + } + } + final class HClientShutdown implements Callback, ArrayList> { - public Deferred call(final ArrayList args) { + public Deferred call(final ArrayList args) { + Callback nextCallback; if (storage_exception_handler != null) { - return client.shutdown().addBoth(new SEHShutdown()); + nextCallback = new SEHShutdown(); + } else { + nextCallback = new FinalShutdown(); } - return client.shutdown().addBoth(new FinalShutdown()); + + if (TSDB.this.rpcResponder.isAsync()) { + client.shutdown().addBoth(new RpcResponsderShutdown()); + } + + return client.shutdown().addBoth(nextCallback); } - public String toString() { + + public String toString() { return "shutdown HBase client"; } } final class ShutdownErrback implements Callback { - public Object call(final Exception e) { + public Object call(final Exception e) { final Logger LOG = LoggerFactory.getLogger(ShutdownErrback.class); if (e instanceof DeferredGroupException) { final DeferredGroupException ge = (DeferredGroupException) e; @@ -1684,13 +1713,14 @@ public Object call(final Exception e) { } return new HClientShutdown().call(null); } - public String toString() { + + public String toString() { return "shutdown HBase client after error"; } } final class CompactCB implements Callback> { - public Object call(ArrayList compactions) throws Exception { + public Object call(ArrayList compactions) throws Exception { return null; } } @@ -2189,4 +2219,9 @@ final Deferred delete(final byte[] key, final byte[][] qualifiers) { return client.delete(new DeleteRequest(table, key, FAMILY, qualifiers)); } + /** Do response by RpcResponder */ + public void response(Runnable run) { + rpcResponder.response(run); + } + } diff --git a/src/core/TsdbQuery.java b/src/core/TsdbQuery.java index a0dd841d6a..9b56c81ad1 100644 --- a/src/core/TsdbQuery.java +++ b/src/core/TsdbQuery.java @@ -1544,14 +1544,15 @@ private long getScanStartTimeSeconds() { } /** Returns the UNIX timestamp at which we must stop scanning. */ - private long getScanEndTimeSeconds() { + @VisibleForTesting + protected long getScanEndTimeSeconds() { // Begin with the raw query end time. long end = getEndTime(); // Convert to seconds if we have a query in ms. if ((end & Const.SECOND_MASK) != 0L) { end /= 1000L; - if (end - (end * 1000) < 1) { + if (end == 0) { // handle an edge case where a user may request a ms time between // 0 and 1 seconds. Just bump it a second. end++; diff --git a/src/query/expression/ExpressionFactory.java b/src/query/expression/ExpressionFactory.java index e0fbdd44e8..43358e6eb6 100644 --- a/src/query/expression/ExpressionFactory.java +++ b/src/query/expression/ExpressionFactory.java @@ -37,6 +37,7 @@ public final class ExpressionFactory { available_functions.put("highestMax", new HighestMax()); available_functions.put("shift", new TimeShift()); available_functions.put("timeShift", new TimeShift()); + available_functions.put("firstDiff", new FirstDifference()); } /** Don't instantiate me! */ diff --git a/src/query/expression/FirstDifference.java b/src/query/expression/FirstDifference.java new file mode 100644 index 0000000000..6dc75bc088 --- /dev/null +++ b/src/query/expression/FirstDifference.java @@ -0,0 +1,101 @@ +// This file is part of OpenTSDB. +// Copyright (C) 2015 The OpenTSDB Authors. +// +// This program is free software: you can redistribute it and/or modify it +// under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 2.1 of the License, or (at your +// option) any later version. This program is distributed in the hope that it +// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty +// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser +// General Public License for more details. You should have received a copy +// of the GNU Lesser General Public License along with this program. If not, +// see . +package net.opentsdb.query.expression; + +import java.util.ArrayList; + +import java.util.List; + +import net.opentsdb.core.DataPoint; +import net.opentsdb.core.DataPoints; +import net.opentsdb.core.IllegalDataException; +import net.opentsdb.core.MutableDataPoint; +import net.opentsdb.core.SeekableView; +import net.opentsdb.core.TSQuery; +import net.opentsdb.core.Aggregators.Interpolation; + +/** + * Implements a difference function, calculates the first difference of a given series + * + * @since 2.3 + */ +public class FirstDifference implements net.opentsdb.query.expression.Expression { + + @Override + public DataPoints[] evaluate(final TSQuery data_query, + final List query_results, final List params) { + if (data_query == null) { + throw new IllegalArgumentException("Missing time series query"); + } + if (query_results == null || query_results.isEmpty()) { + return new DataPoints[]{}; + } + + + int num_results = 0; + for (final DataPoints[] results : query_results) { + num_results += results.length; + } + final DataPoints[] results = new DataPoints[num_results]; + + int ix = 0; + // one or more sub queries (m=...&m=...&m=...) + for (final DataPoints[] sub_query_result : query_results) { + // group bys (m=sum:foo{host=*}) + for (final DataPoints dps : sub_query_result) { + results[ix++] = firstDiff(dps); + } + } + + return results; + + } + + /** + * return the first difference of datapoints + * + * @param points The data points to do difference + * @return The resulting data points + */ + private DataPoints firstDiff(final DataPoints points) { + final List dps = new ArrayList(); + final SeekableView view = points.iterator(); + List nums = new ArrayList(); + List times = new ArrayList(); + while (view.hasNext()) { + DataPoint pt = view.next(); + nums.add(pt.toDouble()); + times.add(pt.timestamp()); + } + List diff = new ArrayList(); + diff.add(0.0); + for (int j =0;j query_params, + final String inner_expression) { + return "firstDiff(" + inner_expression + ")"; + } + +} \ No newline at end of file diff --git a/src/tools/FsckOptions.java b/src/tools/FsckOptions.java index 9112b45564..9fc005e71e 100644 --- a/src/tools/FsckOptions.java +++ b/src/tools/FsckOptions.java @@ -96,6 +96,7 @@ public static void addDataOptions(final ArgP argp) { "Delete compacted columns that cannot be parsed."); argp.addOption("--threads", "NUMBER", "Number of threads to use when executing a full table scan."); + argp.addOption("--sync", "Wait for each fix operation to finish to continue."); } /** @return Whether or not to fix errors while processing. Does not affect diff --git a/src/tools/UidGarbageCollector.java b/src/tools/UidGarbageCollector.java new file mode 100644 index 0000000000..6ab5ec8942 --- /dev/null +++ b/src/tools/UidGarbageCollector.java @@ -0,0 +1,262 @@ +// This file is part of OpenTSDB. +// Copyright (C) 2013 The OpenTSDB Authors. +// +// This program is free software: you can redistribute it and/or modify it +// under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 2.1 of the License, or (at your +// option) any later version. This program is distributed in the hope that it +// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty +// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser +// General Public License for more details. You should have received a copy +// of the GNU Lesser General Public License along with this program. If not, +// see . +package net.opentsdb.tools; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.HashMap; + +import net.opentsdb.core.Const; +import net.opentsdb.core.TSDB; +import net.opentsdb.meta.TSMeta; +import net.opentsdb.meta.UIDMeta; +import net.opentsdb.uid.NoSuchUniqueId; +import net.opentsdb.uid.UniqueId; +import net.opentsdb.uid.UniqueId.UniqueIdType; + +import org.hbase.async.Bytes; +import org.hbase.async.KeyValue; +import org.hbase.async.Scanner; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.stumbleupon.async.Callback; +import com.stumbleupon.async.Deferred; +import com.stumbleupon.async.DeferredGroupException; + +/** + * Tool helper class that removes known UIDs from in-memory structure. + * This class should only be used by CLI tools as it can take a long + * time to complete. + */ +final class UidGarbageCollector extends Thread { + private static final Logger LOG = LoggerFactory.getLogger(UidGarbageCollector.class); + + /** TSDB to use for storage access */ + final TSDB tsdb; + + /** Map type of UID to class containing it. */ + final HashMap name2uids; + + /** Diagnostic ID for this thread */ + final int thread_id; + + /** The scanner for this worker */ + final Scanner scanner; + + /** + * Constructor that sets local variables + * @param tsdb The TSDB to process with + * @param scanner The scanner to use for this worker + * @param thread_id The ID of this thread (starts at 0) + * @param name2uids The currently known UIDs. + */ + public UidGarbageCollector(final TSDB tsdb, + final Scanner scanner, + final int thread_id, + HashMap name2uids + ) { + this.tsdb = tsdb; + this.scanner = scanner; + this.thread_id = thread_id; + this.name2uids = name2uids; + } + + /** + * Loops through the entire TSDB data set and exits when complete. + */ + public void run() { + // list of deferred calls used to act as a buffer + final ArrayList> storage_calls = + new ArrayList>(); + final Deferred result = new Deferred(); + + final class ErrBack implements Callback { + @Override + public Object call(Exception e) throws Exception { + Throwable ex = e; + while (ex.getClass().equals(DeferredGroupException.class)) { + if (ex.getCause() == null) { + LOG.warn("Unable to get to the root cause of the DGE"); + break; + } + ex = ex.getCause(); + } + LOG.error("Sync thread failed with exception", ex); + result.callback(null); + return null; + } + } + final ErrBack err_back = new ErrBack(); + + /** + * Scanner callback that recursively loops through all of the data point + * rows. Note that we don't process the actual data points, just the row + * keys. + */ + final class MetaScanner implements Callback>> { + + private byte[] last_tsuid = null; + private String tsuid_string = ""; + + /** + * Fetches the next set of rows from the scanner and adds this class as + * a callback + * @return A meaningless deferred to wait on until all data rows have + * been processed. + */ + public Object scan() { + return scanner.nextRows().addCallback(this).addErrback(err_back); + } + + @Override + public Object call(ArrayList> rows) + throws Exception { + if (rows == null) { + result.callback(null); + return null; + } + + final Uids metricsUids = name2uids.get(CliUtils.METRICS); + final Uids tagkUids = name2uids.get(CliUtils.TAGK); + final Uids tagvUids = name2uids.get(CliUtils.TAGV); + + for (final ArrayList row : rows) { + try { + final byte[] tsuid = UniqueId.getTSUIDFromKey(row.get(0).key(), + TSDB.metrics_width(), Const.TIMESTAMP_BYTES); + + // if the current tsuid is the same as the last, just continue + // so we save time + if (last_tsuid != null && Arrays.equals(last_tsuid, tsuid)) { + continue; + } + last_tsuid = tsuid; + + tsuid_string = UniqueId.uidToString(tsuid); + + /** + * An error callback used to catch issues with a particular timeseries + * or UIDMeta such as a missing UID name. We want to continue + * processing when this happens so we'll just log the error and + * the user can issue a command later to clean up orphaned meta + * entries. + */ + final class RowErrBack implements Callback { + @Override + public Object call(Exception e) throws Exception { + Throwable ex = e; + while (ex.getClass().equals(DeferredGroupException.class)) { + if (ex.getCause() == null) { + LOG.warn("Unable to get to the root cause of the DGE"); + break; + } + ex = ex.getCause(); + } + if (ex.getClass().equals(IllegalStateException.class)) { + LOG.error("Invalid data when processing TSUID [" + + tsuid_string + "]: " + ex.getMessage()); + } else if (ex.getClass().equals(IllegalArgumentException.class)) { + LOG.error("Invalid data when processing TSUID [" + + tsuid_string + "]: " + ex.getMessage()); + } else if (ex.getClass().equals(NoSuchUniqueId.class)) { + LOG.warn("Timeseries [" + tsuid_string + + "] includes a non-existant UID: " + ex.getMessage()); + } else { + LOG.error("Unknown exception processing row: " + row, ex); + } + return null; + } + } + + LOG.debug("[" + thread_id + "] Processing TSUID: " + tsuid_string); + + // now mark the UIDs as present: + final byte[] metric_uid_bytes = + Arrays.copyOfRange(tsuid, 0, TSDB.metrics_width()); + final String metric_uid = UniqueId.uidToString(metric_uid_bytes); + metricsUids.localRemoveUid(metric_uid); + + // loop through the tags and mark them as present too: + final List tags = UniqueId.getTagsFromTSUID(tsuid_string); + int idx = 0; + for (byte[] tag : tags) { + final UniqueIdType type = (idx % 2 == 0) ? UniqueIdType.TAGK : + UniqueIdType.TAGV; + idx++; + final String uid = UniqueId.uidToString(tag); + if (type == UniqueIdType.TAGK) { + tagkUids.localRemoveUid(uid); + } else { + tagvUids.localRemoveUid(uid); + } + } + } catch (RuntimeException e) { + LOG.error("Processing row " + row + " failed with exception: " + + e.getMessage()); + LOG.debug("Row: " + row + " stack trace: ", e); + } + } + + /** + * A buffering callback used to avoid StackOverflowError exceptions + * where the list of deferred calls can exceed the limit. Instead we'll + * process the Scanner's limit in rows, wait for all of the storage + * calls to complete, then continue on to the next set. + */ + final class ContinueCB implements Callback> { + + @Override + public Object call(ArrayList puts) + throws Exception { + storage_calls.clear(); + return scan(); + } + } + + /** + * Catch exceptions in one of the grouped calls and continue scanning. + * Without this the user may not see the exception and the thread will + * just die silently. + */ + final class ContinueEB implements Callback { + @Override + public Object call(Exception e) throws Exception { + + Throwable ex = e; + while (ex.getClass().equals(DeferredGroupException.class)) { + if (ex.getCause() == null) { + LOG.warn("Unable to get to the root cause of the DGE"); + break; + } + ex = ex.getCause(); + } + LOG.error("[" + thread_id + "] Upstream Exception: ", ex); + return scan(); + } + } + + // call ourself again but wait for the current set of storage calls to + // complete so we don't OOM + Deferred.group(storage_calls).addCallback(new ContinueCB()) + .addErrback(new ContinueEB()); + return null; + } + } + } + +} diff --git a/src/tools/UidManager.java b/src/tools/UidManager.java index 1eead8bf10..efd1a0a907 100644 --- a/src/tools/UidManager.java +++ b/src/tools/UidManager.java @@ -22,6 +22,7 @@ import java.util.Set; import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; +import com.stumbleupon.async.Deferred; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -454,177 +455,17 @@ private static int fsck(final HBaseClient client, final byte[] table, } else { LOG.info("Running in log only mode"); } - - final class Uids { - int errors; - long maxid; - long max_found_id; - short width; - final HashMap id2name = new HashMap(); - final HashMap name2id = new HashMap(); - - void error(final KeyValue kv, final String msg) { - error(msg + ". kv=" + kv); - } - - void error(final String msg) { - LOG.error(msg); - errors++; - } - - /* - * Replaces or creates the reverse map in storage and in the local map - */ - void restoreReverseMap(final String kind, final String name, - final String uid) { - final PutRequest put = new PutRequest(table, - UniqueId.stringToUid(uid), CliUtils.NAME_FAMILY, CliUtils.toBytes(kind), - CliUtils.toBytes(name)); - client.put(put); - id2name.put(uid, name); - LOG.info("FIX: Restoring " + kind + " reverse mapping: " - + uid + " -> " + name); - } - - /* - * Removes the reverse map from storage only - */ - void removeReverseMap(final String kind, final String name, - final String uid) { - // clean up meta data too - final byte[][] qualifiers = new byte[2][]; - qualifiers[0] = CliUtils.toBytes(kind); - if (Bytes.equals(CliUtils.METRICS, qualifiers[0])) { - qualifiers[1] = CliUtils.METRICS_META; - } else if (Bytes.equals(CliUtils.TAGK, qualifiers[0])) { - qualifiers[1] = CliUtils.TAGK_META; - } else if (Bytes.equals(CliUtils.TAGV, qualifiers[0])) { - qualifiers[1] = CliUtils.TAGV_META; - } - - final DeleteRequest delete = new DeleteRequest(table, - UniqueId.stringToUid(uid), CliUtils.NAME_FAMILY, qualifiers); - client.delete(delete); - // can't remove from the id2name map as this will be called while looping - LOG.info("FIX: Removed " + kind + " reverse mapping: " + uid + " -> " - + name); - } - } - final long start_time = System.nanoTime(); - final HashMap name2uids = new HashMap(); - final Scanner scanner = client.newScanner(table); - scanner.setMaxNumRows(1024); + HashMap name2uids = Uids.loadUids(client, table, LOG, fix, + fix_unknowns); int kvcount = 0; - try { - ArrayList> rows; - while ((rows = scanner.nextRows().joinUninterruptibly()) != null) { - for (final ArrayList row : rows) { - for (final KeyValue kv : row) { - kvcount++; - final byte[] qualifier = kv.qualifier(); - // TODO - validate meta data in the future, for now skip it - if (Bytes.equals(qualifier, TSMeta.META_QUALIFIER()) || - Bytes.equals(qualifier, TSMeta.COUNTER_QUALIFIER()) || - Bytes.equals(qualifier, CliUtils.METRICS_META) || - Bytes.equals(qualifier, CliUtils.TAGK_META) || - Bytes.equals(qualifier, CliUtils.TAGV_META)) { - continue; - } - - if (!Bytes.equals(qualifier, CliUtils.METRICS) && - !Bytes.equals(qualifier, CliUtils.TAGK) && - !Bytes.equals(qualifier, CliUtils.TAGV)) { - LOG.warn("Unknown qualifier " + UniqueId.uidToString(qualifier) - + " in row " + UniqueId.uidToString(kv.key())); - if (fix && fix_unknowns) { - final DeleteRequest delete = new DeleteRequest(table, kv.key(), - kv.family(), qualifier); - client.delete(delete); - LOG.info("FIX: Removed unknown qualifier " - + UniqueId.uidToString(qualifier) - + " in row " + UniqueId.uidToString(kv.key())); - } - continue; - } - - final String kind = CliUtils.fromBytes(kv.qualifier()); - Uids uids = name2uids.get(kind); - if (uids == null) { - uids = new Uids(); - name2uids.put(kind, uids); - } - final byte[] key = kv.key(); - final byte[] family = kv.family(); - final byte[] value = kv.value(); - if (Bytes.equals(key, CliUtils.MAXID_ROW)) { - if (value.length != 8) { - uids.error(kv, "Invalid maximum ID for " + kind - + ": should be on 8 bytes: "); - // TODO - a fix would be to find the max used ID for the type - // and store that in the max row. - } else { - uids.maxid = Bytes.getLong(value); - LOG.info("Maximum ID for " + kind + ": " + uids.maxid); - } - } else { - short idwidth = 0; - if (Bytes.equals(family, CliUtils.ID_FAMILY)) { - idwidth = (short) value.length; - final String skey = CliUtils.fromBytes(key); - final String svalue = UniqueId.uidToString(value); - final long max_found_id; - if (Bytes.equals(qualifier, CliUtils.METRICS)) { - max_found_id = UniqueId.uidToLong(value, TSDB.metrics_width()); - } else if (Bytes.equals(qualifier, CliUtils.TAGK)) { - max_found_id = UniqueId.uidToLong(value, TSDB.tagk_width()); - } else { - max_found_id = UniqueId.uidToLong(value, TSDB.tagv_width()); - } - if (uids.max_found_id < max_found_id) { - uids.max_found_id = max_found_id; - } - final String id = uids.name2id.put(skey, svalue); - if (id != null) { - uids.error(kv, "Duplicate forward " + kind + " mapping: " - + skey + " -> " + id - + " and " + skey + " -> " + svalue); - } - } else if (Bytes.equals(family, CliUtils.NAME_FAMILY)) { - final String skey = UniqueId.uidToString(key); - final String svalue = CliUtils.fromBytes(value); - idwidth = (short) key.length; - final String name = uids.id2name.put(skey, svalue); - if (name != null) { - uids.error(kv, "Duplicate reverse " + kind + " mapping: " - + svalue + " -> " + name - + " and " + svalue + " -> " + skey); - } - } - if (uids.width == 0) { - uids.width = idwidth; - } else if (uids.width != idwidth) { - uids.error(kv, "Invalid " + kind + " ID of length " + idwidth - + " (expected: " + uids.width + ')'); - } - } - } - } - } - } catch (HBaseException e) { - LOG.error("Error while scanning HBase, scanner=" + scanner, e); - throw e; - } catch (Exception e) { - LOG.error("WTF? Unexpected exception type, scanner=" + scanner, e); - throw new AssertionError("Should never happen"); - } - // Match up all forward mappings with their reverse mappings and vice // versa and make sure they agree. int errors = 0; for (final Map.Entry entry : name2uids.entrySet()) { final String kind = entry.getKey(); final Uids uids = entry.getValue(); + kvcount += uids.id2name.size(); // This will be used in the event that we run into an inconsistent forward // mapping that could mean a single UID was assigned to different names. @@ -1039,6 +880,86 @@ private static int metaSync(final TSDB tsdb) throws Exception { duration + "] seconds"); return 0; } + + /** + * Runs through the entire data table and delete UIDs that are no longer used + * from the UID table. + * + * The process is as follows: + *
    + *
  • Fetch all known UIDs from the UID table, store as unusedUIDs.
    • + *
  • Fetch the max number of Metric UIDs as we'll use those to match + * on the data rows
    • + *
  • Split the # of UIDs amongst worker threads
    • + *
  • Setup a scanner in each thread for the range it will be working on and + * start iterating
    • + *
  • Fetch the TSUID from the row key
    • + *
  • For each unprocessed TSUID, remove the metric, tagk, and tagv UIDs + * from unusedUIDs.
    • + *
  • When done iterating, any remaining UIDs in unusedUIDS are deleted from + * the UID table.
+ * @param tsdb The tsdb to use for processing. + * @param table The table name for where data is stored. + * @return 0 if completed successfully, something else if it dies + */ + private static int uidGarbageCollect(final TSDB tsdb, final byte[] table) throws Exception { + final long start_time = System.currentTimeMillis() / 1000; + + // get current uids: + HashMap unusedUids = Uids.loadUids(tsdb.getClient(), table, + LOG, false, false); + + // now figure out how many IDs to divy up between the workers + final int workers = Runtime.getRuntime().availableProcessors() * 2; + final Set processed_tsuids = + Collections.synchronizedSet(new HashSet()); + final ConcurrentHashMap metric_uids = + new ConcurrentHashMap(); + final ConcurrentHashMap tagk_uids = + new ConcurrentHashMap(); + final ConcurrentHashMap tagv_uids = + new ConcurrentHashMap(); + + // TODO optimize by making this key only, and first key only, and batching, + // etc. + final List scanners = CliUtils.getDataTableScanners(tsdb, workers); + LOG.info("Spooling up [" + scanners.size() + "] worker threads"); + final List threads = new ArrayList(scanners.size()); + int i = 0; + for (final Scanner scanner : scanners) { + final UidGarbageCollector worker = new UidGarbageCollector(tsdb, scanner, i++, + unusedUids); + worker.setName("UID GC Scan #" + i); + worker.start(); + threads.add(worker); + } + + for (final Thread thread : threads) { + thread.join(); + LOG.info("Thread [" + thread + "] Finished"); + } + LOG.info("All UID GC Scan threads have completed"); + + // At this point the UidGarbageCollector threads should have deleted all + // UIDs that are actually being used. The remaining UIDs are garbage and can + // be deleted. + final ArrayList> deletes = new ArrayList>(); + for (final String kind: unusedUids.keySet()) { + Uids uids = unusedUids.get(kind); + for (final String name: uids.name2id.keySet()) { + deletes.add(tsdb.deleteUidAsync(kind, name)); + } + } + Deferred.group(deletes).join(); + + // make sure buffered data is flushed to storage before exiting + tsdb.flush().joinUninterruptibly(); + + final long duration = (System.currentTimeMillis() / 1000) - start_time; + LOG.info("Completed UID garbage collection in [" + + duration + "] seconds"); + return 0; + } /** * Runs through the tsdb-uid table and removes TSMeta, UIDMeta and TSUID diff --git a/src/tools/Uids.java b/src/tools/Uids.java new file mode 100644 index 0000000000..f5d4dc3c6b --- /dev/null +++ b/src/tools/Uids.java @@ -0,0 +1,232 @@ +// This file is part of OpenTSDB. +// Copyright (C) 2014 The OpenTSDB Authors. +// +// This program is free software: you can redistribute it and/or modify it +// under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 2.1 of the License, or (at your +// option) any later version. This program is distributed in the hope that it +// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty +// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser +// General Public License for more details. You should have received a copy +// of the GNU Lesser General Public License along with this program. If not, +// see . +package net.opentsdb.tools; + +import java.util.ArrayList; +import java.util.concurrent.ConcurrentHashMap; +import java.util.HashMap; + +import org.slf4j.Logger; + +import org.hbase.async.Bytes; +import org.hbase.async.DeleteRequest; +import org.hbase.async.KeyValue; +import org.hbase.async.HBaseClient; +import org.hbase.async.HBaseException; +import org.hbase.async.PutRequest; +import org.hbase.async.Scanner; + +import net.opentsdb.core.TSDB; +import net.opentsdb.meta.TSMeta; +import net.opentsdb.uid.UniqueId; + +/** + * Utility class for fsck and garbage collection of UIDs. + * + * Stores mapping of name to UID and vice versa for particular kind. + */ + +final class Uids { + /* Number of errors found. */ + int errors; + /* Highest possible UID. */ + long maxid; + /* Found highest UID. */ + long max_found_id; + /* Width in bytes of UIDs. */ + short width; + final ConcurrentHashMap id2name = new ConcurrentHashMap(); + final ConcurrentHashMap name2id = new ConcurrentHashMap(); + Logger log; + byte[] table; + HBaseClient client; + + Uids(HBaseClient client, byte[] table, Logger log) { + this.client = client; + this.log = log; + this.table = table; + } + + void error(final KeyValue kv, final String msg) { + error(msg + ". kv=" + kv); + } + + void error(final String msg) { + log.error(msg); + errors++; + } + + /** + * Remove UID from local map (not the storage!). + */ + void localRemoveUid(final String uid) { + // TODO if doesn't exist, don't blow up + final String name = name2id.get(uid); + name2id.remove(uid); + id2name.remove(name); + } + + /** + * Replaces or creates the reverse map in storage and in the local map + */ + void restoreReverseMap(final String kind, final String name, + final String uid) { + final PutRequest put = new PutRequest(table, + UniqueId.stringToUid(uid), CliUtils.NAME_FAMILY, CliUtils.toBytes(kind), + CliUtils.toBytes(name)); + client.put(put); + id2name.put(uid, name); + log.info("FIX: Restoring " + kind + " reverse mapping: " + + uid + " -> " + name); + } + + /** + * Removes the reverse map from storage only + */ + void removeReverseMap(final String kind, final String name, + final String uid) { + // clean up meta data too + final byte[][] qualifiers = new byte[2][]; + qualifiers[0] = CliUtils.toBytes(kind); + if (Bytes.equals(CliUtils.METRICS, qualifiers[0])) { + qualifiers[1] = CliUtils.METRICS_META; + } else if (Bytes.equals(CliUtils.TAGK, qualifiers[0])) { + qualifiers[1] = CliUtils.TAGK_META; + } else if (Bytes.equals(CliUtils.TAGV, qualifiers[0])) { + qualifiers[1] = CliUtils.TAGV_META; + } + + final DeleteRequest delete = new DeleteRequest(table, + UniqueId.stringToUid(uid), CliUtils.NAME_FAMILY, qualifiers); + client.delete(delete); + // can't remove from the id2name map as this will be called while looping + log.info("FIX: Removed " + kind + " reverse mapping: " + uid + " -> " + + name); + } + + /** + * Return mapping from kind (metric/tagk/tagv) to its Uids. + */ + static HashMap loadUids(final HBaseClient client, + final byte[] table, + final Logger log, + final boolean fix, + final boolean fix_unknowns) { + final HashMap name2uids = new HashMap(); + final Scanner scanner = client.newScanner(table); + scanner.setMaxNumRows(1024); + try { + ArrayList> rows; + while ((rows = scanner.nextRows().joinUninterruptibly()) != null) { + for (final ArrayList row : rows) { + for (final KeyValue kv : row) { + final byte[] qualifier = kv.qualifier(); + // TODO - validate meta data in the future, for now skip it + if (Bytes.equals(qualifier, TSMeta.META_QUALIFIER()) || + Bytes.equals(qualifier, TSMeta.COUNTER_QUALIFIER()) || + Bytes.equals(qualifier, CliUtils.METRICS_META) || + Bytes.equals(qualifier, CliUtils.TAGK_META) || + Bytes.equals(qualifier, CliUtils.TAGV_META)) { + continue; + } + + if (!Bytes.equals(qualifier, CliUtils.METRICS) && + !Bytes.equals(qualifier, CliUtils.TAGK) && + !Bytes.equals(qualifier, CliUtils.TAGV)) { + log.warn("Unknown qualifier " + UniqueId.uidToString(qualifier) + + " in row " + UniqueId.uidToString(kv.key())); + if (fix && fix_unknowns) { + final DeleteRequest delete = new DeleteRequest(table, kv.key(), + kv.family(), qualifier); + client.delete(delete); + log.info("FIX: Removed unknown qualifier " + + UniqueId.uidToString(qualifier) + + " in row " + UniqueId.uidToString(kv.key())); + } + continue; + } + + final String kind = CliUtils.fromBytes(kv.qualifier()); + Uids uids = name2uids.get(kind); + if (uids == null) { + uids = new Uids(client, table, log); + name2uids.put(kind, uids); + } + final byte[] key = kv.key(); + final byte[] family = kv.family(); + final byte[] value = kv.value(); + if (Bytes.equals(key, CliUtils.MAXID_ROW)) { + if (value.length != 8) { + uids.error(kv, "Invalid maximum ID for " + kind + + ": should be on 8 bytes: "); + // TODO - a fix would be to find the max used ID for the type + // and store that in the max row. + } else { + uids.maxid = Bytes.getLong(value); + log.info("Maximum ID for " + kind + ": " + uids.maxid); + } + } else { + short idwidth = 0; + if (Bytes.equals(family, CliUtils.ID_FAMILY)) { + idwidth = (short) value.length; + final String skey = CliUtils.fromBytes(key); + final String svalue = UniqueId.uidToString(value); + final long max_found_id; + if (Bytes.equals(qualifier, CliUtils.METRICS)) { + max_found_id = UniqueId.uidToLong(value, TSDB.metrics_width()); + } else if (Bytes.equals(qualifier, CliUtils.TAGK)) { + max_found_id = UniqueId.uidToLong(value, TSDB.tagk_width()); + } else { + max_found_id = UniqueId.uidToLong(value, TSDB.tagv_width()); + } + if (uids.max_found_id < max_found_id) { + uids.max_found_id = max_found_id; + } + final String id = uids.name2id.put(skey, svalue); + if (id != null) { + uids.error(kv, "Duplicate forward " + kind + " mapping: " + + skey + " -> " + id + + " and " + skey + " -> " + svalue); + } + } else if (Bytes.equals(family, CliUtils.NAME_FAMILY)) { + final String skey = UniqueId.uidToString(key); + final String svalue = CliUtils.fromBytes(value); + idwidth = (short) key.length; + final String name = uids.id2name.put(skey, svalue); + if (name != null) { + uids.error(kv, "Duplicate reverse " + kind + " mapping: " + + svalue + " -> " + name + + " and " + svalue + " -> " + skey); + } + } + if (uids.width == 0) { + uids.width = idwidth; + } else if (uids.width != idwidth) { + uids.error(kv, "Invalid " + kind + " ID of length " + idwidth + + " (expected: " + uids.width + ')'); + } + } + } + } + } + } catch (HBaseException e) { + log.error("Error while scanning HBase, scanner=" + scanner, e); + throw e; + } catch (Exception e) { + log.error("WTF? Unexpected exception type, scanner=" + scanner, e); + throw new AssertionError("Should never happen"); + } + return name2uids; + } + +} diff --git a/src/tsd/PutDataPointRpc.java b/src/tsd/PutDataPointRpc.java index f5d82a7643..094fb34696 100644 --- a/src/tsd/PutDataPointRpc.java +++ b/src/tsd/PutDataPointRpc.java @@ -616,60 +616,65 @@ class GroupCB implements Callback> { public GroupCB(final int queued) { this.queued = queued; } - + @Override public Object call(final ArrayList results) { - if (sending_response.get()) { - if (LOG.isDebugEnabled()) { - LOG.debug("Put data point call " + query + " was marked as timedout"); - } - return null; - } else { - sending_response.set(true); - if (timeout != null) { - timeout.cancel(); - } - } - int good_writes = 0; - int failed_writes = 0; - for (final boolean result : results) { - if (result) { - ++good_writes; - } else { - ++failed_writes; - } - } - - final int failures = dps.size() - queued; - if (!show_summary && !show_details) { - if (failures + failed_writes > 0) { - query.sendReply(HttpResponseStatus.BAD_REQUEST, - query.serializer().formatErrorV1( - new BadRequestException(HttpResponseStatus.BAD_REQUEST, - "One or more data points had errors", - "Please see the TSD logs or append \"details\" to the put request"))); - } else { - query.sendReply(HttpResponseStatus.NO_CONTENT, "".getBytes()); - } - } else { - final HashMap summary = new HashMap(); - if (sync_timeout > 0) { - summary.put("timeouts", 0); - } - summary.put("success", results.isEmpty() ? queued : good_writes); - summary.put("failed", failures + failed_writes); - if (show_details) { - summary.put("errors", details); - } - - if (failures > 0) { - query.sendReply(HttpResponseStatus.BAD_REQUEST, - query.serializer().formatPutV1(summary)); - } else { - query.sendReply(query.serializer().formatPutV1(summary)); + tsdb.response(new Runnable() { + @Override + public void run() { + if (sending_response.get()) { + if (LOG.isDebugEnabled()) { + LOG.debug("Put data point call " + query + " was marked as timedout"); + } + return; + } else { + sending_response.set(true); + if (timeout != null) { + timeout.cancel(); + } + } + int good_writes = 0; + int failed_writes = 0; + for (final boolean result : results) { + if (result) { + ++good_writes; + } else { + ++failed_writes; + } + } + + final int failures = dps.size() - queued; + if (!show_summary && !show_details) { + if (failures + failed_writes > 0) { + query.sendReply(HttpResponseStatus.BAD_REQUEST, + query.serializer().formatErrorV1( + new BadRequestException(HttpResponseStatus.BAD_REQUEST, + "One or more data points had errors", + "Please see the TSD logs or append \"details\" to the put request"))); + } else { + query.sendReply(HttpResponseStatus.NO_CONTENT, "".getBytes()); + } + } else { + final HashMap summary = new HashMap(); + if (sync_timeout > 0) { + summary.put("timeouts", 0); + } + summary.put("success", results.isEmpty() ? queued : good_writes); + summary.put("failed", failures + failed_writes); + if (show_details) { + summary.put("errors", details); + } + + if (failures > 0) { + query.sendReply(HttpResponseStatus.BAD_REQUEST, + query.serializer().formatPutV1(summary)); + } else { + query.sendReply(query.serializer().formatPutV1(summary)); + } + } } - } - + }); + return null; } @Override diff --git a/src/utils/Config.java b/src/utils/Config.java index f92f2f5b56..59f7ebd90f 100644 --- a/src/utils/Config.java +++ b/src/utils/Config.java @@ -21,6 +21,7 @@ import java.util.Map; import java.util.Properties; +import net.opentsdb.core.RpcResponder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -340,6 +341,23 @@ public final int getInt(final String property) { return Integer.parseInt(sanitize(properties.get(property))); } + /** + * Returns the given property as an integer. + * If no such property is specified, or if the specified value is not a valid + * Int, then default_val is returned. + * + * @param property The property to load + * @param default_val default value + * @return A parsed integer or default_val. + */ + public final int getInt(final String property, final int default_val) { + try { + return getInt(property); + } catch (Exception e) { + return default_val; + } + } + /** * Returns the given string trimed or null if is null * @param string The string be trimmed of @@ -420,6 +438,23 @@ public final boolean getBoolean(final String property) { return false; } + /** + * Returns the given property as an boolean. + * If no such property is specified, or if the specified value is not a valid + * boolean, then default_val is returned. + * + * @param property The property to load + * @param default_val default value + * @return A parsed boolean or default_val. + */ + public final boolean getBoolean(final String property, final boolean default_val) { + try { + return getBoolean(property); + } catch (Exception e) { + return default_val; + } + } + /** * Returns the directory name, making sure the end is an OS dependent slash * @param property The property to load diff --git a/test/core/TestRpcResponsder.java b/test/core/TestRpcResponsder.java new file mode 100644 index 0000000000..5ddbf73baf --- /dev/null +++ b/test/core/TestRpcResponsder.java @@ -0,0 +1,65 @@ +// This file is part of OpenTSDB. +// Copyright (C) 2010-2017 The OpenTSDB Authors. +// +// This program is free software: you can redistribute it and/or modify it +// under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 2.1 of the License, or (at your +// option) any later version. This program is distributed in the hope that it +// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty +// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser +// General Public License for more details. You should have received a copy +// of the GNU Lesser General Public License along with this program. If not, +// see . +package net.opentsdb.core; + + +import net.opentsdb.utils.Config; +import org.jboss.netty.util.internal.ThreadLocalRandom; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.atomic.AtomicInteger; + +public class TestRpcResponsder { + + private final AtomicInteger complete_counter = new AtomicInteger(0); + + @Test(timeout = 60000) + public void testGracefulShutdown() throws InterruptedException { + RpcResponder rpcResponder = new RpcResponder(new Config()); + + final int n = 100; + for (int i = 0; i < n; i++) { + rpcResponder.response(new MockResponseProcess()); + } + + Thread.sleep(500); + rpcResponder.close(); + + try { + rpcResponder.response(new MockResponseProcess()); + Assert.fail("Expect an IllegalStateException"); + } catch (IllegalStateException ignore) { + } + + Assert.assertEquals(n, complete_counter.get()); + } + + private class MockResponseProcess implements Runnable { + + @Override + public void run() { + long duration = ThreadLocalRandom.current().nextInt(5000); + while (duration > 0) { + try { + Thread.sleep(100); + } catch (InterruptedException ignore) { + } + duration -= 100; + } + complete_counter.incrementAndGet(); + } + } + + +} diff --git a/test/core/TestTsdbQuery.java b/test/core/TestTsdbQuery.java index 366e9951b2..b73e8830e0 100644 --- a/test/core/TestTsdbQuery.java +++ b/test/core/TestTsdbQuery.java @@ -32,6 +32,8 @@ import net.opentsdb.uid.NoSuchUniqueName; import net.opentsdb.utils.DateTime; +import org.jboss.netty.util.internal.ThreadLocalRandom; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -97,6 +99,24 @@ public void setEndTime() throws Exception { assertEquals(1356998400L, query.getEndTime()); } + @Test + public void getScanEndTimeSeconds() { + long now = System.currentTimeMillis() / 1000; + long baseTime = now - (now % Const.MAX_TIMESPAN); + long expectedEndScanTime = baseTime + Const.MAX_TIMESPAN; + + for (int i = 0; i < 3600; i++) { + long sec = baseTime + i; + long ms = sec * 1000 + ThreadLocalRandom.current().nextInt(1000); + query.setEndTime(sec); + Assert.assertEquals("EndTime=" + sec, expectedEndScanTime, + query.getScanEndTimeSeconds()); + query.setEndTime(ms); + Assert.assertEquals("EndTime=" + ms, expectedEndScanTime, + query.getScanEndTimeSeconds()); + } + } + @Test (expected = IllegalStateException.class) public void getStartTimeNotSet() throws Exception { query.getStartTime(); diff --git a/test/query/expression/TestFirstDifference.java b/test/query/expression/TestFirstDifference.java new file mode 100644 index 0000000000..031f0e5e34 --- /dev/null +++ b/test/query/expression/TestFirstDifference.java @@ -0,0 +1,348 @@ +// This file is part of OpenTSDB. +// Copyright (C) 2015 The OpenTSDB Authors. +// +// This program is free software: you can redistribute it and/or modify it +// under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 2.1 of the License, or (at your +// option) any later version. This program is distributed in the hope that it +// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty +// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser +// General Public License for more details. You should have received a copy +// of the GNU Lesser General Public License along with this program. If not, +// see . +package net.opentsdb.query.expression; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import net.opentsdb.core.DataPoint; +import net.opentsdb.core.DataPoints; +import net.opentsdb.core.SeekableView; +import net.opentsdb.core.SeekableViewsForTest; +import net.opentsdb.core.TSQuery; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PowerMockIgnore; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import com.stumbleupon.async.Deferred; + +@RunWith(PowerMockRunner.class) +@PowerMockIgnore({"javax.management.*", "javax.xml.*", + "ch.qos.*", "org.slf4j.*", + "com.sum.*", "org.xml.*"}) +@PrepareForTest({TSQuery.class}) +public class TestFirstDifference { + + private static long START_TIME = 1356998400000L; + private static int INTERVAL = 60000; + private static int NUM_POINTS = 5; + private static String METRIC = "sys.cpu"; + + private TSQuery data_query; + private SeekableView view; + private DataPoints dps; + private DataPoints[] group_bys; + private List query_results; + private List params; + private net.opentsdb.query.expression.FirstDifference func; + + @Before + public void before() throws Exception { + view = SeekableViewsForTest.generator(START_TIME, INTERVAL, + NUM_POINTS, true, 1, 1); + data_query = mock(TSQuery.class); + when(data_query.startTime()).thenReturn(START_TIME); + when(data_query.endTime()).thenReturn(START_TIME + (INTERVAL * NUM_POINTS)); + + dps = PowerMockito.mock(DataPoints.class); + when(dps.iterator()).thenReturn(view); + when(dps.metricNameAsync()).thenReturn(Deferred.fromResult(METRIC)); + + group_bys = new DataPoints[]{dps}; + + query_results = new ArrayList(1); + query_results.add(group_bys); + + params = new ArrayList(1); + func = new net.opentsdb.query.expression.FirstDifference(); + } + + @Test + public void evaluatePositiveGroupByLong() throws Exception { + SeekableView view2 = SeekableViewsForTest.generator(START_TIME, INTERVAL, + NUM_POINTS, true, 10, 1); + DataPoints dps2 = PowerMockito.mock(DataPoints.class); + when(dps2.iterator()).thenReturn(view2); + when(dps2.metricNameAsync()).thenReturn(Deferred.fromResult("sys.mem")); + group_bys = new DataPoints[]{dps, dps2}; + query_results.clear(); + query_results.add(group_bys); + + final DataPoints[] results = func.evaluate(data_query, query_results, params); + + assertEquals(2, results.length); + assertEquals(METRIC, results[0].metricName()); + assertEquals("sys.mem", results[1].metricName()); + + long ts = START_TIME; + long v = 0; + for (DataPoint dp : results[0]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(),0.001); + ts += INTERVAL; + v = 1; + } + ts = START_TIME; + v = 0; + for (DataPoint dp : results[1]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = 1; + } + } + + @Test + public void evaluatePositiveGroupByDouble() throws Exception { + SeekableView view2 = SeekableViewsForTest.generator(START_TIME, INTERVAL, + NUM_POINTS, false, 10, 1); + DataPoints dps2 = PowerMockito.mock(DataPoints.class); + when(dps2.iterator()).thenReturn(view2); + when(dps2.metricNameAsync()).thenReturn(Deferred.fromResult("sys.mem")); + group_bys = new DataPoints[]{dps, dps2}; + query_results.clear(); + query_results.add(group_bys); + + final DataPoints[] results = func.evaluate(data_query, query_results, params); + + assertEquals(2, results.length); + assertEquals(METRIC, results[0].metricName()); + assertEquals("sys.mem", results[1].metricName()); + + long ts = START_TIME; + double v = 0; + for (DataPoint dp : results[0]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v =1; + } + ts = START_TIME; + v = 0; + for (DataPoint dp : results[1]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = 1; + } + } + + @Test + public void evaluatePositiveGroupBy1point5Double() throws Exception { + SeekableView view2 = SeekableViewsForTest.generator(START_TIME, INTERVAL, + NUM_POINTS, false, 10, 1.5); + DataPoints dps2 = PowerMockito.mock(DataPoints.class); + when(dps2.iterator()).thenReturn(view2); + when(dps2.metricNameAsync()).thenReturn(Deferred.fromResult("sys.mem")); + group_bys = new DataPoints[]{dps, dps2}; + query_results.clear(); + query_results.add(group_bys); + + final DataPoints[] results = func.evaluate(data_query, query_results, params); + + assertEquals(2, results.length); + assertEquals(METRIC, results[0].metricName()); + assertEquals("sys.mem", results[1].metricName()); + + long ts = START_TIME; + double v = 0; + for (DataPoint dp : results[0]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v =1; + } + ts = START_TIME; + v = 0; + for (DataPoint dp : results[1]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = 1.5; + } + } + + @Test + public void evaluateFactorNegativeGroupByLong() throws Exception { + SeekableView view2 = SeekableViewsForTest.generator(START_TIME, INTERVAL, + NUM_POINTS, true, -10, -1); + DataPoints dps2 = PowerMockito.mock(DataPoints.class); + when(dps2.iterator()).thenReturn(view2); + when(dps2.metricNameAsync()).thenReturn(Deferred.fromResult("sys.mem")); + group_bys = new DataPoints[]{dps, dps2}; + query_results.clear(); + query_results.add(group_bys); + + final DataPoints[] results = func.evaluate(data_query, query_results, params); + + assertEquals(2, results.length); + assertEquals(METRIC, results[0].metricName()); + assertEquals("sys.mem", results[1].metricName()); + + long ts = START_TIME; + long v = 0; + for (DataPoint dp : results[0]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = 1; + } + ts = START_TIME; + v = 0; + for (DataPoint dp : results[1]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = -1; + } + } + + @Test + public void evaluateNegativeGroupByDouble() throws Exception { + SeekableView view2 = SeekableViewsForTest.generator(START_TIME, INTERVAL, + NUM_POINTS, false, -10, -1); + DataPoints dps2 = PowerMockito.mock(DataPoints.class); + when(dps2.iterator()).thenReturn(view2); + when(dps2.metricNameAsync()).thenReturn(Deferred.fromResult("sys.mem")); + group_bys = new DataPoints[]{dps, dps2}; + query_results.clear(); + query_results.add(group_bys); + + final DataPoints[] results = func.evaluate(data_query, query_results, params); + + assertEquals(2, results.length); + assertEquals(METRIC, results[0].metricName()); + assertEquals("sys.mem", results[1].metricName()); + + long ts = START_TIME; + double v = 0; + for (DataPoint dp : results[0]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = 1; + } + ts = START_TIME; + v = 0; + for (DataPoint dp : results[1]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = -1; + } + } + + @Test + public void evaluateNegativeSubQuerySeries() throws Exception { + params.add("1"); + SeekableView view2 = SeekableViewsForTest.generator(START_TIME, INTERVAL, + NUM_POINTS, true, -10, -1); + DataPoints dps2 = PowerMockito.mock(DataPoints.class); + when(dps2.iterator()).thenReturn(view2); + when(dps2.metricNameAsync()).thenReturn(Deferred.fromResult("sys.mem")); + group_bys = new DataPoints[]{dps, dps2}; + query_results.clear(); + query_results.add(group_bys); + + final DataPoints[] results = func.evaluate(data_query, query_results, params); + + assertEquals(2, results.length); + assertEquals(METRIC, results[0].metricName()); + assertEquals("sys.mem", results[1].metricName()); + + long ts = START_TIME; + long v = 0; + for (DataPoint dp : results[0]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = 1; + } + ts = START_TIME; + v = 0; + for (DataPoint dp : results[1]) { + assertEquals(ts, dp.timestamp()); + assertFalse(dp.isInteger()); + assertEquals(v, dp.doubleValue(), 0.001); + ts += INTERVAL; + v = -1; + } + } + + @Test(expected = IllegalArgumentException.class) + public void evaluateNullQuery() throws Exception { + params.add("1"); + func.evaluate(null, query_results, params); + } + + @Test + public void evaluateNullResults() throws Exception { + params.add("1"); + final DataPoints[] results = func.evaluate(data_query, null, params); + assertEquals(0, results.length); + } + + @Test + public void evaluateNullParams() throws Exception { + assertNotNull(func.evaluate(data_query, query_results, null)); + } + + @Test + public void evaluateEmptyResults() throws Exception { + params.add("1"); + final DataPoints[] results = func.evaluate(data_query, + Collections.emptyList(), params); + assertEquals(0, results.length); + } + + @Test + public void evaluateEmptyParams() throws Exception { + assertNotNull(func.evaluate(data_query, query_results, null)); + } + + @Test + public void writeStringField() throws Exception { + params.add("1"); + assertEquals("firstDiff(inner_expression)", + func.writeStringField(params, "inner_expression")); + assertEquals("firstDiff(null)", func.writeStringField(params, null)); + assertEquals("firstDiff()", func.writeStringField(params, "")); + assertEquals("firstDiff(inner_expression)", + func.writeStringField(null, "inner_expression")); + } +}