From 477617eec99ec6eeaea6b755aa8de3b460da9455 Mon Sep 17 00:00:00 2001
From: larryruili <larryruili@google.com>
Date: Tue, 8 Aug 2017 15:46:59 -0700
Subject: [PATCH] Add activity report generation code

This adds Bigquery API client code to generate the activity reports from our
now standardSQL queries. The naming mirrors that of RDE (Staging generates the
reports and uploads them to GCS).

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=164656344
---
 .../registry/bigquery/BigqueryConnection.java |  23 ++-
 .../env/common/backend/WEB-INF/web.xml        |  11 ++
 .../backend/BackendRequestComponent.java      |   2 +
 .../ActivityReportingQueryBuilder.java        |  84 +++++---
 java/google/registry/reporting/BUILD          |   4 +
 .../reporting/IcannReportingModule.java       |  36 ++++
 .../IcannReportingStagingAction.java          | 185 ++++++++++++++++++
 .../reporting/IcannReportingUploadAction.java |  27 +--
 .../sql/activity_report_aggregation.sql       |  59 +++---
 .../registry/reporting/sql/dns_counts.sql     |   1 +
 .../registry/reporting/sql/epp_metrics.sql    |   1 +
 .../registry/reporting/sql/monthly_logs.sql   |   1 +
 .../sql/registrar_operating_status.sql        |   1 +
 .../registry/reporting/sql/whois_counts.sql   |   1 +
 .../backend/testdata/backend_routing.txt      |   1 +
 .../ActivityReportingQueryBuilderTest.java    |  38 ++--
 javatests/google/registry/reporting/BUILD     |   2 +
 .../IcannReportingStagingActionTest.java      | 149 ++++++++++++++
 .../IcannReportingUploadActionTest.java       |  35 +++-
 .../activity_report_aggregation_test.sql      |  59 +++---
 .../reporting/testdata/dns_counts_test.sql    |   1 +
 .../reporting/testdata/epp_metrics_test.sql   |   3 +-
 .../reporting/testdata/monthly_logs_test.sql  |   1 +
 .../registrar_operating_status_test.sql       |   1 +
 .../reporting/testdata/whois_counts_test.sql  |   3 +-
 25 files changed, 612 insertions(+), 117 deletions(-)
 create mode 100644 java/google/registry/reporting/IcannReportingStagingAction.java
 create mode 100644 javatests/google/registry/reporting/IcannReportingStagingActionTest.java

diff --git a/java/google/registry/bigquery/BigqueryConnection.java b/java/google/registry/bigquery/BigqueryConnection.java
index e19b1829d..38ef5e9ef 100644
--- a/java/google/registry/bigquery/BigqueryConnection.java
+++ b/java/google/registry/bigquery/BigqueryConnection.java
@@ -466,6 +466,25 @@ public class BigqueryConnection implements AutoCloseable {
         directExecutor());
   }
 
+  /**
+   * Returns the result of calling queryToLocalTable, but synchronously to avoid spawning new
+   * background threads, which App Engine doesn't support.
+   *
+   * @see <a href="https://cloud.google.com/appengine/docs/standard/java/runtime#Threads">
+   *   App Engine Runtime</a>
+   *
+   * <p>Returns the results of the query in an ImmutableTable on success.
+   */
+  public ImmutableTable<Integer, TableFieldSchema, Object> queryToLocalTableSync(String querySql)
+      throws Exception {
+    Job job = new Job()
+        .setConfiguration(new JobConfiguration()
+            .setQuery(new JobConfigurationQuery()
+                .setQuery(querySql)
+                .setDefaultDataset(getDataset())));
+    return getQueryResults(runJob(job));
+  }
+
   /**
    * Returns the query results for the given job as an ImmutableTable, row-keyed by row number
    * (indexed from 1), column-keyed by the TableFieldSchema for that field, and with the value
@@ -590,7 +609,7 @@ public class BigqueryConnection implements AutoCloseable {
   }
 
   /**
-   * Lanuch a job, wait for it to complete, but <i>do not</i> check for errors.
+   * Launch a job, wait for it to complete, but <i>do not</i> check for errors.
    *
    * @throws BigqueryJobFailureException
    */
@@ -599,7 +618,7 @@ public class BigqueryConnection implements AutoCloseable {
   }
 
   /**
-   * Lanuch a job, but do not wait for it to complete.
+   * Launch a job, but do not wait for it to complete.
    *
    * @throws BigqueryJobFailureException
    */
diff --git a/java/google/registry/env/common/backend/WEB-INF/web.xml b/java/google/registry/env/common/backend/WEB-INF/web.xml
index 716ab146c..d976425d5 100644
--- a/java/google/registry/env/common/backend/WEB-INF/web.xml
+++ b/java/google/registry/env/common/backend/WEB-INF/web.xml
@@ -57,6 +57,17 @@
     <url-pattern>/_dr/task/brdaCopy</url-pattern>
   </servlet-mapping>
 
+  <!--
+    Monthly ICANN transaction and activity reports. This task generates report
+    files (in CSV format) and stores them in GCS under
+    gs://domain-registry-reporting/icann/monthly/YYYY-MM
+    by default.
+  -->
+  <servlet-mapping>
+    <servlet-name>backend-servlet</servlet-name>
+    <url-pattern>/_dr/task/icannReportingStaging</url-pattern>
+  </servlet-mapping>
+
   <!--
     Monthly ICANN transaction and activity reports. This task uploads the generated
     report files (in CSV format) via an HTTP PUT to ICANN's endpoint.
diff --git a/java/google/registry/module/backend/BackendRequestComponent.java b/java/google/registry/module/backend/BackendRequestComponent.java
index 177664104..cf4543b02 100644
--- a/java/google/registry/module/backend/BackendRequestComponent.java
+++ b/java/google/registry/module/backend/BackendRequestComponent.java
@@ -64,6 +64,7 @@ import google.registry.rde.imports.RdeHostImportAction;
 import google.registry.rde.imports.RdeHostLinkAction;
 import google.registry.rde.imports.RdeImportsModule;
 import google.registry.reporting.IcannReportingModule;
+import google.registry.reporting.IcannReportingStagingAction;
 import google.registry.reporting.IcannReportingUploadAction;
 import google.registry.request.RequestComponentBuilder;
 import google.registry.request.RequestModule;
