Implement a util class to manage push queues using Cloud Tasks API (#1290)

* Implement a util class to manage push queues using Cloud Tasks API

Push queues were part of App Engine when they debuted. As a result the
Task Queue API were part of the App Engine SDK and can only be used in
App Engine classic runtime. The new Cloud Tasks API can be used in any
runtime but it only supports push queues. In this PR we implement a util
class (CloudTasksUtils) like TaskQueueUtils to handle enqueuing tasks to
push queues using Cloud Tasks. One action (TldFanoutAction) was
converted to use the new API as a demo. Mass migration of other call sites of
the old API will follow in a separate PR.

TESTED=deployed to alpha and verified that tasks are corrected enqueued
and executed.

core/src/main/java/google/registry/config/CloudTasksUtilsModule.java

@@ -0,0 +1,65 @@
+// Copyright 2021 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.config;
+
+import com.google.api.gax.core.FixedCredentialsProvider;
+import com.google.cloud.tasks.v2.CloudTasksClient;
+import com.google.cloud.tasks.v2.CloudTasksSettings;
+import dagger.Module;
+import dagger.Provides;
+import google.registry.config.CredentialModule.DefaultCredential;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.util.CloudTasksUtils;
+import google.registry.util.GoogleCredentialsBundle;
+import google.registry.util.Retrier;
+import java.io.IOException;
+import javax.inject.Provider;
+import javax.inject.Singleton;
+
+/**
+ * A {@link Module} that provides {@link CloudTasksUtils}.
+ *
+ * <p>The class itself cannot be annotated as {@code Inject} because its requested dependencies use
+ * the {@link Config} qualifier which is not available in the {@code util} package.
+ */
+@Module
+public abstract class CloudTasksUtilsModule {
+
+  @Singleton
+  @Provides
+  public static CloudTasksUtils provideCloudTasksUtils(
+      @Config("projectId") String projectId,
+      @Config("locationId") String locationId,
+      // Use a provider so that we can use try-with-resources with the client, which implements
+      // Autocloseable.
+      Provider<CloudTasksClient> clientProvider,
+      Retrier retrier) {
+    return new CloudTasksUtils(retrier, projectId, locationId, clientProvider);
+  }
+
+  @Provides
+  public static CloudTasksClient provideCloudTasksClient(
+      @DefaultCredential GoogleCredentialsBundle credentials) {
+    try {
+      return CloudTasksClient.create(
+          CloudTasksSettings.newBuilder()
+              .setCredentialsProvider(
+                  FixedCredentialsProvider.create(credentials.getGoogleCredentials()))
+              .build());
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

core/src/main/java/google/registry/config/RegistryConfig.java

@@ -105,6 +105,12 @@ public final class RegistryConfig {
       return config.appEngine.projectId;
     }
 
+    @Provides
+    @Config("locationId")
+    public static String provideLocationId(RegistryConfigSettings config) {
+      return config.appEngine.locationId;
+    }
+
     /**
      * The filename of the logo to be displayed in the header of the registrar console.
      *

core/src/main/java/google/registry/config/RegistryConfigSettings.java

@@ -45,6 +45,7 @@ public class RegistryConfigSettings {
   /** Configuration options that apply to the entire App Engine project. */
   public static class AppEngine {
     public String projectId;
+    public String locationId;
     public boolean isLocal;
     public String defaultServiceUrl;
     public String backendServiceUrl;

core/src/main/java/google/registry/config/files/default-config.yaml

@@ -8,6 +8,10 @@
 appEngine:
   # Globally unique App Engine project ID
   projectId: registry-project-id
+  # Location of the App engine project, note that us-central1 and europe-west1 are special in that
+  # they are used without the trailing number in App Engine commands and Google Cloud Console.
+  # See: https://cloud.google.com/appengine/docs/locations
+  locationId: registry-location-id
 
   # whether to use local/test credentials when connecting to the servers
   isLocal: true

core/src/main/java/google/registry/cron/TldFanoutAction.java

@@ -14,14 +14,10 @@
 
 package google.registry.cron;
 
-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
-import static com.google.appengine.api.taskqueue.TaskOptions.Builder.withUrl;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Predicates.in;
 import static com.google.common.base.Predicates.not;
-import static com.google.common.base.Strings.nullToEmpty;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
-import static com.google.common.collect.Iterables.getFirst;
 import static com.google.common.collect.Multimaps.filterKeys;
 import static com.google.common.net.MediaType.PLAIN_TEXT_UTF_8;
 import static google.registry.cron.CronModule.ENDPOINT_PARAM;
@@ -36,21 +32,24 @@ import static google.registry.model.tld.Registry.TldType.REAL;
 import static google.registry.model.tld.Registry.TldType.TEST;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
-import com.google.appengine.api.taskqueue.Queue;
-import com.google.appengine.api.taskqueue.TaskHandle;
-import com.google.appengine.api.taskqueue.TaskOptions;
+import com.google.cloud.tasks.v2.Task;
+import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ImmutableListMultimap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Streams;
 import com.google.common.flogger.FluentLogger;
+import com.google.protobuf.Timestamp;
 import google.registry.request.Action;
+import google.registry.request.Action.Service;
 import google.registry.request.Parameter;
 import google.registry.request.ParameterMap;
 import google.registry.request.RequestParameters;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
-import google.registry.util.TaskQueueUtils;
+import google.registry.util.Clock;
+import google.registry.util.CloudTasksUtils;
+import java.time.Instant;
 import java.util.Optional;
 import java.util.Random;
 import java.util.stream.Stream;
@@ -105,7 +104,8 @@ public final class TldFanoutAction implements Runnable {
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
-  @Inject TaskQueueUtils taskQueueUtils;
+  @Inject Clock clock;
+  @Inject CloudTasksUtils cloudTasksUtils;
   @Inject Response response;
   @Inject @Parameter(ENDPOINT_PARAM) String endpoint;
   @Inject @Parameter(QUEUE_PARAM) String queue;
@@ -115,7 +115,9 @@ public final class TldFanoutAction implements Runnable {
   @Inject @Parameter(EXCLUDE_PARAM) ImmutableSet<String> excludes;
   @Inject @Parameter(JITTER_SECONDS_PARAM) Optional<Integer> jitterSeconds;
   @Inject @ParameterMap ImmutableListMultimap<String, String> params;
-  @Inject TldFanoutAction() {}
+
+  @Inject
+  TldFanoutAction() {}
 
   @Override
   public void run() {
@@ -126,8 +128,7 @@ public final class TldFanoutAction implements Runnable {
         runInEmpty || forEachTestTld || forEachRealTld,
         "At least one of runInEmpty, forEachTestTld, forEachRealTld must be given");
     checkArgument(
-        !(runInEmpty && !excludes.isEmpty()),
-        "Can't specify 'exclude' with 'runInEmpty'");
+        !(runInEmpty && !excludes.isEmpty()), "Can't specify 'exclude' with 'runInEmpty'");
     ImmutableSet<String> tlds =
         runInEmpty
             ? ImmutableSet.of("")
@@ -137,7 +138,6 @@ public final class TldFanoutAction implements Runnable {
                 .filter(not(in(excludes)))
                 .collect(toImmutableSet());
     Multimap<String, String> flowThruParams = filterKeys(params, not(in(CONTROL_PARAMS)));
-    Queue taskQueue = getQueue(queue);
     StringBuilder outputPayload =
         new StringBuilder(
             String.format("OK: Launched the following %d tasks in queue %s\n", tlds.size(), queue));
@@ -146,33 +146,41 @@ public final class TldFanoutAction implements Runnable {
       logger.atWarning().log("No TLDs to fan-out!");
     }
     for (String tld : tlds) {
-      TaskOptions taskOptions = createTaskOptions(tld, flowThruParams);
-      TaskHandle taskHandle = taskQueueUtils.enqueue(taskQueue, taskOptions);
+      Task task = createTask(tld, flowThruParams);
+      Task createdTask = cloudTasksUtils.enqueue(queue, task);
       outputPayload.append(
           String.format(
               "- Task: '%s', tld: '%s', endpoint: '%s'\n",
-              taskHandle.getName(), tld, taskOptions.getUrl()));
+              createdTask.getName(), tld, createdTask.getAppEngineHttpRequest().getRelativeUri()));
       logger.atInfo().log(
-          "Task: '%s', tld: '%s', endpoint: '%s'", taskHandle.getName(), tld, taskOptions.getUrl());
+          "Task: '%s', tld: '%s', endpoint: '%s'",
+          createdTask.getName(), tld, createdTask.getAppEngineHttpRequest().getRelativeUri());
     }
     response.setContentType(PLAIN_TEXT_UTF_8);
     response.setPayload(outputPayload.toString());
   }
 
-  private TaskOptions createTaskOptions(String tld, Multimap<String, String> params) {
-    TaskOptions options =
-        withUrl(endpoint)
-            .countdownMillis(
-                jitterSeconds
-                    .map(seconds -> random.nextInt((int) SECONDS.toMillis(seconds)))
-                    .orElse(0));
+  private Task createTask(String tld, Multimap<String, String> params) {
     if (!tld.isEmpty()) {
-      options.param(RequestParameters.PARAM_TLD, tld);
+      params = ArrayListMultimap.create(params);
+      params.put(RequestParameters.PARAM_TLD, tld);
     }
-    for (String param : params.keySet()) {
-      // TaskOptions.param() does not accept null values.
-      options.param(param, nullToEmpty(getFirst(params.get(param), null)));
-    }
-    return options;
+    Instant scheduleTime =
+        Instant.ofEpochMilli(
+            clock
+                .nowUtc()
+                .plusMillis(
+                    jitterSeconds
+                        .map(seconds -> random.nextInt((int) SECONDS.toMillis(seconds)))
+                        .orElse(0))
+                .getMillis());
+    return Task.newBuilder(
+            CloudTasksUtils.createPostTask(endpoint, Service.BACKEND.toString(), params))
+        .setScheduleTime(
+            Timestamp.newBuilder()
+                .setSeconds(scheduleTime.getEpochSecond())
+                .setNanos(scheduleTime.getNano())
+                .build())
+        .build();
   }
 }

core/src/main/java/google/registry/module/backend/BackendComponent.java

@@ -18,6 +18,7 @@ import com.google.monitoring.metrics.MetricReporter;
 import dagger.Component;
 import dagger.Lazy;
 import google.registry.bigquery.BigqueryModule;
+import google.registry.config.CloudTasksUtilsModule;
 import google.registry.config.CredentialModule;
 import google.registry.config.RegistryConfig.ConfigModule;
 import google.registry.dns.writer.VoidDnsWriterModule;
@@ -56,6 +57,7 @@ import javax.inject.Singleton;
       BackendRequestComponentModule.class,
       BigqueryModule.class,
       ConfigModule.class,
+      CloudTasksUtilsModule.class,
       CredentialModule.class,
       CustomLogicFactoryModule.class,
       DatastoreAdminModule.class,