Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-21800: RegionServer aborted due to NPE from MetaTableMetrics coprocessor 

Have included code refactoring in MetaTableMetrics & LossyCounting

Signed-off-by: Xu Cang <xucang@apache.org>
This commit is contained in:
Sakthi 2019-03-20 13:55:08 -07:00 committed by Xu Cang
parent 3d3832ebbc
commit f4de34a03d
No known key found for this signature in database
GPG Key ID: 8E6C8FEDCA866394
3 changed files with 48 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MetaTableMetrics.java
View File

@ -64,52 +64,42 @@ public class MetaTableMetrics extends BaseRegionObserver {
.build();
@Override
public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get,
List<Cell> results) throws IOException {
if (!active || !isMetaTableOp(e)) {
return;
}
tableMetricRegisterAndMark(e, get);
clientMetricRegisterAndMark(e);
regionMetricRegisterAndMark(e, get);
opMetricRegisterAndMark(e, get);
opWithClientMetricRegisterAndMark(e, get);
public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get, List<Cell> results)
throws IOException {
registerAndMarkMetrics(e, get);
}
@Override
public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit,
Durability durability) throws IOException {
if (!active || !isMetaTableOp(e)) {
return;
}
tableMetricRegisterAndMark(e, put);
clientMetricRegisterAndMark(e);
regionMetricRegisterAndMark(e, put);
opMetricRegisterAndMark(e, put);
opWithClientMetricRegisterAndMark(e, put);
registerAndMarkMetrics(e, put);
}
@Override
public void preDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete,
WALEdit edit, Durability durability) throws IOException {
registerAndMarkMetrics(e, delete);
}
private void registerAndMarkMetrics(ObserverContext<RegionCoprocessorEnvironment> e, Row row) {
if (!active || !isMetaTableOp(e)) {
return;
}
tableMetricRegisterAndMark(e, delete);
tableMetricRegisterAndMark(e, row);
clientMetricRegisterAndMark(e);
regionMetricRegisterAndMark(e, delete);
opMetricRegisterAndMark(e, delete);
opWithClientMetricRegisterAndMark(e, delete);
regionMetricRegisterAndMark(e, row);
opMetricRegisterAndMark(e, row);
opWithClientMetricRegisterAndMark(e, row);
}
private void markMeterIfPresent(String requestMeter) {
if (requestMeter.isEmpty()) {
return;
}
Metric metric =
requestsMap.get(requestMeter).isPresent() ? requestsMap.get(requestMeter).get() : null;
if (metric != null) {
((Meter) metric).mark();
if (requestsMap.containsKey(requestMeter) && requestsMap.get(requestMeter).isPresent()) {
Meter metric = (Meter) requestsMap.get(requestMeter).get();
metric.mark();
}
}
@ -128,7 +118,7 @@ public class MetaTableMetrics extends BaseRegionObserver {
/**
* Registers and counts lossyCount for Meters that kept by lossy counting.
* By using lossy count to maintain meters, at most 7 / e meters will be kept (e is error rate)
* e.g. when e is 0.02 by default, at most 50 Clients request metrics will be kept
* e.g. when e is 0.02 by default, at most 350 Clients request metrics will be kept
* also, all kept elements have frequency higher than e * N. (N is total count)
* @param e Region coprocessor environment
* @param requestMeter meter to be registered
@ -192,6 +182,7 @@ public class MetaTableMetrics extends BaseRegionObserver {
}
private void clientMetricRegisterAndMark(ObserverContext<RegionCoprocessorEnvironment> e) {
// Mark client metric
String clientIP = RpcServer.getRemoteIp() != null ? RpcServer.getRemoteIp().toString() : "";
String clientRequestMeter = clientRequestMeterName(clientIP);
@ -199,39 +190,43 @@ public class MetaTableMetrics extends BaseRegionObserver {
markMeterIfPresent(clientRequestMeter);
}
private void tableMetricRegisterAndMark(ObserverContext<RegionCoprocessorEnvironment> e,
Row op) {
// Mark the meta table meter whenever the coprocessor is called
private void tableMetricRegisterAndMark(ObserverContext<RegionCoprocessorEnvironment> e, Row op) {
// Mark table metric
String tableName = getTableNameFromOp(op);
String tableRequestMeter = tableMeterName(tableName);
registerMeterIfNotPresent(e, tableRequestMeter);
markMeterIfPresent(tableRequestMeter);
registerAndMarkMeterIfNotPresent(e, tableRequestMeter);
}
private void regionMetricRegisterAndMark(ObserverContext<RegionCoprocessorEnvironment> e,
Row op) {
// Mark the meta table meter whenever the coprocessor is called
// Mark region metric
String regionId = getRegionIdFromOp(op);
String regionRequestMeter = regionMeterName(regionId);
registerMeterIfNotPresent(e, regionRequestMeter);
markMeterIfPresent(regionRequestMeter);
registerAndMarkMeterIfNotPresent(e, regionRequestMeter);
}
private void opMetricRegisterAndMark(ObserverContext<RegionCoprocessorEnvironment> e,
Row op) {
private void opMetricRegisterAndMark(ObserverContext<RegionCoprocessorEnvironment> e, Row op) {
// Mark access type ["get", "put", "delete"] metric
String opMeterName = opMeterName(op);
registerMeterIfNotPresent(e, opMeterName);
markMeterIfPresent(opMeterName);
registerAndMarkMeterIfNotPresent(e, opMeterName);
}
private void opWithClientMetricRegisterAndMark(ObserverContext<RegionCoprocessorEnvironment> e,
Object op) {
// // Mark client + access type metric
String opWithClientMeterName = opWithClientMeterName(op);
registerMeterIfNotPresent(e, opWithClientMeterName);
markMeterIfPresent(opWithClientMeterName);
registerAndMarkMeterIfNotPresent(e, opWithClientMeterName);
}
// Helper function to register and mark meter if not present
private void registerAndMarkMeterIfNotPresent(ObserverContext<RegionCoprocessorEnvironment> e,
String name) {
registerMeterIfNotPresent(e, name);
markMeterIfPresent(name);
}
private String opWithClientMeterName(Object op) {
// Extract meter name containing the client IP
String clientIP = RpcServer.getRemoteIp() != null ? RpcServer.getRemoteIp().toString() : "";
if (clientIP.isEmpty()) {
return "";
@ -274,10 +269,12 @@ public class MetaTableMetrics extends BaseRegionObserver {
}
private String tableMeterName(String tableName) {
// Extract meter name containing the table name
return String.format("MetaTable_table_%s_request", tableName);
}
private String clientRequestMeterName(String clientIP) {
// Extract meter name containing the client IP
if (clientIP.isEmpty()) {
return "";
}
@ -285,6 +282,7 @@ public class MetaTableMetrics extends BaseRegionObserver {
}
private String regionMeterName(String regionId) {
// Extract meter name containing the region ID
return String.format("MetaTable_region_%s_request", regionId);
}
@ -306,11 +304,10 @@ public class MetaTableMetrics extends BaseRegionObserver {
public void stop(CoprocessorEnvironment env) throws IOException {
// since meta region can move around, clear stale metrics when stop.
if (requestsMap != null) {
MetricRegistry registry = regionCoprocessorEnv.getMetricRegistryForRegionServer();
for (String meterName : requestsMap.keySet()) {
MetricRegistry registry = regionCoprocessorEnv.getMetricRegistryForRegionServer();
registry.remove(meterName);
}
}
}
}

21
hbase-server/src/main/java/org/apache/hadoop/hbase/util/LossyCounting.java
View File

@ -24,7 +24,6 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
@ -60,27 +59,19 @@ public class LossyCounting {
this.bucketSize = (long) Math.ceil(1 / errorRate);
this.currentTerm = 1;
this.totalDataCount = 0;
this.errorRate = errorRate;
this.data = new ConcurrentHashMap<>();
calculateCurrentTerm();
}
public LossyCounting() {
Configuration conf = HBaseConfiguration.create();
this.errorRate = conf.getDouble(HConstants.DEFAULT_LOSSY_COUNTING_ERROR_RATE, 0.02);
this.bucketSize = (long) Math.ceil(1.0 / errorRate);
this.currentTerm = 1;
this.totalDataCount = 0;
this.data = new ConcurrentHashMap<>();
calculateCurrentTerm();
this(HBaseConfiguration.create().getDouble(HConstants.DEFAULT_LOSSY_COUNTING_ERROR_RATE, 0.02));
}
public Set<String> addByOne(String key) {
if(data.containsKey(key)) {
data.put(key, data.get(key) +1);
} else {
data.put(key, 1);
if (!data.containsKey(key)) {
data.put(key, 0);
}
data.put(key, data.get(key) + 1);
totalDataCount++;
calculateCurrentTerm();
Set<String> dataToBeSwept = new HashSet<>();
@ -104,7 +95,7 @@ public class LossyCounting {
for(String key : dataToBeSwept) {
data.remove(key);
}
LOG.debug(String.format("Swept %d of elements.", dataToBeSwept.size()));
LOG.debug(String.format("Swept %d elements.", dataToBeSwept.size()));
return dataToBeSwept;
}
@ -115,7 +106,7 @@ public class LossyCounting {
this.currentTerm = (int) Math.ceil(1.0 * totalDataCount / bucketSize);
}
public long getBuketSize(){
public long getBucketSize(){
return bucketSize;
}

4
hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestLossyCounting.java
View File

@ -34,9 +34,9 @@ public class TestLossyCounting {
@Test
public void testBucketSize() {
LossyCounting lossyCounting = new LossyCounting(0.01);
assertEquals(100L, lossyCounting.getBuketSize());
assertEquals(100L, lossyCounting.getBucketSize());
LossyCounting lossyCounting2 = new LossyCounting();
assertEquals(50L, lossyCounting2.getBuketSize());
assertEquals(50L, lossyCounting2.getBucketSize());
}
@Test