@@ -112,6 +113,7 @@ interface BackendRequestComponent {
   ExportDomainListsAction exportDomainListsAction();
   ExportReservedTermsAction exportReservedTermsAction();
   ExportSnapshotAction exportSnapshotAction();
+  IcannReportingStagingAction icannReportingStagingAction();
   IcannReportingUploadAction icannReportingUploadAction();
   LoadSnapshotAction loadSnapshotAction();
   MapreduceEntityCleanupAction mapreduceEntityCleanupAction();
diff --git a/java/google/registry/reporting/ActivityReportingQueryBuilder.java b/java/google/registry/reporting/ActivityReportingQueryBuilder.java
index 2aeae3f0e..90516216b 100644
--- a/java/google/registry/reporting/ActivityReportingQueryBuilder.java
+++ b/java/google/registry/reporting/ActivityReportingQueryBuilder.java
@@ -16,10 +16,13 @@ package google.registry.reporting;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.io.Resources;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.request.Parameter;
 import google.registry.util.ResourceUtils;
 import google.registry.util.SqlTemplate;
 import java.io.IOException;
 import java.net.URL;
+import javax.inject.Inject;
 import org.joda.time.LocalDate;
 import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
@@ -31,24 +34,44 @@ public final class ActivityReportingQueryBuilder {
 
   // Names for intermediary tables for overall activity reporting query.
   static final String ICANN_REPORTING_DATA_SET = "icann_reporting";
-  static final String MONTHLY_LOGS_TABLE = "monthly_logs";
+  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 EPP_METRICS = "epp_metrics";
   static final String WHOIS_COUNTS = "whois_counts";
 
-  /** Sets the month we're doing activity reporting for, and initializes the query map. */
-  static ImmutableMap<String, String> getQueryMap(
-      LocalDate reportingMonth, String projectId) throws IOException {
-    // Convert reportingMonth into YYYYMM01 format for Bigquery table partition pattern-matching.
-    DateTimeFormatter formatter = DateTimeFormat.forPattern("YYYYMM01");
-    String startOfMonth = formatter.print(reportingMonth);
-    String endOfMonth = formatter.print(reportingMonth.plusMonths(1));
-    return createQueryMap(startOfMonth, endOfMonth, projectId);
+  @Inject @Config("projectId") String projectId;
+  @Inject @Parameter(IcannReportingModule.PARAM_YEAR_MONTH) String yearMonth;
+  @Inject ActivityReportingQueryBuilder() {}
+
+  /** Returns the aggregate query which generates the activity report from the saved view. */
+  String getActivityReportQuery() throws IOException {
+    return String.format(
+        "#standardSQL\nSELECT * FROM `%s.%s.%s`",
+        projectId,
+        ICANN_REPORTING_DATA_SET,
+        getTableName(ACTIVITY_REPORT_AGGREGATION));
   }
 
-  private static ImmutableMap<String, String> createQueryMap(
-      String startOfMonth, String endOfMonth, String projectId) throws IOException {
+  /** 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("-", ""));
+  }
+
+  /** Sets the month we're doing activity reporting for, and returns the view query map. */
+  ImmutableMap<String, String> getViewQueryMap() throws IOException {
+    LocalDate reportDate = DateTimeFormat.forPattern("yyyy-MM").parseLocalDate(yearMonth);
+    // Convert reportingMonth into YYYYMM01 format for Bigquery table partition pattern-matching.
+    DateTimeFormatter formatter = DateTimeFormat.forPattern("YYYYMM01");
+    String startOfMonth = formatter.print(reportDate);
+    String endOfMonth = formatter.print(reportDate.plusMonths(1));
+    return createQueryMap(startOfMonth, endOfMonth);
+  }
+
+  /** Returns a map from view name to its associated SQL query. */
+  private ImmutableMap<String, String> createQueryMap(
+      String startOfMonth, String endOfMonth) throws IOException {
 
     ImmutableMap.Builder<String, String> queriesBuilder = ImmutableMap.builder();
     String operationalRegistrarsQuery =
@@ -57,12 +80,12 @@ public final class ActivityReportingQueryBuilder {
             .put("REGISTRAR_DATA_SET", "registrar_data")
             .put("REGISTRAR_STATUS_TABLE", "registrar_status")
             .build();
-    queriesBuilder.put(REGISTRAR_OPERATING_STATUS, operationalRegistrarsQuery);
+    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(DNS_COUNTS, dnsCountsQuery);
+    queriesBuilder.put(getTableName(DNS_COUNTS), dnsCountsQuery);
 
     // The monthly logs query is a shared dependency for epp counts and whois metrics
     String monthlyLogsQuery =
@@ -73,35 +96,37 @@ public final class ActivityReportingQueryBuilder {
             .put("START_OF_MONTH", startOfMonth)
             .put("END_OF_MONTH", endOfMonth)
             .build();
-    queriesBuilder.put("monthly_logs", monthlyLogsQuery);
+    queriesBuilder.put(getTableName(MONTHLY_LOGS), monthlyLogsQuery);
 
     String eppQuery =
         SqlTemplate.create(getQueryFromFile("epp_metrics.sql"))
             .put("PROJECT_ID", projectId)
             .put("ICANN_REPORTING_DATA_SET", ICANN_REPORTING_DATA_SET)
-            .put("MONTHLY_LOGS_TABLE", MONTHLY_LOGS_TABLE)
+            .put("MONTHLY_LOGS_TABLE", getTableName(MONTHLY_LOGS))
             .build();
-    queriesBuilder.put(EPP_METRICS, eppQuery);
+    queriesBuilder.put(getTableName(EPP_METRICS), eppQuery);
 
     String whoisQuery =
         SqlTemplate.create(getQueryFromFile("whois_counts.sql"))
             .put("PROJECT_ID", projectId)
-            .put("ICANN_REPORTING_DATA_SET", MONTHLY_LOGS_TABLE)
-            .put("MONTHLY_LOGS_TABLE", MONTHLY_LOGS_TABLE)
-            .build();
-    queriesBuilder.put(WHOIS_COUNTS, whoisQuery);
-
-    String activityQuery =
-        SqlTemplate.create(getQueryFromFile("activity_report_aggregation.sql"))
             .put("ICANN_REPORTING_DATA_SET", ICANN_REPORTING_DATA_SET)
-            .put("REGISTRAR_OPERATING_STATUS_TABLE", REGISTRAR_OPERATING_STATUS)
-            .put("DNS_COUNTS_TABLE", DNS_COUNTS)
-            .put("EPP_METRICS_TABLE", EPP_METRICS)
-            .put("WHOIS_COUNTS_TABLE", WHOIS_COUNTS)
-            .put("LATEST_SNAPSHOT_DATA_SET", "latest_snapshot")
+            .put("MONTHLY_LOGS_TABLE", getTableName(MONTHLY_LOGS))
+            .build();
+    queriesBuilder.put(getTableName(WHOIS_COUNTS), whoisQuery);
+
+    String aggregateQuery =
+        SqlTemplate.create(getQueryFromFile("activity_report_aggregation.sql"))
+            .put("PROJECT_ID", projectId)
+            .put("ICANN_REPORTING_DATA_SET", ICANN_REPORTING_DATA_SET)
+            .put("REGISTRAR_OPERATING_STATUS_TABLE", getTableName(REGISTRAR_OPERATING_STATUS))
+            .put("DNS_COUNTS_TABLE", getTableName(DNS_COUNTS))
+            .put("EPP_METRICS_TABLE", getTableName(EPP_METRICS))
+            .put("WHOIS_COUNTS_TABLE", getTableName(WHOIS_COUNTS))
+            // TODO(larryruili): Change to "latest_datastore_export" when cl/163124895 in prod.
+            .put("LATEST_DATASTORE_EXPORT", "latest_datastore_views")
             .put("REGISTRY_TABLE", "Registry")
             .build();
-    queriesBuilder.put("activity_report_aggregation", activityQuery);
+    queriesBuilder.put(getTableName(ACTIVITY_REPORT_AGGREGATION), aggregateQuery);
 
     return queriesBuilder.build();
   }
@@ -115,3 +140,4 @@ public final class ActivityReportingQueryBuilder {
     return Resources.getResource(ActivityReportingQueryBuilder.class, "sql/" + filename);
   }
 }
+
diff --git a/java/google/registry/reporting/BUILD b/java/google/registry/reporting/BUILD
index 6b58cf91c..315e737dc 100644
--- a/java/google/registry/reporting/BUILD
+++ b/java/google/registry/reporting/BUILD
@@ -9,6 +9,7 @@ java_library(
     srcs = glob(["*.java"]),
     resources = glob(["sql/*"]),
     deps = [
+        "//java/google/registry/bigquery",
         "//java/google/registry/config",
         "//java/google/registry/gcs",
         "//java/google/registry/keyring/api",
@@ -18,11 +19,14 @@ java_library(
         "//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_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",
     ],
diff --git a/java/google/registry/reporting/IcannReportingModule.java b/java/google/registry/reporting/IcannReportingModule.java
index 592eaabf0..815d4b831 100644
--- a/java/google/registry/reporting/IcannReportingModule.java
+++ b/java/google/registry/reporting/IcannReportingModule.java
@@ -18,11 +18,18 @@ import static google.registry.request.RequestParameters.extractEnumParameter;
 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.base.Optional;
+import com.google.common.collect.ImmutableList;
 import dagger.Module;
 import dagger.Provides;
+import google.registry.bigquery.BigqueryConnection;
 import google.registry.request.Parameter;
+import java.util.concurrent.Executors;
 import javax.servlet.http.HttpServletRequest;
+import org.joda.time.Duration;
 
 /** Module for dependencies required by ICANN monthly transactions/activity reporting. */
 @Module
@@ -37,6 +44,8 @@ public final class IcannReportingModule {
   static final String PARAM_YEAR_MONTH = "yearMonth";
   static final String PARAM_REPORT_TYPE = "reportType";
   static final String PARAM_SUBDIR = "subdir";
+  private static final String BIGQUERY_SCOPE =  "https://www.googleapis.com/auth/bigquery";
+  private static final String DRIVE_SCOPE = "https://www.googleapis.com/auth/drive.readonly";
 
   @Provides
   @Parameter(PARAM_YEAR_MONTH)
@@ -55,4 +64,31 @@ public final class IcannReportingModule {
   static Optional<String> provideSubdir(HttpServletRequest req) {
     return extractOptionalParameter(req, PARAM_SUBDIR);
   }
+
+  /**
+   * Constructs a BigqueryConnection with default settings.
+   *
+   * <p> We use Bigquery to generate activity reports via large aggregate SQL queries.
+   *
+   * @see ActivityReportingQueryBuilder
+   * @see google.registry.tools.BigqueryParameters for justifications of defaults.
+   */
+  @Provides
+  static BigqueryConnection provideBigqueryConnection(HttpTransport transport) {
+    try {
+      GoogleCredential credential = GoogleCredential
+          .getApplicationDefault(transport, new JacksonFactory());
+      BigqueryConnection connection = new BigqueryConnection.Builder()
+          .setExecutorService(Executors.newFixedThreadPool(20))
+          .setCredential(credential.createScoped(ImmutableList.of(BIGQUERY_SCOPE, DRIVE_SCOPE)))
+          .setDatasetId(ActivityReportingQueryBuilder.ICANN_REPORTING_DATA_SET)
+          .setOverwrite(true)
+          .setPollInterval(Duration.standardSeconds(1))
+          .build();
+      connection.initialize();
+      return connection;
+    } catch (Throwable e) {
+      throw new RuntimeException("Could not initialize BigqueryConnection!", e);
+    }
+  }
 }
diff --git a/java/google/registry/reporting/IcannReportingStagingAction.java b/java/google/registry/reporting/IcannReportingStagingAction.java
new file mode 100644
index 000000000..26fc31b28
--- /dev/null
+++ b/java/google/registry/reporting/IcannReportingStagingAction.java
@@ -0,0 +1,185 @@
+// 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.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.Function;
+import com.google.common.base.Optional;
+import com.google.common.base.Throwables;
+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.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.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+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
+ */
+@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 ActivityReportingQueryBuilder queryBuilder;
+  @Inject BigqueryConnection bigquery;
+  @Inject GcsUtils gcsUtils;
+  @Inject Response response;
+  @Inject IcannReportingStagingAction() {}
+
+  @Override
+  public void run() {
+    try {
+      // Generate intermediary views
+      ImmutableMap<String, String> activityQueries =
+          queryBuilder.getViewQueryMap();
+      for (Entry<String, String> entry : activityQueries.entrySet()) {
+        createIntermediaryTableView(entry.getKey(), entry.getValue());
+      }
+
+      // Get an in-memory table of the activity report query
+      ImmutableTable<Integer, TableFieldSchema, Object> reportTable =
+          bigquery.queryToLocalTableSync(queryBuilder.getActivityReportQuery());
+
+      // Get report headers from the table schema and convert into CSV format
+      String headerRow = constructActivityReportRow(getHeaders(reportTable.columnKeySet()));
+      logger.infofmt("Headers: %s", headerRow);
+
+      // Create a report csv for each tld from query table, and upload to GCS
+      for (Map<TableFieldSchema, Object> row : reportTable.rowMap().values()) {
+        // 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!");
+        }
+        uploadReport(tld, createReport(headerRow, row));
+      }
+      response.setStatus(SC_OK);
+      response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
+      response.setPayload("Completed staging action.");
+    } catch (Exception e) {
+      logger.warning(Throwables.getStackTraceAsString(e));
+      response.setStatus(SC_INTERNAL_SERVER_ERROR);
+      response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
+      response.setPayload("Caught exception:\n" + e.getMessage());
+    }
+  }
+
+  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("An intermediary view to generate activity reports for this month.")
+            .type(TableType.VIEW)
+            .build()
+    ).get();
+  }
+
+  private Iterable<String> getHeaders(ImmutableSet<TableFieldSchema> fields) {
+    return Iterables.transform(
+        fields,
+        new Function<TableFieldSchema, String>() {
+          @Override
+          public String apply(TableFieldSchema schema) {
+            // Change from '_' delimiters (Bigquery-compatible) to '-' (ICANN specification)
+            return schema.getName().replace('_', '-');
+          }
+        }
+    );
+  }
+
+  /**
+   * 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 constructActivityReportRow(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();
+  }
+
+  private String createReport(String headers, Map<TableFieldSchema, Object> row) {
+    StringBuilder reportCsv = new StringBuilder(headers);
+    // Add CRLF between rows per ICANN specification
+    reportCsv.append("\r\n");
+    String valuesRow = constructActivityReportRow(row.values());
+    reportCsv.append(valuesRow);
+    logger.infofmt("Created report %s", 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.ACTIVITY);
+    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());
+  }
+}
diff --git a/java/google/registry/reporting/IcannReportingUploadAction.java b/java/google/registry/reporting/IcannReportingUploadAction.java
index 60eabc9d0..fcdf5f530 100644
--- a/java/google/registry/reporting/IcannReportingUploadAction.java
+++ b/java/google/registry/reporting/IcannReportingUploadAction.java
@@ -56,15 +56,15 @@ public final class IcannReportingUploadAction implements Runnable {
 
   private static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
 
-  @Inject GcsUtils gcsUtils;
-  @Inject IcannHttpReporter icannReporter;
-  @Inject Retrier retrier;
+  @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 GcsUtils gcsUtils;
+  @Inject IcannHttpReporter icannReporter;
   @Inject Response response;
+  @Inject Retrier retrier;
 
   @Inject
   IcannReportingUploadAction() {}
@@ -73,11 +73,9 @@ public final class IcannReportingUploadAction implements Runnable {
   public void run() {
     validateParams();
     String reportFilename = createFilename(tld, yearMonth, reportType);
-    logger.infofmt("Reading ICANN report %s from bucket %s", reportFilename, reportingBucket);
-    final GcsFilename gcsFilename =
-        new GcsFilename(
-            reportingBucket + "/" + (subdir.isPresent() ? subdir.get() : DEFAULT_SUBDIR),
-            reportFilename);
+    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",
@@ -103,12 +101,19 @@ public final class IcannReportingUploadAction implements Runnable {
   }
 
   static String createFilename(String tld, String yearMonth, ReportType reportType) {
-    // TODO(b/62585428): Change file naming date format to YYYY-MM for consistency with URL.
-    // Report files use YYYYMM naming instead of standard YYYY-MM.
+    // 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(
diff --git a/java/google/registry/reporting/sql/activity_report_aggregation.sql b/java/google/registry/reporting/sql/activity_report_aggregation.sql
index 0182d4b64..d1ee1434b 100644
--- a/java/google/registry/reporting/sql/activity_report_aggregation.sql
+++ b/java/google/registry/reporting/sql/activity_report_aggregation.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
@@ -16,7 +17,7 @@
   -- report csv, via a table transpose and sum over all activity report fields.
 
 SELECT
-  Tld.tld AS tld,
+  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,
@@ -59,37 +60,41 @@ SELECT
   SUM(IF(metricName = 'srs-cont-transfer-query', count, 0)) AS srs_cont_transfer_query,
   SUM(IF(metricName = 'srs-cont-transfer-reject', count, 0)) AS srs_cont_transfer_reject,
   SUM(IF(metricName = 'srs-cont-transfer-request', count, 0)) AS srs_cont_transfer_request,
-  SUM(IF(metricName = 'srs-cont-update', count, 0)) AS srs_cont_update,
+  SUM(IF(metricName = 'srs-cont-update', count, 0)) AS srs_cont_update
   -- Cross join a list of all TLDs against TLD-specific metrics and then
   -- filter so that only metrics with that TLD or a NULL TLD are counted
   -- towards a given TLD.
 FROM (
-  SELECT
-    tldStr AS tld
-  FROM
-    [%LATEST_SNAPSHOT_DATA_SET%.%REGISTRY_TABLE%]
-    -- Include all real TLDs that are not in pre-delegation testing.
-  WHERE
-    tldType = 'REAL'
-  OMIT
-    RECORD IF SOME(tldStateTransitions.tldState = 'PDT') ) AS Tld
-  -- TODO(larryruili): Use LEFT JOIN on Tld.tld = TldMetrics.tld instead.
-  -- Also obsoletes dummy data.
-LEFT OUTER JOIN (
-  SELECT
+SELECT tldStr as tld
+FROM `%PROJECT_ID%.%LATEST_DATASTORE_EXPORT%.%REGISTRY_TABLE%`
+WHERE tldType = 'REAL'
+) as RealTlds
+CROSS JOIN(
+   SELECT
     tld,
     metricName,
-    count FROM
+    count
+    FROM
+    (
     -- BEGIN INTERMEDIARY DATA SOURCES --
-    [%ICANN_REPORTING_DATA_SET%.%REGISTRAR_OPERATING_STATUS_TABLE%],
-    [%ICANN_REPORTING_DATA_SET%.%DNS_COUNTS_TABLE%],
-    [%ICANN_REPORTING_DATA_SET%.%EPP_METRICS_TABLE%],
-    [%ICANN_REPORTING_DATA_SET%.%WHOIS_COUNTS_TABLE%],
+    -- Dummy data source to ensure all TLDs appear in report, even if
+    -- they have no recorded metrics for the month.
+    SELECT STRING(NULL) AS tld, STRING(NULL) AS metricName, 0 as count
+    UNION ALL
+    SELECT * FROM
+    `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%REGISTRAR_OPERATING_STATUS_TABLE%`
+    UNION ALL
+    SELECT * FROM
+    `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%DNS_COUNTS_TABLE%`
+    UNION ALL
+    SELECT * FROM
+    `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%EPP_METRICS_TABLE%`
+    UNION ALL
+    SELECT * FROM
+    `%PROJECT_ID%.%ICANN_REPORTING_DATA_SET%.%WHOIS_COUNTS_TABLE%`
     -- END INTERMEDIARY DATA SOURCES --
-    ) AS TldMetrics
-ON
-  Tld.tld = TldMetrics.tld
-GROUP BY
-  tld
-ORDER BY
-  tld
+    )) AS TldMetrics
+WHERE RealTlds.tld = TldMetrics.tld OR TldMetrics.tld IS NULL
+GROUP BY tld
+ORDER BY tld
+
diff --git a/java/google/registry/reporting/sql/dns_counts.sql b/java/google/registry/reporting/sql/dns_counts.sql
index 123782f2f..62d822a6b 100644
--- a/java/google/registry/reporting/sql/dns_counts.sql
+++ b/java/google/registry/reporting/sql/dns_counts.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/java/google/registry/reporting/sql/epp_metrics.sql b/java/google/registry/reporting/sql/epp_metrics.sql
index 7786d7037..b14918e17 100644
--- a/java/google/registry/reporting/sql/epp_metrics.sql
+++ b/java/google/registry/reporting/sql/epp_metrics.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/java/google/registry/reporting/sql/monthly_logs.sql b/java/google/registry/reporting/sql/monthly_logs.sql
index a491c2375..77afdbe09 100644
--- a/java/google/registry/reporting/sql/monthly_logs.sql
+++ b/java/google/registry/reporting/sql/monthly_logs.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/java/google/registry/reporting/sql/registrar_operating_status.sql b/java/google/registry/reporting/sql/registrar_operating_status.sql
index 716df387a..0065e1b37 100644
--- a/java/google/registry/reporting/sql/registrar_operating_status.sql
+++ b/java/google/registry/reporting/sql/registrar_operating_status.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/java/google/registry/reporting/sql/whois_counts.sql b/java/google/registry/reporting/sql/whois_counts.sql
index dc1b57266..d118d1c33 100644
--- a/java/google/registry/reporting/sql/whois_counts.sql
+++ b/java/google/registry/reporting/sql/whois_counts.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/javatests/google/registry/module/backend/testdata/backend_routing.txt b/javatests/google/registry/module/backend/testdata/backend_routing.txt
index c02665e05..a571e697a 100644
--- a/javatests/google/registry/module/backend/testdata/backend_routing.txt
+++ b/javatests/google/registry/module/backend/testdata/backend_routing.txt
@@ -14,6 +14,7 @@ PATH                                   CLASS                              METHOD
 /_dr/task/exportDomainLists            ExportDomainListsAction            POST     n  INTERNAL     APP IGNORED
 /_dr/task/exportReservedTerms          ExportReservedTermsAction          POST     n  INTERNAL     APP IGNORED
 /_dr/task/exportSnapshot               ExportSnapshotAction               POST     y  INTERNAL     APP IGNORED
+/_dr/task/icannReportingStaging        IcannReportingStagingAction        POST     n  INTERNAL     APP IGNORED
 /_dr/task/icannReportingUpload         IcannReportingUploadAction         POST     n  INTERNAL,API APP ADMIN
 /_dr/task/importRdeContacts            RdeContactImportAction             GET      n  INTERNAL     APP IGNORED
 /_dr/task/importRdeDomains             RdeDomainImportAction              GET      n  INTERNAL     APP IGNORED
diff --git a/javatests/google/registry/reporting/ActivityReportingQueryBuilderTest.java b/javatests/google/registry/reporting/ActivityReportingQueryBuilderTest.java
index 31ff02e4f..cec4c3eec 100644
--- a/javatests/google/registry/reporting/ActivityReportingQueryBuilderTest.java
+++ b/javatests/google/registry/reporting/ActivityReportingQueryBuilderTest.java
@@ -19,7 +19,6 @@ import static com.google.common.truth.Truth.assertThat;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import java.io.IOException;
-import org.joda.time.LocalDate;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -28,28 +27,43 @@ import org.junit.runners.JUnit4;
 @RunWith(JUnit4.class)
 public class ActivityReportingQueryBuilderTest {
 
+  private ActivityReportingQueryBuilder getQueryBuilder() {
+    ActivityReportingQueryBuilder queryBuilder = new ActivityReportingQueryBuilder();
+    queryBuilder.yearMonth = "2017-05";
+    queryBuilder.projectId = "domain-registry-alpha";
+    return queryBuilder;
+  }
+
   @Test
-  public void testQueryMatch() throws IOException {
+  public void testAggregateQueryMatch() throws IOException {
+    ActivityReportingQueryBuilder queryBuilder = getQueryBuilder();
+    assertThat(queryBuilder.getActivityReportQuery())
+        .isEqualTo(
+            "#standardSQL\nSELECT * FROM "
+                + "`domain-registry-alpha.icann_reporting.activity_report_aggregation_201705`");
+  }
+
+  @Test
+  public void testIntermediaryQueryMatch() throws IOException {
+    ActivityReportingQueryBuilder queryBuilder = getQueryBuilder();
     ImmutableList<String> queryNames =
         ImmutableList.of(
             ActivityReportingQueryBuilder.REGISTRAR_OPERATING_STATUS,
             ActivityReportingQueryBuilder.DNS_COUNTS,
-            ActivityReportingQueryBuilder.MONTHLY_LOGS_TABLE,
+            ActivityReportingQueryBuilder.MONTHLY_LOGS,
             ActivityReportingQueryBuilder.EPP_METRICS,
             ActivityReportingQueryBuilder.WHOIS_COUNTS,
-            "activity_report_aggregation");
+            ActivityReportingQueryBuilder.ACTIVITY_REPORT_AGGREGATION);
 
     ImmutableMap.Builder<String, String> testQueryBuilder = ImmutableMap.builder();
     for (String queryName : queryNames) {
       String testFilename = String.format("%s_test.sql", queryName);
-      testQueryBuilder.put(queryName, ReportingTestData.getString(testFilename));
-    }
-    ImmutableMap<String, String> testQueries = testQueryBuilder.build();
-    ImmutableMap<String, String> queries =
-        ActivityReportingQueryBuilder.getQueryMap(
-            new LocalDate(2017, 05, 15), "domain-registry-alpha");
-    for (String query : queryNames) {
-      assertThat(queries.get(query)).isEqualTo(testQueries.get(query));
+      testQueryBuilder.put(
+          String.format("%s_201705", queryName), ReportingTestData.getString(testFilename));
     }
+    ImmutableMap<String, String> expectedQueries = testQueryBuilder.build();
+    ImmutableMap<String, String> actualQueries = queryBuilder.getViewQueryMap();
+    assertThat(actualQueries).isEqualTo(expectedQueries);
   }
 }
+
diff --git a/javatests/google/registry/reporting/BUILD b/javatests/google/registry/reporting/BUILD
index 419b3d0d8..aa5a6119b 100644
--- a/javatests/google/registry/reporting/BUILD
+++ b/javatests/google/registry/reporting/BUILD
@@ -12,11 +12,13 @@ java_library(
     srcs = glob(["*.java"]),
     resources = glob(["testdata/*"]),
     deps = [
+        "//java/google/registry/bigquery",
         "//java/google/registry/gcs",
         "//java/google/registry/reporting",
         "//java/google/registry/request",
         "//java/google/registry/util",
         "//javatests/google/registry/testing",
+        "@com_google_apis_google_api_services_bigquery",
         "@com_google_appengine_tools_appengine_gcs_client",
         "@com_google_code_findbugs_jsr305",
         "@com_google_dagger",
diff --git a/javatests/google/registry/reporting/IcannReportingStagingActionTest.java b/javatests/google/registry/reporting/IcannReportingStagingActionTest.java
new file mode 100644
index 000000000..545a60791
--- /dev/null
+++ b/javatests/google/registry/reporting/IcannReportingStagingActionTest.java
@@ -0,0 +1,149 @@
+// 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.truth.Truth.assertThat;
+import static google.registry.testing.GcsTestingUtils.readGcsFile;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.api.services.bigquery.model.TableFieldSchema;
+import com.google.appengine.tools.cloudstorage.GcsFilename;
+import com.google.appengine.tools.cloudstorage.GcsService;
+import com.google.appengine.tools.cloudstorage.GcsServiceFactory;
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableTable;
+import com.google.common.util.concurrent.ListenableFuture;
+import google.registry.bigquery.BigqueryConnection;
+import google.registry.bigquery.BigqueryConnection.DestinationTable;
+import google.registry.bigquery.BigqueryUtils.TableType;
+import google.registry.gcs.GcsUtils;
+import google.registry.testing.AppEngineRule;
+import google.registry.testing.FakeResponse;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Unit tests for {@link google.registry.reporting.IcannReportingStagingAction}.
+ */
+@RunWith(JUnit4.class)
+public class IcannReportingStagingActionTest {
+
+  BigqueryConnection bigquery = mock(BigqueryConnection.class);
+  FakeResponse response = new FakeResponse();
+  ActivityReportingQueryBuilder queryBuilder;
+  GcsService gcsService = GcsServiceFactory.createGcsService();
+
+  @Rule
+  public final AppEngineRule appEngine = AppEngineRule.builder()
+      .withDatastore()
+      .withLocalModules()
+      .build();
+
+  private IcannReportingStagingAction createAction() {
+    IcannReportingStagingAction action = new IcannReportingStagingAction();
+    queryBuilder = new ActivityReportingQueryBuilder();
+    queryBuilder.projectId = "test-project";
+    queryBuilder.yearMonth = "2017-05";
+    action.reportingBucket = "test-bucket";
+    action.yearMonth = "2017-05";
+    action.subdir = Optional.absent();
+    action.queryBuilder = queryBuilder;
+    action.bigquery = bigquery;
+    action.gcsUtils = new GcsUtils(gcsService, 1024);
+    action.response = response;
+    return action;
+  }
+
+  @Test
+  public void testRunSuccess() throws Exception {
+    when(bigquery.query(any(), any())).thenReturn(fakeFuture());
+    DestinationTable.Builder tableBuilder = new DestinationTable.Builder()
+        .datasetId("testdataset")
+        .type(TableType.TABLE)
+        .name("tablename")
+        .overwrite(true);
+    when(bigquery.buildDestinationTable(any())).thenReturn(tableBuilder);
+
+    ImmutableTable<Integer, TableFieldSchema, Object> reportTable =
+        new ImmutableTable.Builder<Integer, TableFieldSchema, Object>()
+            .put(1, new TableFieldSchema().setName("tld"), "fooTld")
+            .put(1, new TableFieldSchema().setName("fooField"), "12")
+            .put(1, new TableFieldSchema().setName("barField"), "34")
+            .put(2, new TableFieldSchema().setName("tld"), "barTld")
+            .put(2, new TableFieldSchema().setName("fooField"), "56")
+            .put(2, new TableFieldSchema().setName("barField"), "78")
+            .build();
+    when(bigquery.queryToLocalTableSync(any())).thenReturn(reportTable);
+    IcannReportingStagingAction action = createAction();
+    action.run();
+
+    String expectedReport1 = "fooField,barField\r\n12,34";
+    String expectedReport2 = "fooField,barField\r\n56,78";
+    byte[] generatedFile1 =
+        readGcsFile(
+            gcsService,
+            new GcsFilename("test-bucket/icann/monthly/2017-05", "fooTld-activity-201705.csv"));
+    assertThat(new String(generatedFile1, UTF_8)).isEqualTo(expectedReport1);
+    byte[] generatedFile2 =
+        readGcsFile(
+            gcsService,
+            new GcsFilename("test-bucket/icann/monthly/2017-05", "barTld-activity-201705.csv"));
+    assertThat(new String(generatedFile2, UTF_8)).isEqualTo(expectedReport2);
+  }
+
+  private ListenableFuture<DestinationTable> fakeFuture() {
+    return new ListenableFuture<DestinationTable>() {
+      @Override
+      public void addListener(Runnable runnable, Executor executor) {
+        // No-op
+      }
+
+      @Override
+      public boolean cancel(boolean mayInterruptIfRunning) {
+        return false;
+      }
+
+      @Override
+      public boolean isCancelled() {
+        return false;
+      }
+
+      @Override
+      public boolean isDone() {
+        return false;
+      }
+
+      @Override
+      public DestinationTable get() throws InterruptedException, ExecutionException {
+        return null;
+      }
+
+      @Override
+      public DestinationTable get(long timeout, TimeUnit unit)
+          throws InterruptedException, ExecutionException, TimeoutException {
+        return null;
+      }
+    };
+  }
+}
diff --git a/javatests/google/registry/reporting/IcannReportingUploadActionTest.java b/javatests/google/registry/reporting/IcannReportingUploadActionTest.java
index aa63ba92d..e0f128a8f 100644
--- a/javatests/google/registry/reporting/IcannReportingUploadActionTest.java
+++ b/javatests/google/registry/reporting/IcannReportingUploadActionTest.java
@@ -16,6 +16,7 @@ package google.registry.reporting;
 
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
+import static google.registry.reporting.IcannReportingModule.ReportType.ACTIVITY;
 import static google.registry.reporting.IcannReportingModule.ReportType.TRANSACTIONS;
 import static google.registry.testing.DatastoreHelper.createTld;
 import static google.registry.testing.GcsTestingUtils.writeGcsFile;
@@ -54,18 +55,18 @@ public class IcannReportingUploadActionTest {
   private final FakeResponse response = new FakeResponse();
   private final GcsService gcsService = GcsServiceFactory.createGcsService();
   private final GcsFilename reportFile =
-      new GcsFilename("basin/icann/monthly", "test-transactions-201706.csv");
+      new GcsFilename("basin/icann/monthly/2017-05", "test-transactions-201705.csv");
 
   private IcannReportingUploadAction createAction() {
     IcannReportingUploadAction action = new IcannReportingUploadAction();
     action.icannReporter = mockReporter;
     action.gcsUtils = new GcsUtils(gcsService, 1024);
     action.retrier = new Retrier(new FakeSleeper(new FakeClock()), 3);
-    action.yearMonth = "2017-06";
+    action.yearMonth = "2017-05";
     action.reportType = TRANSACTIONS;
     action.subdir = Optional.absent();
     action.tld = "test";
-    action.reportingBucket = "basin";
+    action.icannReportingBucket = "basin";
     action.response = response;
     return action;
   }
@@ -80,7 +81,7 @@ public class IcannReportingUploadActionTest {
   public void testSuccess() throws Exception {
     IcannReportingUploadAction action = createAction();
     action.run();
-    verify(mockReporter).send(FAKE_PAYLOAD, "test", "2017-06", TRANSACTIONS);
+    verify(mockReporter).send(FAKE_PAYLOAD, "test", "2017-05", TRANSACTIONS);
     verifyNoMoreInteractions(mockReporter);
     assertThat(((FakeResponse) action.response).getPayload())
         .isEqualTo("OK, sending: test,csv\n13,37");
@@ -92,9 +93,9 @@ public class IcannReportingUploadActionTest {
     doThrow(new IOException("Expected exception."))
         .doNothing()
         .when(mockReporter)
-        .send(FAKE_PAYLOAD, "test", "2017-06", TRANSACTIONS);
+        .send(FAKE_PAYLOAD, "test", "2017-05", TRANSACTIONS);
     action.run();
-    verify(mockReporter, times(2)).send(FAKE_PAYLOAD, "test", "2017-06", TRANSACTIONS);
+    verify(mockReporter, times(2)).send(FAKE_PAYLOAD, "test", "2017-05", TRANSACTIONS);
     verifyNoMoreInteractions(mockReporter);
     assertThat(((FakeResponse) action.response).getPayload())
         .isEqualTo("OK, sending: test,csv\n13,37");
@@ -155,7 +156,27 @@ public class IcannReportingUploadActionTest {
           .hasMessageThat()
           .isEqualTo(
               "ICANN report object test-transactions-123456.csv "
-                  + "in bucket basin/icann/monthly not found");
+                  + "in bucket basin/icann/monthly/1234-56 not found");
     }
   }
+
+  @Test
+  public void testSuccess_CreateFilename() throws Exception{
+    assertThat(IcannReportingUploadAction.createFilename("test", "2017-05", ACTIVITY))
+        .isEqualTo("test-activity-201705.csv");
+    assertThat(IcannReportingUploadAction.createFilename("foo", "1234-56", TRANSACTIONS))
+        .isEqualTo("foo-transactions-123456.csv");
+  }
+
+  @Test
+  public void testSuccess_CreateBucketname() throws Exception{
+    assertThat(
+        IcannReportingUploadAction
+            .createReportingBucketName("gs://my-reporting", Optional.absent(), "2017-05"))
+        .isEqualTo("gs://my-reporting/icann/monthly/2017-05");
+    assertThat(
+        IcannReportingUploadAction
+            .createReportingBucketName("gs://my-reporting", Optional.of("manual"), "2017-05"))
+        .isEqualTo("gs://my-reporting/manual");
+  }
 }
diff --git a/javatests/google/registry/reporting/testdata/activity_report_aggregation_test.sql b/javatests/google/registry/reporting/testdata/activity_report_aggregation_test.sql
index e0ffac326..06f4ca910 100644
--- a/javatests/google/registry/reporting/testdata/activity_report_aggregation_test.sql
+++ b/javatests/google/registry/reporting/testdata/activity_report_aggregation_test.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
@@ -16,7 +17,7 @@
   -- report csv, via a table transpose and sum over all activity report fields.
 
 SELECT
-  Tld.tld AS tld,
+  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,
@@ -59,37 +60,41 @@ SELECT
   SUM(IF(metricName = 'srs-cont-transfer-query', count, 0)) AS srs_cont_transfer_query,
   SUM(IF(metricName = 'srs-cont-transfer-reject', count, 0)) AS srs_cont_transfer_reject,
   SUM(IF(metricName = 'srs-cont-transfer-request', count, 0)) AS srs_cont_transfer_request,
-  SUM(IF(metricName = 'srs-cont-update', count, 0)) AS srs_cont_update,
+  SUM(IF(metricName = 'srs-cont-update', count, 0)) AS srs_cont_update
   -- Cross join a list of all TLDs against TLD-specific metrics and then
   -- filter so that only metrics with that TLD or a NULL TLD are counted
   -- towards a given TLD.
 FROM (
-  SELECT
-    tldStr AS tld
-  FROM
-    [latest_snapshot.Registry]
-    -- Include all real TLDs that are not in pre-delegation testing.
-  WHERE
-    tldType = 'REAL'
-  OMIT
-    RECORD IF SOME(tldStateTransitions.tldState = 'PDT') ) AS Tld
-  -- TODO(larryruili): Use LEFT JOIN on Tld.tld = TldMetrics.tld instead.
-  -- Also obsoletes dummy data.
-LEFT OUTER JOIN (
-  SELECT
+SELECT tldStr as tld
+FROM `domain-registry-alpha.latest_datastore_views.Registry`
+WHERE tldType = 'REAL'
+) as RealTlds
+CROSS JOIN(
+   SELECT
     tld,
     metricName,
-    count FROM
+    count
+    FROM
+    (
     -- BEGIN INTERMEDIARY DATA SOURCES --
-    [icann_reporting.registrar_operating_status],
-    [icann_reporting.dns_counts],
-    [icann_reporting.epp_metrics],
-    [icann_reporting.whois_counts],
+    -- Dummy data source to ensure all TLDs appear in report, even if
+    -- they have no recorded metrics for the month.
+    SELECT STRING(NULL) AS tld, STRING(NULL) AS metricName, 0 as count
+    UNION ALL
+    SELECT * FROM
+    `domain-registry-alpha.icann_reporting.registrar_operating_status_201705`
+    UNION ALL
+    SELECT * FROM
+    `domain-registry-alpha.icann_reporting.dns_counts_201705`
+    UNION ALL
+    SELECT * FROM
+    `domain-registry-alpha.icann_reporting.epp_metrics_201705`
+    UNION ALL
+    SELECT * FROM
+    `domain-registry-alpha.icann_reporting.whois_counts_201705`
     -- END INTERMEDIARY DATA SOURCES --
-    ) AS TldMetrics
-ON
-  Tld.tld = TldMetrics.tld
-GROUP BY
-  tld
-ORDER BY
-  tld
+    )) AS TldMetrics
+WHERE RealTlds.tld = TldMetrics.tld OR TldMetrics.tld IS NULL
+GROUP BY tld
+ORDER BY tld
+
diff --git a/javatests/google/registry/reporting/testdata/dns_counts_test.sql b/javatests/google/registry/reporting/testdata/dns_counts_test.sql
index 123782f2f..62d822a6b 100644
--- a/javatests/google/registry/reporting/testdata/dns_counts_test.sql
+++ b/javatests/google/registry/reporting/testdata/dns_counts_test.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/javatests/google/registry/reporting/testdata/epp_metrics_test.sql b/javatests/google/registry/reporting/testdata/epp_metrics_test.sql
index e8d57eda2..50f940ae6 100644
--- a/javatests/google/registry/reporting/testdata/epp_metrics_test.sql
+++ b/javatests/google/registry/reporting/testdata/epp_metrics_test.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
@@ -38,7 +39,7 @@ FROM (
       -- Extract the logged JSON payload.
       REGEXP_EXTRACT(logMessage, r'FLOW-LOG-SIGNATURE-METADATA: (.*)\n?$')
       AS json
-    FROM `domain-registry-alpha.icann_reporting.monthly_logs` AS logs
+    FROM `domain-registry-alpha.icann_reporting.monthly_logs_201705` AS logs
     JOIN
       UNNEST(logs.logMessage) AS logMessage
     WHERE
diff --git a/javatests/google/registry/reporting/testdata/monthly_logs_test.sql b/javatests/google/registry/reporting/testdata/monthly_logs_test.sql
index 929270028..ad5e65024 100644
--- a/javatests/google/registry/reporting/testdata/monthly_logs_test.sql
+++ b/javatests/google/registry/reporting/testdata/monthly_logs_test.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/javatests/google/registry/reporting/testdata/registrar_operating_status_test.sql b/javatests/google/registry/reporting/testdata/registrar_operating_status_test.sql
index 9fb50733c..e6fe201f5 100644
--- a/javatests/google/registry/reporting/testdata/registrar_operating_status_test.sql
+++ b/javatests/google/registry/reporting/testdata/registrar_operating_status_test.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/javatests/google/registry/reporting/testdata/whois_counts_test.sql b/javatests/google/registry/reporting/testdata/whois_counts_test.sql
index 11d27db11..d2051932e 100644
--- a/javatests/google/registry/reporting/testdata/whois_counts_test.sql
+++ b/javatests/google/registry/reporting/testdata/whois_counts_test.sql
@@ -1,3 +1,4 @@
+#standardSQL
   -- Copyright 2017 The Nomulus Authors. All Rights Reserved.
   --
   -- Licensed under the Apache License, Version 2.0 (the "License");
@@ -25,7 +26,7 @@ SELECT
   END AS metricName,
   COUNT(requestPath) AS count
 FROM
-  `domain-registry-alpha.monthly_logs.monthly_logs`
+  `domain-registry-alpha.icann_reporting.monthly_logs_201705`
 GROUP BY
   metricName
 HAVING