Add publish functionality to billing pipeline

This closes the end-to-end billing pipeline, allowing us to share generated detail reports with registrars via Drive and e-mail the invoicing team a link to the generated invoice.

This also factors out the email configs from ICANN reporting into the common 'misc' config, since we'll likely need alert e-mails for future periodic tasks.

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

java/google/registry/billing/BUILD

@@ -12,12 +12,16 @@ java_library(
     ],
     deps = [
         "//java/google/registry/config",
+        "//java/google/registry/gcs",
+        "//java/google/registry/model",
         "//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",

java/google/registry/billing/BillingEmailUtils.java

@@ -0,0 +1,85 @@
+// 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.billing;
+
+import com.google.common.collect.ImmutableList;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.util.FormattingLogger;
+import google.registry.util.SendEmailService;
+import javax.inject.Inject;
+import javax.mail.Message;
+import javax.mail.Message.RecipientType;
+import javax.mail.MessagingException;
+import javax.mail.internet.InternetAddress;
+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 ImmutableList<String> invoiceEmailRecipients;
+  // TODO(larryruili): Replace this bucket after verifying 2017-12 output.
+  private final String beamBucketUrl;
+
+  @Inject
+  BillingEmailUtils(
+      SendEmailService emailService,
+      YearMonth yearMonth,
+      @Config("alertSenderEmailAddress") String alertSenderAddress,
+      @Config("invoiceEmailRecipients") ImmutableList<String> invoiceEmailRecipients,
+      @Config("apacheBeamBucketUrl") String beamBucketUrl) {
+    this.emailService = emailService;
+    this.yearMonth = yearMonth;
+    this.alertSenderAddress = alertSenderAddress;
+    this.invoiceEmailRecipients = invoiceEmailRecipients;
+    this.beamBucketUrl = beamBucketUrl;
+  }
+
+  private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
+
+  /**
+   * Sends a link to the generated overall invoice in GCS.
+   *
+   * <p>Note the users receiving the e-mail should have access to the object or bucket, via an
+   * authorization mechanism such as IAM.
+   */
+  void emailInvoiceLink() {
+    // TODO(larryruili): Add read permissions for appropriate buckets.
+    try {
+      String beamBucket = beamBucketUrl.replaceFirst("gs://", "");
+      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()));
+      msg.setText(
+          String.format(
+              "Link to invoice on GCS:\nhttps://storage.cloud.google.com/%s/%s",
+              beamBucket,
+              String.format(
+                  "%s%s-%s.csv",
+                  BillingModule.RESULTS_DIRECTORY_PREFIX,
+                  BillingModule.OVERALL_INVOICE_PREFIX,
+                  yearMonth.toString())));
+      emailService.sendMessage(msg);
+    } catch (MessagingException e) {
+      // TODO(larryruili): Replace with retrier with final failure email settings.
+      logger.warning(e, "E-mail service failed due to %s");
+    }
+  }
+}

java/google/registry/billing/BillingModule.java

