Migrate the billing pipeline to flex template (#1100)

This is similar to the migration of the spec11 pipeline in #1073. Also removed a few Dagger providers that are no longer needed. TESTED=tested the dataflow job on alpha.  --- This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/google/nomulus/1100)
2026-09-12 11:06:29 +00:00 · 2021-04-22 10:26:15 -04:00
parent 09b6e300fc
commit 217b37b9d5
26 changed files with 554 additions and 807 deletions
@@ -260,6 +260,11 @@ public abstract class BillingEvent implements Serializable {
        poNumber());
  }

+  /** Returns the grouping key for this {@code BillingEvent}, to generate the detailed report. */
+  String getDetailedReportGroupingKey() {
+    return String.format("%s_%s", registrarId(), tld());
+  }
+
  /** Key for each {@code BillingEvent}, when aggregating for the overall invoice. */
  @AutoValue
  abstract static class InvoiceGroupingKey implements Serializable {
@@ -14,28 +14,28 @@

 package google.registry.beam.invoicing;

-import com.google.auth.oauth2.GoogleCredentials;
+import static google.registry.beam.BeamUtils.getQueryFromFile;
+import static org.apache.beam.sdk.values.TypeDescriptors.strings;
+
+import com.google.common.annotations.VisibleForTesting;
 import google.registry.beam.invoicing.BillingEvent.InvoiceGroupingKey;
 import google.registry.beam.invoicing.BillingEvent.InvoiceGroupingKey.InvoiceGroupingKeyCoder;
-import google.registry.config.CredentialModule.LocalCredential;
-import google.registry.config.RegistryConfig.Config;
 import google.registry.reporting.billing.BillingModule;
-import google.registry.reporting.billing.GenerateInvoicesAction;
-import google.registry.util.GoogleCredentialsBundle;
+import google.registry.util.SqlTemplate;
 import java.io.Serializable;
-import javax.inject.Inject;
-import org.apache.beam.runners.dataflow.DataflowRunner;
-import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.YearMonth;
+import java.time.format.DateTimeFormatter;
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.coders.SerializableCoder;
-import org.apache.beam.sdk.io.DefaultFilenamePolicy.Params;
-import org.apache.beam.sdk.io.FileBasedSink;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.FileIO;
 import org.apache.beam.sdk.io.TextIO;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
-import org.apache.beam.sdk.options.Description;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
-import org.apache.beam.sdk.options.ValueProvider;
-import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
+import org.apache.beam.sdk.transforms.Contextful;
 import org.apache.beam.sdk.transforms.Count;
 import org.apache.beam.sdk.transforms.Filter;
 import org.apache.beam.sdk.transforms.MapElements;
@@ -43,107 +43,54 @@ import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.TypeDescriptor;
-import org.apache.beam.sdk.values.TypeDescriptors;

 /**
- * Definition of a Dataflow pipeline template, which generates a given month's invoices.
+ * Definition of a Dataflow Flex pipeline template, which generates a given month's invoices.
 *
- * <p>To stage this template on GCS, run the {@link
- * google.registry.tools.DeployInvoicingPipelineCommand} Nomulus command.
+ * <p>To stage this template locally, run the {@code stage_beam_pipeline.sh} shell script.
 *
 * <p>Then, you can run the staged template via the API client library, gCloud or a raw REST call.
- * For an example using the API client library, see {@link GenerateInvoicesAction}.
 *
- * @see <a href="https://cloud.google.com/dataflow/docs/templates/overview">Dataflow Templates</a>
+ * @see <a href="https://cloud.google.com/dataflow/docs/guides/templates/using-flex-templates">Using
+ *     Flex Templates</a>
 */
 public class InvoicingPipeline implements Serializable {

-  private final String projectId;
-  private final String beamJobRegion;
-  private final String beamBucketUrl;
-  private final String invoiceTemplateUrl;
-  private final String beamStagingUrl;
-  private final String billingBucketUrl;
-  private final String invoiceFilePrefix;
-  private final GoogleCredentials googleCredentials;
+  private static final DateTimeFormatter TIMESTAMP_FORMATTER =
+      DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");

-  @Inject
-  public InvoicingPipeline(
-      @Config("projectId") String projectId,
-      @Config("defaultJobRegion") String beamJobRegion,
-      @Config("apacheBeamBucketUrl") String beamBucketUrl,
-      @Config("invoiceTemplateUrl") String invoiceTemplateUrl,
-      @Config("beamStagingUrl") String beamStagingUrl,
-      @Config("billingBucketUrl") String billingBucketUrl,
-      @Config("invoiceFilePrefix") String invoiceFilePrefix,
-      @LocalCredential GoogleCredentialsBundle googleCredentialsBundle) {
-    this.projectId = projectId;
-    this.beamJobRegion = beamJobRegion;
-    this.beamBucketUrl = beamBucketUrl;
-    this.invoiceTemplateUrl = invoiceTemplateUrl;
-    this.beamStagingUrl = beamStagingUrl;
-    this.billingBucketUrl = billingBucketUrl;
-    this.invoiceFilePrefix = invoiceFilePrefix;
-    this.googleCredentials = googleCredentialsBundle.getGoogleCredentials();
+  private final InvoicingPipelineOptions options;
+  private final Pipeline pipeline;
+
+  @VisibleForTesting
+  InvoicingPipeline(InvoicingPipelineOptions options, Pipeline pipeline) {
+    this.options = options;
+    this.pipeline = pipeline;
  }

-  /** Custom options for running the invoicing pipeline. */
-  public interface InvoicingPipelineOptions extends DataflowPipelineOptions {
-    /** Returns the yearMonth we're generating invoices for, in yyyy-MM format. */
-    @Description("The yearMonth we generate invoices for, in yyyy-MM format.")
-    ValueProvider<String> getYearMonth();
-    /**
-     * Sets the yearMonth we generate invoices for.
-     *
-     * <p>This is implicitly set when executing the Dataflow template, by specifying the 'yearMonth
-     * parameter.
-     */
-    void setYearMonth(ValueProvider<String> value);
+  InvoicingPipeline(InvoicingPipelineOptions options) {
+    this(options, Pipeline.create(options));
  }

-  /** Deploys the invoicing pipeline as a template on GCS, for a given projectID and GCS bucket. */
-  public void deploy() {
-    // We can't store options as a member variable due to serialization concerns.
-    InvoicingPipelineOptions options = PipelineOptionsFactory.as(InvoicingPipelineOptions.class);
-    options.setProject(projectId);
-    options.setRegion(beamJobRegion);
-    options.setRunner(DataflowRunner.class);
-    // This causes p.run() to stage the pipeline as a template on GCS, as opposed to running it.
-    options.setTemplateLocation(invoiceTemplateUrl);
-    options.setStagingLocation(beamStagingUrl);
-    // This credential is used when Dataflow deploys the template to GCS in target GCP project.
-    // So, make sure the credential has write permission to GCS in that project.
-    options.setGcpCredential(googleCredentials);
-
-    Pipeline p = Pipeline.create(options);
+  PipelineResult run() {
+    setupPipeline();
+    return pipeline.run();
+  }

+  void setupPipeline() {
    PCollection<BillingEvent> billingEvents =
-        p.apply(
+        pipeline.apply(
            "Read BillingEvents from Bigquery",
            BigQueryIO.read(BillingEvent::parseFromRecord)
-                .fromQuery(InvoicingUtils.makeQueryProvider(options.getYearMonth(), projectId))
+                .fromQuery(makeQuery(options.getYearMonth(), options.getProject()))
                .withCoder(SerializableCoder.of(BillingEvent.class))
                .usingStandardSql()
                .withoutValidation()
                .withTemplateCompatibility());
-    applyTerminalTransforms(billingEvents, options.getYearMonth());
-    p.run();
-  }

-  /**
-   * Applies output transforms to the {@code BillingEvent} source collection.
-   *
-   * <p>This is factored out purely to facilitate testing.
-   */
-  void applyTerminalTransforms(
-      PCollection<BillingEvent> billingEvents, ValueProvider<String> yearMonthProvider) {
-    billingEvents
-        .apply("Generate overall invoice rows", new GenerateInvoiceRows())
-        .apply("Write overall invoice to CSV", writeInvoice(yearMonthProvider));
+    saveInvoiceCsv(billingEvents, options);

-    billingEvents.apply(
-        "Write detail reports to separate CSVs keyed by registrarId_tld pair",
-        writeDetailReports(yearMonthProvider));
+    saveDetailedCsv(billingEvents, options);
  }

  /** Transform that converts a {@code BillingEvent} into an invoice CSV row. */
@@ -156,49 +103,85 @@ public class InvoicingPipeline implements Serializable {
              "Map to invoicing key",
              MapElements.into(TypeDescriptor.of(InvoiceGroupingKey.class))
                  .via(BillingEvent::getInvoiceGroupingKey))
-          .apply(Filter.by((InvoiceGroupingKey key) -> key.unitPrice() != 0))
+          .apply(
+              "Filter out free events", Filter.by((InvoiceGroupingKey key) -> key.unitPrice() != 0))
          .setCoder(new InvoiceGroupingKeyCoder())
          .apply("Count occurrences", Count.perElement())
          .apply(
              "Format as CSVs",
-              MapElements.into(TypeDescriptors.strings())
+              MapElements.into(strings())
                  .via((KV<InvoiceGroupingKey, Long> kv) -> kv.getKey().toCsv(kv.getValue())));
    }
  }

-  /** Returns an IO transform that writes the overall invoice to a single CSV file. */
-  private TextIO.Write writeInvoice(ValueProvider<String> yearMonthProvider) {
-    return TextIO.write()
-        .to(
-            NestedValueProvider.of(
-                yearMonthProvider,
-                yearMonth ->
+  /** Saves the billing events to a single overall invoice CSV file. */
+  static void saveInvoiceCsv(
+      PCollection<BillingEvent> billingEvents, InvoicingPipelineOptions options) {
+    billingEvents
+        .apply("Generate overall invoice rows", new GenerateInvoiceRows())
+        .apply(
+            "Write overall invoice to CSV",
+            TextIO.write()
+                .to(
                    String.format(
                        "%s/%s/%s/%s-%s",
-                        billingBucketUrl,
+                        options.getBillingBucketUrl(),
                        BillingModule.INVOICES_DIRECTORY,
-                        yearMonth,
-                        invoiceFilePrefix,
-                        yearMonth)))
-        .withHeader(InvoiceGroupingKey.invoiceHeader())
-        .withoutSharding()
-        .withSuffix(".csv");
+                        options.getYearMonth(),
+                        options.getInvoiceFilePrefix(),
+                        options.getYearMonth()))
+                .withHeader(InvoiceGroupingKey.invoiceHeader())
+                .withoutSharding()
+                .withSuffix(".csv"));
  }

-  /** Returns an IO transform that writes detail reports to registrar-tld keyed CSV files. */
-  private TextIO.TypedWrite<BillingEvent, Params> writeDetailReports(
-      ValueProvider<String> yearMonthProvider) {
-    return TextIO.<BillingEvent>writeCustomType()
-        .to(
-            InvoicingUtils.makeDestinationFunction(
-                String.format("%s/%s", billingBucketUrl, BillingModule.INVOICES_DIRECTORY),
-                yearMonthProvider),
-            InvoicingUtils.makeEmptyDestinationParams(billingBucketUrl + "/errors"))
-        .withFormatFunction(BillingEvent::toCsv)
-        .withoutSharding()
-        .withTempDirectory(
-            FileBasedSink.convertToFileResourceIfPossible(beamBucketUrl + "/temporary"))
-        .withHeader(BillingEvent.getHeader())
-        .withSuffix(".csv");
+  /** Saves the billing events to detailed report CSV files keyed by registrar-tld pairs. */
+  static void saveDetailedCsv(
+      PCollection<BillingEvent> billingEvents, InvoicingPipelineOptions options) {
+    String yearMonth = options.getYearMonth();
+    billingEvents.apply(
+        "Write detailed report for each registrar-tld pair",
+        FileIO.<String, BillingEvent>writeDynamic()
+            .to(
+                String.format(
+                    "%s/%s/%s",
+                    options.getBillingBucketUrl(), BillingModule.INVOICES_DIRECTORY, yearMonth))
+            .by(BillingEvent::getDetailedReportGroupingKey)
+            .withNumShards(1)
+            .withDestinationCoder(StringUtf8Coder.of())
+            .withNaming(
+                key ->
+                    (window, pane, numShards, shardIndex, compression) ->
+                        String.format(
+                            "%s_%s_%s.csv", BillingModule.DETAIL_REPORT_PREFIX, yearMonth, key))
+            .via(
+                Contextful.fn(BillingEvent::toCsv),
+                TextIO.sink().withHeader(BillingEvent.getHeader())));
+  }
+
+  /** Create the Bigquery query for a given project and yearMonth at runtime. */
+  static String makeQuery(String yearMonth, String projectId) {
+    // Get the timestamp endpoints capturing the entire month with microsecond precision
+    YearMonth reportingMonth = YearMonth.parse(yearMonth);
+    LocalDateTime firstMoment = reportingMonth.atDay(1).atTime(LocalTime.MIDNIGHT);
+    LocalDateTime lastMoment = reportingMonth.atEndOfMonth().atTime(LocalTime.MAX);
+    // Construct the month's query by filling in the billing_events.sql template
+    return SqlTemplate.create(getQueryFromFile(InvoicingPipeline.class, "billing_events.sql"))
+        .put("FIRST_TIMESTAMP_OF_MONTH", firstMoment.format(TIMESTAMP_FORMATTER))
+        .put("LAST_TIMESTAMP_OF_MONTH", lastMoment.format(TIMESTAMP_FORMATTER))
+        .put("PROJECT_ID", projectId)
+        .put("DATASTORE_EXPORT_DATA_SET", "latest_datastore_export")
+        .put("ONETIME_TABLE", "OneTime")
+        .put("REGISTRY_TABLE", "Registry")
+        .put("REGISTRAR_TABLE", "Registrar")
+        .put("CANCELLATION_TABLE", "Cancellation")
+        .build();
+  }
+
+  public static void main(String[] args) {
+    PipelineOptionsFactory.register(InvoicingPipelineOptions.class);
+    InvoicingPipelineOptions options =
+        PipelineOptionsFactory.fromArgs(args).withValidation().as(InvoicingPipelineOptions.class);
+    new InvoicingPipeline(options).run();
  }
 }
@@ -0,0 +1,37 @@
+// Copyright 2021 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed 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 google.registry.beam.invoicing;
+
+import google.registry.beam.common.RegistryPipelineOptions;
+import org.apache.beam.sdk.options.Description;
+
+/** Custom options for running the invoicing pipeline. */
+public interface InvoicingPipelineOptions extends RegistryPipelineOptions {
+
+  @Description("The year and month we generate invoices for, in yyyy-MM format.")
+  String getYearMonth();
+
+  void setYearMonth(String value);
+
+  @Description("Filename prefix for the invoice CSV file.")
+  String getInvoiceFilePrefix();
+
+  void setInvoiceFilePrefix(String value);
+
+  @Description("The GCS bucket URL for invoices and detailed  reports to be uploaded.")
+  String getBillingBucketUrl();
+
+  void setBillingBucketUrl(String value);
+}
@@ -1,106 +0,0 @@
-// Copyright 2018 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed 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 google.registry.beam.invoicing;
-
-import static google.registry.beam.BeamUtils.getQueryFromFile;
-
-import google.registry.util.SqlTemplate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.YearMonth;
-import java.time.format.DateTimeFormatter;
-import org.apache.beam.sdk.io.DefaultFilenamePolicy.Params;
-import org.apache.beam.sdk.io.FileBasedSink;
-import org.apache.beam.sdk.options.ValueProvider;
-import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-
-/** Pipeline helper functions used to generate invoices from instances of {@link BillingEvent}. */
-public class InvoicingUtils {
-
-  private InvoicingUtils() {}
-
-  private static final DateTimeFormatter TIMESTAMP_FORMATTER =
-      DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
-
-  /**
-   * Returns a function mapping from {@code BillingEvent} to filename {@code Params}.
-   *
-   * <p>Beam uses this to determine which file a given {@code BillingEvent} should get placed into.
-   *
-   * @param outputBucket the GCS bucket we're outputting reports to
-   * @param yearMonthProvider a runtime provider for the yyyy-MM we're generating the invoice for
-   */
-  static SerializableFunction<BillingEvent, Params> makeDestinationFunction(
-      String outputBucket, ValueProvider<String> yearMonthProvider) {
-    return billingEvent ->
-        new Params()
-            .withShardTemplate("")
-            .withSuffix(".csv")
-            .withBaseFilename(
-                NestedValueProvider.of(
-                    yearMonthProvider,
-                    yearMonth ->
-                        FileBasedSink.convertToFileResourceIfPossible(
-                            String.format(
-                                "%s/%s/%s",
-                                outputBucket, yearMonth, billingEvent.toFilename(yearMonth)))));
-  }
-
-  /**
-   * Returns the default filename parameters for an unmappable {@code BillingEvent}.
-   *
-   * <p>The "failed" file should only be populated when an error occurs, which warrants further
-   * investigation.
-   */
-  static Params makeEmptyDestinationParams(String outputBucket) {
-    return new Params()
-        .withBaseFilename(
-            FileBasedSink.convertToFileResourceIfPossible(
-                String.format("%s/%s", outputBucket, "FAILURES")));
-  }
-
-  /**
-   * Returns a provider that creates a Bigquery query for a given project and yearMonth at runtime.
-   *
-   * <p>We only know yearMonth at runtime, so this provider fills in the {@code
-   * sql/billing_events.sql} template at runtime.
-   *
-   * @param yearMonthProvider a runtime provider that returns which month we're invoicing for.
-   * @param projectId the projectId we're generating invoicing for.
-   */
-  static ValueProvider<String> makeQueryProvider(
-      ValueProvider<String> yearMonthProvider, String projectId) {
-    return NestedValueProvider.of(
-        yearMonthProvider,
-        (yearMonth) -> {
-          // Get the timestamp endpoints capturing the entire month with microsecond precision
-          YearMonth reportingMonth = YearMonth.parse(yearMonth);
-          LocalDateTime firstMoment = reportingMonth.atDay(1).atTime(LocalTime.MIDNIGHT);
-          LocalDateTime lastMoment = reportingMonth.atEndOfMonth().atTime(LocalTime.MAX);
-          // Construct the month's query by filling in the billing_events.sql template
-          return SqlTemplate.create(getQueryFromFile(InvoicingPipeline.class, "billing_events.sql"))
-              .put("FIRST_TIMESTAMP_OF_MONTH", firstMoment.format(TIMESTAMP_FORMATTER))
-              .put("LAST_TIMESTAMP_OF_MONTH", lastMoment.format(TIMESTAMP_FORMATTER))
-              .put("PROJECT_ID", projectId)
-              .put("DATASTORE_EXPORT_DATA_SET", "latest_datastore_export")
-              .put("ONETIME_TABLE", "OneTime")
-              .put("REGISTRY_TABLE", "Registry")
-              .put("REGISTRAR_TABLE", "Registrar")
-              .put("CANCELLATION_TABLE", "Cancellation")
-              .build();
-        });
-  }
-}
@@ -384,19 +384,6 @@ public final class RegistryConfig {
      return Duration.standardHours(1);
    }

-    /**
-     * Number of sharded entity group roots used for performing strongly consistent scans.
-     *
-     * <p><b>Warning:</b> This number may increase but never decrease.
-     *
-     * @see google.registry.model.index.EppResourceIndex
-     */
-    @Provides
-    @Config("eppResourceIndexBucketCount")
-    public static int provideEppResourceIndexBucketCount(RegistryConfigSettings config) {
-      return config.datastore.eppResourceIndexBucketsNum;
-    }
-
    @Provides
    @Config("cloudSqlJdbcUrl")
    public static String providesCloudSqlJdbcUrl(RegistryConfigSettings config) {
@@ -564,53 +551,6 @@ public final class RegistryConfig {
      return config.gSuite.outgoingEmailDisplayName;
    }

-    /**
-     * Returns the name of the GCS bucket for storing Beam templates and results.
-     *
-     * @see google.registry.reporting.billing.GenerateInvoicesAction
-     */
-    @Provides
-    @Config("apacheBeamBucket")
-    public static String provideApacheBeamBucket(@Config("projectId") String projectId) {
-      return projectId + "-beam";
-    }
-
-    /**
-     * Returns the URL of the GCS location for storing Apache Beam related objects.
-     *
-     * @see google.registry.reporting.billing.GenerateInvoicesAction
-     */
-    @Provides
-    @Config("apacheBeamBucketUrl")
-    public static String provideApacheBeamBucketUrl(@Config("apacheBeamBucket") String beamBucket) {
-      return "gs://" + beamBucket;
-    }
-
-    /**
-     * Returns the URL of the GCS location for storing the monthly invoicing Beam template.
-     *
-     * @see google.registry.reporting.billing.GenerateInvoicesAction
-     * @see google.registry.beam.invoicing.InvoicingPipeline
-     */
-    @Provides
-    @Config("invoiceTemplateUrl")
-    public static String provideInvoiceTemplateUrl(
-        @Config("apacheBeamBucketUrl") String beamBucketUrl) {
-      return beamBucketUrl + "/templates/invoicing";
-    }
-
-    /**
-     * Returns the URL of the GCS location for storing the monthly spec11 Beam template.
-     *
-     * @see google.registry.beam.spec11.Spec11Pipeline
-     */
-    @Provides
-    @Config("spec11TemplateUrl")
-    public static String provideSpec11TemplateUrl(
-        @Config("apacheBeamBucketUrl") String beamBucketUrl) {
-      return beamBucketUrl + "/templates/spec11";
-    }
-
    /**
     * Returns whether an SSL certificate hash is required to log in via EPP and run flows.
     *
@@ -634,18 +574,6 @@ public final class RegistryConfig {
      return config.beam.defaultJobRegion;
    }

-    /**
-     * Returns the default job zone to run Apache Beam (Cloud Dataflow) jobs in.
-     *
-     * @see google.registry.reporting.billing.GenerateInvoicesAction
-     * @see google.registry.reporting.spec11.GenerateSpec11ReportAction
-     */
-    @Provides
-    @Config("defaultJobZone")
-    public static String provideDefaultJobZone(RegistryConfigSettings config) {
-      return config.beam.defaultJobZone;
-    }
-
    /** Returns the GCS bucket URL with all staged BEAM flex templates. */
    @Provides
    @Config("beamStagingBucketUrl")
@@ -653,19 +581,6 @@ public final class RegistryConfig {
      return config.beam.stagingBucketUrl;
    }

-    /**
-     * Returns the URL of the GCS location we store jar dependencies for beam pipelines.
-     *
-     * @see google.registry.beam.invoicing.InvoicingPipeline
-     * @see google.registry.beam.spec11.Spec11Pipeline
-     */
-    @Provides
-    @Config("beamStagingUrl")
-    public static String provideInvoiceStagingUrl(
-        @Config("apacheBeamBucketUrl") String beamBucketUrl) {
-      return beamBucketUrl + "/staging";
-    }
-
    /**
     * Returns the Google Cloud Storage bucket for Spec11 and ICANN transaction and activity reports
     * to be uploaded.
@@ -1227,14 +1142,6 @@ public final class RegistryConfig {
      return formatComments(config.registryPolicy.reservedTermsExportDisclaimer);
    }

-    /** Returns the clientId of the registrar used by the {@code CheckApiServlet}. */
-    // TODO(b/80417678): remove this once CheckApiAction no longer uses this id.
-    @Provides
-    @Config("checkApiServletRegistrarClientId")
-    public static String provideCheckApiServletRegistrarClientId(RegistryConfigSettings config) {
-      return config.registryPolicy.checkApiServletClientId;
-    }
-
    /**
     * Returns the clientId of the registrar that admins are automatically logged in as if they
     * aren't otherwise associated with one.
@@ -133,7 +133,6 @@ public class RegistryConfigSettings {
  /** Configuration for Apache Beam (Cloud Dataflow). */
  public static class Beam {
    public String defaultJobRegion;
-    public String defaultJobZone;
    public String stagingBucketUrl;
  }

@@ -420,9 +420,6 @@ misc:
 beam:
  # The default region to run Apache Beam (Cloud Dataflow) jobs in.
  defaultJobRegion: us-east1
-  # The default zone to run Apache Beam (Cloud Dataflow) jobs in.
-  # TODO(weiminyu): consider dropping zone config. No obvious needs for this.
-  defaultJobZone: us-east1-c
  stagingBucketUrl: gcs-bucket-with-staged-templates

 keyring:
@@ -14,6 +14,7 @@

 package google.registry.reporting.billing;

+import static google.registry.beam.BeamUtils.createJobName;
 import static google.registry.reporting.ReportingUtils.enqueueBeamReportingTask;
 import static google.registry.reporting.billing.BillingModule.PARAM_SHOULD_PUBLISH;
 import static google.registry.request.Action.Method.POST;
@@ -21,9 +22,9 @@ import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_OK;

 import com.google.api.services.dataflow.Dataflow;
-import com.google.api.services.dataflow.model.LaunchTemplateParameters;
-import com.google.api.services.dataflow.model.LaunchTemplateResponse;
-import com.google.api.services.dataflow.model.RuntimeEnvironment;
+import com.google.api.services.dataflow.model.LaunchFlexTemplateParameter;
+import com.google.api.services.dataflow.model.LaunchFlexTemplateRequest;
+import com.google.api.services.dataflow.model.LaunchFlexTemplateResponse;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
@@ -33,6 +34,7 @@ import google.registry.request.Action;
 import google.registry.request.Parameter;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
+import google.registry.util.Clock;
 import java.io.IOException;
 import java.util.Map;
 import javax.inject.Inject;
@@ -42,9 +44,8 @@ import org.joda.time.YearMonth;
 * Invokes the {@code InvoicingPipeline} beam template via the REST api, and enqueues the {@link
 * PublishInvoicesAction} to publish the subsequent output.
 *
- * <p>This action runs the {@link google.registry.beam.invoicing.InvoicingPipeline} beam template,
- * staged at gs://&lt;projectId&gt;-beam/templates/invoicing. The pipeline then generates invoices
- * for the month and stores them on GCS.
+ * <p>This action runs the {@link google.registry.beam.invoicing.InvoicingPipeline} beam flex
+ * template. The pipeline then generates invoices for the month and stores them on GCS.
 */
@Action(
    service = Action.Service.BACKEND,
@@ -56,57 +57,73 @@ public class GenerateInvoicesAction implements Runnable {
  private static final FluentLogger logger = FluentLogger.forEnclosingClass();

  static final String PATH = "/_dr/task/generateInvoices";
+  static final String PIPELINE_NAME = "invoicing_pipeline";

  private final String projectId;
-  private final String beamBucketUrl;
-  private final String invoiceTemplateUrl;
-  private final String jobZone;
+  private final String jobRegion;
+  private final String stagingBucketUrl;
+  private final String billingBucketUrl;
+  private final String invoiceFilePrefix;
  private final boolean shouldPublish;
  private final YearMonth yearMonth;
-  private final Dataflow dataflow;
-  private final Response response;
  private final BillingEmailUtils emailUtils;
+  private final Clock clock;
+  private final Response response;
+  private final Dataflow dataflow;

  @Inject
  GenerateInvoicesAction(
      @Config("projectId") String projectId,
-      @Config("apacheBeamBucketUrl") String beamBucketUrl,
-      @Config("invoiceTemplateUrl") String invoiceTemplateUrl,
-      @Config("defaultJobZone") String jobZone,
+      @Config("defaultJobRegion") String jobRegion,
+      @Config("beamStagingBucketUrl") String stagingBucketUrl,
+      @Config("billingBucketUrl") String billingBucketUrl,
+      @Config("invoiceFilePrefix") String invoiceFilePrefix,
      @Parameter(PARAM_SHOULD_PUBLISH) boolean shouldPublish,
      YearMonth yearMonth,
-      Dataflow dataflow,
+      BillingEmailUtils emailUtils,
+      Clock clock,
      Response response,
-      BillingEmailUtils emailUtils) {
+      Dataflow dataflow) {
    this.projectId = projectId;
-    this.beamBucketUrl = beamBucketUrl;
-    this.invoiceTemplateUrl = invoiceTemplateUrl;
-    this.jobZone = jobZone;
+    this.jobRegion = jobRegion;
+    this.stagingBucketUrl = stagingBucketUrl;
+    this.billingBucketUrl = billingBucketUrl;
+    this.invoiceFilePrefix = invoiceFilePrefix;
    this.shouldPublish = shouldPublish;
    this.yearMonth = yearMonth;
-    this.dataflow = dataflow;
-    this.response = response;
    this.emailUtils = emailUtils;
+    this.clock = clock;
+    this.response = response;
+    this.dataflow = dataflow;
  }

  @Override
  public void run() {
+    response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
    logger.atInfo().log("Launching invoicing pipeline for %s", yearMonth);
    try {
-      LaunchTemplateParameters params =
-          new LaunchTemplateParameters()
-              .setJobName(String.format("invoicing-%s", yearMonth))
-              .setEnvironment(
-                  new RuntimeEnvironment()
-                      .setZone(jobZone)
-                      .setTempLocation(beamBucketUrl + "/temporary"))
-              .setParameters(ImmutableMap.of("yearMonth", yearMonth.toString("yyyy-MM")));
-      LaunchTemplateResponse launchResponse =
+      LaunchFlexTemplateParameter parameter =
+          new LaunchFlexTemplateParameter()
+              .setJobName(createJobName("invoicing", clock))
+              .setContainerSpecGcsPath(
+                  String.format("%s/%s_metadata.json", stagingBucketUrl, PIPELINE_NAME))
+              .setParameters(
+                  ImmutableMap.of(
+                      "yearMonth",
+                      yearMonth.toString("yyyy-MM"),
+                      "invoiceFilePrefix",
+                      invoiceFilePrefix,
+                      "billingBucketUrl",
+                      billingBucketUrl));
+      LaunchFlexTemplateResponse launchResponse =
          dataflow
              .projects()
-              .templates()
-              .launch(projectId, params)
-              .setGcsPath(invoiceTemplateUrl)
+              .locations()
+              .flexTemplates()
+              .launch(
+                  projectId,
+                  jobRegion,
+                  new LaunchFlexTemplateRequest().setLaunchParameter(parameter))
              .execute();
      logger.atInfo().log("Got response: %s", launchResponse.getJob().toPrettyString());
      String jobId = launchResponse.getJob().getId();
@@ -123,12 +140,10 @@ public class GenerateInvoicesAction implements Runnable {
      logger.atWarning().withCause(e).log("Template Launch failed");
      emailUtils.sendAlertEmail(String.format("Template Launch failed due to %s", e.getMessage()));
      response.setStatus(SC_INTERNAL_SERVER_ERROR);
-      response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
      response.setPayload(String.format("Template launch failed: %s", e.getMessage()));
      return;
    }
    response.setStatus(SC_OK);
-    response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
    response.setPayload("Launched dataflow template.");
  }
 }
@@ -59,6 +59,7 @@ public class PublishInvoicesAction implements Runnable {
  private static final String JOB_FAILED = "JOB_STATE_FAILED";

  private final String projectId;
+  private final String jobRegion;
  private final String jobId;
  private final BillingEmailUtils emailUtils;
  private final Dataflow dataflow;
@@ -68,12 +69,14 @@ public class PublishInvoicesAction implements Runnable {
  @Inject
  PublishInvoicesAction(
      @Config("projectId") String projectId,
+      @Config("defaultJobRegion") String jobRegion,
      @Parameter(ReportingModule.PARAM_JOB_ID) String jobId,
      BillingEmailUtils emailUtils,
      Dataflow dataflow,
      Response response,
      YearMonth yearMonth) {
    this.projectId = projectId;
+    this.jobRegion = jobRegion;
    this.jobId = jobId;
    this.emailUtils = emailUtils;
    this.dataflow = dataflow;
@@ -87,7 +90,7 @@ public class PublishInvoicesAction implements Runnable {
  public void run() {
    try {
      logger.atInfo().log("Starting publish job.");
-      Job job = dataflow.projects().jobs().get(projectId, jobId).execute();
+      Job job = dataflow.projects().locations().jobs().get(projectId, jobRegion, jobId).execute();
      String state = job.getCurrentState();
      switch (state) {
        case JOB_DONE:
@@ -1,31 +0,0 @@
-// Copyright 2017 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed 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 google.registry.tools;
-
-import com.beust.jcommander.Parameters;
-import google.registry.beam.invoicing.InvoicingPipeline;
-import javax.inject.Inject;
-
-/** Nomulus command that deploys the {@link InvoicingPipeline} template. */
-@Parameters(commandDescription = "Deploy the invoicing pipeline to GCS.")
-public class DeployInvoicingPipelineCommand implements Command {
-
-  @Inject InvoicingPipeline invoicingPipeline;
-
-  @Override
-  public void run() {
-    invoicingPipeline.deploy();
-  }
-}
@@ -62,7 +62,6 @@ public final class RegistryTool {
          .put("delete_premium_list", DeletePremiumListCommand.class)
          .put("delete_reserved_list", DeleteReservedListCommand.class)
          .put("delete_tld", DeleteTldCommand.class)
-          .put("deploy_invoicing_pipeline", DeployInvoicingPipelineCommand.class)
          .put("encrypt_escrow_deposit", EncryptEscrowDepositCommand.class)
          .put("execute_epp", ExecuteEppCommand.class)
          .put("generate_allocation_tokens", GenerateAllocationTokensCommand.class)
@@ -105,8 +105,6 @@ interface RegistryToolComponent {

  void inject(DeleteContactByRoidCommand command);

-  void inject(DeployInvoicingPipelineCommand command);
-
  void inject(EncryptEscrowDepositCommand command);

  void inject(GenerateAllocationTokensCommand command);
@@ -5,7 +5,7 @@
    {
      "name": "registryEnvironment",
      "label": "The Registry environment.",
-      "helpText": "The Registry environment, required if environment-specific initialization is needed on worker VMs.",
+      "helpText": "The Registry environment, required if environment-specific initialization (such as JPA) is needed on worker VMs.",
      "is_optional": true,
      "regexes": [
        "^[0-9A-Z_]+$"
@@ -0,0 +1,66 @@
+{
+  "name": "Invoice and Detailed Reports Generation",
+  "description": "An Apache Beam batch pipeline that reads from a Datastore export and generate monthly invoice and detailed reports, saving them on GCS.",
+  "parameters": [
+    {
+      "name": "registryEnvironment",
+      "label": "The Registry environment.",
+      "helpText": "The Registry environment, required if environment-specific initialization (such as JPA) is needed on worker VMs.",
+      "is_optional": true,
+      "regexes": [
+        "^[0-9A-Z_]+$"
+      ]
+    },
+    {
+      "name": "isolationOverride",
+      "label": "The desired SQL transaction isolation level.",
+      "helpText": "The desired SQL transaction isolation level.",
+      "is_optional": true,
+      "regexes": [
+        "^[0-9A-Z_]+$"
+      ]
+    },
+    {
+      "name": "sqlWriteBatchSize",
+      "label": "SQL write batch size.",
+      "helpText": "The number of entities to write to the SQL database in one operation.",
+      "is_optional": true,
+      "regexes": [
+        "^[1-9][0-9]*$"
+      ]
+    },
+    {
+      "name": "sqlWriteShards",
+      "label": "Number of output shards to create when writing to SQL.",
+      "helpText": "Number of shards to create out of the data before writing to the SQL database. Please refer to the Javadoc of RegistryJpaIO.Write.shards() for how to choose this value.",
+      "is_optional": true,
+      "regexes": [
+        "^[1-9][0-9]*$"
+      ]
+    },
+    {
+      "name": "yearMonth",
+      "label": "The year and month we generate invoice and detailed reports for.",
+      "helpText": "The year and month for which the invoice and detailed reports are generated, in yyyy-MM format.",
+      "regexes": [
+        "^2[0-9]{3}-(0[1-9]|1[0-2])$"
+      ]
+    },
+    {
+      "name": "invoiceFilePrefix",
+      "label": "Filename prefix for the invoice CSV file.",
+      "helpText": "The prefix that will be applied to the invoice file.",
+      "regexes": [
+        "^[a-zA-Z0-9_\\-]+$"
+      ]
+    },
+    {
+      "name": "billingBucketUrl",
+      "label": "invoice and detailed reports upload dir.",
+      "helpText": "The root directory of the reports to upload.",
+      "regexes": [
+        "^gs:\\/\\/[^\\n\\r]+$"
+      ]
+    }
+  ]
+}
@@ -5,7 +5,7 @@
    {
      "name": "registryEnvironment",
      "label": "The Registry environment.",
-      "helpText": "The Registry environment, required if environment-specific initialization is needed on worker VMs.",
+      "helpText": "The Registry environment, required if environment-specific initialization (such as JPA) is needed on worker VMs.",
      "is_optional": true,
      "regexes": [
        "^[0-9A-Z_]+$"
@@ -41,7 +41,7 @@
    {
      "name": "date",
      "label": "The date when the pipeline runs",
-      "helpText": "The date then the threat scan is performed, in yyyy-MM-dd format.",
+      "helpText": "The date when the threat scan is performed, in yyyy-MM-dd format.",
      "regexes": [
        "^2[0-9]{3}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])$"
      ]