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.analysis.salsa.visualization;
  
  
  import prefuse.data.*;
  import prefuse.action.*;
  import prefuse.action.layout.*;
  import prefuse.action.assignment.*;
  import prefuse.visual.*;
  import prefuse.render.*;
  import prefuse.util.*;
  import prefuse.*;
  
  import org.apache.hadoop.chukwa.hicc.OfflineTimeHandler;
  import org.apache.hadoop.chukwa.hicc.TimeHandler;
  import org.apache.hadoop.chukwa.util.DatabaseWriter;
  import org.apache.hadoop.chukwa.database.Macro;
  import org.apache.hadoop.chukwa.util.XssFilter;
  
  import javax.servlet.http.*;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import java.sql.*;
  import java.util.*;
  
  import java.awt.Font;
  import java.awt.geom.Rectangle2D;
  import java.awt.Color;
  
  /**
   * Static image rendering for heatmap visualization of spatial HDFS 
   * activity patterns for scalable rendering on front-end (web-browser)
   * Handles database data retrieval, transforming data to form for 
   * visualization elements, and initializing and calling visualization
   * elements 
   */
  public class Heatmap {
  
    /**
     * Internal representation of all data needed to render heatmap;
     * data-handling code populates this data structure
     */
    protected static class HeatmapData {
      public Table agg_tab;
      public long [][] stats;
      public long min;
      public long max;
      public int num_hosts;
      public String [] hostnames;
      public HeatmapData() {
      }
    }
  
    private static Log log = LogFactory.getLog(Heatmap.class);
  
    static final String START_FIELD_NAME = "start_time_num";
    static final String END_FIELD_NAME = "finish_time_num";
  
    int BOXWIDTH = 250;
    int SIZE_X = 1600, SIZE_Y=1600;
    final int [] BORDER = {200,150,150,150};
    final int LEGEND_X_OFFSET = 10;
    final int LEGEND_Y_OFFSET = 0;
    final int LEGEND_TEXT_OFFSET = 10;
    final int LEGEND_FONT_SIZE = 24;
    final int AXIS_NAME_FONT_SIZE = 24;
  
    protected boolean offline_use = true;
    protected HttpServletRequest request;
  
    // for offline use only
    // keys that need to be filled:
    // period (last1/2/3/6/12/24hr,last7d,last30d), time_type (range/last), start, end
    protected HashMap<String, String> param_map;
    
    protected String cluster;
    protected String timezone;
    protected String query_state;
    protected String query_stat_type;
    protected final String table = "filesystem_fsm";
    protected boolean plot_legend = false; // controls whether to plot hostnames
    protected boolean sort_nodes = true;
    protected boolean plot_additional_info = true;
    protected String add_info_extra = null;
    
    protected Display dis;
    protected Visualization viz;
    
    protected Rectangle2D dataBound = new Rectangle2D.Double();
    protected Rectangle2D xlabBound = new Rectangle2D.Double();
    protected Rectangle2D ylabBound = new Rectangle2D.Double();
    protected Rectangle2D labelBottomBound = new Rectangle2D.Double();
    
    protected HashMap<String, String> prettyStateNames;
    
    /* Different group names allow control of what Renderers to use */
    final String maingroup = "Data";
    final String othergroup = "Misc";
    final String labelgroup = "Label";
    final String legendgroup = "Legend";
    final String legendshapegroup = "LegendShape";
    final String addinfogroup = "AddInfo";
    final String addinfoshapegroup = "AddInfoShape";
    
    public Heatmap() {
      this.cluster = "";
      this.timezone = "";
      this.query_state = "";
      this.query_stat_type = "";
      param_map = new HashMap<String, String>();
    }
    
    /**
     * Constructor for Swimlanes visualization object
     * @param timezone Timezone string from environment
     * @param cluster Cluster name from environment
     * @param event_type Whether to display shuffles or not
     * @param query_stat_type Query state type
     * @param valmap HashMap of key/value pairs simulating parameters from a HttpRequest
     */
    public Heatmap
      (String timezone, String cluster, String event_type, 
       String query_stat_type,
       HashMap<String, String> valmap) 
    {
      this.cluster = cluster;
      if (timezone != null) {
        this.timezone = timezone;
      } else {
        this.timezone = null;
      }    
      this.query_state = event_type;
      this.query_stat_type = query_stat_type;
  
      /* This should "simulate" an HttpServletRequest
       * Need to have "start" and "end" in seconds since Epoch
       */
      this.param_map = valmap; 
    }
    
    public Heatmap
      (String timezone, String cluster, String query_state, 
       String query_stat_type,
       HashMap<String, String> valmap, String shuffles) 
    {
      
      this.cluster = cluster;
      if (timezone != null) {
        this.timezone = timezone;
      } else {
        this.timezone = null;
      }
      this.query_state = query_state;
      this.query_stat_type = query_stat_type;
  
      /* This should "simulate" an HttpServletRequest
       * Need to have "start" and "end" in seconds since Epoch
       */
      this.param_map = valmap; 
      
    }
    
    public Heatmap
      (String timezone, String cluster, String query_state, 
       String query_stat_type,
       HashMap<String, String> valmap, 
       int w, int h) 
    {
      
      this.cluster = cluster;
      if (timezone != null) {
        this.timezone = timezone;
      } else {
        this.timezone = null;
      }
      this.query_state = query_state;
      this.query_stat_type = query_stat_type;
  
      /* This should "simulate" an HttpServletRequest
       * Need to have "start" and "end" in seconds since Epoch
       */
      this.param_map = valmap; 
          
      this.SIZE_X = w;
      this.SIZE_Y = h;
    }
    
    public Heatmap(HttpServletRequest request) {
      XssFilter xf = new XssFilter(request);
      this.offline_use = false;
      this.request = request;
      HttpSession session = request.getSession();
      this.cluster = session.getAttribute("cluster").toString();
      String query_state = xf.getParameter("query_state");
      if (query_state != null) {
        this.query_state = query_state;
      } else {
        this.query_state = "read";
      }
      String query_stat_type = xf.getParameter("query_stat_type");
      if (query_stat_type != null) {
        this.query_stat_type = query_stat_type;
      } else {
        this.query_stat_type = "transaction_count";
      }
      this.timezone = session.getAttribute("time_zone").toString();
    }
  
    /**
     * Set dimensions of image to be generated
     * Call before calling @see #run
     * @param width Image width in pixels
     * @param height Image height in pixels
     */
    public void setDimensions(int width, int height) {
      this.SIZE_X=width;
      this.SIZE_Y=height;
    }
    
    /**
     * Specify whether to print labels of hosts along axes
     * Call before calling @see #run
     * @param legendopt Flag to control plot legends
     */
    public void setLegend(boolean legendopt) {
      if (legendopt) {
        this.plot_legend = true;
      } else {
        this.plot_legend = false;
      }
    }
    
    
    /**
     * Generates image in specified format, and writes image as binary
     * output to supplied output stream 
     * @param output Image output stream
     * @param img_fmt Image format
     * @param scale Image scale
     * @return true if image is saved
     */
    public boolean getImage(java.io.OutputStream output, String img_fmt, double scale) {
      dis = new Display(this.viz);
      dis.setSize(SIZE_X,SIZE_Y);
      dis.setHighQuality(true);
      dis.setFont(new Font(Font.SANS_SERIF,Font.PLAIN,24));
      return dis.saveImage(output, img_fmt, scale);
    } 
    
    protected void setupRenderer() {
      this.viz.setRendererFactory(new RendererFactory(){
        AbstractShapeRenderer sr = new ShapeRenderer();
        ShapeRenderer sr_big = new ShapeRenderer(BOXWIDTH);
        LabelRenderer lr = new LabelRenderer("label");
        LabelRenderer lr_legend = new LabelRenderer("label");
              
        public Renderer getRenderer(VisualItem item) {
          lr_legend.setHorizontalAlignment(Constants.LEFT);
          lr_legend.setVerticalAlignment(Constants.CENTER);
          lr.setHorizontalAlignment(Constants.CENTER);
          lr.setVerticalAlignment(Constants.CENTER);
          if (item.isInGroup(maingroup)) {
            return sr_big;
          } else if (item.isInGroup(legendgroup)) {
            return lr_legend;
          } else if (item.isInGroup(addinfogroup)) {
            return lr;
          }
          return sr;
        }
      });
    }
    
    // setup columns: add additional time fields
    protected HeatmapData setupDataTable() {
      HeatmapData hd = this.getData();    
      return hd;
    }
    
    protected void setupHeatmap(VisualTable vtab, HeatmapData hd) 
    {
      long [][] stats = hd.stats;
      int i, j, curr_idx;
      long curr_val;
      int num_hosts = hd.num_hosts;
      ColorMap cm = new ColorMap(
        ColorLib.getInterpolatedPalette(
          ColorLib.color(ColorLib.getColor(32,0,0)),
          ColorLib.color(Color.WHITE)
        ),
        (double)hd.min,(double)hd.max
      );
      
      for (i = 0; i < num_hosts; i++) {
        for (j = 0; j < num_hosts; j++) {
          curr_idx = j+(i*num_hosts);
          curr_val = stats[i][j]; 
          if (curr_val >= hd.min) {
            vtab.setFillColor(curr_idx, cm.getColor((double)curr_val));
          } else if (curr_val == 0) {
            vtab.setFillColor(curr_idx, ColorLib.color(Color.BLACK));
          }
        }
      }
      
      // gridlayout puts tiles on row-wise (row1, followed by row2, etc.)
      GridLayout gl = new GridLayout(maingroup, num_hosts, num_hosts);
      gl.setLayoutBounds(this.dataBound);
      ActionList gl_list = new ActionList();
      gl_list.add(gl);
      this.viz.putAction("gridlayout",gl_list);
      this.viz.run("gridlayout");
    }
    
    protected void addHostLabels(HeatmapData hd) {
      Table legend_labels_table = new Table();
      legend_labels_table.addColumn("label",String.class);
      legend_labels_table.addRows(hd.hostnames.length);
      for (int i = 0; i < hd.hostnames.length; i++) {
        legend_labels_table.setString(i,"label",hd.hostnames[i]);
      }
      float start_x = LEGEND_X_OFFSET;
      float start_y = LEGEND_Y_OFFSET + BORDER[1] + (BOXWIDTH/2);    
      float incr = this.BOXWIDTH;
      VisualTable legend_labels_table_viz = this.viz.addTable(legendgroup, legend_labels_table);
      for (int i = 0; i < hd.hostnames.length; i++) {
        legend_labels_table_viz.setFloat(i, VisualItem.X, start_x + LEGEND_TEXT_OFFSET);
        legend_labels_table_viz.setFloat(i, VisualItem.Y, start_y + (i * incr));
        legend_labels_table_viz.setTextColor(i,ColorLib.color(java.awt.Color.BLACK));
        legend_labels_table_viz.setFont(i,new Font(Font.SANS_SERIF,Font.PLAIN,LEGEND_FONT_SIZE));
      } 
    }
    
    protected void addAddlInfo(HeatmapData hd) {
      Table legend_labels_table = new Table();
      legend_labels_table.addColumn("label",String.class);
      legend_labels_table.addRows(3);
      
      String hostnumstring = "Number of hosts: " + hd.num_hosts;
      if (sort_nodes) {
        hostnumstring += " (nodes sorted)";
      } else {
        hostnumstring += " (nodes not sorted)";
      }
      if (add_info_extra != null) hostnumstring += add_info_extra;
      legend_labels_table.setString(0,"label",hostnumstring);
      legend_labels_table.setString(1,"label","Src. Hosts");
      legend_labels_table.setString(2,"label","Dest. Hosts");
      
      VisualTable legend_labels_table_viz = this.viz.addTable(addinfogroup, legend_labels_table);
  
      legend_labels_table_viz.setFloat(0, VisualItem.X, this.SIZE_X/2f);
      legend_labels_table_viz.setFloat(0, VisualItem.Y, BORDER[1]/2f);
      legend_labels_table_viz.setTextColor(0,ColorLib.color(java.awt.Color.BLACK));
      legend_labels_table_viz.setFont(0,new Font(Font.SANS_SERIF,Font.PLAIN,LEGEND_FONT_SIZE));
  
      legend_labels_table_viz.setFloat(1, VisualItem.X, this.SIZE_X/2f);
      legend_labels_table_viz.setFloat(1, VisualItem.Y, BORDER[1] + (BOXWIDTH*hd.num_hosts) + BORDER[3]/2f);
      legend_labels_table_viz.setTextColor(1,ColorLib.color(java.awt.Color.BLACK));
      legend_labels_table_viz.setFont(1,new Font(Font.SANS_SERIF,Font.PLAIN,LEGEND_FONT_SIZE));
  
      legend_labels_table_viz.setFloat(2, VisualItem.X, BORDER[0] + (BOXWIDTH*hd.num_hosts) + BORDER[2]/2f);
      legend_labels_table_viz.setFloat(2, VisualItem.Y, this.SIZE_Y/2f);
      legend_labels_table_viz.setTextColor(2,ColorLib.color(java.awt.Color.BLACK));
      legend_labels_table_viz.setFont(2,new Font(Font.SANS_SERIF,Font.PLAIN,LEGEND_FONT_SIZE));
  
    }
    
    protected void initPrettyNames() {
      this.prettyStateNames = new HashMap<String, String>();
  
      prettyStateNames.put("read","Block Reads");
      prettyStateNames.put("write","Block Writes");
      prettyStateNames.put("read_local", "Local Block Reads");
      prettyStateNames.put("write_local", "Local Block Writes");
      prettyStateNames.put("read_remote", "Remote Block Reads");
      prettyStateNames.put("write_remote", "Remote Block Writes");
      prettyStateNames.put("write_replicated", "Replicated Block Writes");    
    }
    
    /**
     * Actual code that calls data, generates heatmap, and saves it
     */
    public void run() {
      initPrettyNames();
      
      // setup visualization
      this.viz = new Visualization();
      
      // add table to visualization
      HeatmapData hd = this.setupDataTable();
  
      // setup bounds
      int width;
      if (SIZE_X-BORDER[0]-BORDER[2] < SIZE_Y-BORDER[1]-BORDER[3]) {
        BOXWIDTH = (SIZE_X-BORDER[0]-BORDER[2]) / hd.num_hosts;
      } else {
        BOXWIDTH = (SIZE_Y-BORDER[1]-BORDER[3]) / hd.num_hosts;
      }
      width = hd.num_hosts * BOXWIDTH;
      this.dataBound.setRect(
        BORDER[0]+BOXWIDTH/2,
        BORDER[1]+BOXWIDTH/2,
        width-BOXWIDTH,width-BOXWIDTH
      );
      this.SIZE_X = BORDER[0] + BORDER[2] + (hd.num_hosts * BOXWIDTH);
      this.SIZE_Y = BORDER[1] + BORDER[3] + (hd.num_hosts * BOXWIDTH);
      
      log.debug("width total: " + width + " width per state: " + BOXWIDTH + " xstart: " 
        + (BORDER[0]+BOXWIDTH/2) 
        + " ystart: " + (BORDER[1]+BOXWIDTH/2) + " (num hosts: "+hd.num_hosts+")");
      log.debug("X size: " + this.SIZE_X + " Y size: " + this.SIZE_Y);
      
      this.setupRenderer();
      VisualTable data_tab_viz = viz.addTable(maingroup, hd.agg_tab);
      setupHeatmap(data_tab_viz, hd);
      
      ShapeAction legend_sa1 = null, legend_sa2 = null;
      SpecifiedLayout legendlabels_sl1 = null, legendlabels_sl2 = null;
      
      if (plot_legend) {
        addHostLabels(hd);
        legend_sa1 = new ShapeAction(legendshapegroup);
        legendlabels_sl1 = new SpecifiedLayout(legendgroup, VisualItem.X, VisualItem.Y);
        ActionList legenddraw = new ActionList();
        legenddraw.add(legend_sa1);
        this.viz.putAction(legendshapegroup, legenddraw);
        ActionList legendlabelsdraw = new ActionList();
        legendlabelsdraw.add(legendlabels_sl1);
        this.viz.putAction(legendgroup,legendlabelsdraw);
      }
  
      if (plot_additional_info) {
        addAddlInfo(hd);
        legend_sa2 = new ShapeAction(addinfoshapegroup);
        legendlabels_sl2 = new SpecifiedLayout(addinfogroup, VisualItem.X, VisualItem.Y);    
        ActionList legenddraw = new ActionList();
        legenddraw.add(legend_sa2);
        this.viz.putAction(addinfoshapegroup, legenddraw);
        ActionList legendlabelsdraw = new ActionList();
        legendlabelsdraw.add(legendlabels_sl2);
        this.viz.putAction(addinfogroup,legendlabelsdraw);
      }
  
    }
    
    protected boolean checkDone(int [] clustId) {
      for (int i = 1; i < clustId.length; i++) {
        if (clustId[i] != clustId[0]) return false;
      }
      return true;
    }
    
    /**
     * Sort data for better visualization of patterns
     */
    protected int [] hClust (long [][] stat) 
    {
      int statlen = stat.length;
      long [] rowSums = new long[statlen];
      int [] permute = new int[statlen];
      int i,j;
  
      // initialize permutation
      for (i = 0; i < statlen; i++) {
        permute[i] = i;
      }
      
      for (i = 0; i < statlen; i++) {
        rowSums[i] = 0;
        for (j = 0; j < statlen; j++) {
          rowSums[i] += stat[i][j];
        }
      }
      
      // insertion sort
      for (i = 0; i < statlen-1; i++) {
        long val = rowSums[i];
        int thispos = permute[i];
        j = i-1;
        while (j >= 0 && rowSums[j] > val) {
          rowSums[j+1] = rowSums[j];
          permute[j+1] = permute[j];
          j--;
        }
        rowSums[j+1] = val; 
        permute[j+1] = thispos;
      }
      
      return permute;
        
    }
    
    /**
     * Reorder rows (and columns) according to a given ordering
     * Maintains same ordering along rows and columns
     */
    protected long [][] doPermute (long [][] stat, int [] permute) {
      int statlen = stat.length;
      int i, j, curr_pos;
      long [][] stat2 = new long[statlen][statlen];
      
      assert(stat.length == permute.length);
      
      for (i = 0; i < statlen; i++) {
        curr_pos = permute[i];
        for (j = 0; j < statlen; j++) {
          stat2[i][j] = stat[curr_pos][permute[j]];
        }
      }
      
      return stat2;
    }
    
    /**
     * Interfaces with database to get data and 
     * populate data structures for rendering
     * @return heat map data JSON
     */
    @edu.umd.cs.findbugs.annotations.SuppressWarnings(value =
        "SQL_NONCONSTANT_STRING_PASSED_TO_EXECUTE", 
        justification = "Dynamic based upon tables in the database")
    public HeatmapData getData() {
      // preliminary setup
      OfflineTimeHandler time_offline;
      TimeHandler time_online;
      long start, end, min, max;
      
      if (offline_use) {
        time_offline = new OfflineTimeHandler(param_map, this.timezone);
        start = time_offline.getStartTime();
        end = time_offline.getEndTime();
      } else {
        time_online = new TimeHandler(this.request, this.timezone);
        start = time_online.getStartTime();
        end = time_online.getEndTime();
      }
      
      DatabaseWriter dbw = new DatabaseWriter(this.cluster);
      
      // setup query
      
      String sqlTemplate = "select block_id,start_time,finish_time,start_time_millis,finish_time_millis,status,state_name,hostname,other_host,bytes from [%s] where finish_time between '[start]' and '[end]' and (%s) order by start_time";
      String query;
      if (this.query_state != null && this.query_state.equals("read")) {
        query = String.format(sqlTemplate, table,"state_name like 'read_local' or state_name like 'read_remote'");
      } else if (this.query_state != null && this.query_state.equals("write")) {
        query = String.format(sqlTemplate, table, "state_name like 'write_local' or state_name like 'write_remote' or state_name like 'write_replicated'");
      } else {
        query = String.format(sqlTemplate, table, "state_name like '" + query_state + "'");
      } 
      Macro mp = new Macro(start,end,query);
      String q = mp.toString();
      
      ArrayList<HashMap<String, Object>> events = new ArrayList<HashMap<String, Object>>();
  
      ResultSet rs = null;
      
      log.debug("Query: " + q);
      // run query, extract results
      try {
        rs = dbw.query(q);
        ResultSetMetaData rmeta = rs.getMetaData();
        int col = rmeta.getColumnCount();
        while (rs.next()) {
          HashMap<String, Object> event = new HashMap<String, Object>();
          for(int i=1;i<=col;i++) {
            if(rmeta.getColumnType(i)==java.sql.Types.TIMESTAMP) {
              event.put(rmeta.getColumnName(i),rs.getTimestamp(i).getTime());
            } else {
              event.put(rmeta.getColumnName(i),rs.getString(i));
            }
          }
          events.add(event);
        }
      } catch (SQLException ex) {
        // handle any errors
        log.error("SQLException: " + ex.getMessage());
        log.error("SQLState: " + ex.getSQLState());
        log.error("VendorError: " + ex.getErrorCode());
      } finally {
        dbw.close();
      }    
  
      log.info(events.size() + " results returned.");
  
      HashSet<String> host_set = new HashSet<String>();
      HashMap<String, Integer> host_indices = new HashMap<String, Integer>();
      HashMap<Integer, String> host_rev_indices = new HashMap<Integer, String>();
  
      // collect hosts, name unique hosts
      for(int i = 0; i < events.size(); i++) {
        HashMap<String, Object> event = events.get(i);
        String curr_host = (String) event.get("hostname");
        String other_host = (String) event.get("other_host");
        host_set.add(curr_host);
        host_set.add(other_host);
      }
      int num_hosts = host_set.size();
  
      Iterator<String> host_iter = host_set.iterator();
      for (int i = 0; i < num_hosts && host_iter.hasNext(); i++) {
        String curr_host = host_iter.next();
        host_indices.put(curr_host, i);
        host_rev_indices.put(i,curr_host);
      }
  
      System.out.println("Number of hosts: " + num_hosts);
      long stats[][] = new long[num_hosts][num_hosts];
      long count[][] = new long[num_hosts][num_hosts]; // used for averaging
  
      int start_millis = 0, end_millis = 0;
  
      // deliberate design choice to duplicate code PER possible operation
      // otherwise we have to do the mode check N times, for N states returned
      //
      // compute aggregate statistics
      log.info("Query statistic type: "+this.query_stat_type);
      if (this.query_stat_type.equals("transaction_count")) {
        for(int i=0;i<events.size();i++) {
          HashMap<String, Object> event = events.get(i);
          start=(Long)event.get("start_time");
          end=(Long)event.get("finish_time");
          start_millis = Integer.parseInt(((String)event.get("start_time_millis")));
          end_millis = Integer.parseInt(((String)event.get("finish_time_millis")));      
          String this_host = (String) event.get("hostname");
          String other_host = (String) event.get("other_host");
          int this_host_idx = host_indices.get(this_host).intValue();
          int other_host_idx = host_indices.get(other_host).intValue();
  
          // to, from
          stats[other_host_idx][this_host_idx] += 1;
        }
      } else if (this.query_stat_type.equals("avg_duration")) {
        for(int i=0;i<events.size();i++) {
          HashMap<String, Object> event = events.get(i);
          start=(Long)event.get("start_time");
          end=(Long)event.get("finish_time");
          start_millis = Integer.parseInt(((String)event.get("start_time_millis")));
          end_millis = Integer.parseInt(((String)event.get("finish_time_millis")));      
          String this_host = (String) event.get("hostname");
          String other_host = (String) event.get("other_host");
          int this_host_idx = host_indices.get(this_host).intValue();
          int other_host_idx = host_indices.get(other_host).intValue();
  
          long curr_val = end_millis - start_millis + ((end - start)*1000);
  
          // to, from
          stats[other_host_idx][this_host_idx] += curr_val;
          count[other_host_idx][this_host_idx] += 1;
        }    
        for (int i = 0; i < num_hosts; i++) {
          for (int j = 0; j < num_hosts; j++) {
            if (count[i][j] > 0) stats[i][j] = stats[i][j] / count[i][j];
          }
        }
      } else if (this.query_stat_type.equals("avg_volume")) {
        for(int i=0;i<events.size();i++) {
          HashMap<String, Object> event = events.get(i);
          start=(Long)event.get("start_time");
          end=(Long)event.get("finish_time");
          start_millis = Integer.parseInt(((String)event.get("start_time_millis")));
          end_millis = Integer.parseInt(((String)event.get("finish_time_millis")));      
          String this_host = (String) event.get("hostname");
          String other_host = (String) event.get("other_host");
          int this_host_idx = host_indices.get(this_host).intValue();
          int other_host_idx = host_indices.get(other_host).intValue();
  
          long curr_val = Long.parseLong((String)event.get("bytes"));
  
          // to, from
          stats[other_host_idx][this_host_idx] += curr_val;
          count[other_host_idx][this_host_idx] += 1;
        }    
        for (int i = 0; i < num_hosts; i++) {
          for (int j = 0; j < num_hosts; j++) {
            if (count[i][j] > 0) stats[i][j] = stats[i][j] / count[i][j];
          }
        }
      } else if (this.query_stat_type.equals("total_duration")) {
        for(int i=0;i<events.size();i++) {
          HashMap<String, Object> event = events.get(i);
          start=(Long)event.get("start_time");
          end=(Long)event.get("finish_time");
          start_millis = Integer.parseInt(((String)event.get("start_time_millis")));
          end_millis = Integer.parseInt(((String)event.get("finish_time_millis")));      
          String this_host = (String) event.get("hostname");
          String other_host = (String) event.get("other_host");
          int this_host_idx = host_indices.get(this_host).intValue();
          int other_host_idx = host_indices.get(other_host).intValue();
  
          double curr_val = end_millis - start_millis + ((end - start)*1000);
  
          // to, from
          stats[other_host_idx][this_host_idx] += curr_val;
        } 
      } else if (this.query_stat_type.equals("total_volume")) {
        for(int i=0;i<events.size();i++) {
          HashMap<String, Object> event = events.get(i);
          start=(Long)event.get("start_time");
          end=(Long)event.get("finish_time");
          start_millis = Integer.parseInt(((String)event.get("start_time_millis")));
          end_millis = Integer.parseInt(((String)event.get("finish_time_millis")));      
          String this_host = (String) event.get("hostname");
          String other_host = (String) event.get("other_host");
          int this_host_idx = host_indices.get(this_host).intValue();
          int other_host_idx = host_indices.get(other_host).intValue();
  
          long curr_val = Long.parseLong((String)event.get("bytes"));
  
          // to, from
          stats[other_host_idx][this_host_idx] += curr_val;
        }    
      }
      
      int [] permute = null;
      if (sort_nodes) {
        permute = hClust(stats);
        stats = doPermute(stats,permute);
      }
      
      Table agg_tab = new Table();
      agg_tab.addColumn("stat", long.class);
      min = Long.MAX_VALUE;
      max = Long.MIN_VALUE;
      agg_tab.addRows(num_hosts*num_hosts);
      
      // row-wise placement (row1, followed by row2, etc.)
      for (int i = 0; i < num_hosts; i++) {
        for (int j = 0; j < num_hosts; j++) {
          agg_tab.setLong((i*num_hosts)+j,"stat",stats[i][j]);
          if (stats[i][j] > max) max = stats[i][j];
          if (stats[i][j] > 0 && stats[i][j] < min) min = stats[i][j];
        }
      }
      if (min == Long.MAX_VALUE) min = 0;
      
      log.info(agg_tab);
      
      // collate data
      HeatmapData hd = new HeatmapData();
      hd.stats = stats;
      hd.min = min;
      hd.max = max;
      hd.num_hosts = num_hosts;
      hd.agg_tab = agg_tab;
      
      this.add_info_extra = new StringBuilder().append("\nState: ").append(this.prettyStateNames.get(this.query_state)).
        append(" (").append(events.size()).append(" ").append(this.query_state).
        append("'s [").append(this.query_stat_type).append("])\n").
        append("Plotted value range: [").append(hd.min).append(",").append(hd.max).
        append("] (Zeros in black)").toString();
  
      hd.hostnames = new String [num_hosts];
      for (int i = 0; i < num_hosts; i++) {
        String curr_host = host_rev_indices.get(permute[i]);
        hd.hostnames[i] = curr_host;
      }
      
      return hd;
    }
    
  }
  