Creates eventarc channel in before deploying custom event function (#5226)

* Ensure eventarc channels exist on deploy

* create channel before deploying custom event fn

* add unit tests

* minor comment edits

* remove confusing comment about channel name

Co-authored-by: Thomas Bouldin <[email protected]>

Author: blidd-google

src/api.ts

@@ -58,6 +58,7 @@ export const dynamicLinksKey = utils.envOverride(
   "FIREBASE_DYNAMIC_LINKS_KEY",
   "AIzaSyB6PtY5vuiSB8MNgt20mQffkOlunZnHYiQ"
 );
+export const eventarcOrigin = utils.envOverride("EVENTARC_URL", "https://eventarc.googleapis.com");
 export const firebaseApiOrigin = utils.envOverride(
   "FIREBASE_API_URL",
   "https://firebase.googleapis.com"

src/deploy/functions/release/fabricator.ts

@@ -6,14 +6,15 @@ import { SourceTokenScraper } from "./sourceTokenScraper";
 import { Timer } from "./timer";
 import { assertExhaustive } from "../../../functional";
 import { getHumanFriendlyRuntimeName } from "../runtimes";
-import { functionsOrigin, functionsV2Origin } from "../../../api";
+import { eventarcOrigin, functionsOrigin, functionsV2Origin } from "../../../api";
 import { logger } from "../../../logger";
 import * as args from "../args";
 import * as backend from "../backend";
 import * as cloudtasks from "../../../gcp/cloudtasks";
 import * as deploymentTool from "../../../deploymentTool";
 import * as gcf from "../../../gcp/cloudfunctions";
 import * as gcfV2 from "../../../gcp/cloudfunctionsv2";
+import * as eventarc from "../../../gcp/eventarc";
 import * as helper from "../functionsDeployHelper";
 import * as planner from "./planner";
 import * as poller from "../../../operation-poller";
@@ -41,6 +42,13 @@ const gcfV2PollerOptions: Omit<poller.OperationPollerOptions, "operationResource
   maxBackoff: 10_000,
 };
 
+const eventarcPollerOptions: Omit<poller.OperationPollerOptions, "operationResourceName"> = {
+  apiOrigin: eventarcOrigin,
+  apiVersion: "v1",
+  masterTimeout: 25 * 60 * 1_000, // 25 minutes is the maximum build time for a function
+  maxBackoff: 10_000,
+};
+
 export interface FabricatorArgs {
   executor: Executor;
   functionExecutor: Executor;
@@ -296,6 +304,37 @@ export class Fabricator {
         .catch(rethrowAs(endpoint, "create topic"));
     }
 
+    // Like Pub/Sub, GCF requires a channel to exist before allowing the function
+    // to be created. Like Pub/Sub we currently only support setting the name
+    // of a channel, so we can do this once during createFunction alone. But if
+    // Eventarc adds new features that we indulge in (e.g. 2P event providers)
+    // things will get much more complicated. We'll have to make sure we keep
+    // up to date on updates, and we will also have to worry about channels leftover
+    // after deletion possibly incurring bills due to events still being sent.
+    const channel = apiFunction.eventTrigger?.channel;
+    if (channel) {
+      await this.executor
+        .run(async () => {
+          try {
+            const op: { name: string } = await eventarc.createChannel({ name: channel });
+            return await poller.pollOperation<eventarc.Channel>({
+              ...eventarcPollerOptions,
+              pollerName: `create-${channel}-${endpoint.region}-${endpoint.id}`,
+              operationResourceName: op.name,
+            });
+          } catch (err: any) {
+            // if error status is 409, the channel already exists and we can deploy safely
+            if (err.status === 409) {
+              return;
+            }
+            throw new FirebaseError("Unexpected error creating Eventarc channel", {
+              original: err as Error,
+            });
+          }
+        })
+        .catch(rethrowAs(endpoint, "upsert eventarc channel"));
+    }
+
     const resultFunction = await this.functionExecutor
       .run(async () => {
         const op: { name: string } = await gcfV2.createFunction(apiFunction);
@@ -571,6 +610,10 @@ export class Fabricator {
     } else if (backend.isBlockingTriggered(endpoint)) {
       await this.unregisterBlockingTrigger(endpoint);
     }
+    // N.B. Like Pub/Sub topics, we don't delete Eventarc channels because we
+    // don't know if there are any subscriers or not. If we start supporting 2P
+    // channels, we might need to revist this or else the events will still get
+    // published and the customer will still get charged.
   }
 
   async upsertScheduleV1(endpoint: backend.Endpoint & backend.ScheduleTriggered): Promise<void> {

src/deploy/functions/release/reporter.ts

@@ -25,6 +25,7 @@ export type OperationType =
   | "upsert schedule"
   | "delete schedule"
   | "upsert task queue"
+  | "upsert eventarc channel"
   | "disable task queue"
   | "create topic"
   | "delete topic"

src/gcp/eventarc.ts

@@ -0,0 +1,94 @@
+import { Client } from "../apiv2";
+import { eventarcOrigin } from "../api";
+import { last } from "lodash";
+import { fieldMasks } from "./proto";
+
+export const API_VERSION = "v1";
+
+export interface Channel {
+  name: string;
+
+  /** Server-assigned uinique identifier. Format is a UUID4 */
+  uid?: string;
+
+  createTime?: string;
+  updateTime?: string;
+
+  /** If set, the channel will grant publish permissions to the 2P provider. */
+  provider?: string;
+
+  // BEGIN oneof transport
+  pubsubTopic?: string;
+  // END oneof transport
+
+  state?: "PENDING" | "ACTIVE" | "INACTIVE";
+
+  /** When the channel is `PENDING`, this token must be sent to the provider */
+  activationToken?: string;
+
+  cryptoKeyName?: string;
+}
+
+interface OperationMetadata {
+  createTime: string;
+  target: string;
+  verb: string;
+  requestedCancellation: boolean;
+  apiVersion: string;
+}
+
+interface Operation {
+  name: string;
+  metadata: OperationMetadata;
+  done: boolean;
+}
+
+const client = new Client({
+  urlPrefix: eventarcOrigin,
+  auth: true,
+  apiVersion: API_VERSION,
+});
+
+/**
+ * Gets a Channel.
+ */
+export async function getChannel(name: string): Promise<Channel | undefined> {
+  const res = await client.get<Channel>(name);
+  if (res.status === 404) {
+    return undefined;
+  }
+  return res.body;
+}
+
+/**
+ * Creates a channel.
+ */
+export async function createChannel(channel: Channel): Promise<Operation> {
+  // const body: Partial<Channel> = cloneDeep(channel);
+  const pathParts = channel.name.split("/");
+
+  const res = await client.post<Channel, Operation>(pathParts.slice(0, -1).join("/"), channel, {
+    queryParams: { channelId: last(pathParts)! },
+  });
+  return res.body;
+}
+
+/**
+ * Updates a channel to match the new spec.
+ * Only set fields are updated.
+ */
+export async function updateChannel(channel: Channel): Promise<Channel> {
+  const res = await client.put<Channel, Channel>(channel.name, channel, {
+    queryParams: {
+      updateMask: fieldMasks(channel).join(","),
+    },
+  });
+  return res.body;
+}
+
+/**
+ * Deletes a channel.
+ */
+export async function deleteChannel(name: string): Promise<void> {
+  await client.delete(name);
+}

src/test/deploy/functions/release/fabricator.spec.ts

@@ -6,6 +6,7 @@ import * as reporter from "../../../../deploy/functions/release/reporter";
 import * as executor from "../../../../deploy/functions/release/executor";
 import * as gcfNSV2 from "../../../../gcp/cloudfunctionsv2";
 import * as gcfNS from "../../../../gcp/cloudfunctions";
+import * as eventarcNS from "../../../../gcp/eventarc";
 import * as pollerNS from "../../../../operation-poller";
 import * as pubsubNS from "../../../../gcp/pubsub";
 import * as schedulerNS from "../../../../gcp/cloudscheduler";
@@ -24,6 +25,7 @@ describe("Fabricator", () => {
   // Stub all GCP APIs to make sure this test is hermetic
   let gcf: sinon.SinonStubbedInstance<typeof gcfNS>;
   let gcfv2: sinon.SinonStubbedInstance<typeof gcfNSV2>;
+  let eventarc: sinon.SinonStubbedInstance<typeof eventarcNS>;
   let poller: sinon.SinonStubbedInstance<typeof pollerNS>;
   let pubsub: sinon.SinonStubbedInstance<typeof pubsubNS>;
   let scheduler: sinon.SinonStubbedInstance<typeof schedulerNS>;
@@ -35,6 +37,7 @@ describe("Fabricator", () => {
   beforeEach(() => {
     gcf = sinon.stub(gcfNS);
     gcfv2 = sinon.stub(gcfNSV2);
+    eventarc = sinon.stub(eventarcNS);
     poller = sinon.stub(pollerNS);
     pubsub = sinon.stub(pubsubNS);
     scheduler = sinon.stub(schedulerNS);
@@ -58,6 +61,10 @@ describe("Fabricator", () => {
     gcfv2.createFunction.rejects(new Error("unexpected gcfv2.createFunction"));
     gcfv2.updateFunction.rejects(new Error("unexpected gcfv2.updateFunction"));
     gcfv2.deleteFunction.rejects(new Error("unexpected gcfv2.deleteFunction"));
+    eventarc.createChannel.rejects(new Error("unexpected eventarc.createChannel"));
+    eventarc.deleteChannel.rejects(new Error("unexpected eventarc.deleteChannel"));
+    eventarc.getChannel.rejects(new Error("unexpected eventarc.getChannel"));
+    eventarc.updateChannel.rejects(new Error("unexpected eventarc.updateChannel"));
     run.getIamPolicy.rejects(new Error("unexpected run.getIamPolicy"));
     run.setIamPolicy.rejects(new Error("unexpected run.setIamPolicy"));
     run.setInvokerCreate.rejects(new Error("unexpected run.setInvokerCreate"));
@@ -481,6 +488,97 @@ describe("Fabricator", () => {
       );
     });
 
+    it("handles already existing eventarc channels", async () => {
+      eventarc.createChannel.callsFake(({ name }) => {
+        expect(name).to.equal("channel");
+        const err = new Error("Already exists");
+        (err as any).status = 409;
+        return Promise.reject(err);
+      });
+      gcfv2.createFunction.resolves({ name: "op", done: false });
+      poller.pollOperation.resolves({ serviceConfig: { service: "service" } });
+
+      const ep = endpoint(
+        {
+          eventTrigger: {
+            eventType: "custom.test.event",
+            channel: "channel",
+            retry: false,
+          },
+        },
+        {
+          platform: "gcfv2",
+        }
+      );
+
+      await fab.createV2Function(ep);
+      expect(eventarc.createChannel).to.have.been.called;
+      expect(gcfv2.createFunction).to.have.been.called;
+    });
+
+    it("creates channels if necessary", async () => {
+      const channelName = "channel";
+      eventarc.createChannel.callsFake(({ name }) => {
+        expect(name).to.equal(channelName);
+        return Promise.resolve({
+          name: "op-resource-name",
+          metadata: {
+            createTime: "",
+            target: "",
+            verb: "",
+            requestedCancellation: false,
+            apiVersion: "",
+          },
+          done: false,
+        });
+      });
+      gcfv2.createFunction.resolves({ name: "op", done: false });
+      poller.pollOperation.resolves({ serviceConfig: { service: "service" } });
+
+      const ep = endpoint(
+        {
+          eventTrigger: {
+            eventType: "custom.test.event",
+            channel: channelName,
+            retry: false,
+          },
+        },
+        {
+          platform: "gcfv2",
+        }
+      );
+
+      await fab.createV2Function(ep);
+      expect(eventarc.createChannel).to.have.been.calledOnceWith({ name: channelName });
+      expect(poller.pollOperation).to.have.been.called;
+    });
+
+    it("wraps errors thrown while creating channels", async () => {
+      eventarc.createChannel.callsFake(() => {
+        const err = new Error("🤷‍♂️");
+        (err as any).status = 400;
+        return Promise.reject(err);
+      });
+
+      const ep = endpoint(
+        {
+          eventTrigger: {
+            eventType: "custom.test.event",
+            channel: "channel",
+            retry: false,
+          },
+        },
+        {
+          platform: "gcfv2",
+        }
+      );
+
+      await expect(fab.createV2Function(ep)).to.eventually.be.rejectedWith(
+        reporter.DeploymentError,
+        "upsert eventarc channel"
+      );
+    });
+
     it("throws on create function failure", async () => {
       gcfv2.createFunction.rejects(new Error("Server failure"));
 
@@ -1168,7 +1266,7 @@ describe("Fabricator", () => {
       await fab.setTrigger(endpoint({ httpsTrigger: {} }));
     });
 
-    it("does nothing for event triggers", async () => {
+    it("does nothing for event triggers without channels", async () => {
       // all APIs throw by default
       const ep = endpoint({
         eventTrigger: {