Enable Requester Pays support (#5140)

* Enable Requester Pays support

Google Cloud Storage has "requester pays" buckets. When reading from
those buckets, the requester is billed for the bandwidth (normally, it's
the bucket owner who is billed).

This pull request enables this feature with GATK. By default it is
turned off (so there are no unexpected charges). To turn it on, use the
command line argument: "--gcs-project-for-requester-pays"

Example:

$ ./gatk PrintReads --input $INPUT --output /tmp/reads.bam
fails with: com.google.cloud.storage.StorageException: Bucket is requester pays bucket but no user project provided.

$ ./gatk PrintReads --input $INPUT --output /tmp/reads.bam --gcs-project-for-requester-pays $PROJECT
works

This PR also removes the argumentless version of
setGlobalNIODefaultOptions() because it was confusing (it uses the
default values, not those indicated by the user - usually we'd expect
the user's values to be taken into account).

Author: jean-philippe-martin

src/main/java/org/broadinstitute/hellbender/cmdline/CommandLineProgram.java

@@ -1,6 +1,7 @@
 package org.broadinstitute.hellbender.cmdline;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
 import com.intel.gkl.compression.IntelDeflaterFactory;
 import com.intel.gkl.compression.IntelInflaterFactory;
 import htsjdk.samtools.Defaults;
@@ -82,6 +83,9 @@ public abstract class CommandLineProgram implements CommandLinePluginProvider {
     @Argument(fullName = StandardArgumentDefinitions.NIO_MAX_REOPENS_LONG_NAME, shortName = StandardArgumentDefinitions.NIO_MAX_REOPENS_SHORT_NAME, doc = "If the GCS bucket channel errors out, how many times it will attempt to re-initiate the connection", optional = true)
     public int NIO_MAX_REOPENS = ConfigFactory.getInstance().getGATKConfig().gcsMaxRetries();
 
+    @Argument(fullName = StandardArgumentDefinitions.NIO_PROJECT_FOR_REQUESTER_PAYS_LONG_NAME, doc = "Project to bill when accessing \"requester pays\" buckets. If unset, these buckets cannot be accessed.", optional = true)
+    public String NIO_PROJECT_FOR_REQUESTER_PAYS = ConfigFactory.getInstance().getGATKConfig().gcsProjectForRequesterPays();
+
     // This option is here for documentation completeness.
     // This is actually parsed out in Main to initialize configuration files because
     // we need to have the configuration completely set up before we create our CommandLinePrograms.
@@ -176,7 +180,7 @@ public Object instanceMainPostParseArgs() {
             BlockGunzipper.setDefaultInflaterFactory(new IntelInflaterFactory());
         }
 
-        BucketUtils.setGlobalNIODefaultOptions(NIO_MAX_REOPENS);
+        BucketUtils.setGlobalNIODefaultOptions(NIO_MAX_REOPENS, NIO_PROJECT_FOR_REQUESTER_PAYS);
 
         if (!QUIET) {
             printStartupMessage(startDateTime);
@@ -435,7 +439,12 @@ protected void printSettings() {
         final boolean usingIntelInflater = (BlockGunzipper.getDefaultInflaterFactory() instanceof IntelInflaterFactory && ((IntelInflaterFactory)BlockGunzipper.getDefaultInflaterFactory()).usingIntelInflater());
         logger.info("Inflater: " + (usingIntelInflater ? "IntelInflater": "JdkInflater"));
 
-        logger.info("GCS max retries/reopens: " + BucketUtils.getCloudStorageConfiguration(NIO_MAX_REOPENS).maxChannelReopens());
+        logger.info("GCS max retries/reopens: " + BucketUtils.getCloudStorageConfiguration(NIO_MAX_REOPENS, "").maxChannelReopens());
+        if (Strings.isNullOrEmpty(NIO_PROJECT_FOR_REQUESTER_PAYS)) {
+            logger.info("Requester pays: disabled");
+        } else {
+            logger.info("Requester pays: enabled. Billed to: " + NIO_PROJECT_FOR_REQUESTER_PAYS);
+        }
     }
 
     /**

src/main/java/org/broadinstitute/hellbender/cmdline/StandardArgumentDefinitions.java

@@ -92,4 +92,5 @@ private StandardArgumentDefinitions(){}
     public static final String USE_JDK_INFLATER_SHORT_NAME = "jdk-inflater";
     public static final String NIO_MAX_REOPENS_LONG_NAME = "gcs-max-retries";
     public static final String NIO_MAX_REOPENS_SHORT_NAME = "gcs-retries";
+    public static final String NIO_PROJECT_FOR_REQUESTER_PAYS_LONG_NAME = "gcs-project-for-requester-pays";
 }

src/main/java/org/broadinstitute/hellbender/utils/config/GATKConfig.java

@@ -157,6 +157,9 @@ public interface GATKConfig extends Mutable, Accessible {
     @DefaultValue("20")
     int gcsMaxRetries();
 
+    @DefaultValue("")
+    String gcsProjectForRequesterPays();
+
     @DefaultValue("true")
     boolean createOutputBamIndex();
 }

src/main/java/org/broadinstitute/hellbender/utils/gcs/BucketUtils.java

@@ -3,12 +3,13 @@
 import com.google.cloud.http.HttpTransportOptions;
 import com.google.cloud.storage.StorageOptions;
 import com.google.cloud.storage.contrib.nio.CloudStorageConfiguration;
+import com.google.cloud.storage.contrib.nio.CloudStorageConfiguration.Builder;
 import com.google.cloud.storage.contrib.nio.CloudStorageFileSystem;
 import com.google.cloud.storage.contrib.nio.CloudStorageFileSystemProvider;
+import com.google.common.base.Strings;
 import com.google.common.io.ByteStreams;
 import htsjdk.samtools.util.IOUtil;
 import htsjdk.samtools.util.RuntimeIOException;
-import htsjdk.tribble.AbstractFeatureReader;
 import htsjdk.tribble.Tribble;
 import htsjdk.tribble.util.TabixUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -19,7 +20,6 @@
 import org.apache.logging.log4j.Logger;
 import org.broadinstitute.hellbender.exceptions.UserException;
 import org.broadinstitute.hellbender.utils.Utils;
-import org.broadinstitute.hellbender.utils.config.ConfigFactory;
 import org.broadinstitute.hellbender.utils.io.IOUtils;
 import shaded.cloud_nio.com.google.api.gax.retrying.RetrySettings;
 import shaded.cloud_nio.com.google.auth.oauth2.GoogleCredentials;
@@ -343,20 +343,20 @@ public static String getBucket(String path) {
      */
     public static String getPathWithoutBucket(String path) {
         final String[] split = path.split("/");
-        final String BUCKET = split[2];
         return String.join("/", Arrays.copyOfRange(split, 3, split.length));
-
     }
 
     /**
-     * Sets NIO_MAX_REOPENS and generous timeouts as the global default.
+     * Sets max_reopens, requester_pays, and generous timeouts as the global default.
      * These will apply even to library code that creates its own paths to access with NIO.
+     *
+     * @param maxReopens If the GCS bucket channel errors out, how many times it will attempt to
+     *                   re-initiate the connection.
+     * @param requesterProject Project to bill when accessing "requester pays" buckets. If unset,
+     *                         these buckets cannot be accessed.
      */
-    public static void setGlobalNIODefaultOptions() {
-        setGlobalNIODefaultOptions(ConfigFactory.getInstance().getGATKConfig().gcsMaxRetries());
-    }
-    public static void setGlobalNIODefaultOptions(int maxReopens) {
-        CloudStorageFileSystemProvider.setDefaultCloudStorageConfiguration(getCloudStorageConfiguration(maxReopens));
+    public static void setGlobalNIODefaultOptions(int maxReopens, String requesterProject) {
+        CloudStorageFileSystemProvider.setDefaultCloudStorageConfiguration(getCloudStorageConfiguration(maxReopens, requesterProject));
         CloudStorageFileSystemProvider.setStorageOptions(setGenerousTimeouts(StorageOptions.newBuilder()).build());
     }
 
@@ -373,12 +373,24 @@ public static java.nio.file.Path getPathOnGcs(String gcsUrl) {
         return CloudStorageFileSystem.forBucket(BUCKET).getPath(pathWithoutBucket);
     }
 
-    /** The config we want to use. **/
-    public static CloudStorageConfiguration getCloudStorageConfiguration(int maxReopens) {
-        return CloudStorageConfiguration.builder()
+    /**
+     * The config we want to use.
+     *
+     * @param maxReopens If the GCS bucket channel errors out, how many times it will attempt to
+     *                   re-initiate the connection.
+     * @param requesterProject Project to bill when accessing "requester pays" buckets. If unset,
+     *                         these buckets cannot be accessed.
+     *
+     **/
+    public static CloudStorageConfiguration getCloudStorageConfiguration(int maxReopens, String requesterProject) {
+        Builder builder = CloudStorageConfiguration.builder()
             // if the channel errors out, re-open up to this many times
-            .maxChannelReopens(maxReopens)
-            .build();
+            .maxChannelReopens(maxReopens);
+        if (!Strings.isNullOrEmpty(requesterProject)) {
+            // enable requester pays and indicate who pays
+            builder = builder.autoDetectRequesterPays(true).userProject(requesterProject);
+        }
+        return builder.build();
     }
 
     private static StorageOptions.Builder setGenerousTimeouts(StorageOptions.Builder builder) {

src/test/java/org/broadinstitute/hellbender/utils/gcs/BucketUtilsTest.java

@@ -1,11 +1,13 @@
 package org.broadinstitute.hellbender.utils.gcs;
 
+import com.google.cloud.storage.contrib.nio.CloudStorageConfiguration;
 import htsjdk.samtools.util.IOUtil;
 import java.net.URI;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import org.broadinstitute.hellbender.GATKBaseTest;
 import org.broadinstitute.hellbender.testutils.MiniClusterUtils;
+import org.broadinstitute.hellbender.utils.config.ConfigFactory;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -17,7 +19,9 @@
 public final class BucketUtilsTest extends GATKBaseTest {
 
     static {
-        BucketUtils.setGlobalNIODefaultOptions();
+        BucketUtils.setGlobalNIODefaultOptions(
+            ConfigFactory.getInstance().getGATKConfig().gcsMaxRetries(),
+            ConfigFactory.getInstance().getGATKConfig().gcsProjectForRequesterPays());
     }
 
     @Test(groups={"bucket"})
@@ -36,6 +40,15 @@ public void testIsCloudStorageURL(){
         String x = "" + null + "://";
     }
 
+    @Test
+    public void testGetCloudStorageConfiguration() {
+        String mockProject = "yes";
+        int mockReopens = 100;
+        CloudStorageConfiguration config = BucketUtils.getCloudStorageConfiguration(mockReopens, mockProject);
+        Assert.assertEquals(config.maxChannelReopens(), mockReopens);
+        Assert.assertEquals(config.userProject(), mockProject);
+    }
+
     @Test
     public void testIsHadoopURL(){
         Assert.assertFalse(BucketUtils.isHadoopUrl("gs://abucket/bucket"));