tajo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jihoon...@apache.org
Subject [2/2] git commit: TAJO-333: Add metric system to Tajo. (hyoungjunkim via jihoon)
Date Sat, 14 Dec 2013 03:33:49 GMT
TAJO-333: Add metric system to Tajo. (hyoungjunkim via jihoon)


Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/62c49c05
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/62c49c05
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/62c49c05

Branch: refs/heads/master
Commit: 62c49c05f522158d75d818df49100a0b1fd354bb
Parents: 1d0d458
Author: Jihoon Son <jihoonson@apache.org>
Authored: Sat Dec 14 12:32:41 2013 +0900
Committer: Jihoon Son <jihoonson@apache.org>
Committed: Sat Dec 14 12:33:40 2013 +0900

----------------------------------------------------------------------
 CHANGES.txt                                     |   2 +
 .../java/org/apache/tajo/conf/TajoConf.java     |   5 +-
 tajo-core/tajo-core-backend/pom.xml             |  16 ++
 .../src/main/java/log4j.properties              |   6 +-
 .../org/apache/tajo/master/GlobalEngine.java    |   7 +-
 .../java/org/apache/tajo/master/TajoMaster.java |  26 +-
 .../master/metrics/CatalogMetricsGaugeSet.java  |  54 ++++
 .../metrics/WorkerResourceMetricsGaugeSet.java  |  74 ++++++
 .../querymaster/QueryMasterManagerService.java  |   3 +
 .../master/querymaster/QueryMasterTask.java     |  13 +
 .../util/metrics/GroupNameMetricsFilter.java    |  43 ++++
 .../tajo/util/metrics/LogEventGaugeSet.java     |  64 +++++
 .../tajo/util/metrics/MetricsFilterList.java    |  43 ++++
 .../tajo/util/metrics/RegexpMetricsFilter.java  |  51 ++++
 .../tajo/util/metrics/TajoLogEventCounter.java  |  86 +++++++
 .../apache/tajo/util/metrics/TajoMetrics.java   | 133 ++++++++++
 .../tajo/util/metrics/TajoSystemMetrics.java    | 213 +++++++++++++++
 .../util/metrics/reporter/GangliaReporter.java  | 258 +++++++++++++++++++
 .../reporter/MetricsConsoleReporter.java        |  80 ++++++
 .../MetricsConsoleScheduledReporter.java        |  32 +++
 .../reporter/MetricsFileScheduledReporter.java  |  57 ++++
 .../MetricsStreamScheduledReporter.java         | 179 +++++++++++++
 .../util/metrics/reporter/NullReporter.java     |  31 +++
 .../metrics/reporter/TajoMetricsReporter.java   | 232 +++++++++++++++++
 .../reporter/TajoMetricsScheduledReporter.java  | 206 +++++++++++++++
 .../java/org/apache/tajo/worker/TajoWorker.java |  40 +++
 .../tajo/worker/TajoWorkerManagerService.java   |   1 +
 .../java/org/apache/tajo/worker/TaskRunner.java |   2 +-
 .../apache/tajo/worker/TaskRunnerManager.java   |   4 +
 .../src/main/resources/tajo-metrics.properties  |  75 ++++++
 .../tajo/util/metrics/TestMetricsFilter.java    |  52 ++++
 .../tajo/util/metrics/TestSystemMetrics.java    | 133 ++++++++++
 32 files changed, 2215 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 452cace..2f7d1db 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,8 @@ Release 0.8.0 - unreleased
 
   NEW FEATURES
 
