merge HADOOP-7933 from trunk

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.23@1226917 13f79535-47bb-0310-9956-ffa450edef68

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -48,6 +48,9 @@ Release 0.23.1 - Unreleased
 
     HADOOP-7504. Add the missing Ganglia31 opts to hadoop-metrics.properties as a comment. (harsh)
 
+    HADOOP-7933. Add a getDelegationTokens api to FileSystem which checks
+    for known tokens in the passed Credentials object. (sseth)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java

@@ -47,6 +47,7 @@ import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.fs.Options.Rename;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.io.MultipleIOException;
+import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
@@ -393,6 +394,40 @@ public abstract class FileSystem extends Configured implements Closeable {
   public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
     return new ArrayList<Token<?>>(0);
   }
+  
+  /**
+   * @see #getDelegationTokens(String)
+   * This is similar to getDelegationTokens, with the added restriction that if
+   * a token is already present in the passed Credentials object - that token
+   * is returned instead of a new delegation token. 
+   * 
+   * If the token is found to be cached in the Credentials object, this API does
+   * not verify the token validity or the passed in renewer. 
+   * 
+   * 
+   * @param renewer the account name that is allowed to renew the token.
+   * @param credentials a Credentials object containing already knowing 
+   *   delegationTokens.
+   * @return a list of delegation tokens.
+   * @throws IOException
+   */
+  @InterfaceAudience.LimitedPrivate({ "HDFS", "MapReduce" })
+  public List<Token<?>> getDelegationTokens(String renewer,
+      Credentials credentials) throws IOException {
+    List<Token<?>> allTokens = getDelegationTokens(renewer);
+    List<Token<?>> newTokens = new ArrayList<Token<?>>();
+    if (allTokens != null) {
+      for (Token<?> token : allTokens) {
+        Token<?> knownToken = credentials.getToken(token.getService());
+        if (knownToken == null) {
+          newTokens.add(token);
+        } else {
+          newTokens.add(knownToken);
+        }
+      }
+    }
+    return newTokens;
+  }
 
   /** create a file with the provided permission
    * The permission of the file is set to be the provided permission as in

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FilterFileSystem.java

@@ -27,6 +27,7 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Progressable;
 
@@ -388,4 +389,11 @@ public class FilterFileSystem extends FileSystem {
   public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
     return fs.getDelegationTokens(renewer);
   }
-}
+  
+  @Override
+  // FileSystem
+  public List<Token<?>> getDelegationTokens(String renewer,
+      Credentials credentials) throws IOException {
+    return fs.getDelegationTokens(renewer, credentials);
+  }
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java

@@ -24,7 +24,9 @@ import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.StringTokenizer;
 import java.util.Map.Entry;
 
@@ -45,7 +47,9 @@ import org.apache.hadoop.fs.UnsupportedFileSystemException;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.viewfs.InodeTree.INode;
 import org.apache.hadoop.fs.viewfs.InodeTree.INodeLink;
+import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Progressable;
@@ -495,7 +499,40 @@ public class ViewFileSystem extends FileSystem {
     }
     return result;
   }
-  
+
+  @Override
+  public List<Token<?>> getDelegationTokens(String renewer,
+      Credentials credentials) throws IOException {
+    List<InodeTree.MountPoint<FileSystem>> mountPoints =
+        fsState.getMountPoints();
+    int initialListSize = 0;
+    for (InodeTree.MountPoint<FileSystem> im : mountPoints) {
+      initialListSize += im.target.targetDirLinkList.length;
+    }
+    Set<String> seenServiceNames = new HashSet<String>();
+    List<Token<?>> result = new ArrayList<Token<?>>(initialListSize);
+    for (int i = 0; i < mountPoints.size(); ++i) {
+      String serviceName =
+          mountPoints.get(i).target.targetFileSystem.getCanonicalServiceName();
+      if (seenServiceNames.contains(serviceName)) {
+        continue;
+      }
+      seenServiceNames.add(serviceName);
+      Token<?> knownToken = credentials.getToken(new Text(serviceName));
+      if (knownToken != null) {
+        result.add(knownToken);
+      } else {
+        List<Token<?>> tokens =
+            mountPoints.get(i).target.targetFileSystem
+                .getDelegationTokens(renewer);
+        if (tokens != null) {
+          result.addAll(tokens);
+        }
+      }
+    }
+    return result;
+  }
+
   /*
    * An instance of this class represents an internal dir of the viewFs 
    * that is internal dir of the mount table.

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java

@@ -34,6 +34,7 @@ import org.apache.hadoop.fs.viewfs.ConfigUtil;
 import org.apache.hadoop.fs.viewfs.ViewFileSystem;
 import org.apache.hadoop.fs.viewfs.ViewFileSystem.MountPoint;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.token.Token;
 import org.junit.After;
 import org.junit.Assert;
@@ -89,6 +90,16 @@ public class ViewFileSystemBaseTest {
     // Set up the defaultMT in the config with our mount point links
     //Configuration conf = new Configuration();
     conf = ViewFileSystemTestSetup.configWithViewfsScheme();
+    setupMountPoints();
+    fsView = FileSystem.get(FsConstants.VIEWFS_URI, conf);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    fsTarget.delete(FileSystemTestHelper.getTestRootPath(fsTarget), true);
+  }
+  
+  void setupMountPoints() {
     ConfigUtil.addLink(conf, "/user", new Path(targetTestRoot,"user").toUri());
     ConfigUtil.addLink(conf, "/user2", new Path(targetTestRoot,"user").toUri());
     ConfigUtil.addLink(conf, "/data", new Path(targetTestRoot,"data").toUri());
@@ -100,20 +111,17 @@ public class ViewFileSystemBaseTest {
         new Path(targetTestRoot,"missingTarget").toUri());
     ConfigUtil.addLink(conf, "/linkToAFile",
         new Path(targetTestRoot,"aFile").toUri());
-    
-    fsView = FileSystem.get(FsConstants.VIEWFS_URI, conf);
-  }
-
-  @After
-  public void tearDown() throws Exception {
-    fsTarget.delete(FileSystemTestHelper.getTestRootPath(fsTarget), true);
   }
   
   @Test
   public void testGetMountPoints() {
     ViewFileSystem viewfs = (ViewFileSystem) fsView;
     MountPoint[] mountPoints = viewfs.getMountPoints();
-    Assert.assertEquals(7, mountPoints.length); 
+    Assert.assertEquals(getExpectedMountPoints(), mountPoints.length); 
+  }
+  
+  int getExpectedMountPoints() {
+    return 7;
   }
   
   /**
@@ -125,9 +133,46 @@ public class ViewFileSystemBaseTest {
   public void testGetDelegationTokens() throws IOException {
     List<Token<?>> delTokens = 
         fsView.getDelegationTokens("sanjay");
-    Assert.assertEquals(0, delTokens.size()); 
+    Assert.assertEquals(getExpectedDelegationTokenCount(), delTokens.size()); 
   }
   
+  int getExpectedDelegationTokenCount() {
+    return 0;
+  }
+
+  @Test
+  public void testGetDelegationTokensWithCredentials() throws IOException {
+    Credentials credentials = new Credentials();
+    List<Token<?>> delTokens =
+        fsView.getDelegationTokens("sanjay", credentials);
+
+    int expectedTokenCount = getExpectedDelegationTokenCountWithCredentials();
+
+    Assert.assertEquals(expectedTokenCount, delTokens.size());
+    for (int i = 0; i < expectedTokenCount / 2; i++) {
+      Token<?> token = delTokens.get(i);
+      credentials.addToken(token.getService(), token);
+    }
+
+    List<Token<?>> delTokens2 =
+        fsView.getDelegationTokens("sanjay", credentials);
+    Assert.assertEquals(expectedTokenCount, delTokens2.size());
+
+    for (int i = 0; i < delTokens2.size(); i++) {
+      for (int j = 0; j < delTokens.size(); j++) {
+        if (delTokens.get(j) == delTokens2.get(i)) {
+          delTokens.remove(j);
+          break;
+        }
+      }
+    }
+    Assert.assertEquals(expectedTokenCount / 2, delTokens.size());
+  }
+
+  int getExpectedDelegationTokenCountWithCredentials() {
+    return 0;
+  }
+
   @Test
   public void testBasicPaths() {
     Assert.assertEquals(FsConstants.VIEWFS_URI,
@@ -340,7 +385,7 @@ public class ViewFileSystemBaseTest {
     
     FileStatus[] dirPaths = fsView.listStatus(new Path("/"));
     FileStatus fs;
-    Assert.assertEquals(6, dirPaths.length);
+    Assert.assertEquals(getExpectedDirPaths(), dirPaths.length);
     fs = FileSystemTestHelper.containsPath(fsView, "/user", dirPaths);
       Assert.assertNotNull(fs);
       Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
@@ -372,6 +417,10 @@ public class ViewFileSystemBaseTest {
         Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
   }
   
+  int getExpectedDirPaths() {
+    return 6;
+  }
+  
   @Test
   public void testListOnMountTargetDirs() throws IOException {
     FileStatus[] dirPaths = fsView.listStatus(new Path("/data"));