@@ -33,10 +33,18 @@ import javax.servlet.http.HttpServletRequest;
 @Module
 public final class BillingModule {
 
+  public static final String DETAIL_REPORT_PREFIX = "invoice_details";
+  public static final String OVERALL_INVOICE_PREFIX = "CRR-INV";
+
+  static final String PARAM_JOB_ID = "jobId";
+  static final String PARAM_DIRECTORY_PREFIX = "directoryPrefix";
+  static final String BILLING_QUEUE = "billing";
+  static final String CRON_QUEUE = "retryable-cron-tasks";
+  // TODO(larryruili): Replace with invoices/yyyy-MM after verifying 2017-12 invoice.
+  static final String RESULTS_DIRECTORY_PREFIX = "results/";
+
   private static final String CLOUD_PLATFORM_SCOPE =
       "https://www.googleapis.com/auth/cloud-platform";
-  static final String BILLING_QUEUE = "billing";
-  static final String PARAM_JOB_ID = "jobId";
 
   /** Provides the invoicing Dataflow jobId enqueued by {@link GenerateInvoicesAction}. */
   @Provides
@@ -45,6 +53,13 @@ public final class BillingModule {
     return extractRequiredParameter(req, PARAM_JOB_ID);
   }
 
+  /** Provides the subdirectory under a GCS bucket that we copy detail reports from. */
+  @Provides
+  @Parameter(PARAM_DIRECTORY_PREFIX)
+  static String provideDirectoryPrefix(HttpServletRequest req) {
+    return extractRequiredParameter(req, PARAM_DIRECTORY_PREFIX);
+  }
+
   /** Constructs a {@link Dataflow} API client with default settings. */
   @Provides
   static Dataflow provideDataflow(

java/google/registry/billing/CopyDetailReportsAction.java

@@ -0,0 +1,129 @@
+// 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.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.request.Action;
+import google.registry.request.Parameter;
+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();
+
+  // TODO(larryruili): Replace this bucket with the billing bucket after verifying 2017-12 output.
+  private final String beamBucketUrl;
+  private final String folderPrefix;
+  private final DriveConnection driveConnection;
+  private final GcsUtils gcsUtils;
+  private final Retrier retrier;
+  private final Response response;
+
+  @Inject
+  CopyDetailReportsAction(
+      @Config("apacheBeamBucketUrl") String beamBucketUrl,
+      @Parameter(BillingModule.PARAM_DIRECTORY_PREFIX) String folderPrefix,
+      DriveConnection driveConnection,
+      GcsUtils gcsUtils,
+      Retrier retrier,
+      Response response) {
+    this.beamBucketUrl = beamBucketUrl;
+    this.folderPrefix = folderPrefix;
+    this.driveConnection = driveConnection;
+    this.gcsUtils = gcsUtils;
+    this.retrier = retrier;
+    this.response = response;
+  }
+
+  @Override
+  public void run() {
+    // Strip the URL prefix from the beam bucket
+    String beamBucket = beamBucketUrl.replace("gs://", "");
+    ImmutableList<String> detailReportObjectNames;
+    try {
+      detailReportObjectNames =
+          gcsUtils
+              .listFolderObjects(beamBucket, folderPrefix)
+              .stream()
+              .filter(objectName -> objectName.startsWith(BillingModule.DETAIL_REPORT_PREFIX))
+              .collect(ImmutableList.toImmutableList());
+    } catch (IOException e) {
+      logger.severefmt("Copy failed due to %s", e.getMessage());
+      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);
+      // TODO(larryruili): Send an email alert if any report fails to be copied for any reason.
+      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(beamBucket, folderPrefix + 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, beamBucket, detailReportName);
+            }
+          },
+          IOException.class);
+    }
+    response.setStatus(SC_OK);
+    response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
+    response.setPayload("Copied detail reports.");
+  }
+}

java/google/registry/billing/GenerateInvoicesAction.java

