Prepare ICANN reporting for production

This originally started as a small change, but quickly grew into a major refactor as I realized the original parameter structure wasn't conducive to a cron task and manual re-runs.

The changes are as follows:
1. Adds DNS metrics to activity reports, thanks to Nick's work with the Zoneman Dremel -> #plx workflow.
2. Surrounds registrar names in transactions reports with quotes, to escape possible commas.
3. Factors out the report generation logic into IcannReportingStager.
4. Assigns default values to the three main parameters
  - yearMonth defaults to the previous month
  - subdir defaults to "icann/monthly/yearMonth", i.e. "gs://domain-registry-reporting/icann/monthly/yyyy-MM"
  - reportType defaults to both reports
5. Adds "Total" row generation logic to transactions reports
  - This was a previously overlooked requirement.
6. Adds "MANIFEST.txt" generation and upload logic.
  - The MANIFEST lists out which files need to be uploaded in the subdirectory.
7. Increases urlfetch timeout from 5s to 10s in backend tasks.
  - Backend tasks should be more latency tolerant anyway, and this reduces the number of incorrect timeouts we see for services like Bigquery which might take some time to respond.

TESTED=Extensive testing in alpha, and ran FOSS test.
TODO: send out an e-mail for report generation and upload, and add reporting to cron.xml

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

java/google/registry/reporting/ActivityReportingQueryBuilder.java

@@ -36,12 +36,12 @@ import org.joda.time.format.DateTimeFormatter;
 public final class ActivityReportingQueryBuilder implements QueryBuilder {
 
   // Names for intermediary tables for overall activity reporting query.
-  static final String ACTIVITY_REPORT_AGGREGATION = "activity_report_aggregation";
-  static final String MONTHLY_LOGS = "monthly_logs";
   static final String REGISTRAR_OPERATING_STATUS = "registrar_operating_status";
   static final String DNS_COUNTS = "dns_counts";
+  static final String MONTHLY_LOGS = "monthly_logs";
   static final String EPP_METRICS = "epp_metrics";
   static final String WHOIS_COUNTS = "whois_counts";
+  static final String ACTIVITY_REPORT_AGGREGATION = "activity_report_aggregation";
 
   @Inject @Config("projectId") String projectId;
   @Inject @Parameter(IcannReportingModule.PARAM_YEAR_MONTH) String yearMonth;
@@ -81,7 +81,6 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
             .build();
     queriesBuilder.put(getTableName(REGISTRAR_OPERATING_STATUS), operationalRegistrarsQuery);
 
-    // TODO(b/62626209): Make this use the CloudDNS counts instead.
     String dnsCountsQuery =
         SqlTemplate.create(getQueryFromFile("dns_counts.sql")).build();
     queriesBuilder.put(getTableName(DNS_COUNTS), dnsCountsQuery);
@@ -135,6 +134,7 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
     return queriesBuilder.build();
   }
 
+
   /** Returns the table name of the query, suffixed with the yearMonth in _YYYYMM format. */
   private String getTableName(String queryName) {
     return String.format("%s_%s", queryName, yearMonth.replace("-", ""));
@@ -149,4 +149,3 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
     return Resources.getResource(ActivityReportingQueryBuilder.class, "sql/" + filename);
   }
 }
-

java/google/registry/reporting/IcannHttpReporter.java

@@ -14,7 +14,9 @@
 
 package google.registry.reporting;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.net.MediaType.CSV_UTF_8;
+import static google.registry.model.registry.Registries.assertTldExists;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import com.google.api.client.http.ByteArrayContent;
@@ -23,6 +25,8 @@ import com.google.api.client.http.HttpHeaders;
 import com.google.api.client.http.HttpRequest;
 import com.google.api.client.http.HttpResponse;
 import com.google.api.client.http.HttpTransport;
+import com.google.common.base.Ascii;
+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;
@@ -35,16 +39,22 @@ import google.registry.xjc.iirdea.XjcIirdeaResult;
 import google.registry.xml.XmlException;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.util.List;
 import javax.inject.Inject;
