#16 annotation timelines

.gitignore

@@ -7,3 +7,5 @@ build
 config.log
 config.status
 configure
+
+.DS_Store

Makefile.am

@@ -29,6 +29,8 @@ dist_pkgdata_SCRIPTS = src/mygnuplot.sh
 tsdb_SRC := \
 	src/core/Aggregator.java	\
 	src/core/Aggregators.java	\
+	src/core/Annotation.java	\
+	src/core/AnnotationQuery.java	\
 	src/core/CompactionQueue.java	\
 	src/core/Const.java	\
 	src/core/DataPoint.java	\
@@ -108,11 +110,13 @@ test_DEPS = \
 
 httpui_SRC := \
 	src/tsd/client/DateTimeBox.java	\
+	src/tsd/client/AnnotationsForm.java \
 	src/tsd/client/EventsHandler.java	\
 	src/tsd/client/GotJsonCallback.java	\
 	src/tsd/client/MetricForm.java	\
 	src/tsd/client/QueryUi.java	\
 	src/tsd/client/RemoteOracle.java	\
+	src/tsd/client/TagsPanel.java	\
 	src/tsd/client/ValidatedTextBox.java
 
 httpui_DEPS = src/tsd/QueryUi.gwt.xml

src/core/Annotation.java

@@ -0,0 +1,23 @@
+package net.opentsdb.core;
+
+/**
+ * Represents a single annotation value to mark meta data in a graph at a
+ * certain time.
+ */
+public class Annotation {
+    private long timestamp;
+    private byte[] value;
+
+    public Annotation(long timestamp, byte[] value) {
+        this.timestamp = timestamp;
+        this.value = value;
+    }
+
+    public long getTimestamp() {
+        return timestamp;
+    }
+
+    public byte[] getValue() {
+        return value;
+    }
+}

src/core/AnnotationQuery.java

