Basic heat map implementation at region level

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java

@@ -30,6 +30,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
@@ -131,6 +132,12 @@
 import org.apache.hadoop.hbase.regionserver.handler.CloseRegionHandler;
 import org.apache.hadoop.hbase.regionserver.handler.RSProcedureHandler;
 import org.apache.hadoop.hbase.regionserver.handler.RegionReplicaFlushHandler;
+import org.apache.hadoop.hbase.regionserver.stats.AccessStats;
+import org.apache.hadoop.hbase.regionserver.stats.AccessStats.AccessStatsType;
+import org.apache.hadoop.hbase.regionserver.stats.AccessStatsRecorderTableImpl;
+import org.apache.hadoop.hbase.regionserver.stats.AccessStatsRecorderUtils;
+import org.apache.hadoop.hbase.regionserver.stats.IAccessStatsRecorder;
+import org.apache.hadoop.hbase.regionserver.stats.RegionAccessStats;
 import org.apache.hadoop.hbase.regionserver.throttle.FlushThroughputControllerFactory;
 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
 import org.apache.hadoop.hbase.replication.regionserver.ReplicationLoad;
@@ -385,6 +392,11 @@ public class HRegionServer extends HasThread implements
    * Check for flushes
    */
   ScheduledChore periodicFlusher;
+
+  /*
+   * Record periodic region counters (read and write)
+   */
+  ScheduledChore regionStatsRecorder;
 
   protected volatile WALFactory walFactory;
 
@@ -1818,6 +1830,83 @@ protected void chore() {
       }
     }
   }
