HDDS-3955. Unable to list intermediate paths on keys created using S3G.

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -2478,4 +2478,19 @@
       rules in Amazon S3's object key naming guide.
     </description>
   </property>
+
+  <property>
+    <name>ozone.om.enable.filesystem.paths</name>
+    <tag>OZONE, OM</tag>
+    <value>false</value>
+    <description>If true, key names will be interpreted as file system paths.
+      "/" will be treated as a special character and paths will be normalized
+      and must follow Unix filesystem path naming conventions. This flag will
+      be helpful when objects created by S3G need to be accessed using OFS/O3Fs.
+      If false, it will fallback to default behavior of Key/MPU create
+      requests where key paths are not normalized and any intermediate
+      directories will not be created or any file checks happens to check
+      filesystem semantics.
+    </description>
+  </property>
 </configuration>

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -239,4 +239,11 @@ private OMConfigKeys() {
           "ozone.om.keyname.character.check.enabled";
   public static final boolean OZONE_OM_KEYNAME_CHARACTER_CHECK_ENABLED_DEFAULT =
           false;
+
+  // This config needs to be enabled, when S3G created objects used via
+  // FileSystem API.
+  public static final String OZONE_OM_ENABLE_FILESYSTEM_PATHS =
+      "ozone.om.enable.filesystem.paths";
+  public static final boolean OZONE_OM_ENABLE_FILESYSTEM_PATHS_DEFAULT =
+      false;
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSWithObjectStoreCreate.java

@@ -0,0 +1,132 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.ozone;
+
+import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.TestDataUtil;
+import org.apache.hadoop.ozone.client.OzoneBucket;
+import org.apache.hadoop.ozone.client.OzoneVolume;
+import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+
+import java.net.URI;
+import java.util.Arrays;
+
+import static org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_SCHEME;
+
+/**
+ * Class tests create with object store and getFileStatus.
+ */
+public class TestOzoneFSWithObjectStoreCreate {
+
+  @Rule
+  public Timeout timeout = new Timeout(300000);
+
+  private String rootPath;
+
+  private MiniOzoneCluster cluster = null;
+
+  private OzoneFileSystem o3fs;
+
+  private String volumeName;
+
+  private String bucketName;
+
+
+  @Before
+  public void init() throws Exception {
+    volumeName = RandomStringUtils.randomAlphabetic(10).toLowerCase();
+    bucketName = RandomStringUtils.randomAlphabetic(10).toLowerCase();
+
+    OzoneConfiguration conf = new OzoneConfiguration();
+
+    conf.setBoolean(OMConfigKeys.OZONE_OM_ENABLE_FILESYSTEM_PATHS, true);
+    cluster = MiniOzoneCluster.newBuilder(conf)
+        .setNumDatanodes(3)
+        .build();
+    cluster.waitForClusterToBeReady();
+
+    // create a volume and a bucket to be used by OzoneFileSystem
+    OzoneBucket bucket =
+        TestDataUtil.createVolumeAndBucket(cluster, volumeName, bucketName);
+
+    rootPath = String.format("%s://%s.%s/", OZONE_URI_SCHEME, bucketName,
+        volumeName);
+    o3fs = (OzoneFileSystem) FileSystem.get(new URI(rootPath), conf);
+  }
+
+
+  @Test
+  public void test() throws Exception {
+
+    OzoneVolume ozoneVolume =
+        cluster.getRpcClient().getObjectStore().getVolume(volumeName);
+
+    OzoneBucket ozoneBucket = ozoneVolume.getBucket(bucketName);
+
+    String key1 = "///dir1/dir2/file1";
+    String key2 = "///dir1/dir2/file2";
+    int length = 10;
+    OzoneOutputStream ozoneOutputStream = ozoneBucket.createKey(key1, length);
+    byte[] b = new byte[10];
+    Arrays.fill(b, (byte)96);
+    ozoneOutputStream.write(b);
+    ozoneOutputStream.close();
+
+    ozoneOutputStream = ozoneBucket.createKey(key2, length);
+    ozoneOutputStream.write(b);
+    ozoneOutputStream.close();
+
+    // Adding "/" here otherwise Path will be considered as relative path and
+    // workingDir will be added.
+    key1 = "///dir1/dir2/file1";
+    Path p = new Path(key1);
+    Assert.assertTrue(o3fs.getFileStatus(p).isFile());
+
+    p = p.getParent();
+    checkAncestors(p);
+
+
+    key2 = "///dir1/dir2/file2";
+    p = new Path(key2);
+    Assert.assertTrue(o3fs.getFileStatus(p).isFile());
+    checkAncestors(p);
+
+  }
+
+  private void checkAncestors(Path p) throws Exception {
+    p = p.getParent();
+    while(p.getParent() != null) {
+      FileStatus fileStatus = o3fs.getFileStatus(p);
+      Assert.assertTrue(fileStatus.isDirectory());
+      p = p.getParent();
+    }
+  }
+
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -200,6 +200,8 @@
 import static org.apache.hadoop.ozone.OzoneConsts.OM_METRICS_TEMP_FILE;
 import static org.apache.hadoop.ozone.OzoneConsts.RPC_PORT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ENABLE_FILESYSTEM_PATHS;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ENABLE_FILESYSTEM_PATHS_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HANDLER_COUNT_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HANDLER_COUNT_KEY;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_KERBEROS_KEYTAB_FILE_KEY;
@@ -3494,4 +3496,10 @@ private Pair<String, String> resolveBucketLink(
   void setExitManagerForTesting(ExitManager exitManagerForTesting) {
     this.exitManager = exitManagerForTesting;
   }
+
+
+  public boolean getEnableFileSystemPaths() {
+    return configuration.getBoolean(OZONE_OM_ENABLE_FILESYSTEM_PATHS,
+        OZONE_OM_ENABLE_FILESYSTEM_PATHS_DEFAULT);
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/OMClientRequest.java

@@ -20,6 +20,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.ipc.ProtobufRpcEngine;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -38,20 +39,27 @@
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
 import java.io.IOException;
 import java.net.InetAddress;
+import java.nio.file.Paths;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
+import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_KEY_NAME;
 
 /**
  * OMClientRequest provides methods which every write OM request should
  * implement.
  */
 public abstract class OMClientRequest implements RequestAuditor {
 
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OMClientRequest.class);
   private OMRequest omRequest;
 
   /**
@@ -265,4 +273,72 @@ public Map<String, String> buildVolumeAuditMap(String volume) {
     auditMap.put(OzoneConsts.VOLUME, volume);
     return auditMap;
   }
+
+
+  public static String validateAndNormalizeKey(boolean enableFileSystemPaths,
+      String keyName) throws OMException {
+    if (enableFileSystemPaths) {
+      return validateAndNormalizeKey(keyName);
+    } else {
+      return keyName;
+    }
+  }
+
+  @SuppressFBWarnings("DMI_HARDCODED_ABSOLUTE_FILENAME")
+  public static String validateAndNormalizeKey(String keyName)
+      throws OMException {
+    String normalizedKeyName;
+    if (keyName.startsWith(OM_KEY_PREFIX)) {
+      normalizedKeyName = Paths.get(keyName).toUri().normalize().getPath();
+    } else {
+      normalizedKeyName = Paths.get(OM_KEY_PREFIX, keyName).toUri()
+          .normalize().getPath();
+    }
+    if (!keyName.equals(normalizedKeyName)) {
+      LOG.debug("Normalized key {} to {} ", keyName,
+          normalizedKeyName.substring(1));
+    }
+    return isValidKeyPath(normalizedKeyName.substring(1));
+  }
+
+  /**
+   * Whether the pathname is valid.  Check key names which contain a
+   * ":", ".", "..", "//", "". If it has any of these characters throws
+   * OMException, else return the path.
+   */
+  private static String isValidKeyPath(String path) throws OMException {
+    boolean isValid = true;
+
+    // If keyName is empty string throw error.
+    if (path.length() == 0) {
+      throw new OMException("Invalid KeyPath, empty keyName" + path,
+          INVALID_KEY_NAME);
+    } else if(path.startsWith("/")) {
+      isValid = false;
+    } else {
+      // Check for ".." "." ":" "/"
+      String[] components = StringUtils.split(path, '/');
+      for (int i = 0; i < components.length; i++) {
+        String element = components[i];
+        if (element.equals(".") ||
+            (element.contains(":")) ||
+            (element.contains("/") || element.equals(".."))) {
+          isValid = false;
+          break;
+        }
+
+        // The string may end with a /, but not have
+        // "//" in the middle.
+        if (element.isEmpty() && i != components.length - 1) {
+          isValid = false;
+        }
+      }
+    }
+
+    if (isValid) {
+      return path;
+    } else {
+      throw new OMException("Invalid KeyPath " + path, INVALID_KEY_NAME);
+    }
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileRequest.java

@@ -197,7 +197,7 @@ public boolean directParentExists() {
   /**
    * Return codes used by verifyFilesInPath method.
    */
-  enum OMDirectoryResult {
+  public enum OMDirectoryResult {
 
     // In below examples path is assumed as "a/b/c" in volume volume1 and
     // bucket b1.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java

@@ -110,9 +110,11 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
             ozoneManager.getPreallocateBlocksMax(),
             ozoneManager.isGrpcBlockTokenEnabled(), ozoneManager.getOMNodeId());
 
-    // Set modification time
+    // Set modification time and normalize key if required.
     KeyArgs.Builder newKeyArgs = keyArgs.toBuilder()
-        .setModificationTime(Time.now());
+        .setModificationTime(Time.now())
+        .setKeyName(validateAndNormalizeKey(
+            ozoneManager.getEnableFileSystemPaths(), keyArgs.getKeyName()));
 
     AllocateBlockRequest.Builder newAllocatedBlockRequest =
         AllocateBlockRequest.newBuilder()

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -91,7 +91,9 @@ public OMRequest preExecute(OzoneManager ozoneManager) throws IOException {
     }
 
     KeyArgs.Builder newKeyArgs =
-        keyArgs.toBuilder().setModificationTime(Time.now());
+        keyArgs.toBuilder().setModificationTime(Time.now())
+            .setKeyName(validateAndNormalizeKey(
+                ozoneManager.getEnableFileSystemPaths(), keyArgs.getKeyName()));
 
     return getOmRequest().toBuilder()
         .setCommitKeyRequest(commitKeyRequest.toBuilder()