@@ -0,0 +1,168 @@
+package net.opentsdb.core;
+
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import net.opentsdb.uid.NoSuchUniqueName;
+
+import org.hbase.async.Bytes;
+import org.hbase.async.HBaseException;
+import org.hbase.async.KeyValue;
+import org.hbase.async.Scanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A query to retrieve annotation data from the TSDB.
+ */
+public class AnnotationQuery {
+  private static final Logger LOGGER = LoggerFactory
+      .getLogger(AnnotationQuery.class);
+
+  private TSDB tsdb;
+  private byte[] metric;
+  private long startTime;
+  private long endTime;
+  private List<byte[]> tags;
+
+  public AnnotationQuery(TSDB tsdb, long startTime, long endTime,
+      Map<String, String> tags) {
+    this.tsdb = tsdb;
+    this.startTime = startTime;
+    this.endTime = endTime;
+    this.tags = Tags.resolveAll(tsdb, tags);
+    try {
+      this.metric = tsdb.metrics.getId(Const.ANNOTATION_NAME);
+    } catch (NoSuchUniqueName e) {
+      LOGGER.debug("not yet stored a timeline annotation", e);
+    }
+  }
+
+  public List<Annotation> run() {
+    final List<Annotation> result = new ArrayList<Annotation>();
+
+    if (metric != null) {
+      final Scanner scanner = getScanner();
+      final short metricWidth = tsdb.metrics.width();
+      ArrayList<ArrayList<KeyValue>> rows;
+
+      try {
+        while ((rows = scanner.nextRows().joinUninterruptibly()) != null) {
+          for (final ArrayList<KeyValue> row : rows) {
+            final byte[] key = row.get(0).key();
+            if (Bytes.memcmp(metric, key, 0, metricWidth) != 0) {
+              throw new IllegalDataException(
+                  "HBase returned a row that doesn't match" + " our scanner ("
+                      + scanner + ")! " + row + " does not start" + " with "
+                      + Arrays.toString(metric));
+            }
+            for (KeyValue keyValue : row) {
+              result.add(new Annotation(getTimestamp(keyValue.key(),
+                  keyValue.qualifier()), keyValue.value()));
+            }
+          }
+        }
+      } catch (Exception e) {
+        throw new RuntimeException("Should never be here", e);
+      }
+    }
+
+    return result;
+  }
+
+  /**
+   * Creates the {@link Scanner} to use for this query.
+   */
+  private Scanner getScanner() throws HBaseException {
+    final short metricWidth = tsdb.metrics.width();
+    final byte[] startRow = new byte[metricWidth + Const.TIMESTAMP_BYTES];
+    final byte[] endRow = new byte[metricWidth + Const.TIMESTAMP_BYTES];
+
+    Bytes.setInt(startRow, (int) startTime, metricWidth);
+    Bytes.setInt(endRow, (int) endTime, metricWidth);
+    System.arraycopy(metric, 0, startRow, 0, metricWidth);
+    System.arraycopy(metric, 0, endRow, 0, metricWidth);
+
+    final Scanner scanner = tsdb.client.newScanner(tsdb.table);
+    scanner.setStartKey(startRow);
+    scanner.setStopKey(endRow);
+    if (tags.size() > 0) {
+      createAndSetFilter(scanner);
+    }
+    scanner.setFamily(TSDB.FAMILY);
+
+    return scanner;
+  }
+
+  /**
+   * Sets the server-side regexp filter on the scanner. In order to find the
+   * rows with the relevant tags, we use a server-side filter that matches a
+   * regular expression on the row key.
+   * 
+   * @param scanner
+   *          The scanner on which to add the filter.
+   */
+  void createAndSetFilter(final Scanner scanner) {
+    final short nameWidth = tsdb.tag_names.width();
+    final short valueWidth = tsdb.tag_values.width();
+    final short tagsize = (short) (nameWidth + valueWidth);
+    // Generate a regexp for our tags. Say we have 2 tags: { 0 0 1 0 0 2 }
+    // and { 4 5 6 9 8 7 }, the regexp will be:
+    // "^.{7}(?:.{6})*\\Q\000\000\001\000\000\002\\E(?:.{6})*\\Q\004\005\006\011\010\007\\E(?:.{6})*$"
+    final StringBuilder buf = new StringBuilder(15 // "^.{N}" + "(?:.{M})*"
+                                                   // + "$"
+        + ((13 + tagsize) // "(?:.{M})*\\Q" + tagsize bytes + "\\E"
+        * tags.size()));
+    // In order to avoid re-allocations, reserve a bit more w/ groups ^^^
+
+    // Alright, let's build this regexp. From the beginning...
+    buf.append("(?s)" // Ensure we use the DOTALL flag.
+        + "^.{")
+    // ... start by skipping the metric ID and timestamp.
+        .append(tsdb.metrics.width() + Const.TIMESTAMP_BYTES).append("}");
+    final Iterator<byte[]> tags = this.tags.iterator();
+    byte[] tag = tags.hasNext() ? tags.next() : null;
+    // Tags and group_bys are already sorted. We need to put them in the
+    // regexp in order by ID, which means we just merge two sorted lists.
+    do {
+      // Skip any number of tags.
+      buf.append("(?:.{").append(tagsize).append("})*\\Q");
+      addId(buf, tag);
+      tag = tags.hasNext() ? tags.next() : null;
+    } while (tag != null);
+    // Skip any number of tags before the end.
+    buf.append("(?:.{").append(tagsize).append("})*$");
+    scanner.setKeyRegexp(buf.toString(), Charset.forName("ISO-8859-1"));
+  }
+
+  /**
+   * Appends the given ID to the given buffer, followed by "\\E".
+   */
+  private void addId(final StringBuilder buf, final byte[] id) {
+    boolean backslash = false;
+    for (final byte b : id) {
+      buf.append((char) (b & 0xFF));
+      if (b == 'E' && backslash) { // If we saw a `\' and now we have a
+                                   // `E'.
+        // So we just terminated the quoted section because we just
+        // added \E
+        // to `buf'. So let's put a litteral \E now and start quoting
+        // again.
+        buf.append("\\\\E\\Q");
+      } else {
+        backslash = b == '\\';
+      }
+    }
+    buf.append("\\E");
+  }
+
+  private long getTimestamp(byte[] key, byte[] qualifier) {
+    final int baseTime = Bytes.getInt(key, tsdb.metrics.width());
+    final short delta = (short) ((Bytes.getShort(qualifier) & 0xFFFF) >>> Const.FLAG_BITS);
+    return baseTime + delta;
+  }
+}

src/core/Const.java

