Class PutDataPointRpc

src/tsd/PutDataPointRpc.java:73–891

This class handles the parsing and writing of data points over any of the implemented RPCs. Each put method should track the deferred of the tsdb.addPoint() method so that we know whether or not the write was actually successful. Note that if HBase is backed up, then the arguments will hang

 * data points such as Rollups or Pre-Aggregates
 */
class PutDataPointRpc implements TelnetRpc, HttpRpc {
  protected static final Logger LOG = LoggerFactory.getLogger(PutDataPointRpc.class);
  protected static final ArrayList<Boolean> EMPTY_DEFERREDS = 
      new ArrayList<Boolean>(0);
  protected static final AtomicLong telnet_requests = new AtomicLong();
  protected static final AtomicLong telnet_requests_unauthorized = new AtomicLong();
  protected static final AtomicLong telnet_requests_forbidden = new AtomicLong();
  protected static final AtomicLong http_requests = new AtomicLong();
  protected static final AtomicLong http_requests_unauthorized = new AtomicLong();
  protected static final AtomicLong http_requests_forbidden = new AtomicLong();
  protected static final AtomicLong raw_dps = new AtomicLong();
  protected static final AtomicLong raw_histograms = new AtomicLong();
  protected static final AtomicLong rollup_dps = new AtomicLong();
  protected static final AtomicLong raw_stored = new AtomicLong();
  protected static final AtomicLong raw_histograms_stored = new AtomicLong();
  protected static final AtomicLong rollup_stored = new AtomicLong();
  protected static final AtomicLong hbase_errors = new AtomicLong();
  protected static final AtomicLong unknown_errors = new AtomicLong();
  protected static final AtomicLong invalid_values = new AtomicLong();
  protected static final AtomicLong illegal_arguments = new AtomicLong();
  protected static final AtomicLong unknown_metrics = new AtomicLong();
  protected static final AtomicLong inflight_exceeded = new AtomicLong();
  protected static final AtomicLong writes_blocked = new AtomicLong();
  protected static final AtomicLong writes_timedout = new AtomicLong();
  protected static final AtomicLong requests_timedout = new AtomicLong();
  
  /** Whether or not to send error messages back over telnet */
  private final boolean send_telnet_errors;
  
  /** The type of data point we're writing.
   * @since 2.4 */
  public enum DataPointType {
    PUT("put"),
    ROLLUP("rollup"),
    HISTOGRAM("histogram");
    
    private final String name;
    DataPointType(final String name) {
      this.name = name;
    }
    
    @Override
    public String toString() {
      return name;
    }
  }
  
  /**
   * Default Ctor
   * @param config The TSDB config to pull from
   */
  public PutDataPointRpc(final Config config) {
    send_telnet_errors = config.getBoolean("tsd.rpc.telnet.return_errors");
  }
  
  @Override
  public Deferred<Object> execute(final TSDB tsdb, final Channel chan,
                                  final String[] cmd) {