content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.hadoop.chukwa.datastore;
  
  import java.io.IOException;
  import java.net.URI;
  import java.net.URISyntaxException;
  import java.nio.ByteBuffer;
  import java.nio.charset.Charset;
  import java.util.ArrayList;
  import java.util.Calendar;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Set;
  import java.util.TimeZone;
  import java.util.UUID;
  import java.util.concurrent.CopyOnWriteArraySet;
  import java.util.concurrent.TimeUnit;
  
  import org.apache.hadoop.chukwa.hicc.bean.Chart;
  import org.apache.hadoop.chukwa.hicc.bean.Dashboard;
  import org.apache.hadoop.chukwa.hicc.bean.HeatMapPoint;
  import org.apache.hadoop.chukwa.hicc.bean.Heatmap;
  import org.apache.hadoop.chukwa.hicc.bean.LineOptions;
  import org.apache.hadoop.chukwa.hicc.bean.Series;
  import org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData;
  import org.apache.hadoop.chukwa.hicc.bean.Widget;
  import org.apache.hadoop.chukwa.hicc.rest.Examples;
  import org.apache.hadoop.chukwa.util.ExceptionUtil;
  import org.apache.hadoop.chukwa.util.HBaseUtil;
  import org.apache.hadoop.hbase.Cell;
  import org.apache.hadoop.hbase.CellUtil;
  import org.apache.hadoop.hbase.TableName;
  import org.apache.hadoop.hbase.client.Connection;
  import org.apache.hadoop.hbase.client.ConnectionFactory;
  import org.apache.hadoop.hbase.client.Delete;
  import org.apache.hadoop.hbase.client.Get;
  import org.apache.hadoop.hbase.client.Put;
  import org.apache.hadoop.hbase.client.Result;
  import org.apache.hadoop.hbase.client.ResultScanner;
  import org.apache.hadoop.hbase.client.Scan;
  import org.apache.hadoop.hbase.client.Table;
  import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
  import org.apache.hadoop.hbase.filter.Filter;
  import org.apache.hadoop.hbase.util.Bytes;
  import org.apache.log4j.Logger;
  import org.json.simple.JSONObject;
  import org.json.simple.JSONValue;
  
  import com.google.gson.Gson;
  
  public class ChukwaHBaseStore {
    static Logger LOG = Logger.getLogger(ChukwaHBaseStore.class);
    static int MINUTES_IN_HOUR = 60;
    static double RESOLUTION = 360;
    static int MINUTE = 60000; //60 milliseconds
    final static int SECOND = (int) TimeUnit.SECONDS.toMillis(1);
    private final static Charset UTF8 = Charset.forName("UTF-8");
  
    final static byte[] COLUMN_FAMILY = "t".getBytes(UTF8);
    final static byte[] ANNOTATION_FAMILY = "a".getBytes(UTF8);
    final static byte[] KEY_NAMES = "k".getBytes(UTF8);
    final static byte[] CHART_TYPE = "chart_meta".getBytes(UTF8);
    final static byte[] CHART_FAMILY = "c".getBytes(UTF8);
    final static byte[] COMMON_FAMILY = "c".getBytes(UTF8);
    final static byte[] WIDGET_TYPE = "widget_meta".getBytes(UTF8);
    final static byte[] DASHBOARD_TYPE = "dashboard_meta".getBytes(UTF8);
    private static final String CHUKWA = "chukwa";
    private static final String CHUKWA_META = "chukwa_meta";
    private static long MILLISECONDS_IN_DAY = 86400000L;
    private static Connection connection = null;
  
    public ChukwaHBaseStore() {
      super();
    }
  
    public static synchronized void getHBaseConnection() throws IOException {
      if (connection == null || connection.isClosed()) {
        connection = ConnectionFactory.createConnection();
      }
    }
    
    public static synchronized void closeHBase() {
      try {
        if(connection != null) {
          connection.close();
        }
      } catch(IOException e) {
        LOG.warn("Unable to release HBase connection.");
      }
    }
    
    /**
     * Scan chukwa table for a particular metric group and metric name based on
     * time ranges.
     * 
     * @param metricGroup metric group name
     * @param metric metric name
     * @param source source of the metric
     * @param startTime start time
     * @param endTime end time
     * @return Series object
     */
    public static Series getSeries(String metricGroup, String metric,
        String source, long startTime, long endTime) {
      String fullMetricName = new StringBuilder(metricGroup).append(".")
          .append(metric).toString();
      return getSeries(fullMetricName, source, startTime, endTime);
    }
  
    /**
     * Scan chukwa table for a full metric name based on time ranges.
     * 
     * @param metric metric group name and metric name combined
     * @param source source of the metric
     * @param startTime start time
     * @param endTime end time
     * @return Series object
     */
    public static synchronized Series getSeries(String metric, String source, long startTime,
        long endTime) {
      String seriesName = new StringBuilder(metric).append(":").append(source).toString();
      Series series = new Series(seriesName);
      try {
        // Swap start and end if the values are inverted.
        if (startTime > endTime) {
          long temp = endTime;
          startTime = endTime;
          endTime = temp;
        }
  
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA));
        Scan scan = new Scan();
        Calendar c = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
        c.setTimeInMillis(startTime);
        int startDay = c.get(Calendar.DAY_OF_YEAR);
        c.setTimeInMillis(endTime);
        int endDay = c.get(Calendar.DAY_OF_YEAR);
        long currentDay = startTime;
        for (int i = startDay; i <= endDay; i++) {
          byte[] rowKey = HBaseUtil.buildKey(currentDay, metric, source);
  
          scan.addFamily(COLUMN_FAMILY);
          scan.setStartRow(rowKey);
          scan.setStopRow(rowKey);
          scan.setTimeRange(startTime, endTime);
          scan.setBatch(10000);
  
          ResultScanner results = table.getScanner(scan);
          Iterator<Result> it = results.iterator();
  
          while (it.hasNext()) {
            Result result = it.next();
            for (Cell kv : result.rawCells()) {
              byte[] key = CellUtil.cloneQualifier(kv);
              long timestamp = ByteBuffer.wrap(key).getLong();
              double value = Double
                  .parseDouble(new String(CellUtil.cloneValue(kv), UTF8));
              series.add(timestamp, value);
            }
          }
          results.close();
          currentDay = currentDay + (i * MILLISECONDS_IN_DAY);
        }
        table.close();
      } catch (IOException e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return series;
    }
  
    public static Set<String> getMetricNames(String metricGroup) {
      Set<String> familyNames = new CopyOnWriteArraySet<String>();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Get get = new Get(metricGroup.getBytes(UTF8));
        Result result = table.get(get);
        for (Cell kv : result.rawCells()) {
          JSONObject json = (JSONObject) JSONValue.parse(new String(CellUtil.cloneValue(kv), UTF8));
          if (json.get("type").equals("metric")) {
            familyNames.add(new String(CellUtil.cloneQualifier(kv), UTF8));
          }
        }
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return familyNames;
  
    }
  
    public static Set<String> getMetricGroups() {
      Set<String> metricGroups = new CopyOnWriteArraySet<String>();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Scan scan = new Scan();
        scan.addFamily(KEY_NAMES);
        ResultScanner rs = table.getScanner(scan);
        Iterator<Result> it = rs.iterator();
        while (it.hasNext()) {
          Result result = it.next();
          metricGroups.add(new String(result.getRow(), UTF8));
        }
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return metricGroups;
    }
  
    public static Set<String> getSourceNames(String dataType) {
      Set<String> pk = new HashSet<String>();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Scan scan = new Scan();
        scan.addFamily(KEY_NAMES);
        ResultScanner rs = table.getScanner(scan);
        Iterator<Result> it = rs.iterator();
        while (it.hasNext()) {
          Result result = it.next();
          for (Cell cell : result.rawCells()) {
            JSONObject json = (JSONObject) JSONValue.parse(new String(CellUtil.cloneValue(cell), UTF8));
            if (json!=null && json.get("type")!=null && json.get("type").equals("source")) {
              pk.add(new String(CellUtil.cloneQualifier(cell), UTF8));
            }
          }
        }
        rs.close();
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return pk;
    }
  
    public static Heatmap getHeatmap(String metricGroup, String metric,
        long startTime, long endTime, double max, double scale, int width, int height) {
      Heatmap heatmap = new Heatmap();
      Set<String> sources = getSourceNames(metricGroup);
      Set<String> metrics = getMetricNames(metricGroup);
      List<Get> series = new ArrayList<Get>();
      String fullName = new StringBuilder(metricGroup).append(".").append(metric).toString();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA));
        Calendar c = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
        c.setTimeInMillis(startTime);
        int startDay = c.get(Calendar.DAY_OF_YEAR);
        c.setTimeInMillis(endTime);
        int endDay = c.get(Calendar.DAY_OF_YEAR);
        long currentDay = startTime;
        for (int i = startDay; i <= endDay; i++) {
          for (String m : metrics) {
            if (m.startsWith(fullName)) {
              for (String source : sources) {
                byte[] rowKey = HBaseUtil.buildKey(currentDay, m, source);
                Get serie = new Get(rowKey);
                serie.addFamily(COLUMN_FAMILY);
                serie.setTimeRange(startTime, endTime);
                series.add(serie);
              }
            }
          }
          currentDay = currentDay + (i * MILLISECONDS_IN_DAY);
        }
        long timeRange = (endTime - startTime);
        Result[] rs = table.get(series);
        int index = 1;
        // Series display in y axis
        int y = 0;
        HashMap<String, Integer> keyMap = new HashMap<String, Integer>();
        for (Result result : rs) {
          for(Cell cell : result.rawCells()) {
            byte[] dest = new byte[5];
            System.arraycopy(CellUtil.cloneRow(cell), 3, dest, 0, 5);
            String source = new String(dest, UTF8);
            long time = cell.getTimestamp();
            // Time display in x axis
            long delta = time - startTime;
            double f = (double) delta / timeRange;
            f = (double) f * width;
            int x = (int) Math.round(f);
            if (keyMap.containsKey(source)) {
              y = keyMap.get(source);
            } else {
              keyMap.put(source, Integer.valueOf(index));
              y = index;
              index++;
            }
            double v = Double.parseDouble(new String(CellUtil.cloneValue(cell), UTF8));
            heatmap.put(x, y, v);
            if (v > max) {
              max = v;
            }
          }
        }
        table.close();
        int radius = height / index;
        // Usually scale max from 0 to 100 for visualization
        heatmap.putMax(scale);
        for (HeatMapPoint point : heatmap.getHeatmap()) {
          double round = point.count / max * scale;
          round = Math.round(round * 100.0) / 100.0;
          point.put(point.x, point.y * radius, round);
        }
        heatmap.putRadius(radius);
        heatmap.putSeries(index -1);
      } catch (IOException e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return heatmap;
    }
  
    /**
     * Scan chukwa table and find cluster tag from annotation column family from a
     * range of entries.
     * 
     * @param startTime start time in epoch
     * @param endTime start time in epoch
     * @return Set of cluster names
     */
    public static Set<String> getClusterNames(long startTime, long endTime) {
      Set<String> clusters = new HashSet<String>();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Scan scan = new Scan();
        scan.addFamily(KEY_NAMES);
        ResultScanner rs = table.getScanner(scan);
        Iterator<Result> it = rs.iterator();
        while (it.hasNext()) {
          Result result = it.next();
          for (Cell cell : result.rawCells()) {
            JSONObject json = (JSONObject) JSONValue.parse(new String(CellUtil.cloneValue(cell), UTF8));
            if (json.get("type").equals("cluster")) {
              clusters.add(new String(CellUtil.cloneQualifier(cell), UTF8));
            }
          }
        }
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return clusters;
    }
  
    /**
     * Get a chart from HBase by ID.
     * 
     * @param id Chart ID
     * @return Chart object
     */
    public static Chart getChart(String id) {
      Chart chart = null;
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Get get = new Get(CHART_TYPE);
        Result r = table.get(get);
        byte[] value = r.getValue(CHART_FAMILY, id.getBytes(UTF8));
        Gson gson = new Gson();
        if(value!=null) {
          chart = gson.fromJson(new String(value, UTF8), Chart.class);
        }
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return chart;
    }
  
    /**
     * Update a chart in HBase by ID.
     * 
     * @param id Chart ID
     * @param chart Chart Object
     */
    public static void putChart(String id, Chart chart) {
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Put put = new Put(CHART_TYPE);
        Gson gson = new Gson();
        String buffer = gson.toJson(chart);
        put.addColumn(CHART_FAMILY, id.getBytes(UTF8), buffer.getBytes(UTF8));
        table.put(put);
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      
    }
  
    /**
     * Create a chart in HBase by specifying parameters.
     * @param id is unique chart identifier
     * @param title is searchable name of the chart
     * @param metrics is list of metric names to render chart
     * @param source is data source name
     * @param yunitType is y axis unit type
     * @return Chart ID
     * @throws URISyntaxException if metrics name can not compose valid URL syntax
     */
    public static synchronized String createChart(String id,
        String title, String[] metrics, String source, String yunitType) throws URISyntaxException {
      Chart chart = new Chart(id);
      chart.setYUnitType(yunitType);
      chart.setTitle(title);
      ArrayList<SeriesMetaData> series = new ArrayList<SeriesMetaData>();
      for(String metric : metrics) {
        SeriesMetaData s = new SeriesMetaData();
        s.setLabel(metric + "/" + source);
        s.setUrl(new URI("/hicc/v1/metrics/series/" + metric + "/"
          + source));
        LineOptions l = new LineOptions();
        s.setLineOptions(l);
        series.add(s);
      }
      chart.setSeries(series);
      return createChart(chart);
      
    }
  
    /**
     * Create a chart in HBase by specifying parameters.
     * @param id is unique chart identifier
     * @param title is searchable name of the chart
     * @param metrics is list of metric names to render ring chart
     * @param source is data source name
     * @param suffixLabel is text label to append to metric values
     * @param direction sets the threshold to have either upper limit or lower limit
     * @return Chart ID
     * @throws URISyntaxException if metrics name can not compose valid URL syntax
     */
    public static synchronized String createCircle(String id,
        String title, String[] metrics, String source, String suffixLabel, String direction) throws URISyntaxException {
      Chart chart = new Chart(id);
      chart.setSuffixText(suffixLabel);
      chart.setTitle(title);
      chart.setThreshold(direction);
      ArrayList<SeriesMetaData> series = new ArrayList<SeriesMetaData>();
      for(String metric : metrics) {
        SeriesMetaData s = new SeriesMetaData();
        s.setLabel(metric + "/" + source);
        s.setUrl(new URI("/hicc/v1/metrics/series/" + metric + "/"
          + source));
        series.add(s);
      }
      chart.setSeries(series);
      return createChart(chart);
      
    }
  
    /**
     * Create a tile in HBase by specifying parameters.
     * @param id is unique tile identifier
     * @param title is searchable name of the tile widget
     * @param bannerText is description of the tile widget
     * @param suffixLabel is text label to append to metric values
     * @param metrics is list of metric names to render tile widget
     * @param source is data source name
     * @param icon is emoji symbol to render beside tile widget
     * @return Widget ID
     * @throws URISyntaxException if metrics name can not compose valid URL syntax
     */
    public static synchronized String createTile(String id, String title, 
        String bannerText, String suffixLabel, String[] metrics, String source, 
        String icon) throws URISyntaxException {
      Chart chart = new Chart(id);
      chart.setTitle(title);
      chart.setBannerText(bannerText);
      chart.setSuffixText(suffixLabel);
      chart.setIcon(icon);
      List<SeriesMetaData> smd = new ArrayList<SeriesMetaData>();
      for (String metric : metrics) {
        SeriesMetaData series = new SeriesMetaData();
        series.setUrl(new URI("/hicc/v1/metrics/series/" + metric + "/" + source));
        smd.add(series);
      }
      chart.setSeries(smd);
      return createChart(chart);
    }
  
    /**
     * Create a chart in HBase.
     * 
     * @param chart is a chukwa Chart object
     * @return id of newly created chart
     */
    public static synchronized String createChart(Chart chart) {
      String id = chart.getId();
      try {
        getHBaseConnection();
        if (id != null) {
          // Check if there is existing chart with same id.
          Chart test = getChart(id);
          if (test != null) {
            // If id already exists, randomly generate an id.
            id = String.valueOf(UUID.randomUUID());
          }
        } else {
          // If id is not provided, randomly generate an id.
          id = String.valueOf(UUID.randomUUID());
        }
        chart.setId(id);
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Put put = new Put(CHART_TYPE);
        Gson gson = new Gson();
        String buffer = gson.toJson(chart);
        put.addColumn(CHART_FAMILY, id.getBytes(UTF8), buffer.getBytes(UTF8));
        table.put(put);
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
        id = null;
      }
      return id;
    }
  
    /**
     * Return data for multiple series of metrics stored in HBase.
     * 
     * @param series is SeriesMetaData object
     * @param startTime sets the start time of metrics
     * @param endTime sets the end time of metrics
     * @return A list of Series meta data
     */
    public static synchronized ArrayList<org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData> getChartSeries(ArrayList<org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData> series, long startTime, long endTime) {
      ArrayList<org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData> list = new ArrayList<org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData>();
      try {
        // Swap start and end if the values are inverted.
        if (startTime > endTime) {
          long temp = endTime;
          startTime = endTime;
          endTime = temp;
        }
        // Figure out the time range and determine the best resolution
        // to fetch the data
        long range = (endTime - startTime)
          / (long) (MINUTES_IN_HOUR * MINUTE);
        long sampleRate = 1;
        if (range <= 1) {
          sampleRate = 5;
        } else if(range <= 24) {
          sampleRate = 240;
        } else if (range <= 720) {
          sampleRate = 7200;
        } else if(range >= 720) {
          sampleRate = 87600;
        }
        double smoothing = (endTime - startTime)
            / (double) (sampleRate * SECOND ) / (double) RESOLUTION;
  
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA));
        Scan scan = new Scan();
        Calendar c = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
        c.setTimeInMillis(startTime);
        int startDay = c.get(Calendar.DAY_OF_YEAR);
        c.setTimeInMillis(endTime);
        int endDay = c.get(Calendar.DAY_OF_YEAR);
        for (org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData s : series) {
          org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData clone = (org.apache.hadoop.chukwa.hicc.bean.SeriesMetaData) s.clone();
          long currentDay = startTime;
          String[] parts = s.getUrl().toString().split("/");
          String metric = parts[5];
          String source = parts[6];
          ArrayList<ArrayList<Number>> data = new ArrayList<ArrayList<Number>>();
          for (int i = startDay; i <= endDay; i++) {
            byte[] rowKey = HBaseUtil.buildKey(currentDay, metric, source);
            scan.addFamily(COLUMN_FAMILY);
            scan.setStartRow(rowKey);
            scan.setStopRow(rowKey);
            scan.setTimeRange(startTime, endTime);
            scan.setBatch(10000);
  
            ResultScanner results = table.getScanner(scan);
            Iterator<Result> it = results.iterator();
            double filteredValue = 0.0d;
            long lastTime = startTime;
            long totalElapsedTime = 0;
            int initial = 0;
            
            while (it.hasNext()) {
              Result result = it.next();
              for (Cell kv : result.rawCells()) {
                byte[] key = CellUtil.cloneQualifier(kv);
                long timestamp = ByteBuffer.wrap(key).getLong();
                double value = Double.parseDouble(new String(CellUtil.cloneValue(kv),
                    UTF8));
                if(initial==0) {
                  filteredValue = value;
                }
                long elapsedTime = (timestamp - lastTime) / SECOND;
                lastTime = timestamp;
                // Determine if there is any gap, if there is gap in data, reset
                // calculation.
                if (elapsedTime > (sampleRate * 5)) {
                  filteredValue = 0.0d;
                } else {
                  if (smoothing != 0.0d) {
                    // Apply low pass filter to calculate
                    filteredValue = filteredValue + (double) ((double) elapsedTime * (double) ((double) (value - filteredValue) / smoothing));
                  } else {
                    // Use original value
                    filteredValue = value;
                  }
                }
                totalElapsedTime = totalElapsedTime + elapsedTime;
                if (totalElapsedTime >= sampleRate) {
                  ArrayList<Number> points = new ArrayList<Number>();
                  points.add(timestamp);
                  points.add(filteredValue);
                  data.add(points);
                  totalElapsedTime = 0;
                }
              }
              initial++;
            }
            results.close();
            currentDay = currentDay + (i * MILLISECONDS_IN_DAY);
          }
          clone.setData(data);
          list.add(clone);
        }
        table.close();
      } catch (IOException|CloneNotSupportedException e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return list;
    }
  
    /**
     * List widgets stored in HBase.
     * 
     * @param limit sets the number of widgets to return
     * @param offset sets the starting point to return widgets
     * @return List of Widgets
     */
    public static synchronized List<Widget> listWidget(int limit, int offset) {
      ArrayList<Widget> list = new ArrayList<Widget>();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Scan scan = new Scan();
        scan.setStartRow(WIDGET_TYPE);
        scan.setStopRow(WIDGET_TYPE);
        ResultScanner rs = table.getScanner(scan);
        Iterator<Result> it = rs.iterator();
        int c = 0;
        while(it.hasNext()) {
          Result result = it.next();
          for(Cell kv : result.rawCells()) {
            if(c > limit) {
              break;
            }
            if(c < offset) {
              continue;
            }
            Gson gson = new Gson();
            Widget widget = gson.fromJson(new String(CellUtil.cloneValue(kv), UTF8), Widget.class);
            list.add(widget);
            c++;
          }
        }
        rs.close();
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return list;
    }
  
    public static synchronized List<String> getData(ArrayList<SeriesMetaData> series, long startTime, long endTime) {
      ArrayList<String> data = new ArrayList<String>();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA));
        Scan scan = new Scan();
        for(SeriesMetaData s : series) {
          String snapshot = "";
          String[] parts = s.getUrl().toString().split("/");
          String metric = parts[5];
          String source = parts[6];
          long currentDay = startTime;
          byte[] rowKey = HBaseUtil.buildKey(currentDay, metric, source);
          scan.addFamily(COLUMN_FAMILY);
          scan.setStartRow(rowKey);
          scan.setStopRow(rowKey);
          scan.setTimeRange(startTime, endTime);
          scan.setBatch(10000);
          ResultScanner rs = table.getScanner(scan);
          Iterator<Result> it = rs.iterator();
          while(it.hasNext()) {
            Result result = it.next();
            for(Cell kv : result.rawCells()) {
              snapshot = new String(CellUtil.cloneValue(kv));
              if(snapshot.matches("-?\\d+(\\.\\d+)?")) {
                int endOffset = snapshot.length();
                if(snapshot.length() - snapshot.indexOf(".") > 2) {
                  endOffset = snapshot.indexOf(".") + 2;
                }
                snapshot = snapshot.substring(0, endOffset);
              }
            }
          }
          data.add(snapshot);
          rs.close();
        }
        table.close();
      } catch(Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return data;
    }
  
    /**
     * Find widget by title prefix in HBase.
     * 
     * @param query is prefix query of widget title.
     * @return List of Widgets
     */
    public static synchronized List<Widget> searchWidget(String query) {
      ArrayList<Widget> list = new ArrayList<Widget>();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Filter filter = new ColumnPrefixFilter(Bytes.toBytes(query));
        Scan scan = new Scan();
        scan.setStartRow(WIDGET_TYPE);
        scan.setStopRow(WIDGET_TYPE);
        scan.setFilter(filter);
        ResultScanner rs = table.getScanner(scan);
        Iterator<Result> it = rs.iterator();
        while(it.hasNext()) {
          Result result = it.next();
          for(Cell kv : result.rawCells()) {
            Gson gson = new Gson();
            Widget widget = gson.fromJson(new String(CellUtil.cloneValue(kv), UTF8), Widget.class);
            list.add(widget);
          }
        }
        rs.close();
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return list;
    }
  
    /**
     * View a widget information in HBase.
     * 
     * @param title is title of the widget.
     * @return List of Widgets
     */
    public static synchronized Widget viewWidget(String title) {
      Widget w = null;
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Get widget = new Get(WIDGET_TYPE);
        widget.addColumn(COMMON_FAMILY, title.getBytes(UTF8));
        Result rs = table.get(widget);
        byte[] buffer = rs.getValue(COMMON_FAMILY, title.getBytes(UTF8));
        Gson gson = new Gson();
        w = gson.fromJson(new String(buffer, UTF8), Widget.class);
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return w;
    }
  
    /**
     * Create a widget in HBase.
     * 
     * @param widget is chukwa Widget object
     * @return true if widget is created
     */
    public static synchronized boolean createWidget(Widget widget) {
      boolean created = false;
      try {
        widget.tokenize();
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Get widgetTest = new Get(WIDGET_TYPE);
        widgetTest.addColumn(COMMON_FAMILY, widget.getTitle().getBytes(UTF8));
        if (table.exists(widgetTest)) {
          LOG.warn("Widget: " + widget.getTitle() + " already exists.");
          created = false;
        } else {
          Put put = new Put(WIDGET_TYPE);
          Gson gson = new Gson();
          String buffer = gson.toJson(widget);
          put.addColumn(COMMON_FAMILY, widget.getTitle().getBytes(UTF8), buffer.getBytes(UTF8));
          table.put(put);
          created = true;
        }
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return created;
    }
  
    /**
     * Update a widget in HBase.
     * 
     * @param title is searchable title in a widget
     * @param widget is Chukwa Widget object
     * @return true if widget has been updated
     */
    public static synchronized boolean updateWidget(String title, Widget widget) {
      boolean result = false;
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Delete oldWidget = new Delete(WIDGET_TYPE);
        oldWidget.addColumn(COMMON_FAMILY, title.getBytes(UTF8));
        table.delete(oldWidget);
        Put put = new Put(WIDGET_TYPE);
        Gson gson = new Gson();
        String buffer = gson.toJson(widget);
        put.addColumn(COMMON_FAMILY, title.getBytes(UTF8), buffer.getBytes(UTF8));
        table.put(put);
        table.close();
        result = true;
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
        LOG.error("Error in updating widget, original title: " + 
          title + " new title:" + widget.getTitle());
      }
      return result;
    }
  
    /**
     * Delete a widget in HBase.
     * 
     * @param title is searchable title in a widget
     * @return true if widget has been deleted
     */
    public static synchronized boolean deleteWidget(String title) {
      boolean result = false;
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Delete oldWidget = new Delete(WIDGET_TYPE);
        oldWidget.addColumn(COMMON_FAMILY, title.getBytes(UTF8));
        table.delete(oldWidget);
        table.close();
        result = true;
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
        LOG.error("Error in deleting widget: "+ title);
      }
      return result;
    }
  
    public static boolean isDefaultExists() {
      boolean exists = false;
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Get dashboardTest = new Get(DASHBOARD_TYPE);
        dashboardTest.addColumn(COMMON_FAMILY, "default".getBytes(UTF8));
        exists = table.exists(dashboardTest);
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
      }
      return exists;
    }
  
    public static void populateDefaults() {
      boolean defaultExists = isDefaultExists();
      try {
        if(defaultExists) {
          return;
        }
        // Populate example chart widgets
        createChart(Examples.SYSTEM_LOAD_AVERAGE);
        createChart(Examples.CPU_UTILIZATION);
        createChart(Examples.MEMORY_UTILIZATION);
        createChart(Examples.DISK_UTILIZATION);
        createChart(Examples.NETWORK_UTILIZATION);
        createChart(Examples.SWAP_UTILIZATION);
        
        // Namenode heap usage
        createChart(Examples.NAMENODE_MEMORY);
        
        // HDFS Usage
        createChart(Examples.HDFS_USAGE);
  
        // Resource Manager Memory
        createChart(Examples.RESOURCE_MANAGER_MEMORY);
  
        // Node Managers Health
        createChart(Examples.NODE_MANAGER_HEALTH);
  
        // High Availability State
        createChart(Examples.HDFS_HA);
  
        // HDFS Load
        createChart(Examples.HDFS_LOAD);
  
        // Namenode RPC Latency
        createChart(Examples.NAMENODE_RPC_LATENCY);
  
        // Datanode Health
        createChart(Examples.DATANODES);
  
        // HBase Master Memory
        createChart(Examples.HBASE_MASTER_MEMORY);
  
        // Populate default widgets
        createWidget(Examples.SYSTEM_LOAD_AVERAGE_WIDGET);
        createWidget(Examples.WELCOME_PAGE_WIDGET);
        createWidget(Examples.TRIAL_DOWNLOAD_WIDGET);
        createWidget(Examples.CLUSTER_RUNNING_WIDGET);
        createWidget(Examples.USER_WORKING_WIDGET);
        createWidget(Examples.APP_RUNNING_WIDGET);
        createWidget(Examples.TRIAL_ABANDON_RATE_WIDGET);
        createWidget(Examples.CLUSTERS_HEALTH_WIDGET);
        createWidget(Examples.TOP_ACTIVE_CLUSTERS_WIDGET);
        createWidget(Examples.TOP_APP_WIDGET);
        
        // User widgets
        createWidget(Examples.APP_USAGE_WIDGET);
        createWidget(Examples.QUICK_LINKS_WIDGET);
        createWidget(Examples.LOG_SEARCH_WIDGET);
        createWidget(Examples.YARN_APP_WIDGET);
        createWidget(Examples.HDFS_WIDGET);
        createWidget(Examples.HBASE_TABLE_WIDGET);
        createWidget(Examples.TOP_USER_WIDGET);
  
        // System widgets
        createWidget(Examples.HDFS_HA_STATE_WIDGET);
        createWidget(Examples.HDFS_LOAD_WIDGET);
        createWidget(Examples.HDFS_NAMENODE_LATENCY_WIDGET);
        createWidget(Examples.DATANODES_HEALTH_WIDGET);
        createWidget(Examples.NODE_MANAGERS_HEALTH_WIDGET);
        createWidget(Examples.HDFS_REMAINING_WIDGET);
        createWidget(Examples.NAMENODE_MEMORY_WIDGET);
        createWidget(Examples.RESOURCE_MANAGER_MEMORY_WIDGET);
        createWidget(Examples.HBASE_MASTER_MOMORY_WIDGET);
        createWidget(Examples.CPU_UTILIZATION_WIDGET);
        createWidget(Examples.MEMORY_UTILIZATION_WIDGET);
        createWidget(Examples.SWAP_UTILIZATION_WIDGET);
        createWidget(Examples.DISK_UTILIZATION_WIDGET);
        createWidget(Examples.NETWORK_UTILIZATION_WIDGET);
        createWidget(Examples.CPU_HEAPMAP_WIDGET);
        createWidget(Examples.HDFS_UI_WIDGET);
        createWidget(Examples.HBASE_MASTER_UI_WIDGET);
        
        // Populate default dashboard
        updateDashboard("default", "", Examples.DEFAULT_DASHBOARD);
        updateDashboard("user", "", Examples.USER_DASHBOARD);
        updateDashboard("system", "", Examples.SYSTEM_DASHBOARD);
      } catch (Throwable ex) {
        LOG.error(ExceptionUtil.getStackTrace(ex));
      }
    }
  
    public static synchronized Dashboard getDashboard(String id, String user) {
      Dashboard dash = null;
      String key = new StringBuilder().append(id).
          append("|").append(user).toString();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Get dashboard = new Get(DASHBOARD_TYPE);
        dashboard.addColumn(COMMON_FAMILY, key.getBytes(UTF8));
        Result rs = table.get(dashboard);
        byte[] buffer = rs.getValue(COMMON_FAMILY, key.getBytes(UTF8));
        if(buffer == null) {
          // If user dashboard is not found, use default dashboard.
          key = new StringBuilder().append(id).append("|").toString();
          dashboard = new Get(DASHBOARD_TYPE);
          dashboard.addColumn(COMMON_FAMILY, key.getBytes(UTF8));
          rs = table.get(dashboard);
          buffer = rs.getValue(COMMON_FAMILY, key.getBytes(UTF8));        
        }
        Gson gson = new Gson();
        dash = gson.fromJson(new String(buffer, UTF8), Dashboard.class);
        table.close();
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
        LOG.error("Error retrieving dashboard, id: " + 
          id + " user:" + user);
      }
      return dash;
    }
  
    public static boolean updateDashboard(String id, String user, Dashboard dash) {
      boolean result = false;
      String key = new StringBuilder().append(id).
          append("|").append(user).toString();
      try {
        getHBaseConnection();
        Table table = connection.getTable(TableName.valueOf(CHUKWA_META));
        Put put = new Put(DASHBOARD_TYPE);
        Gson gson = new Gson();
        String buffer = gson.toJson(dash);
        put.addColumn(COMMON_FAMILY, key.getBytes(UTF8), buffer.getBytes(UTF8));
        table.put(put);
        table.close();
        result = true;
      } catch (Exception e) {
        closeHBase();
        LOG.error(ExceptionUtil.getStackTrace(e));
        LOG.error("Error in updating dashboard, id: " + 
          id + " user:" + user);
      }
      return result;
    }
  
  }
  