HADOOP-19343. Add additional authentication support

Closes #7779

Co-authored-by: Chris Nauroth <cnauroth@apache.org>
Signed-off-by: Chris Nauroth <cnauroth@apache.org>

Author: arunkumarchacko

hadoop-tools/hadoop-gcp/dev-support/findbugs-exclude.xml

@@ -26,4 +26,9 @@
     <Method name="createItemInfoForBlob" />
     <Bug pattern="NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE" />
   </Match>
+    <Match>
+    <Class name="org.apache.hadoop.fs.gs.GoogleCloudStorageExceptions" />
+    <Method name="createCompositeException" />
+    <Bug pattern="NP_NULL_ON_SOME_PATH" />
+  </Match>
 </FindBugsFilter>

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GcsInstrumentation.java

@@ -0,0 +1,63 @@
+/*
+ * 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.gs;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.EnumSet;
+
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.statistics.impl.IOStatisticsStore;
+import org.apache.hadoop.fs.statistics.impl.IOStatisticsStoreBuilder;
+
+import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.iostatisticsStore;
+
+class GcsInstrumentation implements Closeable, IOStatisticsSource {
+  private final IOStatisticsStore instanceIOStatistics;
+
+  GcsInstrumentation() {
+    IOStatisticsStoreBuilder storeBuilder = iostatisticsStore();
+
+    // declare all counter statistics
+    EnumSet.allOf(GcsStatistics.class).stream()
+        .filter(statistic ->
+            statistic.getType() == StatisticTypeEnum.TYPE_COUNTER)
+        .forEach(stat -> {
+          storeBuilder.withCounters(stat.getSymbol());
+        });
+
+    EnumSet.allOf(GcsStatistics.class).stream()
+        .filter(statistic ->
+            statistic.getType() == StatisticTypeEnum.TYPE_DURATION)
+        .forEach(stat -> {
+          storeBuilder.withDurationTracking(stat.getSymbol());
+        });
+
+    this.instanceIOStatistics = storeBuilder.build();
+  }
+
+  @Override
+  public void close() throws IOException {
+  }
+
+  @Override
+  public IOStatisticsStore getIOStatistics() {
+    return instanceIOStatistics;
+  }
+}

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GcsListOperation.java

@@ -24,8 +24,6 @@
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.Storage;
 
-import static org.apache.hadoop.thirdparty.com.google.common.base.Preconditions.checkArgument;
-
 final class GcsListOperation {
   private static final int ALL = 0;
   private final Storage.BlobListOption[] listOptions;
@@ -72,7 +70,11 @@ Builder forRecursiveListing() {
     }
 
     GcsListOperation build() {
-      blobListOptions.add(Storage.BlobListOption.prefix(prefix));
+      // Can be null while listing the root directory.
+      if (prefix != null) {
+        blobListOptions.add(Storage.BlobListOption.prefix(prefix));
+      }
+
       return new GcsListOperation(this);
     }
 
@@ -82,13 +84,11 @@ Builder forCurrentDirectoryListing() {
       return this;
     }
 
-    Builder forCurrentDirectoryListingWithLimit(int theLimit) {
-      checkArgument(
-          theLimit > 0,
-          "limit should be greater than 0. found %d; prefix=%s", theLimit, prefix);
-
-      this.limit = theLimit;
-      prefix = StringPaths.toDirectoryPath(prefix);
+    Builder forImplicitDirectoryCheck() {
+      this.limit = 1;
+      if (prefix != null) {
+        prefix = StringPaths.toDirectoryPath(prefix);
+      }
 
       blobListOptions.add(Storage.BlobListOption.pageSize(1));
       forCurrentDirectoryListing();

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GcsStatistics.java

@@ -0,0 +1,72 @@
+/*
+ * 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.gs;
+
+import org.apache.hadoop.fs.statistics.StoreStatisticNames;
+
+import static org.apache.hadoop.fs.gs.StatisticTypeEnum.TYPE_DURATION;
+
+enum GcsStatistics {
+  INVOCATION_GET_FILE_STATUS(
+      StoreStatisticNames.OP_GET_FILE_STATUS,
+      "Calls of getFileStatus()",
+      TYPE_DURATION),
+  INVOCATION_CREATE(
+      StoreStatisticNames.OP_CREATE,
+      "Calls of create()",
+      TYPE_DURATION),
+  INVOCATION_DELETE(
+      StoreStatisticNames.OP_DELETE,
+      "Calls of delete()",
+      TYPE_DURATION),
+  INVOCATION_RENAME(
+      StoreStatisticNames.OP_RENAME,
+      "Calls of rename()",
+      TYPE_DURATION),
+  INVOCATION_OPEN(
+      StoreStatisticNames.OP_OPEN,
+      "Calls of open()",
+      TYPE_DURATION),
+  INVOCATION_MKDIRS(
+      StoreStatisticNames.OP_MKDIRS,
+      "Calls of mkdirs()",
+      TYPE_DURATION),
+  INVOCATION_LIST_STATUS(
+      StoreStatisticNames.OP_LIST_STATUS,
+      "Calls of listStatus()",
+      TYPE_DURATION);
+
+  private final String description;
+  private final StatisticTypeEnum type;
+  private final String symbol;
+
+  StatisticTypeEnum getType() {
+    return this.type;
+  }
+
+  String getSymbol() {
+    return this.symbol;
+  }
+
+  GcsStatistics(String symbol, String description, StatisticTypeEnum type) {
+    this.symbol = symbol;
+    this.description = description;
+    this.type = type;
+  }
+}

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GcsStorageStatistics.java

@@ -0,0 +1,52 @@
+/*
+ * 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.gs;
+
+import java.util.Iterator;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.statistics.IOStatistics;
+import org.apache.hadoop.fs.statistics.impl.StorageStatisticsFromIOStatistics;
+
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+class GcsStorageStatistics
+    extends StorageStatisticsFromIOStatistics {
+  static final String NAME = "GhfsStorageStatistics";
+
+  GcsStorageStatistics(final IOStatistics ioStatistics) {
+    super(NAME, Constants.SCHEME, ioStatistics);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    for (Iterator<LongStatistic> it = this.getLongStatistics(); it.hasNext();) {
+      LongStatistic statistic = it.next();
+
+      if (sb.length() != 0) {
+        sb.append(", ");
+      }
+      sb.append(String.format("%s=%s", statistic.getName(), statistic.getValue()));
+    }
+
+    return String.format("[%s]", sb);
+  }
+}

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GoogleCloudStorage.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.fs.gs;
 
+import static org.apache.hadoop.thirdparty.com.google.common.base.Preconditions.checkArgument;
 import static org.apache.hadoop.thirdparty.com.google.common.base.Preconditions.*;
 import static org.apache.hadoop.thirdparty.com.google.common.base.Strings.isNullOrEmpty;
 import static java.lang.Math.toIntExact;
@@ -27,7 +28,9 @@
 import com.google.api.client.util.ExponentialBackOff;
 import com.google.api.client.util.Sleeper;
 import com.google.api.gax.paging.Page;
+import com.google.auth.Credentials;
 import com.google.cloud.storage.*;
+import org.apache.hadoop.thirdparty.com.google.common.base.Strings;
 import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList;
 import org.apache.hadoop.thirdparty.com.google.common.collect.Maps;
 import org.apache.hadoop.thirdparty.com.google.common.io.BaseEncoding;
@@ -55,7 +58,7 @@
  * client</a>.
  */
 class GoogleCloudStorage {
-  static final Logger LOG = LoggerFactory.getLogger(GoogleHadoopFileSystem.class);
+  static final Logger LOG = LoggerFactory.getLogger(GoogleCloudStorage.class);
   static final List<Storage.BlobField> BLOB_FIELDS =
       ImmutableList.of(
           Storage.BlobField.BUCKET, Storage.BlobField.CONTENT_ENCODING,
@@ -76,18 +79,19 @@ class GoogleCloudStorage {
    * Having an instance of gscImpl to redirect calls to Json client while new client implementation
    * is in WIP.
    */
-  GoogleCloudStorage(GoogleHadoopFileSystemConfiguration configuration) throws IOException {
-    // TODO: Set credentials
-    this.storage = createStorage(configuration.getProjectId());
+  GoogleCloudStorage(GoogleHadoopFileSystemConfiguration configuration, Credentials credentials)
+          throws IOException {
+    this.storage = createStorage(configuration.getProjectId(), credentials);
     this.configuration = configuration;
   }
 
-  private static Storage createStorage(String projectId) {
+  private static Storage createStorage(String projectId, Credentials credentials) {
+    StorageOptions.Builder builder = StorageOptions.newBuilder();
     if (projectId != null) {
-      return StorageOptions.newBuilder().setProjectId(projectId).build().getService();
+      builder.setProjectId(projectId);
     }
 
-    return StorageOptions.newBuilder().build().getService();
+    return builder.setCredentials(credentials).build().getService();
   }
 
   WritableByteChannel create(final StorageResourceId resourceId, final CreateFileOptions options)
@@ -494,7 +498,9 @@ List<GoogleCloudStorageItemInfo> listDirectoryRecursive(String bucketName, Strin
     // TODO: Take delimiter from config
     // TODO: Set specific fields
 
-    checkArgument(objectName.endsWith("/"), String.format("%s should end with /", objectName));
+    checkArgument(
+            objectName == null || objectName.endsWith("/"),
+            String.format("%s should end with /", objectName));
     try {
       List<Blob> blobs = new GcsListOperation.Builder(bucketName, objectName, storage)
           .forRecursiveListing().build()
@@ -887,7 +893,7 @@ List<GoogleCloudStorageItemInfo> getItemInfos(List<StorageResourceId> resourceId
   List<GoogleCloudStorageItemInfo> listDirectory(String bucketName, String objectNamePrefix)
       throws IOException {
     checkArgument(
-        objectNamePrefix.endsWith("/"),
+        objectNamePrefix == null || objectNamePrefix.endsWith("/"),
         String.format("%s should end with /", objectNamePrefix));
 
     try {
@@ -971,7 +977,7 @@ GoogleCloudStorageItemInfo getFileOrDirectoryInfo(StorageResourceId resourceId)
   GoogleCloudStorageItemInfo getImplicitDirectory(StorageResourceId resourceId) {
     List<Blob> blobs = new GcsListOperation
         .Builder(resourceId.getBucketName(), resourceId.getObjectName(), storage)
-        .forCurrentDirectoryListingWithLimit(1).build()
+        .forImplicitDirectoryCheck().build()
         .execute();
 
     if (blobs.isEmpty()) {
@@ -981,6 +987,34 @@ GoogleCloudStorageItemInfo getImplicitDirectory(StorageResourceId resourceId) {
     return GoogleCloudStorageItemInfo.createInferredDirectory(resourceId.toDirectoryId());
   }
 
+  public void deleteBuckets(List<String> bucketNames) throws IOException {
+    LOG.trace("deleteBuckets({})", bucketNames);
+
+    // Validate all the inputs first.
+    for (String bucketName : bucketNames) {
+      checkArgument(!Strings.isNullOrEmpty(bucketName), "bucketName must not be null or empty");
+    }
+
+    // Gather exceptions to wrap in a composite exception at the end.
+    List<IOException> innerExceptions = new ArrayList<>();
+
+    for (String bucketName : bucketNames) {
+      try {
+        boolean isDeleted = storage.delete(bucketName);
+        if (!isDeleted) {
+          innerExceptions.add(createFileNotFoundException(bucketName, null, null));
+        }
+      } catch (StorageException e) {
+        innerExceptions.add(
+                new IOException(String.format("Error deleting '%s' bucket", bucketName), e));
+      }
+    }
+
+    if (!innerExceptions.isEmpty()) {
+      throw GoogleCloudStorageExceptions.createCompositeException(innerExceptions);
+    }
+  }
+
   // Helper class to capture the results of list operation.
   private class ListOperationResult {
     private final Map<String, Blob> prefixes = new HashMap<>();

hadoop-tools/hadoop-gcp/src/main/java/org/apache/hadoop/fs/gs/GoogleCloudStorageExceptions.java

@@ -18,12 +18,16 @@
 
 package org.apache.hadoop.fs.gs;
 
+import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
+
 import static org.apache.hadoop.thirdparty.com.google.common.base.Preconditions.checkArgument;
 import static org.apache.hadoop.thirdparty.com.google.common.base.Strings.isNullOrEmpty;
 import static org.apache.hadoop.thirdparty.com.google.common.base.Strings.nullToEmpty;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.util.Collection;
+import java.util.Iterator;
 import javax.annotation.Nullable;
 
 /**
@@ -55,4 +59,23 @@ static FileNotFoundException createFileNotFoundException(
     return createFileNotFoundException(
         resourceId.getBucketName(), resourceId.getObjectName(), cause);
   }
+
+  public static IOException createCompositeException(Collection<IOException> innerExceptions) {
+    Preconditions.checkArgument(
+            innerExceptions != null && !innerExceptions.isEmpty(),
+            "innerExceptions (%s) must be not null and contain at least one element",
+            innerExceptions);
+
+    Iterator<IOException> innerExceptionIterator = innerExceptions.iterator();
+
+    if (innerExceptions.size() == 1) {
+      return innerExceptionIterator.next();
+    }
+
+    IOException combined = new IOException("Multiple IOExceptions.");
+    while (innerExceptionIterator.hasNext()) {
+      combined.addSuppressed(innerExceptionIterator.next());
+    }
+    return combined;
+  }
 }