+
+  /*
+   * This class is to periodically store information about region access stats.
+   */
+  private static class RegionStatsRecorder extends ScheduledChore {
+    private final HRegionServer instance;
+    IAccessStatsRecorder accessStatsRecorder;
+
+    Map<String, Long> regionLastReadCountMap = new HashMap<String, Long>();
+    Map<String, Long> regionLastWriteCountMap = new HashMap<String, Long>();
+
+    RegionStatsRecorder(final HRegionServer h, final Stoppable stopper, final int durationInMinutes,
+        final Configuration configuration) {
+      super("RegionStatsRecorder", stopper, durationInMinutes * 60 * 1000);
+      this.instance = h;
+
+      accessStatsRecorder = new AccessStatsRecorderTableImpl(configuration);
+
+      LOG.info(this.getName() + " runs every " + durationInMinutes
+          + " minutes with initial delay of " + durationInMinutes + " minutes.");
+    }
+
+    @Override
+    protected void chore() {
+      List<AccessStats> accessStatsList = new ArrayList<>();
+
+      for (Region region : this.instance.onlineRegions.values()) {
+        if (region == null) continue;
+
+        String regionName = region.getRegionInfo().getRegionNameAsString();
+
+        long currentReadRequestCount = region.getReadRequestsCount();
+        long currentWriteRequestCount = region.getWriteRequestsCount();
+
+        Long lastReadRequestCount = regionLastReadCountMap.get(regionName);
+        Long lastWriteRequestCount = regionLastWriteCountMap.get(regionName);
+
+        long readRequestCountSinceLastIteration = currentReadRequestCount;
+        long writeRequestCountSinceLastIteration = currentWriteRequestCount;
+
+        if (lastReadRequestCount != null) {
+          readRequestCountSinceLastIteration = currentReadRequestCount - lastReadRequestCount;
+        }
+
+        if (lastWriteRequestCount != null) {
+          writeRequestCountSinceLastIteration = currentWriteRequestCount - lastWriteRequestCount;
+        }
+
+        RegionAccessStats regionAccessStats =
+            new RegionAccessStats(region.getRegionInfo().getTable(), AccessStatsType.READCOUNT,
+                region.getRegionInfo().getStartKey(), region.getRegionInfo().getEndKey(),
+                readRequestCountSinceLastIteration);
+        regionAccessStats.setRegionName(regionName);
+        accessStatsList.add(regionAccessStats);
+
+        regionAccessStats = new RegionAccessStats(region.getRegionInfo().getTable(),
+            AccessStatsType.WRITECOUNT, region.getRegionInfo().getStartKey(),
+            region.getRegionInfo().getEndKey(), writeRequestCountSinceLastIteration);
+        regionAccessStats.setRegionName(regionName);
+        accessStatsList.add(regionAccessStats);
+
+        regionLastReadCountMap.put(regionName, currentReadRequestCount);
+        regionLastWriteCountMap.put(regionName, currentWriteRequestCount);
+      }
+
+      accessStatsRecorder.writeAccessStats(accessStatsList);
+    }
+
+    @Override
+    protected synchronized void cleanup() {
+      try {
+        accessStatsRecorder.close();
+      } catch (IOException e) {
+        LOG.error("Exception in cleanup of RegionStatsRecorder - "+e.getMessage());
+      }
+    }
+  }
 
   /**
    * Report the status of the server. A server is online once all the startup is
@@ -1977,6 +2066,7 @@ private void startServices() throws IOException {
     if (this.nonceManagerChore != null) choreService.scheduleChore(nonceManagerChore);
     if (this.storefileRefresher != null) choreService.scheduleChore(storefileRefresher);
     if (this.movedRegionsCleaner != null) choreService.scheduleChore(movedRegionsCleaner);
+    if (this.regionStatsRecorder != null) choreService.scheduleChore(regionStatsRecorder);
     if (this.fsUtilizationChore != null) choreService.scheduleChore(fsUtilizationChore);
 
     // Leases is not a Thread. Internally it runs a daemon thread. If it gets
@@ -2019,6 +2109,11 @@ private void initializeThreads() throws IOException {
     // in a while. It will take care of not checking too frequently on store-by-store basis.
     this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency, this);
     this.periodicFlusher = new PeriodicMemStoreFlusher(this.threadWakeFrequency, this);
+
+    AccessStatsRecorderUtils.createInstance(conf);
+    this.regionStatsRecorder = new RegionStatsRecorder(this, this,
+        AccessStatsRecorderUtils.getInstance().getIterationDuration(), conf);
+
     this.leases = new Leases(this.threadWakeFrequency);
 
     // Create the thread to clean the moved regions list
@@ -2131,7 +2226,8 @@ private boolean isHealthy() {
         && (this.cacheFlusher == null || this.cacheFlusher.isAlive())
         && (this.walRoller == null || this.walRoller.isAlive())
         && (this.compactionChecker == null || this.compactionChecker.isScheduled())
-        && (this.periodicFlusher == null || this.periodicFlusher.isScheduled());
+        && (this.periodicFlusher == null || this.periodicFlusher.isScheduled()
+        && (this.regionStatsRecorder == null || this.regionStatsRecorder.isScheduled()));
     if (!healthy) {
       stop("One or more threads are no longer alive -- stop");
     }
@@ -2473,6 +2569,7 @@ protected void stopServiceThreads() {
       choreService.cancelChore(healthCheckChore);
       choreService.cancelChore(storefileRefresher);
       choreService.cancelChore(movedRegionsCleaner);
+      choreService.cancelChore(regionStatsRecorder);
       choreService.cancelChore(fsUtilizationChore);
       // clean up the remaining scheduled chores (in case we missed out any)
       choreService.shutdown();
@@ -3837,4 +3934,4 @@ public void run() {
       System.exit(1);
     }
   }
-}
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/stats/AccessStats.java

@@ -0,0 +1,131 @@
+/**
+ *
+ * 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.stats;
+
+import java.util.List;
+
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * Generic class to hold AccessStats information together, its an abstract class and certain methods
+ * needs to be implemented for given granularity of access stats. Currently its implemented only for
+ * REGION.
+ */
+@InterfaceAudience.Private
+public abstract class AccessStats {
+  protected AccessStatsType accessStatsType;
+  TableName table;
+  protected long normalizedTimeInEpoch;
+  protected long value;
+  protected byte[] keyRangeStart;
+  protected byte[] keyRangeEnd;
+
+  public AccessStats(TableName table, AccessStatsType accessStatsType, byte[] keyRangeStart,
+      byte[] keyRangeEnd, long value) {
+    this(table, accessStatsType, keyRangeStart, keyRangeEnd,
+        AccessStatsRecorderUtils.getInstance().getNormalizedTimeCurrent(), value);
+  }
+
+  public AccessStats(TableName table, AccessStatsType accessStatsType, byte[] keyRangeStart,
+      byte[] keyRangeEnd, long time, long value) {
+    this.accessStatsType = accessStatsType;
+    this.table = table;
+    this.normalizedTimeInEpoch = time;
+    this.value = value;
+    this.keyRangeStart = keyRangeStart;
+    this.keyRangeEnd = keyRangeEnd;
+  }
+
+  public long getEpochTime() {
+    return normalizedTimeInEpoch;
+  }
+
+  public byte[] getValueInBytes() {
+    return Bytes.toBytes(value);
+  }
+
+  public long getValue() {
+    return value;
+  }
+
+  public byte[] getAccessStatsType() {
+    return Bytes.toBytes(accessStatsType.toString());
+  }
+
+  public TableName getTable() {
+    return table;
+  }
+
+  public byte[] getKeyRangeStart() {
+    return keyRangeStart;
+  }
+
+  public byte[] getKeyRangeEnd() {
+    return keyRangeEnd;
+  }
+
+  /*
+   * Each AccessStats record is uniquely identified by a row key which is a combination of table
+   * name and time, followed by fixed set of KeyPartDescriptors; all encoded in fix length binary
+   * encoding except time. For a given granularity, this method will return list of
+   * KeyPartDescriptor which will be used in generating row key.
+   */
+  protected abstract List<KeyPartDescriptor> getKeyPartDescriptors();
+
+  /*
+   * Since only this class knows how to divide byte encoded row key suffix into different parts,
+   * this method is supposed to do that.
+   */
+  protected abstract byte[][] convertKeySuffixToByteArrayList(byte[] keyByteArray, int indexStart);
+
+  /*
+   * This method should set the fields specific to given granularity by using decoded row key parts.
+   */
+  protected abstract void setFieldsUsingKeyParts(List<String> strings);
+
+  class KeyPartDescriptor {
+    private String keyStr;
+    private int uidLengthInBytes;
+
+    public KeyPartDescriptor(String keyStr, int uidLengthInBytes) {
+      this.keyStr = keyStr;
+      this.uidLengthInBytes = uidLengthInBytes;
+    }
+
+    public String getKeyStr() {
+      return keyStr;
+    }
+
+    public int getUidLengthInBytes() {
+      return uidLengthInBytes;
+    }
+
+  }
+
+  public enum AccessStatsType {
+    READCOUNT, WRITECOUNT
+  }
+
+  public enum AccessStatsGranularity {
+    REGION
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/stats/AccessStatsFactory.java

@@ -0,0 +1,48 @@
+/**
+ *
+ * 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.stats;
+
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.regionserver.stats.AccessStats.AccessStatsGranularity;
+import org.apache.hadoop.hbase.regionserver.stats.AccessStats.AccessStatsType;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public class AccessStatsFactory {
+  public static AccessStats getAccessStatsObj(TableName table,
+      AccessStatsGranularity accessStatsGranularity, AccessStatsType accessStatsType,
+      byte[] keyRangeStart, byte[] keyRangeEnd, long timeInEpoch, long value) {
+    AccessStats accessStats = null;
+
+    switch (accessStatsGranularity) {
+    case REGION:
+      accessStats = new RegionAccessStats(table, accessStatsType, keyRangeStart, keyRangeEnd,
+          timeInEpoch, value);
+      break;
+    default:
+      accessStats = new RegionAccessStats(table, accessStatsType, keyRangeStart, keyRangeEnd,
+          timeInEpoch, value);
+      break;
+    }
+
+    return accessStats;
+  }
+
+}