A Environment Monitor

EnvMonitor.java

package org.apache.hadoop.hbase.regionserver;

import java.awt.Color;
import java.awt.Font;
import java.io.File;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
import java.lang.management.ThreadMXBean;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.ArrayList;
//import java.util.Set;
//import java.util.TreeSet;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jrobin.core.FetchData;
import org.jrobin.core.FetchRequest;
import org.jrobin.core.RrdDb;
import org.jrobin.core.RrdDef;
import org.jrobin.core.Sample;
import org.jrobin.core.Util;
import org.jrobin.graph.RrdGraph;
import org.jrobin.graph.RrdGraphDef;

public class EnvMonitor extends Thread {
	static final Log LOG = LogFactory.getLog(EnvMonitor.class);
	
	private final HRegionServer server;
	RuntimeMXBean rtbean = ManagementFactory.getRuntimeMXBean();
	ThreadMXBean thbean = ManagementFactory.getThreadMXBean();
	final String pid;
	protected boolean running = false;
	
	class Data {
		long bin;
		long bout;
		long timestamp;

		Data() {
			this(0, 0);
		}
		Data(long in, long out) {
			update(in, out);
		}
		void update(long in, long out) {
			update(in, out, System.nanoTime());
		}
		void update(long rec, long tran, long stamp) {
			this.bin = rec;
			this.bout = tran;
			this.timestamp = stamp;
		}
	}
	
	Thread updater = new Thread(new Updater());
	Thread adapter = new Thread(new Adapter());
	Thread timer = new Thread(new Timer());

	private HashMap<String, Data> ifdata;
	private HashMap<String, Double> netStatus = new HashMap<String, Double>();
	private String dataFormat = "%5.1lf %sB";
	protected int width = 1280;
	protected int height = 400;
	private long timeStart = 0;
	private long genStart = 0;
	private String rrdPath = "";
	private String dataDir = "/tmp" + File.separator +"envmon";
	private int step = 10;
	private String xmlPath;
	Color[] colors = new Color[] { //rainbow(16);
			new Color(0xFF0000), 
			new Color(0xFF6000),
			new Color(0xFFBF00),
			new Color(0xDFFF00),
			new Color(0x80FF00),
			new Color(0x20FF00),
			new Color(0x00FF40),
			new Color(0x00FF9F),
			new Color(0x00FFFF),
			new Color(0x009FFF),
			new Color(0x0040FF),
			new Color(0x2000FF),
			new Color(0x8000FF),
			new Color(0xDF00FF),
			new Color(0xFF00BF),
			new Color(0xFF0060)
	};
	
	class Updater implements Runnable {