+import org.joda.time.YearMonth;
+import org.joda.time.format.DateTimeFormat;
 
 /**
  * Class that uploads a CSV file to ICANN's endpoint via an HTTP PUT call.
  *
- * <p> It uses basic authorization credentials as specified in the "Registry Interfaces" draft.
+ * <p>It uses basic authorization credentials as specified in the "Registry Interfaces" draft.
+ *
+ * <p>Note that there's a lot of hard-coded logic extracting parameters from the report filenames.
+ * These are safe, as long as they follow the tld-reportType-yearMonth.csv filename format.
  *
  * @see IcannReportingUploadAction
- * @see <a href=https://tools.ietf.org/html/draft-lozano-icann-registry-interfaces-07#page-9>
- *   ICANN Reporting Specification</a>
+ * @see <a href=https://tools.ietf.org/html/draft-lozano-icann-registry-interfaces-07#page-9>ICANN
+ *     Reporting Specification</a>
  */
 public class IcannHttpReporter {
 
@@ -57,29 +67,24 @@ public class IcannHttpReporter {
   @Inject IcannHttpReporter() {}
 
   /** Uploads {@code reportBytes} to ICANN. */
-  public void send(
-      byte[] reportBytes,
-      String tld,
-      String yearMonth,
-      ReportType reportType) throws XmlException, IOException {
-    GenericUrl uploadUrl = new GenericUrl(makeUrl(tld, yearMonth, reportType));
+  public void send(byte[] reportBytes, String reportFilename) throws XmlException, IOException {
+    validateReportFilename(reportFilename);
+    GenericUrl uploadUrl = new GenericUrl(makeUrl(reportFilename));
     HttpRequest request =
         httpTransport
             .createRequestFactory()
             .buildPutRequest(uploadUrl, new ByteArrayContent(CSV_UTF_8.toString(), reportBytes));
 
     HttpHeaders headers = request.getHeaders();
-    headers.setBasicAuthentication(tld + "_ry", password);
+    headers.setBasicAuthentication(getTld(reportFilename) + "_ry", password);
     headers.setContentType(CSV_UTF_8.toString());
     request.setHeaders(headers);
     request.setFollowRedirects(false);
 
     HttpResponse response = null;
     logger.infofmt(
-        "Sending %s report to %s with content length %s",
-        reportType,
-        uploadUrl.toString(),
-        request.getContent().getLength());
+        "Sending report to %s with content length %s",
+        uploadUrl.toString(), request.getContent().getLength());
     try {
       response = request.execute();
       byte[] content;
@@ -117,9 +122,31 @@ public class IcannHttpReporter {
     return result;
   }
 
-  private String makeUrl(String tld, String yearMonth, ReportType reportType) {
-    String urlPrefix = getUrlPrefix(reportType);
-    return String.format("%s/%s/%s", urlPrefix, tld, yearMonth);
+  /** Verifies a given report filename matches the pattern tld-reportType-yyyyMM.csv. */
+  private void validateReportFilename(String filename) {
+    checkArgument(
+        filename.matches("[a-z0-9.\\-]+-((activity)|(transactions))-[0-9]{6}\\.csv"),
+        "Expected file format: tld-reportType-yyyyMM.csv, got %s instead",
+        filename);
+    assertTldExists(getTld(filename));
+  }
+
+  private String getTld(String filename) {
+    // Extract the TLD, up to second-to-last hyphen in the filename (works with international TLDs)
+    return filename.substring(0, filename.lastIndexOf('-', filename.lastIndexOf('-') - 1));
+  }
+
+  private String makeUrl(String filename) {
+    // Filename is in the format tld-reportType-yearMonth.csv
+    String tld = getTld(filename);
+    // Remove the tld- prefix and csv suffix
+    String remainder = filename.substring(tld.length() + 1, filename.length() - 4);
+    List<String> elements = Splitter.on('-').splitToList(remainder);
+    ReportType reportType = ReportType.valueOf(Ascii.toUpperCase(elements.get(0)));
+    // Re-add hyphen between year and month, because ICANN is inconsistent between filename and URL
+    String yearMonth =
+        YearMonth.parse(elements.get(1), DateTimeFormat.forPattern("yyyyMM")).toString("yyyy-MM");
+    return String.format("%s/%s/%s", getUrlPrefix(reportType), tld, yearMonth);
   }
 
   private String getUrlPrefix(ReportType reportType) {
@@ -131,7 +158,7 @@ public class IcannHttpReporter {
       default:
         throw new IllegalStateException(
             String.format(
-                "Received invalid reportType! Expected ACTIVITY or TRANSACTIONS, got %s.",
+                "Received invalid reportTypes! Expected ACTIVITY or TRANSACTIONS, got %s.",
                 reportType));
     }
   }

java/google/registry/reporting/IcannReportingModule.java

@@ -14,22 +14,24 @@
 
 package google.registry.reporting;
 
-import static google.registry.request.RequestParameters.extractEnumParameter;
+import static google.registry.request.RequestParameters.extractOptionalEnumParameter;
 import static google.registry.request.RequestParameters.extractOptionalParameter;
-import static google.registry.request.RequestParameters.extractRequiredParameter;
 
 import com.google.api.client.googleapis.auth.oauth2.GoogleCredential;
 import com.google.api.client.http.HttpTransport;
 import com.google.api.client.json.jackson2.JacksonFactory;
 import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.MoreExecutors;
 import dagger.Module;
 import dagger.Provides;
 import google.registry.bigquery.BigqueryConnection;
+import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.Parameter;
+import google.registry.util.Clock;
 import java.util.Optional;
-import java.util.concurrent.Executors;
 import javax.servlet.http.HttpServletRequest;
 import org.joda.time.Duration;
+import org.joda.time.format.DateTimeFormat;
 
 /** Module for dependencies required by ICANN monthly transactions/activity reporting. */
 @Module
@@ -41,43 +43,78 @@ public final class IcannReportingModule {
     ACTIVITY
   }
 
+  static final String PARAM_OPTIONAL_YEAR_MONTH = "yearMonthOptional";
   static final String PARAM_YEAR_MONTH = "yearMonth";
-  static final String PARAM_REPORT_TYPE = "reportType";
+  static final String PARAM_OPTIONAL_SUBDIR = "subdirOptional";
   static final String PARAM_SUBDIR = "subdir";
+  static final String PARAM_REPORT_TYPE = "reportType";
   static final String ICANN_REPORTING_DATA_SET = "icann_reporting";
   static final String DATASTORE_EXPORT_DATA_SET = "latest_datastore_export";
-  private static final String BIGQUERY_SCOPE =  "https://www.googleapis.com/auth/bigquery";
+  static final String MANIFEST_FILE_NAME = "MANIFEST.txt";
+  private static final String DEFAULT_SUBDIR = "icann/monthly";
+  private static final String BIGQUERY_SCOPE = "https://www.googleapis.com/auth/cloud-platform";
 
+  /** Extracts an optional yearMonth in yyyy-MM format from the request. */
+  @Provides
+  @Parameter(PARAM_OPTIONAL_YEAR_MONTH)
+  static Optional<String> provideYearMonthOptional(HttpServletRequest req) {
+    return extractOptionalParameter(req, PARAM_YEAR_MONTH);
+  }
+
+  /** Provides the yearMonth in yyyy-MM format, defaults to one month prior to run time. */
   @Provides
   @Parameter(PARAM_YEAR_MONTH)
-  static String provideYearMonth(HttpServletRequest req) {
-    return extractRequiredParameter(req, PARAM_YEAR_MONTH);
+  static String provideYearMonth(
+      @Parameter(PARAM_OPTIONAL_YEAR_MONTH) Optional<String> yearMonthOptional, Clock clock) {
+    String yearMonth =
+        yearMonthOptional.orElse(
+            DateTimeFormat.forPattern("yyyy-MM").print(clock.nowUtc().minusMonths(1)));
+    if (!yearMonth.matches("[0-9]{4}-[0-9]{2}")) {
+      throw new BadRequestException(
+          String.format("yearMonth must be in yyyy-MM format, got %s instead", yearMonth));
+    }
+    return yearMonth;
   }
 
+  /** Provides an optional subdirectory to store/upload reports to, extracted from the request. */
   @Provides
-  @Parameter(PARAM_REPORT_TYPE)
-  static ReportType provideReportType(HttpServletRequest req) {
-    return extractEnumParameter(req, ReportType.class, PARAM_REPORT_TYPE);
-  }
-
-  @Provides
-  @Parameter(PARAM_SUBDIR)
-  static Optional<String> provideSubdir(HttpServletRequest req) {
+  @Parameter(PARAM_OPTIONAL_SUBDIR)
+  static Optional<String> provideSubdirOptional(HttpServletRequest req) {
     return extractOptionalParameter(req, PARAM_SUBDIR);
   }
 
+  /** Provides the subdirectory to store/upload reports to, defaults to icann/monthly/yearMonth. */
   @Provides
-  static QueryBuilder provideQueryBuilder(
-      @Parameter(PARAM_REPORT_TYPE) ReportType reportType,
-      ActivityReportingQueryBuilder activityBuilder,
-      TransactionsReportingQueryBuilder transactionsBuilder) {
-    return reportType == ReportType.ACTIVITY ? activityBuilder : transactionsBuilder;
+  @Parameter(PARAM_SUBDIR)
+  static String provideSubdir(
+      @Parameter(PARAM_OPTIONAL_SUBDIR) Optional<String> subdirOptional,
+      @Parameter(PARAM_YEAR_MONTH) String yearMonth) {
+    String subdir = subdirOptional.orElse(String.format("%s/%s", DEFAULT_SUBDIR, yearMonth));
+    if (subdir.startsWith("/") || subdir.endsWith("/")) {
+      throw new BadRequestException(
+          String.format("subdir must not start or end with a \"/\", got %s instead.", subdir));
+    }
+    return subdir;
+  }
+
+  /** Provides an optional reportType to store/upload reports to, extracted from the request. */
+  @Provides
+  static Optional<ReportType> provideReportTypeOptional(HttpServletRequest req) {
+    return extractOptionalEnumParameter(req, ReportType.class, PARAM_REPORT_TYPE);
+  }
+
+  /** Provides a list of reportTypes specified. If absent, we default to both report types. */
+  @Provides
+  @Parameter(PARAM_REPORT_TYPE)
+  static ImmutableList<ReportType> provideReportTypes(Optional<ReportType> reportTypeOptional) {
+    return reportTypeOptional.map(ImmutableList::of)
+        .orElseGet(() -> ImmutableList.of(ReportType.ACTIVITY, ReportType.TRANSACTIONS));
   }
 
   /**
    * Constructs a BigqueryConnection with default settings.
    *
-   * <p> We use Bigquery to generate activity reports via large aggregate SQL queries.
+   * <p>We use Bigquery to generate ICANN monthly reports via large aggregate SQL queries.
    *
    * @see ActivityReportingQueryBuilder
    * @see google.registry.tools.BigqueryParameters for justifications of defaults.
@@ -87,13 +124,14 @@ public final class IcannReportingModule {
     try {
       GoogleCredential credential = GoogleCredential
           .getApplicationDefault(transport, new JacksonFactory());
-      BigqueryConnection connection = new BigqueryConnection.Builder()
-          .setExecutorService(Executors.newFixedThreadPool(20))
-          .setCredential(credential.createScoped(ImmutableList.of(BIGQUERY_SCOPE)))
-          .setDatasetId(ICANN_REPORTING_DATA_SET)
-          .setOverwrite(true)
-          .setPollInterval(Duration.standardSeconds(1))
-          .build();
+      BigqueryConnection connection =
+          new BigqueryConnection.Builder()
+              .setExecutorService(MoreExecutors.newDirectExecutorService())
+              .setCredential(credential.createScoped(ImmutableList.of(BIGQUERY_SCOPE)))
+              .setDatasetId(ICANN_REPORTING_DATA_SET)
+              .setOverwrite(true)
+              .setPollInterval(Duration.standardSeconds(1))
+              .build();
       connection.initialize();
       return connection;
     } catch (Throwable e) {
@@ -101,3 +139,4 @@ public final class IcannReportingModule {
     }
   }
 }
+

java/google/registry/reporting/IcannReportingStager.java

@@ -0,0 +1,264 @@
+// 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 com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static google.registry.reporting.IcannReportingModule.MANIFEST_FILE_NAME;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.api.services.bigquery.model.TableFieldSchema;
+import com.google.appengine.tools.cloudstorage.GcsFilename;
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ImmutableCollection;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableTable;
+import com.google.common.collect.ListMultimap;
+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.request.Parameter;
+import google.registry.util.FormattingLogger;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import javax.inject.Inject;
+
+/**
+ * Class containing methods for staging ICANN monthly reports on GCS.
+ *
+ * <p>The main entrypoint is stageReports, which generates a given type of reports.
+ */
+public class IcannReportingStager {
+
+  private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
+
+  @Inject @Config("icannReportingBucket") String reportingBucket;
+  @Inject @Parameter(IcannReportingModule.PARAM_YEAR_MONTH) String yearMonth;
+
+  @Inject
+  @Parameter(IcannReportingModule.PARAM_SUBDIR)
+  String subdir;
+
+  @Inject ActivityReportingQueryBuilder activityQueryBuilder;
+  @Inject TransactionsReportingQueryBuilder transactionsQueryBuilder;
+  @Inject GcsUtils gcsUtils;
+  @Inject BigqueryConnection bigquery;
+
+  @Inject
+  IcannReportingStager() {}
+
+  /**
+   * Creates and stores reports of a given type on GCS.
+   *
+   * <p>This is factored out to facilitate choosing which reports to upload,
+   */
+  ImmutableList<String> stageReports(ReportType reportType) throws Exception {
+    QueryBuilder queryBuilder =
+        (reportType == ReportType.ACTIVITY) ? activityQueryBuilder : transactionsQueryBuilder;
+
+
+    ImmutableMap<String, String> viewQueryMap = queryBuilder.getViewQueryMap();
+    // Generate intermediary views
+    for (Entry<String, String> entry : viewQueryMap.entrySet()) {
+      createIntermediaryTableView(entry.getKey(), entry.getValue(), reportType);
+    }
+
+    // Get an in-memory table of the aggregate query's result
+    ImmutableTable<Integer, TableFieldSchema, Object> reportTable =
+        bigquery.queryToLocalTableSync(queryBuilder.getReportQuery());
+
+    // Get report headers from the table schema and convert into CSV format
+    String headerRow = constructRow(getHeaders(reportTable.columnKeySet()));
+    logger.infofmt("Headers: %s", headerRow);
+
+    return (reportType == ReportType.ACTIVITY)
+        ? stageActivityReports(headerRow, reportTable.rowMap().values())
+        : stageTransactionsReports(headerRow, reportTable.rowMap().values());
+  }
+
+  private void createIntermediaryTableView(String queryName, String query, ReportType reportType)
+      throws ExecutionException, InterruptedException {
+    // Later views depend on the results of earlier ones, so query everything synchronously
+    logger.infofmt("Generating intermediary view %s", queryName);
+    bigquery.query(
+        query,
+        bigquery.buildDestinationTable(queryName)
+            .description(String.format(
+                "An intermediary view to generate %s reports for this month.", reportType))
+            .type(TableType.VIEW)
+            .build()
+    ).get();
+  }
+
+  private Iterable<String> getHeaders(ImmutableSet<TableFieldSchema> fields) {
+    return fields
+        .stream()
+        .map((schema) -> schema.getName().replace('_', '-'))
+        .collect(ImmutableList.toImmutableList());
+  }
+
+  /** Creates and stores activity reports on GCS, returns a list of files stored. */
+  private ImmutableList<String> stageActivityReports(
+      String headerRow, ImmutableCollection<Map<TableFieldSchema, Object>> rows)
+      throws IOException {
+    ImmutableList.Builder<String> manifestBuilder = new ImmutableList.Builder<>();
+    // Create a report csv for each tld from query table, and upload to GCS
+    for (Map<TableFieldSchema, Object> row : rows) {
+      // Get the tld (first cell in each row)
+      String tld = row.values().iterator().next().toString();
+      if (isNullOrEmpty(tld)) {
+        throw new RuntimeException("Found an empty row in the activity report table!");
+      }
+      ImmutableList<String> rowStrings = ImmutableList.of(constructRow(row.values()));
+      // Create and upload the activity report with a single row
+      manifestBuilder.add(
+          saveReportToGcs(tld, createReport(headerRow, rowStrings), ReportType.ACTIVITY));
+    }
+    return manifestBuilder.build();
+  }
+
+  /** Creates and stores transactions reports on GCS, returns a list of files stored. */
+  private ImmutableList<String> stageTransactionsReports(
+      String headerRow, ImmutableCollection<Map<TableFieldSchema, Object>> rows)
+      throws IOException {
+    // Map from tld to rows
+    ListMultimap<String, String> tldToRows = ArrayListMultimap.create();
+    // Map from tld to totals
+    HashMap<String, List<Integer>> tldToTotals = new HashMap<>();
+    for (Map<TableFieldSchema, Object> row : rows) {
+      // Get the tld (first cell in each row)
+      String tld = row.values().iterator().next().toString();
+      if (isNullOrEmpty(tld)) {
+        throw new RuntimeException("Found an empty row in the transactions report table!");
+      }
+      tldToRows.put(tld, constructRow(row.values()));
+      // Construct totals for each tld, skipping non-summable columns (TLD, registrar name, iana-id)
+      if (!tldToTotals.containsKey(tld)) {
+        tldToTotals.put(tld, new ArrayList<>(Collections.nCopies(row.values().size() - 3, 0)));
+      }
+      addToTotal(tldToTotals.get(tld), row);
+    }
+    ImmutableList.Builder<String> manifestBuilder = new ImmutableList.Builder<>();
+    // Create and upload a transactions report for each tld via its rows
+    for (String tld : tldToRows.keySet()) {
+      // Append the totals row
+      tldToRows.put(tld, constructTotalRow(tldToTotals.get(tld)));
+      manifestBuilder.add(
+          saveReportToGcs(
+              tld, createReport(headerRow, tldToRows.get(tld)), ReportType.TRANSACTIONS));
+    }
+    return manifestBuilder.build();
+  }
+
+  /** Adds a row's values to an existing list of integers (totals). */
+  private void addToTotal(List<Integer> totals, Map<TableFieldSchema, Object> row) {
+    List<Integer> rowVals =
+        row.values()
+            .stream()
+            // Ignore TLD, Registrar name and IANA id
+            .skip(3)
+            .map((Object o) -> Integer.parseInt(o.toString()))
+            .collect(Collectors.toList());
+    checkState(
+        rowVals.size() == totals.size(),
+        "Number of elements in totals not equal to number of elements in row!");
+    for (int i = 0; i < rowVals.size(); i++) {
+      totals.set(i, totals.get(i) + rowVals.get(i));
+    }
+  }
+
+  /** Returns a list of integers (totals) as a comma separated string. */
+  private String constructTotalRow(List<Integer> totals) {
+    StringBuilder rowString = new StringBuilder("Totals,,");
+    rowString.append(
+        totals.stream().map((Integer i) -> i.toString()).collect(Collectors.joining(",")));
+    return rowString.toString();
+  }
+
+  /**
+   * Makes a row of the report by appending the string representation of all objects in an iterable
+   * with commas separating individual fields.
+   *
+   * <p>This discards the first object, which is assumed to be the TLD field.
+   * */
+  private String constructRow(Iterable<? extends Object> iterable) {
+    Iterator<? extends Object> rowIter = iterable.iterator();
+    StringBuilder rowString = new StringBuilder();
+    // Skip the TLD column
+    rowIter.next();
+    while (rowIter.hasNext()) {
+      rowString.append(String.format("%s,", rowIter.next().toString()));
+    }
+    // Remove trailing comma
+    rowString.deleteCharAt(rowString.length() - 1);
+    return rowString.toString();
+  }
+
+  /**
+   * Constructs a report given its headers and rows as a string.
+   *
+   * <p>Note that activity reports will only have one row, while transactions reports may have
+   * multiple rows.
+   */
+  private String createReport(String headers, List<String> rows) {
+    StringBuilder reportCsv = new StringBuilder(headers);
+    for (String row : rows) {
+      // Add CRLF between rows per ICANN specification
+      reportCsv.append("\r\n");
+      reportCsv.append(row);
+    }
+    logger.infofmt("Created report:\n%s", reportCsv.toString());
+    return reportCsv.toString();
+  }
+
+  /** Stores a report on GCS, returning the name of the file stored. */
+  private String saveReportToGcs(String tld, String reportCsv, ReportType reportType)
+      throws IOException {
+    // Upload resulting CSV file to GCS
+    byte[] reportBytes = reportCsv.getBytes(UTF_8);
+    String reportFilename = ReportingUtils.createFilename(tld, yearMonth, reportType);
+    String reportBucketname = ReportingUtils.createReportingBucketName(reportingBucket, subdir);
+    final GcsFilename gcsFilename = new GcsFilename(reportBucketname, reportFilename);
+    gcsUtils.createFromBytes(gcsFilename, reportBytes);
+    logger.infofmt(
+        "Wrote %d bytes to file location %s",
+        reportBytes.length,
+        gcsFilename.toString());
+    return reportFilename;
+  }
+
+  /** Creates and stores a manifest file on GCS, indicating which reports were generated. */
+  void createAndUploadManifest(ImmutableList<String> filenames) throws IOException {
+    String reportBucketname = ReportingUtils.createReportingBucketName(reportingBucket, subdir);
+    final GcsFilename gcsFilename = new GcsFilename(reportBucketname, MANIFEST_FILE_NAME);
+    StringBuilder manifestString = new StringBuilder();
+    filenames.forEach((filename) -> manifestString.append(filename).append("\n"));
+    gcsUtils.createFromBytes(gcsFilename, manifestString.toString().getBytes(UTF_8));
+    logger.infofmt(
+        "Wrote %d filenames to manifest at %s", filenames.size(), gcsFilename.toString());
+  }
+}

java/google/registry/reporting/IcannReportingStagingAction.java

@@ -14,99 +14,72 @@
 
 package google.registry.reporting;
 
-import static com.google.common.base.Strings.isNullOrEmpty;
 import static google.registry.request.Action.Method.POST;
-import static java.nio.charset.StandardCharsets.UTF_8;
 import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
 
-import com.google.api.services.bigquery.model.TableFieldSchema;
-import com.google.appengine.tools.cloudstorage.GcsFilename;
 import com.google.common.base.Throwables;
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.ImmutableTable;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.ListMultimap;
 import com.google.common.net.MediaType;
-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.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 java.io.OutputStream;
 import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Optional;
-import java.util.concurrent.ExecutionException;
 import javax.inject.Inject;
 
 /**
  * Action that generates monthly ICANN activity and transactions reports.
  *
- * <p> The reports are then uploaded to GCS under
- * gs://domain-registry-reporting/icann/monthly/YYYY-MM
+ * <p>The reports are stored in GCS under gs://[project-id]-reporting/[subdir]. We also store a
+ * MANIFEST.txt file that contains a list of filenames generated, to facilitate subsequent uploads.
+ *
+ * <p>Parameters:
+ *
+ * <p>yearMonth: the reporting month in yyyy-MM format. Defaults to the previous month at runtime
+ * (i.e. a run on 2017-09-01 defaults to 2017-08's reports).
+ *
+ * <p>subdir: the subdirectory of gs://[project-id]-reporting/ to upload to. For example:
+ * "manual/dir" means reports will be stored under gs://[project-id]-reporting/manual/dir. Defaults
+ * to "icann/monthly/[yearMonth]".
+ *
+ * <p>reportTypes: the type of reports to generate. You can specify either 'activity' or
+ * 'transactions'. Defaults to generating both.
  */
-@Action(
-    path = IcannReportingStagingAction.PATH,
-    method = POST,
-    auth = Auth.AUTH_INTERNAL_ONLY
-)
+@Action(path = IcannReportingStagingAction.PATH, method = POST, auth = Auth.AUTH_INTERNAL_ONLY)
 public final class IcannReportingStagingAction implements Runnable {
 
   static final String PATH = "/_dr/task/icannReportingStaging";
 
   private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
 
-  @Inject @Config("icannReportingBucket") String reportingBucket;
-  @Inject @Parameter(IcannReportingModule.PARAM_YEAR_MONTH) String yearMonth;
-  @Inject @Parameter(IcannReportingModule.PARAM_SUBDIR) Optional<String> subdir;
-  @Inject @Parameter(IcannReportingModule.PARAM_REPORT_TYPE) ReportType reportType;
-  @Inject QueryBuilder queryBuilder;
-  @Inject BigqueryConnection bigquery;
-  @Inject GcsUtils gcsUtils;
+  @Inject
+  @Parameter(IcannReportingModule.PARAM_REPORT_TYPE)
+  ImmutableList<ReportType> reportTypes;
+
+  @Inject IcannReportingStager stager;
   @Inject Response response;
   @Inject IcannReportingStagingAction() {}
 
   @Override
   public void run() {
     try {
-      ImmutableMap<String, String> viewQueryMap = queryBuilder.getViewQueryMap();
-      // Generate intermediary views
-      for (Entry<String, String> entry : viewQueryMap.entrySet()) {
-        createIntermediaryTableView(entry.getKey(), entry.getValue());
+      ImmutableList.Builder<String> manifestedFilesBuilder = new ImmutableList.Builder<>();
+      for (ReportType reportType : reportTypes) {
+        manifestedFilesBuilder.addAll(stager.stageReports(reportType));
       }
+      ImmutableList<String> manifestedFiles = manifestedFilesBuilder.build();
+      stager.createAndUploadManifest(manifestedFiles);
 
-      // Get an in-memory table of the aggregate query's result
-      ImmutableTable<Integer, TableFieldSchema, Object> reportTable =
-          bigquery.queryToLocalTableSync(queryBuilder.getReportQuery());
-
-      // Get report headers from the table schema and convert into CSV format
-      String headerRow = constructRow(getHeaders(reportTable.columnKeySet()));
-      logger.infofmt("Headers: %s", headerRow);
-
-      if (reportType == ReportType.ACTIVITY) {
-        stageActivityReports(headerRow, reportTable.rowMap().values());
-      } else {
-        stageTransactionsReports(headerRow, reportTable.rowMap().values());
-      }
+      logger.infofmt("Completed staging %d report files.", manifestedFiles.size());
       response.setStatus(SC_OK);
       response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
       response.setPayload("Completed staging action.");
     } catch (Exception e) {
-      logger.warning(Throwables.getStackTraceAsString(e));
+      logger.severe("Reporting staging action failed!");
+      logger.severe(Throwables.getStackTraceAsString(e));
       response.setStatus(SC_INTERNAL_SERVER_ERROR);
       response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
       response.setPayload(
@@ -114,109 +87,4 @@ public final class IcannReportingStagingAction implements Runnable {
               Arrays.toString(e.getStackTrace())));
     }
   }
-
-  private void createIntermediaryTableView(String queryName, String query)
-      throws ExecutionException, InterruptedException {
-    // Later views depend on the results of earlier ones, so query everything synchronously
-    bigquery.query(
-        query,
-        bigquery.buildDestinationTable(queryName)
-            .description(String.format(
-                "An intermediary view to generate %s reports for this month.", reportType))
-            .type(TableType.VIEW)
-            .build()
-    ).get();
-  }
-
-  private Iterable<String> getHeaders(ImmutableSet<TableFieldSchema> fields) {
-    return Iterables.transform(fields, schema -> schema.getName().replace('_', '-'));
-  }
-
-  private void stageActivityReports (
-      String headerRow, ImmutableCollection<Map<TableFieldSchema, Object>> rows)
-      throws IOException {
-    // Create a report csv for each tld from query table, and upload to GCS
-    for (Map<TableFieldSchema, Object> row : rows) {
-      // Get the tld (first cell in each row)
-      String tld = row.values().iterator().next().toString();
-      if (isNullOrEmpty(tld)) {
-        throw new RuntimeException("Found an empty row in the activity report table!");
-      }
-      ImmutableList<String> rowStrings = ImmutableList.of(constructRow(row.values()));
-      // Create and upload the activity report with a single row
-      uploadReport(tld, createReport(headerRow, rowStrings));
-    }
-  }
-
-  private void stageTransactionsReports(
-      String headerRow, ImmutableCollection<Map<TableFieldSchema, Object>> rows)
-      throws IOException {
-    // Map from tld to rows
-    ListMultimap<String, String> tldToRows = ArrayListMultimap.create();
-    for (Map<TableFieldSchema, Object> row : rows) {
-      // Get the tld (first cell in each row)
-      String tld = row.values().iterator().next().toString();
-      if (isNullOrEmpty(tld)) {
-        throw new RuntimeException("Found an empty row in the activity report table!");
-      }
-      tldToRows.put(tld, constructRow(row.values()));
-    }
-    // Create and upload a transactions report for each tld via its rows
-    for (String tld : tldToRows.keySet()) {
-      uploadReport(tld, createReport(headerRow, tldToRows.get(tld)));
-    }
-  }
-
-  /**
-   * Makes a row of the report by appending the string representation of all objects in an iterable
-   * with commas separating individual fields.
-   *
-   * <p>This discards the first object, which is assumed to be the TLD field.
-   * */
-  private String constructRow(Iterable<? extends Object> iterable) {
-    Iterator<? extends Object> rowIter = iterable.iterator();
-    StringBuilder rowString = new StringBuilder();
-    // Skip the TLD column
-    rowIter.next();
-    while (rowIter.hasNext()) {
-      rowString.append(String.format("%s,", rowIter.next().toString()));
-    }
-    // Remove trailing comma
-    rowString.deleteCharAt(rowString.length() - 1);
-    return rowString.toString();
-  }
-
-  /**
-   * Constructs a report given its headers and rows as a string.
-   *
-   * <p>Note that activity reports will only have one row, while transactions reports may have
-   * multiple rows.
-   */
-  private String createReport(String headers, List<String> rows) {
-    StringBuilder reportCsv = new StringBuilder(headers);
-    for (String row : rows) {
-      // Add CRLF between rows per ICANN specification
-      reportCsv.append("\r\n");
-      reportCsv.append(row);
-    }
-    logger.infofmt("Created %s report:\n%s", reportType, reportCsv.toString());
-    return reportCsv.toString();
-  }
-
-  private void uploadReport(String tld, String reportCsv) throws IOException {
-    // Upload resulting CSV file to GCS
-    byte[] reportBytes = reportCsv.getBytes(UTF_8);
-    String reportFilename =
-        IcannReportingUploadAction.createFilename(tld, yearMonth, reportType);
-    String reportBucketname =
-        IcannReportingUploadAction.createReportingBucketName(reportingBucket, subdir, yearMonth);
-    final GcsFilename gcsFilename = new GcsFilename(reportBucketname, reportFilename);
-    try (OutputStream gcsOutput = gcsUtils.openOutputStream(gcsFilename)) {
-      gcsOutput.write(reportBytes);
-    }
-    logger.infofmt(
-        "Wrote %d bytes to file location %s",
-        reportBytes.length,
-        gcsFilename.toString());
-  }
 }

java/google/registry/reporting/IcannReportingUploadAction.java

@@ -14,115 +14,108 @@
 
 package google.registry.reporting;
 
-import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.net.MediaType.PLAIN_TEXT_UTF_8;
-import static google.registry.model.registry.Registries.assertTldExists;
+import static google.registry.reporting.IcannReportingModule.MANIFEST_FILE_NAME;
 import static google.registry.request.Action.Method.POST;
+import static java.nio.charset.StandardCharsets.UTF_8;
 
 import com.google.appengine.tools.cloudstorage.GcsFilename;
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableList;
 import com.google.common.io.ByteStreams;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.gcs.GcsUtils;
-import google.registry.reporting.IcannReportingModule.ReportType;
 import google.registry.request.Action;
 import google.registry.request.Parameter;
-import google.registry.request.RequestParameters;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.util.FormattingLogger;
 import google.registry.util.Retrier;
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.Optional;
 import javax.inject.Inject;
 
 /**
- * Action that uploads the monthly transaction and activity reports from Cloud Storage to ICANN via
- * an HTTP PUT.
+ * Action that uploads the monthly activity/transactions reports from GCS to ICANN via an HTTP PUT.
  *
+ * <p>This should be run after {@link IcannReportingStagingAction}, which writes out the month's
+ * reports and a MANIFEST.txt file. This action reads the filenames from the MANIFEST.txt, and
+ * attempts to upload every file in the manifest to ICANN's endpoint.
+ *
+ * <p>Parameters:
+ *
+ * <p>subdir: the subdirectory of gs://[project-id]-reporting/ to retrieve reports from. For
+ * example: "manual/dir" means reports will be stored under gs://[project-id]-reporting/manual/dir.
+ * Defaults to "icann/monthly/[last month in yyyy-MM format]".
  */
-@Action(
-  path = IcannReportingUploadAction.PATH,
-  method = POST,
-  auth = Auth.AUTH_INTERNAL_OR_ADMIN
-)
+@Action(path = IcannReportingUploadAction.PATH, method = POST, auth = Auth.AUTH_INTERNAL_OR_ADMIN)
 public final class IcannReportingUploadAction implements Runnable {
 
   static final String PATH = "/_dr/task/icannReportingUpload";
-  static final String DEFAULT_SUBDIR = "icann/monthly";
 
   private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
 
-  @Inject @Config("icannReportingBucket") String icannReportingBucket;
-  @Inject @Parameter(RequestParameters.PARAM_TLD) String tld;
-  @Inject @Parameter(IcannReportingModule.PARAM_YEAR_MONTH) String yearMonth;
-  @Inject @Parameter(IcannReportingModule.PARAM_REPORT_TYPE) ReportType reportType;
-  @Inject @Parameter(IcannReportingModule.PARAM_SUBDIR) Optional<String> subdir;
+  @Inject
+  @Config("icannReportingBucket")
+  String reportingBucket;
+
+  @Inject
+  @Parameter(IcannReportingModule.PARAM_SUBDIR)
+  String subdir;
+
   @Inject GcsUtils gcsUtils;
   @Inject IcannHttpReporter icannReporter;
-  @Inject Response response;
   @Inject Retrier retrier;
+  @Inject Response response;
 
   @Inject
   IcannReportingUploadAction() {}
 
   @Override
   public void run() {
-    validateParams();
-    String reportFilename = createFilename(tld, yearMonth, reportType);
-    String reportBucketname = createReportingBucketName(icannReportingBucket, subdir, yearMonth);
-    logger.infofmt("Reading ICANN report %s from bucket %s", reportFilename, reportBucketname);
-    final GcsFilename gcsFilename = new GcsFilename(reportBucketname, reportFilename);
-    checkState(
-        gcsUtils.existsAndNotEmpty(gcsFilename),
-        "ICANN report object %s in bucket %s not found",
-        gcsFilename.getObjectName(),
-        gcsFilename.getBucketName());
+    String reportBucketname = ReportingUtils.createReportingBucketName(reportingBucket, subdir);
+    ImmutableList<String> manifestedFiles = getManifestedFiles(reportBucketname);
+    // Report on all manifested files
+    for (String reportFilename : manifestedFiles) {
+      logger.infofmt("Reading ICANN report %s from bucket %s", reportFilename, reportBucketname);
+      final GcsFilename gcsFilename = new GcsFilename(reportBucketname, reportFilename);
+      verifyFileExists(gcsFilename);
+      retrier.callWithRetry(
+          () -> {
+            final byte[] payload = readBytesFromGcs(gcsFilename);
+            icannReporter.send(payload, reportFilename);
+            response.setContentType(PLAIN_TEXT_UTF_8);
+            response.setPayload(String.format("OK, sending: %s", new String(payload, UTF_8)));
+            return null;
+          },
+          IOException.class);
+    }
+  }
 
-    retrier.callWithRetry(
-        () -> {
-          final byte[] payload = readReportFromGcs(gcsFilename);
-          icannReporter.send(payload, tld, yearMonth, reportType);
-          response.setContentType(PLAIN_TEXT_UTF_8);
-          response.setPayload(
-              String.format("OK, sending: %s", new String(payload, StandardCharsets.UTF_8)));
-          return null;
-        },
+  private ImmutableList<String> getManifestedFiles(String reportBucketname) {
+    GcsFilename manifestFilename = new GcsFilename(reportBucketname, MANIFEST_FILE_NAME);
+    verifyFileExists(manifestFilename);
+    return retrier.callWithRetry(
+        () ->
+            ImmutableList.copyOf(
+                Splitter.on('\n')
+                    .omitEmptyStrings()
+                    .split(new String(readBytesFromGcs(manifestFilename), UTF_8))),
         IOException.class);
   }
 
-  private byte[] readReportFromGcs(GcsFilename reportFilename) throws IOException {
+  private byte[] readBytesFromGcs(GcsFilename reportFilename) throws IOException {
     try (InputStream gcsInput = gcsUtils.openInputStream(reportFilename)) {
       return ByteStreams.toByteArray(gcsInput);
     }
   }
 
-  static String createFilename(String tld, String yearMonth, ReportType reportType) {
-    // Report files use YYYYMM naming instead of standard YYYY-MM, per ICANN requirements.
-    String fileYearMonth = yearMonth.substring(0, 4) + yearMonth.substring(5, 7);
-    return String.format("%s-%s-%s.csv", tld, reportType.toString().toLowerCase(), fileYearMonth);
-  }
-
-  static String createReportingBucketName(
-      String reportingBucket, Optional<String> subdir, String yearMonth) {
-    return subdir.isPresent()
-        ? String.format("%s/%s", reportingBucket, subdir.get())
-        : String.format("%s/%s/%s", reportingBucket, DEFAULT_SUBDIR, yearMonth);
-
-  }
-
-  private void validateParams() {
-    assertTldExists(tld);
-    checkState(
-        yearMonth.matches("[0-9]{4}-[0-9]{2}"),
-        "yearMonth must be in YYYY-MM format, got %s instead.",
-        yearMonth);
-    if (subdir.isPresent()) {
-      checkState(
-          !subdir.get().startsWith("/") && !subdir.get().endsWith("/"),
-          "subdir must not start or end with a \"/\", got %s instead.",
-          subdir.get());
-    }
+  private void verifyFileExists(GcsFilename gcsFilename) {
+    checkArgument(
+        gcsUtils.existsAndNotEmpty(gcsFilename),
+        "Object %s in bucket %s not found",
+        gcsFilename.getObjectName(),
+        gcsFilename.getBucketName());
   }
 }

java/google/registry/reporting/ReportingUtils.java

@@ -0,0 +1,34 @@
+// 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 com.google.common.base.Ascii;
+import google.registry.reporting.IcannReportingModule.ReportType;
+
+/** Static utils for reporting. */
+public final class ReportingUtils {
+
+  /** Generates a report filename in accord with ICANN's specifications. */
+  static String createFilename(String tld, String yearMonth, ReportType reportType) {
+    // Report files use YYYYMM naming instead of standard YYYY-MM, per ICANN requirements.
+    return String.format(
+        "%s-%s-%s.csv", tld, Ascii.toLowerCase(reportType.toString()), yearMonth.replace("-", ""));
+  }
+
+  /** Constructs the bucket name to store/upload reports to. */
+  static String createReportingBucketName(String reportingBucket, String subdir) {
+    return String.format("%s/%s", reportingBucket, subdir);
+  }
+}

java/google/registry/reporting/TransactionsReportingQueryBuilder.java

@@ -158,6 +158,8 @@ public final class TransactionsReportingQueryBuilder implements QueryBuilder {
     String aggregateQuery =
         SqlTemplate.create(getQueryFromFile("transactions_report_aggregation.sql"))
             .put("PROJECT_ID", projectId)
+            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+            .put("REGISTRY_TABLE", "Registry")
             .put("ICANN_REPORTING_DATA_SET", ICANN_REPORTING_DATA_SET)
             .put("REGISTRAR_IANA_ID_TABLE", getTableName(REGISTRAR_IANA_ID))
             .put("TOTAL_DOMAINS_TABLE", getTableName(TOTAL_DOMAINS))

java/google/registry/reporting/sql/activity_report_aggregation.sql

@@ -19,8 +19,6 @@
 SELECT
   RealTlds.tld AS tld,
   SUM(IF(metricName = 'operational-registrars', count, 0)) AS operational_registrars,
-  SUM(IF(metricName = 'ramp-up-registrars', count, 0)) AS ramp_up_registrars,
-  SUM(IF(metricName = 'pre-ramp-up-registrars', count, 0)) AS pre_ramp_up_registrars,
   -- We use the Centralized Zone Data Service.
   "CZDS" AS zfa_passwords,
   SUM(IF(metricName = 'whois-43-queries', count, 0)) AS whois_43_queries,
@@ -65,7 +63,7 @@ SELECT
   -- filter so that only metrics with that TLD or a NULL TLD are counted
   -- towards a given TLD.
 FROM (
-SELECT tldStr as tld
+SELECT tldStr AS tld
 FROM `%PROJECT_ID%.%DATASTORE_EXPORT_DATA_SET%.%REGISTRY_TABLE%`
 WHERE tldType = 'REAL'
 ) as RealTlds

java/google/registry/reporting/sql/dns_counts.sql

@@ -15,14 +15,13 @@
 
   -- Query for DNS metrics.
 
-  -- This is a no-op until after we transition to Google Cloud DNS, which
-  -- will likely export metrics via Stackdriver.
+  -- You must configure this yourself to enable activity reporting, according
+  -- to whatever metrics your DNS provider makes available. We hope to make
+  -- this available in the open-source build in the near future.
 
 SELECT
-  -- DNS metrics apply to all tlds, which requires the 'null' magic value.
   STRING(NULL) AS tld,
   metricName,
-  -- TODO(b/63388735): Change this to actually query Google Cloud DNS when ready.
   -1 AS count
 FROM ((
   SELECT 'dns-udp-queries' AS metricName)

java/google/registry/reporting/sql/registrar_iana_id.sql

@@ -26,5 +26,5 @@ FROM
   UNNEST(allowedTlds) as allowed_tlds
 WHERE (type = 'REAL' OR type = 'INTERNAL')
 -- Filter out prober data
-AND NOT ENDS_WITH(allowed_tlds, "test")
+AND NOT ENDS_WITH(allowed_tlds, ".test")
 ORDER BY tld, registrarName

java/google/registry/reporting/sql/registrar_operating_status.sql

@@ -23,5 +23,5 @@ SELECT
 FROM
   `%PROJECT_ID%.%DATASTORE_EXPORT_DATA_SET%.%REGISTRAR_TABLE%`
 WHERE
-  type = 'REAL'
+  (type = 'REAL' OR type = 'INTERNAL')
 GROUP BY metricName

java/google/registry/reporting/sql/total_domains.sql

@@ -32,7 +32,7 @@ JOIN
 ON
   currentSponsorClientId = registrar_table.__key__.name
 WHERE
-  domain_table._d = "DomainResource"
-  AND (registrar_table.type = "REAL" OR registrar_table.type = "INTERNAL")
+  domain_table._d = 'DomainResource'
+  AND (registrar_table.type = 'REAL' OR registrar_table.type = 'INTERNAL')
 GROUP BY tld, registrarName
 ORDER BY tld, registrarName

java/google/registry/reporting/sql/transaction_counts.sql

@@ -65,8 +65,6 @@ FROM (
   WHERE reportingTime
   BETWEEN TIMESTAMP('%EARLIEST_REPORT_TIME%')
   AND TIMESTAMP('%LATEST_REPORT_TIME%')
-    -- Ignore prober data
-  AND NOT ENDS_WITH(tld, "test")
   GROUP BY
     tld,
     clientId,

java/google/registry/reporting/sql/transactions_report_aggregation.sql

@@ -20,7 +20,8 @@
 
 SELECT
   registrars.tld as tld,
-  registrars.registrar_name as registrar_name,
+  -- Surround registrar names with quotes to handle names containing a comma.
+  FORMAT("\"%s\"", registrars.registrar_name) as registrar_name,
   registrars.iana_id as iana_id,
   SUM(IF(metrics.metricName = 'TOTAL_DOMAINS', metrics.metricValue, 0)) AS total_domains,
   SUM(IF(metrics.metricName = 'TOTAL_NAMESERVERS', metrics.metricValue, 0)) AS total_nameservers,
@@ -62,9 +63,16 @@ SELECT
   0 AS agp_exemptions_granted,
   0 AS agp_exempted_domains,
   SUM(IF(metrics.metricName = 'ATTEMPTED_ADDS', metrics.metricValue, 0)) AS attempted_adds
-FROM (
-  SELECT *
-  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%REGISTRAR_IANA_ID_TABLE%`) AS registrars
+FROM
+-- Only produce reports for real TLDs
+(SELECT tldStr AS tld
+  FROM `%PROJECT_ID%.%DATASTORE_EXPORT_DATA_SET%.%REGISTRY_TABLE%`
+  WHERE tldType = 'REAL') AS registries
+JOIN
+(SELECT *
+  FROM `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%REGISTRAR_IANA_ID_TABLE%`)
+  AS registrars
+ON registries.tld = registrars.tld
 -- We LEFT JOIN to produce reports even if the registrar made no transactions
 LEFT OUTER JOIN (
   -- Gather all intermediary data views

java/google/registry/request/RequestParameters.java

@@ -98,6 +98,20 @@ public final class RequestParameters {
     return parameters == null ? ImmutableSet.<String>of() : ImmutableSet.copyOf(parameters);
   }
 
+  /**
+   * Returns the first GET or POST parameter associated with {@code name}, absent otherwise.
+   *
+   * @throws BadRequestException if request parameter named {@code name} is not equal to any of the
+   *     values in {@code enumClass}
+   */
+  public static <C extends Enum<C>> Optional<C> extractOptionalEnumParameter(
+      HttpServletRequest req, Class<C> enumClass, String name) {
+    String stringParam = req.getParameter(name);
+    return isNullOrEmpty(stringParam)
+        ? Optional.empty()
+        : Optional.of(extractEnumParameter(req, enumClass, name));
+  }
+
   /**
    * Returns the first GET or POST parameter associated with {@code name}.
    *