HBASE-9194 Break HMaster metrics into multiple contexts

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1514513 13f79535-47bb-0310-9956-ffa450edef68

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsAssignmentManagerSource.java

@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSource;
+
+public interface MetricsAssignmentManagerSource extends BaseSource {
+
+  /**
+   * The name of the metrics
+   */
+  String METRICS_NAME = "AssignmentManger";
+
+  /**
+   * The context metrics will be under.
+   */
+  String METRICS_CONTEXT = "master";
+
+  /**
+   * The name of the metrics context that metrics will be under in jmx
+   */
+  String METRICS_JMX_CONTEXT = "Master,sub=" + METRICS_NAME;
+
+  /**
+   * Description
+   */
+  String METRICS_DESCRIPTION = "Metrics about HBase master assingment manager.";
+
+  String RIT_COUNT_NAME = "ritCount";
+  String RIT_COUNT_OVER_THRESHOLD_NAME = "ritCountOverThreshold";
+  String RIT_OLDEST_AGE_NAME = "ritOldestAge";
+  String ASSIGN_TIME_NAME = "assign";
+  String BULK_ASSIGN_TIME_NAME = "bulkAssign";
+
+  void updateAssignmentTime(long time);
+
+  void updateBulkAssignTime(long time);
+
+  /**
+   * Set the number of regions in transition.
+   *
+   * @param ritCount count of the regions in transition.
+   */
+  void setRIT(int ritCount);
+
+  /**
+   * Set the count of the number of regions that have been in transition over the threshold time.
+   *
+   * @param ritCountOverThreshold number of regions in transition for longer than threshold.
+   */
+  void setRITCountOverThreshold(int ritCountOverThreshold);
+
+  /**
+   * Set the oldest region in transition.
+   *
+   * @param age age of the oldest RIT.
+   */
+  void setRITOldestAge(long age);
+}

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterFileSystemSource.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.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSource;
+
+public interface MetricsMasterFileSystemSource extends BaseSource {
+
+  /**
+   * The name of the metrics
+   */
+  String METRICS_NAME = "FileSystem";
+
+  /**
+   * The context metrics will be under.
+   */
+  String METRICS_CONTEXT = "master";
+
+  /**
+   * The name of the metrics context that metrics will be under in jmx
+   */
+  String METRICS_JMX_CONTEXT = "Master,sub=" + METRICS_NAME;
+
+  /**
+   * Description
+   */
+  String METRICS_DESCRIPTION = "Metrics about HBase master file system.";
+
+  String META_SPLIT_TIME_NAME = "metaHlogSplitTime";
+  String META_SPLIT_SIZE_NAME = "metaHlogSplitSize";
+  String SPLIT_TIME_NAME = "hlogSplitTime";
+  String SPLIT_SIZE_NAME = "hlogSplitSize";
+
+  String META_SPLIT_TIME_DESC = "Time it takes to finish splitMetaLog()";
+  String META_SPLIT_SIZE_DESC = "Size of META HLog files being split";
+  String SPLIT_TIME_DESC = "Time it takes to finish HLog.splitLog()";
+  String SPLIT_SIZE_DESC = "Size of HLog files being split";
+
+
+  void updateMetaWALSplitTime(long time);
+
+  void updateMetaWALSplitSize(long size);
+
+  void updateSplitTime(long time);
+
+  void updateSplitSize(long size);
+
+}

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterSource.java

