You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by cn...@apache.org on 2013/06/21 08:37:39 UTC
svn commit: r1495297 [13/46] - in /hadoop/common/branches/branch-1-win: ./
bin/ conf/ ivy/ lib/jdiff/ src/c++/libhdfs/docs/
src/c++/libhdfs/tests/conf/ src/contrib/capacity-scheduler/ivy/
src/contrib/capacity-scheduler/src/java/org/apache/hadoop/mapred...
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryPolicies.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryPolicies.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryPolicies.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryPolicies.java Fri Jun 21 06:37:27 2013
@@ -17,13 +17,17 @@
*/
package org.apache.hadoop.io.retry;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
-import java.util.Random;
-import java.util.Set;
import java.util.Map.Entry;
+import java.util.Random;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.ipc.RemoteException;
/**
@@ -32,7 +36,15 @@ import org.apache.hadoop.ipc.RemoteExcep
* </p>
*/
public class RetryPolicies {
-
+ private static final Log LOG = LogFactory.getLog(RetryPolicies.class);
+
+ private static ThreadLocal<Random> RANDOM = new ThreadLocal<Random>() {
+ @Override
+ protected Random initialValue() {
+ return new Random();
+ }
+ };
+
/**
* <p>
* Try once, and fail by re-throwing the exception.
@@ -139,17 +151,35 @@ public class RetryPolicies {
}
}
+ /**
+ * Retry up to maxRetries.
+ * The actual sleep time of the n-th retry is f(n, sleepTime),
+ * where f is a function provided by the subclass implementation.
+ *
+ * The object of the subclasses should be immutable;
+ * otherwise, the subclass must override hashCode(), equals(..) and toString().
+ */
static abstract class RetryLimited implements RetryPolicy {
- int maxRetries;
- long sleepTime;
- TimeUnit timeUnit;
+ final int maxRetries;
+ final long sleepTime;
+ final TimeUnit timeUnit;
+
+ private String myString;
- public RetryLimited(int maxRetries, long sleepTime, TimeUnit timeUnit) {
+ RetryLimited(int maxRetries, long sleepTime, TimeUnit timeUnit) {
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries = " + maxRetries+" < 0");
+ }
+ if (sleepTime < 0) {
+ throw new IllegalArgumentException("sleepTime = " + sleepTime + " < 0");
+ }
+
this.maxRetries = maxRetries;
this.sleepTime = sleepTime;
this.timeUnit = timeUnit;
}
+ @Override
public boolean shouldRetry(Exception e, int retries) throws Exception {
if (retries >= maxRetries) {
throw e;
@@ -163,6 +193,30 @@ public class RetryPolicies {
}
protected abstract long calculateSleepTime(int retries);
+
+ @Override
+ public int hashCode() {
+ return toString().hashCode();
+ }
+
+ @Override
+ public boolean equals(final Object that) {
+ if (this == that) {
+ return true;
+ } else if (that == null || this.getClass() != that.getClass()) {
+ return false;
+ }
+ return this.toString().equals(that.toString());
+ }
+
+ @Override
+ public String toString() {
+ if (myString == null) {
+ myString = getClass().getSimpleName() + "(maxRetries=" + maxRetries
+ + ", sleepTime=" + sleepTime + " " + timeUnit + ")";
+ }
+ return myString;
+ }
}
static class RetryUpToMaximumCountWithFixedSleep extends RetryLimited {
@@ -192,6 +246,169 @@ public class RetryPolicies {
return sleepTime * (retries + 1);
}
}
+
+ /**
+ * Given pairs of number of retries and sleep time (n0, t0), (n1, t1), ...,
+ * the first n0 retries sleep t0 milliseconds on average,
+ * the following n1 retries sleep t1 milliseconds on average, and so on.
+ *
+ * For all the sleep, the actual sleep time is randomly uniform distributed
+ * in the close interval [0.5t, 1.5t], where t is the sleep time specified.
+ *
+ * The objects of this class are immutable.
+ */
+ public static class MultipleLinearRandomRetry implements RetryPolicy {
+ /** Pairs of numRetries and sleepSeconds */
+ public static class Pair {
+ final int numRetries;
+ final int sleepMillis;
+
+ public Pair(final int numRetries, final int sleepMillis) {
+ if (numRetries < 0) {
+ throw new IllegalArgumentException("numRetries = " + numRetries+" < 0");
+ }
+ if (sleepMillis < 0) {
+ throw new IllegalArgumentException("sleepMillis = " + sleepMillis + " < 0");
+ }
+
+ this.numRetries = numRetries;
+ this.sleepMillis = sleepMillis;
+ }
+
+ @Override
+ public String toString() {
+ return numRetries + "x" + sleepMillis + "ms";
+ }
+ }
+
+ private final List<Pair> pairs;
+ private String myString;
+
+ public MultipleLinearRandomRetry(List<Pair> pairs) {
+ if (pairs == null || pairs.isEmpty()) {
+ throw new IllegalArgumentException("pairs must be neither null nor empty.");
+ }
+ this.pairs = Collections.unmodifiableList(pairs);
+ }
+
+ @Override
+ public boolean shouldRetry(Exception e, int curRetry) throws Exception {
+ final Pair p = searchPair(curRetry);
+ if (p == null) {
+ //no more retries, re-throw the original exception.
+ throw e;
+ }
+
+ //sleep and return true.
+ //If the sleep is interrupted, throw the InterruptedException out.
+ final double ratio = RANDOM.get().nextDouble() + 0.5;//0.5 <= ratio <=1.5
+ Thread.sleep(Math.round(p.sleepMillis * ratio));
+ return true;
+ }
+
+ /**
+ * Given the current number of retry, search the corresponding pair.
+ * @return the corresponding pair,
+ * or null if the current number of retry > maximum number of retry.
+ */
+ private Pair searchPair(int curRetry) {
+ int i = 0;
+ for(; i < pairs.size() && curRetry > pairs.get(i).numRetries; i++) {
+ curRetry -= pairs.get(i).numRetries;
+ }
+ return i == pairs.size()? null: pairs.get(i);
+ }
+
+ @Override
+ public int hashCode() {
+ return toString().hashCode();
+ }
+
+ @Override
+ public boolean equals(final Object that) {
+ if (this == that) {
+ return true;
+ } else if (that == null || this.getClass() != that.getClass()) {
+ return false;
+ }
+ return this.toString().equals(that.toString());
+ }
+
+ @Override
+ public String toString() {
+ if (myString == null) {
+ myString = getClass().getSimpleName() + pairs;
+ }
+ return myString;
+ }
+
+ /**
+ * Parse the given string as a MultipleLinearRandomRetry object.
+ * The format of the string is "t_1, n_1, t_2, n_2, ...",
+ * where t_i and n_i are the i-th pair of sleep time and number of retires.
+ * Note that the white spaces in the string are ignored.
+ *
+ * @return the parsed object, or null if the parsing fails.
+ */
+ public static MultipleLinearRandomRetry parseCommaSeparatedString(String s) {
+ final String[] elements = s.split(",");
+ if (elements.length == 0) {
+ LOG.warn("Illegal value: there is no element in \"" + s + "\".");
+ return null;
+ }
+ if (elements.length % 2 != 0) {
+ LOG.warn("Illegal value: the number of elements in \"" + s + "\" is "
+ + elements.length + " but an even number of elements is expected.");
+ return null;
+ }
+
+ final List<RetryPolicies.MultipleLinearRandomRetry.Pair> pairs
+ = new ArrayList<RetryPolicies.MultipleLinearRandomRetry.Pair>();
+
+ for(int i = 0; i < elements.length; ) {
+ //parse the i-th sleep-time
+ final int sleep = parsePositiveInt(elements, i++, s);
+ if (sleep == -1) {
+ return null; //parse fails
+ }
+
+ //parse the i-th number-of-retries
+ final int retries = parsePositiveInt(elements, i++, s);
+ if (retries == -1) {
+ return null; //parse fails
+ }
+
+ pairs.add(new RetryPolicies.MultipleLinearRandomRetry.Pair(retries, sleep));
+ }
+ return new RetryPolicies.MultipleLinearRandomRetry(pairs);
+ }
+
+ /**
+ * Parse the i-th element as an integer.
+ * @return -1 if the parsing fails or the parsed value <= 0;
+ * otherwise, return the parsed value.
+ */
+ private static int parsePositiveInt(final String[] elements,
+ final int i, final String originalString) {
+ final String s = elements[i].trim();
+ final int n;
+ try {
+ n = Integer.parseInt(s);
+ } catch(NumberFormatException nfe) {
+ LOG.warn("Failed to parse \"" + s + "\", which is the index " + i
+ + " element in \"" + originalString + "\"", nfe);
+ return -1;
+ }
+
+ if (n <= 0) {
+ LOG.warn("The value " + n + " <= 0: it is parsed from the string \""
+ + s + "\" which is the index " + i + " element in \""
+ + originalString + "\"");
+ return -1;
+ }
+ return n;
+ }
+ }
static class ExceptionDependentRetry implements RetryPolicy {
@@ -248,6 +465,13 @@ public class RetryPolicies {
public ExponentialBackoffRetry(
int maxRetries, long sleepTime, TimeUnit timeUnit) {
super(maxRetries, sleepTime, timeUnit);
+
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries = " + maxRetries + " < 0");
+ } else if (maxRetries > 30) {
+ //if maxRetries > 30, calculateSleepTime will overflow.
+ throw new IllegalArgumentException("maxRetries = " + maxRetries + " > 30");
+ }
}
@Override
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryProxy.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryProxy.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryProxy.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryProxy.java Fri Jun 21 06:37:27 2013
@@ -65,4 +65,13 @@ public class RetryProxy {
new RetryInvocationHandler(implementation, methodNameToPolicyMap)
);
}
+
+ public static Object create(Class<?> iface, Object implementation,
+ RetryPolicy defaultPolicy, Map<String,RetryPolicy> methodNameToPolicyMap) {
+ return Proxy.newProxyInstance(
+ implementation.getClass().getClassLoader(),
+ new Class<?>[] { iface },
+ new RetryInvocationHandler(implementation, defaultPolicy, methodNameToPolicyMap)
+ );
+ }
}
Added: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryUtils.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryUtils.java?rev=1495297&view=auto
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryUtils.java (added)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/io/retry/RetryUtils.java Fri Jun 21 06:37:27 2013
@@ -0,0 +1,155 @@
+/**
+ * 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.io.retry;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RemoteException;
+
+public class RetryUtils {
+ public static final Log LOG = LogFactory.getLog(RetryUtils.class);
+
+ /**
+ * Return the default retry policy set in conf.
+ *
+ * If the value retryPolicyEnabledKey is set to false in conf,
+ * use TRY_ONCE_THEN_FAIL.
+ *
+ * Otherwise, get the MultipleLinearRandomRetry policy specified in the conf
+ * and then
+ * (1) use multipleLinearRandomRetry for
+ * - remoteExceptionToRetry, or
+ * - IOException other than RemoteException; and
+ * (2) use TRY_ONCE_THEN_FAIL for
+ * - RemoteException other than remoteExceptionToRetry, or
+ * - non-IOException.
+ *
+ * @param conf
+ * @param retryPolicyEnabledKey conf property key for enabling retry
+ * @param defaultRetryPolicyEnabled default retryPolicyEnabledKey conf value
+ * @param retryPolicySpecKey conf property key for retry policy spec
+ * @param defaultRetryPolicySpec default retryPolicySpecKey conf value
+ * @param remoteExceptionsToRetry The particular RemoteExceptions to retry
+ * @return the default retry policy.
+ */
+ public static RetryPolicy getDefaultRetryPolicy(
+ Configuration conf,
+ String retryPolicyEnabledKey,
+ boolean defaultRetryPolicyEnabled,
+ String retryPolicySpecKey,
+ String defaultRetryPolicySpec,
+ final Class<? extends Exception> ... remoteExceptionsToRetry
+ ) {
+
+ final RetryPolicy multipleLinearRandomRetry =
+ getMultipleLinearRandomRetry(
+ conf,
+ retryPolicyEnabledKey, defaultRetryPolicyEnabled,
+ retryPolicySpecKey, defaultRetryPolicySpec
+ );
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("multipleLinearRandomRetry = " + multipleLinearRandomRetry);
+ }
+
+ if (multipleLinearRandomRetry == null) {
+ //no retry
+ return RetryPolicies.TRY_ONCE_THEN_FAIL;
+ } else {
+ return new RetryPolicy() {
+ @Override
+ public boolean shouldRetry(Exception e, int retries) throws Exception {
+ //see (1) and (2) in the javadoc of this method.
+ final RetryPolicy p;
+ if (e instanceof RemoteException) {
+ final RemoteException re = (RemoteException)e;
+ RetryPolicy found = null;
+ for(Class<? extends Exception> reToRetry : remoteExceptionsToRetry) {
+ if (reToRetry.getName().equals(re.getClassName())) {
+ found = multipleLinearRandomRetry;
+ break;
+ }
+ }
+ p = found != null? found: RetryPolicies.TRY_ONCE_THEN_FAIL;
+ } else if (e instanceof IOException) {
+ p = multipleLinearRandomRetry;
+ } else { //non-IOException
+ p = RetryPolicies.TRY_ONCE_THEN_FAIL;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("RETRY " + retries + ") policy="
+ + p.getClass().getSimpleName() + ", exception=" + e);
+ }
+ return p.shouldRetry(e, retries);
+ }
+
+ @Override
+ public String toString() {
+ return "RetryPolicy[" + multipleLinearRandomRetry + ", "
+ + RetryPolicies.TRY_ONCE_THEN_FAIL.getClass().getSimpleName()
+ + "]";
+ }
+ };
+ }
+ }
+
+ /**
+ * Return the MultipleLinearRandomRetry policy specified in the conf,
+ * or null if the feature is disabled.
+ * If the policy is specified in the conf but the policy cannot be parsed,
+ * the default policy is returned.
+ *
+ * Retry policy spec:
+ * N pairs of sleep-time and number-of-retries "s1,n1,s2,n2,..."
+ *
+ * @param conf
+ * @param retryPolicyEnabledKey conf property key for enabling retry
+ * @param defaultRetryPolicyEnabled default retryPolicyEnabledKey conf value
+ * @param retryPolicySpecKey conf property key for retry policy spec
+ * @param defaultRetryPolicySpec default retryPolicySpecKey conf value
+ * @return the MultipleLinearRandomRetry policy specified in the conf,
+ * or null if the feature is disabled.
+ */
+ public static RetryPolicy getMultipleLinearRandomRetry(
+ Configuration conf,
+ String retryPolicyEnabledKey,
+ boolean defaultRetryPolicyEnabled,
+ String retryPolicySpecKey,
+ String defaultRetryPolicySpec
+ ) {
+ final boolean enabled =
+ conf.getBoolean(retryPolicyEnabledKey, defaultRetryPolicyEnabled);
+ if (!enabled) {
+ return null;
+ }
+
+ final String policy = conf.get(retryPolicySpecKey, defaultRetryPolicySpec);
+
+ final RetryPolicy r =
+ RetryPolicies.MultipleLinearRandomRetry.parseCommaSeparatedString(
+ policy);
+ return (r != null) ?
+ r :
+ RetryPolicies.MultipleLinearRandomRetry.parseCommaSeparatedString(
+ defaultRetryPolicySpec);
+ }
+}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Client.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Client.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Client.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Client.java Fri Jun 21 06:37:27 2013
@@ -18,40 +18,41 @@
package org.apache.hadoop.ipc;
-import java.net.InetAddress;
-import java.net.Socket;
-import java.net.InetSocketAddress;
-import java.net.SocketTimeoutException;
-import java.net.UnknownHostException;
-import java.net.ConnectException;
-
-import java.io.IOException;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.FilterInputStream;
+import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-
+import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
import java.security.PrivilegedExceptionAction;
import java.util.Hashtable;
import java.util.Iterator;
+import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
-import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import javax.net.SocketFactory;
-import org.apache.commons.logging.*;
-
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
-import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.KerberosInfo;
import org.apache.hadoop.security.SaslRpcClient;
@@ -60,8 +61,8 @@ import org.apache.hadoop.security.Securi
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenIdentifier;
-import org.apache.hadoop.security.token.TokenSelector;
import org.apache.hadoop.security.token.TokenInfo;
+import org.apache.hadoop.security.token.TokenSelector;
import org.apache.hadoop.util.ReflectionUtils;
/** A client for an IPC service. IPC calls take a single {@link Writable} as a
@@ -71,7 +72,9 @@ import org.apache.hadoop.util.Reflection
* @see Server
*/
public class Client {
-
+ public static final String IPC_CLIENT_CONNECT_MAX_RETRIES_KEY = "ipc.client.connect.max.retries";
+ public static final int IPC_CLIENT_CONNECT_MAX_RETRIES_DEFAULT = 10;
+
public static final Log LOG =
LogFactory.getLog(Client.class);
private Hashtable<ConnectionId, Connection> connections =
@@ -111,6 +114,21 @@ public class Client {
}
/**
+ * The time after which a RPC will timeout. If ping is not enabled (via
+ * ipc.client.ping), then the timeout value is the same as the pingInterval.
+ * If ping is enabled, then there is no timeout value.
+ *
+ * @param conf Configuration
+ * @return the timeout period in milliseconds. -1 if no timeout value is set
+ */
+ final public static int getTimeout(Configuration conf) {
+ if (!conf.getBoolean("ipc.client.ping", true)) {
+ return getPingInterval(conf);
+ }
+ return -1;
+ }
+
+ /**
* Increment this client's reference count
*
*/
@@ -197,9 +215,10 @@ public class Client {
private int rpcTimeout;
private int maxIdleTime; //connections will be culled if it was idle for
//maxIdleTime msecs
- private int maxRetries; //the max. no. of retries for socket connections
+ private final RetryPolicy connectionRetryPolicy;
private boolean tcpNoDelay; // if T then disable Nagle's Algorithm
private int pingInterval; // how often sends ping to the server in msecs
+
// currently active calls
private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>();
@@ -215,7 +234,7 @@ public class Client {
remoteId.getAddress().getHostName());
}
this.maxIdleTime = remoteId.getMaxIdleTime();
- this.maxRetries = remoteId.getMaxRetries();
+ this.connectionRetryPolicy = remoteId.connectionRetryPolicy;
this.tcpNoDelay = remoteId.getTcpNoDelay();
this.pingInterval = remoteId.getPingInterval();
if (LOG.isDebugEnabled()) {
@@ -453,7 +472,7 @@ public class Client {
if (updateAddress()) {
timeoutFailures = ioFailures = 0;
}
- handleConnectionFailure(ioFailures++, maxRetries, ie);
+ handleConnectionFailure(ioFailures++, ie);
}
}
}
@@ -663,8 +682,26 @@ public class Client {
Thread.sleep(1000);
} catch (InterruptedException ignored) {}
- LOG.info("Retrying connect to server: " + server +
- ". Already tried " + curRetries + " time(s).");
+ LOG.info("Retrying connect to server: " + server + ". Already tried "
+ + curRetries + " time(s); maxRetries=" + maxRetries);
+ }
+
+ private void handleConnectionFailure(int curRetries, IOException ioe
+ ) throws IOException {
+ closeConnection();
+
+ final boolean retry;
+ try {
+ retry = connectionRetryPolicy.shouldRetry(ioe, curRetries);
+ } catch(Exception e) {
+ throw e instanceof IOException? (IOException)e: new IOException(e);
+ }
+ if (!retry) {
+ throw ioe;
+ }
+
+ LOG.info("Retrying connect to server: " + server + ". Already tried "
+ + curRetries + " time(s); retry policy is " + connectionRetryPolicy);
}
/* Write the RPC header */
@@ -1220,14 +1257,15 @@ public class Client {
private String serverPrincipal;
private int maxIdleTime; //connections will be culled if it was idle for
//maxIdleTime msecs
- private int maxRetries; //the max. no. of retries for socket connections
+ private final RetryPolicy connectionRetryPolicy;
private boolean tcpNoDelay; // if T then disable Nagle's Algorithm
private int pingInterval; // how often sends ping to the server in msecs
+
ConnectionId(InetSocketAddress address, Class<?> protocol,
UserGroupInformation ticket, int rpcTimeout,
String serverPrincipal, int maxIdleTime,
- int maxRetries, boolean tcpNoDelay,
+ RetryPolicy connectionRetryPolicy, boolean tcpNoDelay,
int pingInterval) {
this.protocol = protocol;
this.address = address;
@@ -1235,7 +1273,7 @@ public class Client {
this.rpcTimeout = rpcTimeout;
this.serverPrincipal = serverPrincipal;
this.maxIdleTime = maxIdleTime;
- this.maxRetries = maxRetries;
+ this.connectionRetryPolicy = connectionRetryPolicy;
this.tcpNoDelay = tcpNoDelay;
this.pingInterval = pingInterval;
}
@@ -1264,10 +1302,6 @@ public class Client {
return maxIdleTime;
}
- int getMaxRetries() {
- return maxRetries;
- }
-
boolean getTcpNoDelay() {
return tcpNoDelay;
}
@@ -1285,11 +1319,26 @@ public class Client {
static ConnectionId getConnectionId(InetSocketAddress addr,
Class<?> protocol, UserGroupInformation ticket, int rpcTimeout,
Configuration conf) throws IOException {
+ return getConnectionId(addr, protocol, ticket, rpcTimeout, null, conf);
+ }
+
+ static ConnectionId getConnectionId(InetSocketAddress addr,
+ Class<?> protocol, UserGroupInformation ticket, int rpcTimeout,
+ RetryPolicy connectionRetryPolicy, Configuration conf) throws IOException {
+
+ if (connectionRetryPolicy == null) {
+ final int max = conf.getInt(
+ IPC_CLIENT_CONNECT_MAX_RETRIES_KEY,
+ IPC_CLIENT_CONNECT_MAX_RETRIES_DEFAULT);
+ connectionRetryPolicy = RetryPolicies.retryUpToMaximumCountWithFixedSleep(
+ max, 1, TimeUnit.SECONDS);
+ }
+
String remotePrincipal = getRemotePrincipal(conf, addr, protocol);
return new ConnectionId(addr, protocol, ticket,
rpcTimeout, remotePrincipal,
conf.getInt("ipc.client.connection.maxidletime", 10000), // 10s
- conf.getInt("ipc.client.connect.max.retries", 10),
+ connectionRetryPolicy,
conf.getBoolean("ipc.client.tcpnodelay", false),
Client.getPingInterval(conf));
}
@@ -1326,7 +1375,7 @@ public class Client {
ConnectionId that = (ConnectionId) obj;
return isEqual(this.address, that.address)
&& this.maxIdleTime == that.maxIdleTime
- && this.maxRetries == that.maxRetries
+ && isEqual(this.connectionRetryPolicy, that.connectionRetryPolicy)
&& this.pingInterval == that.pingInterval
&& isEqual(this.protocol, that.protocol)
&& this.rpcTimeout == that.rpcTimeout
@@ -1339,10 +1388,9 @@ public class Client {
@Override
public int hashCode() {
- int result = 1;
+ int result = connectionRetryPolicy.hashCode();
result = PRIME * result + ((address == null) ? 0 : address.hashCode());
result = PRIME * result + maxIdleTime;
- result = PRIME * result + maxRetries;
result = PRIME * result + pingInterval;
result = PRIME * result + ((protocol == null) ? 0 : protocol.hashCode());
result = PRIME * rpcTimeout;
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/RPC.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/RPC.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/RPC.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/RPC.java Fri Jun 21 06:37:27 2013
@@ -18,31 +18,35 @@
package org.apache.hadoop.ipc;
-import java.lang.reflect.Proxy;
-import java.lang.reflect.Method;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.lang.reflect.Array;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
-
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
-import java.io.*;
-import java.util.Map;
import java.util.HashMap;
+import java.util.Map;
import javax.net.SocketFactory;
-import org.apache.commons.logging.*;
-
-import org.apache.hadoop.io.*;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.ObjectWritable;
+import org.apache.hadoop.io.UTF8;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.SaslRpcServer;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.SecretManager;
import org.apache.hadoop.security.token.TokenIdentifier;
-import org.apache.hadoop.conf.*;
-
-import org.apache.hadoop.net.NetUtils;
/** A simple RPC mechanism.
*
@@ -205,12 +209,12 @@ public class RPC {
private Client client;
private boolean isClosed = false;
- public Invoker(Class<? extends VersionedProtocol> protocol,
+ private Invoker(Class<? extends VersionedProtocol> protocol,
InetSocketAddress address, UserGroupInformation ticket,
Configuration conf, SocketFactory factory,
- int rpcTimeout) throws IOException {
+ int rpcTimeout, RetryPolicy connectionRetryPolicy) throws IOException {
this.remoteId = Client.ConnectionId.getConnectionId(address, protocol,
- ticket, rpcTimeout, conf);
+ ticket, rpcTimeout, connectionRetryPolicy, conf);
this.client = CLIENTS.getClient(conf, factory);
}
@@ -385,19 +389,39 @@ public class RPC {
Class<? extends VersionedProtocol> protocol,
long clientVersion, InetSocketAddress addr, UserGroupInformation ticket,
Configuration conf, SocketFactory factory, int rpcTimeout) throws IOException {
+ return getProxy(protocol, clientVersion, addr, ticket, conf, factory,
+ rpcTimeout, null, true);
+ }
+
+ /** Construct a client-side proxy object that implements the named protocol,
+ * talking to a server at the named address. */
+ public static VersionedProtocol getProxy(
+ Class<? extends VersionedProtocol> protocol,
+ long clientVersion, InetSocketAddress addr, UserGroupInformation ticket,
+ Configuration conf, SocketFactory factory, int rpcTimeout,
+ RetryPolicy connectionRetryPolicy,
+ boolean checkVersion) throws IOException {
if (UserGroupInformation.isSecurityEnabled()) {
SaslRpcServer.init(conf);
}
- VersionedProtocol proxy =
- (VersionedProtocol) Proxy.newProxyInstance(
- protocol.getClassLoader(), new Class[] { protocol },
- new Invoker(protocol, addr, ticket, conf, factory, rpcTimeout));
+ final Invoker invoker = new Invoker(protocol, addr, ticket, conf, factory,
+ rpcTimeout, connectionRetryPolicy);
+ VersionedProtocol proxy = (VersionedProtocol)Proxy.newProxyInstance(
+ protocol.getClassLoader(), new Class[]{protocol}, invoker);
+
+ if (checkVersion) {
+ checkVersion(protocol, clientVersion, proxy);
+ }
+ return proxy;
+ }
+
+ /** Get server version and then compare it with client version. */
+ public static void checkVersion(Class<? extends VersionedProtocol> protocol,
+ long clientVersion, VersionedProtocol proxy) throws IOException {
long serverVersion = proxy.getProtocolVersion(protocol.getName(),
clientVersion);
- if (serverVersion == clientVersion) {
- return proxy;
- } else {
+ if (serverVersion != clientVersion) {
throw new VersionMismatch(protocol.getName(), clientVersion,
serverVersion);
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Server.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Server.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Server.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/ipc/Server.java Fri Jun 21 06:37:27 2013
@@ -43,11 +43,13 @@ import java.nio.channels.WritableByteCha
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
@@ -95,7 +97,44 @@ import org.apache.hadoop.util.StringUtil
public abstract class Server {
private final boolean authorize;
private boolean isSecurityEnabled;
-
+ private ExceptionsHandler exceptionsHandler = new ExceptionsHandler();
+
+ public void addTerseExceptions(Class<?>... exceptionClass) {
+ exceptionsHandler.addTerseExceptions(exceptionClass);
+ }
+
+ /**
+ * ExceptionsHandler manages Exception groups for special handling e.g., terse
+ * exception group for concise logging messages
+ */
+ static class ExceptionsHandler {
+ private volatile Set<String> terseExceptions = new HashSet<String>();
+
+ /**
+ * Add exception class so server won't log its stack trace. Modifying the
+ * terseException through this method is thread safe.
+ *
+ * @param exceptionClass
+ * exception classes
+ */
+ void addTerseExceptions(Class<?>... exceptionClass) {
+
+ // Make a copy of terseException for performing modification
+ final HashSet<String> newSet = new HashSet<String>(terseExceptions);
+
+ // Add all class names into the HashSet
+ for (Class<?> name : exceptionClass) {
+ newSet.add(name.toString());
+ }
+ // Replace terseException set
+ terseExceptions = Collections.unmodifiableSet(newSet);
+ }
+
+ boolean isTerse(Class<?> t) {
+ return terseExceptions.contains(t.toString());
+ }
+ }
+
/**
* The first four bytes of Hadoop RPC connections
*/
@@ -290,7 +329,6 @@ public abstract class Server {
private long cleanupInterval = 10000; //the minimum interval between
//two cleanup runs
private int backlogLength = conf.getInt("ipc.server.listen.queue.size", 128);
- private ExecutorService readPool;
public Listener() throws IOException {
address = new InetSocketAddress(bindAddress, port);
@@ -304,12 +342,12 @@ public abstract class Server {
// create a selector;
selector= Selector.open();
readers = new Reader[readThreads];
- readPool = Executors.newFixedThreadPool(readThreads);
for (int i = 0; i < readThreads; i++) {
Selector readSelector = Selector.open();
- Reader reader = new Reader(readSelector);
+ Reader reader = new Reader("Socket Reader #" + (i + 1) + " for port " + port,
+ readSelector);
readers[i] = reader;
- readPool.execute(reader);
+ reader.start();
}
// Register accepts on the server socket with the selector.
@@ -318,15 +356,16 @@ public abstract class Server {
this.setDaemon(true);
}
- private class Reader implements Runnable {
+ private class Reader extends Thread {
private volatile boolean adding = false;
private Selector readSelector = null;
- Reader(Selector readSelector) {
+ Reader(String name, Selector readSelector) {
+ super(name);
this.readSelector = readSelector;
}
public void run() {
- LOG.info("Starting SocketReader");
+ LOG.info("Starting " + getName());
synchronized (this) {
while (running) {
SelectionKey key = null;
@@ -380,6 +419,16 @@ public abstract class Server {
adding = false;
this.notify();
}
+
+ void shutdown() {
+ assert !running;
+ readSelector.wakeup();
+ try {
+ join();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ }
}
/** cleanup connections from connectionList. Choose a random range
@@ -568,7 +617,9 @@ public abstract class Server {
LOG.info(getName() + ":Exception in closing listener socket. " + e);
}
}
- readPool.shutdown();
+ for (Reader r : readers) {
+ r.shutdown();
+ }
}
// The method that will return the next reader to work with
@@ -1398,7 +1449,19 @@ public abstract class Server {
);
}
} catch (Throwable e) {
- LOG.info(getName()+", call "+call+": error: " + e, e);
+ String logMsg = getName() + ", call " + call + ": error: " + e;
+ if (e instanceof RuntimeException || e instanceof Error) {
+ // These exception types indicate something is probably wrong
+ // on the server side, as opposed to just a normal exceptional
+ // result.
+ LOG.warn(logMsg, e);
+ } else if (exceptionsHandler.isTerse(e.getClass())) {
+ // Don't log the whole stack trace of these exceptions.
+ // Way too noisy!
+ LOG.info(logMsg);
+ } else {
+ LOG.info(logMsg, e);
+ }
errorClass = e.getClass().getName();
error = StringUtils.stringifyException(e);
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/jmx/JMXJsonServlet.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/jmx/JMXJsonServlet.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/jmx/JMXJsonServlet.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/jmx/JMXJsonServlet.java Fri Jun 21 06:37:27 2013
@@ -134,8 +134,8 @@ public class JMXJsonServlet extends Http
public void doGet(HttpServletRequest request, HttpServletResponse response) {
try {
// Do the authorization
- if (!HttpServer.hasAdministratorAccess(getServletContext(), request,
- response)) {
+ if (!HttpServer.isInstrumentationAccessAllowed(getServletContext(),
+ request, response)) {
return;
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/metrics/MetricsServlet.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/metrics/MetricsServlet.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/metrics/MetricsServlet.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/metrics/MetricsServlet.java Fri Jun 21 06:37:27 2013
@@ -109,8 +109,8 @@ public class MetricsServlet extends Http
throws ServletException, IOException {
// Do the authorization
- if (!HttpServer.hasAdministratorAccess(getServletContext(), request,
- response)) {
+ if (!HttpServer.isInstrumentationAccessAllowed(getServletContext(),
+ request, response)) {
return;
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/DNS.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/DNS.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/DNS.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/DNS.java Fri Jun 21 06:37:27 2013
@@ -22,7 +22,9 @@ import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.net.UnknownHostException;
+import java.util.Collections;
import java.util.Enumeration;
+import java.util.LinkedHashSet;
import java.util.Vector;
import javax.naming.NamingException;
@@ -30,6 +32,9 @@ import javax.naming.directory.Attributes
import javax.naming.directory.DirContext;
import javax.naming.directory.InitialDirContext;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
/**
*
* A class that provides direct and reverse lookup functionalities, allowing
@@ -39,6 +44,8 @@ import javax.naming.directory.InitialDir
*/
public class DNS {
+ public static final Log LOG = LogFactory.getLog(DNS.class);
+
/**
* Returns the hostname associated with the specified IP address by the
* provided nameserver.
@@ -69,7 +76,60 @@ public class DNS {
"/" + reverseIP, new String[] { "PTR" });
ictx.close();
- return attribute.get("PTR").get().toString();
+ String hostname = attribute.get("PTR").get().toString();
+ int hostnameLength = hostname.length();
+ if (hostname.charAt(hostnameLength - 1) == '.') {
+ hostname = hostname.substring(0, hostnameLength - 1);
+ }
+ return hostname;
+ }
+
+ /**
+ * @return NetworkInterface for the given subinterface name (eg eth0:0)
+ * or null if no interface with the given name can be found
+ */
+ private static NetworkInterface getSubinterface(String strInterface)
+ throws SocketException {
+ Enumeration<NetworkInterface> nifs =
+ NetworkInterface.getNetworkInterfaces();
+
+ while (nifs.hasMoreElements()) {
+ Enumeration<NetworkInterface> subNifs =
+ nifs.nextElement().getSubInterfaces();
+
+ while (subNifs.hasMoreElements()) {
+ NetworkInterface nif = subNifs.nextElement();
+ if (nif.getName().equals(strInterface)) {
+ return nif;
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * @param nif network interface to get addresses for
+ * @return set containing addresses for each subinterface of nif,
+ * see below for the rationale for using an ordered set
+ */
+ private static LinkedHashSet<InetAddress> getSubinterfaceInetAddrs(
+ NetworkInterface nif) {
+ LinkedHashSet<InetAddress> addrs = new LinkedHashSet<InetAddress>();
+ Enumeration<NetworkInterface> subNifs = nif.getSubInterfaces();
+ while (subNifs.hasMoreElements()) {
+ NetworkInterface subNif = subNifs.nextElement();
+ addrs.addAll(Collections.list(subNif.getInetAddresses()));
+ }
+ return addrs;
+ }
+
+ /**
+ * Like {@link DNS#getIPs(String, boolean)}, but returns all
+ * IPs associated with the given interface and its subinterfaces.
+ */
+ public static String[] getIPs(String strInterface)
+ throws UnknownHostException {
+ return getIPs(strInterface, true);
}
/**
@@ -77,31 +137,56 @@ public class DNS {
* textual form.
*
* @param strInterface
- * The name of the network interface to query (e.g. eth0)
+ * The name of the network interface or subinterface to query
+ * (eg eth0 or eth0:0) or the string "default"
+ * @param returnSubinterfaces
+ * Whether to return IPs associated with subinterfaces of
+ * the given interface
* @return A string vector of all the IPs associated with the provided
* interface
* @throws UnknownHostException
* If an UnknownHostException is encountered in querying the
- * default interface
+ * default interface or the given interface can not be found
*
*/
- public static String[] getIPs(String strInterface)
- throws UnknownHostException {
+ public static String[] getIPs(String strInterface,
+ boolean returnSubinterfaces) throws UnknownHostException {
+ if ("default".equals(strInterface)) {
+ return new String[] {
+ InetAddress.getLocalHost().getHostAddress()
+ };
+ }
+ NetworkInterface netIf;
try {
- NetworkInterface netIF = NetworkInterface.getByName(strInterface);
- if (netIF == null)
- return new String[] { InetAddress.getLocalHost()
- .getHostAddress() };
- else {
- Vector<String> ips = new Vector<String>();
- Enumeration e = netIF.getInetAddresses();
- while (e.hasMoreElements())
- ips.add(((InetAddress) e.nextElement()).getHostAddress());
- return ips.toArray(new String[] {});
+ netIf = NetworkInterface.getByName(strInterface);
+ if (netIf == null) {
+ netIf = getSubinterface(strInterface);
+ if (netIf == null) {
+ throw new UnknownHostException("Unknown interface " + strInterface);
+ }
}
} catch (SocketException e) {
- return new String[] { InetAddress.getLocalHost().getHostAddress() };
+ LOG.warn("Unable to get IP for interface " + strInterface, e);
+ return new String[] {
+ InetAddress.getLocalHost().getHostAddress()
+ };
+ }
+
+ // NB: Using a LinkedHashSet to preserve the order for callers
+ // that depend on a particular element being 1st in the array.
+ // For example, getDefaultIP always returns the first element.
+ LinkedHashSet<InetAddress> allAddrs = new LinkedHashSet<InetAddress>();
+ allAddrs.addAll(Collections.list(netIf.getInetAddresses()));
+ if (!returnSubinterfaces) {
+ allAddrs.removeAll(getSubinterfaceInetAddrs(netIf));
}
+
+ String ips[] = new String[allAddrs.size()];
+ int i = 0;
+ for (InetAddress addr : allAddrs) {
+ ips[i++] = addr.getHostAddress();
+ }
+ return ips;
}
/**
@@ -109,7 +194,8 @@ public class DNS {
* network interface
*
* @param strInterface
- * The name of the network interface to query (e.g. eth0)
+ * The name of the network interface or subinterface to query
+ * (e.g. eth0 or eth0:0) or the string "default"
* @return The IP address in text form
* @throws UnknownHostException
* If one is encountered in querying the default interface
@@ -125,7 +211,8 @@ public class DNS {
* address bound to the specified network interface
*
* @param strInterface
- * The name of the network interface to query (e.g. eth0)
+ * The name of the network interface or subinterface to query
+ * (e.g. eth0 or eth0:0) or the string "default"
* @param nameserver
* The DNS host name
* @return A string vector of all host names associated with the IPs tied to
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetUtils.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetUtils.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetUtils.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetUtils.java Fri Jun 21 06:37:27 2013
@@ -39,6 +39,8 @@ import javax.net.SocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.commons.net.util.SubnetUtils;
+import org.apache.commons.net.util.SubnetUtils.SubnetInfo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.ipc.VersionedProtocol;
@@ -455,7 +457,7 @@ public class NetUtils {
return (socket.getChannel() == null) ?
socket.getOutputStream() : new SocketOutputStream(socket, timeout);
}
-
+
/**
* This is a drop-in replacement for
* {@link Socket#connect(SocketAddress, int)}.
@@ -470,11 +472,27 @@ public class NetUtils {
* @see java.net.Socket#connect(java.net.SocketAddress, int)
*
* @param socket
- * @param endpoint
- * @param timeout - timeout in milliseconds
+ * @param address the remote address
+ * @param timeout timeout in milliseconds
+ */
+ public static void connect(Socket socket,
+ SocketAddress address,
+ int timeout) throws IOException {
+ connect(socket, address, null, timeout);
+ }
+
+ /**
+ * Like {@link NetUtils#connect(Socket, SocketAddress, int)} but
+ * also takes a local address and port to bind the socket to.
+ *
+ * @param socket
+ * @param endpoint the remote address
+ * @param localAddr the local address to bind the socket to
+ * @param timeout timeout in milliseconds
*/
public static void connect(Socket socket,
- SocketAddress endpoint,
+ SocketAddress endpoint,
+ SocketAddress localAddr,
int timeout) throws IOException {
if (socket == null || endpoint == null || timeout < 0) {
throw new IllegalArgumentException("Illegal argument for connect()");
@@ -482,6 +500,10 @@ public class NetUtils {
SocketChannel ch = socket.getChannel();
+ if (localAddr != null) {
+ socket.bind(localAddr);
+ }
+
if (ch == null) {
// let the default implementation handle it.
socket.connect(endpoint, timeout);
@@ -589,4 +611,70 @@ public class NetUtils {
} catch (UnknownHostException ignore) { }
return addr;
}
+
+ /**
+ * @return true if the given string is a subnet specified
+ * using CIDR notation, false otherwise
+ */
+ public static boolean isValidSubnet(String subnet) {
+ try {
+ new SubnetUtils(subnet);
+ return true;
+ } catch (IllegalArgumentException iae) {
+ return false;
+ }
+ }
+
+ /**
+ * Add all addresses associated with the given nif in the
+ * given subnet to the given list.
+ */
+ private static void addMatchingAddrs(NetworkInterface nif,
+ SubnetInfo subnetInfo, List<InetAddress> addrs) {
+ Enumeration<InetAddress> ifAddrs = nif.getInetAddresses();
+ while (ifAddrs.hasMoreElements()) {
+ InetAddress ifAddr = ifAddrs.nextElement();
+ if (subnetInfo.isInRange(ifAddr.getHostAddress())) {
+ addrs.add(ifAddr);
+ }
+ }
+ }
+
+ /**
+ * Return an InetAddress for each interface that matches the
+ * given subnet specified using CIDR notation.
+ *
+ * @param subnet subnet specified using CIDR notation
+ * @param returnSubinterfaces
+ * whether to return IPs associated with subinterfaces
+ * @throws IllegalArgumentException if subnet is invalid
+ */
+ public static List<InetAddress> getIPs(String subnet,
+ boolean returnSubinterfaces) {
+ List<InetAddress> addrs = new ArrayList<InetAddress>();
+ SubnetInfo subnetInfo = new SubnetUtils(subnet).getInfo();
+ Enumeration<NetworkInterface> nifs;
+
+ try {
+ nifs = NetworkInterface.getNetworkInterfaces();
+ } catch (SocketException e) {
+ LOG.error("Unable to get host interfaces", e);
+ return addrs;
+ }
+
+ while (nifs.hasMoreElements()) {
+ NetworkInterface nif = nifs.nextElement();
+ // NB: adding addresses even if the nif is not up
+ addMatchingAddrs(nif, subnetInfo, addrs);
+
+ if (!returnSubinterfaces) {
+ continue;
+ }
+ Enumeration<NetworkInterface> subNifs = nif.getSubInterfaces();
+ while (subNifs.hasMoreElements()) {
+ addMatchingAddrs(subNifs.nextElement(), subnetInfo, addrs);
+ }
+ }
+ return addrs;
+ }
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopology.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopology.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopology.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopology.java Fri Jun 21 06:37:27 2013
@@ -46,6 +46,13 @@ public class NetworkTopology {
public static final Log LOG =
LogFactory.getLog(NetworkTopology.class);
+ public static class InvalidTopologyException extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+ public InvalidTopologyException(String msg) {
+ super(msg);
+ }
+ }
+
/* Inner Node represent a switch/router of a data center or rack.
* Different from a leave node, it has non-null children.
*/
@@ -330,6 +337,8 @@ public class NetworkTopology {
} // end of InnerNode
InnerNode clusterMap;
+ /** Depth of all leaf nodes */
+ private int depthOfAllLeaves = -1;
protected int numOfRacks = 0; // rack counter
protected ReadWriteLock netlock = new ReentrantReadWriteLock();
@@ -359,6 +368,7 @@ public class NetworkTopology {
*/
public void add(Node node) {
if (node==null) return;
+ String oldTopoStr = this.toString();
if( node instanceof InnerNode ) {
throw new IllegalArgumentException(
"Not allow to add an inner node: "+NodeBase.getPath(node));
@@ -376,6 +386,19 @@ public class NetworkTopology {
if (rack == null) {
numOfRacks++;
}
+ if (!(node instanceof InnerNode)) {
+ if (depthOfAllLeaves == -1) {
+ depthOfAllLeaves = node.getLevel();
+ } else {
+ if (depthOfAllLeaves != node.getLevel()) {
+ LOG.error("Error: can't add leaf node at depth " +
+ node.getLevel() + " to topology:\n" + oldTopoStr);
+ throw new InvalidTopologyException("Invalid network topology. " +
+ "You cannot have a rack and a non-rack node at the same " +
+ "level of the network topology.");
+ }
+ }
+ }
}
LOG.debug("NetworkTopology became:\n" + this.toString());
} finally {
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/net/NetworkTopologyWithNodeGroup.java Fri Jun 21 06:37:27 2013
@@ -190,7 +190,12 @@ public class NetworkTopologyWithNodeGrou
}
rack = getNode(nodeGroup.getNetworkLocation());
- if (rack != null && !(rack instanceof InnerNode)) {
+ // rack should be an innerNode and with parent.
+ // note: rack's null parent case is: node's topology only has one layer,
+ // so rack is recognized as "/" and no parent.
+ // This will be recognized as a node with fault topology.
+ if (rack != null &&
+ (!(rack instanceof InnerNode)|| rack.getParent() == null)) {
throw new IllegalArgumentException("Unexpected data node "
+ node.toString()
+ " at an illegal network location");
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/AuthenticationFilterInitializer.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/AuthenticationFilterInitializer.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/AuthenticationFilterInitializer.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/AuthenticationFilterInitializer.java Fri Jun 21 06:37:27 2013
@@ -30,7 +30,7 @@ import java.util.Map;
/**
* Initializes Alfredo AuthenticationFilter which provides support for
- * Kerberos HTTP SPENGO authentication.
+ * Kerberos HTTP SPNEGO authentication.
* <p/>
* It enables anonymous access, simple/speudo and Kerberos HTTP SPNEGO
* authentication for Hadoop JobTracker, NameNode, DataNodes and
Added: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsMappingWithFallback.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsMappingWithFallback.java?rev=1495297&view=auto
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsMappingWithFallback.java (added)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsMappingWithFallback.java Fri Jun 21 06:37:27 2013
@@ -0,0 +1,63 @@
+/**
+ * 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.security;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.NativeCodeLoader;
+
+public class JniBasedUnixGroupsMappingWithFallback implements
+ GroupMappingServiceProvider {
+
+ private static final Log LOG = LogFactory
+ .getLog(JniBasedUnixGroupsMappingWithFallback.class);
+
+ private GroupMappingServiceProvider impl;
+
+ public JniBasedUnixGroupsMappingWithFallback() {
+ if (NativeCodeLoader.isNativeCodeLoaded()) {
+ this.impl = new JniBasedUnixGroupsMapping();
+ } else {
+ LOG.info("Falling back to shell based");
+ this.impl = new ShellBasedUnixGroupsMapping();
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Group mapping impl=" + impl.getClass().getName());
+ }
+ }
+
+ @Override
+ public List<String> getGroups(String user) throws IOException {
+ return impl.getGroups(user);
+ }
+
+ @Override
+ public void cacheGroupsRefresh() throws IOException {
+ impl.cacheGroupsRefresh();
+ }
+
+ @Override
+ public void cacheGroupsAdd(List<String> groups) throws IOException {
+ impl.cacheGroupsAdd(groups);
+ }
+
+}
Added: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsNetgroupMappingWithFallback.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsNetgroupMappingWithFallback.java?rev=1495297&view=auto
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsNetgroupMappingWithFallback.java (added)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/JniBasedUnixGroupsNetgroupMappingWithFallback.java Fri Jun 21 06:37:27 2013
@@ -0,0 +1,63 @@
+/**
+ * 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.security;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.NativeCodeLoader;
+
+public class JniBasedUnixGroupsNetgroupMappingWithFallback implements
+ GroupMappingServiceProvider {
+
+ private static final Log LOG = LogFactory
+ .getLog(JniBasedUnixGroupsNetgroupMappingWithFallback.class);
+
+ private GroupMappingServiceProvider impl;
+
+ public JniBasedUnixGroupsNetgroupMappingWithFallback() {
+ if (NativeCodeLoader.isNativeCodeLoaded()) {
+ this.impl = new JniBasedUnixGroupsNetgroupMapping();
+ } else {
+ LOG.info("Falling back to shell based");
+ this.impl = new ShellBasedUnixGroupsNetgroupMapping();
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Group mapping impl=" + impl.getClass().getName());
+ }
+ }
+
+ @Override
+ public List<String> getGroups(String user) throws IOException {
+ return impl.getGroups(user);
+ }
+
+ @Override
+ public void cacheGroupsRefresh() throws IOException {
+ impl.cacheGroupsRefresh();
+ }
+
+ @Override
+ public void cacheGroupsAdd(List<String> groups) throws IOException {
+ impl.cacheGroupsAdd(groups);
+ }
+
+}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/SecurityUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/SecurityUtil.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/SecurityUtil.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/SecurityUtil.java Fri Jun 21 06:37:27 2013
@@ -19,16 +19,17 @@ package org.apache.hadoop.security;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URL;
+import java.net.URLConnection;
import java.net.UnknownHostException;
import java.security.AccessController;
import java.util.Arrays;
import java.util.List;
+import java.util.Locale;
import java.util.Set;
import javax.security.auth.Subject;
@@ -40,6 +41,9 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
+import org.apache.hadoop.security.authentication.client.AuthenticationException;
+import org.apache.hadoop.security.authentication.util.KerberosUtil;
import org.apache.hadoop.security.authorize.AccessControlList;
import org.apache.hadoop.security.token.Token;
@@ -55,12 +59,19 @@ public class SecurityUtil {
// by the user; visible for testing
static boolean useIpForTokenService;
static HostResolver hostResolver;
+
+ private static final boolean useKsslAuth;
static {
- boolean useIp = new Configuration().getBoolean(
+ Configuration conf = new Configuration();
+ boolean useIp = conf.getBoolean(
CommonConfigurationKeys.HADOOP_SECURITY_TOKEN_SERVICE_USE_IP,
CommonConfigurationKeys.HADOOP_SECURITY_TOKEN_SERVICE_USE_IP_DEFAULT);
setTokenServiceUseIp(useIp);
+
+ useKsslAuth = conf.getBoolean(
+ CommonConfigurationKeys.HADOOP_SECURITY_USE_WEAK_HTTP_CRYPTO_KEY,
+ CommonConfigurationKeys.HADOOP_SECURITY_USE_WEAK_HTTP_CRYPTO_DEFAULT);
}
/**
@@ -126,7 +137,8 @@ public class SecurityUtil {
if(!UserGroupInformation.isSecurityEnabled())
return;
- String serviceName = "host/" + remoteHost.getHost();
+ String serviceName = KerberosUtil.getServicePrincipal("host",
+ remoteHost.getHost());
if (LOG.isDebugEnabled())
LOG.debug("Fetching service ticket for host at: " + serviceName);
Object serviceCred = null;
@@ -245,10 +257,17 @@ public class SecurityUtil {
if (fqdn == null || fqdn.equals("") || fqdn.equals("0.0.0.0")) {
fqdn = getLocalHostName();
}
- return components[0] + "/" + fqdn.toLowerCase() + "@" + components[2];
+ return components[0] + "/" + fqdn.toLowerCase(Locale.US) + "@" + components[2];
}
- static String getLocalHostName() throws UnknownHostException {
+ /**
+ * Get the fqdn for the current host.
+ *
+ * @return fqdn of the current host.
+ * @throws UnknownHostException
+ * if no IP address for the local host could be found.
+ */
+ public static String getLocalHostName() throws UnknownHostException {
return InetAddress.getLocalHost().getCanonicalHostName();
}
@@ -381,8 +400,45 @@ public class SecurityUtil {
public static String getHostFromPrincipal(String principalName) {
return new KerberosName(principalName).getHostName();
}
+
+ /**
+ * @return true if we should use KSSL to authenticate NN HTTP endpoints,
+ * false to use SPNEGO or if security is disabled.
+ */
+ public static boolean useKsslAuth() {
+ return UserGroupInformation.isSecurityEnabled() && useKsslAuth;
+ }
/**
+ * Open a (if need be) secure connection to a URL in a secure environment
+ * that is using SPNEGO or KSSL to authenticate its URLs. All Namenode and
+ * Secondary Namenode URLs that are protected via SPNEGO or KSSL should be
+ * accessed via this method.
+ *
+ * @param url to authenticate via SPNEGO.
+ * @return A connection that has been authenticated via SPNEGO
+ * @throws IOException If unable to authenticate via SPNEGO
+ */
+ public static URLConnection openSecureHttpConnection(URL url)
+ throws IOException {
+ if (useKsslAuth) {
+ // Avoid Krb bug with cross-realm hosts
+ fetchServiceTicket(url);
+ }
+ if (!UserGroupInformation.isSecurityEnabled() || useKsslAuth) {
+ return url.openConnection();
+ } else {
+ AuthenticatedURL.Token token = new AuthenticatedURL.Token();
+ try {
+ return new AuthenticatedURL().openConnection(url, token);
+ } catch (AuthenticationException e) {
+ throw new IOException("Exception trying to open authenticated connection to "
+ + url, e);
+ }
+ }
+ }
+
+ /**
* Resolves a host subject to the security requirements determined by
* hadoop.security.token.service.use_ip.
*
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/UserGroupInformation.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/UserGroupInformation.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/UserGroupInformation.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/UserGroupInformation.java Fri Jun 21 06:37:27 2013
@@ -69,6 +69,7 @@ public class UserGroupInformation {
*/
private static final float TICKET_RENEW_WINDOW = 0.80f;
static final String HADOOP_USER_NAME = "HADOOP_USER_NAME";
+ static final String HADOOP_PROXY_USER = "HADOOP_PROXY_USER";
/**
* A login module that looks at the Kerberos, Unix, or Windows principal and
@@ -252,16 +253,30 @@ public class UserGroupInformation {
private static String OS_LOGIN_MODULE_NAME;
private static Class<? extends Principal> OS_PRINCIPAL_CLASS;
- private static final boolean windows =
- System.getProperty("os.name").startsWith("Windows");
+
+ private static final boolean windows =
+ System.getProperty("os.name").startsWith("Windows");
+ private static final boolean is64Bit =
+ System.getProperty("os.arch").contains("64");
+
+ private static final boolean ibmJava = System.getProperty("java.vendor").contains("IBM");
+ private static final boolean aix = System.getProperty("os.name").equals("AIX");
+
private static Thread renewerThread = null;
private static volatile boolean shouldRunRenewerThread = true;
/* Return the OS login module class name */
private static String getOSLoginModuleName() {
- if (System.getProperty("java.vendor").contains("IBM")) {
- return windows ? "com.ibm.security.auth.module.NTLoginModule"
- : "com.ibm.security.auth.module.LinuxLoginModule";
+ if (ibmJava) {
+ if (windows) {
+ return is64Bit ? "com.ibm.security.auth.module.Win64LoginModule"
+ : "com.ibm.security.auth.module.NTLoginModule";
+ } else if (aix) {
+ return is64Bit ? "com.ibm.security.auth.module.AIX64LoginModule"
+ : "com.ibm.security.auth.module.AIXLoginModule";
+ } else {
+ return "com.ibm.security.auth.module.LinuxLoginModule";
+ }
} else {
return windows ? "com.sun.security.auth.module.NTLoginModule"
: "com.sun.security.auth.module.UnixLoginModule";
@@ -273,21 +288,24 @@ public class UserGroupInformation {
private static Class<? extends Principal> getOsPrincipalClass() {
ClassLoader cl = ClassLoader.getSystemClassLoader();
try {
- if (System.getProperty("java.vendor").contains("IBM")) {
- if (windows) {
- return (Class<? extends Principal>)
- cl.loadClass("com.ibm.security.auth.UsernamePrincipal");
+ String principalClass = null;
+ if (ibmJava) {
+ if (is64Bit) {
+ principalClass = "com.ibm.security.auth.UsernamePrincipal";
} else {
- return (Class<? extends Principal>)
- (System.getProperty("os.arch").contains("64")
- ? cl.loadClass("com.ibm.security.auth.UsernamePrincipal")
- : cl.loadClass("com.ibm.security.auth.LinuxPrincipal"));
+ if (windows) {
+ principalClass = "com.ibm.security.auth.NTUserPrincipal";
+ } else if (aix) {
+ principalClass = "com.ibm.security.auth.AIXPrincipal";
+ } else {
+ principalClass = "com.ibm.security.auth.LinuxPrincipal";
+ }
}
} else {
- return (Class<? extends Principal>) (windows
- ? cl.loadClass("com.sun.security.auth.NTUserPrincipal")
- : cl.loadClass("com.sun.security.auth.UnixPrincipal"));
+ principalClass = windows ? "com.sun.security.auth.NTUserPrincipal"
+ : "com.sun.security.auth.UnixPrincipal";
}
+ return (Class<? extends Principal>) cl.loadClass(principalClass);
} catch (ClassNotFoundException e) {
LOG.error("Unable to find JAAS classes:" + e.getMessage());
}
@@ -358,12 +376,21 @@ public class UserGroupInformation {
private static final Map<String,String> USER_KERBEROS_OPTIONS =
new HashMap<String,String>();
static {
- USER_KERBEROS_OPTIONS.put("doNotPrompt", "true");
- USER_KERBEROS_OPTIONS.put("useTicketCache", "true");
- USER_KERBEROS_OPTIONS.put("renewTGT", "true");
+ if (ibmJava) {
+ USER_KERBEROS_OPTIONS.put("useDefaultCcache", "true");
+ } else {
+ USER_KERBEROS_OPTIONS.put("doNotPrompt", "true");
+ USER_KERBEROS_OPTIONS.put("useTicketCache", "true");
+ USER_KERBEROS_OPTIONS.put("renewTGT", "true");
+ }
String ticketCache = System.getenv("KRB5CCNAME");
if (ticketCache != null) {
- USER_KERBEROS_OPTIONS.put("ticketCache", ticketCache);
+ if (ibmJava) {
+ // The first value searched when "useDefaultCcache" is used.
+ System.setProperty("KRB5CCNAME", ticketCache);
+ } else {
+ USER_KERBEROS_OPTIONS.put("ticketCache", ticketCache);
+ }
}
}
private static final AppConfigurationEntry USER_KERBEROS_LOGIN =
@@ -373,9 +400,14 @@ public class UserGroupInformation {
private static final Map<String,String> KEYTAB_KERBEROS_OPTIONS =
new HashMap<String,String>();
static {
- KEYTAB_KERBEROS_OPTIONS.put("doNotPrompt", "true");
- KEYTAB_KERBEROS_OPTIONS.put("useKeyTab", "true");
- KEYTAB_KERBEROS_OPTIONS.put("storeKey", "true");
+ if (ibmJava) {
+ KEYTAB_KERBEROS_OPTIONS.put("credsType", "both");
+ } else {
+ KEYTAB_KERBEROS_OPTIONS.put("doNotPrompt", "true");
+ KEYTAB_KERBEROS_OPTIONS.put("useKeyTab", "true");
+ KEYTAB_KERBEROS_OPTIONS.put("storeKey", "true");
+ KEYTAB_KERBEROS_OPTIONS.put("refreshKrb5Config", "true");
+ }
}
private static final AppConfigurationEntry KEYTAB_KERBEROS_LOGIN =
new AppConfigurationEntry(KerberosUtil.getKrb5LoginModuleName(),
@@ -399,7 +431,12 @@ public class UserGroupInformation {
} else if (USER_KERBEROS_CONFIG_NAME.equals(appName)) {
return USER_KERBEROS_CONF;
} else if (KEYTAB_KERBEROS_CONFIG_NAME.equals(appName)) {
- KEYTAB_KERBEROS_OPTIONS.put("keyTab", keytabFile);
+ if (ibmJava) {
+ KEYTAB_KERBEROS_OPTIONS.put("useKeytab",
+ prependFileAuthority(keytabFile));
+ } else {
+ KEYTAB_KERBEROS_OPTIONS.put("keyTab", keytabFile);
+ }
KEYTAB_KERBEROS_OPTIONS.put("principal", keytabPrincipal);
return KEYTAB_KERBEROS_CONF;
}
@@ -430,6 +467,11 @@ public class UserGroupInformation {
user.setLogin(login);
}
+ private static String prependFileAuthority(String keytabPath) {
+ return keytabPath.startsWith("file://") ? keytabPath
+ : "file://" + keytabPath;
+ }
+
/**
* Create a UserGroupInformation for the given subject.
* This does not change the subject or acquire new credentials.
@@ -459,7 +501,11 @@ public class UserGroupInformation {
static UserGroupInformation getCurrentUser() throws IOException {
AccessControlContext context = AccessController.getContext();
Subject subject = Subject.getSubject(context);
- return subject == null ? getLoginUser() : new UserGroupInformation(subject);
+ if (subject == null || subject.getPrincipals(User.class).isEmpty()) {
+ return getLoginUser();
+ } else {
+ return new UserGroupInformation(subject);
+ }
}
/**
@@ -479,12 +525,20 @@ public class UserGroupInformation {
login = newLoginContext(HadoopConfiguration.SIMPLE_CONFIG_NAME, subject);
}
login.login();
- loginUser = new UserGroupInformation(subject);
- loginUser.setLogin(login);
- loginUser.setAuthenticationMethod(isSecurityEnabled() ?
+ UserGroupInformation realUser = new UserGroupInformation(subject);
+ realUser.setLogin(login);
+ realUser.setAuthenticationMethod(isSecurityEnabled() ?
AuthenticationMethod.KERBEROS :
AuthenticationMethod.SIMPLE);
- loginUser = new UserGroupInformation(login.getSubject());
+ realUser = new UserGroupInformation(login.getSubject());
+ // If the HADOOP_PROXY_USER environment variable or property
+ // is specified, create a proxy user as the logged in user.
+ String proxyUser = System.getenv(HADOOP_PROXY_USER);
+ if (proxyUser == null) {
+ proxyUser = System.getProperty(HADOOP_PROXY_USER);
+ }
+ loginUser = proxyUser == null ? realUser : createProxyUser(proxyUser, realUser);
+
String fileLocation = System.getenv(HADOOP_TOKEN_FILE_LOCATION);
if (fileLocation != null && isSecurityEnabled()) {
// load the token storage file and put all of the tokens into the
@@ -497,6 +551,7 @@ public class UserGroupInformation {
}
loginUser.spawnAutoRenewalThreadForUserCreds();
} catch (LoginException le) {
+ LOG.debug("failure to login", le);
throw new IOException("failure to login", le);
}
if (LOG.isDebugEnabled()) {
@@ -1040,7 +1095,7 @@ public class UserGroupInformation {
List<String> result = groups.getGroups(getShortUserName());
return result.toArray(new String[result.size()]);
} catch (IOException ie) {
- LOG.warn("No groups available for user " + getShortUserName(), ie);
+ LOG.warn("No groups available for user " + getShortUserName());
return new String[0];
}
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/AuthenticatedURL.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/AuthenticatedURL.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/AuthenticatedURL.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/AuthenticatedURL.java Fri Jun 21 06:37:27 2013
@@ -266,6 +266,7 @@ public class AuthenticatedURL {
}
}
} else {
+ token.set(null);
throw new AuthenticationException("Authentication failed, status: " + conn.getResponseCode() +
", message: " + conn.getResponseMessage());
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java Fri Jun 21 06:37:27 2013
@@ -194,7 +194,8 @@ public class KerberosAuthenticator imple
GSSContext gssContext = null;
try {
GSSManager gssManager = GSSManager.getInstance();
- String servicePrincipal = "HTTP/" + KerberosAuthenticator.this.url.getHost();
+ String servicePrincipal = KerberosUtil.getServicePrincipal("HTTP",
+ KerberosAuthenticator.this.url.getHost());
Oid oid = KerberosUtil.getOidInstance("NT_GSS_KRB5_PRINCIPAL");
GSSName serviceName = gssManager.createName(servicePrincipal,
oid);
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/server/AuthenticationFilter.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/server/AuthenticationFilter.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/server/AuthenticationFilter.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/server/AuthenticationFilter.java Fri Jun 21 06:37:27 2013
@@ -306,8 +306,8 @@ public class AuthenticationFilter implem
if (!token.getType().equals(authHandler.getType())) {
throw new AuthenticationException("Invalid AuthenticationToken type");
}
- if (token.isExpired()) {
- throw new AuthenticationException("AuthenticationToken expired");
+ if (LOG.isDebugEnabled() && token.isExpired ()) {
+ LOG.debug("Authentication Token expired");
}
}
return token;
@@ -327,12 +327,21 @@ public class AuthenticationFilter implem
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain filterChain)
throws IOException, ServletException {
+ boolean unauthorizedResponse = true;
+ String unauthorizedMsg = "";
HttpServletRequest httpRequest = (HttpServletRequest) request;
HttpServletResponse httpResponse = (HttpServletResponse) response;
try {
boolean newToken = false;
- AuthenticationToken token = getToken(httpRequest);
- if (token == null) {
+ AuthenticationToken token;
+ try {
+ token = getToken(httpRequest);
+ }
+ catch (AuthenticationException ex) {
+ LOG.warn("AuthenticationToken ignored: " + ex.getMessage());
+ token = null;
+ }
+ if (token == null || token.isExpired ()) {
if (LOG.isDebugEnabled()) {
LOG.debug("Request [{}] triggering authentication", getRequestURL(httpRequest));
}
@@ -343,6 +352,7 @@ public class AuthenticationFilter implem
newToken = true;
}
if (token != null) {
+ unauthorizedResponse = false;
if (LOG.isDebugEnabled()) {
LOG.debug("Request [{}] user [{}] authenticated", getRequestURL(httpRequest), token.getUserName());
}
@@ -372,13 +382,16 @@ public class AuthenticationFilter implem
filterChain.doFilter(httpRequest, httpResponse);
}
} catch (AuthenticationException ex) {
+ unauthorizedMsg = ex.toString();
+ LOG.warn("Authentication exception: " + ex.getMessage(), ex);
+ }
+ if (unauthorizedResponse) {
if (!httpResponse.isCommitted()) {
Cookie cookie = createCookie("");
cookie.setMaxAge(0);
httpResponse.addCookie(cookie);
- httpResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED, ex.getMessage());
+ httpResponse.sendError(HttpServletResponse.SC_UNAUTHORIZED, unauthorizedMsg);
}
- LOG.warn("Authentication exception: " + ex.getMessage(), ex);
}
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/util/KerberosUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/util/KerberosUtil.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/util/KerberosUtil.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authentication/util/KerberosUtil.java Fri Jun 21 06:37:27 2013
@@ -20,7 +20,10 @@ package org.apache.hadoop.security.authe
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+import java.net.UnknownHostException;
+import java.util.Locale;
+import org.apache.hadoop.security.SecurityUtil;
import org.ietf.jgss.GSSException;
import org.ietf.jgss.Oid;
@@ -65,4 +68,26 @@ public class KerberosUtil {
new Class[0]);
return (String)getDefaultRealmMethod.invoke(kerbConf, new Object[0]);
}
+
+ /**
+ * Create Kerberos principal for a given service and hostname. It converts
+ * hostname to lower case. If hostname is null or "0.0.0.0", it uses
+ * dynamically looked-up fqdn of the current host instead.
+ *
+ * @param service
+ * Service for which you want to generate the principal.
+ * @param hostname
+ * Fully-qualified domain name.
+ * @return Converted Kerberos principal name.
+ * @throws UnknownHostException
+ * If no IP address for the local host could be found.
+ */
+ public static final String getServicePrincipal(String service, String hostname)
+ throws UnknownHostException {
+ String fqdn = hostname;
+ if (null == fqdn || fqdn.equals("") || fqdn.equals("0.0.0.0")) {
+ fqdn = SecurityUtil.getLocalHostName();
+ }
+ return service + "/" + fqdn.toLowerCase(Locale.US);
+ }
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authorize/ProxyUsers.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authorize/ProxyUsers.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authorize/ProxyUsers.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/authorize/ProxyUsers.java Fri Jun 21 06:37:27 2013
@@ -122,7 +122,9 @@ public class ProxyUsers {
Collection<String> allowedUserGroups = proxyGroups.get(
getProxySuperuserGroupConfKey(superUser.getShortUserName()));
- if (allowedUserGroups != null && !allowedUserGroups.isEmpty()) {
+ if (isWildcardList(allowedUserGroups)) {
+ groupAuthorized = true;
+ } else if (allowedUserGroups != null && !allowedUserGroups.isEmpty()) {
for (String group : user.getGroupNames()) {
if (allowedUserGroups.contains(group)) {
groupAuthorized = true;
@@ -138,8 +140,10 @@ public class ProxyUsers {
Collection<String> ipList = proxyHosts.get(
getProxySuperuserIpConfKey(superUser.getShortUserName()));
-
- if (ipList != null && !ipList.isEmpty()) {
+
+ if (isWildcardList(ipList)) {
+ ipAuthorized = true;
+ } else if (ipList != null && !ipList.isEmpty()) {
for (String allowedHost : ipList) {
InetAddress hostAddr;
try {
@@ -158,4 +162,15 @@ public class ProxyUsers {
+ superUser.getUserName() + " from IP " + remoteAddress);
}
}
+
+ /**
+ * Return true if the configuration specifies the special configuration value
+ * "*", indicating that any group or host list is allowed to use this configuration.
+ */
+ private static boolean isWildcardList(Collection<String> list) {
+ return (list != null) &&
+ (list.size() == 1) &&
+ (list.contains("*"));
+ }
+
}
Modified: hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/token/Token.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/token/Token.java?rev=1495297&r1=1495296&r2=1495297&view=diff
==============================================================================
--- hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/token/Token.java (original)
+++ hadoop/common/branches/branch-1-win/src/core/org/apache/hadoop/security/token/Token.java Fri Jun 21 06:37:27 2013
@@ -265,8 +265,6 @@ public class Token<T extends TokenIdenti
StringBuilder buffer = new StringBuilder();
buffer.append("Ident: ");
addBinaryBuffer(buffer, identifier);
- buffer.append(", Pass: ");
- addBinaryBuffer(buffer, password);
buffer.append(", Kind: ");
buffer.append(kind.toString());
buffer.append(", Service: ");