Repository: hbase Updated Branches: refs/heads/master e93e1ea73 -> f28e39529
HBASE-13375 Provide HBase superuser higher priority over other users in the RPC handling. (Mikhail Antonov) Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/f28e3952 Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/f28e3952 Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/f28e3952 Branch: refs/heads/master Commit: f28e3952905592c6cc3c6c07b2509795e1210e70 Parents: e93e1ea Author: anoopsjohn <anoopsamj...@gmail.com> Authored: Sat May 23 21:12:47 2015 +0530 Committer: anoopsjohn <anoopsamj...@gmail.com> Committed: Sat May 23 21:12:47 2015 +0530 ---------------------------------------------------------------------- .../hadoop/hbase/ipc/PriorityFunction.java | 4 +- .../hadoop/hbase/ipc/SimpleRpcScheduler.java | 2 +- .../AnnotationReadingPriorityFunction.java | 50 +++++++++++++++++++- .../hbase/regionserver/RSRpcServices.java | 5 +- .../hbase/ipc/TestSimpleRpcScheduler.java | 11 +++-- .../hbase/master/TestMasterPriorityRpc.java | 7 ++- .../hbase/regionserver/TestPriorityRpc.java | 38 ++++++++++++--- .../hbase/regionserver/TestQosFunction.java | 22 +++++---- 8 files changed, 112 insertions(+), 27 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/PriorityFunction.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/PriorityFunction.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/PriorityFunction.java index e323e78..f56bf6f 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/PriorityFunction.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/PriorityFunction.java @@ -22,6 +22,7 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.classification.InterfaceStability; import org.apache.hadoop.hbase.HBaseInterfaceAudience; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; +import org.apache.hadoop.hbase.security.User; /** * Function to figure priority of incoming request. @@ -34,9 +35,10 @@ public interface PriorityFunction { * The returned value is mainly used to select the dispatch queue. * @param header * @param param + * @param user * @return Priority of this request. */ - int getPriority(RequestHeader header, Message param); + int getPriority(RequestHeader header, Message param, User user); /** * Returns the deadline of the specified request. http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java index 2889d8d..b8e9c52 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java @@ -192,7 +192,7 @@ public class SimpleRpcScheduler extends RpcScheduler { @Override public void dispatch(CallRunner callTask) throws InterruptedException { RpcServer.Call call = callTask.getCall(); - int level = priority.getPriority(call.getHeader(), call.param); + int level = priority.getPriority(call.getHeader(), call.param, call.getRequestUser()); if (priorityExecutor != null && level > highPriorityLevel) { priorityExecutor.dispatch(callTask); } else if (replicationExecutor != null && level == HConstants.REPLICATION_QOS) { http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AnnotationReadingPriorityFunction.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AnnotationReadingPriorityFunction.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AnnotationReadingPriorityFunction.java index 880710a..c96619b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AnnotationReadingPriorityFunction.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/AnnotationReadingPriorityFunction.java @@ -17,8 +17,11 @@ */ package org.apache.hadoop.hbase.regionserver; +import java.io.IOException; import java.lang.reflect.Method; import java.util.HashMap; +import java.util.HashSet; +import java.util.List; import java.util.Map; import org.apache.commons.logging.Log; @@ -49,7 +52,9 @@ import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; import com.google.common.annotations.VisibleForTesting; import com.google.protobuf.Message; import com.google.protobuf.TextFormat; - +import org.apache.hadoop.hbase.security.User; +import org.apache.hadoop.hbase.security.visibility.VisibilityUtils; +import org.apache.hadoop.hbase.util.Pair; /** * Reads special method annotations and table names to figure a priority for use by QoS facility in @@ -105,6 +110,11 @@ class AnnotationReadingPriorityFunction implements PriorityFunction { private final float scanVirtualTimeWeight; + // lists of super users and super groups, used to route rpc calls made by + // superusers through high-priority (ADMIN_QOS) thread pool. + // made protected for tests + protected final HashSet<String> superUsers; + protected final HashSet<String> superGroups; /** * Calls {@link #AnnotationReadingPriorityFunction(RSRpcServices, Class)} using the result of * {@code rpcServices#getClass()} @@ -158,6 +168,16 @@ class AnnotationReadingPriorityFunction implements PriorityFunction { Configuration conf = rpcServices.getConfiguration(); scanVirtualTimeWeight = conf.getFloat(SCAN_VTIME_WEIGHT_CONF_KEY, 1.0f); + + try { + // TODO Usage of VisibilityUtils API to be avoided with HBASE-13755 + Pair<List<String>, List<String>> pair = VisibilityUtils.getSystemAndSuperUsers(rpcServices + .getConfiguration()); + superUsers = new HashSet<>(pair.getFirst()); + superGroups = new HashSet<>(pair.getSecond()); + } catch (IOException e) { + throw new RuntimeException(e); + } } private String capitalize(final String s) { @@ -175,12 +195,18 @@ class AnnotationReadingPriorityFunction implements PriorityFunction { * NORMAL_QOS (user requests). */ @Override - public int getPriority(RequestHeader header, Message param) { + public int getPriority(RequestHeader header, Message param, User user) { String methodName = header.getMethodName(); Integer priorityByAnnotation = annotatedQos.get(methodName); if (priorityByAnnotation != null) { return priorityByAnnotation; } + + // all requests executed by super users have high QoS + if (isExecutedBySuperUser(user)) { + return HConstants.ADMIN_QOS; + } + if (param == null) { return HConstants.NORMAL_QOS; } @@ -280,4 +306,24 @@ class AnnotationReadingPriorityFunction implements PriorityFunction { void setRegionServer(final HRegionServer hrs) { this.rpcServices = hrs.getRSRpcServices(); } + + /** + * @param user user running request + * @return true if user is super user, false otherwise + */ + private boolean isExecutedBySuperUser(User user) { + if (superUsers.contains(user.getShortName())) { + return true; + } + + String[] groups = user.getGroupNames(); + if (groups != null) { + for (String group : groups) { + if (superGroups.contains(group)) { + return true; + } + } + } + return false; + } } http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java index 7bdbed2..aedd351 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java @@ -161,6 +161,7 @@ import org.apache.hadoop.hbase.regionserver.ScannerContext.LimitScope; import org.apache.hadoop.hbase.regionserver.handler.OpenMetaHandler; import org.apache.hadoop.hbase.regionserver.handler.OpenRegionHandler; import org.apache.hadoop.hbase.regionserver.wal.WALEdit; +import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Counter; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; @@ -1008,8 +1009,8 @@ public class RSRpcServices implements HBaseRPCErrorHandler, } @Override - public int getPriority(RequestHeader header, Message param) { - return priority.getPriority(header, param); + public int getPriority(RequestHeader header, Message param, User user) { + return priority.getPriority(header, param, user); } @Override http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java index 144da14..33d44da 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java @@ -27,6 +27,7 @@ import com.google.protobuf.Message; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.hbase.testclassification.RPCTests; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.apache.hadoop.hbase.ipc.RpcServer.Call; @@ -125,7 +126,8 @@ public class TestSimpleRpcScheduler { scheduler.init(CONTEXT); scheduler.start(); for (CallRunner task : tasks) { - when(qosFunction.getPriority((RPCProtos.RequestHeader) anyObject(), (Message) anyObject())) + when(qosFunction.getPriority((RPCProtos.RequestHeader) anyObject(), + (Message) anyObject(), (User) anyObject())) .thenReturn(qos.get(task)); scheduler.dispatch(task); } @@ -157,7 +159,8 @@ public class TestSimpleRpcScheduler { schedConf.set(SimpleRpcScheduler.CALL_QUEUE_TYPE_CONF_KEY, queueType); PriorityFunction priority = mock(PriorityFunction.class); - when(priority.getPriority(any(RequestHeader.class), any(Message.class))) + when(priority.getPriority(any(RequestHeader.class), + any(Message.class), any(User.class))) .thenReturn(HConstants.NORMAL_QOS); RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 1, 1, 1, priority, @@ -235,8 +238,8 @@ public class TestSimpleRpcScheduler { schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0.5f); PriorityFunction priority = mock(PriorityFunction.class); - when(priority.getPriority(any(RequestHeader.class), any(Message.class))) - .thenReturn(HConstants.NORMAL_QOS); + when(priority.getPriority(any(RequestHeader.class), any(Message.class), + any(User.class))).thenReturn(HConstants.NORMAL_QOS); RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 3, 1, 1, priority, HConstants.QOS_THRESHOLD); http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterPriorityRpc.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterPriorityRpc.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterPriorityRpc.java index a72f96b..3aafe0f 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterPriorityRpc.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterPriorityRpc.java @@ -28,6 +28,7 @@ import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.ipc.PriorityFunction; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; +import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.hbase.testclassification.MasterTests; import org.apache.hadoop.hbase.testclassification.MediumTests; import org.junit.Before; @@ -43,6 +44,7 @@ import com.google.common.collect.Sets; public class TestMasterPriorityRpc { private HMaster master = null; private PriorityFunction priority = null; + private User user = null; private final Set<String> ADMIN_METHODS = Sets.newHashSet("GetLastFlushedSequenceId", "RegionServerReport", "RegionServerStartup", "ReportRSFatalError", @@ -58,6 +60,7 @@ public class TestMasterPriorityRpc { CoordinatedStateManager cp = CoordinatedStateManagerFactory.getCoordinatedStateManager(conf); master = HMaster.constructMaster(HMaster.class, conf, cp); priority = master.getMasterRpcServices().getPriority(); + user = User.createUserForTesting(conf, "someuser", new String[]{"somegroup"}); } /** @@ -69,8 +72,8 @@ public class TestMasterPriorityRpc { * The expected priority. */ private void assertPriority(String methodName, int expectedPriority) { - assertEquals(methodName + " had unexpected priority", expectedPriority, - priority.getPriority(RequestHeader.newBuilder().setMethodName(methodName).build(), null)); + assertEquals(methodName + " had unexpected priority", expectedPriority, priority.getPriority( + RequestHeader.newBuilder().setMethodName(methodName).build(), null, user)); } @Test http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPriorityRpc.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPriorityRpc.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPriorityRpc.java index dc18408..3cd6f3d 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPriorityRpc.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPriorityRpc.java @@ -24,6 +24,7 @@ import static org.junit.Assert.assertTrue; import java.io.IOException; +import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.hbase.testclassification.MediumTests; import org.apache.hadoop.hbase.testclassification.RegionServerTests; import org.apache.hadoop.hbase.util.ByteStringer; @@ -90,12 +91,13 @@ public class TestPriorityRpc { RSRpcServices mockRpc = Mockito.mock(RSRpcServices.class); Mockito.when(mockRS.getRSRpcServices()).thenReturn(mockRpc); HRegionInfo mockRegionInfo = Mockito.mock(HRegionInfo.class); - Mockito.when(mockRpc.getRegion((RegionSpecifier)Mockito.any())).thenReturn(mockRegion); + Mockito.when(mockRpc.getRegion((RegionSpecifier) Mockito.any())).thenReturn(mockRegion); Mockito.when(mockRegion.getRegionInfo()).thenReturn(mockRegionInfo); Mockito.when(mockRegionInfo.isSystemTable()).thenReturn(true); // Presume type. ((AnnotationReadingPriorityFunction)priority).setRegionServer(mockRS); - assertEquals(HConstants.SYSTEMTABLE_QOS, priority.getPriority(header, getRequest)); + assertEquals(HConstants.SYSTEMTABLE_QOS, priority.getPriority(header, getRequest, + User.createUserForTesting(regionServer.conf, "someuser", new String[]{"somegroup"}))); } @Test @@ -108,7 +110,28 @@ public class TestPriorityRpc { headerBuilder.setMethodName("foo"); RequestHeader header = headerBuilder.build(); PriorityFunction qosFunc = regionServer.rpcServices.getPriority(); - assertEquals(HConstants.NORMAL_QOS, qosFunc.getPriority(header, null)); + assertEquals(HConstants.NORMAL_QOS, qosFunc.getPriority(header, null, + User.createUserForTesting(regionServer.conf, "someuser", new String[]{"somegroup"}))); + } + + @Test + public void testQosFunctionForRequestCalledBySuperUser() throws Exception { + RequestHeader.Builder headerBuilder = RequestHeader.newBuilder(); + headerBuilder.setMethodName("foo"); + RequestHeader header = headerBuilder.build(); + PriorityFunction qosFunc = regionServer.rpcServices.getPriority(); + + //test superusers + ((AnnotationReadingPriorityFunction) qosFunc).superUsers.add("samplesuperuser"); + assertEquals(HConstants.ADMIN_QOS, qosFunc.getPriority(header, null, + User.createUserForTesting(regionServer.conf, "samplesuperuser", + new String[]{"somegroup"}))); + + //test supergroups + ((AnnotationReadingPriorityFunction) qosFunc).superGroups.add("samplesupergroup"); + assertEquals(HConstants.ADMIN_QOS, qosFunc.getPriority(header, null, + User.createUserForTesting(regionServer.conf, "regularuser", + new String[]{"samplesupergroup"}))); } @Test @@ -130,7 +153,8 @@ public class TestPriorityRpc { Mockito.when(mockRegionInfo.isSystemTable()).thenReturn(false); // Presume type. ((AnnotationReadingPriorityFunction)priority).setRegionServer(mockRS); - int qos = priority.getPriority(header, scanRequest); + int qos = priority.getPriority(header, scanRequest, + User.createUserForTesting(regionServer.conf, "someuser", new String[]{"somegroup"})); assertTrue ("" + qos, qos == HConstants.NORMAL_QOS); //build a scan request with scannerID @@ -148,10 +172,12 @@ public class TestPriorityRpc { // Presume type. ((AnnotationReadingPriorityFunction)priority).setRegionServer(mockRS); - assertEquals(HConstants.SYSTEMTABLE_QOS, priority.getPriority(header, scanRequest)); + assertEquals(HConstants.SYSTEMTABLE_QOS, priority.getPriority(header, scanRequest, + User.createUserForTesting(regionServer.conf, "someuser", new String[]{"somegroup"}))); //the same as above but with non-meta region Mockito.when(mockRegionInfo.isSystemTable()).thenReturn(false); - assertEquals(HConstants.NORMAL_QOS, priority.getPriority(header, scanRequest)); + assertEquals(HConstants.NORMAL_QOS, priority.getPriority(header, scanRequest, + User.createUserForTesting(regionServer.conf, "someuser", new String[]{"somegroup"}))); } } http://git-wip-us.apache.org/repos/asf/hbase/blob/f28e3952/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestQosFunction.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestQosFunction.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestQosFunction.java index 2de4622..a3ad071 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestQosFunction.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestQosFunction.java @@ -28,6 +28,7 @@ import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos; import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos; +import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.hbase.testclassification.RegionServerTests; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest; @@ -55,11 +56,12 @@ public class TestQosFunction { new AnnotationReadingPriorityFunction(rpcServices, RSRpcServices.class); // Set method name in pb style with the method name capitalized. - checkMethod("ReplicateWALEntry", HConstants.REPLICATION_QOS, qosFunction); + checkMethod(conf, "ReplicateWALEntry", HConstants.REPLICATION_QOS, qosFunction); // Set method name in pb style with the method name capitalized. - checkMethod("OpenRegion", HConstants.ADMIN_QOS, qosFunction); + checkMethod(conf, "OpenRegion", HConstants.ADMIN_QOS, qosFunction); // Check multi works. - checkMethod("Multi", HConstants.NORMAL_QOS, qosFunction, MultiRequest.getDefaultInstance()); + checkMethod(conf, "Multi", HConstants.NORMAL_QOS, qosFunction, + MultiRequest.getDefaultInstance()); } @@ -103,19 +105,21 @@ public class TestQosFunction { .addTransition(normalTransition).build(); final String reportFuncName = "ReportRegionStateTransition"; - checkMethod(reportFuncName, HConstants.SYSTEMTABLE_QOS, qosFunction, metaTransitionRequest); - checkMethod(reportFuncName, HConstants.NORMAL_QOS, qosFunction, normalTransitionRequest); + checkMethod(conf, reportFuncName, HConstants.SYSTEMTABLE_QOS, qosFunction, + metaTransitionRequest); + checkMethod(conf, reportFuncName, HConstants.NORMAL_QOS, qosFunction, normalTransitionRequest); } - private void checkMethod(final String methodName, final int expected, + private void checkMethod(Configuration conf, final String methodName, final int expected, final AnnotationReadingPriorityFunction qosf) { - checkMethod(methodName, expected, qosf, null); + checkMethod(conf, methodName, expected, qosf, null); } - private void checkMethod(final String methodName, final int expected, + private void checkMethod(Configuration conf, final String methodName, final int expected, final AnnotationReadingPriorityFunction qosf, final Message param) { RequestHeader.Builder builder = RequestHeader.newBuilder(); builder.setMethodName(methodName); - assertEquals(methodName, expected, qosf.getPriority(builder.build(), param)); + assertEquals(methodName, expected, qosf.getPriority(builder.build(), param, + User.createUserForTesting(conf, "someuser", new String[]{"somegroup"}))); } } \ No newline at end of file