Repository: hadoop Updated Branches: refs/heads/branch-2 02627f8c5 -> b9f2effaf
HADOOP-14640. Azure: Support affinity for service running on localhost and reuse SPNEGO hadoop.auth cookie for authorization, SASKey and delegation token generation. Contributed by Santhosh G Nayak. Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/b9f2effa Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/b9f2effa Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/b9f2effa Branch: refs/heads/branch-2 Commit: b9f2effaf6782885fbade07436fcf26aad4713a4 Parents: 02627f8c Author: Jitendra Pandey <jiten...@apache.org> Authored: Mon Jul 17 02:27:55 2017 -0700 Committer: Jitendra Pandey <jiten...@apache.org> Committed: Mon Jul 17 16:25:05 2017 -0700 ---------------------------------------------------------------------- .../fs/azure/RemoteSASKeyGeneratorImpl.java | 8 +- .../fs/azure/RemoteWasbAuthorizerImpl.java | 8 +- .../fs/azure/SecureWasbRemoteCallHelper.java | 86 ++++++++++++-------- .../hadoop/fs/azure/WasbRemoteCallHelper.java | 61 +++++++++++--- .../hadoop/fs/azure/security/Constants.java | 19 +++-- .../RemoteWasbDelegationTokenManager.java | 27 +++--- .../hadoop/fs/azure/security/SpnegoToken.java | 49 +++++++++++ .../fs/azure/TestWasbRemoteCallHelper.java | 58 ++++++++++++- 8 files changed, 245 insertions(+), 71 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteSASKeyGeneratorImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteSASKeyGeneratorImpl.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteSASKeyGeneratorImpl.java index 2b55d92..cc27905 100644 --- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteSASKeyGeneratorImpl.java +++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteSASKeyGeneratorImpl.java @@ -104,10 +104,11 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl { */ private static final String SAS_KEY_GENERATOR_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT = - "1000,3,10000,2"; + "10,3,100,2"; private WasbRemoteCallHelper remoteCallHelper = null; private boolean isKerberosSupportEnabled; + private boolean isSpnegoTokenCacheEnabled; private RetryPolicy retryPolicy; private String[] commaSeparatedUrls; @@ -126,13 +127,16 @@ public class RemoteSASKeyGeneratorImpl extends SASKeyGeneratorImpl { this.isKerberosSupportEnabled = conf.getBoolean(Constants.AZURE_KERBEROS_SUPPORT_PROPERTY_NAME, false); + this.isSpnegoTokenCacheEnabled = + conf.getBoolean(Constants.AZURE_ENABLE_SPNEGO_TOKEN_CACHE, true); this.commaSeparatedUrls = conf.getTrimmedStrings(KEY_CRED_SERVICE_URLS); if (this.commaSeparatedUrls == null || this.commaSeparatedUrls.length <= 0) { throw new IOException( KEY_CRED_SERVICE_URLS + " config not set" + " in configuration."); } if (isKerberosSupportEnabled && UserGroupInformation.isSecurityEnabled()) { - this.remoteCallHelper = new SecureWasbRemoteCallHelper(retryPolicy, false); + this.remoteCallHelper = new SecureWasbRemoteCallHelper(retryPolicy, false, + isSpnegoTokenCacheEnabled); } else { this.remoteCallHelper = new WasbRemoteCallHelper(retryPolicy); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteWasbAuthorizerImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteWasbAuthorizerImpl.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteWasbAuthorizerImpl.java index a0eb39e..f50fc01 100644 --- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteWasbAuthorizerImpl.java +++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteWasbAuthorizerImpl.java @@ -93,10 +93,11 @@ public class RemoteWasbAuthorizerImpl implements WasbAuthorizerInterface { * Authorization Remote http client retry policy spec default value. {@value} */ private static final String AUTHORIZER_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT = - "1000,3,10000,2"; + "10,3,100,2"; private WasbRemoteCallHelper remoteCallHelper = null; private boolean isKerberosSupportEnabled; + private boolean isSpnegoTokenCacheEnabled; private RetryPolicy retryPolicy; private String[] commaSeparatedUrls = null; @@ -111,6 +112,8 @@ public class RemoteWasbAuthorizerImpl implements WasbAuthorizerInterface { LOG.debug("Initializing RemoteWasbAuthorizerImpl instance"); this.isKerberosSupportEnabled = conf.getBoolean(Constants.AZURE_KERBEROS_SUPPORT_PROPERTY_NAME, false); + this.isSpnegoTokenCacheEnabled = + conf.getBoolean(Constants.AZURE_ENABLE_SPNEGO_TOKEN_CACHE, true); this.commaSeparatedUrls = conf.getTrimmedStrings(KEY_REMOTE_AUTH_SERVICE_URLS); if (this.commaSeparatedUrls == null @@ -123,7 +126,8 @@ public class RemoteWasbAuthorizerImpl implements WasbAuthorizerInterface { AUTHORIZER_HTTP_CLIENT_RETRY_POLICY_SPEC_SPEC, AUTHORIZER_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT); if (isKerberosSupportEnabled && UserGroupInformation.isSecurityEnabled()) { - this.remoteCallHelper = new SecureWasbRemoteCallHelper(retryPolicy, false); + this.remoteCallHelper = new SecureWasbRemoteCallHelper(retryPolicy, false, + isSpnegoTokenCacheEnabled); } else { this.remoteCallHelper = new WasbRemoteCallHelper(retryPolicy); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SecureWasbRemoteCallHelper.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SecureWasbRemoteCallHelper.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SecureWasbRemoteCallHelper.java index 7f8bc0e..a0204be 100644 --- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SecureWasbRemoteCallHelper.java +++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SecureWasbRemoteCallHelper.java @@ -6,9 +6,9 @@ * 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 - * <p> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p> + * + * 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. @@ -20,6 +20,7 @@ package org.apache.hadoop.fs.azure; import org.apache.commons.lang.Validate; import org.apache.hadoop.fs.azure.security.Constants; +import org.apache.hadoop.fs.azure.security.SpnegoToken; import org.apache.hadoop.fs.azure.security.WasbDelegationTokenIdentifier; import org.apache.hadoop.io.retry.RetryPolicy; import org.apache.hadoop.security.UserGroupInformation; @@ -39,6 +40,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; +import java.net.InetAddress; import java.net.URISyntaxException; import java.security.PrivilegedExceptionAction; import java.util.List; @@ -69,10 +71,21 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper { */ private boolean alwaysRequiresKerberosAuth; + /** + * Enable caching of Spnego token. + */ + private boolean isSpnegoTokenCachingEnabled; + + /** + * Cached SPNEGO token. + */ + private SpnegoToken spnegoToken; + public SecureWasbRemoteCallHelper(RetryPolicy retryPolicy, - boolean alwaysRequiresKerberosAuth) { + boolean alwaysRequiresKerberosAuth, boolean isSpnegoTokenCachingEnabled) { super(retryPolicy); this.alwaysRequiresKerberosAuth = alwaysRequiresKerberosAuth; + this.isSpnegoTokenCachingEnabled = isSpnegoTokenCachingEnabled; } @Override @@ -81,32 +94,6 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper { final String httpMethod) throws IOException { final UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); UserGroupInformation connectUgi = ugi.getRealUser(); - if (connectUgi == null) { - connectUgi = ugi; - } - if (delegationToken == null) { - connectUgi.checkTGTAndReloginFromKeytab(); - } - String s = null; - try { - s = connectUgi.doAs(new PrivilegedExceptionAction<String>() { - @Override public String run() throws Exception { - return retryableRequest(urls, path, queryParams, httpMethod); - } - }); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new IOException(e.getMessage(), e); - } - return s; - } - - @Override - public HttpUriRequest getHttpRequest(String[] urls, String path, - List<NameValuePair> queryParams, int urlIndex, String httpMethod) - throws URISyntaxException, IOException { - final UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); - UserGroupInformation connectUgi = ugi.getRealUser(); if (connectUgi != null) { queryParams.add(new NameValuePair() { @Override public String getName() { @@ -117,6 +104,8 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper { return ugi.getShortUserName(); } }); + } else { + connectUgi = ugi; } final Token delegationToken = getDelegationToken(ugi); @@ -134,8 +123,32 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper { }); } + if (delegationToken == null) { + connectUgi.checkTGTAndReloginFromKeytab(); + } + String s = null; + try { + s = connectUgi.doAs(new PrivilegedExceptionAction<String>() { + @Override public String run() throws Exception { + return retryableRequest(urls, path, queryParams, httpMethod); + } + }); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e.getMessage(), e); + } + return s; + } + + @Override + public HttpUriRequest getHttpRequest(String[] urls, String path, + List<NameValuePair> queryParams, int urlIndex, String httpMethod, + boolean requiresNewAuth) throws URISyntaxException, IOException { URIBuilder uriBuilder = new URIBuilder(urls[urlIndex]).setPath(path).setParameters(queryParams); + if (uriBuilder.getHost().equals("localhost")) { + uriBuilder.setHost(InetAddress.getLocalHost().getCanonicalHostName()); + } HttpUriRequest httpUriRequest = null; switch (httpMethod) { case HttpPut.METHOD_NAME: @@ -152,11 +165,18 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper { LOG.debug("SecureWasbRemoteCallHelper#getHttpRequest() {}", uriBuilder.build().toURL()); if (alwaysRequiresKerberosAuth || delegationToken == null) { - AuthenticatedURL.Token token = new AuthenticatedURL.Token(); + AuthenticatedURL.Token token = null; final Authenticator kerberosAuthenticator = new KerberosDelegationTokenAuthenticator(); try { - kerberosAuthenticator.authenticate(uriBuilder.build().toURL(), token); + if (isSpnegoTokenCachingEnabled && !requiresNewAuth + && spnegoToken != null && spnegoToken.isTokenValid()){ + token = spnegoToken.getToken(); + } else { + token = new AuthenticatedURL.Token(); + kerberosAuthenticator.authenticate(uriBuilder.build().toURL(), token); + spnegoToken = new SpnegoToken(token); + } } catch (AuthenticationException e) { throw new WasbRemoteCallException( Constants.AUTHENTICATION_FAILED_ERROR_MESSAGE, e); @@ -170,7 +190,7 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper { return httpUriRequest; } - private synchronized Token<?> getDelegationToken( + private Token<?> getDelegationToken( UserGroupInformation userGroupInformation) throws IOException { if (this.delegationToken == null) { Token<?> token = null; http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/WasbRemoteCallHelper.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/WasbRemoteCallHelper.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/WasbRemoteCallHelper.java index 2b22242..cacb424 100644 --- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/WasbRemoteCallHelper.java +++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/WasbRemoteCallHelper.java @@ -6,9 +6,9 @@ * 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 - * <p> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p> + * + * 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. @@ -41,6 +41,7 @@ import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.io.InterruptedIOException; +import java.net.InetAddress; import java.net.URISyntaxException; import java.nio.charset.StandardCharsets; import java.util.List; @@ -85,8 +86,7 @@ public class WasbRemoteCallHelper { this.retryPolicy = retryPolicy; } - @VisibleForTesting - public void updateHttpClient(HttpClient client) { + @VisibleForTesting public void updateHttpClient(HttpClient client) { this.client = client; } @@ -112,25 +112,57 @@ public class WasbRemoteCallHelper { HttpResponse response = null; HttpUriRequest httpRequest = null; - for (int retry = 0, index = - random.nextInt(urls.length);; retry++, index++) { + /** + * Get the index of local url if any. If list of urls contains strings like + * "https://localhost:"; or "http://localhost";, consider it as local url and + * give it affinity more than other urls in the list. + */ + + int indexOfLocalUrl = -1; + for (int i = 0; i < urls.length; i++) { + if (urls[i].toLowerCase().startsWith("https://localhost:";) || urls[i] + .toLowerCase().startsWith("http://localhost:";)) { + indexOfLocalUrl = i; + } + } + + boolean requiresNewAuth = false; + for (int retry = 0, index = (indexOfLocalUrl != -1) + ? indexOfLocalUrl + : random + .nextInt(urls.length);; retry++, index++) { if (index >= urls.length) { index = index % urls.length; } - + /** + * If the first request fails to localhost, then randomly pick the next url + * from the remaining urls in the list, so that load can be balanced. + */ + if (indexOfLocalUrl != -1 && retry == 1) { + index = (index + random.nextInt(urls.length)) % urls.length; + if (index == indexOfLocalUrl) { + index = (index + 1) % urls.length; + } + } try { httpRequest = - getHttpRequest(urls, path, queryParams, index, httpMethod); - + getHttpRequest(urls, path, queryParams, index, httpMethod, + requiresNewAuth); httpRequest.setHeader("Accept", APPLICATION_JSON); response = client.execute(httpRequest); StatusLine statusLine = response.getStatusLine(); if (statusLine == null || statusLine.getStatusCode() != HttpStatus.SC_OK) { + requiresNewAuth = + (statusLine == null) + || (statusLine.getStatusCode() == HttpStatus.SC_UNAUTHORIZED); + throw new WasbRemoteCallException( httpRequest.getURI().toString() + ":" + ((statusLine != null) ? statusLine.toString() : "NULL")); + } else { + requiresNewAuth = false; } Header contentTypeHeader = response.getFirstHeader("Content-Type"); @@ -201,11 +233,14 @@ public class WasbRemoteCallHelper { } protected HttpUriRequest getHttpRequest(String[] urls, String path, - List<NameValuePair> queryParams, int urlIndex, String httpMethod) - throws URISyntaxException, IOException { + List<NameValuePair> queryParams, int urlIndex, String httpMethod, + boolean requiresNewAuth) throws URISyntaxException, IOException { URIBuilder uriBuilder = null; uriBuilder = new URIBuilder(urls[urlIndex]).setPath(path).setParameters(queryParams); + if (uriBuilder.getHost().equals("localhost")) { + uriBuilder.setHost(InetAddress.getLocalHost().getCanonicalHostName()); + } HttpUriRequest httpUriRequest = null; switch (httpMethod) { case HttpPut.METHOD_NAME: @@ -247,7 +282,7 @@ public class WasbRemoteCallHelper { Thread.sleep(a.delayMillis); return; } - } catch(InterruptedIOException e) { + } catch (InterruptedIOException e) { LOG.warn(e.getMessage(), e); Thread.currentThread().interrupt(); return; http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/Constants.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/Constants.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/Constants.java index cacdfc5..fa63837 100644 --- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/Constants.java +++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/Constants.java @@ -23,22 +23,27 @@ package org.apache.hadoop.fs.azure.security; */ public final class Constants { - private Constants() { - } - /** * The configuration property to enable Kerberos support. */ - public static final String AZURE_KERBEROS_SUPPORT_PROPERTY_NAME = "fs.azure.enable.kerberos.support"; - + public static final String AZURE_KERBEROS_SUPPORT_PROPERTY_NAME = + "fs.azure.enable.kerberos.support"; + /** + * The configuration property to enable SPNEGO token cache. + */ + public static final String AZURE_ENABLE_SPNEGO_TOKEN_CACHE = + "fs.azure.enable.spnego.token.cache"; /** * Parameter to be used for impersonation. */ public static final String DOAS_PARAM = "doas"; - /** * Error message for Authentication failures. */ - public static final String AUTHENTICATION_FAILED_ERROR_MESSAGE = "Authentication Failed "; + public static final String AUTHENTICATION_FAILED_ERROR_MESSAGE = + "Authentication Failed "; + + private Constants() { + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/RemoteWasbDelegationTokenManager.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/RemoteWasbDelegationTokenManager.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/RemoteWasbDelegationTokenManager.java index 1078f88..36381dc 100644 --- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/RemoteWasbDelegationTokenManager.java +++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/RemoteWasbDelegationTokenManager.java @@ -6,9 +6,9 @@ * 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 - * <p> - * http://www.apache.org/licenses/LICENSE-2.0 - * <p> + * + * 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. @@ -34,7 +34,7 @@ import java.io.IOException; import java.util.Map; /** - * Class to manage delegation token operations by making rest call to remote service. + * Class to manage delegation token operations by making rest call to remote service. */ public class RemoteWasbDelegationTokenManager implements WasbDelegationTokenManager { @@ -64,24 +64,26 @@ public class RemoteWasbDelegationTokenManager * Default for delegation token service http retry policy spec. */ private static final String DT_MANAGER_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT = - "1000,3,10000,2"; + "10,3,100,2"; private static final boolean DT_MANAGER_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT = true; private static final Text WASB_DT_SERVICE_NAME = new Text("WASB_DT_SERVICE"); /** - * Query parameter value for Getting delegation token http request + * Query parameter value for Getting delegation token http request */ private static final String GET_DELEGATION_TOKEN_OP = "GETDELEGATIONTOKEN"; /** * Query parameter value for renewing delegation token http request */ - private static final String RENEW_DELEGATION_TOKEN_OP = "RENEWDELEGATIONTOKEN"; + private static final String RENEW_DELEGATION_TOKEN_OP = + "RENEWDELEGATIONTOKEN"; /** * Query parameter value for canceling the delegation token http request */ - private static final String CANCEL_DELEGATION_TOKEN_OP = "CANCELDELEGATIONTOKEN"; + private static final String CANCEL_DELEGATION_TOKEN_OP = + "CANCELDELEGATIONTOKEN"; /** * op parameter to represent the operation. */ @@ -100,6 +102,7 @@ public class RemoteWasbDelegationTokenManager private static final String TOKEN_PARAM_KEY_NAME = "token"; private WasbRemoteCallHelper remoteCallHelper; private String[] dtServiceUrls; + private boolean isSpnegoTokenCacheEnabled; public RemoteWasbDelegationTokenManager(Configuration conf) throws IOException { @@ -108,8 +111,11 @@ public class RemoteWasbDelegationTokenManager DT_MANAGER_HTTP_CLIENT_RETRY_POLICY_ENABLED_DEFAULT, DT_MANAGER_HTTP_CLIENT_RETRY_POLICY_SPEC_KEY, DT_MANAGER_HTTP_CLIENT_RETRY_POLICY_SPEC_DEFAULT); + this.isSpnegoTokenCacheEnabled = + conf.getBoolean(Constants.AZURE_ENABLE_SPNEGO_TOKEN_CACHE, true); - remoteCallHelper = new SecureWasbRemoteCallHelper(retryPolicy, true); + remoteCallHelper = new SecureWasbRemoteCallHelper(retryPolicy, true, + isSpnegoTokenCacheEnabled); this.dtServiceUrls = conf.getTrimmedStrings(KEY_DELEGATION_TOKEN_SERVICE_URLS); if (this.dtServiceUrls == null || this.dtServiceUrls.length <= 0) { @@ -126,7 +132,8 @@ public class RemoteWasbDelegationTokenManager new URIBuilder().setPath(DEFAULT_DELEGATION_TOKEN_MANAGER_ENDPOINT) .addParameter(OP_PARAM_KEY_NAME, GET_DELEGATION_TOKEN_OP) .addParameter(RENEWER_PARAM_KEY_NAME, renewer) - .addParameter(SERVICE_PARAM_KEY_NAME, WASB_DT_SERVICE_NAME.toString()); + .addParameter(SERVICE_PARAM_KEY_NAME, + WASB_DT_SERVICE_NAME.toString()); String responseBody = remoteCallHelper .makeRemoteRequest(dtServiceUrls, uriBuilder.getPath(), uriBuilder.getQueryParams(), HttpGet.METHOD_NAME); http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/SpnegoToken.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/SpnegoToken.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/SpnegoToken.java new file mode 100644 index 0000000..fba4e41 --- /dev/null +++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/SpnegoToken.java @@ -0,0 +1,49 @@ +/** + * 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.fs.azure.security; + +import org.apache.hadoop.security.authentication.client.AuthenticatedURL; + +/** + * Class to represent SPNEGO token. + */ +public class SpnegoToken { + private AuthenticatedURL.Token token; + private long expiryTime; + private static final long TOKEN_VALIDITY_TIME_IN_MS = 60 * 60 * 1000L; + + public SpnegoToken(AuthenticatedURL.Token token) { + this.token = token; + //set the expiry time of the token to be 60 minutes, + // actual token will be valid for more than few hours and treating token as opaque. + this.expiryTime = System.currentTimeMillis() + TOKEN_VALIDITY_TIME_IN_MS; + } + + public AuthenticatedURL.Token getToken() { + return token; + } + + public long getExpiryTime() { + return expiryTime; + } + + public boolean isTokenValid() { + return (expiryTime >= System.currentTimeMillis()); + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/b9f2effa/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestWasbRemoteCallHelper.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestWasbRemoteCallHelper.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestWasbRemoteCallHelper.java index efdea8b..7aaefea 100644 --- a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestWasbRemoteCallHelper.java +++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestWasbRemoteCallHelper.java @@ -43,6 +43,8 @@ import org.mockito.ArgumentMatcher; import org.mockito.Mockito; import java.io.ByteArrayInputStream; +import java.net.InetAddress; +import java.net.UnknownHostException; import java.nio.charset.StandardCharsets; import static org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.KEY_USE_SECURE_MODE; @@ -62,7 +64,7 @@ public class TestWasbRemoteCallHelper protected AzureBlobStorageTestAccount createTestAccount() throws Exception { Configuration conf = new Configuration(); conf.set(NativeAzureFileSystem.KEY_AZURE_AUTHORIZATION, "true"); - conf.set(RemoteWasbAuthorizerImpl.KEY_REMOTE_AUTH_SERVICE_URLS, "http://localhost1/,http://localhost2/";); + conf.set(RemoteWasbAuthorizerImpl.KEY_REMOTE_AUTH_SERVICE_URLS, "http://localhost1/,http://localhost2/,http://localhost:8080";); return AzureBlobStorageTestAccount.create(conf); } @@ -304,6 +306,18 @@ public class TestWasbRemoteCallHelper Mockito.when(mockHttpResponseService2.getEntity()) .thenReturn(mockHttpEntity); + HttpResponse mockHttpResponseServiceLocal = Mockito.mock(HttpResponse.class); + Mockito.when(mockHttpResponseServiceLocal.getStatusLine()) + .thenReturn(newStatusLine(HttpStatus.SC_INTERNAL_SERVER_ERROR)); + Mockito.when(mockHttpResponseServiceLocal.getFirstHeader("Content-Type")) + .thenReturn(newHeader("Content-Type", "application/json")); + Mockito.when(mockHttpResponseServiceLocal.getFirstHeader("Content-Length")) + .thenReturn(newHeader("Content-Length", "1024")); + Mockito.when(mockHttpResponseServiceLocal.getEntity()) + .thenReturn(mockHttpEntity); + + + class HttpGetForService1 extends ArgumentMatcher<HttpGet>{ @Override public boolean matches(Object o) { return checkHttpGetMatchHost((HttpGet) o, "localhost1"); @@ -314,10 +328,21 @@ public class TestWasbRemoteCallHelper return checkHttpGetMatchHost((HttpGet) o, "localhost2"); } } + class HttpGetForServiceLocal extends ArgumentMatcher<HttpGet>{ + @Override public boolean matches(Object o) { + try { + return checkHttpGetMatchHost((HttpGet) o, InetAddress.getLocalHost().getCanonicalHostName()); + } catch (UnknownHostException e) { + return checkHttpGetMatchHost((HttpGet) o, "localhost"); + } + } + } Mockito.when(mockHttpClient.execute(argThat(new HttpGetForService1()))) .thenReturn(mockHttpResponseService1); Mockito.when(mockHttpClient.execute(argThat(new HttpGetForService2()))) .thenReturn(mockHttpResponseService2); + Mockito.when(mockHttpClient.execute(argThat(new HttpGetForServiceLocal()))) + .thenReturn(mockHttpResponseServiceLocal); //Need 3 times because performop() does 3 fs operations. Mockito.when(mockHttpEntity.getContent()) @@ -331,6 +356,7 @@ public class TestWasbRemoteCallHelper performop(mockHttpClient); + Mockito.verify(mockHttpClient, times(3)).execute(Mockito.argThat(new HttpGetForServiceLocal())); Mockito.verify(mockHttpClient, times(3)).execute(Mockito.argThat(new HttpGetForService2())); } @@ -362,6 +388,17 @@ public class TestWasbRemoteCallHelper Mockito.when(mockHttpResponseService2.getEntity()) .thenReturn(mockHttpEntity); + HttpResponse mockHttpResponseService3 = Mockito.mock(HttpResponse.class); + Mockito.when(mockHttpResponseService3.getStatusLine()) + .thenReturn(newStatusLine( + HttpStatus.SC_INTERNAL_SERVER_ERROR)); + Mockito.when(mockHttpResponseService3.getFirstHeader("Content-Type")) + .thenReturn(newHeader("Content-Type", "application/json")); + Mockito.when(mockHttpResponseService3.getFirstHeader("Content-Length")) + .thenReturn(newHeader("Content-Length", "1024")); + Mockito.when(mockHttpResponseService3.getEntity()) + .thenReturn(mockHttpEntity); + class HttpGetForService1 extends ArgumentMatcher<HttpGet>{ @Override public boolean matches(Object o) { return checkHttpGetMatchHost((HttpGet) o, "localhost1"); @@ -372,10 +409,21 @@ public class TestWasbRemoteCallHelper return checkHttpGetMatchHost((HttpGet) o, "localhost2"); } } + class HttpGetForService3 extends ArgumentMatcher<HttpGet> { + @Override public boolean matches(Object o){ + try { + return checkHttpGetMatchHost((HttpGet) o, InetAddress.getLocalHost().getCanonicalHostName()); + } catch (UnknownHostException e) { + return checkHttpGetMatchHost((HttpGet) o, "localhost"); + } + } + } Mockito.when(mockHttpClient.execute(argThat(new HttpGetForService1()))) .thenReturn(mockHttpResponseService1); Mockito.when(mockHttpClient.execute(argThat(new HttpGetForService2()))) .thenReturn(mockHttpResponseService2); + Mockito.when(mockHttpClient.execute(argThat(new HttpGetForService3()))) + .thenReturn(mockHttpResponseService3); //Need 3 times because performop() does 3 fs operations. Mockito.when(mockHttpEntity.getContent()) @@ -390,10 +438,12 @@ public class TestWasbRemoteCallHelper performop(mockHttpClient); }catch (WasbAuthorizationException e){ e.printStackTrace(); - Mockito.verify(mockHttpClient, atLeast(3)) + Mockito.verify(mockHttpClient, atLeast(2)) .execute(argThat(new HttpGetForService1())); - Mockito.verify(mockHttpClient, atLeast(3)) + Mockito.verify(mockHttpClient, atLeast(2)) .execute(argThat(new HttpGetForService2())); + Mockito.verify(mockHttpClient, atLeast(3)) + .execute(argThat(new HttpGetForService3())); Mockito.verify(mockHttpClient, times(7)).execute(Mockito.<HttpGet>any()); } } @@ -425,7 +475,7 @@ public class TestWasbRemoteCallHelper expectedEx.expectMessage(new MatchesPattern( "org\\.apache\\.hadoop\\.fs\\.azure\\.WasbRemoteCallException: " + "Encountered error while making remote call to " - + "http:\\/\\/localhost1\\/,http:\\/\\/localhost2\\/ retried 6 time\\(s\\)\\.")); + + "http:\\/\\/localhost1\\/,http:\\/\\/localhost2\\/,http:\\/\\/localhost:8080 retried 6 time\\(s\\)\\.")); } private void performop(HttpClient mockHttpClient) throws Throwable { --------------------------------------------------------------------- To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-commits-h...@hadoop.apache.org
