Refactor ICANN reporting and billing into common package

This moves the default yearMonth logic into a common ReportingModule, rather than the coarse-scoped BackendModule, which may not want the default parameter extraction logic, as well as moving the 'yearMonth' parameter constant to the common package it's used in. This also provides a basis for future consolidation of the ReportingEmailUtils and BillingEmailUtils classes, which have modest overlap.

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

java/google/registry/reporting/BUILD

@@ -7,27 +7,11 @@ licenses(["notice"])  # Apache 2.0
 java_library(
     name = "reporting",
     srcs = glob(["*.java"]),
-    resources = glob(["sql/*"]),
     deps = [
-        "//java/google/registry/bigquery",
-        "//java/google/registry/config",
-        "//java/google/registry/gcs",
-        "//java/google/registry/keyring/api",
-        "//java/google/registry/model",
         "//java/google/registry/request",
-        "//java/google/registry/request/auth",
         "//java/google/registry/util",
-        "//java/google/registry/xjc",
-        "//java/google/registry/xml",
-        "@com_google_api_client",
-        "@com_google_apis_google_api_services_bigquery",
-        "@com_google_appengine_api_1_0_sdk",
-        "@com_google_appengine_tools_appengine_gcs_client",
         "@com_google_code_findbugs_jsr305",
         "@com_google_dagger",
-        "@com_google_guava",
-        "@com_google_http_client",
-        "@com_google_http_client_jackson2",
         "@javax_servlet_api",
         "@joda_time",
     ],

java/google/registry/reporting/ReportingModule.java

@@ -0,0 +1,67 @@
+// 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.reporting;
+
+import static google.registry.request.RequestParameters.extractOptionalParameter;
+
+import dagger.Module;
+import dagger.Provides;
+import google.registry.request.HttpException.BadRequestException;
+import google.registry.request.Parameter;
+import google.registry.util.Clock;
+import java.util.Optional;
+import javax.servlet.http.HttpServletRequest;
+import org.joda.time.YearMonth;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+
+/**
+ * Dagger module for injecting common settings for all Backend tasks.
+ */
+@Module
+public class ReportingModule {
+
+  /**
+   * The request parameter name used by reporting actions that takes a year/month parameter, which
+   * defaults to the last month.
+   */
+  public static final String PARAM_YEAR_MONTH = "yearMonth";
+
+  /** Extracts an optional YearMonth in yyyy-MM format from the request. */
+  @Provides
+  @Parameter(PARAM_YEAR_MONTH)
+  static Optional<YearMonth> provideYearMonthOptional(HttpServletRequest req) {
+    DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-MM");
+    Optional<String> optionalYearMonthStr = extractOptionalParameter(req, PARAM_YEAR_MONTH);
+    try {
+      return optionalYearMonthStr.map(s -> YearMonth.parse(s, formatter));
+    } catch (IllegalArgumentException e) {
+      throw new BadRequestException(
+          String.format(
+              "yearMonth must be in yyyy-MM format, got %s instead",
+              optionalYearMonthStr.orElse("UNSPECIFIED YEARMONTH")));
+    }
+  }
+
+  /**
+   * Provides the yearMonth in yyyy-MM format, if not specified in the request, defaults to one
+   * month prior to run time.
+   */
+  @Provides
+  static YearMonth provideYearMonth(
+      @Parameter(PARAM_YEAR_MONTH) Optional<YearMonth> yearMonthOptional, Clock clock) {
+    return yearMonthOptional.orElseGet(() -> new YearMonth(clock.nowUtc().minusMonths(1)));
+  }
+}

java/google/registry/reporting/billing/BUILD

@@ -0,0 +1,36 @@
+package(
+    default_visibility = ["//visibility:public"],
+)
+
+licenses(["notice"])  # Apache 2.0
+
+java_library(
+    name = "billing",
+    srcs = glob(["*.java"]),
+    runtime_deps = [
+        "@com_google_apis_google_api_services_bigquery",
+    ],
+    deps = [
+        "//java/google/registry/config",
+        "//java/google/registry/gcs",
+        "//java/google/registry/model",
+        "//java/google/registry/reporting",
+        "//java/google/registry/request",
+        "//java/google/registry/request/auth",
+        "//java/google/registry/storage/drive",
+        "//java/google/registry/util",
+        "@com_google_api_client_appengine",
+        "@com_google_apis_google_api_services_dataflow",
+        "@com_google_appengine_api_1_0_sdk",
+        "@com_google_appengine_tools_appengine_gcs_client",
+        "@com_google_dagger",
+        "@com_google_guava",
+        "@com_google_http_client",
+        "@javax_servlet_api",
+        "@joda_time",
+        "@org_apache_beam_runners_direct_java",
+        "@org_apache_beam_runners_google_cloud_dataflow_java",
+        "@org_apache_beam_sdks_java_core",
+        "@org_apache_beam_sdks_java_io_google_cloud_platform",
+    ],
+)

java/google/registry/reporting/billing/BillingEmailUtils.java

@@ -0,0 +1,148 @@
+// 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.reporting.billing;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.appengine.tools.cloudstorage.GcsFilename;
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.CharStreams;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.gcs.GcsUtils;
+import google.registry.reporting.billing.BillingModule.InvoiceDirectoryPrefix;
+import google.registry.util.FormattingLogger;
+import google.registry.util.Retrier;
+import google.registry.util.SendEmailService;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import javax.inject.Inject;
+import javax.mail.BodyPart;
+import javax.mail.Message;
+import javax.mail.Message.RecipientType;
+import javax.mail.MessagingException;
+import javax.mail.Multipart;
+import javax.mail.internet.InternetAddress;
+import javax.mail.internet.MimeBodyPart;
+import javax.mail.internet.MimeMultipart;
+import org.joda.time.YearMonth;
+
+/** Utility functions for sending emails involving monthly invoices. */
+class BillingEmailUtils {
+
+  private final SendEmailService emailService;
+  private final YearMonth yearMonth;
+  private final String alertSenderAddress;
+  private final String alertRecipientAddress;
+  private final ImmutableList<String> invoiceEmailRecipients;
+  private final String billingBucket;
+  private final String invoiceDirectoryPrefix;
+  private final GcsUtils gcsUtils;
+  private final Retrier retrier;
+
+  @Inject
+  BillingEmailUtils(
+      SendEmailService emailService,
+      YearMonth yearMonth,
+      @Config("alertSenderEmailAddress") String alertSenderAddress,
+      @Config("alertRecipientEmailAddress") String alertRecipientAddress,
+      @Config("invoiceEmailRecipients") ImmutableList<String> invoiceEmailRecipients,
+      @Config("billingBucket") String billingBucket,
+      @InvoiceDirectoryPrefix String invoiceDirectoryPrefix,
+      GcsUtils gcsUtils,
+      Retrier retrier) {
+    this.emailService = emailService;
+    this.yearMonth = yearMonth;
+    this.alertSenderAddress = alertSenderAddress;
+    this.alertRecipientAddress = alertRecipientAddress;
+    this.invoiceEmailRecipients = invoiceEmailRecipients;
+    this.billingBucket = billingBucket;
+    this.invoiceDirectoryPrefix = invoiceDirectoryPrefix;
+    this.gcsUtils = gcsUtils;
+    this.retrier = retrier;
+  }
+
+  private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
+
+  /** Sends an e-mail to all expected recipients with an attached overall invoice from GCS. */
+  void emailOverallInvoice() {
+    retrier.callWithRetry(
+        () -> {
+          String invoiceFile =
+              String.format(
+                  "%s-%s.csv", BillingModule.OVERALL_INVOICE_PREFIX, yearMonth.toString());
+          GcsFilename invoiceFilename =
+              new GcsFilename(billingBucket, invoiceDirectoryPrefix + invoiceFile);
+          try (InputStream in = gcsUtils.openInputStream(invoiceFilename)) {
+            Message msg = emailService.createMessage();
+            msg.setFrom(new InternetAddress(alertSenderAddress));
+            for (String recipient : invoiceEmailRecipients) {
+              msg.addRecipient(RecipientType.TO, new InternetAddress(recipient));
+            }
+            msg.setSubject(String.format("Domain Registry invoice data %s", yearMonth.toString()));
+            Multipart multipart = new MimeMultipart();
+            BodyPart textPart = new MimeBodyPart();
+            textPart.setText(
+                String.format(
+                    "Attached is the %s invoice for the domain registry.", yearMonth.toString()));
+            multipart.addBodyPart(textPart);
+            BodyPart invoicePart = new MimeBodyPart();
+            String invoiceData = CharStreams.toString(new InputStreamReader(in, UTF_8));
+            invoicePart.setContent(invoiceData, "text/csv; charset=utf-8");
+            invoicePart.setFileName(invoiceFile);
+            multipart.addBodyPart(invoicePart);
+            msg.setContent(multipart);
+            msg.saveChanges();
+            emailService.sendMessage(msg);
+          }
+        },
+        new Retrier.FailureReporter() {
+          @Override
+          public void beforeRetry(Throwable thrown, int failures, int maxAttempts) {}
+
+          @Override
+          public void afterFinalFailure(Throwable thrown, int failures) {
+            sendAlertEmail(
+                String.format("Emailing invoice failed due to %s", thrown.getMessage()));
+          }
+        },
+        IOException.class,
+        MessagingException.class);
+  }
+
+  /** Sends an e-mail to the provided alert e-mail address indicating a billing failure. */
+  void sendAlertEmail(String body) {
+    retrier.callWithRetry(
+        () -> {
+          Message msg = emailService.createMessage();
+          msg.setFrom(new InternetAddress(alertSenderAddress));
+          msg.addRecipient(RecipientType.TO, new InternetAddress(alertRecipientAddress));
+          msg.setSubject(String.format("Billing Pipeline Alert: %s", yearMonth.toString()));
+          msg.setText(body);
+          emailService.sendMessage(msg);
+          return null;
+        },
+        new Retrier.FailureReporter() {
+          @Override
+          public void beforeRetry(Throwable thrown, int failures, int maxAttempts) {}
+
+          @Override
+          public void afterFinalFailure(Throwable thrown, int failures) {
+            logger.severe(thrown, "The alert e-mail system failed.");
+          }
+        },
+        MessagingException.class);
+  }
+}

java/google/registry/reporting/billing/BillingModule.java

@@ -0,0 +1,86 @@
+// 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.reporting.billing;
+
+import static google.registry.request.RequestParameters.extractRequiredParameter;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import com.google.api.client.googleapis.extensions.appengine.auth.oauth2.AppIdentityCredential;
+import com.google.api.client.http.HttpTransport;
+import com.google.api.client.json.JsonFactory;
+import com.google.api.services.dataflow.Dataflow;
+import com.google.common.collect.ImmutableSet;
+import dagger.Module;
+import dagger.Provides;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.request.Parameter;
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.util.Set;
+import java.util.function.Function;
+import javax.inject.Qualifier;
+import javax.servlet.http.HttpServletRequest;
+import org.joda.time.YearMonth;
+
+/** Module for dependencies required by monthly billing actions. */
+@Module
+public final class BillingModule {
+
+  public static final String DETAIL_REPORT_PREFIX = "invoice_details";
+  public static final String OVERALL_INVOICE_PREFIX = "CRR-INV";
+  public static final String INVOICES_DIRECTORY = "invoices";
+
+  static final String PARAM_JOB_ID = "jobId";
+  static final String BILLING_QUEUE = "billing";
+  static final String CRON_QUEUE = "retryable-cron-tasks";
+
+  private static final String CLOUD_PLATFORM_SCOPE =
+      "https://www.googleapis.com/auth/cloud-platform";
+
+  /** Provides the invoicing Dataflow jobId enqueued by {@link GenerateInvoicesAction}. */
+  @Provides
+  @Parameter(PARAM_JOB_ID)
+  static String provideJobId(HttpServletRequest req) {
+    return extractRequiredParameter(req, PARAM_JOB_ID);
+  }
+
+  @Provides
+  @InvoiceDirectoryPrefix
+  static String provideDirectoryPrefix(YearMonth yearMonth) {
+    return String.format("%s/%s/", INVOICES_DIRECTORY, yearMonth.toString());
+  }
+
+  /** Constructs a {@link Dataflow} API client with default settings. */
+  @Provides
+  static Dataflow provideDataflow(
+      @Config("projectId") String projectId,
+      HttpTransport transport,
+      JsonFactory jsonFactory,
+      Function<Set<String>, AppIdentityCredential> appIdentityCredentialFunc) {
+
+    return new Dataflow.Builder(
+            transport,
+            jsonFactory,
+            appIdentityCredentialFunc.apply(ImmutableSet.of(CLOUD_PLATFORM_SCOPE)))
+        .setApplicationName(String.format("%s billing", projectId))
+        .build();
+  }
+
+  /** Dagger qualifier for the subdirectory we stage to/upload from. */
+  @Qualifier
+  @Documented
+  @Retention(RUNTIME)
+  @interface InvoiceDirectoryPrefix{}
+}

java/google/registry/reporting/billing/CopyDetailReportsAction.java

@@ -0,0 +1,140 @@
+// 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.reporting.billing;
+
+import static google.registry.request.Action.Method.POST;
+import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
+import static javax.servlet.http.HttpServletResponse.SC_OK;
+
+import com.google.appengine.tools.cloudstorage.GcsFilename;
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.ByteStreams;
+import com.google.common.net.MediaType;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.gcs.GcsUtils;
+import google.registry.model.registrar.Registrar;
+import google.registry.reporting.billing.BillingModule.InvoiceDirectoryPrefix;
+import google.registry.request.Action;
+import google.registry.request.Response;
+import google.registry.request.auth.Auth;
+import google.registry.storage.drive.DriveConnection;
+import google.registry.util.FormattingLogger;
+import google.registry.util.Retrier;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Optional;
+import javax.inject.Inject;
+
+/** Copy all registrar detail reports in a given bucket's subdirectory from GCS to Drive. */
+@Action(path = CopyDetailReportsAction.PATH, method = POST, auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+public final class CopyDetailReportsAction implements Runnable {
+
+  public static final String PATH = "/_dr/task/copyDetailReports";
+
+  private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
+
+  private final String billingBucket;
+  private final String invoiceDirectoryPrefix;
+  private final DriveConnection driveConnection;
+  private final GcsUtils gcsUtils;
+  private final Retrier retrier;
+  private final Response response;
+  private final BillingEmailUtils emailUtils;
+
+  @Inject
+  CopyDetailReportsAction(
+      @Config("billingBucket") String billingBucket,
+      @InvoiceDirectoryPrefix String invoiceDirectoryPrefix,
+      DriveConnection driveConnection,
+      GcsUtils gcsUtils,
+      Retrier retrier,
+      Response response,
+      BillingEmailUtils emailUtils) {
+    this.billingBucket = billingBucket;
+    this.invoiceDirectoryPrefix = invoiceDirectoryPrefix;
+    this.driveConnection = driveConnection;
+    this.gcsUtils = gcsUtils;
+    this.retrier = retrier;
+    this.response = response;
+    this.emailUtils = emailUtils;
+  }
+
+  @Override
+  public void run() {
+    ImmutableList<String> detailReportObjectNames;
+    try {
+      detailReportObjectNames =
+          gcsUtils
+              .listFolderObjects(billingBucket, invoiceDirectoryPrefix)
+              .stream()
+              .filter(objectName -> objectName.startsWith(BillingModule.DETAIL_REPORT_PREFIX))
+              .collect(ImmutableList.toImmutableList());
+    } catch (IOException e) {
+      logger.severe(e, "Copying registrar detail report failed");
+      response.setStatus(SC_INTERNAL_SERVER_ERROR);
+      response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
+      response.setPayload(String.format("Failure, encountered %s", e.getMessage()));
+      return;
+    }
+    for (String detailReportName : detailReportObjectNames) {
+      // The standard report format is "invoice_details_yyyy-MM_registrarId_tld.csv
+      // TODO(larryruili): Determine a safer way of enforcing this.
+      String registrarId = detailReportName.split("_")[3];
+      Optional<Registrar> registrar = Registrar.loadByClientId(registrarId);
+      if (!registrar.isPresent()) {
+        logger.warningfmt(
+            "Registrar %s not found in database for file %s", registrar, detailReportName);
+        continue;
+      }
+      String driveFolderId = registrar.get().getDriveFolderId();
+      if (driveFolderId == null) {
+        logger.warningfmt("Drive folder id not found for registrar %s", registrarId);
+        continue;
+      }
+      // Attempt to copy each detail report to its associated registrar's drive folder.
+      retrier.callWithRetry(
+          () -> {
+            try (InputStream input =
+                gcsUtils.openInputStream(
+                    new GcsFilename(billingBucket, invoiceDirectoryPrefix + detailReportName))) {
+              driveConnection.createFile(
+                  detailReportName,
+                  MediaType.CSV_UTF_8,
+                  driveFolderId,
+                  ByteStreams.toByteArray(input));
+              logger.infofmt(
+                  "Published detail report for %s to folder %s using GCS file gs://%s/%s.",
+                  registrarId, driveFolderId, billingBucket, detailReportName);
+            }
+          },
+          new Retrier.FailureReporter() {
+            @Override
+            public void beforeRetry(Throwable thrown, int failures, int maxAttempts) {}
+
+            @Override
+            public void afterFinalFailure(Throwable thrown, int failures) {
+              emailUtils.sendAlertEmail(
+                  String.format(
+                      "Warning: CopyDetailReportsAction failed.\nEncountered: %s on file: %s",
+                      thrown.getMessage(), detailReportName));
+            }
+          },
+          IOException.class);
+    }
+    response.setStatus(SC_OK);
+    response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
+    response.setPayload("Copied detail reports.");
+  }
+}

java/google/registry/reporting/billing/GenerateInvoicesAction.java

@@ -0,0 +1,127 @@
+// 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.reporting.billing;
+
+import static google.registry.reporting.ReportingModule.PARAM_YEAR_MONTH;
+import static google.registry.request.Action.Method.POST;
+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.appengine.api.taskqueue.QueueFactory;
+import com.google.appengine.api.taskqueue.TaskOptions;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.net.MediaType;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.request.Action;
+import google.registry.request.Response;
+import google.registry.request.auth.Auth;
+import google.registry.util.FormattingLogger;
+import java.io.IOException;
+import javax.inject.Inject;
+import org.joda.time.Duration;
+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.InvoicingPipeline} beam template, staged at
+ * gs://<projectId>-beam/templates/invoicing. The pipeline then generates invoices for the month and
+ * stores them on GCS.
+ */
+@Action(path = GenerateInvoicesAction.PATH, method = POST, auth = Auth.AUTH_INTERNAL_ONLY)
+public class GenerateInvoicesAction implements Runnable {
+
+  private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
+
+  static final String PATH = "/_dr/task/generateInvoices";
+
+  private final String projectId;
+  private final String beamBucketUrl;
+  private final String invoiceTemplateUrl;
+  private final YearMonth yearMonth;
+  private final Dataflow dataflow;
+  private final Response response;
+  private final BillingEmailUtils emailUtils;
+
+  @Inject
+  GenerateInvoicesAction(
+      @Config("projectId") String projectId,
+      @Config("apacheBeamBucketUrl") String beamBucketUrl,
+      @Config("invoiceTemplateUrl") String invoiceTemplateUrl,
+      YearMonth yearMonth,
+      Dataflow dataflow,
+      Response response,
+      BillingEmailUtils emailUtils) {
+    this.projectId = projectId;
+    this.beamBucketUrl = beamBucketUrl;
+    this.invoiceTemplateUrl = invoiceTemplateUrl;
+    this.yearMonth = yearMonth;
+    this.dataflow = dataflow;
+    this.response = response;
+    this.emailUtils = emailUtils;
+  }
+
+  @Override
+  public void run() {
+    logger.infofmt("Launching invoicing pipeline for %s", yearMonth);
+    try {
+      LaunchTemplateParameters params =
+          new LaunchTemplateParameters()
+              .setJobName(String.format("invoicing-%s", yearMonth))
+              .setEnvironment(
+                  new RuntimeEnvironment()
+                      .setZone("us-east1-c")
+                      .setTempLocation(beamBucketUrl + "/temporary"))
+              .setParameters(ImmutableMap.of("yearMonth", yearMonth.toString("yyyy-MM")));
+      LaunchTemplateResponse launchResponse =
+          dataflow
+              .projects()
+              .templates()
+              .launch(projectId, params)
+              .setGcsPath(invoiceTemplateUrl)
+              .execute();
+      logger.infofmt("Got response: %s", launchResponse.getJob().toPrettyString());
+      String jobId = launchResponse.getJob().getId();
+      enqueuePublishTask(jobId);
+    } catch (IOException e) {
+      logger.warning(e, "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.");
+  }
+
+  private void enqueuePublishTask(String jobId) {
+    TaskOptions publishTask =
+        TaskOptions.Builder.withUrl(PublishInvoicesAction.PATH)
+            .method(TaskOptions.Method.POST)
+            // Dataflow jobs tend to take about 10 minutes to complete.
+            .countdownMillis(Duration.standardMinutes(10).getMillis())
+            .param(BillingModule.PARAM_JOB_ID, jobId)
+            // Need to pass this through to ensure transitive yearMonth dependencies are satisfied.
+            .param(PARAM_YEAR_MONTH, yearMonth.toString());
+    QueueFactory.getQueue(BillingModule.BILLING_QUEUE).add(publishTask);
+  }
+}

java/google/registry/reporting/billing/PublishInvoicesAction.java

@@ -0,0 +1,119 @@
+// 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.reporting.billing;
+
+import static google.registry.reporting.ReportingModule.PARAM_YEAR_MONTH;
+import static google.registry.request.Action.Method.POST;
+import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
+import static javax.servlet.http.HttpServletResponse.SC_NOT_MODIFIED;
+import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
+import static javax.servlet.http.HttpServletResponse.SC_OK;
+
+import com.google.api.services.dataflow.Dataflow;
+import com.google.api.services.dataflow.model.Job;
+import com.google.appengine.api.taskqueue.QueueFactory;
+import com.google.appengine.api.taskqueue.TaskOptions;
+import com.google.common.net.MediaType;
+import google.registry.config.RegistryConfig.Config;
+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.FormattingLogger;
+import java.io.IOException;
+import javax.inject.Inject;
+import org.joda.time.YearMonth;
+
+/**
+ * Uploads the results of the {@link google.registry.beam.InvoicingPipeline}.
+ *
+ * <p>This relies on the retry semantics in {@code queue.xml} to ensure proper upload, in spite of
+ * fluctuations in generation timing.
+ *
+ * @see <a href=https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs#Job.JobState>
+ *   Job States</a>
+ */
+@Action(path = PublishInvoicesAction.PATH, method = POST, auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+public class PublishInvoicesAction implements Runnable {
+
+  private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
+  private static final String JOB_DONE = "JOB_STATE_DONE";
+  private static final String JOB_FAILED = "JOB_STATE_FAILED";
+
+  private final String projectId;
+  private final String jobId;
+  private final BillingEmailUtils emailUtils;
+  private final Dataflow dataflow;
+  private final Response response;
+  private final YearMonth yearMonth;
+
+  @Inject
+  PublishInvoicesAction(
+      @Config("projectId") String projectId,
+      @Parameter(BillingModule.PARAM_JOB_ID) String jobId,
+      BillingEmailUtils emailUtils,
+      Dataflow dataflow,
+      Response response,
+      YearMonth yearMonth) {
+    this.projectId = projectId;
+    this.jobId = jobId;
+    this.emailUtils = emailUtils;
+    this.dataflow = dataflow;
+    this.response = response;
+    this.yearMonth = yearMonth;
+  }
+
+  static final String PATH = "/_dr/task/publishInvoices";
+
+  @Override
+  public void run() {
+    try {
+      logger.info("Starting publish job.");
+      Job job = dataflow.projects().jobs().get(projectId, jobId).execute();
+      String state = job.getCurrentState();
+      switch (state) {
+        case JOB_DONE:
+          logger.infofmt("Dataflow job %s finished successfully, publishing results.", jobId);
+          response.setStatus(SC_OK);
+          enqueueCopyDetailReportsTask();
+          emailUtils.emailOverallInvoice();
+          break;
+        case JOB_FAILED:
+          logger.severefmt("Dataflow job %s finished unsuccessfully.", jobId);
+          response.setStatus(SC_NO_CONTENT);
+          emailUtils.sendAlertEmail(
+              String.format("Dataflow job %s ended in status failure.", jobId));
+          break;
+        default:
+          logger.infofmt("Job in non-terminal state %s, retrying:", state);
+          response.setStatus(SC_NOT_MODIFIED);
+          break;
+      }
+    } catch (IOException e) {
+      emailUtils.sendAlertEmail(String.format("Publish action 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()));
+    }
+  }
+
+  private void enqueueCopyDetailReportsTask() {
+    TaskOptions copyDetailTask =
+        TaskOptions.Builder.withUrl(CopyDetailReportsAction.PATH)
+            .method(TaskOptions.Method.POST)
+            .param(PARAM_YEAR_MONTH, yearMonth.toString());
+    QueueFactory.getQueue(BillingModule.CRON_QUEUE).add(copyDetailTask);
+  }
+}

java/google/registry/reporting/icann/ActivityReportingQueryBuilder.java

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
-import static google.registry.reporting.IcannReportingModule.DATASTORE_EXPORT_DATA_SET;
-import static google.registry.reporting.IcannReportingModule.ICANN_REPORTING_DATA_SET;
+import static google.registry.reporting.icann.IcannReportingModule.DATASTORE_EXPORT_DATA_SET;
+import static google.registry.reporting.icann.IcannReportingModule.ICANN_REPORTING_DATA_SET;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.Resources;

java/google/registry/reporting/icann/BUILD

@@ -0,0 +1,34 @@
+package(
+    default_visibility = ["//visibility:public"],
+)
+
+licenses(["notice"])  # Apache 2.0
+
+java_library(
+    name = "icann",
+    srcs = glob(["*.java"]),
+    resources = glob(["sql/*"]),
+    deps = [
+        "//java/google/registry/bigquery",
+        "//java/google/registry/config",
+        "//java/google/registry/gcs",
+        "//java/google/registry/keyring/api",
+        "//java/google/registry/model",
+        "//java/google/registry/request",
+        "//java/google/registry/request/auth",
+        "//java/google/registry/util",
+        "//java/google/registry/xjc",
+        "//java/google/registry/xml",
+        "@com_google_api_client",
+        "@com_google_apis_google_api_services_bigquery",
+        "@com_google_appengine_api_1_0_sdk",
+        "@com_google_appengine_tools_appengine_gcs_client",
+        "@com_google_code_findbugs_jsr305",
+        "@com_google_dagger",
+        "@com_google_guava",
+        "@com_google_http_client",
+        "@com_google_http_client_jackson2",
+        "@javax_servlet_api",
+        "@joda_time",
+    ],
+)

java/google/registry/reporting/icann/IcannHttpReporter.java

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.net.MediaType.CSV_UTF_8;
@@ -30,7 +30,7 @@ import com.google.common.base.Splitter;
 import com.google.common.io.ByteStreams;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.keyring.api.KeyModule.Key;
-import google.registry.reporting.IcannReportingModule.ReportType;
+import google.registry.reporting.icann.IcannReportingModule.ReportType;
 import google.registry.util.FormattingLogger;
 import google.registry.xjc.XjcXmlTransformer;
 import google.registry.xjc.iirdea.XjcIirdeaResponseElement;

java/google/registry/reporting/icann/IcannReportingModule.java

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
 import static google.registry.request.RequestParameters.extractOptionalEnumParameter;
 import static google.registry.request.RequestParameters.extractOptionalParameter;

java/google/registry/reporting/icann/IcannReportingStager.java

@@ -12,12 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Strings.isNullOrEmpty;
 import static com.google.common.collect.ImmutableList.toImmutableList;
-import static google.registry.reporting.IcannReportingModule.MANIFEST_FILE_NAME;
+import static google.registry.reporting.icann.IcannReportingModule.MANIFEST_FILE_NAME;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import com.google.api.services.bigquery.model.TableFieldSchema;
@@ -34,8 +34,8 @@ import google.registry.bigquery.BigqueryConnection;
 import google.registry.bigquery.BigqueryUtils.TableType;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.gcs.GcsUtils;
-import google.registry.reporting.IcannReportingModule.ReportType;
-import google.registry.reporting.IcannReportingModule.ReportingSubdir;
+import google.registry.reporting.icann.IcannReportingModule.ReportType;
+import google.registry.reporting.icann.IcannReportingModule.ReportingSubdir;
 import google.registry.util.FormattingLogger;
 import java.io.IOException;
 import java.util.ArrayList;

java/google/registry/reporting/icann/IcannReportingStagingAction.java

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
 import static google.registry.request.Action.Method.POST;
 import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
@@ -25,8 +25,8 @@ import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.net.MediaType;
 import google.registry.bigquery.BigqueryJobFailureException;
-import google.registry.reporting.IcannReportingModule.ReportType;
-import google.registry.reporting.IcannReportingModule.ReportingSubdir;
+import google.registry.reporting.icann.IcannReportingModule.ReportType;
+import google.registry.reporting.icann.IcannReportingModule.ReportingSubdir;
 import google.registry.request.Action;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;

java/google/registry/reporting/icann/IcannReportingUploadAction.java

@@ -12,11 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.net.MediaType.PLAIN_TEXT_UTF_8;
-import static google.registry.reporting.IcannReportingModule.MANIFEST_FILE_NAME;
+import static google.registry.reporting.icann.IcannReportingModule.MANIFEST_FILE_NAME;
 import static google.registry.request.Action.Method.POST;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
@@ -28,7 +28,7 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.io.ByteStreams;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.gcs.GcsUtils;
-import google.registry.reporting.IcannReportingModule.ReportingSubdir;
+import google.registry.reporting.icann.IcannReportingModule.ReportingSubdir;
 import google.registry.request.Action;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;

java/google/registry/reporting/icann/QueryBuilder.java

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
 import com.google.common.collect.ImmutableMap;
 import java.io.IOException;

java/google/registry/reporting/icann/ReportingEmailUtils.java

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
 import google.registry.config.RegistryConfig.Config;
 import google.registry.util.FormattingLogger;

java/google/registry/reporting/icann/TransactionsReportingQueryBuilder.java

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.reporting;
+package google.registry.reporting.icann;
 
-import static google.registry.reporting.IcannReportingModule.DATASTORE_EXPORT_DATA_SET;
-import static google.registry.reporting.IcannReportingModule.ICANN_REPORTING_DATA_SET;
+import static google.registry.reporting.icann.IcannReportingModule.DATASTORE_EXPORT_DATA_SET;
+import static google.registry.reporting.icann.IcannReportingModule.ICANN_REPORTING_DATA_SET;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.Resources;