		@Override
		public void run() {
			while (running) {
				try {
					updateRrd();
					try {
						Thread.sleep(0, 100000);
					} catch (InterruptedException e) {
						e.printStackTrace();
						continue;
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			LOG.info(Thread.currentThread().getName() + " exiting");
		}
	}
	
	class Timer implements Runnable {

		@Override
		public void run() {
			while (running) {
				try {
					Thread.sleep(1000 * 3600 * 2);
				} catch (InterruptedException e) {
					e.printStackTrace();
					continue;
				}
				updateTime(Util.getTimestamp());
			}
			LOG.info(Thread.currentThread().getName() + " exiting");
		}
		
	}
	
	class Adapter implements Runnable {

		public Adapter() {
			
		}
		
		@Override
		public void run() {
			
		}
		
	}

	public EnvMonitor() {
		this(null);
	}
	
	public EnvMonitor(final HRegionServer server) {
		this.server = server;
		this.rrdPath = this.dataDir + File.separator + "envmon.rrd";
		this.xmlPath = this.dataDir + File.separator + "envmonRRDTool.xml";
		this.ifdata = new HashMap<String, Data>();
		this.pid = rtbean.getName().split("@")[0];
		this.running = true;
		
		// init /tmp/envmon
		File monDir = new File(dataDir);
		if (!monDir.exists()) {
			monDir.mkdir();
		}
		
		initDev();
		
		try {
			this.timeStart = Util.getTimestamp();
			this.genStart = timeStart;
			initRrd();
		} catch (Exception e) {e.printStackTrace();}
	}
	
	public static String [] exec(final String ... commands) {
		String[] ret = null;
		try {			
			ProcessBuilder pb = new ProcessBuilder(commands);
			Process p = pb.start();
			p.waitFor();
			
			String line = null;
			BufferedReader stdout = new BufferedReader(new InputStreamReader(p.getInputStream()));
			List<String> stdoutList = new ArrayList<String>();
			while ((line = stdout.readLine()) != null) {
				stdoutList.add(line);
			}
			BufferedReader stderr = new BufferedReader(new InputStreamReader(p.getErrorStream()));
			List<String> stderrList = new ArrayList<String>();
			while ((line = stderr.readLine()) != null) {
				stderrList.add(line);
			}
			ret = stdoutList.toArray(new String[0]);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return ret;
	}
	
	private void initDev() {
		// read directly from /proc/net/dev
		String[] content = readFile("/proc/net/dev");
		if (content != null) {
			for (int i = 0; i < content.length; ++i) {
				if (content[i].startsWith("eth")) {
			    	String[] value = content[i].split(" +");
			    	String[] devs = content[i].split(":");
			    	long in = Long.parseLong(value[1]);// 1 = bytes, 2 = packets;
			    	long out = Long.parseLong(value[9]);// 9 = bytes, 10 = packets;
			    	Data dt = new Data(in, out);
			    	this.ifdata.put(devs[0], dt);
				}
			}
		}
	}
	
	private static String[] list2array(List<String> list) {
		return list == null? null : list.toArray(new String[0]);
	}
	
	public String[] getLatency(String dest) {
		List<String> results = new ArrayList<String>();
		String[] get = exec("ping", "-c4", dest);
		if (get != null && get.length > 0) {
			for (int i = 0; i < get.length; ++i) {
				String str = get[i].trim();
				if (str.startsWith("64 bytes from")) {
					int a = str.indexOf("time=");
					int b = str.indexOf(" ", a);
					results.add(str.substring(a+5, b));
					continue;
				}
			}
		}
		return list2array(results);
	}
	
	public static String[] ifstat(String iface) {
		String[] res = exec("ifstat", "-i", iface, ".1", "4");
		return res;
	}
	
	public static String[] readDir(String dir) {
		// e.g. "/proc/PID/task/TID/fd"
		File d = new File(dir);
		return d.list();
	}
	
	public static File[] readEntry(String dir) {
		File d = new File(dir);
		return d.listFiles();
	}
	
	public static String[] readFile(String f) {
		List<String> ret = new ArrayList<String>();
		BufferedReader buffer = null;
		try {
			buffer = new BufferedReader(new FileReader(f));
		} catch (Exception e) {}
		String line = null;
		try {
			while((line = buffer.readLine()) != null) {
				ret.add(line.trim());
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		return list2array(ret);
	}
	
	public static void printContent(String[] str) {
		if (str != null) {
			for (int i = 0; i < str.length; ++i) {
				System.out.println(str[i]);
			}
		}
	}
	
	public void getThroughput() {
		for (Iterator<String> it = ifdata.keySet().iterator(); it.hasNext();) {
			String key = it.next();		
			String[] strs = ifstat(key);
			double in = 0.0;
			double out = 0.0;
			if (strs != null && strs.length > 0) {
				for (int j = 2; j < strs.length; ++j) {
					String strValue = strs[j].trim();
					String[] value = strValue.split(" +");
					assert value.length == 2;
					in += Double.parseDouble(value[0]);
					out += Double.parseDouble(value[1]);
				}
				in /= 4;
				out /= 4;
				netStatus.put(key + "_recv", in * 1000);
				netStatus.put(key + "_xmit", out * 1000);
			}
		}
	}
	
	public void parseDev() {
    	long stamp = System.nanoTime();
		String[] content = readFile("/proc/net/dev");
		String dev = null;
		long in = 0L;
		long out = 0L;
		if (content != null) {
			for (int i = 0; i < content.length; ++i) {
				if (content[i].startsWith("eth")) {
			    	String[] value = content[i].split(" +");
			    	String[] devs = content[i].split(":");
			    	dev = devs[0];
			    	in = Long.parseLong(value[1]);// 1 = bytes, 2 = packets;
			    	out = Long.parseLong(value[9]);// 9 = bytes, 10 = packets;
			    	long bytesin = ifdata.get(dev).bin;
			    	long bytesout = ifdata.get(dev).bout;
			    	long tm = stamp - ifdata.get(dev).timestamp;
			    	if (bytesin > in || bytesout > out) {
			    		LOG.warn("rollover for interface " + dev + ", reinitialising.");
			    	}
			    	else {
			    		netStatus.put(dev + "_recv",  ((double)(in - bytesin)) / tm * 1000000000);
			    		netStatus.put(dev + "_xmit", ((double)(out - bytesout)) / tm * 1000000000);
			    	}
			    	ifdata.get(dev).update(in, out);
				}
			}
		}
		return;
	}

	protected void updateTime(final long stamp) {
		this.genStart = stamp - 1;
	}
	@Override
	public void run() {
		
		updater.setName(getName() + ".RrdUpdater");
		updater.setDaemon(true);
		timer.setName(getName() + ".TimeUpdater");
		timer.setDaemon(true);
		updater.start();
		timer.start();
		adapter.setName(getName() + ".Adapter");
		adapter.setDaemon(true);
		adapter.start();
		
		while(running && checkServer()) {
			try {
				generate(genStart, Util.getTimestamp());
			} catch (Exception e) {
				e.printStackTrace();
			}
			try {
				Thread.sleep(60 * 1000);
			} catch (InterruptedException e) {
				continue;
			}
		}
		this.running = false;

		this.timer.interrupt();
		this.updater.interrupt();
		this.adapter.interrupt();
		
		LOG.info(getName() + " exiting");
	}

	private boolean checkServer() {
		return server == null? true : !server.isStopped();
	}

	public String fetchLatest(long start, long end, String func, String device) {
		RrdDb rrdDb = null;
		try {
			rrdDb = new RrdDb(rrdPath);
			FetchRequest req = rrdDb.createFetchRequest(func, start, end);
			FetchData fetch = req.fetchData();
			double [] v = fetch.getValues(device + "_recv");
			double [] u = fetch.getValues(device + "_xmit");			
			return v[v.length - 1] + " " + u[u.length - 1];
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (rrdDb != null) {
				try {
					rrdDb.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return null;
	}
	
	public HashMap<String, Double> current() {
		return netStatus;
	}
	
	public double current(String index) {
		return netStatus.get(index);
	}
	
	public String generate() {
		return generate(genStart, Util.getTimestamp());
	}

	public String generate(long start, long end) {
		RrdDb rrdDb = null;
		try {
			String graphPath = this.dataDir + File.separator + "envmon.png";
			RrdGraphDef gDef = new RrdGraphDef();
			gDef.setWidth(this.width);
			gDef.setHeight(this.height);
			gDef.setFilename(graphPath);
			gDef.setStartTime(start);
			gDef.setEndTime(end);
			gDef.setTitle("Environment Monitor - Throughputs");
			gDef.setVerticalLabel("Bytes/s");
			
			String [] dsNames = null;
			rrdDb = new RrdDb(rrdPath);
			dsNames = rrdDb.getDsNames();
			for (int i = 0; i < dsNames.length; ++i) {
				String dsName = dsNames[i];
				gDef.datasource(dsName, rrdPath, dsName, "AVERAGE");
				gDef.line(dsName, colors[i % colors.length], dsName, 4);
				
				gDef.gprint(dsName, "MIN", this.dataFormat + " Min");
				gDef.gprint(dsName, "AVERAGE", this.dataFormat + " Avg");
				gDef.gprint(dsName, "MAX", this.dataFormat + " Max");
				gDef.gprint(dsName, "LAST", this.dataFormat + " Last\\r");
				//gDef.print(dsName, "MIN", "min" + dsName + " = %.3f");
				//gDef.print(dsName, "AVERAGE", "avg" + dsName + " = %.3f");
				//gDef.print(dsName, "MAX", "max" + dsName + " = %.3f");
				//gDef.print(dsName, "LAST", "last" + dsName + " = %.3f");
			}
			gDef.setPoolUsed(false);
			gDef.setImageFormat("png");
			gDef.setSmallFont(new Font("Monospaced", Font.PLAIN, 14));
			gDef.setLargeFont(new Font("Sanserif", Font.BOLD, 18));
			
			RrdGraph graph = new RrdGraph(gDef);
			return graph.getRrdGraphInfo().getFilename();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (rrdDb != null) {
				try {
					rrdDb.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return null;
	}
	
	public void initRrd() throws Exception {
		RrdDb rrdDb = null;
		try {
			File f = new File(rrdPath);
			if (f.exists()) f.delete();
		} catch (Exception e) {}
		LOG.info("creating new Rrd file - @" + timeStart);
		RrdDef rrdDef = new RrdDef(rrdPath, timeStart - 1, step);

		for (Iterator<String> it = ifdata.keySet().iterator(); it.hasNext();) {
			String key = it.next();
			LOG.info("adding interface: " + key);
			rrdDef.addDatasource(key + "_recv", "GAUGE", 2 * step, 0, Double.NaN);
			rrdDef.addDatasource(key + "_xmit", "GAUGE", 2 * step, 0, Double.NaN);
		}
		
		rrdDef.addArchive("AVERAGE", 0.5, 1, 24 * 3600 / step);
		//rrdDef.addArchive("AVERAGE", 0.5, 300 / step, 7 * 288);
		rrdDb = new RrdDb(rrdDef);
		if (rrdDb != null) rrdDb.close();
	}
	
	public void updateRrd() throws Exception {
		RrdDb rrdDb = null;
		try {
			rrdDb = new RrdDb(rrdPath);
			String [] dsNames = rrdDb.getDsNames();
			long lastUpdateTime = rrdDb.getLastUpdateTime();
			long timestamp = Util.getTimestamp();
			if (timestamp > lastUpdateTime) {
				rrdDb.setInfo("T=" + timestamp);
				Sample sample = rrdDb.createSample();
				sample.setTime(timestamp);
				parseDev();
				//System.out.println(map);
				Double value = null;
				for (int i = 0; i < dsNames.length; ++i) {
					String dsName = dsNames[i];
					value = netStatus.get(dsName);
					if (value == null) value = Double.NaN;
					sample.setValue(dsName, value);
				}
				sample.update();
			}
		} finally {
			if (rrdDb != null) {
				try{
					rrdDb.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	public String getImagePath() {
		return "/monitor" + File.separator + "envmon.png";
	}
	
	public void cancel() {
		this.running = false;
	}
	
	public static void main(String[] args) {
        EnvMonitor mon = new EnvMonitor();
        mon.setName("EnvMonitor");
        mon.setDaemon(false);
        mon.start();
        try {
            Thread.sleep(1 * 60 * 1000);
        } catch (InterruptedException e) {}
        mon.cancel();
        //System.exit(0);
	}
	
	public void dump() {
		try {
			RrdDb rrdDb = new RrdDb(this.rrdPath);
			rrdDb.dumpXml(xmlPath);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
}

regionserver.jsp

<%@ page contentType="text/html;charset=UTF-8"
  import="java.util.*"
  import="java.io.IOException"
  import="org.apache.hadoop.io.Text"
  import="org.apache.hadoop.hbase.regionserver.HRegionServer"
  import="org.apache.hadoop.hbase.regionserver.HRegion"
  import="org.apache.hadoop.hbase.regionserver.EnvMonitor"
  import="org.apache.hadoop.hbase.regionserver.metrics.RegionServerMetrics"
  import="org.apache.hadoop.hbase.util.Bytes"
  import="org.apache.hadoop.hbase.HConstants"
  import="org.apache.hadoop.hbase.HServerInfo"
  import="org.apache.hadoop.hbase.HServerLoad"
  import="org.apache.hadoop.hbase.HRegionInfo" %><%
  HRegionServer regionServer = (HRegionServer)getServletContext().getAttribute(HRegionServer.REGIONSERVER);
  HServerInfo serverInfo = null;
  try {
    serverInfo = regionServer.getHServerInfo();
  } catch (IOException e) {
    e.printStackTrace();
  }
  RegionServerMetrics metrics = regionServer.getMetrics();
  List<HRegionInfo> onlineRegions = regionServer.getOnlineRegions();
  int interval = regionServer.getConfiguration().getInt("hbase.regionserver.msginterval", 3000)/1000;
  EnvMonitor mon = regionServer.getMonitor();

%><?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" 
  "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd"> 
<html xmlns="http://www.w3.org/1999/xhtml">
<head><meta http-equiv="Content-Type" content="text/html;charset=UTF-8"/>
<title>HBase Region Server: <%= serverInfo.getServerAddress().getHostname() %>:<%= serverInfo.getServerAddress().getPort() %></title>
<link rel="stylesheet" type="text/css" href="/static/hbase.css" />
<style type="text/css">
#links_menu { border: none; }
</style>
</head>

<body>
<a id="logo" href="http://wiki.apache.org/lucene-hadoop/Hbase"><img src="/static/hbase_logo_med.gif" alt="HBase Logo" title="HBase Logo" /></a>
<h1 id="page_title">Region Server: <%= serverInfo.getServerAddress().getHostname() %>:<%= serverInfo.getServerAddress().getPort() %></h1>
<table id="links_menu">
<tr><td><a href="/logs/">Local logs</a>, </td><td><a href="/stacks">Thread Dump</a>, </td><td><a href="/logLevel">Log Level</a>, </td><td>
<form method="get">
<input type="hidden" name="action" value="refresh">
<input type="submit" value="Refresh Image">
</form>
</td>
</tr>
</table>
<% String action = request.getParameter("action");
  if (action != null && action.equals("refresh")) {
    mon.interrupt();
  } 
%>
<hr id="head_rule" />

<h2>Region Server Attributes</h2>
<table>
<tr><th>Attribute Name</th><th>Value</th><th>Description</th></tr>
<tr><td>HBase Version</td><td><%= org.apache.hadoop.hbase.util.VersionInfo.getVersion() %>, r<%= org.apache.hadoop.hbase.util.VersionInfo.getRevision() %></td><td>HBase version and svn revision</td></tr>
<tr><td>HBase Compiled</td><td><%= org.apache.hadoop.hbase.util.VersionInfo.getDate() %>, <%= org.apache.hadoop.hbase.util.VersionInfo.getUser() %></td><td>When HBase version was compiled and by whom</td></tr>
<tr><td>Metrics</td><td><%= metrics.toString() %></td><td>RegionServer Metrics; file and heap sizes are in megabytes</td></tr>
<tr><td>Zookeeper Quorum</td><td><%= regionServer.getZooKeeper().getQuorum() %></td><td>Addresses of all registered ZK servers</td></tr>
</table>

<h2>Region Server Monitor</h2>
<div>
<img src="/monitor/envmon.png" alt="Monitoring status" title="Network Status" />
</div>

<h2>Online Regions</h2>
<% if (onlineRegions != null && onlineRegions.size() > 0) { %>
<table>
<tr><th>Region Name</th><th>Start Key</th><th>End Key</th><th>Metrics</th></tr>
<%   
  Collections.sort(onlineRegions);
  for (HRegionInfo r: onlineRegions) { 
        HServerLoad.RegionLoad load = regionServer.createRegionLoad(r.getEncodedName());
 %>
<tr><td><%= r.getRegionNameAsString() %></td>
    <td><%= Bytes.toStringBinary(r.getStartKey()) %></td><td><%= Bytes.toStringBinary(r.getEndKey()) %></td>
    <td><%= load == null? "null": load.toString() %></td>
    </tr>
<%   } %>
</table>
<p>Region names are made of the containing table's name, a comma,
the start key, a comma, and a randomly generated region id.  To illustrate,
the region named
<em>domains,apache.org,5464829424211263407</em> is party to the table 
<em>domains</em>, has an id of <em>5464829424211263407</em> and the first key
in the region is <em>apache.org</em>.  The <em>-ROOT-</em>
and <em>.META.</em> 'tables' are internal sytem tables (or 'catalog' tables in db-speak).
The -ROOT- keeps a list of all regions in the .META. table.  The .META. table
keeps a list of all regions in the system. The empty key is used to denote
table start and table end.  A region with an empty start key is the first region in a table.
If region has both an empty start and an empty end key, its the only region in the table.  See
<a href="http://hbase.org">HBase Home</a> for further explication.<p>
<% } else { %>
<p>Not serving regions</p>
<% } %>
</body>
</html>