Add support for cloud tasks within the emulator. (#7475)

* register cloud task triggers * Create and register task queue emulator * Add basic features of task queue emulator * Implement a queue to improve performance of tasks emulator * Remove unused import * Register task queues from functions emulator * change task queue endpoint to accept IDs in body instead of query params * set enviornment variable when task queue emulator is running * Remove bug where tasks were getting dispatched too fast * Rework task enqueueing API and support task deletion * properly support task deletion * Refactor Task Dispatching Algorithm * Add task deduplication for all tasks that have ever been added * Refactor Task Queue * Add tests for task queue emulator * respond to PR review * update firebase-config.json * use node-fetch * fix tests * Make tasks emulator start up whenever functions emulator starts * fix error message * Add statistics endpoint * Add in global task deduplication * Add headers to request * Review Changes * Remove console logs * Update backoff calculations * add changelog --------- Co-authored-by: blidd-google <112491344+blidd-google@users.noreply.github.com> Co-authored-by: joehan <joehanley@google.com> Co-authored-by: Brian Li <blidd@google.com>
firebase · Aug 27, 2024 · 2b1a314 · 2b1a314
1 parent dee2c89
commit 2b1a314
Show file tree

Hide file tree

Showing 14 changed files with 1,367 additions and 4 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,2 +1 @@
-- Released Firestore Emulator v 1.19.8 which adds support for `FindNearest.distanceResultField` and `FindNearest.distanceThreshold` parameters.
-- Temporarily removing extensions from functions deploy because it was causing unexpected extension deletions.
+- Add support for cloud tasks within the emulator. (#7475)
diff --git a/schema/firebase-config.json b/schema/firebase-config.json
@@ -500,6 +500,18 @@
                     },
                     "type": "object"
                 },