@@ -49,23 +49,28 @@ public class GenerateInvoicesAction implements Runnable {
 
   private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
 
-  @Inject
-  @Config("projectId")
-  String projectId;
-
-  @Inject
-  @Config("apacheBeamBucketUrl")
-  String beamBucketUrl;
-
-  @Inject YearMonth yearMonth;
-  @Inject Dataflow dataflow;
-  @Inject Response response;
-
-  @Inject
-  GenerateInvoicesAction() {}
-
   static final String PATH = "/_dr/task/generateInvoices";
 
+  private final String projectId;
+  private final String beamBucketUrl;
+  private final YearMonth yearMonth;
+  private final Dataflow dataflow;
+  private final Response response;
+
+  @Inject
+  GenerateInvoicesAction(
+      @Config("projectId") String projectId,
+      @Config("apacheBeamBucketUrl") String beamBucketUrl,
+      YearMonth yearMonth,
+      Dataflow dataflow,
+      Response response) {
+    this.projectId = projectId;
+    this.beamBucketUrl = beamBucketUrl;
+    this.yearMonth = yearMonth;
+    this.dataflow = dataflow;
+    this.response = response;
+  }
+
   @Override
   public void run() {
     logger.infofmt("Launching invoicing pipeline for %s", yearMonth);
@@ -87,13 +92,7 @@ public class GenerateInvoicesAction implements Runnable {
               .execute();
       logger.infofmt("Got response: %s", launchResponse.getJob().toPrettyString());
       String jobId = launchResponse.getJob().getId();
-      TaskOptions uploadTask =
-          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);
-      QueueFactory.getQueue(BillingModule.BILLING_QUEUE).add(uploadTask);
+      enqueuePublishTask(jobId);
     } catch (IOException e) {
       logger.warningfmt("Template Launch failed due to: %s", e.getMessage());
       response.setStatus(SC_INTERNAL_SERVER_ERROR);
@@ -105,4 +104,14 @@ public class GenerateInvoicesAction implements Runnable {
     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);
+    QueueFactory.getQueue(BillingModule.BILLING_QUEUE).add(publishTask);
+  }
 }

java/google/registry/billing/PublishInvoicesAction.java

@@ -22,6 +22,8 @@ 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;
@@ -48,31 +50,45 @@ public class PublishInvoicesAction implements Runnable {
   private static final String JOB_DONE = "JOB_STATE_DONE";
   private static final String JOB_FAILED = "JOB_STATE_FAILED";
 
-  @Inject @Config("projectId") String projectId;
-  @Inject @Parameter(BillingModule.PARAM_JOB_ID) String jobId;
-  @Inject Dataflow dataflow;
-  @Inject Response response;
-  @Inject PublishInvoicesAction() {}
+  private final String projectId;
+  private final String jobId;
+  private final BillingEmailUtils emailUtils;
+  private final Dataflow dataflow;
+  private final Response response;
+
+  @Inject
+  PublishInvoicesAction(
+      @Config("projectId") String projectId,
+      @Parameter(BillingModule.PARAM_JOB_ID) String jobId,
+      BillingEmailUtils emailUtils,
+      Dataflow dataflow,
+      Response response) {
+    this.projectId = projectId;
+    this.jobId = jobId;
+    this.emailUtils = emailUtils;
+    this.dataflow = dataflow;
+    this.response = response;
+  }
 
   static final String PATH = "/_dr/task/publishInvoices";
 
   @Override
   public void run() {
-    logger.info("Starting publish job.");
     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.", jobId);
+          logger.infofmt("Dataflow job %s finished successfully, publishing results.", jobId);
           response.setStatus(SC_OK);
-          // TODO(larryruili): Implement upload logic.
+          enqueueCopyDetailReportsTask();
+          emailUtils.emailInvoiceLink();
           break;
         case JOB_FAILED:
           logger.severefmt("Dataflow job %s finished unsuccessfully.", jobId);
-          // Return a 'success' code to stop task queue retry.
           response.setStatus(SC_NO_CONTENT);
-          // TODO(larryruili): Implement failure response.
+          // TODO(larryruili): Email failure message
           break;
         default:
           logger.infofmt("Job in non-terminal state %s, retrying:", state);
@@ -80,10 +96,18 @@ public class PublishInvoicesAction implements Runnable {
           break;
       }
     } catch (IOException e) {
-      logger.warningfmt("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()));
     }
   }
+
+  private static void enqueueCopyDetailReportsTask() {
+    TaskOptions copyDetailTask =
+        TaskOptions.Builder.withUrl(CopyDetailReportsAction.PATH)
+            .method(TaskOptions.Method.POST)
+            .param(
+                BillingModule.PARAM_DIRECTORY_PREFIX, BillingModule.RESULTS_DIRECTORY_PREFIX);
+    QueueFactory.getQueue(BillingModule.CRON_QUEUE).add(copyDetailTask);
+  }
 }