- * http://www.apache.org/licenses/LICENSE-2.0 - *
+ *
+ * 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,9 +94,35 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper {
final String httpMethod) throws IOException {
final UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
UserGroupInformation connectUgi = ugi.getRealUser();
- if (connectUgi == null) {
+ if (connectUgi != null) {
+ queryParams.add(new NameValuePair() {
+ @Override public String getName() {
+ return Constants.DOAS_PARAM;
+ }
+
+ @Override public String getValue() {
+ return ugi.getShortUserName();
+ }
+ });
+ } else {
connectUgi = ugi;
}
+
+ final Token delegationToken = getDelegationToken(ugi);
+ if (!alwaysRequiresKerberosAuth && delegationToken != null) {
+ final String delegationTokenEncodedUrlString =
+ delegationToken.encodeToUrlString();
+ queryParams.add(new NameValuePair() {
+ @Override public String getName() {
+ return DELEGATION_TOKEN_QUERY_PARAM_NAME;
+ }
+
+ @Override public String getValue() {
+ return delegationTokenEncodedUrlString;
+ }
+ });
+ }
+
if (delegationToken == null) {
connectUgi.checkTGTAndReloginFromKeytab();
}
@@ -103,39 +142,13 @@ public class SecureWasbRemoteCallHelper extends WasbRemoteCallHelper {
@Override
public HttpUriRequest getHttpRequest(String[] urls, String path,
- List
- * http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
+ * 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.
@@ -40,6 +40,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;
@@ -84,8 +85,7 @@ public class WasbRemoteCallHelper {
this.retryPolicy = retryPolicy;
}
- @VisibleForTesting
- public void updateHttpClient(HttpClient client) {
+ @VisibleForTesting public void updateHttpClient(HttpClient client) {
this.client = client;
}
@@ -111,25 +111,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");
@@ -200,11 +232,14 @@ public class WasbRemoteCallHelper {
}
protected HttpUriRequest getHttpRequest(String[] urls, String path,
- List
- * http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
+ * 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);
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 00000000000..fba4e4142f5
--- /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());
+ }
+}
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 f459b242a2b..efda15def45 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