hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From st...@apache.org
Subject svn commit: r1329958 - in /hbase/branches/0.94/src: main/java/org/apache/hadoop/hbase/regionserver/ main/java/org/apache/hadoop/hbase/regionserver/metrics/ test/java/org/apache/hadoop/hbase/regionserver/
Date Tue, 24 Apr 2012 19:28:40 GMT
Author: stack
Date: Tue Apr 24 19:28:40 2012
New Revision: 1329958

URL: http://svn.apache.org/viewvc?rev=1329958&view=rev
Log:
HBASE-5836 Backport per region metrics from HBASE-3614 to 0.94.1

Added:
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/OperationMetrics.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionMetricsStorage.java
Modified:
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerDynamicMetrics.java
    hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java
    hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerMetrics.java

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1329958&r1=1329957&r2=1329958&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Tue
Apr 24 19:28:40 2012
@@ -38,6 +38,7 @@ import java.util.Map;
 import java.util.NavigableMap;
 import java.util.NavigableSet;
 import java.util.Random;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.UUID;
 import java.util.concurrent.Callable;
@@ -104,6 +105,7 @@ import org.apache.hadoop.hbase.ipc.HBase
 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
 import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
+import org.apache.hadoop.hbase.regionserver.metrics.OperationMetrics;
 import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics;
 import org.apache.hadoop.hbase.regionserver.wal.HLog;
 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
@@ -322,83 +324,8 @@ public class HRegion implements HeapSize
   public final static String REGIONINFO_FILE = ".regioninfo";
   private HTableDescriptor htableDescriptor = null;
   private RegionSplitPolicy splitPolicy;
+  private final OperationMetrics opMetrics;
 
