commit e92cc905c1e56a44e428690cca5c9b1c49518611

Author: Suresh Srinivas <sureshms@yahoo-inc.com>
Date:   Fri Jul 16 13:57:48 2010 -0700

    HDFS-1289 from https://issues.apache.org/jira/secure/attachment/12449701/HDFS-1298.y20.patch
    
    +++ b/YAHOO-CHANGES.txt
    +    HDFS-1298 - Add support in HDFS for new statistics added in FileSystem
    +    to track the file system operations (suresh)
    +


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

src/hdfs/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -174,6 +174,7 @@ public class DistributedFileSystem extends FileSystem {
     if (file == null) {
       return null;
     }
+    statistics.incrementReadOps(1);
     return dfs.getBlockLocations(getPathName(file.getPath()), start, len);
   }
 
@@ -182,6 +183,7 @@ public class DistributedFileSystem extends FileSystem {
   }
 
   public FSDataInputStream open(Path f, int bufferSize) throws IOException {
+    statistics.incrementReadOps(1);
     return new DFSClient.DFSDataInputStream(
           dfs.open(getPathName(f), bufferSize, verifyChecksum, statistics));
   }
@@ -190,6 +192,7 @@ public class DistributedFileSystem extends FileSystem {
   public FSDataOutputStream append(Path f, int bufferSize,
       Progressable progress) throws IOException {
 
+    statistics.incrementWriteOps(1);
     DFSOutputStream op = (DFSOutputStream)dfs.append(getPathName(f), bufferSize, progress);
     return new FSDataOutputStream(op, statistics, op.getInitialLen());
   }
@@ -199,6 +202,7 @@ public class DistributedFileSystem extends FileSystem {
     int bufferSize, short replication, long blockSize,
     Progressable progress) throws IOException {
 
+    statistics.incrementWriteOps(1);
     return new FSDataOutputStream
        (dfs.create(getPathName(f), permission,
                    overwrite, replication, blockSize, progress, bufferSize),
@@ -208,6 +212,7 @@ public class DistributedFileSystem extends FileSystem {
   public boolean setReplication(Path src, 
                                 short replication
                                ) throws IOException {
+    statistics.incrementWriteOps(1);
     return dfs.setReplication(getPathName(src), replication);
   }
 
@@ -215,6 +220,7 @@ public class DistributedFileSystem extends FileSystem {
    * Rename files/dirs
    */
   public boolean rename(Path src, Path dst) throws IOException {
+    statistics.incrementWriteOps(1);
     return dfs.rename(getPathName(src), getPathName(dst));
   }
 
@@ -223,6 +229,7 @@ public class DistributedFileSystem extends FileSystem {
    */
   @Deprecated
   public boolean delete(Path f) throws IOException {
+    statistics.incrementWriteOps(1);
     return dfs.delete(getPathName(f));
   }
   
@@ -231,11 +238,13 @@ public class DistributedFileSystem extends FileSystem {
    * empty directory recursively.
    */
   public boolean delete(Path f, boolean recursive) throws IOException {
-   return dfs.delete(getPathName(f), recursive);
+    statistics.incrementWriteOps(1);
+    return dfs.delete(getPathName(f), recursive);
   }
   
   /** {@inheritDoc} */
   public ContentSummary getContentSummary(Path f) throws IOException {
+    statistics.incrementReadOps(1);
     return dfs.getContentSummary(getPathName(f));
   }
 
@@ -281,6 +290,7 @@ public class DistributedFileSystem extends FileSystem {
       for (int i = 0; i < partialListing.length; i++) {
         stats[i] = makeQualified(partialListing[i], p);
       }
+      statistics.incrementReadOps(1);
       return stats;
     }
     
@@ -294,6 +304,7 @@ public class DistributedFileSystem extends FileSystem {
     for (HdfsFileStatus fileStatus : partialListing) {
       listing.add(makeQualified(fileStatus, p));
     }
+    statistics.incrementLargeReadOps(1);
 
     // now fetch more entries
     do {
@@ -307,12 +318,14 @@ public class DistributedFileSystem extends FileSystem {
       for (HdfsFileStatus fileStatus : partialListing) {
         listing.add(makeQualified(fileStatus, p));
       }
+      statistics.incrementLargeReadOps(1);
     } while (thisListing.hasMore());
 
     return listing.toArray(new FileStatus[listing.size()]);
   }
 
   public boolean mkdirs(Path f, FsPermission permission) throws IOException {
+    statistics.incrementWriteOps(1);
     return dfs.mkdirs(getPathName(f), permission);
   }
 
@@ -504,6 +517,7 @@ public class DistributedFileSystem extends FileSystem {
    * @throws FileNotFoundException if the file does not exist.
    */
   public FileStatus getFileStatus(Path f) throws IOException {
+    statistics.incrementReadOps(1);
     HdfsFileStatus fi = dfs.getFileInfo(getPathName(f));
     if (fi != null) {
       return makeQualified(fi, f);
@@ -514,12 +528,14 @@ public class DistributedFileSystem extends FileSystem {
 
   /** {@inheritDoc} */
   public MD5MD5CRC32FileChecksum getFileChecksum(Path f) throws IOException {
+    statistics.incrementReadOps(1);
     return dfs.getFileChecksum(getPathName(f));
   }
 
   /** {@inheritDoc }*/
   public void setPermission(Path p, FsPermission permission
       ) throws IOException {
+    statistics.incrementWriteOps(1);
     dfs.setPermission(getPathName(p), permission);
   }
 
@@ -529,12 +545,14 @@ public class DistributedFileSystem extends FileSystem {
     if (username == null && groupname == null) {
       throw new IOException("username == null && groupname == null");
     }
+    statistics.incrementWriteOps(1);
     dfs.setOwner(getPathName(p), username, groupname);
   }
 
   /** {@inheritDoc }*/
   public void setTimes(Path p, long mtime, long atime
       ) throws IOException {
+    statistics.incrementWriteOps(1);
     dfs.setTimes(getPathName(p), mtime, atime);
   }
 

src/test/org/apache/hadoop/hdfs/DFSTestUtil.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem.Statistics;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.ShellBasedUnixGroupsMapping;
@@ -357,4 +358,8 @@ public class DFSTestUtil {
                     }
                   });
   }  
+
+  public static Statistics getStatistics(FileSystem fs) {
+    return FileSystem.getStatistics(fs.getUri().getScheme(), fs.getClass());
+  }
 }

src/test/org/apache/hadoop/hdfs/TestDistributedFileSystem.java

@@ -27,11 +27,16 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileChecksum;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.log4j.Level;
+import org.junit.Test;
+import static org.junit.Assert.*;
 
-public class TestDistributedFileSystem extends junit.framework.TestCase {
+public class TestDistributedFileSystem {
   private static final Random RAN = new Random();
 
   private boolean dualPortTesting = false;
@@ -45,6 +50,7 @@ public class TestDistributedFileSystem extends junit.framework.TestCase {
     return conf;
   }
 
+  @Test
   public void testFileSystemCloseAll() throws Exception {
     Configuration conf = getTestConfiguration();
     MiniDFSCluster cluster = new MiniDFSCluster(conf, 0, true, null);
@@ -68,6 +74,7 @@ public class TestDistributedFileSystem extends junit.framework.TestCase {
    * Tests DFSClient.close throws no ConcurrentModificationException if 
    * multiple files are open.
    */
+  @Test
   public void testDFSClose() throws Exception {
     Configuration conf = getTestConfiguration();
     MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
@@ -85,6 +92,7 @@ public class TestDistributedFileSystem extends junit.framework.TestCase {
     }
   }
 
+  @Test
   public void testDFSClient() throws Exception {
     Configuration conf = getTestConfiguration();
     MiniDFSCluster cluster = null;
@@ -129,6 +137,7 @@ public class TestDistributedFileSystem extends junit.framework.TestCase {
     }
   }
   
+  @Test
   public void testFileChecksum() throws IOException {
     ((Log4JLogger)HftpFileSystem.LOG).getLogger().setLevel(Level.ALL);
 
@@ -202,6 +211,7 @@ public class TestDistributedFileSystem extends junit.framework.TestCase {
     cluster.shutdown();
   }
   
+  @Test
   public void testAllWithDualPort() throws Exception {
     dualPortTesting = true;
 
@@ -210,4 +220,96 @@ public class TestDistributedFileSystem extends junit.framework.TestCase {
     testDFSClient();
     testFileChecksum();
   }
+  
+  @Test
+  public void testStatistics() throws Exception {
+    int lsLimit = 2;
+    final Configuration conf = getTestConfiguration();
+    conf.setInt(DFSConfigKeys.DFS_LIST_LIMIT, lsLimit);
+    final MiniDFSCluster cluster = new MiniDFSCluster(conf, 1, true, null);
+    try {
+      final FileSystem fs = cluster.getFileSystem();
+      Path dir = new Path("/test");
+      Path file = new Path(dir, "file");
+      
+      int readOps = DFSTestUtil.getStatistics(fs).getReadOps();
+      int writeOps = DFSTestUtil.getStatistics(fs).getWriteOps();
+      int largeReadOps = DFSTestUtil.getStatistics(fs).getLargeReadOps();
+      fs.mkdirs(dir);
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+      FSDataOutputStream out = fs.create(file, (short)1);
+      out.close();
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+      FileStatus status = fs.getFileStatus(file);
+      checkStatistics(fs, ++readOps, writeOps, largeReadOps);
+      
+      fs.getFileBlockLocations(status, 0, 0);
+      checkStatistics(fs, ++readOps, writeOps, largeReadOps);
+      
+      FSDataInputStream in = fs.open(file);
+      in.close();
+      checkStatistics(fs, ++readOps, writeOps, largeReadOps);
+      
+      fs.setReplication(file, (short)2);
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+      Path file1 = new Path(dir, "file1");
+      fs.rename(file, file1);
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+      fs.getContentSummary(file1);
+      checkStatistics(fs, ++readOps, writeOps, largeReadOps);
+      
+      
+      // Iterative ls test
+      for (int i = 0; i < 10; i++) {
+        Path p = new Path(dir, Integer.toString(i));
+        fs.mkdirs(p);
+        FileStatus[] list = fs.listStatus(dir);
+        if (list.length > lsLimit) {
+          // if large directory, then count readOps and largeReadOps by 
+          // number times listStatus iterates
+          int iterations = (int)Math.ceil((double)list.length/lsLimit);
+          largeReadOps += iterations;
+          readOps += iterations;
+        } else {
+          // Single iteration in listStatus - no large read operation done
+          readOps++;
+        }
+        
+        // writeOps incremented by 1 for mkdirs
+        // readOps and largeReadOps incremented by 1 or more
+        checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      }
+      
+      fs.getFileChecksum(file1);
+      checkStatistics(fs, ++readOps, writeOps, largeReadOps);
+      
+      fs.setPermission(file1, new FsPermission((short)0777));
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+      fs.setTimes(file1, 0L, 0L);
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+      UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+      fs.setOwner(file1, ugi.getUserName(), ugi.getGroupNames()[0]);
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+      fs.delete(dir, true);
+      checkStatistics(fs, readOps, ++writeOps, largeReadOps);
+      
+    } finally {
+      if (cluster != null) cluster.shutdown();
+    }
+    
+  }
+  
+  /** Checks statistics. -1 indicates do not check for the operations */
+  private void checkStatistics(FileSystem fs, int readOps, int writeOps, int largeReadOps) {
+    assertEquals(readOps, DFSTestUtil.getStatistics(fs).getReadOps());
+    assertEquals(writeOps, DFSTestUtil.getStatistics(fs).getWriteOps());
+    assertEquals(largeReadOps, DFSTestUtil.getStatistics(fs).getLargeReadOps());
+  }
 }