Atomic increment/add #265
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
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,135 @@ | ||
| package net.opentsdb.tsd; | ||
|
There was a problem hiding this comment. Styling and would need to support HTTP writes |
||
|
|
||
| import java.util.HashMap; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
|
|
||
| import net.opentsdb.core.TSDB; | ||
| import net.opentsdb.core.Tags; | ||
| import net.opentsdb.stats.StatsCollector; | ||
| import net.opentsdb.uid.NoSuchUniqueName; | ||
|
|
||
| import org.jboss.netty.channel.Channel; | ||
|
|
||
| import com.stumbleupon.async.Callback; | ||
| import com.stumbleupon.async.Deferred; | ||
|
|
||
| /** | ||
| * Increment/add datapoint rpc. | ||
| * | ||
| */ | ||
| public class IncDataPointRpc implements TelnetRpc { | ||
|
|
||
| private static final AtomicLong requests = new AtomicLong(); | ||
| private static final AtomicLong hbase_errors = new AtomicLong(); | ||
| private static final AtomicLong invalid_values = new AtomicLong(); | ||
| private static final AtomicLong illegal_arguments = new AtomicLong(); | ||
| private static final AtomicLong unknown_metrics = new AtomicLong(); | ||
|
|
||
| @Override | ||
| public Deferred<? extends Object> execute(final TSDB tsdb, final Channel chan, | ||
| final String[] cmd) { | ||
|
|
||
| requests.incrementAndGet(); | ||
| String errmsg = null; | ||
| try { | ||
| final class PutErrback implements Callback<Exception, Exception> { | ||
| public Exception call(final Exception arg) { | ||
| if (chan.isConnected()) { | ||
| chan.write("inc: HBase error: " + arg.getMessage() | ||
| + '\n'); | ||
| } | ||
| hbase_errors.incrementAndGet(); | ||
| return arg; | ||
| } | ||
|
|
||
| public String toString() { | ||
| return "report error to channel"; | ||
| } | ||
| } | ||
| return importDataPoint(tsdb, cmd).addErrback(new PutErrback()); | ||
| } catch (NumberFormatException x) { | ||
| errmsg = "inc: invalid value: " + x.getMessage() + '\n'; | ||
| invalid_values.incrementAndGet(); | ||
| } catch (IllegalArgumentException x) { | ||
| errmsg = "inc: illegal argument: " + x.getMessage() + '\n'; | ||
| illegal_arguments.incrementAndGet(); | ||
| } catch (NoSuchUniqueName x) { | ||
| errmsg = "inc: unknown metric: " + x.getMessage() + '\n'; | ||
| unknown_metrics.incrementAndGet(); | ||
| } | ||
| if (errmsg != null && chan.isConnected()) { | ||
| chan.write(errmsg); | ||
| } | ||
| return Deferred.fromResult(null); | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Collects the stats and metrics tracked by this instance. | ||
| * | ||
| * @param collector | ||
| * The collector to use. | ||
| */ | ||
| public static void collectStats(final StatsCollector collector) { | ||
| collector.record("rpc.received", requests, "type=inc"); | ||
| collector.record("rpc.errors", hbase_errors, "type=hbase_errors"); | ||
| collector.record("rpc.errors", invalid_values, "type=invalid_values"); | ||
| collector.record("rpc.errors", illegal_arguments, | ||
| "type=illegal_arguments"); | ||
| collector.record("rpc.errors", unknown_metrics, "type=unknown_metrics"); | ||
| } | ||
|
|
||
| /** | ||
| * Increments a single data point. | ||
| * | ||
| * @param tsdb | ||
| * The TSDB to increment the data point into. | ||
| * @param words | ||
| * The words describing the data point to import, in the | ||
| * following format: {@code [metric, timestamp, value, ..tags..]} | ||
| * @return A deferred Long that indicates the completion of the request. | ||
| * @throws NumberFormatException | ||
| * if the timestamp or value is invalid. | ||
| * @throws IllegalArgumentException | ||
| * if any other argument is invalid. | ||
| * @throws net.opentsdb.uid.NoSuchUniqueName | ||
| * if the metric isn't registered. | ||
| */ | ||
| private Deferred<? extends Object> importDataPoint(final TSDB tsdb, | ||
| final String[] words) { | ||
|
|
||
| words[0] = null; // Ditch the "inc". | ||
|
|
||
|
|
||
| if (words.length < 5) { // Need at least: metric timestamp value tag | ||
| // ^ 5 and not 4 because words[0] is "inc". | ||
| throw new IllegalArgumentException("not enough arguments" | ||
| + " (need least 4, got " + (words.length - 1) + ')'); | ||
| } | ||
| final String metric = words[1]; | ||
| if (metric.length() <= 0) { | ||
| throw new IllegalArgumentException("empty metric name"); | ||
| } | ||
| final long timestamp = Tags.parseLong(words[2]); | ||
| if (timestamp <= 0) { | ||
| throw new IllegalArgumentException("invalid timestamp: " | ||
| + timestamp); | ||
| } | ||
| final String value = words[3]; | ||
| if (value.length() <= 0) { | ||
| throw new IllegalArgumentException("empty value"); | ||
| } | ||
| final HashMap<String, String> tags = new HashMap<String, String>(); | ||
| for (int i = 4; i < words.length; i++) { | ||
| if (!words[i].isEmpty()) { | ||
| Tags.parse(tags, words[i]); | ||
| } | ||
| } | ||
| if (value.indexOf('.') < 0) { // integer value | ||
| return tsdb.incPoint(metric, timestamp, Tags.parseLong(value), tags); | ||
| } else { // floating point value | ||
| throw new IllegalArgumentException("Submitted value for increment does not work: " + timestamp); | ||
| } | ||
| } | ||
|
|
||
| } | ||
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
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Avoid re-arranging imports if possible. Just clutters PRs.