-  // for simple numeric metrics (# of blocks read from block cache)
-  public static final ConcurrentMap<String, AtomicLong> numericMetrics = new ConcurrentHashMap<String,
AtomicLong>();
-
-  // for simple numeric metrics (current block cache size)
-  // These ones are not reset to zero when queried, unlike the previous.
-  public static final ConcurrentMap<String, AtomicLong> numericPersistentMetrics =
new ConcurrentHashMap<String, AtomicLong>();
-
-  /**
-   * Used for metrics where we want track a metrics (such as latency) over a
-   * number of operations.
-   */
-  public static final ConcurrentMap<String, Pair<AtomicLong, AtomicInteger>>
-      timeVaryingMetrics = new ConcurrentHashMap<String, 
-          Pair<AtomicLong, AtomicInteger>>();
-
-  public static void incrNumericMetric(String key, long amount) {
-    AtomicLong oldVal = numericMetrics.get(key);
-    if (oldVal == null) {
-      oldVal = numericMetrics.putIfAbsent(key, new AtomicLong(amount));
-      if (oldVal == null)
-        return;
-    }
-    oldVal.addAndGet(amount);
-  }
-
-  public static void setNumericMetric(String key, long amount) {
-    numericMetrics.put(key, new AtomicLong(amount));
-  }
-
-  public static void incrTimeVaryingMetric(String key, long amount) {
-    Pair<AtomicLong, AtomicInteger> oldVal = timeVaryingMetrics.get(key);
-    if (oldVal == null) {
-      oldVal = timeVaryingMetrics.putIfAbsent(key,
-          new Pair<AtomicLong, AtomicInteger>(new AtomicLong(amount),
-              new AtomicInteger(1)));
-      if (oldVal == null)
-        return;
-    }
-    oldVal.getFirst().addAndGet(amount); // total time
-    oldVal.getSecond().incrementAndGet(); // increment ops by 1
-  }
-
-  public static void incrNumericPersistentMetric(String key, long amount) {
-    AtomicLong oldVal = numericPersistentMetrics.get(key);
-    if (oldVal == null) {
-      oldVal = numericPersistentMetrics
-          .putIfAbsent(key, new AtomicLong(amount));
-      if (oldVal == null)
-        return;
-    }
-    oldVal.addAndGet(amount);
-  }
-
-  public static long getNumericMetric(String key) {
-    AtomicLong m = numericMetrics.get(key);
-    if (m == null)
-      return 0;
-    return m.get();
-  }
-
-  public static Pair<Long, Integer> getTimeVaryingMetric(String key) {
-    Pair<AtomicLong, AtomicInteger> pair = timeVaryingMetrics.get(key);
-    if (pair == null) {
-      return new Pair<Long, Integer>(0L, 0);
-    }
-
-    return new Pair<Long, Integer>(pair.getFirst().get(),
-        pair.getSecond().get());
-  }
-
-  static long getNumericPersistentMetric(String key) {
-    AtomicLong m = numericPersistentMetrics.get(key);
-    if (m == null)
-      return 0;
-    return m.get();
-  }
 
   /**
    * Should only be used for testing purposes
@@ -419,6 +346,7 @@ public class HRegion implements HeapSize
     this.threadWakeFrequency = 0L;
     this.coprocessorHost = null;
     this.scannerReadPoints = new ConcurrentHashMap<RegionScanner, Long>();
+    this.opMetrics = new OperationMetrics();
   }
 
   /**
@@ -463,6 +391,7 @@ public class HRegion implements HeapSize
     setHTableSpecificConf();
     this.regiondir = getRegionDir(this.tableDir, encodedNameStr);
     this.scannerReadPoints = new ConcurrentHashMap<RegionScanner, Long>();
+    this.opMetrics = new OperationMetrics(conf, this.regionInfo);
 
     /*
      * timestamp.slop provides a server-side constraint on the timestamp. This
@@ -1851,11 +1780,7 @@ public class HRegion implements HeapSize
       coprocessorHost.postDelete(delete, walEdit, writeToWAL);
     }
     final long after = EnvironmentEdgeManager.currentTimeMillis();
-    final String metricPrefix = SchemaMetrics.generateSchemaMetricsPrefix(
-        getTableDesc().getNameAsString(), familyMap.keySet());
-    if (!metricPrefix.isEmpty()) {
-      HRegion.incrTimeVaryingMetric(metricPrefix + "delete_", after - now);
-    }
+    this.opMetrics.updateDeleteMetrics(familyMap.keySet(), after-now);
 
     if (flush) {
       // Request a cache flush.  Do it outside update lock.
@@ -2001,9 +1926,10 @@ public class HRegion implements HeapSize
   @SuppressWarnings("unchecked")
   private long doMiniBatchPut(
       BatchOperationInProgress<Pair<Put, Integer>> batchOp) throws IOException
{
-    String metricPrefix = null;
     final String tableName = getTableDesc().getNameAsString();
 
+    // The set of columnFamilies first seen.
+    Set<byte[]> cfSet = null;
     // variable to note if all Put items are for the same CF -- metrics related
     boolean cfSetConsistent = true;
     long startTimeMs = EnvironmentEdgeManager.currentTimeMillis();
@@ -2085,19 +2011,13 @@ public class HRegion implements HeapSize
         lastIndexExclusive++;
         numReadyToWrite++;
 
-        // If first time around, designate a prefix for metrics based on the CF
-        // set. After that, watch for inconsistencies.
-        final String curMetricPrefix =
-            SchemaMetrics.generateSchemaMetricsPrefix(tableName,
-                put.getFamilyMap().keySet());
-
-        if (metricPrefix == null) {
-          metricPrefix = curMetricPrefix;
-        } else if (cfSetConsistent && !metricPrefix.equals(curMetricPrefix)) {
-          // The column family set for this batch put is undefined.
-          cfSetConsistent = false;
-          metricPrefix = SchemaMetrics.generateSchemaMetricsPrefix(tableName,
-              SchemaMetrics.UNKNOWN);
+        // If Column Families stay consistent through out all of the
+        // individual puts then metrics can be reported as a mutliput across
+        // column families in the first put.
+        if (cfSet == null) {
+          cfSet = put.getFamilyMap().keySet();
+        } else {
+          cfSetConsistent = cfSetConsistent && put.getFamilyMap().keySet().equals(cfSet);
         }
       }
 
@@ -2242,11 +2162,12 @@ public class HRegion implements HeapSize
 
       // do after lock
       final long endTimeMs = EnvironmentEdgeManager.currentTimeMillis();
-      if (metricPrefix == null) {
-        metricPrefix = SchemaMetrics.CF_BAD_FAMILY_PREFIX;
-      }
-      HRegion.incrTimeVaryingMetric(metricPrefix + "multiput_",
-          endTimeMs - startTimeMs);
+            
+      // See if the column families were consistent through the whole thing.
+      // if they were then keep them. If they were not then pass a null.
+      // null will be treated as unknown.
+      final Set<byte[]> keptCfs = cfSetConsistent ? cfSet : null;
+      this.opMetrics.updateMultiPutMetrics(keptCfs, endTimeMs - startTimeMs);
 
       if (!success) {
         for (int i = firstIndex; i < lastIndexExclusive; i++) {
@@ -2501,12 +2422,8 @@ public class HRegion implements HeapSize
 
     // do after lock
     final long after = EnvironmentEdgeManager.currentTimeMillis();
-    final String metricPrefix = SchemaMetrics.generateSchemaMetricsPrefix(
-        this.getTableDesc().getNameAsString(), familyMap.keySet());
-    if (!metricPrefix.isEmpty()) {
-      HRegion.incrTimeVaryingMetric(metricPrefix + "put_", after - now);
-    }
-
+    this.opMetrics.updatePutMetrics(familyMap.keySet(), after - now);
+    
     if (flush) {
       // Request a cache flush.  Do it outside update lock.
       requestFlush();
@@ -4128,12 +4045,8 @@ public class HRegion implements HeapSize
 
     // do after lock
     final long after = EnvironmentEdgeManager.currentTimeMillis();
-    final String metricPrefix = SchemaMetrics.generateSchemaMetricsPrefix(
-        this.getTableDesc().getNameAsString(), get.familySet());
-    if (!metricPrefix.isEmpty()) {
-      HRegion.incrTimeVaryingMetric(metricPrefix + "get_", after - now);
-    }
-
+    this.opMetrics.updateGetMetrics(get.familySet(), after - now);
+    
     return results;
   }
 
@@ -4454,6 +4367,10 @@ public class HRegion implements HeapSize
       closeRegionOperation();
     }
 
+    
+    long after = EnvironmentEdgeManager.currentTimeMillis();
+    this.opMetrics.updateAppendMetrics(append.getFamilyMap().keySet(), after - now);
+    
     if (flush) {
       // Request a cache flush. Do it outside update lock.
       requestFlush();
@@ -4564,6 +4481,9 @@ public class HRegion implements HeapSize
     } finally {
       closeRegionOperation();
     }
+    
+    long after = EnvironmentEdgeManager.currentTimeMillis();
+    this.opMetrics.updateIncrementMetrics(increment.getFamilyMap().keySet(), after - now);
 
     if (flush) {
       // Request a cache flush.  Do it outside update lock.
@@ -4661,9 +4581,7 @@ public class HRegion implements HeapSize
 
     // do after lock
     long after = EnvironmentEdgeManager.currentTimeMillis();
-    String metricPrefix = SchemaMetrics.generateSchemaMetricsPrefix(
-        getTableDesc().getName(), family);
-    HRegion.incrTimeVaryingMetric(metricPrefix + "increment_", after - before);
+    this.opMetrics.updateIncrementColumnValueMetrics(family, after - before);
 
     if (flush) {
       // Request a cache flush.  Do it outside update lock.
@@ -4693,7 +4611,7 @@ public class HRegion implements HeapSize
       ClassSize.OBJECT +
       ClassSize.ARRAY +
       30 * ClassSize.REFERENCE + Bytes.SIZEOF_INT +
-      (5 * Bytes.SIZEOF_LONG) +
+      (6 * Bytes.SIZEOF_LONG) +
       Bytes.SIZEOF_BOOLEAN);
 
   public static final long DEEP_OVERHEAD = FIXED_OVERHEAD +

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1329958&r1=1329957&r2=1329958&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
(original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
Tue Apr 24 19:28:40 2012
@@ -128,6 +128,7 @@ import org.apache.hadoop.hbase.regionser
 import org.apache.hadoop.hbase.regionserver.handler.OpenMetaHandler;
 import org.apache.hadoop.hbase.regionserver.handler.OpenRegionHandler;
 import org.apache.hadoop.hbase.regionserver.handler.OpenRootHandler;
+import org.apache.hadoop.hbase.regionserver.metrics.RegionMetricsStorage;
 import org.apache.hadoop.hbase.regionserver.metrics.RegionServerDynamicMetrics;
 import org.apache.hadoop.hbase.regionserver.metrics.RegionServerMetrics;
 import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics;
@@ -1399,7 +1400,7 @@ public class HRegionServer implements HR
     }
 
     for (Entry<String, MutableDouble> e : tempVals.entrySet()) {
-      HRegion.setNumericMetric(e.getKey(), e.getValue().longValue());
+      RegionMetricsStorage.setNumericMetric(e.getKey(), e.getValue().longValue());
     }
 
     this.metrics.stores.set(stores);

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java?rev=1329958&r1=1329957&r2=1329958&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
(original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
Tue Apr 24 19:28:40 2012
@@ -32,6 +32,7 @@ import org.apache.hadoop.hbase.HConstant
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.regionserver.metrics.RegionMetricsStorage;
 import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -367,7 +368,7 @@ class StoreScanner extends NonLazyKeyVal
             this.heap.next();
           }
 
-          HRegion.incrNumericMetric(metricNameGetSize, kv.getLength());
+          RegionMetricsStorage.incrNumericMetric(metricNameGetSize, kv.getLength());
           if (limit > 0 && (results.size() == limit)) {
             break LOOP;
           }

Added: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/OperationMetrics.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/OperationMetrics.java?rev=1329958&view=auto
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/OperationMetrics.java
(added)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/OperationMetrics.java
Tue Apr 24 19:28:40 2012
@@ -0,0 +1,207 @@
+/*
+ * Copyright The Apache Software Foundation
+ *
+ * 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.regionserver.metrics;
+
+import java.util.Set;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * This class provides a simplified interface to expose time varying metrics
+ * about GET/DELETE/PUT/ICV operations on a region and on Column Families. All
+ * metrics are stored in {@link RegionMetricsStorage} and exposed to hadoop
+ * metrics through {@link RegionServerDynamicMetrics}.
+ */
+@InterfaceAudience.Private
+public class OperationMetrics {
+
+  private static final String DELETE_KEY = "delete_";
+  private static final String PUT_KEY = "put_";
+  private static final String GET_KEY = "get_";
+  private static final String ICV_KEY = "incrementColumnValue_";
+  private static final String INCREMENT_KEY = "increment_";
+  private static final String MULTIPUT_KEY = "multiput_";
+  private static final String APPEND_KEY = "append_";
+  
+  /** Conf key controlling whether we should expose metrics.*/
+  private static final String CONF_KEY =
+      "hbase.metrics.exposeOperationTimes";
+
+  private String tableName = null;
+  private String regionName = null;
+  private String regionMetrixPrefix = null;
+  private Configuration conf = null;
+  
+
+  /**
+   * Create a new OperationMetrics
+   * @param conf The Configuration of the HRegion reporting operations coming in.
+   * @param regionInfo The region info
+   */
+  public OperationMetrics(Configuration conf, HRegionInfo regionInfo) {
+    // Configure SchemaMetrics before trying to create a RegionOperationMetrics instance
as
+    // RegionOperationMetrics relies on SchemaMetrics to do naming.
+    if (conf != null) {
+      SchemaMetrics.configureGlobally(conf);
+
+      this.conf = conf;
+      if (regionInfo != null) {
+        this.tableName = regionInfo.getTableNameAsString();
+        this.regionName = regionInfo.getEncodedName();
+      } else {
+        this.tableName = SchemaMetrics.UNKNOWN;
+        this.regionName = SchemaMetrics.UNKNOWN;
+      }
+      this.regionMetrixPrefix =
+          SchemaMetrics.generateRegionMetricsPrefix(this.tableName, this.regionName);
+    }
+  }
+  
+  /**
+   * This is used in creating a testing HRegion where the regionInfo is unknown
+   * @param conf
+   */
+  public OperationMetrics() {
+    this(null, null);
+  }
+
+
+  /**
+   * Update the stats associated with {@link HTable#put(java.util.List)}.
+   * 
+   * @param columnFamilies Set of CF's this multiput is associated with
+   * @param value the time
+   */
+  public void updateMultiPutMetrics(Set<byte[]> columnFamilies, long value) {
+    doUpdateTimeVarying(columnFamilies, MULTIPUT_KEY, value);
+  }
+
+  /**
+   * Update the metrics associated with a {@link Get}
+   * 
+   * @param columnFamilies
+   *          Set of Column Families in this get.
+   * @param value
+   *          the time
+   */
+  public void updateGetMetrics(Set<byte[]> columnFamilies, long value) {
+    doUpdateTimeVarying(columnFamilies, GET_KEY, value);
+  }
+  
+  /**
+   * Update metrics associated with an {@link Increment}
+   * @param columnFamilies
+   * @param value
+   */
+  public void updateIncrementMetrics(Set<byte[]> columnFamilies, long value) {
+    doUpdateTimeVarying(columnFamilies, INCREMENT_KEY, value);
+  }
+  
+  
+  /**
+   * Update the metrics associated with an {@link Append}
+   * @param columnFamilies
+   * @param value
+   */
+  public void updateAppendMetrics(Set<byte[]> columnFamilies, long value) {
+    doUpdateTimeVarying(columnFamilies, APPEND_KEY, value);
+  }
+
+
+  /**
+   * Update the metrics associated with
+   * {@link HTable#incrementColumnValue(byte[], byte[], byte[], long)}
+   * 
+   * @param columnFamily
+   *          The single column family associated with an ICV
+   * @param value
+   *          the time
+   */
+  public void updateIncrementColumnValueMetrics(byte[] columnFamily, long value) {
+    String cfMetricPrefix =
+        SchemaMetrics.generateSchemaMetricsPrefix(this.tableName, Bytes.toString(columnFamily));
+    doSafeIncTimeVarying(cfMetricPrefix, ICV_KEY, value);
+    doSafeIncTimeVarying(this.regionMetrixPrefix, ICV_KEY, value);
+  }
+
+  /**
+   * update metrics associated with a {@link Put}
+   * 
+   * @param columnFamilies
+   *          Set of column families involved.
+   * @param value
+   *          the time.
+   */
+  public void updatePutMetrics(Set<byte[]> columnFamilies, long value) {
+    doUpdateTimeVarying(columnFamilies, PUT_KEY, value);
+  }
+
+  /**
+   * update metrics associated with a {@link Delete}
+   * 
+   * @param columnFamilies
+   * @param value
+   *          the time.
+   */
+  public void updateDeleteMetrics(Set<byte[]> columnFamilies, long value) {
+    doUpdateTimeVarying(columnFamilies, DELETE_KEY, value);
+  }
+
+  /**
+   * Method to send updates for cf and region metrics. This is the normal method
+   * used if the naming of stats and CF's are in line with put/delete/multiput.
+   * 
+   * @param columnFamilies
+   *          the set of column families involved.
+   * @param key
+   *          the metric name.
+   * @param value
+   *          the time.
+   */
+  private void doUpdateTimeVarying(Set<byte[]> columnFamilies, String key, long value)
{
+    String cfPrefix = null;
+    if (columnFamilies != null) {
+      cfPrefix = SchemaMetrics.generateSchemaMetricsPrefix(tableName, columnFamilies);
+    } else {
+      cfPrefix = SchemaMetrics.generateSchemaMetricsPrefix(tableName, SchemaMetrics.UNKNOWN);
+    }
+
+    doSafeIncTimeVarying(cfPrefix, key, value);
+    doSafeIncTimeVarying(this.regionMetrixPrefix, key, value);
+  }
+
+  private void doSafeIncTimeVarying(String prefix, String key, long value) {
+    if (conf.getBoolean(CONF_KEY, true)) {
+      if (prefix != null && !prefix.isEmpty() && key != null && !key.isEmpty())
{
+        RegionMetricsStorage.incrTimeVaryingMetric(prefix + key, value);
+      }
+    }
+  }
+
+}