@@ -55,17 +55,9 @@ public interface MetricsMasterSource extends BaseSource {
   String SERVER_NAME_NAME = "serverName";
   String CLUSTER_ID_NAME = "clusterId";
   String IS_ACTIVE_MASTER_NAME = "isActiveMaster";
-  String SPLIT_TIME_NAME = "hlogSplitTime";
+
-  String SPLIT_SIZE_NAME = "hlogSplitSize";
+
-  String SNAPSHOT_TIME_NAME = "snapshotTime";
-  String SNAPSHOT_RESTORE_TIME_NAME = "snapshotRestoreTime";
-  String SNAPSHOT_CLONE_TIME_NAME = "snapshotCloneTime";
-  String META_SPLIT_TIME_NAME = "metaHlogSplitTime";
-  String META_SPLIT_SIZE_NAME = "metaHlogSplitSize";
   String CLUSTER_REQUESTS_NAME = "clusterRequests";
-  String RIT_COUNT_NAME = "ritCount";
-  String RIT_COUNT_OVER_THRESHOLD_NAME = "ritCountOverThreshold";
-  String RIT_OLDEST_AGE_NAME = "ritOldestAge";
   String MASTER_ACTIVE_TIME_DESC = "Master Active Time";
   String MASTER_START_TIME_DESC = "Master Start Time";
   String AVERAGE_LOAD_DESC = "AverageLoad";
@@ -75,13 +67,8 @@ public interface MetricsMasterSource extends BaseSource {
   String SERVER_NAME_DESC = "Server Name";
   String CLUSTER_ID_DESC = "Cluster Id";
   String IS_ACTIVE_MASTER_DESC = "Is Active Master";
-  String SPLIT_TIME_DESC = "Time it takes to finish HLog.splitLog()";
+
-  String SPLIT_SIZE_DESC = "Size of HLog files being split";
+
-  String SNAPSHOT_TIME_DESC = "Time it takes to finish snapshot()";
-  String SNAPSHOT_RESTORE_TIME_DESC = "Time it takes to finish restoreSnapshot()";
-  String SNAPSHOT_CLONE_TIME_DESC = "Time it takes to finish cloneSnapshot()";
-  String META_SPLIT_TIME_DESC = "Time it takes to finish splitMetaLog()";
-  String META_SPLIT_SIZE_DESC = "Size of META HLog files being split";
 
   /**
    * Increment the number of requests the cluster has seen.
@@ -90,39 +77,7 @@ public interface MetricsMasterSource extends BaseSource {
    */
   void incRequests(final int inc);
 
-  /**
-   * Set the number of regions in transition.
-   *
-   * @param ritCount count of the regions in transition.
-   */
-  void setRIT(int ritCount);
   
-  /**
-   * Set the count of the number of regions that have been in transition over the threshold time.
-   *
-   * @param ritCountOverThreshold number of regions in transition for longer than threshold.
-   */
-  void setRITCountOverThreshold(int ritCountOverThreshold);
 
-  /**
-   * Set the oldest region in transition.
-   *
-   * @param age age of the oldest RIT.
-   */
-  void setRITOldestAge(long age);
-
-  void updateSplitTime(long time);
-
-  void updateSplitSize(long size);
-
-  void updateSnapshotTime(long time);
-
-  void updateSnapshotCloneTime(long time);
-
-  void updateSnapshotRestoreTime(long time);
-  
-  void updateMetaWALSplitTime(long time);
-
-  void updateMetaWALSplitSize(long size);
 
 }

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsSnapshotSource.java

@@ -0,0 +1,56 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSource;
+
+public interface MetricsSnapshotSource extends BaseSource {
+  /**
+   * The name of the metrics
+   */
+  String METRICS_NAME = "Snapshots";
+
+  /**
+   * The context metrics will be under.
+   */
+  String METRICS_CONTEXT = "master";
+
+  /**
+   * The name of the metrics context that metrics will be under in jmx
+   */
+  String METRICS_JMX_CONTEXT = "Master,sub=" + METRICS_NAME;
+
+  /**
+   * Description
+   */
+  String METRICS_DESCRIPTION = "Metrics about HBase master server";
+
+  String SNAPSHOT_TIME_NAME = "snapshotTime";
+  String SNAPSHOT_RESTORE_TIME_NAME = "snapshotRestoreTime";
+  String SNAPSHOT_CLONE_TIME_NAME = "snapshotCloneTime";
+  String SNAPSHOT_TIME_DESC = "Time it takes to finish snapshot()";
+  String SNAPSHOT_RESTORE_TIME_DESC = "Time it takes to finish restoreSnapshot()";
+  String SNAPSHOT_CLONE_TIME_DESC = "Time it takes to finish cloneSnapshot()";
+
+  void updateSnapshotTime(long time);
+
+  void updateSnapshotCloneTime(long time);
+
+  void updateSnapshotRestoreTime(long time);
+}

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/master/balancer/MetricsBalancerSource.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.hadoop.hbase.master.balancer;
+
+import org.apache.hadoop.hbase.metrics.BaseSource;
+
+public interface MetricsBalancerSource extends BaseSource  {
+
+  /**
+   * The name of the metrics
+   */
+  String METRICS_NAME = "Balancer";
+
+  /**
+   * The context metrics will be under.
+   */
+  String METRICS_CONTEXT = "master";
+
+  /**
+   * The name of the metrics context that metrics will be under in jmx
+   */
+  String METRICS_JMX_CONTEXT = "Master,sub=" + METRICS_NAME;
+
+  String BALANCE_CLUSTER = "balancerCluster";
+  String MISC_INVOATION_COUNT = "miscInvocationCount";
+
+  /**
+   * Description
+   */
+  String METRICS_DESCRIPTION = "Metrics about HBase master balancer";
+
+  void updateBalanceCluster(long time);
+
+  void incrMiscInvocations();
+}

hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsAssignmentManagerSourceImpl.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.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MetricMutableGaugeLong;
+import org.apache.hadoop.metrics2.lib.MetricMutableHistogram;
+
+public class MetricsAssignmentManagerSourceImpl extends BaseSourceImpl implements MetricsAssignmentManagerSource {
+
+  private MetricMutableGaugeLong ritGauge;
+  private MetricMutableGaugeLong ritCountOverThresholdGauge;
+  private MetricMutableGaugeLong ritOldestAgeGauge;
+  private MetricMutableHistogram assignTimeHisto;
+  private MetricMutableHistogram bulkAssignTimeHisto;
+
+  public MetricsAssignmentManagerSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsAssignmentManagerSourceImpl(String metricsName,
+                                            String metricsDescription,
+                                            String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+  @Override
+  public void init() {
+    super.init();
+    ritGauge = metricsRegistry.newGauge(RIT_COUNT_NAME, "", 0l);
+    ritCountOverThresholdGauge = metricsRegistry.newGauge(RIT_COUNT_OVER_THRESHOLD_NAME, "", 0l);
+    ritOldestAgeGauge = metricsRegistry.newGauge(RIT_OLDEST_AGE_NAME, "", 0l);
+    assignTimeHisto = metricsRegistry.newHistogram(ASSIGN_TIME_NAME);
+    bulkAssignTimeHisto = metricsRegistry.newHistogram(BULK_ASSIGN_TIME_NAME);
+  }
+
+  @Override
+  public void updateAssignmentTime(long time) {
+    assignTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateBulkAssignTime(long time) {
+    bulkAssignTimeHisto.add(time);
+  }
+
+  public void setRIT(int ritCount) {
+    ritGauge.set(ritCount);
+  }
+
+  public void setRITCountOverThreshold(int ritCount) {
+    ritCountOverThresholdGauge.set(ritCount);
+  }
+
+  public void setRITOldestAge(long ritCount) {
+    ritOldestAgeGauge.set(ritCount);
+  }
+}

hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterFilesystemSourceImpl.java

@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MetricMutableHistogram;
+
+public class MetricsMasterFilesystemSourceImpl extends BaseSourceImpl implements MetricsMasterFileSystemSource {
+
+  private MetricMutableHistogram splitSizeHisto;
+  private MetricMutableHistogram splitTimeHisto;
+  private MetricMutableHistogram metaSplitTimeHisto;
+  private MetricMutableHistogram metaSplitSizeHisto;
+
+  public MetricsMasterFilesystemSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsMasterFilesystemSourceImpl(String metricsName,
+                                           String metricsDescription,
+                                           String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+  @Override
+  public void init() {
+    splitSizeHisto = metricsRegistry.newHistogram(SPLIT_SIZE_NAME, SPLIT_SIZE_DESC);
+    splitTimeHisto = metricsRegistry.newHistogram(SPLIT_TIME_NAME, SPLIT_TIME_DESC);
+    metaSplitTimeHisto = metricsRegistry.newHistogram(META_SPLIT_TIME_NAME, META_SPLIT_TIME_DESC);
+    metaSplitSizeHisto = metricsRegistry.newHistogram(META_SPLIT_SIZE_NAME, META_SPLIT_SIZE_DESC);
+  }
+
+  @Override
+  public void updateSplitTime(long time) {
+    splitTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateSplitSize(long size) {
+    splitSizeHisto.add(size);
+  }
+
+
+  @Override
+  public void updateMetaWALSplitTime(long time) {
+    metaSplitTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateMetaWALSplitSize(long size) {
+    metaSplitSizeHisto.add(size);
+  }
+}

hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterSourceImpl.java

@@ -40,16 +40,7 @@ public class MetricsMasterSourceImpl
 
   private final MetricsMasterWrapper masterWrapper;
   private MetricMutableCounterLong clusterRequestsCounter;
-  private MetricMutableGaugeLong ritGauge;
+
-  private MetricMutableGaugeLong ritCountOverThresholdGauge;
-  private MetricMutableGaugeLong ritOldestAgeGauge;
-  private MetricMutableHistogram splitTimeHisto;
-  private MetricMutableHistogram splitSizeHisto;
-  private MetricMutableStat snapshotTimeHisto;
-  private MetricMutableStat snapshotCloneTimeHisto;
-  private MetricMutableStat snapshotRestoreTimeHisto;
-  private MetricMutableHistogram metaSplitTimeHisto;
-  private MetricMutableHistogram metaSplitSizeHisto;
 
   public MetricsMasterSourceImpl(MetricsMasterWrapper masterWrapper) {
     this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT, masterWrapper);
@@ -68,72 +59,12 @@ public class MetricsMasterSourceImpl
   public void init() {
     super.init();
     clusterRequestsCounter = metricsRegistry.newCounter(CLUSTER_REQUESTS_NAME, "", 0l);
-    ritGauge = metricsRegistry.newGauge(RIT_COUNT_NAME, "", 0l);
-    ritCountOverThresholdGauge = metricsRegistry.newGauge(RIT_COUNT_OVER_THRESHOLD_NAME, "", 0l);
-    ritOldestAgeGauge = metricsRegistry.newGauge(RIT_OLDEST_AGE_NAME, "", 0l);
-    splitSizeHisto = metricsRegistry.newHistogram(SPLIT_SIZE_NAME, SPLIT_SIZE_DESC);
-    splitTimeHisto = metricsRegistry.newHistogram(SPLIT_TIME_NAME, SPLIT_TIME_DESC);
-    snapshotTimeHisto = metricsRegistry.newStat(
-        SNAPSHOT_TIME_NAME, SNAPSHOT_TIME_DESC, "Ops", "Time", true);
-    snapshotCloneTimeHisto = metricsRegistry.newStat(
-        SNAPSHOT_CLONE_TIME_NAME, SNAPSHOT_CLONE_TIME_DESC, "Ops", "Time", true);
-    snapshotRestoreTimeHisto = metricsRegistry.newStat(
-        SNAPSHOT_RESTORE_TIME_NAME, SNAPSHOT_RESTORE_TIME_DESC, "Ops", "Time", true);
-    metaSplitTimeHisto = metricsRegistry.newHistogram(META_SPLIT_TIME_NAME, META_SPLIT_TIME_DESC);
-    metaSplitSizeHisto = metricsRegistry.newHistogram(META_SPLIT_SIZE_NAME, META_SPLIT_SIZE_DESC);
   }
 
   public void incRequests(final int inc) {
     this.clusterRequestsCounter.incr(inc);
   }
 
-  public void setRIT(int ritCount) {
-    ritGauge.set(ritCount);
-  }
-
-  public void setRITCountOverThreshold(int ritCount) {
-    ritCountOverThresholdGauge.set(ritCount);
-  }
-
-  public void setRITOldestAge(long ritCount) {
-    ritOldestAgeGauge.set(ritCount);
-  }
-
-  @Override
-  public void updateSplitTime(long time) {
-    splitTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateSplitSize(long size) {
-    splitSizeHisto.add(size);
-  }
-
-  @Override
-  public void updateSnapshotTime(long time) {
-    snapshotTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateSnapshotCloneTime(long time) {
-    snapshotCloneTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateSnapshotRestoreTime(long time) {
-    snapshotRestoreTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateMetaWALSplitTime(long time) {
-    metaSplitTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateMetaWALSplitSize(long size) {
-    metaSplitSizeHisto.add(size);
-  }
-
   /**
    * Method to export all the metrics.
    *

hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsSnapshotSourceImpl.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.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MetricMutableHistogram;
+
+public class MetricsSnapshotSourceImpl extends BaseSourceImpl implements MetricsSnapshotSource {
+
+  private MetricMutableHistogram snapshotTimeHisto;
+  private MetricMutableHistogram snapshotCloneTimeHisto;
+  private MetricMutableHistogram snapshotRestoreTimeHisto;
+
+  public MetricsSnapshotSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsSnapshotSourceImpl(String metricsName,
+                                   String metricsDescription,
+                                   String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+  @Override
+  public void init() {
+    snapshotTimeHisto = metricsRegistry.newHistogram(
+        SNAPSHOT_TIME_NAME, SNAPSHOT_TIME_DESC);
+    snapshotCloneTimeHisto = metricsRegistry.newHistogram(
+        SNAPSHOT_CLONE_TIME_NAME, SNAPSHOT_CLONE_TIME_DESC);
+    snapshotRestoreTimeHisto = metricsRegistry.newHistogram(
+        SNAPSHOT_RESTORE_TIME_NAME, SNAPSHOT_RESTORE_TIME_DESC);
+  }
+
+  @Override
+  public void updateSnapshotTime(long time) {
+    snapshotTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateSnapshotCloneTime(long time) {
+    snapshotCloneTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateSnapshotRestoreTime(long time) {
+    snapshotRestoreTimeHisto.add(time);
+  }
+}

hbase-hadoop1-compat/src/main/java/org/apache/hadoop/hbase/master/balancer/MetricsBalancerSourceImpl.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.hadoop.hbase.master.balancer;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MetricMutableCounterLong;
+import org.apache.hadoop.metrics2.lib.MetricMutableHistogram;
+
+public class MetricsBalancerSourceImpl extends BaseSourceImpl implements MetricsBalancerSource{
+
+  private MetricMutableHistogram blanceClusterHisto;
+  private MetricMutableCounterLong miscCount;
+
+  public MetricsBalancerSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsBalancerSourceImpl(String metricsName,
+                                   String metricsDescription,
+                                   String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+
+  @Override
+  public void init() {
+    blanceClusterHisto = metricsRegistry.newHistogram(BALANCE_CLUSTER);
+    miscCount = metricsRegistry.newCounter(MISC_INVOATION_COUNT, "", 0L);
+
+  }
+
+  @Override
+  public void updateBalanceCluster(long time) {
+    blanceClusterHisto.add(time);
+  }
+
+  @Override
+  public void incrMiscInvocations() {
+     miscCount.incr();
+  }
+}

hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.MetricsAssignmentManagerSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.MetricsAssignmentManagerSourceImpl

hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.MetricsMasterFileSystemSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.MetricsMasterFilesystemSourceImpl

hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.MetricsSnapshotSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.MetricsSnapshotSourceImpl

hbase-hadoop1-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.balancer.MetricsBalancerSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.balancer.MetricsBalancerSourceImpl

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsAssignmentManagerSourceImpl.java

@@ -0,0 +1,73 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
+import org.apache.hadoop.metrics2.lib.MutableHistogram;
+
+
+public class MetricsAssignmentManagerSourceImpl extends BaseSourceImpl implements MetricsAssignmentManagerSource {
+
+  private MutableGaugeLong ritGauge;
+  private MutableGaugeLong ritCountOverThresholdGauge;
+  private MutableGaugeLong ritOldestAgeGauge;
+  private MutableHistogram assignTimeHisto;
+  private MutableHistogram bulkAssignTimeHisto;
+
+  public MetricsAssignmentManagerSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsAssignmentManagerSourceImpl(String metricsName,
+                                            String metricsDescription,
+                                            String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+  public void init() {
+    ritGauge = metricsRegistry.newGauge(RIT_COUNT_NAME, "", 0l);
+    ritCountOverThresholdGauge = metricsRegistry.newGauge(RIT_COUNT_OVER_THRESHOLD_NAME, "", 0l);
+    ritOldestAgeGauge = metricsRegistry.newGauge(RIT_OLDEST_AGE_NAME, "", 0l);
+    assignTimeHisto = metricsRegistry.newHistogram(ASSIGN_TIME_NAME);
+    bulkAssignTimeHisto = metricsRegistry.newHistogram(BULK_ASSIGN_TIME_NAME);
+  }
+
+  @Override
+  public void updateAssignmentTime(long time) {
+    assignTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateBulkAssignTime(long time) {
+    bulkAssignTimeHisto.add(time);
+  }
+
+  public void setRIT(int ritCount) {
+    ritGauge.set(ritCount);
+  }
+
+  public void setRITCountOverThreshold(int ritCount) {
+    ritCountOverThresholdGauge.set(ritCount);
+  }
+
+  public void setRITOldestAge(long ritCount) {
+    ritOldestAgeGauge.set(ritCount);
+  }
+}

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterFilesystemSourceImpl.java

@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MutableHistogram;
+
+public class MetricsMasterFilesystemSourceImpl extends BaseSourceImpl implements MetricsMasterFileSystemSource {
+
+  private MutableHistogram splitSizeHisto;
+  private MutableHistogram splitTimeHisto;
+  private MutableHistogram metaSplitTimeHisto;
+  private MutableHistogram metaSplitSizeHisto;
+
+  public MetricsMasterFilesystemSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsMasterFilesystemSourceImpl(String metricsName,
+                                           String metricsDescription,
+                                           String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+  @Override
+  public void init() {
+    splitSizeHisto = metricsRegistry.newHistogram(SPLIT_SIZE_NAME, SPLIT_SIZE_DESC);
+    splitTimeHisto = metricsRegistry.newHistogram(SPLIT_TIME_NAME, SPLIT_TIME_DESC);
+    metaSplitTimeHisto = metricsRegistry.newHistogram(META_SPLIT_TIME_NAME, META_SPLIT_TIME_DESC);
+    metaSplitSizeHisto = metricsRegistry.newHistogram(META_SPLIT_SIZE_NAME, META_SPLIT_SIZE_DESC);
+  }
+
+  @Override
+  public void updateSplitTime(long time) {
+    splitTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateSplitSize(long size) {
+    splitSizeHisto.add(size);
+  }
+
+
+  @Override
+  public void updateMetaWALSplitTime(long time) {
+    metaSplitTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateMetaWALSplitSize(long size) {
+    metaSplitSizeHisto.add(size);
+  }
+}

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterSourceImpl.java

@@ -37,16 +37,6 @@ public class MetricsMasterSourceImpl
 
   private final MetricsMasterWrapper masterWrapper;
   private MutableCounterLong clusterRequestsCounter;
-  private MutableGaugeLong ritGauge;
-  private MutableGaugeLong ritCountOverThresholdGauge;
-  private MutableGaugeLong ritOldestAgeGauge;
-  private MutableHistogram splitTimeHisto;
-  private MutableHistogram splitSizeHisto;
-  private MutableStat snapshotTimeHisto;
-  private MutableStat snapshotCloneTimeHisto;
-  private MutableStat snapshotRestoreTimeHisto;
-  private MutableHistogram metaSplitTimeHisto;
-  private MutableHistogram metaSplitSizeHisto;
 
   public MetricsMasterSourceImpl(MetricsMasterWrapper masterWrapper) {
     this(METRICS_NAME,
@@ -70,72 +60,12 @@ public class MetricsMasterSourceImpl
   public void init() {
     super.init();
     clusterRequestsCounter = metricsRegistry.newCounter(CLUSTER_REQUESTS_NAME, "", 0l);
-    ritGauge = metricsRegistry.newGauge(RIT_COUNT_NAME, "", 0l);
-    ritCountOverThresholdGauge = metricsRegistry.newGauge(RIT_COUNT_OVER_THRESHOLD_NAME, "", 0l);
-    ritOldestAgeGauge = metricsRegistry.newGauge(RIT_OLDEST_AGE_NAME, "", 0l);
-    splitSizeHisto = metricsRegistry.newHistogram(SPLIT_SIZE_NAME, SPLIT_SIZE_DESC);
-    splitTimeHisto = metricsRegistry.newHistogram(SPLIT_TIME_NAME, SPLIT_TIME_DESC);
-    snapshotTimeHisto = metricsRegistry.newStat(
-        SNAPSHOT_TIME_NAME, SNAPSHOT_TIME_DESC, "Ops", "Time", true);
-    snapshotCloneTimeHisto = metricsRegistry.newStat(
-        SNAPSHOT_CLONE_TIME_NAME, SNAPSHOT_CLONE_TIME_DESC, "Ops", "Time", true);
-    snapshotRestoreTimeHisto = metricsRegistry.newStat(
-        SNAPSHOT_RESTORE_TIME_NAME, SNAPSHOT_RESTORE_TIME_DESC, "Ops", "Time", true);
-    metaSplitTimeHisto = metricsRegistry.newHistogram(META_SPLIT_TIME_NAME, META_SPLIT_TIME_DESC);
-    metaSplitSizeHisto = metricsRegistry.newHistogram(META_SPLIT_SIZE_NAME, META_SPLIT_SIZE_DESC);
   }
 
   public void incRequests(final int inc) {
     this.clusterRequestsCounter.incr(inc);
   }
 
-  public void setRIT(int ritCount) {
-    ritGauge.set(ritCount);
-  }
-
-  public void setRITCountOverThreshold(int ritCount) {
-    ritCountOverThresholdGauge.set(ritCount);
-  }
-
-  public void setRITOldestAge(long ritCount) {
-    ritOldestAgeGauge.set(ritCount);
-  }
-
-  @Override
-  public void updateSplitTime(long time) {
-    splitTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateSplitSize(long size) {
-    splitSizeHisto.add(size);
-  }
-
-  @Override
-  public void updateSnapshotTime(long time) {
-    snapshotTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateSnapshotCloneTime(long time) {
-    snapshotCloneTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateSnapshotRestoreTime(long time) {
-    snapshotRestoreTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateMetaWALSplitTime(long time) {
-    metaSplitTimeHisto.add(time);
-  }
-
-  @Override
-  public void updateMetaWALSplitSize(long size) {
-    metaSplitSizeHisto.add(size);
-  }
-
   @Override
   public void getMetrics(MetricsCollector metricsCollector, boolean all) {
 

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/master/MetricsSnapshotSourceImpl.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.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MutableHistogram;
+
+public class MetricsSnapshotSourceImpl extends BaseSourceImpl implements MetricsSnapshotSource {
+
+  private MutableHistogram snapshotTimeHisto;
+  private MutableHistogram snapshotCloneTimeHisto;
+  private MutableHistogram snapshotRestoreTimeHisto;
+
+  public MetricsSnapshotSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsSnapshotSourceImpl(String metricsName,
+                                   String metricsDescription,
+                                   String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+  @Override
+  public void init() {
+    snapshotTimeHisto = metricsRegistry.newHistogram(
+        SNAPSHOT_TIME_NAME, SNAPSHOT_TIME_DESC);
+    snapshotCloneTimeHisto = metricsRegistry.newHistogram(
+        SNAPSHOT_CLONE_TIME_NAME, SNAPSHOT_CLONE_TIME_DESC);
+    snapshotRestoreTimeHisto = metricsRegistry.newHistogram(
+        SNAPSHOT_RESTORE_TIME_NAME, SNAPSHOT_RESTORE_TIME_DESC);
+  }
+
+  @Override
+  public void updateSnapshotTime(long time) {
+    snapshotTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateSnapshotCloneTime(long time) {
+    snapshotCloneTimeHisto.add(time);
+  }
+
+  @Override
+  public void updateSnapshotRestoreTime(long time) {
+    snapshotRestoreTimeHisto.add(time);
+  }
+}

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/master/balancer/MetricsBalancerSourceImpl.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.hadoop.hbase.master.balancer;
+
+import org.apache.hadoop.hbase.metrics.BaseSourceImpl;
+import org.apache.hadoop.metrics2.lib.MutableCounterLong;
+import org.apache.hadoop.metrics2.lib.MutableHistogram;
+
+public class MetricsBalancerSourceImpl extends BaseSourceImpl implements MetricsBalancerSource{
+
+  private MutableHistogram blanceClusterHisto;
+  private MutableCounterLong miscCount;
+
+  public MetricsBalancerSourceImpl() {
+    this(METRICS_NAME, METRICS_DESCRIPTION, METRICS_CONTEXT, METRICS_JMX_CONTEXT);
+  }
+
+  public MetricsBalancerSourceImpl(String metricsName,
+                                   String metricsDescription,
+                                   String metricsContext, String metricsJmxContext) {
+    super(metricsName, metricsDescription, metricsContext, metricsJmxContext);
+  }
+
+
+  @Override
+  public void init() {
+    blanceClusterHisto = metricsRegistry.newHistogram(BALANCE_CLUSTER);
+    miscCount = metricsRegistry.newCounter(MISC_INVOATION_COUNT, "", 0L);
+
+  }
+
+  @Override
+  public void updateBalanceCluster(long time) {
+    blanceClusterHisto.add(time);
+  }
+
+  @Override
+  public void incrMiscInvocations() {
+     miscCount.incr();
+  }
+}

hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.MetricsAssignmentManagerSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.MetricsAssignmentManagerSourceImpl

hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.MetricsMasterFileSystemSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.MetricsMasterFilesystemSourceImpl

hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.MetricsSnapshotSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.MetricsSnapshotSourceImpl

hbase-hadoop2-compat/src/main/resources/META-INF/services/org.apache.hadoop.hbase.master.balancer.MetricsBalancerSource

@@ -0,0 +1 @@
+org.apache.hadoop.hbase.master.balancer.MetricsBalancerSourceImpl

hbase-server/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java

@@ -123,6 +123,8 @@ public class AssignmentManager extends ZooKeeperListener {
 
   private LoadBalancer balancer;
 
+  private final MetricsAssignmentManager metricsAssignmentManager;
+
   private final TableLockManager tableLockManager;
 
   final private KeyLocker<String> locker = new KeyLocker<String>();
@@ -182,9 +184,6 @@ public class AssignmentManager extends ZooKeeperListener {
   private List<EventType> ignoreStatesRSOffline = Arrays.asList(
       EventType.RS_ZK_REGION_FAILED_OPEN, EventType.RS_ZK_REGION_CLOSED);
 
-  // metrics instance to send metrics for RITs
-  MetricsMaster metricsMaster;
-
   private final RegionStates regionStates;
 
   // The threshold to use bulk assigning. Using bulk assignment
@@ -273,7 +272,6 @@ public class AssignmentManager extends ZooKeeperListener {
     int maxThreads = conf.getInt("hbase.assignment.threads.max", 30);
     this.threadPoolExecutorService = Threads.getBoundedCachedThreadPool(
       maxThreads, 60L, TimeUnit.SECONDS, Threads.newDaemonThreadFactory("AM."));
-    this.metricsMaster = metricsMaster;// can be null only with tests.
     this.regionStates = new RegionStates(server, serverManager);
 
     this.bulkAssignWaitTillAllAssigned =
@@ -286,6 +284,8 @@ public class AssignmentManager extends ZooKeeperListener {
     zkEventWorkers = Threads.getBoundedCachedThreadPool(workers, 60L,
             TimeUnit.SECONDS, threadFactory);
     this.tableLockManager = tableLockManager;
+
+    this.metricsAssignmentManager = new MetricsAssignmentManager();
   }
 
   void startTimeOutMonitor() {
@@ -1533,170 +1533,175 @@ public class AssignmentManager extends ZooKeeperListener {
    * @return true if successful
    */
   boolean assign(final ServerName destination, final List<HRegionInfo> regions) {
-    int regionCount = regions.size();
+    long startTime = EnvironmentEdgeManager.currentTimeMillis();
-    if (regionCount == 0) {
-      return true;
-    }
-    LOG.debug("Assigning " + regionCount + " region(s) to " + destination.toString());
-    Set<String> encodedNames = new HashSet<String>(regionCount);
-    for (HRegionInfo region : regions) {
-      encodedNames.add(region.getEncodedName());
-    }
-
-    List<HRegionInfo> failedToOpenRegions = new ArrayList<HRegionInfo>();
-    Map<String, Lock> locks = locker.acquireLocks(encodedNames);
     try {
-      AtomicInteger counter = new AtomicInteger(0);
+      int regionCount = regions.size();
-      Map<String, Integer> offlineNodesVersions = new ConcurrentHashMap<String, Integer>();
+      if (regionCount == 0) {
-      OfflineCallback cb = new OfflineCallback(
+        return true;
-        watcher, destination, counter, offlineNodesVersions);
+      }
-      Map<String, RegionPlan> plans = new HashMap<String, RegionPlan>(regions.size());
+      LOG.debug("Assigning " + regionCount + " region(s) to " + destination.toString());
-      List<RegionState> states = new ArrayList<RegionState>(regions.size());
+      Set<String> encodedNames = new HashSet<String>(regionCount);
       for (HRegionInfo region : regions) {
-        String encodedRegionName = region.getEncodedName();
+        encodedNames.add(region.getEncodedName());
-        RegionState state = forceRegionStateToOffline(region, true);
-        if (state != null && asyncSetOfflineInZooKeeper(state, cb, destination)) {
-          RegionPlan plan = new RegionPlan(region, state.getServerName(), destination);
-          plans.put(encodedRegionName, plan);
-          states.add(state);
-        } else {
-          LOG.warn("failed to force region state to offline or "
-            + "failed to set it offline in ZK, will reassign later: " + region);
-          failedToOpenRegions.add(region); // assign individually later
-          Lock lock = locks.remove(encodedRegionName);
-          lock.unlock();
-        }
       }
 
-      // Wait until all unassigned nodes have been put up and watchers set.
+      List<HRegionInfo> failedToOpenRegions = new ArrayList<HRegionInfo>();
-      int total = states.size();
+      Map<String, Lock> locks = locker.acquireLocks(encodedNames);
-      for (int oldCounter = 0; !server.isStopped();) {
-        int count = counter.get();
-        if (oldCounter != count) {
-          LOG.info(destination.toString() + " unassigned znodes=" + count +
-            " of total=" + total);
-          oldCounter = count;
-        }
-        if (count >= total) break;
-        Threads.sleep(5);
-      }
-
-      if (server.isStopped()) {
-        return false;
-      }
-
-      // Add region plans, so we can updateTimers when one region is opened so
-      // that unnecessary timeout on RIT is reduced.
-      this.addPlans(plans);
-
-      List<Triple<HRegionInfo, Integer, List<ServerName>>> regionOpenInfos =
-        new ArrayList<Triple<HRegionInfo, Integer, List<ServerName>>>(states.size());
-      for (RegionState state: states) {
-        HRegionInfo region = state.getRegion();
-        String encodedRegionName = region.getEncodedName();
-        Integer nodeVersion = offlineNodesVersions.get(encodedRegionName);
-        if (nodeVersion == null || nodeVersion == -1) {
-          LOG.warn("failed to offline in zookeeper: " + region);
-          failedToOpenRegions.add(region); // assign individually later
-          Lock lock = locks.remove(encodedRegionName);
-          lock.unlock();
-        } else {
-          regionStates.updateRegionState(region,
-            RegionState.State.PENDING_OPEN, destination);
-          List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST;
-          if (this.shouldAssignRegionsWithFavoredNodes) {
-            favoredNodes = ((FavoredNodeLoadBalancer)this.balancer).getFavoredNodes(region);
-          }
-          regionOpenInfos.add(new Triple<HRegionInfo, Integer,  List<ServerName>>(
-            region, nodeVersion, favoredNodes));
-        }
-      }
-
-      // Move on to open regions.
       try {
-        // Send OPEN RPC. If it fails on a IOE or RemoteException, the
+        AtomicInteger counter = new AtomicInteger(0);
-        // TimeoutMonitor will pick up the pieces.
+        Map<String, Integer> offlineNodesVersions = new ConcurrentHashMap<String, Integer>();
-        long maxWaitTime = System.currentTimeMillis() +
+        OfflineCallback cb = new OfflineCallback(
-          this.server.getConfiguration().
+          watcher, destination, counter, offlineNodesVersions);
-            getLong("hbase.regionserver.rpc.startup.waittime", 60000);
+        Map<String, RegionPlan> plans = new HashMap<String, RegionPlan>(regions.size());
-        for (int i = 1; i <= maximumAttempts && !server.isStopped(); i++) {
+        List<RegionState> states = new ArrayList<RegionState>(regions.size());
-          try {
+        for (HRegionInfo region : regions) {
-            List<RegionOpeningState> regionOpeningStateList = serverManager
+          String encodedRegionName = region.getEncodedName();
-              .sendRegionOpen(destination, regionOpenInfos);
+          RegionState state = forceRegionStateToOffline(region, true);
-            if (regionOpeningStateList == null) {
+          if (state != null && asyncSetOfflineInZooKeeper(state, cb, destination)) {
-              // Failed getting RPC connection to this server
+            RegionPlan plan = new RegionPlan(region, state.getServerName(), destination);
-              return false;
+            plans.put(encodedRegionName, plan);
+            states.add(state);
+          } else {
+            LOG.warn("failed to force region state to offline or "
+              + "failed to set it offline in ZK, will reassign later: " + region);
+            failedToOpenRegions.add(region); // assign individually later
+            Lock lock = locks.remove(encodedRegionName);
+            lock.unlock();
+          }
+        }
+
+        // Wait until all unassigned nodes have been put up and watchers set.
+        int total = states.size();
+        for (int oldCounter = 0; !server.isStopped();) {
+          int count = counter.get();
+          if (oldCounter != count) {
+            LOG.info(destination.toString() + " unassigned znodes=" + count +
+              " of total=" + total);
+            oldCounter = count;
+          }
+          if (count >= total) break;
+          Threads.sleep(5);
+        }
+
+        if (server.isStopped()) {
+          return false;
+        }
+
+        // Add region plans, so we can updateTimers when one region is opened so
+        // that unnecessary timeout on RIT is reduced.
+        this.addPlans(plans);
+
+        List<Triple<HRegionInfo, Integer, List<ServerName>>> regionOpenInfos =
+          new ArrayList<Triple<HRegionInfo, Integer, List<ServerName>>>(states.size());
+        for (RegionState state: states) {
+          HRegionInfo region = state.getRegion();
+          String encodedRegionName = region.getEncodedName();
+          Integer nodeVersion = offlineNodesVersions.get(encodedRegionName);
+          if (nodeVersion == null || nodeVersion == -1) {
+            LOG.warn("failed to offline in zookeeper: " + region);
+            failedToOpenRegions.add(region); // assign individually later
+            Lock lock = locks.remove(encodedRegionName);
+            lock.unlock();
+          } else {
+            regionStates.updateRegionState(region,
+              RegionState.State.PENDING_OPEN, destination);
+            List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST;
+            if (this.shouldAssignRegionsWithFavoredNodes) {
+              favoredNodes = ((FavoredNodeLoadBalancer)this.balancer).getFavoredNodes(region);
             }
-            for (int k = 0, n = regionOpeningStateList.size(); k < n; k++) {
+            regionOpenInfos.add(new Triple<HRegionInfo, Integer,  List<ServerName>>(
-              RegionOpeningState openingState = regionOpeningStateList.get(k);
+              region, nodeVersion, favoredNodes));
-              if (openingState != RegionOpeningState.OPENED) {
+          }
-                HRegionInfo region = regionOpenInfos.get(k).getFirst();
+        }
-                if (openingState == RegionOpeningState.ALREADY_OPENED) {
+
-                  processAlreadyOpenedRegion(region, destination);
+        // Move on to open regions.
-                } else if (openingState == RegionOpeningState.FAILED_OPENING) {
+        try {
-                  // Failed opening this region, reassign it later
+          // Send OPEN RPC. If it fails on a IOE or RemoteException, the
-                  failedToOpenRegions.add(region);
+          // TimeoutMonitor will pick up the pieces.
-                } else {
+          long maxWaitTime = System.currentTimeMillis() +
-                  LOG.warn("THIS SHOULD NOT HAPPEN: unknown opening state "
+            this.server.getConfiguration().
-                    + openingState + " in assigning region " + region);
+              getLong("hbase.regionserver.rpc.startup.waittime", 60000);
+          for (int i = 1; i <= maximumAttempts && !server.isStopped(); i++) {
+            try {
+              List<RegionOpeningState> regionOpeningStateList = serverManager
+                .sendRegionOpen(destination, regionOpenInfos);
+              if (regionOpeningStateList == null) {
+                // Failed getting RPC connection to this server
+                return false;
+              }
+              for (int k = 0, n = regionOpeningStateList.size(); k < n; k++) {
+                RegionOpeningState openingState = regionOpeningStateList.get(k);
+                if (openingState != RegionOpeningState.OPENED) {
+                  HRegionInfo region = regionOpenInfos.get(k).getFirst();
+                  if (openingState == RegionOpeningState.ALREADY_OPENED) {
+                    processAlreadyOpenedRegion(region, destination);
+                  } else if (openingState == RegionOpeningState.FAILED_OPENING) {
+                    // Failed opening this region, reassign it later
+                    failedToOpenRegions.add(region);
+                  } else {
+                    LOG.warn("THIS SHOULD NOT HAPPEN: unknown opening state "
+                      + openingState + " in assigning region " + region);
+                  }
                 }
               }
-            }
+              break;
-            break;
+            } catch (IOException e) {
-          } catch (IOException e) {
+              if (e instanceof RemoteException) {
-            if (e instanceof RemoteException) {
+                e = ((RemoteException)e).unwrapRemoteException();
-              e = ((RemoteException)e).unwrapRemoteException();
+              }
-            }
+              if (e instanceof RegionServerStoppedException) {
-            if (e instanceof RegionServerStoppedException) {
+                LOG.warn("The region server was shut down, ", e);
-              LOG.warn("The region server was shut down, ", e);
+                // No need to retry, the region server is a goner.
-              // No need to retry, the region server is a goner.
+                return false;
-              return false;
+              } else if (e instanceof ServerNotRunningYetException) {
-            } else if (e instanceof ServerNotRunningYetException) {
+                long now = System.currentTimeMillis();
-              long now = System.currentTimeMillis();
+                if (now < maxWaitTime) {
-              if (now < maxWaitTime) {
+                  LOG.debug("Server is not yet up; waiting up to " +
-                LOG.debug("Server is not yet up; waiting up to " +
+                    (maxWaitTime - now) + "ms", e);
-                  (maxWaitTime - now) + "ms", e);
+                  Thread.sleep(100);
-                Thread.sleep(100);
+                  i--; // reset the try count
-                i--; // reset the try count
+                  continue;
+                }
+              } else if (e instanceof java.net.SocketTimeoutException
+                  && this.serverManager.isServerOnline(destination)) {
+                // In case socket is timed out and the region server is still online,
+                // the openRegion RPC could have been accepted by the server and
+                // just the response didn't go through.  So we will retry to
+                // open the region on the same server.
+                if (LOG.isDebugEnabled()) {
+                  LOG.debug("Bulk assigner openRegion() to " + destination
+                    + " has timed out, but the regions might"
+                    + " already be opened on it.", e);
+                }
                 continue;
               }
-            } else if (e instanceof java.net.SocketTimeoutException
+              throw e;
-                && this.serverManager.isServerOnline(destination)) {
-              // In case socket is timed out and the region server is still online,
-              // the openRegion RPC could have been accepted by the server and
-              // just the response didn't go through.  So we will retry to
-              // open the region on the same server.
-              if (LOG.isDebugEnabled()) {
-                LOG.debug("Bulk assigner openRegion() to " + destination
-                  + " has timed out, but the regions might"
-                  + " already be opened on it.", e);
-              }
-              continue;
             }
-            throw e;
           }
+        } catch (IOException e) {
+          // Can be a socket timeout, EOF, NoRouteToHost, etc
+          LOG.info("Unable to communicate with the region server in order" +
+            " to assign regions", e);
+          return false;
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
+        }
+      } finally {
+        for (Lock lock : locks.values()) {
+          lock.unlock();
         }
-      } catch (IOException e) {
-        // Can be a socket timeout, EOF, NoRouteToHost, etc
-        LOG.info("Unable to communicate with the region server in order" +
-          " to assign regions", e);
-        return false;
-      } catch (InterruptedException e) {
-        throw new RuntimeException(e);
       }
-    } finally {
-      for (Lock lock : locks.values()) {
-        lock.unlock();
-      }
-    }
 
-    if (!failedToOpenRegions.isEmpty()) {
+      if (!failedToOpenRegions.isEmpty()) {
-      for (HRegionInfo region : failedToOpenRegions) {
+        for (HRegionInfo region : failedToOpenRegions) {
-        invokeAssign(region);
+          invokeAssign(region);
+        }
       }
+      LOG.debug("Bulk assigning done for " + destination.toString());
+      return true;
+    } finally {
+      metricsAssignmentManager.updateBulkAssignTime(EnvironmentEdgeManager.currentTimeMillis() - startTime);
     }
-    LOG.debug("Bulk assigning done for " + destination.toString());
-    return true;
   }
 
   /**
@@ -1821,6 +1826,8 @@ public class AssignmentManager extends ZooKeeperListener {
    */
   private void assign(RegionState state,
       final boolean setOfflineInZK, final boolean forceNewPlan) {
+    long startTime = EnvironmentEdgeManager.currentTimeMillis();
+    try {
     RegionState currentState = state;
     int versionOfOfflineNode = -1;
     RegionPlan plan = null;
@@ -2029,6 +2036,9 @@ public class AssignmentManager extends ZooKeeperListener {
     if (!tomActivated) {
       regionStates.updateRegionState(region, RegionState.State.FAILED_OPEN);
     }
+    } finally {
+      metricsAssignmentManager.updateAssignmentTime(EnvironmentEdgeManager.currentTimeMillis() - startTime);
+    }
   }
 
   private void processAlreadyOpenedRegion(HRegionInfo region, ServerName sn) {
@@ -2804,10 +2814,10 @@ public class AssignmentManager extends ZooKeeperListener {
         oldestRITTime = ritTime;
       }
     }
-    if (this.metricsMaster != null) {
+    if (this.metricsAssignmentManager != null) {
-      this.metricsMaster.updateRITOldestAge(oldestRITTime);
+      this.metricsAssignmentManager.updateRITOldestAge(oldestRITTime);
-      this.metricsMaster.updateRITCount(totalRITs);
+      this.metricsAssignmentManager.updateRITCount(totalRITs);
-      this.metricsMaster.updateRITCountOverThreshold(totalRITsOverThreshold);
+      this.metricsAssignmentManager.updateRITCountOverThreshold(totalRITsOverThreshold);
     }
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -764,7 +764,7 @@ MasterServices, Server {
 
     this.masterActiveTime = System.currentTimeMillis();
     // TODO: Do this using Dependency Injection, using PicoContainer, Guice or Spring.
-    this.fileSystemManager = new MasterFileSystem(this, this, metricsMaster, masterRecovery);
+    this.fileSystemManager = new MasterFileSystem(this, this, masterRecovery);
 
     this.tableDescriptors =
       new FSTableDescriptors(this.fileSystemManager.getFileSystem(),

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterFileSystem.java

@@ -72,7 +72,7 @@ public class MasterFileSystem {
   // master status
   Server master;
   // metrics for master
-  MetricsMaster metricsMaster;
+  private final MetricsMasterFileSystem metricsMasterFilesystem = new MetricsMasterFileSystem();
   // Persisted unique cluster ID
   private ClusterId clusterId;
   // Keep around for convenience.
@@ -103,13 +103,11 @@ public class MasterFileSystem {
     }
   };
 
-  public MasterFileSystem(Server master, MasterServices services,
+  public MasterFileSystem(Server master, MasterServices services, boolean masterRecovery)
-      MetricsMaster metricsMaster, boolean masterRecovery)
   throws IOException {
     this.conf = master.getConfiguration();
     this.master = master;
     this.services = services;
-    this.metricsMaster = metricsMaster;
     // Set filesystem to be that of this.rootdir else we get complaints about
     // mismatched filesystems if hbase.rootdir is hdfs and fs.defaultFS is
     // default localfs.  Presumption is that rootdir is fully-qualified before
@@ -410,11 +408,11 @@ public class MasterFileSystem {
     splitLogSize = splitLogManager.splitLogDistributed(serverNames, logDirs, filter);
     splitTime = EnvironmentEdgeManager.currentTimeMillis() - splitTime;
 
-    if (this.metricsMaster != null) {
+    if (this.metricsMasterFilesystem != null) {
       if (filter == META_FILTER) {
-        this.metricsMaster.addMetaWALSplit(splitTime, splitLogSize);
+        this.metricsMasterFilesystem.addMetaWALSplit(splitTime, splitLogSize);
       } else {
-        this.metricsMaster.addSplit(splitTime, splitLogSize);
+        this.metricsMasterFilesystem.addSplit(splitTime, splitLogSize);
       }
     }
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MetricsAssignmentManager.java

@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
+
+public class MetricsAssignmentManager {
+
+  private final MetricsAssignmentManagerSource assignmentManagerSource;
+
+  public MetricsAssignmentManager() {
+    assignmentManagerSource = CompatibilitySingletonFactory.getInstance(
+        MetricsAssignmentManagerSource.class);
+  }
+
+  public void updateAssignmentTime(long time) {
+    assignmentManagerSource.updateAssignmentTime(time);
+  }
+
+  public void updateBulkAssignTime(long time) {
+    assignmentManagerSource.updateBulkAssignTime(time);
+  }
+
+  /**
+   * set new value for number of regions in transition.
+   * @param ritCount
+   */
+  public void updateRITCount(int ritCount) {
+    assignmentManagerSource.setRIT(ritCount);
+  }
+
+  /**
+   * update RIT count that are in this state for more than the threshold
+   * as defined by the property rit.metrics.threshold.time.
+   * @param ritCountOverThreshold
+   */
+  public void updateRITCountOverThreshold(int ritCountOverThreshold) {
+    assignmentManagerSource.setRITCountOverThreshold(ritCountOverThreshold);
+  }
+  /**
+   * update the timestamp for oldest region in transition metrics.
+   * @param timestamp
+   */
+  public void updateRITOldestAge(long timestamp) {
+    assignmentManagerSource.setRITOldestAge(timestamp);
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MetricsMaster.java

@@ -15,6 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.hadoop.hbase.master;
 
 import org.apache.commons.logging.Log;
@@ -48,79 +49,10 @@ public class MetricsMaster {
     return masterSource;
   }
 
-  /**
-   * Record a single instance of a split
-   * @param time time that the split took
-   * @param size length of original HLogs that were split
-   */
-  public synchronized void addSplit(long time, long size) {
-    masterSource.updateSplitTime(time);
-    masterSource.updateSplitSize(size);
-  }
-
-  /**
-   * Record a single instance of a split
-   * @param time time that the split took
-   * @param size length of original HLogs that were split
-   */
-  public synchronized void addMetaWALSplit(long time, long size) {
-    masterSource.updateMetaWALSplitTime(time);
-    masterSource.updateMetaWALSplitSize(size);
-  }
-
   /**
    * @param inc How much to add to requests.
    */
   public void incrementRequests(final int inc) {
     masterSource.incRequests(inc);
-
-  }
-
-  /**
-   * set new value for number of regions in transition.
-   * @param ritCount
-   */
-  public void updateRITCount(int ritCount) {
-    masterSource.setRIT(ritCount);
-  }
-
-  /**
-   * update RIT count that are in this state for more than the threshold
-   * as defined by the property rit.metrics.threshold.time.
-   * @param ritCountOverThreshold
-   */
-  public void updateRITCountOverThreshold(int ritCountOverThreshold) {
-    masterSource.setRITCountOverThreshold(ritCountOverThreshold);
-  }
-  /**
-   * update the timestamp for oldest region in transition metrics.
-   * @param timestamp
-   */
-  public void updateRITOldestAge(long timestamp) {
-    masterSource.setRITOldestAge(timestamp);
-  }
-
-  /**
-   * Record a single instance of a snapshot
-   * @param time time that the snapshot took
-   */
-  public void addSnapshot(long time) {
-    masterSource.updateSnapshotTime(time);
-  }
-
-  /**
-   * Record a single instance of a snapshot
-   * @param time time that the snapshot restore took
-   */
-  public void addSnapshotRestore(long time) {
-    masterSource.updateSnapshotRestoreTime(time);
-  }
-
-  /**
-   * Record a single instance of a snapshot cloned table
-   * @param time time that the snapshot clone took
-   */
-  public void addSnapshotClone(long time) {
-    masterSource.updateSnapshotCloneTime(time);
   }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MetricsMasterFileSystem.java

@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
+
+public class MetricsMasterFileSystem {
+
+  private final MetricsMasterFileSystemSource source;
+
+  public MetricsMasterFileSystem() {
+    source = CompatibilitySingletonFactory.getInstance(MetricsMasterFileSystemSource.class);
+  }
+
+  /**
+   * Record a single instance of a split
+   * @param time time that the split took
+   * @param size length of original HLogs that were split
+   */
+  public synchronized void addSplit(long time, long size) {
+    source.updateSplitTime(time);
+    source.updateSplitSize(size);
+  }
+
+  /**
+   * Record a single instance of a split
+   * @param time time that the split took
+   * @param size length of original HLogs that were split
+   */
+  public synchronized void addMetaWALSplit(long time, long size) {
+    source.updateMetaWALSplitTime(time);
+    source.updateMetaWALSplitSize(size);
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MetricsSnapshot.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.hadoop.hbase.master;
+
+import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
+
+public class MetricsSnapshot {
+
+  private final MetricsSnapshotSource source;
+
+  public MetricsSnapshot() {
+    source = CompatibilitySingletonFactory.getInstance(MetricsSnapshotSource.class);
+  }
+
+  /**
+   * Record a single instance of a snapshot
+   * @param time time that the snapshot took
+   */
+  public void addSnapshot(long time) {
+    source.updateSnapshotTime(time);
+  }
+
+  /**
+   * Record a single instance of a snapshot
+   * @param time time that the snapshot restore took
+   */
+  public void addSnapshotRestore(long time) {
+    source.updateSnapshotRestoreTime(time);
+  }
+
+  /**
+   * Record a single instance of a snapshot cloned table
+   * @param time time that the snapshot clone took
+   */
+  public void addSnapshotClone(long time) {
+    source.updateSnapshotCloneTime(time);
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BaseLoadBalancer.java

@@ -342,6 +342,8 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
   private Configuration config;
   private static final Random RANDOM = new Random(System.currentTimeMillis());
   private static final Log LOG = LogFactory.getLog(BaseLoadBalancer.class);
+
+  protected final MetricsBalancer metricsBalancer = new MetricsBalancer();
   protected MasterServices services;
 
   @Override
@@ -409,6 +411,8 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
    */
   public Map<ServerName, List<HRegionInfo>> roundRobinAssignment(List<HRegionInfo> regions,
       List<ServerName> servers) {
+    metricsBalancer.incrMiscInvocations();
+
     if (regions.isEmpty() || servers.isEmpty()) {
       return null;
     }
@@ -452,6 +456,8 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
    */
   public Map<HRegionInfo, ServerName> immediateAssignment(List<HRegionInfo> regions,
       List<ServerName> servers) {
+    metricsBalancer.incrMiscInvocations();
+
     Map<HRegionInfo, ServerName> assignments = new TreeMap<HRegionInfo, ServerName>();
     for (HRegionInfo region : regions) {
       assignments.put(region, randomAssignment(region, servers));
@@ -463,6 +469,8 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
    * Used to assign a single region to a random server.
    */
   public ServerName randomAssignment(HRegionInfo regionInfo, List<ServerName> servers) {
+    metricsBalancer.incrMiscInvocations();
+
     if (servers == null || servers.isEmpty()) {
       LOG.warn("Wanted to do random assignment but no servers to assign to");
       return null;
@@ -489,6 +497,9 @@ public abstract class BaseLoadBalancer implements LoadBalancer {
    */
   public Map<ServerName, List<HRegionInfo>> retainAssignment(Map<HRegionInfo, ServerName> regions,
       List<ServerName> servers) {
+    // Update metrics
+    metricsBalancer.incrMiscInvocations();
+
     // Group all of the old assignments by their hostname.
     // We can't group directly by ServerName since the servers all have
     // new start-codes.

hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/MetricsBalancer.java

@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.master.balancer;
+
+import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
+
+/**
+ * Faced for exposing metrics about the balancer.
+ */
+public class MetricsBalancer {
+
+  private final MetricsBalancerSource source;
+
+  public MetricsBalancer() {
+    source = CompatibilitySingletonFactory.getInstance(MetricsBalancerSource.class);
+  }
+
+  public void balanceCluster(long time) {
+    source.updateBalanceCluster(time);
+  }
+
+  public void incrMiscInvocations() {
+    source.incrMiscInvocations();
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/StochasticLoadBalancer.java

@@ -246,6 +246,7 @@ public class StochasticLoadBalancer extends BaseLoadBalancer {
 
     long endTime = EnvironmentEdgeManager.currentTimeMillis();
 
+    metricsBalancer.balanceCluster(endTime - startTime);
 
     if (initCost > currentCost) {
       List<RegionPlan> plans = createRegionPlans(cluster);

hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/CloneSnapshotHandler.java

@@ -36,7 +36,7 @@ import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.errorhandling.ForeignException;
 import org.apache.hadoop.hbase.errorhandling.ForeignExceptionDispatcher;
 import org.apache.hadoop.hbase.master.MasterServices;
-import org.apache.hadoop.hbase.master.MetricsMaster;
+import org.apache.hadoop.hbase.master.MetricsSnapshot;
 import org.apache.hadoop.hbase.master.SnapshotSentinel;
 import org.apache.hadoop.hbase.master.handler.CreateTableHandler;
 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
@@ -48,7 +48,6 @@ import org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper;
 import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
 
 import com.google.common.base.Preconditions;
-import org.apache.hadoop.hbase.util.FSUtils;
 
 /**
  * Handler to Clone a snapshot.
@@ -65,17 +64,15 @@ public class CloneSnapshotHandler extends CreateTableHandler implements Snapshot
   private final SnapshotDescription snapshot;
 
   private final ForeignExceptionDispatcher monitor;
-  private final MetricsMaster metricsMaster;
+  private final MetricsSnapshot metricsSnapshot = new MetricsSnapshot();
   private final MonitoredTask status;
 
   private volatile boolean stopped = false;
 
   public CloneSnapshotHandler(final MasterServices masterServices,
-      final SnapshotDescription snapshot, final HTableDescriptor hTableDescriptor,
+      final SnapshotDescription snapshot, final HTableDescriptor hTableDescriptor) {
-      final MetricsMaster metricsMaster) {
     super(masterServices, masterServices.getMasterFileSystem(), hTableDescriptor,
       masterServices.getConfiguration(), null, masterServices);
-    this.metricsMaster = metricsMaster;
 
     // Snapshot information
     this.snapshot = snapshot;
@@ -145,7 +142,7 @@ public class CloneSnapshotHandler extends CreateTableHandler implements Snapshot
     } else {
       status.markComplete("Snapshot '"+ snapshot.getName() +"' clone completed and table enabled!");
     }
-    metricsMaster.addSnapshotClone(status.getCompletionTimestamp() - status.getStartTime());
+    metricsSnapshot.addSnapshotClone(status.getCompletionTimestamp() - status.getStartTime());
     super.completed(exception);
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/DisabledTableSnapshotHandler.java

@@ -64,8 +64,8 @@ public class DisabledTableSnapshotHandler extends TakeSnapshotHandler {
    * @param masterServices master services provider
    */
   public DisabledTableSnapshotHandler(SnapshotDescription snapshot,
-      final MasterServices masterServices, final MetricsMaster metricsMaster) {
+      final MasterServices masterServices) {
-    super(snapshot, masterServices, metricsMaster);
+    super(snapshot, masterServices);
 
     // setup the timer
     timeoutInjector = TakeSnapshotUtils.getMasterTimerAndBindToMonitor(snapshot, conf, monitor);

hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/EnabledTableSnapshotHandler.java

@@ -50,8 +50,8 @@ public class EnabledTableSnapshotHandler extends TakeSnapshotHandler {
   private final ProcedureCoordinator coordinator;
 
   public EnabledTableSnapshotHandler(SnapshotDescription snapshot, MasterServices master,
-      final SnapshotManager manager, final MetricsMaster metricsMaster) {
+      final SnapshotManager manager) {
-    super(snapshot, master, metricsMaster);
+    super(snapshot, master);
     this.coordinator = manager.getCoordinator();
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/RestoreSnapshotHandler.java

@@ -38,7 +38,7 @@ import org.apache.hadoop.hbase.errorhandling.ForeignExceptionDispatcher;
 import org.apache.hadoop.hbase.executor.EventType;
 import org.apache.hadoop.hbase.master.MasterFileSystem;
 import org.apache.hadoop.hbase.master.MasterServices;
-import org.apache.hadoop.hbase.master.MetricsMaster;
+import org.apache.hadoop.hbase.master.MetricsSnapshot;
 import org.apache.hadoop.hbase.master.SnapshotSentinel;
 import org.apache.hadoop.hbase.master.handler.TableEventHandler;
 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
@@ -48,7 +48,6 @@ import org.apache.hadoop.hbase.snapshot.ClientSnapshotDescriptionUtils;
 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotHelper;
 import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
-import org.apache.hadoop.hbase.util.FSUtils;
 
 /**
  * Handler to Restore a snapshot.
@@ -64,16 +63,14 @@ public class RestoreSnapshotHandler extends TableEventHandler implements Snapsho
   private final SnapshotDescription snapshot;
 
   private final ForeignExceptionDispatcher monitor;
-  private final MetricsMaster metricsMaster;
+  private final MetricsSnapshot metricsSnapshot = new MetricsSnapshot();
   private final MonitoredTask status;
 
   private volatile boolean stopped = false;
 
   public RestoreSnapshotHandler(final MasterServices masterServices,
-      final SnapshotDescription snapshot, final HTableDescriptor htd,
+      final SnapshotDescription snapshot, final HTableDescriptor htd) throws IOException {
-      final MetricsMaster metricsMaster) throws IOException {
     super(EventType.C_M_RESTORE_SNAPSHOT, htd.getTableName(), masterServices, masterServices);
-    this.metricsMaster = metricsMaster;
 
     // Snapshot information
     this.snapshot = snapshot;
@@ -153,7 +150,7 @@ public class RestoreSnapshotHandler extends TableEventHandler implements Snapsho
     } else {
       status.markComplete("Restore snapshot '"+ snapshot.getName() +"'!");
     }
-    metricsMaster.addSnapshotRestore(status.getCompletionTimestamp() - status.getStartTime());
+    metricsSnapshot.addSnapshotRestore(status.getCompletionTimestamp() - status.getStartTime());
     super.completed(exception);
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/SnapshotManager.java

@@ -450,7 +450,7 @@ public class SnapshotManager implements Stoppable {
 
     // Take the snapshot of the disabled table
     DisabledTableSnapshotHandler handler =
-        new DisabledTableSnapshotHandler(snapshot, master, metricsMaster);
+        new DisabledTableSnapshotHandler(snapshot, master);
     snapshotTable(snapshot, handler);
   }
 
@@ -466,7 +466,7 @@ public class SnapshotManager implements Stoppable {
 
     // Take the snapshot of the enabled table
     EnabledTableSnapshotHandler handler =
-        new EnabledTableSnapshotHandler(snapshot, master, this, metricsMaster);
+        new EnabledTableSnapshotHandler(snapshot, master, this);
     snapshotTable(snapshot, handler);
   }
 
@@ -645,7 +645,7 @@ public class SnapshotManager implements Stoppable {
 
     try {
       CloneSnapshotHandler handler =
-        new CloneSnapshotHandler(master, snapshot, hTableDescriptor, metricsMaster).prepare();
+        new CloneSnapshotHandler(master, snapshot, hTableDescriptor).prepare();
       this.executorService.submit(handler);
       this.restoreHandlers.put(tableName, handler);
     } catch (Exception e) {
@@ -738,7 +738,7 @@ public class SnapshotManager implements Stoppable {
 
     try {
       RestoreSnapshotHandler handler =
-        new RestoreSnapshotHandler(master, snapshot, hTableDescriptor, metricsMaster).prepare();
+        new RestoreSnapshotHandler(master, snapshot, hTableDescriptor).prepare();
       this.executorService.submit(handler);
       restoreHandlers.put(tableName, handler);
     } catch (Exception e) {

hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/TakeSnapshotHandler.java

@@ -41,13 +41,12 @@ import org.apache.hadoop.hbase.errorhandling.ForeignExceptionSnare;
 import org.apache.hadoop.hbase.executor.EventHandler;
 import org.apache.hadoop.hbase.executor.EventType;
 import org.apache.hadoop.hbase.master.MasterServices;
-import org.apache.hadoop.hbase.master.MetricsMaster;
+import org.apache.hadoop.hbase.master.MetricsSnapshot;
 import org.apache.hadoop.hbase.master.SnapshotSentinel;
 import org.apache.hadoop.hbase.master.TableLockManager;
 import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
-import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
 import org.apache.hadoop.hbase.snapshot.ClientSnapshotDescriptionUtils;
 import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
@@ -72,7 +71,7 @@ public abstract class TakeSnapshotHandler extends EventHandler implements Snapsh
 
   // none of these should ever be null
   protected final MasterServices master;
-  protected final MetricsMaster metricsMaster;
+  protected final MetricsSnapshot metricsSnapshot = new MetricsSnapshot();
   protected final SnapshotDescription snapshot;
   protected final Configuration conf;
   protected final FileSystem fs;
@@ -90,14 +89,12 @@ public abstract class TakeSnapshotHandler extends EventHandler implements Snapsh
    * @param snapshot descriptor of the snapshot to take
    * @param masterServices master services provider
    */
-  public TakeSnapshotHandler(SnapshotDescription snapshot, final MasterServices masterServices,
+  public TakeSnapshotHandler(SnapshotDescription snapshot, final MasterServices masterServices) {
-      final MetricsMaster metricsMaster) {
     super(masterServices, EventType.C_M_SNAPSHOT_TABLE);
     assert snapshot != null : "SnapshotDescription must not be nul1";
     assert masterServices != null : "MasterServices must not be nul1";
 
     this.master = masterServices;
-    this.metricsMaster = metricsMaster;
     this.snapshot = snapshot;
     this.snapshotTable = TableName.valueOf(snapshot.getTable());
     this.conf = this.master.getConfiguration();
@@ -187,7 +184,7 @@ public abstract class TakeSnapshotHandler extends EventHandler implements Snapsh
       completeSnapshot(this.snapshotDir, this.workingDir, this.fs);
       status.markComplete("Snapshot " + snapshot.getName() + " of table " + snapshotTable
           + " completed");
-      metricsMaster.addSnapshot(status.getCompletionTimestamp() - status.getStartTime());
+      metricsSnapshot.addSnapshot(status.getCompletionTimestamp() - status.getStartTime());
     } catch (Exception e) {
       status.abort("Failed to complete snapshot " + snapshot.getName() + " on table " +
           snapshotTable + " because " + e.getMessage());

hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestCatalogJanitor.java

@@ -178,7 +178,7 @@ public class TestCatalogJanitor {
     private final AssignmentManager asm;
 
     MockMasterServices(final Server server) throws IOException {
-      this.mfs = new MasterFileSystem(server, this, null, false);
+      this.mfs = new MasterFileSystem(server, this, false);
       this.asm = Mockito.mock(AssignmentManager.class);
     }