+    TAJO-333: Add metric system to Tajo. (hyoungjunkim via jihoon)
+
     TAJO-413: Implement pi function. (DaeMyung Kang via jihoon)
 
     TAJO-61: Implement Time Datum Type. (DaeMyung Kang via hyunsik)

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
index 0d9bbb0..fb1c29b 100644
--- a/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
+++ b/tajo-common/src/main/java/org/apache/tajo/conf/TajoConf.java
@@ -195,8 +195,11 @@ public class TajoConf extends YarnConfiguration {
     //////////////////////////////////
     // Task Configuration
     TASK_DEFAULT_MEMORY("tajo.task.memory-slot-mb.default", 512),
-    TASK_DEFAULT_DISK("tajo.task.disk-slot.default", 1.0f)
+    TASK_DEFAULT_DISK("tajo.task.disk-slot.default", 1.0f),
     //////////////////////////////////
+
+    // Metrics
+    METRICS_PROPERTY_FILENAME("tajo.metrics.property.file", "tajo-metrics.properties")
     ;
 
     public final String varname;

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/pom.xml
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/pom.xml b/tajo-core/tajo-core-backend/pom.xml
index fca3372..07117a6 100644
--- a/tajo-core/tajo-core-backend/pom.xml
+++ b/tajo-core/tajo-core-backend/pom.xml
@@ -35,6 +35,7 @@
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <antlr4.visitor>true</antlr4.visitor>
     <antlr4.listener>true</antlr4.listener>
+    <metrics.version>3.0.1</metrics.version>
   </properties>
 
   <repositories>
@@ -379,6 +380,21 @@
       <version>6.1.14</version>
     </dependency>
 
+    <dependency>
+      <groupId>com.codahale.metrics</groupId>
+      <artifactId>metrics-core</artifactId>
+      <version>${metrics.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>com.codahale.metrics</groupId>
+      <artifactId>metrics-jvm</artifactId>
+      <version>${metrics.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>info.ganglia.gmetric4j</groupId>
+      <artifactId>gmetric4j</artifactId>
+      <version>1.0.3</version>
+    </dependency>
   </dependencies>
 
   <profiles>

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/log4j.properties
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/log4j.properties b/tajo-core/tajo-core-backend/src/main/java/log4j.properties
index 749124c..15e5778 100644
--- a/tajo-core/tajo-core-backend/src/main/java/log4j.properties
+++ b/tajo-core/tajo-core-backend/src/main/java/log4j.properties
@@ -18,11 +18,13 @@
 
 # log4j configuration used during build and unit tests
 
-log4j.rootLogger=info,stdout
+log4j.rootLogger=info,stdout,EventCounter
 log4j.threshhold=ALL
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} %-5p %c{2} (%F:%M(%L)) - %m%n
 
 log4j.logger.org.apache.hadoop=WARN
-log4j.logger.org.apache.hadoop.conf=ERROR
\ No newline at end of file
+log4j.logger.org.apache.hadoop.conf=ERROR
+
+log4j.appender.EventCounter=org.apache.tajo.util.metrics.TajoLogEventCounter
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
index 2412637..10f42c5 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -92,7 +92,6 @@ public class GlobalEngine extends AbstractService {
       hookManager = new DistributedQueryHookManager();
       hookManager.addHook(new CreateTableHook());
       hookManager.addHook(new InsertHook());
-
     } catch (Throwable t) {
       LOG.error(t.getMessage(), t);
     }
@@ -128,9 +127,12 @@ public class GlobalEngine extends AbstractService {
       LOG.info("hive.query.mode:" + hiveQueryMode);
 
       if (hiveQueryMode) {
+        context.getSystemMetrics().counter("Query", "numHiveMode").inc();
         queryContext.setHiveQueryMode();
       }
 
+      context.getSystemMetrics().counter("Query", "totalQuery").inc();
+
       Expr planningContext = hiveQueryMode ? converter.parse(sql) : analyzer.parse(sql);
 
       LogicalPlan plan = createLogicalPlan(planningContext);
@@ -138,11 +140,13 @@ public class GlobalEngine extends AbstractService {
 
       GetQueryStatusResponse.Builder responseBuilder = GetQueryStatusResponse.newBuilder();
       if (PlannerUtil.checkIfDDLPlan(rootNode)) {
+        context.getSystemMetrics().counter("Query", "numDDLQuery").inc();
         updateQuery(rootNode.getChild());
         responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto());
         responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
         responseBuilder.setState(TajoProtos.QueryState.QUERY_SUCCEEDED);
       } else {
+        context.getSystemMetrics().counter("Query", "numDMLQuery").inc();
         hookManager.doHooks(queryContext, plan);
 
         QueryJobManager queryJobManager = this.context.getQueryJobManager();
@@ -169,6 +173,7 @@ public class GlobalEngine extends AbstractService {
 
       return response;
     } catch (Throwable t) {
+      context.getSystemMetrics().counter("Query", "errorQuery").inc();
       LOG.error("\nStack Trace:\n" + StringUtils.stringifyException(t));
       GetQueryStatusResponse.Builder responseBuilder = GetQueryStatusResponse.newBuilder();
       responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto());

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
index f98cecb..e3d4c01 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TajoMaster.java
@@ -50,6 +50,8 @@ import org.apache.tajo.engine.function.datetime.ToCharTimestamp;
 import org.apache.tajo.engine.function.datetime.ToTimestamp;
 import org.apache.tajo.engine.function.math.*;
 import org.apache.tajo.engine.function.string.*;
+import org.apache.tajo.master.metrics.CatalogMetricsGaugeSet;
+import org.apache.tajo.master.metrics.WorkerResourceMetricsGaugeSet;
 import org.apache.tajo.master.querymaster.QueryJobManager;
 import org.apache.tajo.master.rm.TajoWorkerResourceManager;
 import org.apache.tajo.master.rm.WorkerResourceManager;
@@ -57,6 +59,7 @@ import org.apache.tajo.storage.AbstractStorageManager;
 import org.apache.tajo.storage.StorageManagerFactory;
 import org.apache.tajo.util.CommonTestingUtil;
 import org.apache.tajo.util.NetUtils;
+import org.apache.tajo.util.metrics.TajoSystemMetrics;
 import org.apache.tajo.webapp.QueryExecutorServlet;
 import org.apache.tajo.webapp.StaticHttpServer;
 
@@ -70,6 +73,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 public class TajoMaster extends CompositeService {
+  private static final String METRICS_GROUP_NAME = "tajomaster";
 
   /** Class Logger */
   private static final Log LOG = LogFactory.getLog(TajoMaster.class);
@@ -119,6 +123,8 @@ public class TajoMaster extends CompositeService {
 
   private ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
 
+  private TajoSystemMetrics systemMetrics;
+
   public TajoMaster() throws Exception {
     super(TajoMaster.class.getName());
   }
@@ -180,6 +186,14 @@ public class TajoMaster extends CompositeService {
     LOG.info("Tajo Master is initialized.");
   }
 
+  private void initSystemMetrics() {
+    systemMetrics = new TajoSystemMetrics(systemConf, METRICS_GROUP_NAME, getMasterName());
+    systemMetrics.start();
+
+    systemMetrics.register("resource", new WorkerResourceMetricsGaugeSet(context));
+    systemMetrics.register("catalog", new CatalogMetricsGaugeSet(context));
+  }
+
   private void initResourceManager() throws Exception {
     Class<WorkerResourceManager>  resourceManagerClass = (Class<WorkerResourceManager>)
         systemConf.getClass(ConfVars.RESOURCE_MANAGER_CLASS.varname, TajoWorkerResourceManager.class);
@@ -826,8 +840,10 @@ public class TajoMaster extends CompositeService {
     try {
       writeSystemConf();
     } catch (IOException e) {
-      e.printStackTrace();
+      LOG.error(e.getMessage(), e);
     }
+
+    initSystemMetrics();
   }
 
   private void writeSystemConf() throws IOException {
@@ -862,6 +878,10 @@ public class TajoMaster extends CompositeService {
       }
     }
 
+    if(systemMetrics != null) {
+      systemMetrics.stop();
+    }
+
     super.stop();
     LOG.info("Tajo Master main thread exiting");
   }
@@ -928,6 +948,10 @@ public class TajoMaster extends CompositeService {
     public TajoMasterService getTajoMasterService() {
       return tajoMasterService;
     }
+
+    public TajoSystemMetrics getSystemMetrics() {
+      return systemMetrics;
+    }
   }
 
   String getThreadTaskName(long id, String name) {

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/CatalogMetricsGaugeSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/CatalogMetricsGaugeSet.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/CatalogMetricsGaugeSet.java
new file mode 100644
index 0000000..08fff53
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/CatalogMetricsGaugeSet.java
@@ -0,0 +1,54 @@
+/**
+ * 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.tajo.master.metrics;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricSet;
+import org.apache.tajo.master.TajoMaster;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class CatalogMetricsGaugeSet implements MetricSet {
+  TajoMaster.MasterContext tajoMasterContext;
+  public CatalogMetricsGaugeSet(TajoMaster.MasterContext tajoMasterContext) {
+    this.tajoMasterContext = tajoMasterContext;
+  }
+
+  @Override
+  public Map<String, Metric> getMetrics() {
+    Map<String, Metric> metricsMap = new HashMap<String, Metric>();
+    metricsMap.put("numTables", new Gauge<Integer>() {
+      @Override
+      public Integer getValue() {
+        return tajoMasterContext.getCatalog().getAllTableNames().size();
+      }
+    });
+
+    metricsMap.put("numFunctions", new Gauge<Integer>() {
+      @Override
+      public Integer getValue() {
+        return tajoMasterContext.getCatalog().getFunctions().size();
+      }
+    });
+
+    return metricsMap;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/WorkerResourceMetricsGaugeSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/WorkerResourceMetricsGaugeSet.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/WorkerResourceMetricsGaugeSet.java
new file mode 100644
index 0000000..1924041
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/metrics/WorkerResourceMetricsGaugeSet.java
@@ -0,0 +1,74 @@
+/**
+ * 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.tajo.master.metrics;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricSet;
+import org.apache.tajo.master.TajoMaster;
+import org.apache.tajo.master.rm.WorkerResource;
+import org.apache.tajo.master.rm.WorkerStatus;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class WorkerResourceMetricsGaugeSet implements MetricSet {
+  TajoMaster.MasterContext tajoMasterContext;
+  public WorkerResourceMetricsGaugeSet(TajoMaster.MasterContext tajoMasterContext) {
+    this.tajoMasterContext = tajoMasterContext;
+  }
+
+  @Override
+  public Map<String, Metric> getMetrics() {
+    Map<String, Metric> metricsMap = new HashMap<String, Metric>();
+    metricsMap.put("totalWorkers", new Gauge<Integer>() {
+      @Override
+      public Integer getValue() {
+        return tajoMasterContext.getResourceManager().getWorkers().size();
+      }
+    });
+
+    metricsMap.put("liveWorkers", new Gauge<Integer>() {
+      @Override
+      public Integer getValue() {
+        return getNumWorkers(WorkerStatus.LIVE);
+      }
+    });
+
+    metricsMap.put("deadWorkers", new Gauge<Integer>() {
+      @Override
+      public Integer getValue() {
+        return getNumWorkers(WorkerStatus.DEAD);
+      }
+    });
+
+    return metricsMap;
+  }
+
+  protected int getNumWorkers(WorkerStatus status) {
+    int numWorkers = 0;
+    for(WorkerResource eachWorker: tajoMasterContext.getResourceManager().getWorkers().values()) {
+      if(eachWorker.getWorkerStatus() == status) {
+        numWorkers++;
+      }
+    }
+
+    return numWorkers;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
index 4f0e128..78c417e 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterManagerService.java
@@ -194,6 +194,8 @@ public class QueryMasterManagerService extends CompositeService
                            TajoWorkerProtocol.QueryExecutionRequestProto request,
                            RpcCallback<PrimitiveProtos.BoolProto> done) {
     try {
+      workerContext.getWorkerSystemMetrics().counter("querymaster", "numQuery").inc();
+
       QueryId queryId = new QueryId(request.getQueryId());
       LOG.info("Receive executeQuery request:" + queryId);
       queryMaster.handle(new QueryStartEvent(queryId,
@@ -201,6 +203,7 @@ public class QueryMasterManagerService extends CompositeService
           request.getLogicalPlanJson().getValue()));
       done.run(TajoWorker.TRUE_PROTO);
     } catch (Exception e) {
+      workerContext.getWorkerSystemMetrics().counter("querymaster", "errorQuery").inc();
       LOG.error(e.getMessage(), e);
       done.run(TajoWorker.FALSE_PROTO);
     }

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
index 8cd7d45..828ebfa 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
@@ -51,6 +51,8 @@ import org.apache.tajo.rpc.CallFuture;
 import org.apache.tajo.rpc.NettyClientBase;
 import org.apache.tajo.rpc.RpcConnectionPool;
 import org.apache.tajo.storage.AbstractStorageManager;
+import org.apache.tajo.util.metrics.TajoMetrics;
+import org.apache.tajo.util.metrics.reporter.MetricsConsoleReporter;
 import org.apache.tajo.worker.AbstractResourceAllocator;
 import org.apache.tajo.worker.TajoResourceAllocator;
 import org.apache.tajo.worker.YarnResourceAllocator;
@@ -99,6 +101,8 @@ public class QueryMasterTask extends CompositeService {
 
   private AtomicBoolean stopped = new AtomicBoolean(false);
 
+  private TajoMetrics queryMetrics;
+
   public QueryMasterTask(QueryMaster.QueryMasterContext queryMasterContext,
                          QueryId queryId, QueryContext queryContext, String sql, String logicalPlanJson) {
     super(QueryMasterTask.class.getName());
@@ -136,6 +140,8 @@ public class QueryMasterTask extends CompositeService {
 
       initStagingDir();
 
+      queryMetrics = new TajoMetrics(queryId.toString());
+
       super.init(systemConf);
     } catch (IOException e) {
       LOG.error(e.getMessage(), e);
@@ -186,6 +192,9 @@ public class QueryMasterTask extends CompositeService {
 
     super.stop();
 
+    //TODO change report to tajo master
+    queryMetrics.report(new MetricsConsoleReporter());
+
     LOG.info("Stopped QueryMasterTask:" + queryId);
   }
 
@@ -493,6 +502,10 @@ public class QueryMasterTask extends CompositeService {
     public AbstractResourceAllocator getResourceAllocator() {
       return resourceAllocator;
     }
+
+    public TajoMetrics getQueryMetrics() {
+      return queryMetrics;
+    }
   }
 
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/GroupNameMetricsFilter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/GroupNameMetricsFilter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/GroupNameMetricsFilter.java
new file mode 100644
index 0000000..a273475
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/GroupNameMetricsFilter.java
@@ -0,0 +1,43 @@
+/**
+ * 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.tajo.util.metrics;
+
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricFilter;
+
+public class GroupNameMetricsFilter implements MetricFilter {
+  String groupName;
+
+  public GroupNameMetricsFilter(String groupName) {
+    this.groupName = groupName;
+  }
+  @Override
+  public boolean matches(String name, Metric metric) {
+    if(name != null) {
+      String[] tokens = name.split("\\.");
+      if(groupName.equals(tokens[0])) {
+        return true;
+      } else {
+        return false;
+      }
+    } else {
+      return false;
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/LogEventGaugeSet.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/LogEventGaugeSet.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/LogEventGaugeSet.java
new file mode 100644
index 0000000..6e130ff
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/LogEventGaugeSet.java
@@ -0,0 +1,64 @@
+/**
+ * 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.tajo.util.metrics;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricSet;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class LogEventGaugeSet implements MetricSet {
+
+  @Override
+  public Map<String, Metric> getMetrics() {
+    final Map<String, Metric> gauges = new HashMap<String, Metric>();
+
+    gauges.put("Fatal", new Gauge<Long>() {
+      @Override
+      public Long getValue() {
+        return TajoLogEventCounter.getFatal();
+      }
+    });
+
+    gauges.put("Error", new Gauge<Long>() {
+      @Override
+      public Long getValue() {
+        return TajoLogEventCounter.getError();
+      }
+    });
+
+    gauges.put("Warn", new Gauge<Long>() {
+      @Override
+      public Long getValue() {
+        return TajoLogEventCounter.getWarn();
+      }
+    });
+
+    gauges.put("Info", new Gauge<Long>() {
+      @Override
+      public Long getValue() {
+        return TajoLogEventCounter.getInfo();
+      }
+    });
+
+    return gauges;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/MetricsFilterList.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/MetricsFilterList.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/MetricsFilterList.java
new file mode 100644
index 0000000..b2fc6e4
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/MetricsFilterList.java
@@ -0,0 +1,43 @@
+/**
+ * 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.tajo.util.metrics;
+
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricFilter;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class MetricsFilterList implements MetricFilter {
+  List<MetricFilter> filters = new ArrayList<MetricFilter>();
+
+  public void addMetricFilter(MetricFilter filter) {
+    filters.add(filter);
+  }
+
+  @Override
+  public boolean matches(String name, Metric metric) {
+    for (MetricFilter eachFilter: filters) {
+      if (!eachFilter.matches(name, metric)) {
+        return false;
+      }
+    }
+    return true;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/RegexpMetricsFilter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/RegexpMetricsFilter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/RegexpMetricsFilter.java
new file mode 100644
index 0000000..4faa3e7
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/RegexpMetricsFilter.java
@@ -0,0 +1,51 @@
+/**
+ * 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.tajo.util.metrics;
+
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricFilter;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.regex.Pattern;
+
+public class RegexpMetricsFilter implements MetricFilter {
+  List<Pattern> filterPatterns = new ArrayList<Pattern>();
+
+  public RegexpMetricsFilter(Collection<String> filterExpressions) {
+    for(String eachExpression: filterExpressions) {
+      filterPatterns.add(Pattern.compile(eachExpression));
+    }
+  }
+
+  @Override
+  public boolean matches(String name, Metric metric) {
+    if(filterPatterns.isEmpty()) {
+      return true;
+    }
+
+    for(Pattern eachPattern: filterPatterns) {
+      if(eachPattern.matcher(name).find()) {
+        return true;
+      }
+    }
+    return false;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoLogEventCounter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoLogEventCounter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoLogEventCounter.java
new file mode 100644
index 0000000..3e44b02
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoLogEventCounter.java
@@ -0,0 +1,86 @@
+/**
+ * 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.tajo.util.metrics;
+
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.Level;
+import org.apache.log4j.spi.LoggingEvent;
+
+public class TajoLogEventCounter extends AppenderSkeleton {
+  private static final int FATAL = 0;
+  private static final int ERROR = 1;
+  private static final int WARN = 2;
+  private static final int INFO = 3;
+
+  private static class EventCounts {
+
+    private final long[] counts = {0, 0, 0, 0};
+
+    private synchronized void incr(int i) {
+      ++counts[i];
+    }
+
+    private synchronized long get(int i) {
+      return counts[i];
+    }
+  }
+
+  private static EventCounts counts = new EventCounts();
+
+  public static long getFatal() {
+    return counts.get(FATAL);
+  }
+
+  public static long getError() {
+    return counts.get(ERROR);
+  }
+
+  public static long getWarn() {
+    return counts.get(WARN);
+  }
+
+  public static long getInfo() {
+    return counts.get(INFO);
+  }
+
+  @Override
+  public void append(LoggingEvent event) {
+    Level level = event.getLevel();
+    String levelStr = level.toString();
+
+    if (level == Level.INFO || "INFO".equalsIgnoreCase(levelStr)) {
+      counts.incr(INFO);
+    } else if (level == Level.WARN || "WARN".equalsIgnoreCase(levelStr)) {
+      counts.incr(WARN);
+    } else if (level == Level.ERROR || "ERROR".equalsIgnoreCase(levelStr)) {
+      counts.incr(ERROR);
+    } else if (level == Level.FATAL || "FATAL".equalsIgnoreCase(levelStr)) {
+      counts.incr(FATAL);
+    }
+  }
+
+  @Override
+  public void close() {
+  }
+
+  @Override
+  public boolean requiresLayout() {
+    return false;
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoMetrics.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoMetrics.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoMetrics.java
new file mode 100644
index 0000000..0e378b2
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoMetrics.java
@@ -0,0 +1,133 @@
+/**
+ * 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.tajo.util.metrics;
+
+import com.codahale.metrics.*;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.util.metrics.reporter.TajoMetricsReporter;
+
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class TajoMetrics {
+  private static final Log LOG = LogFactory.getLog(TajoMetrics.class);
+
+  protected MetricRegistry metricRegistry;
+  protected AtomicBoolean stop = new AtomicBoolean(false);
+  protected String metricsGroupName;
+
+  public TajoMetrics(String metricsGroupName) {
+    this.metricsGroupName = metricsGroupName;
+    this.metricRegistry = new MetricRegistry();
+  }
+
+  public void stop() {
+    stop.set(true);
+  }
+
+  public MetricRegistry getRegistry() {
+    return metricRegistry;
+  }
+
+  public void report(TajoMetricsReporter reporter) {
+    try {
+      reporter.report(metricRegistry.getGauges(),
+          metricRegistry.getCounters(),
+          metricRegistry.getHistograms(),
+          metricRegistry.getMeters(),
+          metricRegistry.getTimers());
+    } catch (Exception e) {
+      if(LOG.isDebugEnabled()) {
+        LOG.warn("Metric report error:" + e.getMessage(), e);
+      } else {
+        LOG.warn("Metric report error:" + e.getMessage());
+      }
+    }
+  }
+
+  public Map<String, Metric> getMetrics() {
+    return metricRegistry.getMetrics();
+  }
+
+  public SortedMap<String, Gauge> getGuageMetrics(MetricFilter filter) {
+    if(filter == null) {
+      filter = MetricFilter.ALL;
+    }
+    return metricRegistry.getGauges(filter);
+  }
+
+  public SortedMap<String, Counter> getCounterMetrics(MetricFilter filter) {
+    if(filter == null) {
+      filter = MetricFilter.ALL;
+    }
+    return metricRegistry.getCounters(filter);
+  }
+
+  public SortedMap<String, Histogram> getHistogramMetrics(MetricFilter filter) {
+    if(filter == null) {
+      filter = MetricFilter.ALL;
+    }
+    return metricRegistry.getHistograms(filter);
+  }
+
+  public SortedMap<String, Meter> getMeterMetrics(MetricFilter filter) {
+    if(filter == null) {
+      filter = MetricFilter.ALL;
+    }
+    return metricRegistry.getMeters(filter);
+  }
+
+  public SortedMap<String, Timer> getTimerMetrics(MetricFilter filter) {
+    if(filter == null) {
+      filter = MetricFilter.ALL;
+    }
+    return metricRegistry.getTimers(filter);
+  }
+
+  public void register(String contextName, MetricSet metricSet) {
+    metricRegistry.register(MetricRegistry.name(metricsGroupName, contextName), metricSet);
+  }
+
+  public void register(String contextName, String itemName, Gauge gauge) {
+    metricRegistry.register(makeMetricsName(metricsGroupName, contextName, itemName), gauge);
+  }
+
+  public Counter counter(String contextName, String itemName) {
+    return metricRegistry.counter(makeMetricsName(metricsGroupName, contextName, itemName));
+  }
+
+  public Histogram histogram(String contextName, String itemName) {
+    return metricRegistry.histogram(makeMetricsName(metricsGroupName, contextName, itemName));
+  }
+
+  public Meter meter(String contextName, String itemName) {
+    return metricRegistry.meter(makeMetricsName(metricsGroupName, contextName, itemName));
+  }
+
+  public Timer timer(String contextName, String itemName) {
+    return metricRegistry.timer(makeMetricsName(metricsGroupName, contextName, itemName));
+  }
+
+  public static String makeMetricsName(String metricsGroupName, String contextName, String itemName) {
+    return MetricRegistry.name(metricsGroupName, contextName, itemName);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoSystemMetrics.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoSystemMetrics.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoSystemMetrics.java
new file mode 100644
index 0000000..4192ca0
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/TajoSystemMetrics.java
@@ -0,0 +1,213 @@
+/**
+ * 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.tajo.util.metrics;
+
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.jvm.FileDescriptorRatioGauge;
+import com.codahale.metrics.jvm.GarbageCollectorMetricSet;
+import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
+import com.codahale.metrics.jvm.ThreadStatesGaugeSet;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.commons.configuration.event.ConfigurationEvent;
+import org.apache.commons.configuration.event.ConfigurationListener;
+import org.apache.commons.configuration.reloading.FileChangedReloadingStrategy;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.util.metrics.reporter.TajoMetricsScheduledReporter;
+
+import java.util.*;
+
+public class TajoSystemMetrics extends TajoMetrics {
+  private static final Log LOG = LogFactory.getLog(TajoSystemMetrics.class);
+
+  private PropertiesConfiguration metricsProps;
+
+  private Thread propertyChangeChecker;
+
+  private String hostAndPort;
+
+  private List<TajoMetricsScheduledReporter> metricsReporters = new ArrayList<TajoMetricsScheduledReporter>();
+
+  private boolean inited = false;
+
+  private String metricsPropertyFileName;
+
+  public TajoSystemMetrics(TajoConf tajoConf, String metricsGroupName, String hostAndPort) {
+    super(metricsGroupName);
+
+    this.hostAndPort = hostAndPort;
+    try {
+      this.metricsPropertyFileName = tajoConf.getVar(TajoConf.ConfVars.METRICS_PROPERTY_FILENAME);
+      this.metricsProps = new PropertiesConfiguration(metricsPropertyFileName);
+      this.metricsProps.addConfigurationListener(new MetricsReloadListener());
+      FileChangedReloadingStrategy reloadingStrategy = new FileChangedReloadingStrategy();
+      reloadingStrategy.setRefreshDelay(5 * 1000);
+      this.metricsProps.setReloadingStrategy(reloadingStrategy);
+    } catch (ConfigurationException e) {
+      LOG.warn(e.getMessage(), e);
+    }
+
+    //PropertiesConfiguration fire configurationChanged after getXXX()
+    //So neeaded calling getXXX periodically
+    propertyChangeChecker = new Thread() {
+      public void run() {
+        while(!stop.get()) {
+          String value = metricsProps.getString("reporter.file");
+          try {
+            Thread.sleep(10 * 1000);
+          } catch (InterruptedException e) {
+          }
+        }
+      }
+    };
+
+    propertyChangeChecker.start();
+  }
+
+  public Collection<TajoMetricsScheduledReporter> getMetricsReporters() {
+    synchronized (metricsReporters) {
+      return Collections.unmodifiableCollection(metricsReporters);
+    }
+  }
+
+  @Override
+  public void stop() {
+    super.stop();
+    if(propertyChangeChecker != null) {
+      propertyChangeChecker.interrupt();
+    }
+    stopAndClearReporter();
+  }
+
+  protected void stopAndClearReporter() {
+    synchronized(metricsReporters) {
+      for(TajoMetricsScheduledReporter eachReporter: metricsReporters) {
+        eachReporter.close();
+      }
+
+      metricsReporters.clear();
+    }
+  }
+
+  public void start() {
+    setMetricsReporter(metricsGroupName);
+
+    String jvmMetricsName = metricsGroupName + "-jvm";
+    setMetricsReporter(jvmMetricsName);
+
+    if(!inited) {
+      metricRegistry.register(MetricRegistry.name(jvmMetricsName, "Heap"), new MemoryUsageGaugeSet());
+      metricRegistry.register(MetricRegistry.name(jvmMetricsName, "File"), new FileDescriptorRatioGauge());
+      metricRegistry.register(MetricRegistry.name(jvmMetricsName, "GC"), new GarbageCollectorMetricSet());
+      metricRegistry.register(MetricRegistry.name(jvmMetricsName, "Thread"), new ThreadStatesGaugeSet());
+      metricRegistry.register(MetricRegistry.name(jvmMetricsName, "Log"), new LogEventGaugeSet());
+    }
+    inited = true;
+  }
+
+  private void setMetricsReporter(String groupName) {
+    //reporter name -> class name
+    Map<String, String> reporters = new HashMap<String, String>();
+
+    List<String> reporterNames = metricsProps.getList(groupName + ".reporters");
+    if(reporterNames.isEmpty()) {
+      LOG.warn("No property " + groupName + ".reporters in " + metricsPropertyFileName);
+      return;
+    }
+
+    Map<String, String> allReporterProperties = new HashMap<String, String>();
+
+    Iterator<String> keys = metricsProps.getKeys();
+    while (keys.hasNext()) {
+      String key = keys.next();
+      String value = metricsProps.getString(key);
+      if(key.indexOf("reporter.") == 0) {
+        String[] tokens = key.split("\\.");
+        if(tokens.length == 2) {
+          reporters.put(tokens[1], value);
+        }
+      } else if(key.indexOf(groupName + ".") == 0) {
+        String[] tokens = key.split("\\.");
+        if(tokens.length > 2) {
+          allReporterProperties.put(key, value);
+        }
+      }
+    }
+
+    synchronized(metricsReporters) {
+      for(String eachReporterName: reporterNames) {
+        if("null".equals(eachReporterName)) {
+          continue;
+        }
+        String reporterClass = reporters.get(eachReporterName);
+        if(reporterClass == null) {
+          LOG.warn("No metrics reporter definition[" + eachReporterName + "] in " + metricsPropertyFileName);
+          continue;
+        }
+
+        Map<String, String> eachMetricsReporterProperties = findMetircsProperties(allReporterProperties,
+            groupName + "." + eachReporterName);
+
+        try {
+          Object reporterObject = Class.forName(reporterClass).newInstance();
+          if(!(reporterObject instanceof TajoMetricsScheduledReporter)) {
+            LOG.warn(reporterClass + " is not subclass of " + TajoMetricsScheduledReporter.class.getCanonicalName());
+            continue;
+          }
+          TajoMetricsScheduledReporter reporter = (TajoMetricsScheduledReporter)reporterObject;
+          reporter.init(metricRegistry, groupName, hostAndPort, eachMetricsReporterProperties);
+          reporter.start();
+
+          metricsReporters.add(reporter);
+          LOG.info("Started metrics reporter " + reporter.getClass().getCanonicalName() + " for " + groupName);
+        } catch (ClassNotFoundException e) {
+          LOG.warn("No metrics reporter class[" + eachReporterName + "], required class= " + reporterClass);
+          continue;
+        } catch (Exception e) {
+          LOG.warn("Can't initiate metrics reporter class[" + eachReporterName + "]" + e.getMessage() , e);
+          continue;
+        }
+      }
+    }
+  }
+
+  private Map<String, String> findMetircsProperties(Map<String, String> allReporterProperties, String findKey) {
+    Map<String, String> metricsProperties = new HashMap<String, String>();
+
+    for (Map.Entry<String, String> entry: allReporterProperties.entrySet()) {
+      String eachKey = entry.getKey();
+      if (eachKey.indexOf(findKey) == 0) {
+        metricsProperties.put(eachKey, entry.getValue());
+      }
+    }
+    return metricsProperties;
+  }
+
+  class MetricsReloadListener implements ConfigurationListener {
+    @Override
+    public synchronized void configurationChanged(ConfigurationEvent event) {
+      if (!event.isBeforeUpdate()) {
+        stopAndClearReporter();
+        start();
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/GangliaReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/GangliaReporter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/GangliaReporter.java
new file mode 100644
index 0000000..b9acf0e
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/GangliaReporter.java
@@ -0,0 +1,258 @@
+/**
+ * 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.tajo.util.metrics.reporter;
+
+import com.codahale.metrics.*;
+import info.ganglia.gmetric4j.gmetric.GMetric;
+import info.ganglia.gmetric4j.gmetric.GMetricSlope;
+import info.ganglia.gmetric4j.gmetric.GMetricType;
+import info.ganglia.gmetric4j.gmetric.GangliaException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.SortedMap;
+
+import static com.codahale.metrics.MetricRegistry.name;
+
+public class GangliaReporter extends TajoMetricsScheduledReporter {
+  private static final Logger LOG = LoggerFactory.getLogger(GangliaReporter.class);
+  public static final String REPORTER_NAME = "ganglia";
+
+  private GMetric ganglia;
+  private String prefix;
+  private int tMax = 60;
+  private int dMax = 0;
+
+  @Override
+  protected String getReporterName() {
+    return REPORTER_NAME;
+  }
+
+  @Override
+  protected void afterInit() {
+    String server = metricsProperties.get(metricsPropertyKey + "server");
+    String port = metricsProperties.get(metricsPropertyKey + "port");
+
+    if(server == null || server.isEmpty()) {
+      LOG.warn("No " + metricsPropertyKey + "server property in tajo-metrics.properties");
+      return;
+    }
+
+    if(port == null || port.isEmpty()) {
+      LOG.warn("No " + metricsPropertyKey + "port property in tajo-metrics.properties");
+      return;
+    }
+
+    try {
+      ganglia = new GMetric(server, Integer.parseInt(port), GMetric.UDPAddressingMode.MULTICAST, 1);
+    } catch (Exception e) {
+      LOG.warn(e.getMessage(), e);
+    }
+  }
+
+  public void setPrefix(String prefix) {
+    this.prefix = prefix;
+  }
+
+  public void settMax(int tMax) {
+    this.tMax = tMax;
+  }
+
+  public void setdMax(int dMax) {
+    this.dMax = dMax;
+  }
+
+  @Override
+  public void report(SortedMap<String, Gauge> gauges,
+                     SortedMap<String, Counter> counters,
+                     SortedMap<String, Histogram> histograms,
+                     SortedMap<String, Meter> meters,
+                     SortedMap<String, Timer> timers) {
+    for (Map.Entry<String, Gauge> entry : gauges.entrySet()) {
+      reportGauge(entry.getKey(), entry.getValue());
+    }
+
+    for (Map.Entry<String, Counter> entry : counters.entrySet()) {
+      reportCounter(entry.getKey(), entry.getValue());
+    }
+
+    for (Map.Entry<String, Histogram> entry : histograms.entrySet()) {
+      reportHistogram(entry.getKey(), entry.getValue());
+    }
+
+    for (Map.Entry<String, Meter> entry : meters.entrySet()) {
+      reportMeter(entry.getKey(), entry.getValue());
+    }
+
+    for (Map.Entry<String, Timer> entry : timers.entrySet()) {
+      reportTimer(entry.getKey(), entry.getValue());
+    }
+  }
+
+  private void reportTimer(String name, Timer timer) {
+    final String group = group(name);
+    try {
+      final Snapshot snapshot = timer.getSnapshot();
+
+      announce(prefix(name, "max"), group, convertDuration(snapshot.getMax()), getDurationUnit());
+      announce(prefix(name, "mean"), group, convertDuration(snapshot.getMean()), getDurationUnit());
+      announce(prefix(name, "min"), group, convertDuration(snapshot.getMin()), getDurationUnit());
+      announce(prefix(name, "stddev"), group, convertDuration(snapshot.getStdDev()), getDurationUnit());
+
+      announce(prefix(name, "p50"), group, convertDuration(snapshot.getMedian()), getDurationUnit());
+      announce(prefix(name, "p75"),
+          group,
+          convertDuration(snapshot.get75thPercentile()),
+          getDurationUnit());
+      announce(prefix(name, "p95"),
+          group,
+          convertDuration(snapshot.get95thPercentile()),
+          getDurationUnit());
+      announce(prefix(name, "p98"),
+          group,
+          convertDuration(snapshot.get98thPercentile()),
+          getDurationUnit());
+      announce(prefix(name, "p99"),
+          group,
+          convertDuration(snapshot.get99thPercentile()),
+          getDurationUnit());
+      announce(prefix(name, "p999"),
+          group,
+          convertDuration(snapshot.get999thPercentile()),
+          getDurationUnit());
+
+      reportMetered(name, timer, group, "calls");
+    } catch (GangliaException e) {
+      LOG.warn("Unable to report timer {}", name, e);
+    }
+  }
+
+  private void reportMeter(String name, Meter meter) {
+    final String group = group(name);
+    try {
+      reportMetered(name, meter, group, "events");
+    } catch (GangliaException e) {
+      LOG.warn("Unable to report meter {}", name, e);
+    }
+  }
+
+  private void reportMetered(String name, Metered meter, String group, String eventName) throws GangliaException {
+    final String unit = eventName + '/' + getRateUnit();
+    announce(prefix(name, "count"), group, meter.getCount(), eventName);
+    announce(prefix(name, "m1_rate"), group, convertRate(meter.getOneMinuteRate()), unit);
+    announce(prefix(name, "m5_rate"), group, convertRate(meter.getFiveMinuteRate()), unit);
+    announce(prefix(name, "m15_rate"), group, convertRate(meter.getFifteenMinuteRate()), unit);
+    announce(prefix(name, "mean_rate"), group, convertRate(meter.getMeanRate()), unit);
+  }
+
+  private void reportHistogram(String name, Histogram histogram) {
+    final String group = group(name);
+    try {
+      final Snapshot snapshot = histogram.getSnapshot();
+
+      announce(prefix(name, "count"), group, histogram.getCount(), "");
+      announce(prefix(name, "max"), group, snapshot.getMax(), "");
+      announce(prefix(name, "mean"), group, snapshot.getMean(), "");
+      announce(prefix(name, "min"), group, snapshot.getMin(), "");
+      announce(prefix(name, "stddev"), group, snapshot.getStdDev(), "");
+      announce(prefix(name, "p50"), group, snapshot.getMedian(), "");
+      announce(prefix(name, "p75"), group, snapshot.get75thPercentile(), "");
+      announce(prefix(name, "p95"), group, snapshot.get95thPercentile(), "");
+      announce(prefix(name, "p98"), group, snapshot.get98thPercentile(), "");
+      announce(prefix(name, "p99"), group, snapshot.get99thPercentile(), "");
+      announce(prefix(name, "p999"), group, snapshot.get999thPercentile(), "");
+    } catch (GangliaException e) {
+      LOG.warn("Unable to report histogram {}", name, e);
+    }
+  }
+
+  private void reportCounter(String name, Counter counter) {
+    final String group = group(name);
+    try {
+      announce(prefix(name, "count"), group, counter.getCount(), "");
+    } catch (GangliaException e) {
+      LOG.warn("Unable to report counter {}", name, e);
+    }
+  }
+
+  private void reportGauge(String name, Gauge gauge) {
+    final String group = group(name);
+    final Object obj = gauge.getValue();
+
+    try {
+      ganglia.announce(name(prefix, name), String.valueOf(obj), detectType(obj), "",
+          GMetricSlope.BOTH, tMax, dMax, group);
+    } catch (GangliaException e) {
+      LOG.warn("Unable to report gauge {}", name, e);
+    }
+  }
+
+  private void announce(String name, String group, double value, String units) throws GangliaException {
+    ganglia.announce(name,
+        Double.toString(value),
+        GMetricType.DOUBLE,
+        units,
+        GMetricSlope.BOTH,
+        tMax,
+        dMax,
+        group);
+  }
+
+  private void announce(String name, String group, long value, String units) throws GangliaException {
+    final String v = Long.toString(value);
+    ganglia.announce(name,
+        v,
+        GMetricType.DOUBLE,
+        units,
+        GMetricSlope.BOTH,
+        tMax,
+        dMax,
+        group);
+  }
+
+  private GMetricType detectType(Object o) {
+    if (o instanceof Float) {
+      return GMetricType.FLOAT;
+    } else if (o instanceof Double) {
+      return GMetricType.DOUBLE;
+    } else if (o instanceof Byte) {
+      return GMetricType.INT8;
+    } else if (o instanceof Short) {
+      return GMetricType.INT16;
+    } else if (o instanceof Integer) {
+      return GMetricType.INT32;
+    } else if (o instanceof Long) {
+      return GMetricType.DOUBLE;
+    }
+    return GMetricType.STRING;
+  }
+
+  private String group(String name) {
+    String[] tokens = name.split("\\.");
+    if(tokens.length < 3) {
+      return "";
+    }
+    return tokens[0] + "." + tokens[1];
+  }
+
+  private String prefix(String name, String n) {
+    return name(prefix, name, n);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleReporter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleReporter.java
new file mode 100644
index 0000000..80b77f1
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleReporter.java
@@ -0,0 +1,80 @@
+/**
+ * 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.tajo.util.metrics.reporter;
+
+import com.codahale.metrics.*;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.concurrent.TimeUnit;
+
+public class MetricsConsoleReporter extends TajoMetricsReporter {
+  @Override
+  public void report(SortedMap<String, Gauge> gauges,
+                     SortedMap<String, Counter> counters,
+                     SortedMap<String, Histogram> histograms,
+                     SortedMap<String, Meter> meters,
+                     SortedMap<String, Timer> timers) {
+    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+    final String dateTime = dateFormat.format(new Date());
+    double rateFactor = TimeUnit.SECONDS.toSeconds(1);
+
+    if (!gauges.isEmpty()) {
+      Map<String, Map<String, Gauge>> gaugeGroups = findMetricsItemGroup(gauges);
+
+      for(Map.Entry<String, Map<String, Gauge>> eachGroup: gaugeGroups.entrySet()) {
+        System.out.println(gaugeGroupToString(dateTime, null, rateFactor, eachGroup.getKey(), eachGroup.getValue()));
+      }
+    }
+
+    if (!counters.isEmpty()) {
+      Map<String, Map<String, Counter>> counterGroups = findMetricsItemGroup(counters);
+
+      for(Map.Entry<String, Map<String, Counter>> eachGroup: counterGroups.entrySet()) {
+        System.out.println(counterGroupToString(dateTime, null, rateFactor, eachGroup.getKey(), eachGroup.getValue()));
+      }
+    }
+
+    if (!histograms.isEmpty()) {
+      Map<String, Map<String, Histogram>> histogramGroups = findMetricsItemGroup(histograms);
+
+      for(Map.Entry<String, Map<String, Histogram>> eachGroup: histogramGroups.entrySet()) {
+        System.out.println(histogramGroupToString(dateTime, null, rateFactor, eachGroup.getKey(), eachGroup.getValue()));
+      }
+    }
+
+    if (!meters.isEmpty()) {
+      Map<String, Map<String, Meter>> meterGroups = findMetricsItemGroup(meters);
+
+      for(Map.Entry<String, Map<String, Meter>> eachGroup: meterGroups.entrySet()) {
+        System.out.println(meterGroupToString(dateTime, null, rateFactor, eachGroup.getKey(), eachGroup.getValue()));
+      }
+    }
+
+    if (!timers.isEmpty()) {
+      Map<String, Map<String, Timer>> timerGroups = findMetricsItemGroup(timers);
+
+      for(Map.Entry<String, Map<String, Timer>> eachGroup: timerGroups.entrySet()) {
+        System.out.println(timerGroupToString(dateTime, null, rateFactor, eachGroup.getKey(), eachGroup.getValue()));
+      }
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleScheduledReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleScheduledReporter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleScheduledReporter.java
new file mode 100644
index 0000000..286ef8d
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsConsoleScheduledReporter.java
@@ -0,0 +1,32 @@
+/**
+ * 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.tajo.util.metrics.reporter;
+
+public class MetricsConsoleScheduledReporter extends MetricsStreamScheduledReporter {
+  public static final String REPORTER_NAME = "console";
+  @Override
+  protected String getReporterName() {
+    return REPORTER_NAME;
+  }
+
+  @Override
+  protected void afterInit() {
+    setOutput(System.out);
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java
new file mode 100644
index 0000000..35dd6f1
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsFileScheduledReporter.java
@@ -0,0 +1,57 @@
+/**
+ * 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.tajo.util.metrics.reporter;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+
+public class MetricsFileScheduledReporter extends MetricsStreamScheduledReporter {
+  private static final Log LOG = LogFactory.getLog(MetricsFileScheduledReporter.class);
+  public static final String REPORTER_NAME = "file";
+
+  protected String getReporterName() {
+    return REPORTER_NAME;
+  }
+
+  @Override
+  protected void afterInit() {
+    String fileName = metricsProperties.get(metricsPropertyKey + "filename");
+    if(fileName == null) {
+      LOG.warn("No " + metricsPropertyKey + "filename property in tajo-metrics.properties");
+      return;
+    }
+    try {
+      File file = new File(fileName);
+      File parentFile = file.getParentFile();
+      if(parentFile != null && !parentFile.exists()) {
+        if(!parentFile.mkdirs()) {
+          LOG.warn("Can't create dir for tajo metrics:" + parentFile.getAbsolutePath());
+        }
+      }
+      this.setOutput(new FileOutputStream(fileName, true));
+      this.setDateFormat(null);
+    } catch (FileNotFoundException e) {
+      LOG.warn("Can't open metrics file:" + fileName);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java
new file mode 100644
index 0000000..4fbefd7
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/MetricsStreamScheduledReporter.java
@@ -0,0 +1,179 @@
+/**
+ * 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.tajo.util.metrics.reporter;
+
+import com.codahale.metrics.*;
+import com.codahale.metrics.Timer;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+public abstract class MetricsStreamScheduledReporter extends TajoMetricsScheduledReporter {
+  private static final Log LOG = LogFactory.getLog(MetricsStreamScheduledReporter.class);
+
+  protected OutputStream output;
+  protected Locale locale;
+  protected Clock clock;
+  protected TimeZone timeZone;
+  protected MetricFilter filter;
+  protected DateFormat dateFormat;
+
+  private final byte[] NEW_LINE = "\n".getBytes();
+
+  public MetricsStreamScheduledReporter() {
+    dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+    clock = Clock.defaultClock();
+  }
+
+  public void setOutput(OutputStream output) {
+    this.output = output;
+  }
+
+  public void setLocale(Locale locale) {
+    this.locale = locale;
+  }
+
+  public void setClock(Clock clock) {
+    this.clock = clock;
+  }
+
+  public void setTimeZone(TimeZone timeZone) {
+    this.dateFormat.setTimeZone(timeZone);
+    this.timeZone = timeZone;
+  }
+
+  public void setDateFormat(DateFormat dateFormat) {
+    this.dateFormat = dateFormat;
+  }
+
+  @Override
+  public void report(SortedMap<String, Gauge> gauges,
+                     SortedMap<String, Counter> counters,
+                     SortedMap<String, Histogram> histograms,
+                     SortedMap<String, Meter> meters,
+                     SortedMap<String, Timer> timers) {
+    final String dateTime = dateFormat == null ? "" + clock.getTime() : dateFormat.format(new Date(clock.getTime()));
+
+    if (!gauges.isEmpty()) {
+      Map<String, Map<String, Gauge>> gaugeGroups = findMetricsItemGroup(gauges);
+
+      for(Map.Entry<String, Map<String, Gauge>> eachGroup: gaugeGroups.entrySet()) {
+        printGaugeGroup(dateTime, eachGroup.getKey(), eachGroup.getValue());
+      }
+    }
+
+    if (!counters.isEmpty()) {
+      Map<String, Map<String, Counter>> counterGroups = findMetricsItemGroup(counters);
+
+      for(Map.Entry<String, Map<String, Counter>> eachGroup: counterGroups.entrySet()) {
+        printCounterGroup(dateTime, eachGroup.getKey(), eachGroup.getValue());
+      }
+    }
+
+    if (!histograms.isEmpty()) {
+      Map<String, Map<String, Histogram>> histogramGroups = findMetricsItemGroup(histograms);
+
+      for(Map.Entry<String, Map<String, Histogram>> eachGroup: histogramGroups.entrySet()) {
+        printHistogramGroup(dateTime, eachGroup.getKey(), eachGroup.getValue());
+      }
+    }
+
+    if (!meters.isEmpty()) {
+      Map<String, Map<String, Meter>> meterGroups = findMetricsItemGroup(meters);
+
+      for(Map.Entry<String, Map<String, Meter>> eachGroup: meterGroups.entrySet()) {
+        printMeterGroup(dateTime, eachGroup.getKey(), eachGroup.getValue());
+      }
+    }
+
+    if (!timers.isEmpty()) {
+      Map<String, Map<String, Timer>> timerGroups = findMetricsItemGroup(timers);
+
+      for(Map.Entry<String, Map<String, Timer>> eachGroup: timerGroups.entrySet()) {
+        printTimerGroup(dateTime, eachGroup.getKey(), eachGroup.getValue());
+      }
+    }
+    try {
+      output.flush();
+    } catch (IOException e) {
+    }
+  }
+
+  private void printMeterGroup(String dateTime, String groupName, Map<String, Meter> meters) {
+    try {
+      output.write(meterGroupToString(dateTime, hostAndPort, rateFactor, groupName, meters).getBytes());
+      output.write(NEW_LINE);
+    } catch (Exception e) {
+      LOG.warn(e.getMessage(), e);
+    }
+  }
+
+  private void printCounterGroup(String dateTime, String groupName, Map<String, Counter> counters) {
+    try {
+      output.write(counterGroupToString(dateTime, hostAndPort, rateFactor, groupName, counters).getBytes());
+      output.write(NEW_LINE);
+    } catch (Exception e) {
+      LOG.warn(e.getMessage(), e);
+    }
+  }
+
+  private void printGaugeGroup(String dateTime, String groupName, Map<String, Gauge> gauges) {
+    try {
+      output.write(gaugeGroupToString(dateTime, hostAndPort, rateFactor, groupName, gauges).getBytes());
+      output.write(NEW_LINE);
+    } catch (Exception e) {
+      LOG.warn(e.getMessage(), e);
+    }
+  }
+
+  private void printHistogramGroup(String dateTime, String groupName, Map<String, Histogram> histograms) {
+    try {
+      output.write(histogramGroupToString(dateTime, hostAndPort, rateFactor, groupName, histograms).getBytes());
+      output.write(NEW_LINE);
+    } catch (Exception e) {
+      LOG.warn(e.getMessage(), e);
+    }
+  }
+
+  private void printTimerGroup(String dateTime, String groupName, Map<String, Timer> timers) {
+    try {
+      output.write(timerGroupToString(dateTime, hostAndPort, rateFactor, groupName, timers).getBytes());
+      output.write(NEW_LINE);
+    } catch (Exception e) {
+      LOG.warn(e.getMessage(), e);
+    }
+  }
+
+  @Override
+  public void close() {
+    if(output != null) {
+      try {
+        output.close();
+      } catch (IOException e) {
+      }
+    }
+
+    super.close();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/NullReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/NullReporter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/NullReporter.java
new file mode 100644
index 0000000..9dc1755
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/NullReporter.java
@@ -0,0 +1,31 @@
+/**
+ * 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.tajo.util.metrics.reporter;
+
+import com.codahale.metrics.*;
+
+import java.util.SortedMap;
+
+public class NullReporter extends TajoMetricsReporter {
+  @Override
+  public void report(SortedMap<String, Gauge> gauges, SortedMap<String, Counter> counters,
+                     SortedMap<String, Histogram> histograms, SortedMap<String, Meter> meters,
+                     SortedMap<String, Timer> timers) {
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/62c49c05/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java
new file mode 100644
index 0000000..a32a913
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/util/metrics/reporter/TajoMetricsReporter.java
@@ -0,0 +1,232 @@
+/**
+ * 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.tajo.util.metrics.reporter;
+
+import com.codahale.metrics.*;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedMap;
+
+public abstract class TajoMetricsReporter {
+  public abstract void report(SortedMap<String, Gauge> gauges,
+                              SortedMap<String, Counter> counters,
+                              SortedMap<String, Histogram> histograms,
+                              SortedMap<String, Meter> meters,
+                              SortedMap<String, Timer> timers);
+
+  public <T> Map<String, Map<String, T>> findMetricsItemGroup(SortedMap<String, T> metricsMap) {
+    Map<String, Map<String, T>> metricsGroup = new HashMap<String, Map<String, T>>();
+
+    String previousGroup = null;
+    Map<String, T> groupItems = new HashMap<String, T>();
+
+    for (Map.Entry<String, T> entry : metricsMap.entrySet()) {
+      String key = entry.getKey();
+      String[] keyTokens = key.split("\\.");
+
+      String groupName = null;
+      String itemName = null;
+
+      if (keyTokens.length > 2) {
+        groupName = keyTokens[0] + "." + keyTokens[1];
+        itemName = "";
+        String prefix = "";
+        for (int i = 2; i < keyTokens.length; i++) {
+          itemName += prefix + keyTokens[i];
+          prefix = ".";
+        }
+      } else {
+        groupName = "";
+        itemName = key;
+        if(!metricsGroup.containsKey(groupName)) {
+          metricsGroup.put(groupName, new HashMap<String, T>());
+        }
+        metricsGroup.get(groupName).put(itemName, entry.getValue());
+        continue;
+      }
+
+      if (previousGroup != null && !previousGroup.equals(groupName)) {
+        metricsGroup.put(previousGroup, groupItems);
+        groupItems = new HashMap<String, T>();
+      }
+      groupItems.put(itemName, entry.getValue());
+      previousGroup = groupName;
+    }
+
+    if(groupItems != null && !groupItems.isEmpty()) {
+      metricsGroup.put(previousGroup, groupItems);
+    }
+
+    return metricsGroup;
+  }
+
+  protected String meterGroupToString(String dateTime, String hostAndPort, double rateFactor,
+                                      String groupName, Map<String, Meter> meters) {
+    StringBuilder sb = new StringBuilder();
+    sb.append(dateTime).append(" ");
+    if(hostAndPort != null && !hostAndPort.isEmpty()) {
+      sb.append(hostAndPort).append(" ");
+    }
+    sb.append("meter").append(" ");
+
+    if(!groupName.isEmpty()) {
+      sb.append(groupName).append(" ");
+    }
+    String prefix = "";
+    for(Map.Entry<String, Meter> eachMeter: meters.entrySet()) {
+      String key = eachMeter.getKey();
+      Meter meter = eachMeter.getValue();
+      sb.append(prefix);
+      sb.append(key).append(".count=").append(meter.getCount()).append("|");
+      sb.append(key).append(".mean=").append(String.format("%2.2f",
+          convertRate(meter.getMeanRate(), rateFactor))).append("|");
+      sb.append(key).append(".1minute=").append(String.format("%2.2f",
+          convertRate(meter.getOneMinuteRate(), rateFactor))).append("|");
+      sb.append(key).append(".5minute=").append(String.format("%2.2f",
+          convertRate(meter.getFiveMinuteRate(), rateFactor))).append("|");
+      sb.append(key).append(".15minute=").append(String.format("%2.2f",
+          convertRate(meter.getFifteenMinuteRate(), rateFactor)));
+      prefix = ",";
+    }
+
+    return sb.toString();
+  }
+
+  protected String counterGroupToString(String dateTime, String hostAndPort, double rateFactor,
+                                        String groupName, Map<String, Counter> counters) {
+    StringBuilder sb = new StringBuilder();
+    sb.append(dateTime).append(" ");
+    if(hostAndPort != null && !hostAndPort.isEmpty()) {
+      sb.append(hostAndPort).append(" ");
+    }
+    sb.append("counter").append(" ");
+
+    if(!groupName.isEmpty()) {
+      sb.append(groupName).append(" ");
+    }
+    String prefix = "";
+    for(Map.Entry<String, Counter> eachCounter: counters.entrySet()) {
+      sb.append(prefix);
+      sb.append(eachCounter.getKey()).append("=").append(eachCounter.getValue().getCount());
+      prefix = ",";
+
+    }
+    return sb.toString();
+  }
+
+  protected String gaugeGroupToString(String dateTime, String hostAndPort, double rateFactor,
+                                      String groupName, Map<String, Gauge> gauges) {
+    StringBuilder sb = new StringBuilder();
+    sb.append(dateTime).append(" ");
+    if(hostAndPort != null && !hostAndPort.isEmpty()) {
+      sb.append(hostAndPort).append(" ");
+    }
+    sb.append("guage").append(" ");
+
+    if(!groupName.isEmpty()) {
+      sb.append(groupName).append(" ");
+    }
+    String prefix = "";
+    for(Map.Entry<String, Gauge> eachGauge: gauges.entrySet()) {
+      sb.append(prefix).append(eachGauge.getKey()).append("=").append(eachGauge.getValue().getValue());
+      prefix = ",";
+    }
+    return sb.toString();
+  }
+
+  protected String histogramGroupToString(String dateTime, String hostAndPort, double rateFactor,
+                                          String groupName, Map<String, Histogram> histograms) {
+    StringBuilder sb = new StringBuilder();
+    sb.append(dateTime).append(" ");
+    if(hostAndPort != null && !hostAndPort.isEmpty()) {
+      sb.append(hostAndPort).append(" ");
+    }
+    sb.append("histo").append(" ");
+
+    String prefix = "";
+    for(Map.Entry<String, Histogram> eachHistogram: histograms.entrySet()) {
+      String key = eachHistogram.getKey();
+      Histogram histogram = eachHistogram.getValue();
+      sb.append(prefix);
+      sb.append(key).append(".count=").append(histogram.getCount()).append("|");
+
+      Snapshot snapshot = histogram.getSnapshot();
+
+      sb.append(key).append(".min=").append(snapshot.getMin()).append("|");
+      sb.append(key).append(".max=").append(snapshot.getMax()).append("|");
+      sb.append(key).append(".mean=").append(String.format("%2.2f", snapshot.getMean())).append("|");
+      sb.append(key).append(".stddev=").append(String.format("%2.2f", snapshot.getStdDev())).append("|");
+      sb.append(key).append(".median=").append(String.format("%2.2f", snapshot.getMedian())).append("|");
+      sb.append(key).append(".75%=").append(String.format("%2.2f", snapshot.get75thPercentile())).append("|");
+      sb.append(key).append(".95%=").append(String.format("%2.2f", snapshot.get95thPercentile())).append("|");
+      sb.append(key).append(".98%=").append(String.format("%2.2f", snapshot.get98thPercentile())).append("|");
+      sb.append(key).append(".99%=").append(String.format("%2.2f", snapshot.get99thPercentile())).append("|");
+      sb.append(key).append(".999%=").append(String.format("%2.2f", snapshot.get999thPercentile()));
+      prefix = ",";
+    }
+    return sb.toString();
+  }
+
+  protected String timerGroupToString(String dateTime, String hostAndPort, double rateFactor,
+                                 String groupName, Map<String, Timer> timers) {
+    StringBuilder sb = new StringBuilder();
+
+    sb.append(dateTime).append(" ");
+    if(hostAndPort != null && !hostAndPort.isEmpty()) {
+      sb.append(hostAndPort).append(" ");
+    }
+    sb.append("timer").append(" ");
+
+    if(!groupName.isEmpty()) {
+      sb.append(groupName).append(" ");
+    }
+    String prefix = "";
+    for(Map.Entry<String, Timer> eachTimer: timers.entrySet()) {
+      String key = eachTimer.getKey();
+      Timer timer = eachTimer.getValue();
+      Snapshot snapshot = timer.getSnapshot();
+
+      sb.append(prefix);
+      sb.append(key).append(".count=").append(timer.getCount()).append("|");
+      sb.append(key).append(".meanrate=").append(String.format("%2.2f", convertRate(timer.getMeanRate(), rateFactor))).append("|");
+      sb.append(key).append(".1minuterate=").append(String.format("%2.2f", convertRate(timer.getOneMinuteRate(), rateFactor))).append("|");
+      sb.append(key).append(".5minuterate=").append(String.format("%2.2f", convertRate(timer.getFiveMinuteRate(), rateFactor))).append("|");
+      sb.append(key).append(".15minuterate=").append(String.format("%2.2f", convertRate(timer.getFifteenMinuteRate(),rateFactor))).append("|");
+      sb.append(key).append(".min=").append(String.format("%2.2f", convertRate(snapshot.getMin(), rateFactor))).append("|");
+      sb.append(key).append(".max=").append(String.format("%2.2f", convertRate(snapshot.getMax(),rateFactor))).append("|");
+      sb.append(key).append(".mean=").append(String.format("%2.2f", convertRate(snapshot.getMean(), rateFactor))).append("|");
+      sb.append(key).append(".stddev=").append(String.format("%2.2f", convertRate(snapshot.getStdDev(),rateFactor))).append("|");
+      sb.append(key).append(".median=").append(String.format("%2.2f", convertRate(snapshot.getMedian(), rateFactor))).append("|");
+      sb.append(key).append(".75%=").append(String.format("%2.2f", convertRate(snapshot.get75thPercentile(), rateFactor))).append("|");
+      sb.append(key).append(".95%=").append(String.format("%2.2f", convertRate(snapshot.get95thPercentile(), rateFactor))).append("|");
+      sb.append(key).append(".98%=").append(String.format("%2.2f", convertRate(snapshot.get98thPercentile(), rateFactor))).append("|");
+      sb.append(key).append(".99%=").append(String.format("%2.2f", convertRate(snapshot.get99thPercentile(), rateFactor))).append("|");
+      sb.append(key).append(".999%=").append(String.format("%2.2f", convertRate(snapshot.get999thPercentile(),rateFactor)));
+      prefix = ",";
+    }
+
+    return sb.toString();
+  }
+
+  protected double convertRate(double rate, double rateFactor) {
+    return rate * rateFactor;
+  }
+
+}


Mime
View raw message