Added: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionMetricsStorage.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionMetricsStorage.java?rev=1329958&view=auto
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionMetricsStorage.java
(added)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionMetricsStorage.java
Tue Apr 24 19:28:40 2012
@@ -0,0 +1,130 @@
+/*
+ * Copyright The Apache Software Foundation
+ *
+ * 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.regionserver.metrics;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.Pair;
+
+/**
+ * This class if for maintaining the maps used to power metrics for hfiles,
+ * regions, and regionservers. It has methods to mutate and get state of metrics
+ * numbers. These numbers are exposed to Hadoop metrics through
+ * RegionServerDynamicMetrics.
+ */
+@InterfaceAudience.Private
+public class RegionMetricsStorage {
+
+  // for simple numeric metrics (# of blocks read from block cache)
+  private static final ConcurrentMap<String, AtomicLong> numericMetrics =
+      new ConcurrentHashMap<String, AtomicLong>();
+
+  // for simple numeric metrics (current block cache size)
+  // These ones are not reset to zero when queried, unlike the previous.
+  private static final ConcurrentMap<String, AtomicLong> numericPersistentMetrics =
+      new ConcurrentHashMap<String, AtomicLong>();
+
+  /**
+   * Used for metrics where we want track a metrics (such as latency) over a
+   * number of operations.
+   */
+  private static final ConcurrentMap<String, Pair<AtomicLong, AtomicInteger>>
timeVaryingMetrics =
+      new ConcurrentHashMap<String, Pair<AtomicLong, AtomicInteger>>();
+
+  public static Map<String, AtomicLong> getNumericMetrics() {
+    return numericMetrics;
+  }
+
+  public static Map<String, AtomicLong> getNumericPersistentMetrics() {
+    return numericPersistentMetrics;
+  }
+
+  public static Map<String, Pair<AtomicLong, AtomicInteger>> getTimeVaryingMetrics()
{
+    return timeVaryingMetrics;
+  }
+
+  public static void incrNumericMetric(String key, long amount) {
+    AtomicLong oldVal = numericMetrics.get(key);
+    if (oldVal == null) {
+      oldVal = numericMetrics.putIfAbsent(key, new AtomicLong(amount));
+      if (oldVal == null)
+        return;
+    }
+    oldVal.addAndGet(amount);
+  }
+
+  public static void incrTimeVaryingMetric(String key, long amount) {
+    Pair<AtomicLong, AtomicInteger> oldVal = timeVaryingMetrics.get(key);
+    if (oldVal == null) {
+      oldVal =
+          timeVaryingMetrics.putIfAbsent(key, 
+              new Pair<AtomicLong, AtomicInteger>(
+                  new AtomicLong(amount), 
+                  new AtomicInteger(1)));
+      if (oldVal == null)
+        return;
+    }
+    oldVal.getFirst().addAndGet(amount); // total time
+    oldVal.getSecond().incrementAndGet(); // increment ops by 1
+  }
+
+  public static void incrNumericPersistentMetric(String key, long amount) {
+    AtomicLong oldVal = numericPersistentMetrics.get(key);
+    if (oldVal == null) {
+      oldVal = numericPersistentMetrics.putIfAbsent(key, new AtomicLong(amount));
+      if (oldVal == null)
+        return;
+    }
+    oldVal.addAndGet(amount);
+  }
+
+  public static void setNumericMetric(String key, long amount) {
+    numericMetrics.put(key, new AtomicLong(amount));
+  }
+
+  public static long getNumericMetric(String key) {
+    AtomicLong m = numericMetrics.get(key);
+    if (m == null)
+      return 0;
+    return m.get();
+  }
+
+  public static Pair<Long, Integer> getTimeVaryingMetric(String key) {
+    Pair<AtomicLong, AtomicInteger> pair = timeVaryingMetrics.get(key);
+    if (pair == null) {
+      return new Pair<Long, Integer>(0L, 0);
+    }
+
+    return new Pair<Long, Integer>(pair.getFirst().get(), pair.getSecond().get());
+  }
+
+  public static long getNumericPersistentMetric(String key) {
+    AtomicLong m = numericPersistentMetrics.get(key);
+    if (m == null)
+      return 0;
+    return m.get();
+  }
+
+}

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerDynamicMetrics.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerDynamicMetrics.java?rev=1329958&r1=1329957&r2=1329958&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerDynamicMetrics.java
(original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/RegionServerDynamicMetrics.java
Tue Apr 24 19:28:40 2012
@@ -131,17 +131,17 @@ public class RegionServerDynamicMetrics 
    */
   public void doUpdates(MetricsContext context) {
     /* get dynamically created numeric metrics, and push the metrics */
-    for (Entry<String, AtomicLong> entry : HRegion.numericMetrics.entrySet()) {
+    for (Entry<String, AtomicLong> entry : RegionMetricsStorage.getNumericMetrics().entrySet())
{
       this.setNumericMetric(entry.getKey(), entry.getValue().getAndSet(0));
     }
     /* get dynamically created numeric metrics, and push the metrics.
      * These ones aren't to be reset; they are cumulative. */
-    for (Entry<String, AtomicLong> entry : HRegion.numericPersistentMetrics.entrySet())
{
+    for (Entry<String, AtomicLong> entry : RegionMetricsStorage.getNumericPersistentMetrics().entrySet())
{
       this.setNumericMetric(entry.getKey(), entry.getValue().get());
     }
     /* get dynamically created time varying metrics, and push the metrics */
     for (Entry<String, Pair<AtomicLong, AtomicInteger>> entry :
-          HRegion.timeVaryingMetrics.entrySet()) {
+        RegionMetricsStorage.getTimeVaryingMetrics().entrySet()) {
       Pair<AtomicLong, AtomicInteger> value = entry.getValue();
       this.incrTimeVaryingMetric(entry.getKey(),
           value.getFirst().getAndSet(0),

Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java?rev=1329958&r1=1329957&r2=1329958&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java
(original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/metrics/SchemaMetrics.java
Tue Apr 24 19:28:40 2012
@@ -165,9 +165,10 @@ public class SchemaMetrics {
    */
   public static final String UNKNOWN = "__unknown";
 
-  private static final String TABLE_PREFIX = "tbl.";
+  public static final String TABLE_PREFIX = "tbl.";
   public static final String CF_PREFIX = "cf.";
   public static final String BLOCK_TYPE_PREFIX = "bt.";
+  public static final String REGION_PREFIX = "region.";
 
   public static final String CF_UNKNOWN_PREFIX = CF_PREFIX + UNKNOWN + ".";
   public static final String CF_BAD_FAMILY_PREFIX = CF_PREFIX + "__badfamily.";
@@ -364,7 +365,7 @@ public class SchemaMetrics {
     if (blockCategory == null) {
       blockCategory = BlockCategory.UNKNOWN;  // So that we see this in stats.
     }
-    HRegion.incrNumericMetric(getBlockMetricName(blockCategory,
+    RegionMetricsStorage.incrNumericMetric(getBlockMetricName(blockCategory,
         isCompaction, metricType), 1);
 
     if (blockCategory != BlockCategory.ALL_CATEGORIES) {
@@ -375,7 +376,7 @@ public class SchemaMetrics {
 
   private void addToReadTime(BlockCategory blockCategory,
       boolean isCompaction, long timeMs) {
-    HRegion.incrTimeVaryingMetric(getBlockMetricName(blockCategory,
+    RegionMetricsStorage.incrTimeVaryingMetric(getBlockMetricName(blockCategory,
         isCompaction, BlockMetricType.READ_TIME), timeMs);
 
     // Also update the read time aggregated across all block categories
@@ -431,7 +432,7 @@ public class SchemaMetrics {
    */
   public void updatePersistentStoreMetric(StoreMetricType storeMetricType,
       long value) {
-    HRegion.incrNumericPersistentMetric(
+    RegionMetricsStorage.incrNumericPersistentMetric(
         storeMetricNames[storeMetricType.ordinal()], value);
   }
 
@@ -476,7 +477,7 @@ public class SchemaMetrics {
     if (category == null) {
       category = BlockCategory.ALL_CATEGORIES;
     }
-    HRegion.incrNumericPersistentMetric(getBlockMetricName(category, false,
+    RegionMetricsStorage.incrNumericPersistentMetric(getBlockMetricName(category, false,
         BlockMetricType.CACHE_SIZE), cacheSizeDelta);
 
     if (category != BlockCategory.ALL_CATEGORIES) {
@@ -500,7 +501,7 @@ public class SchemaMetrics {
    * positives/negatives as specified by the argument.
    */
   public void updateBloomMetrics(boolean isInBloom) {
-    HRegion.incrNumericMetric(getBloomMetricName(isInBloom), 1);
+    RegionMetricsStorage.incrNumericMetric(getBloomMetricName(isInBloom), 1);
     if (this != ALL_SCHEMA_METRICS) {
       ALL_SCHEMA_METRICS.updateBloomMetrics(isInBloom);
     }
@@ -615,6 +616,23 @@ public class SchemaMetrics {
   }
 
   /**
+   * Get the prefix for metrics generated about a single region.
+   * 
+   * @param tableName
+   *          the table name or {@link #TOTAL_KEY} for all tables
+   * @param regionName
+   *          regionName
+   * @return the prefix for this table/region combination.
+   */
+  static String generateRegionMetricsPrefix(String tableName, String regionName) {
+    tableName = getEffectiveTableName(tableName);
+    String schemaMetricPrefix = tableName.equals(TOTAL_KEY) ? "" : TABLE_PREFIX + tableName
+ ".";
+    schemaMetricPrefix += regionName.equals(TOTAL_KEY) ? "" : REGION_PREFIX + regionName
+ ".";
+
+    return schemaMetricPrefix;
+  }
+  
+  /**
    * Sets the flag of whether to use table name in metric names. This flag
    * is specified in configuration and is not expected to change at runtime,
    * so we log an error message when it does change.
@@ -729,11 +747,11 @@ public class SchemaMetrics {
         long metricValue;
         if (isTimeVaryingKey(metricName)) {
           Pair<Long, Integer> totalAndCount =
-              HRegion.getTimeVaryingMetric(stripTimeVaryingSuffix(metricName));
+              RegionMetricsStorage.getTimeVaryingMetric(stripTimeVaryingSuffix(metricName));
           metricValue = metricName.endsWith(TOTAL_SUFFIX) ?
               totalAndCount.getFirst() : totalAndCount.getSecond();
         } else {
-          metricValue = HRegion.getNumericMetric(metricName);
+          metricValue = RegionMetricsStorage.getNumericMetric(metricName);
         }
 
         metricsSnapshot.put(metricName, metricValue);

Modified: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerMetrics.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerMetrics.java?rev=1329958&r1=1329957&r2=1329958&view=diff
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerMetrics.java
(original)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionServerMetrics.java
Tue Apr 24 19:28:40 2012
@@ -23,15 +23,25 @@ import static org.junit.Assert.assertEqu
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.MediumTests;
+import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.regionserver.metrics.RegionMetricsStorage;
 import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics;
 import org.apache.hadoop.hbase.regionserver.metrics.SchemaMetrics.
     StoreMetricType;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -77,6 +87,29 @@ public class TestRegionServerMetrics {
     SchemaMetrics.validateMetricChanges(startingMetrics);
   }
 
+  private void assertTimeVaryingMetricCount(int expectedCount, String table, String cf,
+      String regionName, String metricPrefix) {
+
+    Integer expectedCountInteger = new Integer(expectedCount);
+
+    if (cf != null) {
+      String cfKey =
+          SchemaMetrics.TABLE_PREFIX + table + "." + SchemaMetrics.CF_PREFIX + cf + "."
+              + metricPrefix;
+      Pair<Long, Integer> cfPair = RegionMetricsStorage.getTimeVaryingMetric(cfKey);
+      assertEquals(expectedCountInteger, cfPair.getSecond());
+    }
+
+    if (regionName != null) {
+      String rKey =
+          SchemaMetrics.TABLE_PREFIX + table + "." + SchemaMetrics.REGION_PREFIX + regionName
+ "."
+              + metricPrefix;
+
+      Pair<Long, Integer> regionPair = RegionMetricsStorage.getTimeVaryingMetric(rKey);
+      assertEquals(expectedCountInteger, regionPair.getSecond());
+    }
+  }
+  
   private void assertStoreMetricEquals(long expected,
       SchemaMetrics schemaMetrics, StoreMetricType storeMetricType) {
     final String storeMetricName =
@@ -84,10 +117,88 @@ public class TestRegionServerMetrics {
     Long startValue = startingMetrics.get(storeMetricName);
     assertEquals("Invalid value for store metric " + storeMetricName
         + " (type " + storeMetricType + ")", expected,
-        HRegion.getNumericMetric(storeMetricName)
+        RegionMetricsStorage.getNumericMetric(storeMetricName)
             - (startValue != null ? startValue : 0));
   }
-
+  
+    @Test
+    public void testOperationMetrics() throws IOException {
+      String cf = "OPCF";
+      String otherCf = "otherCF";
+      String rk = "testRK";
+      String icvCol = "icvCol";
+      String appendCol = "appendCol";
+      String regionName = null;
+      HTable hTable =
+          TEST_UTIL.createTable(TABLE_NAME.getBytes(),
+              new byte[][] { cf.getBytes(), otherCf.getBytes() });
+      Set<HRegionInfo> regionInfos = hTable.getRegionLocations().keySet();
+  
+      regionName = regionInfos.toArray(new HRegionInfo[regionInfos.size()])[0].getEncodedName();
+  
+      //Do a multi put that has one cf.  Since they are in different rk's
+      //The lock will still be obtained and everything will be applied in one multiput.
+      Put pOne = new Put(rk.getBytes());
+      pOne.add(cf.getBytes(), icvCol.getBytes(), Bytes.toBytes(0L));
+      Put pTwo = new Put("ignored1RK".getBytes());
+      pTwo.add(cf.getBytes(), "ignored".getBytes(), Bytes.toBytes(0L));
+      
+      hTable.put(Arrays.asList(new Put[] {pOne, pTwo}));
+  
+      // Do a multiput where the cf doesn't stay consistent.
+      Put pThree = new Put("ignored2RK".getBytes());
+      pThree.add(cf.getBytes(), "ignored".getBytes(), Bytes.toBytes("TEST1"));
+      Put pFour = new Put("ignored3RK".getBytes());
+      pFour.add(otherCf.getBytes(), "ignored".getBytes(), Bytes.toBytes(0L));
+  
+      hTable.put(Arrays.asList(new Put[] { pThree, pFour }));
+  
+      hTable.incrementColumnValue(rk.getBytes(), cf.getBytes(), icvCol.getBytes(), 1L);
+      
+      Increment i = new Increment(rk.getBytes());
+      i.addColumn(cf.getBytes(), icvCol.getBytes(), 1L);
+      hTable.increment(i);
+  
+      Get g = new Get(rk.getBytes());
+      g.addColumn(cf.getBytes(), appendCol.getBytes());
+      hTable.get(g);
+  
+      Append a = new Append(rk.getBytes());
+      a.add(cf.getBytes(), appendCol.getBytes(), Bytes.toBytes("-APPEND"));
+      hTable.append(a);
+  
+      Delete dOne = new Delete(rk.getBytes());
+      dOne.deleteFamily(cf.getBytes());
+      hTable.delete(dOne);
+  
+      Delete dTwo = new Delete(rk.getBytes());
+      hTable.delete(dTwo);
+  
+      // There should be one multi put where the cf is consistent
+      assertTimeVaryingMetricCount(1, TABLE_NAME, cf, null, "multiput_");
+  
+      // There were two multiputs to the cf.
+      assertTimeVaryingMetricCount(2, TABLE_NAME, null, regionName, "multiput_");
+  
+      // There was one multiput where the cf was not consistent.
+      assertTimeVaryingMetricCount(1, TABLE_NAME, "__unknown", null, "multiput_");
+  
+      // One increment and one append
+      assertTimeVaryingMetricCount(1, TABLE_NAME, cf, regionName, "incrementColumnValue_");
+      assertTimeVaryingMetricCount(1, TABLE_NAME, cf, regionName, "increment_");
+      assertTimeVaryingMetricCount(1, TABLE_NAME, cf, regionName, "append_");
+  
+      // One delete where the cf is known
+      assertTimeVaryingMetricCount(1, TABLE_NAME, cf, null, "delete_");
+  
+      // two deletes in the region.
+      assertTimeVaryingMetricCount(2, TABLE_NAME, null, regionName, "delete_");
+  
+      // Three gets. one for gets. One for append. One for increment.
+      assertTimeVaryingMetricCount(4, TABLE_NAME, cf, regionName, "get_");
+  
+    }
+  
   @Test
   public void testMultipleRegions() throws IOException, InterruptedException {
 
@@ -124,7 +235,7 @@ public class TestRegionServerMetrics {
     final String storeMetricName = ALL_METRICS
         .getStoreMetricNameMax(StoreMetricType.STORE_FILE_COUNT);
     assertEquals("Invalid value for store metric " + storeMetricName,
-        NUM_FLUSHES, HRegion.getNumericMetric(storeMetricName));
+        NUM_FLUSHES, RegionMetricsStorage.getNumericMetric(storeMetricName));
   }
 
 



Mime
View raw message