Use local credential to deploy beam pipelines

We are moving away from using Application Default Credentials generated by "gcloud auth application-default login" in our code base and consolidate on using self-managed credentials provided from AuthModule.

One of the remaining dependencies on the ADCs is from beam pipeline deployment commands, which by default use the ADCs to talk to GCS and upload the jar files and templates. In this CL, we explicitly provide the locally created credential to the Options used in deployments.

Also moved all credential qualifiers to CredentialModule, and removed @AppEngineAdminApiCredential, which is no longer used.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=224199812

java/google/registry/beam/invoicing/BUILD

@@ -15,6 +15,7 @@ java_library(
         "//java/google/registry/reporting/billing",
         "//java/google/registry/util",
         "@com_google_apis_google_api_services_bigquery",
+        "@com_google_auth_library_oauth2_http",
         "@com_google_auto_value",
         "@com_google_dagger",
         "@com_google_flogger",

java/google/registry/beam/invoicing/InvoicingPipeline.java

@@ -14,11 +14,17 @@
 
 package google.registry.beam.invoicing;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.auth.oauth2.GoogleCredentials;
 import google.registry.beam.invoicing.BillingEvent.InvoiceGroupingKey;
 import google.registry.beam.invoicing.BillingEvent.InvoiceGroupingKey.InvoiceGroupingKeyCoder;
+import google.registry.config.CredentialModule.LocalCredentialJson;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.reporting.billing.BillingModule;
 import google.registry.reporting.billing.GenerateInvoicesAction;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.io.Serializable;
 import javax.inject.Inject;
 import org.apache.beam.runners.dataflow.DataflowRunner;
@@ -79,6 +85,8 @@ public class InvoicingPipeline implements Serializable {
   @Config("invoiceFilePrefix")
   String invoiceFilePrefix;
 
+  @Inject @LocalCredentialJson String credentialJson;
+
   @Inject
   InvoicingPipeline() {}
 
@@ -100,6 +108,13 @@ public class InvoicingPipeline implements Serializable {
   public void deploy() {
     // We can't store options as a member variable due to serialization concerns.
     InvoicingPipelineOptions options = PipelineOptionsFactory.as(InvoicingPipelineOptions.class);
+    try {
+      options.setGcpCredential(
+          GoogleCredentials.fromStream(new ByteArrayInputStream(credentialJson.getBytes(UTF_8))));
+    } catch (IOException e) {
+      throw new RuntimeException(
+          "Cannot obtain local credential to deploy the invoicing pipeline", e);
+    }
     options.setProject(projectId);
     options.setRunner(DataflowRunner.class);
     // This causes p.run() to stage the pipeline as a template on GCS, as opposed to running it.

java/google/registry/beam/spec11/BUILD

@@ -12,6 +12,7 @@ java_library(
         "//java/google/registry/beam",
         "//java/google/registry/config",
         "//java/google/registry/util",
+        "@com_google_auth_library_oauth2_http",
         "@com_google_auto_value",
         "@com_google_dagger",
         "@com_google_flogger",

java/google/registry/beam/spec11/Spec11Pipeline.java

@@ -15,11 +15,16 @@
 package google.registry.beam.spec11;
 
 import static google.registry.beam.BeamUtils.getQueryFromFile;
+import static java.nio.charset.StandardCharsets.UTF_8;
 
+import com.google.auth.oauth2.GoogleCredentials;
 import google.registry.beam.spec11.SafeBrowsingTransforms.EvaluateSafeBrowsingFn;
+import google.registry.config.CredentialModule.LocalCredentialJson;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.util.Retrier;
 import google.registry.util.SqlTemplate;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.io.Serializable;
 import javax.inject.Inject;
 import org.apache.beam.runners.dataflow.DataflowRunner;
@@ -89,6 +94,8 @@ public class Spec11Pipeline implements Serializable {
   @Inject
   Retrier retrier;
 
+  @Inject @LocalCredentialJson String credentialJson;
+
   @Inject
   Spec11Pipeline() {}
 
@@ -123,6 +130,12 @@ public class Spec11Pipeline implements Serializable {
   public void deploy() {
     // We can't store options as a member variable due to serialization concerns.
     Spec11PipelineOptions options = PipelineOptionsFactory.as(Spec11PipelineOptions.class);
+    try {
+      options.setGcpCredential(
+          GoogleCredentials.fromStream(new ByteArrayInputStream(credentialJson.getBytes(UTF_8))));
+    } catch (IOException e) {
+      throw new RuntimeException("Cannot obtain local credential to deploy the spec11 pipeline", e);
+    }
     options.setProject(projectId);
     options.setRunner(DataflowRunner.class);
     // This causes p.run() to stage the pipeline as a template on GCS, as opposed to running it.