Migrate Spec11 pipeline to flex template (#1073)

* Migrate Spec11 pipeline to flex template

Unfortunately this PR has turned out to be much bigger than I initially
conceived. However this is no good way to separate it out because the
changes are intertwined. This PR includes 3 main changes:

1. Change the spec11 pipline to use Dataflow Flex Template.
2. Retire the use of the old JPA layer that relies on credential saved
   in KMS.
3. Some extensive refactoring to streamline the logic and improve test
   isolation.

* Fix job name and remove projectId from options

* Add parameter logs

* Set RegistryEnvironment

* Remove logging and modify safe browsing API key regex

* Rename a test method and rebase

* Remove unused Junit extension

* Specify job region

core/src/main/resources/google/registry/beam/spec11/sql/subdomains.sql

@@ -0,0 +1,51 @@
+#standardSQL
+  -- Copyright 2018 The Nomulus Authors. All Rights Reserved.
+  --
+  -- Licensed under the Apache License, Version 2.0 (the "License");
+  -- you may not use this file except in compliance with the License.
+  -- You may obtain a copy of the License at
+  --
+  --     http://www.apache.org/licenses/LICENSE-2.0
+  --
+  -- Unless required by applicable law or agreed to in writing, software
+  -- distributed under the License is distributed on an "AS IS" BASIS,
+  -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  -- See the License for the specific language governing permissions and
+  -- limitations under the License.
+
+  -- This query gathers all Subdomains active within a given yearMonth
+  -- and emits a row containing its fully qualified domain name
+  -- [SLD].[TLD], the current registrar's name, and the current registrar's
+  -- email address.
+
+SELECT
+  domain.fullyQualifiedDomainName AS domainName,
+  domain.__key__.name AS domainRepoId,
+  registrar.clientId AS registrarId,
+  COALESCE(registrar.emailAddress, '') AS registrarEmailAddress
+FROM ( (
+    SELECT
+      __key__,
+      fullyQualifiedDomainName,
+      currentSponsorClientId,
+      creationTime
+    FROM
+      `%PROJECT_ID%.%DATASTORE_EXPORT_DATASET%.%DOMAIN_BASE_TABLE%`
+    WHERE
+      -- Only include active registrations
+      -- Registrations that are active (not deleted) will have null deletionTime
+      -- because END_OF_TIME is an invalid timestamp in standardSQL
+      (SAFE_CAST(deletionTime AS STRING) IS NULL
+        OR deletionTime > CURRENT_TIMESTAMP)) AS domain
+  JOIN (
+    SELECT
+      __key__.name AS clientId,
+      emailAddress
+    FROM
+      `%PROJECT_ID%.%DATASTORE_EXPORT_DATASET%.%REGISTRAR_TABLE%`
+    WHERE
+      type = 'REAL') AS registrar
+  ON
+    domain.currentSponsorClientId = registrar.clientId)
+ORDER BY
+  creationTime DESC

core/src/main/resources/google/registry/beam/spec11_pipeline_metadata.json

@@ -0,0 +1,66 @@
+{
+  "name": "Spec11 Report Generation",
+  "description": "An Apache Beam batch pipeline that reads from a Datastore export and generate Spec11 report for the month, saving it to both SQL and as a JSON file on GCS.",
+  "parameters": [
+    {
+      "name": "registryEnvironment",
+      "label": "The Registry environment.",
+      "helpText": "The Registry environment, required if environment-specific initialization is needed on worker VMs.",
+      "is_optional": true,
+      "regexes": [
+        "^[0-9A-Z_]+$"
+      ]
+    },
+    {
+      "name": "isolationOverride",
+      "label": "The desired SQL transaction isolation level.",
+      "helpText": "The desired SQL transaction isolation level.",
+      "is_optional": true,
+      "regexes": [
+        "^[0-9A-Z_]+$"
+      ]
+    },
+    {
+      "name": "sqlWriteBatchSize",
+      "label": "SQL write batch size.",
+      "helpText": "The number of entities to write to the SQL database in one operation.",
+      "is_optional": true,
+      "regexes": [
+        "^[1-9][0-9]*$"
+      ]
+    },
+    {
+      "name": "sqlWriteShards",
+      "label": "Number of output shards to create when writing to SQL.",
+      "helpText": "Number of shards to create out of the data before writing to the SQL database. Please refer to the Javadoc of RegistryJpaIO.Write.shards() for how to choose this value.",
+      "is_optional": true,
+      "regexes": [
+        "^[1-9][0-9]*$"
+      ]
+    },
+    {
+      "name": "date",
+      "label": "The date when the pipeline runs",
+      "helpText": "The date then the threat scan is performed, in yyyy-MM-dd format.",
+      "regexes": [
+        "^2[0-9]{3}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])$"
+      ]
+    },
+    {
+      "name": "safeBrowsingApiKey",
+      "label": "API Key used to identify the project with Safe Browsing",
+      "helpText": "The earliest CommitLogs to load, in ISO8601 format.",
+      "regexes": [
+        "^[0-9a-zA-Z_]+[\\n]?$"
+      ]
+    },
+    {
+      "name": "reportingBucketUrl",
+      "label": "Spec11 report upload dir.",
+      "helpText": "The root directory of the report to upload.",
+      "regexes": [
+        "^gs:\\/\\/[^\\n\\r]+$"
+      ]
+    }
+  ]
+}