+                "tasks": {
+                    "additionalProperties": false,
+                    "properties": {
+                        "host": {
+                            "type": "string"
+                        },
+                        "port": {
+                            "type": "number"
+                        }
+                    },
+                    "type": "object"
+                },
                 "ui": {
                     "additionalProperties": false,
                     "properties": {

diff --git a/src/emulator/constants.ts b/src/emulator/constants.ts
@@ -14,6 +14,7 @@ export const DEFAULT_PORTS: { [s in Emulators]: number } = {
   storage: 9199,
   eventarc: 9299,
   dataconnect: 9399,
+  tasks: 9499,
 };
 
 export const FIND_AVAILBLE_PORT_BY_DEFAULT: Record<Emulators, boolean> = {
@@ -30,6 +31,7 @@ export const FIND_AVAILBLE_PORT_BY_DEFAULT: Record<Emulators, boolean> = {
   extensions: false,
   eventarc: true,
   dataconnect: true,
+  tasks: true,
 };
 
 export const EMULATOR_DESCRIPTION: Record<Emulators, string> = {
@@ -46,6 +48,7 @@ export const EMULATOR_DESCRIPTION: Record<Emulators, string> = {
   extensions: "Extensions Emulator",
   eventarc: "Eventarc Emulator",
   dataconnect: "Data Connect Emulator",
+  tasks: "Cloud Tasks Emulator",
 };
 
 export const DEFAULT_HOST = "localhost";
@@ -87,6 +90,9 @@ export class Constants {
   // Environment variable to discover the eventarc emulator.
   static CLOUD_EVENTARC_EMULATOR_HOST = "CLOUD_EVENTARC_EMULATOR_HOST";
 
+  // Environment variable to discover the tasks emulator.
+  static CLOUD_TASKS_EMULATOR_HOST = "CLOUD_TASKS_EMULATOR_HOST";
+
   // Environment variable to discover the Emulator HUB
   static FIREBASE_EMULATOR_HUB = "FIREBASE_EMULATOR_HUB";
   static FIREBASE_GA_SESSION = "FIREBASE_GA_SESSION";
@@ -95,7 +101,9 @@ export class Constants {
   static SERVICE_REALTIME_DATABASE = "firebaseio.com";
   static SERVICE_PUBSUB = "pubsub.googleapis.com";
   static SERVICE_EVENTARC = "eventarc.googleapis.com";
+  static SERVICE_CLOUD_TASKS = "cloudtasks.googleapis.com";
   static SERVICE_FIREALERTS = "firebasealerts.googleapis.com";
+
   // Note: the service name below are here solely for logging purposes.
   // There is not an emulator available for these.
   static SERVICE_ANALYTICS = "app-measurement.com";
@@ -127,6 +135,8 @@ export class Constants {
         return "test lab";
       case this.SERVICE_EVENTARC:
         return "eventarc";
+      case this.SERVICE_CLOUD_TASKS:
+        return "tasks";
       default:
         return service;
     }

diff --git a/src/emulator/controller.ts b/src/emulator/controller.ts
@@ -57,6 +57,7 @@ import { HostingEmulator } from "./hostingEmulator";
 import { PubsubEmulator } from "./pubsubEmulator";
 import { StorageEmulator } from "./storage";
 import { readFirebaseJson } from "../dataconnect/fileUtils";
+import { TasksEmulator } from "./tasksEmulator";
 
 const START_LOGGING_EMULATOR = utils.envOverride(
   "START_LOGGING_EMULATOR",
@@ -371,11 +372,13 @@ export async function startAll(
     // If we already know we need Functions (and Eventarc), assign them now.
     listenConfig[Emulators.FUNCTIONS] = getListenConfig(options, Emulators.FUNCTIONS);
     listenConfig[Emulators.EVENTARC] = getListenConfig(options, Emulators.EVENTARC);
+    listenConfig[Emulators.TASKS] = getListenConfig(options, Emulators.TASKS);
   }
   for (const emulator of ALL_EMULATORS) {
     if (
       emulator === Emulators.FUNCTIONS ||
       emulator === Emulators.EVENTARC ||
+      emulator === Emulators.TASKS ||
       // Same port as Functions, no need for separate assignment
       emulator === Emulators.EXTENSIONS ||
       (emulator === Emulators.UI && !showUI)
@@ -527,12 +530,13 @@ export async function startAll(
   }
 
   if (emulatableBackends.length) {
-    if (!listenForEmulator.functions || !listenForEmulator.eventarc) {
+    if (!listenForEmulator.functions || !listenForEmulator.eventarc || !listenForEmulator.tasks) {
       // We did not know that we need Functions and Eventarc earlier but now we do.
       listenForEmulator = await resolveHostAndAssignPorts({
         ...listenForEmulator,
         functions: listenForEmulator.functions ?? getListenConfig(options, Emulators.FUNCTIONS),
         eventarc: listenForEmulator.eventarc ?? getListenConfig(options, Emulators.EVENTARC),
+        tasks: listenForEmulator.eventarc ?? getListenConfig(options, Emulators.TASKS),
       });
       hubLogger.log("DEBUG", "late-assigned ports for functions and eventarc emulators", {
         user: listenForEmulator,
@@ -588,6 +592,14 @@ export async function startAll(
       port: eventarcAddr.port,
     });
     await startEmulator(eventarcEmulator);
+
+    const tasksAddr = legacyGetFirstAddr(Emulators.TASKS);
+    const tasksEmulator = new TasksEmulator({
+      host: tasksAddr.host,
+      port: tasksAddr.port,
+    });
+
+    await startEmulator(tasksEmulator);
   }
 
   if (listenForEmulator.firestore) {

diff --git a/src/emulator/env.ts b/src/emulator/env.ts
@@ -38,6 +38,9 @@ export function setEnvVarsForEmulators(
       case Emulators.EVENTARC:
         env[Constants.CLOUD_EVENTARC_EMULATOR_HOST] = `http://${host}`;
         break;
+      case Emulators.TASKS:
+        env[Constants.CLOUD_TASKS_EMULATOR_HOST] = host;
+        break;
     }
   }
 }
diff --git a/src/emulator/functionsEmulator.ts b/src/emulator/functionsEmulator.ts
@@ -654,6 +654,15 @@ export class FunctionsEmulator implements EmulatorInstance {
           definition.name,
           definition.region,
         );
+        if (definition.taskQueueTrigger) {
+          added = await this.addTaskQueueTrigger(
+            this.args.projectId,
+            definition.region,
+            definition.name,
+            url,
+            definition.taskQueueTrigger,
+          );
+        }
       } else if (definition.eventTrigger) {
         const service: string = getFunctionService(definition);
         const key = this.getTriggerKey(definition);
@@ -1195,6 +1204,35 @@ export class FunctionsEmulator implements EmulatorInstance {
     return true;
   }
 
+  async addTaskQueueTrigger(
+    projectId: string,
+    location: string,
+    entryPoint: string,
+    defaultUri: string,
+    taskQueueTrigger: backend.TaskQueueTrigger,
+  ): Promise<boolean> {
+    logger.debug(`addTaskQueueTrigger`, JSON.stringify(taskQueueTrigger));
+    if (!EmulatorRegistry.isRunning(Emulators.TASKS)) {
+      logger.debug(`addTaskQueueTrigger`, "TQ not running");
+      return Promise.resolve(false);
+    }
+    const bundle = {
+      ...taskQueueTrigger,
+      defaultUri,
+    };
+
+    try {
+      await EmulatorRegistry.client(Emulators.TASKS).post(
+        `/projects/${projectId}/locations/${location}/queues/${entryPoint}`,
+        bundle,
+      );
+      return true;
+    } catch (err) {
+      this.logger.log("WARN", "Error adding Task Queue function: " + err);
+      return false;
+    }
+  }
+
   getProjectId(): string {
     return this.args.projectId;
   }

diff --git a/src/emulator/functionsEmulatorShared.ts b/src/emulator/functionsEmulatorShared.ts
@@ -45,6 +45,7 @@ export interface ParsedTriggerDefinition {
   availableMemoryMb?: backend.MemoryOptions;
   httpsTrigger?: any;
   eventTrigger?: EventTrigger;
+  taskQueueTrigger?: backend.TaskQueueTrigger;
   schedule?: EventSchedule;
   blockingTrigger?: BlockingTrigger;
   labels?: { [key: string]: any };
@@ -241,8 +242,20 @@ export function emulatedFunctionsFromEndpoints(
         options: endpoint.blockingTrigger.options || {},
       };
     } else if (backend.isTaskQueueTriggered(endpoint)) {
-      // Just expose TQ trigger as HTTPS. Useful for debugging.
       def.httpsTrigger = {};
+      def.taskQueueTrigger = {
+        retryConfig: {
+          maxAttempts: endpoint.taskQueueTrigger.retryConfig?.maxAttempts,
+          maxRetrySeconds: endpoint.taskQueueTrigger.retryConfig?.maxRetrySeconds,
+          maxBackoffSeconds: endpoint.taskQueueTrigger.retryConfig?.maxBackoffSeconds,
+          maxDoublings: endpoint.taskQueueTrigger.retryConfig?.maxDoublings,
+          minBackoffSeconds: endpoint.taskQueueTrigger.retryConfig?.minBackoffSeconds,
+        },
+        rateLimits: {
+          maxConcurrentDispatches: endpoint.taskQueueTrigger.rateLimits?.maxConcurrentDispatches,
+          maxDispatchesPerSecond: endpoint.taskQueueTrigger.rateLimits?.maxDispatchesPerSecond,
+        },
+      };
     } else {
       // All other trigger types are not supported by the emulator
       // We leave both eventTrigger and httpTrigger attributes empty
@@ -347,6 +360,9 @@ export function getFunctionService(def: ParsedTriggerDefinition): string {
   if (def.httpsTrigger) {
     return "https";
   }
+  if (def.taskQueueTrigger) {
+    return Constants.SERVICE_CLOUD_TASKS;
+  }
 
   return "unknown";
 }

diff --git a/src/emulator/portUtils.ts b/src/emulator/portUtils.ts
@@ -205,6 +205,7 @@ const EMULATOR_CAN_LISTEN_ON_PRIMARY_ONLY: Record<PortName, boolean> = {
   functions: true,
   logging: true,
   storage: true,
+  tasks: true,
 
   // Only one hostname possible in .server mode, can switch to middleware later.
   hosting: true,

diff --git a/src/emulator/registry.ts b/src/emulator/registry.ts
@@ -88,6 +88,7 @@ export class EmulatorRegistry {
       storage: 3.5,
       eventarc: 3.6,
       dataconnect: 3.7,
+      tasks: 3.8,
 
       // Hub shuts down once almost everything else is done
       hub: 4,