HADOOP-1061. Fix bug in listing files in the S3 filesystem.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@534595 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -300,6 +300,12 @@ Trunk (unreleased changes)
 89. HADOOP-1247.  Add support to contrib/streaming for aggregate
     package, formerly called Abacus.  (Runping Qi via cutting)
 
+90. HADOOP-1061.  Fix bug in listing files in the S3 filesystem.
+    NOTE: this change is not backwards compatible!  You should use the 
+    MigrationTool supplied to migrate existing S3 filesystem data to 
+    the new format.  Please backup your data first before upgrading 
+    (using 'hadoop distcp' for example).  (tomwhite)
+
 
 Release 0.12.3 - 2007-04-06
 

src/java/org/apache/hadoop/fs/s3/FileSystemStore.java

@@ -14,6 +14,7 @@ import org.apache.hadoop.fs.Path;
 public interface FileSystemStore {
   
   void initialize(URI uri, Configuration conf) throws IOException;
+  String getVersion() throws IOException;
 
   void storeINode(Path path, INode inode) throws IOException;
   void storeBlock(Block block, File file) throws IOException;

src/java/org/apache/hadoop/fs/s3/Jets3tFileSystemStore.java

@@ -9,10 +9,9 @@ import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.io.UnsupportedEncodingException;
 import java.net.URI;
-import java.net.URLDecoder;
-import java.net.URLEncoder;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 import java.util.TreeSet;
 
@@ -27,8 +26,26 @@ import org.jets3t.service.model.S3Object;
 import org.jets3t.service.security.AWSCredentials;
 
 class Jets3tFileSystemStore implements FileSystemStore {
+  
+  private static final String FILE_SYSTEM_NAME = "fs";
+  private static final String FILE_SYSTEM_VALUE = "Hadoop";
+
+  private static final String FILE_SYSTEM_TYPE_NAME = "fs-type";
+  private static final String FILE_SYSTEM_TYPE_VALUE = "block";
 
-  private static final String PATH_DELIMITER = urlEncode(Path.SEPARATOR);
+  private static final String FILE_SYSTEM_VERSION_NAME = "fs-version";
+  private static final String FILE_SYSTEM_VERSION_VALUE = "1";
+  
+  private static final Map<String, String> METADATA =
+    new HashMap<String, String>();
+  
+  static {
+    METADATA.put(FILE_SYSTEM_NAME, FILE_SYSTEM_VALUE);
+    METADATA.put(FILE_SYSTEM_TYPE_NAME, FILE_SYSTEM_TYPE_VALUE);
+    METADATA.put(FILE_SYSTEM_VERSION_NAME, FILE_SYSTEM_VERSION_VALUE);
+  }
+
+  private static final String PATH_DELIMITER = Path.SEPARATOR;
   private static final String BLOCK_PREFIX = "block_";
 
   private Configuration conf;
@@ -94,7 +111,7 @@ class Jets3tFileSystemStore implements FileSystemStore {
     createBucket(bucket.getName());
     
     this.bufferSize = conf.getInt("io.file.buffer.size", 4096);
-  }  
+  }
 
   private void createBucket(String bucketName) throws IOException {
     try {
@@ -106,6 +123,10 @@ class Jets3tFileSystemStore implements FileSystemStore {
       throw new S3Exception(e);
     }
   }
+  
+  public String getVersion() throws IOException {
+    return FILE_SYSTEM_VERSION_VALUE;
+  }
 
   private void delete(String key) throws IOException {
     try {
@@ -127,7 +148,7 @@ class Jets3tFileSystemStore implements FileSystemStore {
   }
 
   public boolean inodeExists(Path path) throws IOException {
-    InputStream in = get(pathToKey(path));
+    InputStream in = get(pathToKey(path), true);
     if (in == null) {
       return false;
     }
@@ -136,7 +157,7 @@ class Jets3tFileSystemStore implements FileSystemStore {
   }
   
   public boolean blockExists(long blockId) throws IOException {
-    InputStream in = get(blockToKey(blockId));
+    InputStream in = get(blockToKey(blockId), false);
     if (in == null) {
       return false;
     }
@@ -144,9 +165,14 @@ class Jets3tFileSystemStore implements FileSystemStore {
     return true;
   }
 
-  private InputStream get(String key) throws IOException {
+  private InputStream get(String key, boolean checkMetadata)
+      throws IOException {
+    
     try {
       S3Object object = s3Service.getObject(bucket, key);
+      if (checkMetadata) {
+        checkMetadata(object);
+      }
       return object.getDataInputStream();
     } catch (S3ServiceException e) {
       if (e.getS3ErrorCode().equals("NoSuchKey")) {
@@ -175,8 +201,26 @@ class Jets3tFileSystemStore implements FileSystemStore {
     }
   }
 
+  private void checkMetadata(S3Object object) throws S3FileSystemException,
+      S3ServiceException {
+    
+    String name = (String) object.getMetadata(FILE_SYSTEM_NAME);
+    if (!FILE_SYSTEM_VALUE.equals(name)) {
+      throw new S3FileSystemException("Not a Hadoop S3 file.");
+    }
+    String type = (String) object.getMetadata(FILE_SYSTEM_TYPE_NAME);
+    if (!FILE_SYSTEM_TYPE_VALUE.equals(type)) {
+      throw new S3FileSystemException("Not a block file.");
+    }
+    String dataVersion = (String) object.getMetadata(FILE_SYSTEM_VERSION_NAME);
+    if (!FILE_SYSTEM_VERSION_VALUE.equals(dataVersion)) {
+      throw new VersionMismatchException(FILE_SYSTEM_VERSION_VALUE,
+          dataVersion);
+    }
+  }
+
   public INode retrieveINode(Path path) throws IOException {
-    return INode.deserialize(get(pathToKey(path)));
+    return INode.deserialize(get(pathToKey(path), true));
   }
 
   public File retrieveBlock(Block block, long byteRangeStart)
@@ -224,7 +268,7 @@ class Jets3tFileSystemStore implements FileSystemStore {
       if (!prefix.endsWith(PATH_DELIMITER)) {
         prefix += PATH_DELIMITER;
       }
-      S3Object[] objects = s3Service.listObjects(bucket, prefix, PATH_DELIMITER, 0);
+      S3Object[] objects = s3Service.listObjects(bucket, prefix, PATH_DELIMITER);
       Set<Path> prefixes = new TreeSet<Path>();
       for (int i = 0; i < objects.length; i++) {
         prefixes.add(keyToPath(objects[i].getKey()));
@@ -260,12 +304,17 @@ class Jets3tFileSystemStore implements FileSystemStore {
     }    
   }
 
-  private void put(String key, InputStream in, long length) throws IOException {
+  private void put(String key, InputStream in, long length, boolean storeMetadata)
+      throws IOException {
+    
     try {
       S3Object object = new S3Object(key);
       object.setDataInputStream(in);
       object.setContentType("binary/octet-stream");
       object.setContentLength(length);
+      if (storeMetadata) {
+        object.addAllMetadata(METADATA);
+      }
       s3Service.putObject(bucket, object);
     } catch (S3ServiceException e) {
       if (e.getCause() instanceof IOException) {
@@ -276,14 +325,14 @@ class Jets3tFileSystemStore implements FileSystemStore {
   }
 
   public void storeINode(Path path, INode inode) throws IOException {
-    put(pathToKey(path), inode.serialize(), inode.getSerializedLength());
+    put(pathToKey(path), inode.serialize(), inode.getSerializedLength(), true);
   }
 
   public void storeBlock(Block block, File file) throws IOException {
     BufferedInputStream in = null;
     try {
       in = new BufferedInputStream(new FileInputStream(file));
-      put(blockToKey(block), in, block.getLength());
+      put(blockToKey(block), in, block.getLength(), false);
     } finally {
       closeQuietly(in);
     }    
@@ -303,35 +352,13 @@ class Jets3tFileSystemStore implements FileSystemStore {
     if (!path.isAbsolute()) {
       throw new IllegalArgumentException("Path must be absolute: " + path);
     }
-    return urlEncode(path.toUri().getPath());
+    return path.toUri().getPath();
   }
 
   private Path keyToPath(String key) {
-    return new Path(urlDecode(key));
+    return new Path(key);
   }
   
-  private static String urlEncode(String s) {
-    try {
-      return URLEncoder.encode(s, "UTF-8");
-    } catch (UnsupportedEncodingException e) {
-      // Should never happen since every implementation of the Java Platform
-      // is required to support UTF-8.
-      // See http://java.sun.com/j2se/1.5.0/docs/api/java/nio/charset/Charset.html
-      throw new IllegalStateException(e);
-    }
-  }
-  
-  private static String urlDecode(String s) {
-    try {
-      return URLDecoder.decode(s, "UTF-8");
-    } catch (UnsupportedEncodingException e) {
-      // Should never happen since every implementation of the Java Platform
-      // is required to support UTF-8.
-      // See http://java.sun.com/j2se/1.5.0/docs/api/java/nio/charset/Charset.html
-      throw new IllegalStateException(e);
-    }
-  }
-
   private String blockToKey(long blockId) {
     return BLOCK_PREFIX + blockId;
   }

src/java/org/apache/hadoop/fs/s3/MigrationTool.java

@@ -0,0 +1,278 @@
+/**
+ * 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.s3;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URLDecoder;
+import java.net.URLEncoder;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.ToolBase;
+import org.jets3t.service.S3Service;
+import org.jets3t.service.S3ServiceException;
+import org.jets3t.service.impl.rest.httpclient.RestS3Service;
+import org.jets3t.service.model.S3Bucket;
+import org.jets3t.service.model.S3Object;
+import org.jets3t.service.security.AWSCredentials;
+
+/**
+ * <p>
+ * This class is a tool for migrating data from an older to a newer version
+ * of an S3 filesystem.
+ * </p>
+ * <p>
+ * All files in the filesystem are migrated by re-writing the block metadata
+ * - no datafiles are touched.
+ * </p>
+ */
+public class MigrationTool extends ToolBase {
+  
+  private S3Service s3Service;
+  private S3Bucket bucket;
+  
+  public static void main(String[] args) throws Exception {
+    int res = new MigrationTool().doMain(new Configuration(), args);
+    System.exit(res);
+  }
+  
+  public int run(String[] args) throws Exception {
+    
+    if (args.length == 0) {
+      System.err.println("Usage: MigrationTool <S3 file system URI>");
+      System.err.println("\t<S3 file system URI>\tfilesystem to migrate");
+      return -1;
+    }
+    
+    URI uri = URI.create(args[0]);
+    
+    initialize(uri, conf);
+    
+    FileSystemStore newStore = new Jets3tFileSystemStore();
+    newStore.initialize(uri, conf);
+    
+    if (get("%2F") != null) { 
+      System.err.println("Current version number is [unversioned].");
+      System.err.println("Target version number is " +
+          newStore.getVersion() + ".");
+      Store oldStore = new UnversionedStore();
+      migrate(oldStore, newStore);
+      return 0;
+    } else {
+      S3Object root = get("/");
+      if (root != null) {
+        String version = (String) root.getMetadata("fs-version");
+        if (version == null) {
+          System.err.println("Can't detect version - exiting.");
+        } else {
+          String newVersion = newStore.getVersion();
+          System.err.println("Current version number is " + version + ".");
+          System.err.println("Target version number is " + newVersion + ".");
+          if (version.equals(newStore.getVersion())) {
+            System.err.println("No migration required.");
+            return 0;
+          }
+          // use version number to create Store
+          //Store oldStore = ... 
+          //migrate(oldStore, newStore);
+          System.err.println("Not currently implemented.");
+          return 0;
+        }
+      }
+      System.err.println("Can't detect version - exiting.");
+      return 0;
+    }
+    
+  }
+  
+  public void initialize(URI uri, Configuration conf) throws IOException {
+    
+    this.conf = conf;
+    
+    try {
+      String accessKey = null;
+      String secretAccessKey = null;
+      String userInfo = uri.getUserInfo();
+      if (userInfo != null) {
+        int index = userInfo.indexOf(':');
+        if (index != -1) {
+          accessKey = userInfo.substring(0, index);
+          secretAccessKey = userInfo.substring(index + 1);
+        } else {
+          accessKey = userInfo;
+        }
+      }
+      if (accessKey == null) {
+        accessKey = conf.get("fs.s3.awsAccessKeyId");
+      }
+      if (secretAccessKey == null) {
+        secretAccessKey = conf.get("fs.s3.awsSecretAccessKey");
+      }
+      if (accessKey == null && secretAccessKey == null) {
+        throw new IllegalArgumentException("AWS " +
+                                           "Access Key ID and Secret Access Key " +
+                                           "must be specified as the username " +
+                                           "or password (respectively) of a s3 URL, " +
+                                           "or by setting the " +
+                                           "fs.s3.awsAccessKeyId or " +                         
+                                           "fs.s3.awsSecretAccessKey properties (respectively).");
+      } else if (accessKey == null) {
+        throw new IllegalArgumentException("AWS " +
+                                           "Access Key ID must be specified " +
+                                           "as the username of a s3 URL, or by setting the " +
+                                           "fs.s3.awsAccessKeyId property.");
+      } else if (secretAccessKey == null) {
+        throw new IllegalArgumentException("AWS " +
+                                           "Secret Access Key must be specified " +
+                                           "as the password of a s3 URL, or by setting the " +
+                                           "fs.s3.awsSecretAccessKey property.");         
+      }
+      AWSCredentials awsCredentials =
+        new AWSCredentials(accessKey, secretAccessKey);
+      this.s3Service = new RestS3Service(awsCredentials);
+    } catch (S3ServiceException e) {
+      if (e.getCause() instanceof IOException) {
+        throw (IOException) e.getCause();
+      }
+      throw new S3Exception(e);
+    }
+    bucket = new S3Bucket(uri.getHost());
+  }
+  
+  private void migrate(Store oldStore, FileSystemStore newStore)
+      throws IOException {
+    for (Path path : oldStore.listAllPaths()) {
+      INode inode = oldStore.retrieveINode(path);
+      oldStore.deleteINode(path);
+      newStore.storeINode(path, inode);
+    }
+  }
+  
+  private S3Object get(String key) {
+    try {
+      return s3Service.getObject(bucket, key);
+    } catch (S3ServiceException e) {
+      if (e.getS3ErrorCode().equals("NoSuchKey")) {
+        return null;
+      }
+    }
+    return null;
+  }
+  
+  interface Store {
+
+    Set<Path> listAllPaths() throws IOException;
+    INode retrieveINode(Path path) throws IOException;
+    void deleteINode(Path path) throws IOException;
+    
+  }
+  
+  class UnversionedStore implements Store {
+
+    public Set<Path> listAllPaths() throws IOException {
+      try {
+        String prefix = urlEncode(Path.SEPARATOR);
+        S3Object[] objects = s3Service.listObjects(bucket, prefix, null);
+        Set<Path> prefixes = new TreeSet<Path>();
+        for (int i = 0; i < objects.length; i++) {
+          prefixes.add(keyToPath(objects[i].getKey()));
+        }
+        return prefixes;
+      } catch (S3ServiceException e) {
+        if (e.getCause() instanceof IOException) {
+          throw (IOException) e.getCause();
+        }
+        throw new S3Exception(e);
+      }   
+    }
+
+    public void deleteINode(Path path) throws IOException {
+      delete(pathToKey(path));
+    }
+    
+    private void delete(String key) throws IOException {
+      try {
+        s3Service.deleteObject(bucket, key);
+      } catch (S3ServiceException e) {
+        if (e.getCause() instanceof IOException) {
+          throw (IOException) e.getCause();
+        }
+        throw new S3Exception(e);
+      }
+    }
+    
+    public INode retrieveINode(Path path) throws IOException {
+      return INode.deserialize(get(pathToKey(path)));
+    }
+
+    private InputStream get(String key) throws IOException {
+      try {
+        S3Object object = s3Service.getObject(bucket, key);
+        return object.getDataInputStream();
+      } catch (S3ServiceException e) {
+        if (e.getS3ErrorCode().equals("NoSuchKey")) {
+          return null;
+        }
+        if (e.getCause() instanceof IOException) {
+          throw (IOException) e.getCause();
+        }
+        throw new S3Exception(e);
+      }
+    }
+    
+    private String pathToKey(Path path) {
+      if (!path.isAbsolute()) {
+        throw new IllegalArgumentException("Path must be absolute: " + path);
+      }
+      return urlEncode(path.toUri().getPath());
+    }
+    
+    private Path keyToPath(String key) {
+      return new Path(urlDecode(key));
+    }
+
+    private String urlEncode(String s) {
+      try {
+        return URLEncoder.encode(s, "UTF-8");
+      } catch (UnsupportedEncodingException e) {
+        // Should never happen since every implementation of the Java Platform
+        // is required to support UTF-8.
+        // See http://java.sun.com/j2se/1.5.0/docs/api/java/nio/charset/Charset.html
+        throw new IllegalStateException(e);
+      }
+    }
+    
+    private String urlDecode(String s) {
+      try {
+        return URLDecoder.decode(s, "UTF-8");
+      } catch (UnsupportedEncodingException e) {
+        // Should never happen since every implementation of the Java Platform
+        // is required to support UTF-8.
+        // See http://java.sun.com/j2se/1.5.0/docs/api/java/nio/charset/Charset.html
+        throw new IllegalStateException(e);
+      }
+    }
+    
+  }
+  
+}

src/java/org/apache/hadoop/fs/s3/S3FileSystemException.java

@@ -0,0 +1,29 @@
+/**
+ * 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.s3;
+
+import java.io.IOException;
+
+/**
+ * Thrown when there is a fatal exception while using {@link S3FileSystem}.
+ */
+public class S3FileSystemException extends IOException {
+  public S3FileSystemException(String message) {
+    super(message);
+  }
+}

src/java/org/apache/hadoop/fs/s3/VersionMismatchException.java

@@ -0,0 +1,30 @@
+/**
+ * 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.s3;
+
+/**
+ * Thrown when Hadoop cannot read the version of the data stored
+ * in {@link S3FileSystem}.
+ */
+public class VersionMismatchException extends S3FileSystemException {
+  public VersionMismatchException(String clientVersion, String dataVersion) {
+    super("Version mismatch: client expects version " + clientVersion +
+        ", but data has version " +
+        (dataVersion == null ? "[unversioned]" : dataVersion));
+  }
+}

src/test/org/apache/hadoop/fs/s3/InMemoryFileSystemStore.java

@@ -33,6 +33,10 @@ class InMemoryFileSystemStore implements FileSystemStore {
   public void initialize(URI uri, Configuration conf) {
     this.conf = conf;
   }
+  
+  public String getVersion() throws IOException {
+    return "0";
+  }
 
   public void deleteINode(Path path) throws IOException {
     inodes.remove(path);