HADOOP-17346. Fair call queue is defeated by abusive service principals (#2431)
Co-authored-by: ahussein <ahmed.hussein@verizonmedia.com>
This commit is contained in:
parent
fc961b63d1
commit
5ce18101cb
@ -33,6 +33,7 @@
|
|||||||
import org.apache.hadoop.fs.CommonConfigurationKeys;
|
import org.apache.hadoop.fs.CommonConfigurationKeys;
|
||||||
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto;
|
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto;
|
||||||
|
|
||||||
|
import org.apache.hadoop.security.UserGroupInformation;
|
||||||
import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
|
import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
@ -212,6 +213,19 @@ int getPriorityLevel(Schedulable e) {
|
|||||||
return scheduler.getPriorityLevel(e);
|
return scheduler.getPriorityLevel(e);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
int getPriorityLevel(UserGroupInformation user) {
|
||||||
|
if (scheduler instanceof DecayRpcScheduler) {
|
||||||
|
return ((DecayRpcScheduler)scheduler).getPriorityLevel(user);
|
||||||
|
}
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
void setPriorityLevel(UserGroupInformation user, int priority) {
|
||||||
|
if (scheduler instanceof DecayRpcScheduler) {
|
||||||
|
((DecayRpcScheduler)scheduler).setPriorityLevel(user, priority);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
void setClientBackoffEnabled(boolean value) {
|
void setClientBackoffEnabled(boolean value) {
|
||||||
clientBackOffEnabled = value;
|
clientBackOffEnabled = value;
|
||||||
}
|
}
|
||||||
|
@ -40,6 +40,8 @@
|
|||||||
|
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
import com.fasterxml.jackson.databind.ObjectWriter;
|
import com.fasterxml.jackson.databind.ObjectWriter;
|
||||||
|
|
||||||
|
import org.apache.hadoop.security.UserGroupInformation;
|
||||||
import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
|
import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
|
||||||
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.AtomicDoubleArray;
|
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.AtomicDoubleArray;
|
||||||
import org.apache.commons.lang3.exception.ExceptionUtils;
|
import org.apache.commons.lang3.exception.ExceptionUtils;
|
||||||
@ -193,6 +195,7 @@ public class DecayRpcScheduler implements RpcScheduler,
|
|||||||
private static final double PRECISION = 0.0001;
|
private static final double PRECISION = 0.0001;
|
||||||
private MetricsProxy metricsProxy;
|
private MetricsProxy metricsProxy;
|
||||||
private final CostProvider costProvider;
|
private final CostProvider costProvider;
|
||||||
|
private final Map<String, Integer> staticPriorities = new HashMap<>();
|
||||||
private Set<String> serviceUserNames;
|
private Set<String> serviceUserNames;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -581,7 +584,10 @@ private int computePriorityLevel(long cost, Object identity) {
|
|||||||
if (isServiceUser((String)identity)) {
|
if (isServiceUser((String)identity)) {
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
Integer staticPriority = staticPriorities.get(identity);
|
||||||
|
if (staticPriority != null) {
|
||||||
|
return staticPriority.intValue();
|
||||||
|
}
|
||||||
long totalCallSnapshot = totalDecayedCallCost.get();
|
long totalCallSnapshot = totalDecayedCallCost.get();
|
||||||
|
|
||||||
double proportion = 0;
|
double proportion = 0;
|
||||||
@ -626,6 +632,15 @@ private int cachedOrComputedPriorityLevel(Object identity) {
|
|||||||
return priority;
|
return priority;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private String getIdentity(Schedulable obj) {
|
||||||
|
String identity = this.identityProvider.makeIdentity(obj);
|
||||||
|
if (identity == null) {
|
||||||
|
// Identity provider did not handle this
|
||||||
|
identity = DECAYSCHEDULER_UNKNOWN_IDENTITY;
|
||||||
|
}
|
||||||
|
return identity;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Compute the appropriate priority for a schedulable based on past requests.
|
* Compute the appropriate priority for a schedulable based on past requests.
|
||||||
* @param obj the schedulable obj to query and remember
|
* @param obj the schedulable obj to query and remember
|
||||||
@ -634,15 +649,42 @@ private int cachedOrComputedPriorityLevel(Object identity) {
|
|||||||
@Override
|
@Override
|
||||||
public int getPriorityLevel(Schedulable obj) {
|
public int getPriorityLevel(Schedulable obj) {
|
||||||
// First get the identity
|
// First get the identity
|
||||||
String identity = this.identityProvider.makeIdentity(obj);
|
String identity = getIdentity(obj);
|
||||||
if (identity == null) {
|
// highest priority users may have a negative priority but their
|
||||||
// Identity provider did not handle this
|
// calls will be priority 0.
|
||||||
identity = DECAYSCHEDULER_UNKNOWN_IDENTITY;
|
return Math.max(0, cachedOrComputedPriorityLevel(identity));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@VisibleForTesting
|
||||||
|
int getPriorityLevel(UserGroupInformation ugi) {
|
||||||
|
String identity = getIdentity(newSchedulable(ugi));
|
||||||
|
// returns true priority of the user.
|
||||||
return cachedOrComputedPriorityLevel(identity);
|
return cachedOrComputedPriorityLevel(identity);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@VisibleForTesting
|
||||||
|
void setPriorityLevel(UserGroupInformation ugi, int priority) {
|
||||||
|
String identity = getIdentity(newSchedulable(ugi));
|
||||||
|
priority = Math.min(numLevels - 1, priority);
|
||||||
|
LOG.info("Setting priority for user:" + identity + "=" + priority);
|
||||||
|
staticPriorities.put(identity, priority);
|
||||||
|
}
|
||||||
|
|
||||||
|
// dummy instance to conform to identity provider api.
|
||||||
|
private static Schedulable newSchedulable(UserGroupInformation ugi) {
|
||||||
|
return new Schedulable() {
|
||||||
|
@Override
|
||||||
|
public UserGroupInformation getUserGroupInformation() {
|
||||||
|
return ugi;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public int getPriorityLevel() {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
private boolean isServiceUser(String userName) {
|
private boolean isServiceUser(String userName) {
|
||||||
return this.serviceUserNames.contains(userName);
|
return this.serviceUserNames.contains(userName);
|
||||||
}
|
}
|
||||||
|
@ -51,6 +51,7 @@
|
|||||||
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto;
|
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcStatusProto;
|
||||||
import org.apache.hadoop.net.NetUtils;
|
import org.apache.hadoop.net.NetUtils;
|
||||||
import org.apache.hadoop.security.SaslRpcServer;
|
import org.apache.hadoop.security.SaslRpcServer;
|
||||||
|
import org.apache.hadoop.security.SecurityUtil;
|
||||||
import org.apache.hadoop.security.UserGroupInformation;
|
import org.apache.hadoop.security.UserGroupInformation;
|
||||||
import org.apache.hadoop.security.token.SecretManager;
|
import org.apache.hadoop.security.token.SecretManager;
|
||||||
import org.apache.hadoop.security.token.TokenIdentifier;
|
import org.apache.hadoop.security.token.TokenIdentifier;
|
||||||
@ -980,7 +981,18 @@ void registerProtocolAndImpl(RpcKind rpcKind, Class<?> protocolClass,
|
|||||||
" ProtocolImpl=" + protocolImpl.getClass().getName() +
|
" ProtocolImpl=" + protocolImpl.getClass().getName() +
|
||||||
" protocolClass=" + protocolClass.getName());
|
" protocolClass=" + protocolClass.getName());
|
||||||
}
|
}
|
||||||
}
|
String client = SecurityUtil.getClientPrincipal(protocolClass, getConf());
|
||||||
|
if (client != null) {
|
||||||
|
// notify the server's rpc scheduler that the protocol user has
|
||||||
|
// highest priority. the scheduler should exempt the user from
|
||||||
|
// priority calculations.
|
||||||
|
try {
|
||||||
|
setPriorityLevel(UserGroupInformation.createRemoteUser(client), -1);
|
||||||
|
} catch (Exception ex) {
|
||||||
|
LOG.warn("Failed to set scheduling priority for " + client, ex);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
static class VerProtocolImpl {
|
static class VerProtocolImpl {
|
||||||
final long version;
|
final long version;
|
||||||
|
@ -643,7 +643,22 @@ public static void bind(ServerSocket socket, InetSocketAddress address,
|
|||||||
address.getPort(), e);
|
address.getPort(), e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting
|
||||||
|
int getPriorityLevel(Schedulable e) {
|
||||||
|
return callQueue.getPriorityLevel(e);
|
||||||
|
}
|
||||||
|
|
||||||
|
@org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting
|
||||||
|
int getPriorityLevel(UserGroupInformation ugi) {
|
||||||
|
return callQueue.getPriorityLevel(ugi);
|
||||||
|
}
|
||||||
|
|
||||||
|
@org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting
|
||||||
|
void setPriorityLevel(UserGroupInformation ugi, int priority) {
|
||||||
|
callQueue.setPriorityLevel(ugi, priority);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Returns a handle to the rpcMetrics (required in tests)
|
* Returns a handle to the rpcMetrics (required in tests)
|
||||||
* @return rpc metrics
|
* @return rpc metrics
|
||||||
|
@ -31,6 +31,6 @@ public String makeIdentity(Schedulable obj) {
|
|||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
return ugi.getUserName();
|
return ugi.getShortUserName();
|
||||||
}
|
}
|
||||||
}
|
}
|
@ -380,7 +380,25 @@ public static void setSecurityInfoProviders(SecurityInfo... providers) {
|
|||||||
}
|
}
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Look up the client principal for a given protocol. It searches all known
|
||||||
|
* SecurityInfo providers.
|
||||||
|
* @param protocol the protocol class to get the information for
|
||||||
|
* @param conf configuration object
|
||||||
|
* @return client principal or null if it has no client principal defined.
|
||||||
|
*/
|
||||||
|
public static String getClientPrincipal(Class<?> protocol,
|
||||||
|
Configuration conf) {
|
||||||
|
String user = null;
|
||||||
|
KerberosInfo krbInfo = SecurityUtil.getKerberosInfo(protocol, conf);
|
||||||
|
if (krbInfo != null) {
|
||||||
|
String key = krbInfo.clientPrincipal();
|
||||||
|
user = (key != null && !key.isEmpty()) ? conf.get(key) : null;
|
||||||
|
}
|
||||||
|
return user;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Look up the TokenInfo for a given protocol. It searches all known
|
* Look up the TokenInfo for a given protocol. It searches all known
|
||||||
* SecurityInfo providers.
|
* SecurityInfo providers.
|
||||||
|
@ -28,7 +28,6 @@
|
|||||||
import org.apache.hadoop.classification.InterfaceStability;
|
import org.apache.hadoop.classification.InterfaceStability;
|
||||||
import org.apache.hadoop.conf.Configuration;
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.fs.CommonConfigurationKeys;
|
import org.apache.hadoop.fs.CommonConfigurationKeys;
|
||||||
import org.apache.hadoop.security.KerberosInfo;
|
|
||||||
import org.apache.hadoop.security.SecurityUtil;
|
import org.apache.hadoop.security.SecurityUtil;
|
||||||
import org.apache.hadoop.security.UserGroupInformation;
|
import org.apache.hadoop.security.UserGroupInformation;
|
||||||
import org.apache.hadoop.util.MachineList;
|
import org.apache.hadoop.util.MachineList;
|
||||||
@ -101,21 +100,19 @@ public void authorize(UserGroupInformation user,
|
|||||||
String clientPrincipal = null;
|
String clientPrincipal = null;
|
||||||
if (UserGroupInformation.isSecurityEnabled()) {
|
if (UserGroupInformation.isSecurityEnabled()) {
|
||||||
// get client principal key to verify (if available)
|
// get client principal key to verify (if available)
|
||||||
KerberosInfo krbInfo = SecurityUtil.getKerberosInfo(protocol, conf);
|
clientPrincipal = SecurityUtil.getClientPrincipal(protocol, conf);
|
||||||
if (krbInfo != null) {
|
try {
|
||||||
String clientKey = krbInfo.clientPrincipal();
|
if (clientPrincipal != null) {
|
||||||
if (clientKey != null && !clientKey.isEmpty()) {
|
clientPrincipal =
|
||||||
try {
|
SecurityUtil.getServerPrincipal(clientPrincipal, addr);
|
||||||
clientPrincipal = SecurityUtil.getServerPrincipal(
|
|
||||||
conf.get(clientKey), addr);
|
|
||||||
} catch (IOException e) {
|
|
||||||
throw (AuthorizationException) new AuthorizationException(
|
|
||||||
"Can't figure out Kerberos principal name for connection from "
|
|
||||||
+ addr + " for user=" + user + " protocol=" + protocol)
|
|
||||||
.initCause(e);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
} catch (IOException e) {
|
||||||
|
throw (AuthorizationException) new AuthorizationException(
|
||||||
|
"Can't figure out Kerberos principal name for connection from "
|
||||||
|
+ addr + " for user=" + user + " protocol=" + protocol)
|
||||||
|
.initCause(e);
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
if((clientPrincipal != null && !clientPrincipal.equals(user.getUserName())) ||
|
if((clientPrincipal != null && !clientPrincipal.equals(user.getUserName())) ||
|
||||||
acls.length != 2 || !acls[0].isUserAllowed(user) || acls[1].isUserAllowed(user)) {
|
acls.length != 2 || !acls[0].isUserAllowed(user) || acls[1].isUserAllowed(user)) {
|
||||||
|
@ -48,9 +48,8 @@
|
|||||||
public class TestDecayRpcScheduler {
|
public class TestDecayRpcScheduler {
|
||||||
private Schedulable mockCall(String id) {
|
private Schedulable mockCall(String id) {
|
||||||
Schedulable mockCall = mock(Schedulable.class);
|
Schedulable mockCall = mock(Schedulable.class);
|
||||||
UserGroupInformation ugi = mock(UserGroupInformation.class);
|
UserGroupInformation ugi = UserGroupInformation.createRemoteUser(id);
|
||||||
|
|
||||||
when(ugi.getUserName()).thenReturn(id);
|
|
||||||
when(mockCall.getUserGroupInformation()).thenReturn(ugi);
|
when(mockCall.getUserGroupInformation()).thenReturn(ugi);
|
||||||
|
|
||||||
return mockCall;
|
return mockCall;
|
||||||
|
@ -1294,6 +1294,43 @@ public void testDecayRpcSchedulerMetrics() throws Exception {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test (timeout=30000)
|
||||||
|
public void testProtocolUserPriority() throws Exception {
|
||||||
|
final String ns = CommonConfigurationKeys.IPC_NAMESPACE + ".0";
|
||||||
|
conf.set(CLIENT_PRINCIPAL_KEY, "clientForProtocol");
|
||||||
|
Server server = null;
|
||||||
|
try {
|
||||||
|
server = setupDecayRpcSchedulerandTestServer(ns + ".");
|
||||||
|
|
||||||
|
UserGroupInformation ugi = UserGroupInformation.createRemoteUser("user");
|
||||||
|
// normal users start with priority 0.
|
||||||
|
Assert.assertEquals(0, server.getPriorityLevel(ugi));
|
||||||
|
// calls for a protocol defined client will have priority of 0.
|
||||||
|
Assert.assertEquals(0, server.getPriorityLevel(newSchedulable(ugi)));
|
||||||
|
|
||||||
|
// protocol defined client will have top priority of -1.
|
||||||
|
ugi = UserGroupInformation.createRemoteUser("clientForProtocol");
|
||||||
|
Assert.assertEquals(-1, server.getPriorityLevel(ugi));
|
||||||
|
// calls for a protocol defined client will have priority of 0.
|
||||||
|
Assert.assertEquals(0, server.getPriorityLevel(newSchedulable(ugi)));
|
||||||
|
} finally {
|
||||||
|
stop(server, null);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Schedulable newSchedulable(UserGroupInformation ugi) {
|
||||||
|
return new Schedulable(){
|
||||||
|
@Override
|
||||||
|
public UserGroupInformation getUserGroupInformation() {
|
||||||
|
return ugi;
|
||||||
|
}
|
||||||
|
@Override
|
||||||
|
public int getPriorityLevel() {
|
||||||
|
return 0; // doesn't matter.
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
private Server setupDecayRpcSchedulerandTestServer(String ns)
|
private Server setupDecayRpcSchedulerandTestServer(String ns)
|
||||||
throws Exception {
|
throws Exception {
|
||||||
final int queueSizePerHandler = 3;
|
final int queueSizePerHandler = 3;
|
||||||
|
@ -62,6 +62,8 @@ public class TestRpcBase {
|
|||||||
|
|
||||||
protected final static String SERVER_PRINCIPAL_KEY =
|
protected final static String SERVER_PRINCIPAL_KEY =
|
||||||
"test.ipc.server.principal";
|
"test.ipc.server.principal";
|
||||||
|
protected final static String CLIENT_PRINCIPAL_KEY =
|
||||||
|
"test.ipc.client.principal";
|
||||||
protected final static String ADDRESS = "0.0.0.0";
|
protected final static String ADDRESS = "0.0.0.0";
|
||||||
protected final static int PORT = 0;
|
protected final static int PORT = 0;
|
||||||
protected static InetSocketAddress addr;
|
protected static InetSocketAddress addr;
|
||||||
@ -271,7 +273,8 @@ public Token<TestTokenIdentifier> selectToken(Text service,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@KerberosInfo(serverPrincipal = SERVER_PRINCIPAL_KEY)
|
@KerberosInfo(serverPrincipal = SERVER_PRINCIPAL_KEY,
|
||||||
|
clientPrincipal = CLIENT_PRINCIPAL_KEY)
|
||||||
@TokenInfo(TestTokenSelector.class)
|
@TokenInfo(TestTokenSelector.class)
|
||||||
@ProtocolInfo(protocolName = "org.apache.hadoop.ipc.TestRpcBase$TestRpcService",
|
@ProtocolInfo(protocolName = "org.apache.hadoop.ipc.TestRpcBase$TestRpcService",
|
||||||
protocolVersion = 1)
|
protocolVersion = 1)
|
||||||
|
Loading…
Reference in New Issue
Block a user