-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add support for splitting rollup queries (#1853)
* Add an SLA config flag for rollup intervals Adds a configuration option for rollup intervals to specify their maximum acceptable delay. Queries that cover a time between now and that maximum delay will need to query other tables for that time interval. * Add global config flag to enable splitting queries Adds a global config flag to enable splitting queries that would hit the rollup table, but the rollup table has a delay SLA configured. In that case, this feature allows splitting a query into to; one that gets the data from the rollups table until the time where it's guaranteed to be available, and the rest from the raw table. * Add a new SplitRollupQuery Adds a SplitRollupQuery class that suports splitting a rollup query into two separate queries. This is useful for when a rollup table is filled by e.g. a batch job that processes the data from the previous day on a daily basis. Rollup data for yesterday will then only be available some time today. This delay SLA can be configured on a per-table basis. The delay would specify by how much time the table can be behind real time. If a query comes in that would query data from that blackout period where data is only available in the raw table, but not yet guaranteed to be in the rollup table, the incoming query can be split into two using the SplitRollupQuery class. It wraps a query that queries the rollup table until the last guaranteed to be available timestamp based on the SLA; and one that gets the remaining data from the raw table. * Extract an AbstractQuery Extracts an AbstractQuery from the TsdbQuery implementation since we'd like to reuse some parts of it in other Query classes (in this case SplitRollupQuery) * Extract an AbstractSpanGroup * Avoid NullPointerException when setting start time Avoids a NullPointerException that happened when we were trying to set the start time on a query that would be eligible to split, but due to the SLA config only hit the raw table anyway. * Scale timestamps to milliseconds for split queries Scales all timestamps for split queries to milliseconds. It's important to maintain consistent units between all the partial queries that make up the bigger one. * Fix starting time error for split queries Fixes a bug that would happen when the start time of a query aligns perfectly with the time configured in the SLA for the delay of a rollup table. For a defined SLA, e.g. 1 day, if the start time of the query was exactly 1 day ago, the end time of the rollups part of the query would be updated and then be equal to its start time. That isn't allowed and causes a query exception.
- Loading branch information
Showing
31 changed files
with
2,508 additions
and
139 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| // This file is part of OpenTSDB. | ||
| // Copyright (C) 2010-2012 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 <http://www.gnu.org/licenses/>. | ||
| package net.opentsdb.core; | ||
|
|
||
| import org.hbase.async.HBaseException; | ||
|
|
||
| public abstract class AbstractQuery implements Query { | ||
| /** | ||
| * Runs this query. | ||
| * | ||
| * @return The data points matched by this query. | ||
| * <p> | ||
| * Each element in the non-{@code null} but possibly empty array returned | ||
| * corresponds to one time series for which some data points have been | ||
| * matched by the query. | ||
| * @throws HBaseException if there was a problem communicating with HBase to | ||
| * perform the search. | ||
| */ | ||
| @Override | ||
| public DataPoints[] run() throws HBaseException { | ||
| try { | ||
| return runAsync().joinUninterruptibly(); | ||
| } catch (RuntimeException e) { | ||
| throw e; | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("Should never be here", e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Runs this query. | ||
| * | ||
| * @return The data points matched by this query and applied with percentile calculation | ||
| * <p> | ||
| * Each element in the non-{@code null} but possibly empty array returned | ||
| * corresponds to one time series for which some data points have been | ||
| * matched by the query. | ||
| * @throws HBaseException if there was a problem communicating with HBase to | ||
| * perform the search. | ||
| * @throws IllegalStateException if the query is not a histogram query | ||
| */ | ||
| @Override | ||
| public DataPoints[] runHistogram() throws HBaseException { | ||
| if (!isHistogramQuery()) { | ||
| throw new RuntimeException("Should never be here"); | ||
| } | ||
|
|
||
| try { | ||
| return runHistogramAsync().joinUninterruptibly(); | ||
| } catch (RuntimeException e) { | ||
| throw e; | ||
| } catch (Exception e) { | ||
| throw new RuntimeException("Should never be here", e); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| // This file is part of OpenTSDB. | ||
| // Copyright (C) 2010-2012 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 <http://www.gnu.org/licenses/>. | ||
| package net.opentsdb.core; | ||
|
|
||
| import java.util.*; | ||
|
|
||
| import org.hbase.async.Bytes; | ||
| import org.hbase.async.Bytes.ByteMap; | ||
|
|
||
| import com.stumbleupon.async.Callback; | ||
| import com.stumbleupon.async.Deferred; | ||
|
|
||
| import net.opentsdb.meta.Annotation; | ||
| import net.opentsdb.rollup.RollupQuery; | ||
|
|
||
| /** | ||
| * Groups multiple spans together and offers a dynamic "view" on them. | ||
| * <p> | ||
| * This is used for queries to the TSDB, where we might group multiple | ||
| * {@link Span}s that are for the same time series but different tags | ||
| * together. We need to "hide" data points that are outside of the | ||
| * time period of the query and do on-the-fly aggregation of the data | ||
| * points coming from the different Spans, using an {@link Aggregator}. | ||
| * Since not all the Spans will have their data points at exactly the | ||
| * same time, we also do on-the-fly linear interpolation. If needed, | ||
| * this view can also return the rate of change instead of the actual | ||
| * data points. | ||
| * <p> | ||
| * This is one of the rare (if not the only) implementations of | ||
| * {@link DataPoints} for which {@link #getTags} can potentially return | ||
| * an empty map. | ||
| * <p> | ||
| * The implementation can also dynamically downsample the data when a | ||
| * sampling interval a downsampling function (in the form of an | ||
| * {@link Aggregator}) are given. This is done by using a special | ||
| * iterator when using the {@link Span.DownsamplingIterator}. | ||
| */ | ||
| abstract class AbstractSpanGroup implements DataPoints { | ||
| /** | ||
| * Finds the {@code i}th data point of this group in {@code O(n)}. | ||
| * Where {@code n} is the number of data points in this group. | ||
| */ | ||
| protected DataPoint getDataPoint(int i) { | ||
| if (i < 0) { | ||
| throw new IndexOutOfBoundsException("negative index: " + i); | ||
| } | ||
| final int saved_i = i; | ||
| final SeekableView it = iterator(); | ||
| DataPoint dp = null; | ||
| while (it.hasNext() && i >= 0) { | ||
| dp = it.next(); | ||
| i--; | ||
| } | ||
| if (i != -1 || dp == null) { | ||
| throw new IndexOutOfBoundsException("index " + saved_i | ||
| + " too large (it's >= " + size() + ") for " + this); | ||
| } | ||
| return dp; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| // This file is part of OpenTSDB. | ||
| // Copyright (C) 2010-2012 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 <http://www.gnu.org/licenses/>. | ||
| package net.opentsdb.core; | ||
|
|
||
| import com.stumbleupon.async.Callback; | ||
|
|
||
| import java.util.ArrayList; | ||
|
|
||
| class GroupCallback implements Callback<Object, ArrayList<Object>> { | ||
| /** | ||
| * We're only waiting for all callbacks to complete, ignoring their return values. | ||
| * | ||
| * @param ignored The return values of the individual callbacks - ignored | ||
| * @return null | ||
| */ | ||
| @Override | ||
| public Object call(ArrayList<Object> ignored) { | ||
| return null; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| // 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 <http://www.gnu.org/licenses/>. | ||
| package net.opentsdb.core; | ||
|
|
||
| import java.util.List; | ||
| import java.util.NoSuchElementException; | ||
|
|
||
| public class SeekableViewChain implements SeekableView { | ||
|
|
||
| private final List<SeekableView> iterators; | ||
| private int currentIterator; | ||
|
|
||
| SeekableViewChain(List<SeekableView> iterators) { | ||
| this.iterators = iterators; | ||
| } | ||
|
|
||
| /** | ||
| * Returns {@code true} if this view has more elements. | ||
| */ | ||
| @Override | ||
| public boolean hasNext() { | ||
| SeekableView iterator = getCurrentIterator(); | ||
| return iterator != null && iterator.hasNext(); | ||
| } | ||
|
|
||
| /** | ||
| * Returns a <em>view</em> on the next data point. | ||
| * No new object gets created, the referenced returned is always the same | ||
| * and must not be stored since its internal data structure will change the | ||
| * next time {@code next()} is called. | ||
| * | ||
| * @throws NoSuchElementException if there were no more elements to iterate | ||
| * on (in which case {@link #hasNext} would have returned {@code false}. | ||
| */ | ||
| @Override | ||
| public DataPoint next() { | ||
| SeekableView iterator = getCurrentIterator(); | ||
| if (iterator == null || !iterator.hasNext()) { | ||
| throw new NoSuchElementException("No elements left in iterator"); | ||
| } | ||
|
|
||
| DataPoint next = iterator.next(); | ||
|
|
||
| if (!iterator.hasNext()) { | ||
| currentIterator++; | ||
| } | ||
|
|
||
| return next; | ||
| } | ||
|
|
||
| /** | ||
| * Unsupported operation. | ||
| * | ||
| * @throws UnsupportedOperationException always. | ||
| */ | ||
| @Override | ||
| public void remove() { | ||
| throw new UnsupportedOperationException("Removing items is not supported"); | ||
| } | ||
|
|
||
| /** | ||
| * Advances the iterator to the given point in time. | ||
| * <p> | ||
| * This allows the iterator to skip all the data points that are strictly | ||
| * before the given timestamp. | ||
| * | ||
| * @param timestamp A strictly positive 32 bit UNIX timestamp (in seconds). | ||
| * @throws IllegalArgumentException if the timestamp is zero, or negative, | ||
| * or doesn't fit on 32 bits (think "unsigned int" -- yay Java!). | ||
| */ | ||
| @Override | ||
| public void seek(long timestamp) { | ||
| for (final SeekableView it : iterators) { | ||
| it.seek(timestamp); | ||
| } | ||
| } | ||
|
|
||
| private SeekableView getCurrentIterator() { | ||
| while (currentIterator < iterators.size()) { | ||
| if (iterators.get(currentIterator).hasNext()) { | ||
| return iterators.get(currentIterator); | ||
| } | ||
| currentIterator++; | ||
| } | ||
| return null; | ||
| } | ||
| } |
Oops, something went wrong.