azkaban-memoizeit

diff --git a/azkaban-common/src/main/java/azkaban/executor/ConnectorParams.java b/azkaban-common/src/main/java/azkaban/executor/ConnectorParams.java
index 5eb19a4..8e933ef 100644
--- a/azkaban-common/src/main/java/azkaban/executor/ConnectorParams.java
+++ b/azkaban-common/src/main/java/azkaban/executor/ConnectorParams.java
@@ -92,9 +92,15 @@ public interface ConnectorParams {
   public static final String STATS_SET_CLEANINGINTERVAL = "changeCleaningInterval";
   public static final String STATS_SET_MAXREPORTERPOINTS = "changeEmitterPoints";
   public static final String STATS_SET_ENABLEMETRICS = "enableMetrics";
-  public static final String STATS_SET_DISBLEMETRICS = "disableMetrics";
+  public static final String STATS_SET_DISABLEMETRICS = "disableMetrics";
   public static final String STATS_MAP_METRICNAMEPARAM = "metricName";
-  public static final String STATS_MAP_METRICRETRIVALMODE = "useStats";
+
+  /**
+   * useStats param is used to filter datapoints on /stats graph by using standard deviation and means
+   * By default, we consider only top/bottom 5% datapoints
+   */
+
+  public static final String STATS_MAP_METRICRETRIEVALMODE = "useStats";
   public static final String STATS_MAP_STARTDATE = "from";
   public static final String STATS_MAP_ENDDATE = "to";
   public static final String STATS_MAP_REPORTINGINTERVAL = "interval";

diff --git a/azkaban-common/src/main/java/azkaban/executor/ExecutorManagerAdapter.java b/azkaban-common/src/main/java/azkaban/executor/ExecutorManagerAdapter.java
index dec133e..10379ec 100644
--- a/azkaban-common/src/main/java/azkaban/executor/ExecutorManagerAdapter.java
+++ b/azkaban-common/src/main/java/azkaban/executor/ExecutorManagerAdapter.java
@@ -165,6 +165,19 @@ public interface ExecutorManagerAdapter {
   public String submitExecutableFlow(ExecutableFlow exflow, String userId)
       throws ExecutorManagerException;
 
+  /**
+   * Manage servlet call for stats servlet in Azkaban execution server
+   * Action can take any of the following values
+   * <ul>
+   * <li>{@link azkaban.executor.ConnectorParams#STATS_SET_REPORTINGINTERVAL}<li>
+   * <li>{@link azkaban.executor.ConnectorParams#STATS_SET_CLEANINGINTERVAL}<li>
+   * <li>{@link azkaban.executor.ConnectorParams#STATS_SET_MAXREPORTERPOINTS}<li>
+   * <li>{@link azkaban.executor.ConnectorParams#STATS_GET_ALLMETRICSNAME}<li>
+   * <li>{@link azkaban.executor.ConnectorParams#STATS_GET_METRICHISTORY}<li>
+   * <li>{@link azkaban.executor.ConnectorParams#STATS_SET_ENABLEMETRICS}<li>
+   * <li>{@link azkaban.executor.ConnectorParams#STATS_SET_DISABLEMETRICS}<li>
+   * </ul>
+   */
   public Map<String, Object> callExecutorStats(String action,
       Pair<String, String>... params) throws IOException;
 

diff --git a/azkaban-common/src/main/java/azkaban/metric/AbstractMetric.java b/azkaban-common/src/main/java/azkaban/metric/AbstractMetric.java
index fcce205..6114d36 100644
--- a/azkaban-common/src/main/java/azkaban/metric/AbstractMetric.java
+++ b/azkaban-common/src/main/java/azkaban/metric/AbstractMetric.java
@@ -23,7 +23,7 @@ import org.apache.log4j.Logger;
  * @param <T> Type of Value of a given metric
  */
 public abstract class AbstractMetric<T> implements IMetric<T>, Cloneable{
-  protected static final Logger logger = Logger.getLogger(MetricReportManager.class);
+  protected static final Logger _logger = Logger.getLogger(MetricReportManager.class);
   protected String name;
   protected T value;
   protected String type;
@@ -35,7 +35,7 @@ public abstract class AbstractMetric<T> implements IMetric<T>, Cloneable{
    * @param initialValue Initial Value of a metric
    * @param manager Metric Manager whom a metric will report to
    */
-  public AbstractMetric(String metricName, String metricType, T initialValue, MetricReportManager manager) {
+  protected AbstractMetric(String metricName, String metricType, T initialValue, MetricReportManager manager) {
     name = metricName;
     type = metricType;
     value = initialValue;
@@ -82,13 +82,13 @@ public abstract class AbstractMetric<T> implements IMetric<T>, Cloneable{
    * @see azkaban.metric.IMetric#notifyManager()
    */
   public synchronized void notifyManager() {
-    logger.debug(String.format("Notifying Manager for %s", this.getClass().getName()));
+    _logger.debug(String.format("Notifying Manager for %s", this.getClass().getName()));
     try {
       metricManager.reportMetric( (IMetric<?>) this.clone());
     } catch (NullPointerException ex) {
-      logger.error(String.format("Metric Manager is not set for %s metric %s", this.getClass().getName(), ex.toString()));
+      _logger.error(String.format("Metric Manager is not set for %s metric %s", this.getClass().getName(), ex.toString()));
     } catch (CloneNotSupportedException ex) {
-      logger.error(String.format("Failed to take snapshot for %s metric %s", this.getClass().getName(), ex.toString()));
+      _logger.error(String.format("Failed to take snapshot for %s metric %s", this.getClass().getName(), ex.toString()));
     }
   }
 }

diff --git a/azkaban-common/src/main/java/azkaban/metric/GangliaMetricEmitter.java b/azkaban-common/src/main/java/azkaban/metric/GangliaMetricEmitter.java
index 76fb4d4..a5f2ead 100644
--- a/azkaban-common/src/main/java/azkaban/metric/GangliaMetricEmitter.java
+++ b/azkaban-common/src/main/java/azkaban/metric/GangliaMetricEmitter.java
@@ -16,8 +16,11 @@
 
 package azkaban.metric;
 
+import org.apache.commons.collections.bag.SynchronizedBag;
+
 import azkaban.utils.Props;
 
+
 /**
  * MetricEmitter implementation to report metric to a ganglia gmetric process
  */
@@ -34,7 +37,15 @@ public class GangliaMetricEmitter implements IMetricEmitter {
   }
 
   private String buildCommand(IMetric<?> metric) {
-    return String.format("%s -t %s -n %s -v %s", gmetricPath, metric.getValueType(), metric.getName(), metric.getValue().toString());
+    String cmd = null;
+
+    synchronized (metric) {
+      cmd =
+          String.format("%s -t %s -n %s -v %s", gmetricPath, metric.getValueType(), metric.getName(), metric.getValue()
+              .toString());
+    }
+
+    return cmd;
   }
 
   /**
@@ -45,13 +56,16 @@ public class GangliaMetricEmitter implements IMetricEmitter {
   @Override
   public void reportMetric(final IMetric<?> metric) throws Exception {
     String gangliaCommand = buildCommand(metric);
-    synchronized (metric) {
+
+    if (gangliaCommand != null) {
       // executes shell command to report metric to ganglia dashboard
       Process emission = Runtime.getRuntime().exec(gangliaCommand);
       int exitCode = emission.waitFor();
       if (exitCode != 0) {
         throw new RuntimeException("Failed to report metric using gmetric");
       }
+    } else {
+      throw new Exception("Failed to build ganglia Command");
     }
   }
 

diff --git a/azkaban-common/src/main/java/azkaban/metric/inmemoryemitter/InMemoryMetricEmitter.java b/azkaban-common/src/main/java/azkaban/metric/inmemoryemitter/InMemoryMetricEmitter.java
index febff63..3798689 100644
--- a/azkaban-common/src/main/java/azkaban/metric/inmemoryemitter/InMemoryMetricEmitter.java
+++ b/azkaban-common/src/main/java/azkaban/metric/inmemoryemitter/InMemoryMetricEmitter.java
@@ -44,26 +44,26 @@ public class InMemoryMetricEmitter implements IMetricEmitter {
   Map<String, LinkedList<InMemoryHistoryNode>> historyListMapping;
   private static final String INMEMORY_METRIC_REPORTER_WINDOW = "azkaban.metric.inmemory.interval";
   private static final String INMEMORY_METRIC_NUM_INSTANCES = "azkaban.metric.inmemory.maxinstances";
-  private static final String INMEMORY_METRIC_DEVIAION_FACTOR = "azkaban.metric.inmemory.statisticalDeviationFactor";
+  private static final String INMEMORY_METRIC_STANDARDDEVIATION_FACTOR = "azkaban.metric.inmemory.standardDeviationFactor";
 
-  double statisticalDeviationFactor;
+  private double standardDeviationFactor;
   /**
    * Interval (in millisecond) from today for which we should maintain the in memory snapshots
    */
-  long interval;
+  private long timeWindow;
   /**
    * Maximum number of snapshots that should be displayed on /stats servlet
    */
-  long numInstances;
+  private long numInstances;
 
   /**
    * @param azkProps Azkaban Properties
    */
   public InMemoryMetricEmitter(Props azkProps) {
     historyListMapping = new HashMap<String, LinkedList<InMemoryHistoryNode>>();
-    interval = azkProps.getLong(INMEMORY_METRIC_REPORTER_WINDOW, 60 * 60 * 24 * 7 * 1000);
+    timeWindow = azkProps.getLong(INMEMORY_METRIC_REPORTER_WINDOW, 60 * 60 * 24 * 7 * 1000);
     numInstances = azkProps.getLong(INMEMORY_METRIC_NUM_INSTANCES, 50);
-    statisticalDeviationFactor = azkProps.getDouble(INMEMORY_METRIC_DEVIAION_FACTOR, 2);
+    standardDeviationFactor = azkProps.getDouble(INMEMORY_METRIC_STANDARDDEVIATION_FACTOR, 2);
   }
 
   /**
@@ -71,7 +71,7 @@ public class InMemoryMetricEmitter implements IMetricEmitter {
    * @param val interval in milli seconds
    */
   public synchronized void setReportingInterval(long val) {
-    interval = val;
+    timeWindow = val;
   }
 
   /**
@@ -151,7 +151,7 @@ public class InMemoryMetricEmitter implements IMetricEmitter {
       InMemoryHistoryNode currentNode = ite.next();
       double value = ((Number) currentNode.getValue()).doubleValue();
       // remove all elements which lies in 95% value band
-      if (value < mean + statisticalDeviationFactor * std && value > mean - statisticalDeviationFactor * std) {
+      if (value < mean + standardDeviationFactor * std && value > mean - standardDeviationFactor * std) {
         ite.remove();
       }
     }
@@ -195,7 +195,7 @@ public class InMemoryMetricEmitter implements IMetricEmitter {
         // go ahead for clean up using latest possible value of interval
         // any interval change will not affect on going clean up
         synchronized (this) {
-          localCopyOfInterval = interval;
+          localCopyOfInterval = timeWindow;
         }
 
         // removing objects older than Interval time from firstAllowedDate

diff --git a/azkaban-common/src/main/java/azkaban/metric/MetricReportManager.java b/azkaban-common/src/main/java/azkaban/metric/MetricReportManager.java
index 2a49878..b69f6e5 100644
--- a/azkaban-common/src/main/java/azkaban/metric/MetricReportManager.java
+++ b/azkaban-common/src/main/java/azkaban/metric/MetricReportManager.java
@@ -26,6 +26,12 @@ import org.apache.log4j.Logger;
 
 /**
  * Manager for access or updating metric related functionality of Azkaban
+ * MetricManager is responsible all handling all action requests from statsServlet in Exec server
+ * <p> Metric Manager 'has a' relationship with :-
+ * <ul>
+ * <li>all the metric Azkaban is tracking</li>
+ * <li>all the emitters Azkaban is supposed to report metrics</li>
+ * </ul></p>
  */
 public class MetricReportManager {
   /**
@@ -36,10 +42,14 @@ public class MetricReportManager {
 
   /**
    * List of all the metrics that Azkaban is tracking
+   * Manager is not concerned with type of metric as long as it honors IMetric contracts
    */
   private List<IMetric<?>> metrics;
+
   /**
    * List of all the emitter listening all the metrics
+   * Manager is not concerned with how emitter is reporting value.
+   * Manager is only responsible to notify all emitters whenever an IMetric wants to be notified
    */
   private List<IMetricEmitter> metricEmitters;
   private ExecutorService executorService;

diff --git a/azkaban-common/src/main/java/azkaban/metric/TimeBasedReportingMetric.java b/azkaban-common/src/main/java/azkaban/metric/TimeBasedReportingMetric.java
index 69c23ec..62a2e10 100644
--- a/azkaban-common/src/main/java/azkaban/metric/TimeBasedReportingMetric.java
+++ b/azkaban-common/src/main/java/azkaban/metric/TimeBasedReportingMetric.java
@@ -64,7 +64,7 @@ public abstract class TimeBasedReportingMetric<T> extends AbstractMetric<T> {
    * @param interval
    */
   public void updateInterval(final long interval) {
-    logger.debug(String.format("Updating tracking interval to %d milisecond for %s metric", interval, getName()));
+    _logger.debug(String.format("Updating tracking interval to %d milisecond for %s metric", interval, getName()));
     timer.cancel();
     timer = new Timer();
     timer.schedule(getTimerTask(), interval, interval);

diff --git a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedFlowMetric.java b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedFlowMetric.java
index 99f6b03..4945ecc 100644
--- a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedFlowMetric.java
+++ b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedFlowMetric.java
@@ -31,7 +31,7 @@ public class NumFailedFlowMetric extends TimeBasedReportingMetric<Integer> imple
 
   public NumFailedFlowMetric(MetricReportManager manager, long interval) {
     super(NUM_FAILED_FLOW_METRIC_NAME, NUM_FAILED_FLOW_METRIC_TYPE, 0, manager, interval);
-    logger.debug("Instantiated NumFailedJobMetric");
+    _logger.debug("Instantiated NumFailedJobMetric");
   }
 
   /**

diff --git a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedJobMetric.java b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedJobMetric.java
index 6e16899..94eff06 100644
--- a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedJobMetric.java
+++ b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumFailedJobMetric.java
@@ -33,7 +33,7 @@ public class NumFailedJobMetric extends TimeBasedReportingMetric<Integer> implem
 
   public NumFailedJobMetric(MetricReportManager manager, long interval) {
     super(NUM_FAILED_JOB_METRIC_NAME, NUM_FAILED_JOB_METRIC_TYPE, 0, manager, interval);
-    logger.debug("Instantiated NumFailedJobMetric");
+    _logger.debug("Instantiated NumFailedJobMetric");
   }
 
   /**

diff --git a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumQueuedFlowMetric.java b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumQueuedFlowMetric.java
index 6ecc385..98446b6 100644
--- a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumQueuedFlowMetric.java
+++ b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumQueuedFlowMetric.java
@@ -33,7 +33,7 @@ public class NumQueuedFlowMetric extends TimeBasedReportingMetric<Integer> {
    */
   public NumQueuedFlowMetric(FlowRunnerManager flowRunnerManager, MetricReportManager manager, long interval) {
     super(NUM_QUEUED_FLOW_METRIC_NAME, NUM_QUEUED_FLOW_METRIC_TYPE, 0, manager, interval);
-    logger.debug("Instantiated NumQueuedFlowMetric");
+    _logger.debug("Instantiated NumQueuedFlowMetric");
     flowManager = flowRunnerManager;
   }
 

diff --git a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningFlowMetric.java b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningFlowMetric.java
index b611151..7c87a4e 100644
--- a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningFlowMetric.java
+++ b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningFlowMetric.java
@@ -36,7 +36,7 @@ public class NumRunningFlowMetric extends TimeBasedReportingMetric<Integer> {
    */
   public NumRunningFlowMetric(FlowRunnerManager flowRunnerManager, MetricReportManager manager, long interval) {
     super(NUM_RUNNING_FLOW_METRIC_NAME, NUM_RUNNING_FLOW_METRIC_TYPE, 0, manager, interval);
-    logger.debug("Instantiated NumRunningFlowMetric");
+    _logger.debug("Instantiated NumRunningFlowMetric");
     flowManager = flowRunnerManager;
   }
 

diff --git a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningJobMetric.java b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningJobMetric.java
index 1ad39d1..bc9ffc2 100644
--- a/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningJobMetric.java
+++ b/azkaban-execserver/src/main/java/azkaban/execapp/metric/NumRunningJobMetric.java
@@ -35,7 +35,7 @@ public class NumRunningJobMetric extends TimeBasedReportingMetric<Integer> imple
    */
   public NumRunningJobMetric(MetricReportManager manager, long interval) {
     super(NUM_RUNNING_JOB_METRIC_NAME, NUM_RUNNING_JOB_METRIC_TYPE, 0, manager, interval);
-    logger.debug("Instantiated NumRunningJobMetric");
+    _logger.debug("Instantiated NumRunningJobMetric");
   }
 
   /**
@@ -53,7 +53,7 @@ public class NumRunningJobMetric extends TimeBasedReportingMetric<Integer> imple
   }
 
   @Override
-  protected synchronized void preTrackingEventMethod() {
+  protected void preTrackingEventMethod() {
     // nothing to finalize value is already updated
   }
 

diff --git a/azkaban-execserver/src/main/java/azkaban/execapp/StatsServlet.java b/azkaban-execserver/src/main/java/azkaban/execapp/StatsServlet.java
index a06b859..1cf6b07 100644
--- a/azkaban-execserver/src/main/java/azkaban/execapp/StatsServlet.java
+++ b/azkaban-execserver/src/main/java/azkaban/execapp/StatsServlet.java
@@ -51,11 +51,9 @@ import azkaban.utils.JSONUtils;
 public class StatsServlet extends HttpServlet implements ConnectorParams {
   private static final long serialVersionUID = 2L;
   private static final Logger logger = Logger.getLogger(StatsServlet.class);
-  private AzkabanExecutorServer server;
 
   public void init(ServletConfig config) throws ServletException {
-    server =
-        (AzkabanExecutorServer) config.getServletContext().getAttribute(ServerConstants.AZKABAN_SERVLET_CONTEXT_KEY);
+    // Nothing to initialize
   }
 
   public boolean hasParam(HttpServletRequest request, String param) {
@@ -96,8 +94,10 @@ public class StatsServlet extends HttpServlet implements ConnectorParams {
         handleGetMetricHistory(req, ret);
       } else if (action.equals(STATS_SET_ENABLEMETRICS)) {
         handleChangeManagerStatusRequest(req, ret, true);
-      } else if (action.equals(STATS_SET_DISBLEMETRICS)) {
+      } else if (action.equals(STATS_SET_DISABLEMETRICS)) {
         handleChangeManagerStatusRequest(req, ret, false);
+      } else {
+        ret.put(RESPONSE_ERROR, "Invalid action");
       }
     }
 
@@ -177,7 +177,7 @@ public class StatsServlet extends HttpServlet implements ConnectorParams {
           List<InMemoryHistoryNode> result =
               memoryEmitter.getDrawMetric(getParam(req, STATS_MAP_METRICNAMEPARAM),
                   parseDate(getParam(req, STATS_MAP_STARTDATE)), parseDate(getParam(req, STATS_MAP_ENDDATE)),
-                  getBooleanParam(req, STATS_MAP_METRICRETRIVALMODE));
+                  getBooleanParam(req, STATS_MAP_METRICRETRIEVALMODE));
 
           if (result != null && result.size() > 0) {
             ret.put("data", result);

diff --git a/azkaban-webserver/src/main/java/azkaban/webapp/servlet/StatsServlet.java b/azkaban-webserver/src/main/java/azkaban/webapp/servlet/StatsServlet.java
index 4a99e9b..88b76f3 100644
--- a/azkaban-webserver/src/main/java/azkaban/webapp/servlet/StatsServlet.java
+++ b/azkaban-webserver/src/main/java/azkaban/webapp/servlet/StatsServlet.java
@@ -81,8 +81,8 @@ public class StatsServlet extends LoginAbstractAzkabanServlet {
       handleChangeConfigurationRequest(ConnectorParams.STATS_SET_MAXREPORTERPOINTS, req, ret);
     } else if (actionName.equals(ConnectorParams.STATS_SET_ENABLEMETRICS)) {
       handleChangeConfigurationRequest(ConnectorParams.STATS_SET_ENABLEMETRICS, req, ret);
-    } else if (actionName.equals(ConnectorParams.STATS_SET_DISBLEMETRICS)) {
-      handleChangeConfigurationRequest(ConnectorParams.STATS_SET_DISBLEMETRICS, req, ret);
+    } else if (actionName.equals(ConnectorParams.STATS_SET_DISABLEMETRICS)) {
+      handleChangeConfigurationRequest(ConnectorParams.STATS_SET_DISABLEMETRICS, req, ret);
     }
 
     writeJSON(resp, ret);

diff --git a/azkaban-webserver/src/main/resources/azkaban/webapp/servlet/velocity/statsPage.vm b/azkaban-webserver/src/main/resources/azkaban/webapp/servlet/velocity/statsPage.vm
index 636d989..f853453 100644
--- a/azkaban-webserver/src/main/resources/azkaban/webapp/servlet/velocity/statsPage.vm
+++ b/azkaban-webserver/src/main/resources/azkaban/webapp/servlet/velocity/statsPage.vm
@@ -16,122 +16,107 @@
 
 <!DOCTYPE html>
 <html lang="en">
-  <head>
-
-#parse("azkaban/webapp/servlet/velocity/style.vm")
-#parse("azkaban/webapp/servlet/velocity/javascript.vm")
-
-    <link rel="stylesheet" type="text/css" href="${context}/css/bootstrap-datetimepicker.css" />
-
-    <script type="text/javascript" src="${context}/js/raphael.min.js"></script>
-    <script type="text/javascript" src="${context}/js/morris.min.js"></script>
-    <script type="text/javascript" src="${context}/js/moment.min.js"></script>
-    <script type="text/javascript" src="${context}/js/bootstrap-datetimepicker.min.js"></script>
-    <script type="text/javascript">
-      var contextURL = "${context}";
-      var currentTime = ${currentTime};
-      var timezone = "${timezone}";
-
-      function refreshMetricChart() {
-            var requestURL = '/stats';
-            var requestData = {
-              'action': 'metricHistory',
-              'from': new Date($('#datetimebegin').val()).toUTCString(),
-              'to'  : new Date($('#datetimeend').val()).toUTCString(),
-              'metricName': $('#metricName').val(),
-              'useStats': $("#useStats").is(':checked')
-            };
-            var successHandler = function(responseData) {
-              if(responseData.error != null) {
-                $('#reportedMetric').html(responseData.error);
-              } else {
-                var graphDiv = document.createElement('div');
-                $('#reportedMetric').html(graphDiv);
-
-                Morris.Line({
-                              element: graphDiv,
-                              data: responseData.data,
-                              xkey: 'timestamp',
-                              ykeys: ['value'],
-                              labels: [$('#metricName').val()]
-                            });
-              }
-            };
-            $.get(requestURL, requestData, successHandler, 'json');
-      }
-
-      $(document).ready(function () {
-          $('#datetimebegin').datetimepicker();
-          $('#datetimeend').datetimepicker();
-          $('#datetimebegin').on('change.dp', function(e) {
-              $('#datetimeend').data('DateTimePicker').setStartDate(e.date);
-            });
-          $('#datetimeend').on('change.dp', function(e) {
-              $('#datetimebegin').data('DateTimePicker').setEndDate(e.date);
-            });
-          $('#retrieve').click(refreshMetricChart);
-      });
-
-    </script>
-  </head>
-  <body>
-
-#set ($current_page="Statistics")
-#parse ("azkaban/webapp/servlet/velocity/nav.vm")
-
-#if ($errorMsg)
-  #parse ("azkaban/webapp/servlet/velocity/errormsg.vm")
-#else
-
-  ## Page header.
-
-    <div class="az-page-header">
-      <div class="container-full">
-        <div class="row">
-          <div class="header-title" style="width: 17%;">
-            <h1><a href="${context}/stats">Statistics</a></h1>
-          </div>
-          <div class="header-control" style="width: 900px; padding-top: 5px;">
-
-            <form id="metric-form" method="get">
-                        <label for="metricLabel" >Metric</label>
-              #if (!$metricList.isEmpty())
-                 <select id="metricName"  name="metricName" style="width:200px">
-                    #foreach ($metric in $metricList)
-                              <option value="${metric.name}" style="width:200px">${metric.name}</option>
+   <head>
+      #parse("azkaban/webapp/servlet/velocity/style.vm")
+      #parse("azkaban/webapp/servlet/velocity/javascript.vm")
+      <link rel="stylesheet" type="text/css" href="${context}/css/bootstrap-datetimepicker.css" />
+      <script type="text/javascript" src="${context}/js/raphael.min.js"></script>
+      <script type="text/javascript" src="${context}/js/morris.min.js"></script>
+      <script type="text/javascript" src="${context}/js/moment.min.js"></script>
+      <script type="text/javascript" src="${context}/js/bootstrap-datetimepicker.min.js"></script>
+      <script type="text/javascript">
+         var contextURL = "${context}";
+         var currentTime = ${currentTime};
+         var timezone = "${timezone}";
+
+         function refreshMetricChart() {
+               var requestURL = '/stats';
+               var requestData = {
+                 'action': 'metricHistory',
+                 'from': new Date($('#datetimebegin').val()).toUTCString(),
+                 'to'  : new Date($('#datetimeend').val()).toUTCString(),
+                 'metricName': $('#metricName').val(),
+                 'useStats': $("#useStats").is(':checked')
+               };
+               var successHandler = function(responseData) {
+                 if(responseData.error != null) {
+                   $('#reportedMetric').html(responseData.error);
+                 } else {
+                   var graphDiv = document.createElement('div');
+                   $('#reportedMetric').html(graphDiv);
+
+                   Morris.Line({
+                                 element: graphDiv,
+                                 data: responseData.data,
+                                 xkey: 'timestamp',
+                                 ykeys: ['value'],
+                                 labels: [$('#metricName').val()]
+                               });
+                 }
+               };
+               $.get(requestURL, requestData, successHandler, 'json');
+         }
+
+         $(document).ready(function () {
+             $('#datetimebegin').datetimepicker();
+             $('#datetimeend').datetimepicker();
+             $('#datetimebegin').on('change.dp', function(e) {
+                 $('#datetimeend').data('DateTimePicker').setStartDate(e.date);
+               });
+             $('#datetimeend').on('change.dp', function(e) {
+                 $('#datetimebegin').data('DateTimePicker').setEndDate(e.date);
+               });
+             $('#retrieve').click(refreshMetricChart);
+         });
+
+      </script>
+   </head>
+   <body>
+      #set ($current_page="Statistics")
+      #parse ("azkaban/webapp/servlet/velocity/nav.vm")
+      #if ($errorMsg)
+      #parse ("azkaban/webapp/servlet/velocity/errormsg.vm")
+      #else
+      ## Page header.
+      <div class="az-page-header">
+         <div class="container-full">
+            <div class="row">
+               <div class="header-title" style="width: 17%;">
+                  <h1><a href="${context}/stats">Statistics</a></h1>
+               </div>
+               <div class="header-control" style="width: 900px; padding-top: 5px;">
+                  <form id="metric-form" method="get">
+                     <label for="metricLabel" >Metric</label>
+                     #if (!$metricList.isEmpty())
+                     <select id="metricName"  name="metricName" style="width:200px">
+                        #foreach ($metric in $metricList)
+                        <option value="${metric.name}" style="width:200px">${metric.name}</option>
+                        #end
+                     </select>
                      #end
-                 </select>
-              #end
-                   <label for="datetimebegin" >Between</label>
-
-                    <input type="text" id="datetimebegin" value="" class="ui-datetime-container" style="width:150px">
-
-                   <label for="datetimeend" >and</label>
-
+                     <label for="datetimebegin" >Between</label>
+                     <input type="text" id="datetimebegin" value="" class="ui-datetime-container" style="width:150px">
+                     <label for="datetimeend" >and</label>
                      <input type="text" id="datetimeend" value="" class="ui-datetime-container" style="width:150px">
-                    <input type="checkbox" name="useStats" id="useStats" value="true"> useStats
-                    <input type="button" id="retrieve" value="Retrieve" class="btn btn-success" >
-                </div>
-              </div>
+                     <input type="checkbox" name="useStats" id="useStats" value="true"> useStats
+                     <input type="button" id="retrieve" value="Retrieve" class="btn btn-success" >
+               </div>
+            </div>
             </form>
-          </div>
-        </div>
+         </div>
       </div>
-    </div>
-
-    <div class="container-full">
-
-  #parse ("azkaban/webapp/servlet/velocity/alerts.vm")
-
-      <div class="row">
-        <div id="reportedMetric" style="padding: 60px 10px 10px 10px;height: 750px;">
-        </div>
-      </div><!-- /row -->
-
-
-
-  #parse ("azkaban/webapp/servlet/velocity/invalidsessionmodal.vm")
-    </div><!-- /container-full -->
-#end
-  </body>
-<html>
+      </div>
+      </div>
+      <div class="container-full">
+         #parse ("azkaban/webapp/servlet/velocity/alerts.vm")
+         <div class="row">
+            <div id="reportedMetric" style="padding: 60px 10px 10px 10px;height: 750px;">
+            </div>
+         </div>
+         <!-- /row -->
+         #parse ("azkaban/webapp/servlet/velocity/invalidsessionmodal.vm")
+      </div>
+      <!-- /container-full -->
+      #end
+   </body>
+   <html>
\ No newline at end of file