@@ -15,6 +15,11 @@
 /** Constants used in various places.  */
 public final class Const {
 
+  /**
+   * Name of the annotation "metric"
+   */
+  public static final String ANNOTATION_NAME = "timelineannotation";
+
   /** Number of bytes on which a timestamp is encoded.  */
   public static final short TIMESTAMP_BYTES = 4;
 
@@ -31,6 +36,11 @@ public final class Const {
    */
   static final short FLAG_FLOAT = 0x8;
 
+  /**
+   * When this bit is set, the value is an annotation value.
+   */
+  static final short FLAG_ANNOTATION = 0x1;
+
   /** Mask to select the size of a value from the qualifier.  */
   static final short LENGTH_MASK = 0x7;
 

src/core/Internal.java

@@ -19,9 +19,6 @@
 import org.hbase.async.KeyValue;
 import org.hbase.async.Scanner;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 /**
  * <strong>This class is not part of the public API.</strong>
  * <p><pre>

src/core/TSDB.java

@@ -257,6 +257,35 @@ public Deferred<Object> addPoint(final String metric,
                             tags, flags);
   }
 
+  /**
+   * Adds a single annotation in the TSDB.
+   * @param timestamp The timestamp associated with the value.
+   * @param value The value of the data point.
+   * @param tags The tags on this series.  This map must be non-empty.
+   * @return A deferred object that indicates the completion of the request.
+   * The {@link Object} has not special meaning and can be {@code null} (think
+   * of it as {@code Deferred<Void>}). But you probably want to attach at
+   * least an errback to this {@code Deferred} to handle failures.
+   * @throws IllegalArgumentException if the timestamp is less than or equal
+   * to the previous timestamp added or 0 for the first timestamp, or if the
+   * difference with the previous timestamp is too large.
+   * @throws IllegalArgumentException if the value is empty.
+   * @throws IllegalArgumentException if the tags list is empty or one of the
+   * elements contains illegal characters.
+   * @throws HBaseException (deferred) if there was a problem while persisting
+   * data.
+   */
+  public Deferred<Object> addAnnotation(final long timestamp,
+      final String value, final Map<String, String> tags) {
+    String metric = Const.ANNOTATION_NAME;
+    if (value == null || value.length() == 0) {
+      throw new IllegalArgumentException("annotation value is empty: " + value
+          + " for metric=" + metric + " timestamp=" + timestamp);
+    }
+    final short flags = Const.FLAG_ANNOTATION;
+    return addPointInternal(metric, timestamp, value.getBytes(), tags, flags);
+  }
+
   private Deferred<Object> addPointInternal(final String metric,
                                             final long timestamp,
                                             final byte[] value,
@@ -395,5 +424,4 @@ final Deferred<Object> put(final byte[] key,
   final Deferred<Object> delete(final byte[] key, final byte[][] qualifiers) {
     return client.delete(new DeleteRequest(table, key, FAMILY, qualifiers));
   }
-
 }

src/create_table.sh

@@ -12,6 +12,7 @@ test -d "$HBASE_HOME" || {
 
 TSDB_TABLE=${TSDB_TABLE-'tsdb'}
 UID_TABLE=${UID_TABLE-'tsdb-uid'}
+ANNOTATION_TABLE=${ANNOTATION_TABLE-'tsdb-annotation'}
 BLOOMFILTER=${BLOOMFILTER-'ROW'}
 # LZO requires lzo2 64bit to be installed + the hadoop-gpl-compression jar.
 COMPRESSION=${COMPRESSION-'LZO'}
@@ -28,4 +29,7 @@ create '$UID_TABLE',
 
 create '$TSDB_TABLE',
   {NAME => 't', VERSIONS => 1, COMPRESSION => '$COMPRESSION', BLOOMFILTER => '$BLOOMFILTER'}
+
+create '$ANNOTATION_TABLE',
+  {NAME => 'a', COMPRESSION => '$COMPRESSION', BLOOMFILTER => '$BLOOMFILTER'}
 EOF

src/graph/Plot.java

@@ -15,15 +15,17 @@
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.TimeZone;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import net.opentsdb.core.Annotation;
 import net.opentsdb.core.DataPoint;
 import net.opentsdb.core.DataPoints;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * Produces files to generate graphs with Gnuplot.
  * <p>
@@ -43,6 +45,8 @@ public final class Plot {
   /** All the DataPoints we want to plot. */
   private ArrayList<DataPoints> datapoints =
     new ArrayList<DataPoints>();
+
+  private List<Annotation> annotations = new ArrayList<Annotation>();
 
   /** Per-DataPoints Gnuplot options. */
   private ArrayList<String> options = new ArrayList<String>();
@@ -143,6 +147,12 @@ public Iterable<DataPoints> getDataPoints() {
     return datapoints;
   }
 
+  public void setAnnotations(final List<Annotation> annotations) {
+    if (annotations != null) {
+      this.annotations = annotations;
+    }
+  }
+
   /**
    * Generates the Gnuplot script and data files.
    * @param basepath The base path to use.  A number of new files will be
@@ -258,6 +268,14 @@ private void writeGnuplotScript(final String basepath,
           break;
         }
       }
+
+      for(Annotation annotation : annotations) {
+        String ts = Long.toString(annotation.getTimestamp() + utc_offset);
+        String value = new String(annotation.getValue());
+        gp.append("set arrow from \"").append(ts).append("\", graph 0 to \"").append(ts).append("\", graph 1 nohead ls 3\n");
+        gp.append("set object rectangle at \"").append(ts).append("\", graph 0 size char (strlen(\"").append(value).append("\") + 3), char 1 front fc rgbcolor \"white\"\n");
+        gp.append("set label \"").append(value).append("\" at \"").append(ts).append("\", graph 0 front center\n");
+      }
 
       gp.write("plot ");
       for (int i = 0; i < nseries; i++) {

src/tools/CliQuery.java

@@ -103,7 +103,7 @@ public static void main(String[] args) throws IOException {
     }
 
     final HBaseClient client = CliOptions.clientFromOptions(argp);
-    final TSDB tsdb = new TSDB(client, argp.get("--table", "tsdb"),
+    final TSDB tsdb = new TSDB(client, argp.get("--table", "tsdb"), 
                                argp.get("--uidtable", "tsdb-uid"));
     final String basepath = argp.get("--graph");
     argp = null;