Add events-backend-module-google-pubsub

Signed-off-by: Fredrik Adelöw <freben@gmail.com>
2025-04-15 10:45:12 +02:00
parent 1a85f9d7b8
commit af853ef1bc
19 changed files with 1226 additions and 3 deletions
@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-events-backend-module-google-pubsub': minor
+---
+
+Added a module that is able to transfer messages from Google Pub/Sub subscriptions into the Backstage events system.
@@ -45,6 +45,7 @@
    "@backstage/plugin-catalog-backend-module-unprocessed": "workspace:^",
    "@backstage/plugin-devtools-backend": "workspace:^",
    "@backstage/plugin-events-backend": "workspace:^",
+    "@backstage/plugin-events-backend-module-google-pubsub": "workspace:^",
    "@backstage/plugin-kubernetes-backend": "workspace:^",
    "@backstage/plugin-notifications-backend": "workspace:^",
    "@backstage/plugin-permission-backend": "workspace:^",
@@ -61,4 +61,5 @@ backend.add(import('@backstage/plugin-signals-backend'));
 backend.add(import('@backstage/plugin-notifications-backend'));
 backend.add(import('./instanceMetadata'));

+backend.add(import('@backstage/plugin-events-backend-module-google-pubsub'));
 backend.start();
@@ -0,0 +1 @@
+module.exports = require('@backstage/cli/config/eslint-factory')(__dirname);
@@ -0,0 +1,42 @@
+# @backstage/plugin-events-backend-module-google-pubsub
+
+This package is a module for the `events-backend` backend plugin
+and extends the events system with Google Pub/Sub related capabilities.
+
+## Configuration
+
+The following configuration enables the transfer of messages from a Google
+Pub/Sub subscription into a topic on the Backstage events system.
+
+```yaml
+events:
+  modules:
+    googlePubSub:
+      googlePubSubConsumingEventPublisher:
+        subscriptions:
+          # A unique key for your subscription, to be used in logging and metrics
+          mySubscription:
+            # The fully qualified name of the subscription
+            subscriptionName: 'projects/my-google-project/subscriptions/github-enterprise-events'
+            # The event system topic to transfer to. This can also be just a plain string
+            targetTopic:
+              # This example picks the topic name from a message attribute + a prefix
+              fromMessageAttribute:
+                attributeName: 'x-github-event'
+                withPrefix: 'github.'
+```
+
+## Installation
+
+1. Install this module
+2. Add your configuration.
+
+```bash
+# From your Backstage root directory
+yarn --cwd packages/backend add @backstage/plugin-events-backend-module-google-pubsub
+```
+
+```ts
+// packages/backend/src/index.ts
+backend.add(import('@backstage/plugin-events-backend-module-google-pubsub'));
+```
@@ -0,0 +1,10 @@
+apiVersion: backstage.io/v1alpha1
+kind: Component
+metadata:
+  name: backstage-plugin-events-backend-module-google-pubsub
+  title: '@backstage/plugin-events-backend-module-google-pubsub'
+  description: The google-pubsub backend module for the events plugin.
+spec:
+  lifecycle: experimental
+  type: backstage-backend-plugin-module
+  owner: maintainers
@@ -0,0 +1,96 @@
+/*
+ * Copyright 2022 The Backstage Authors
+ *
+ * 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.
+ */
+
+export interface Config {
+  events?: {
+    modules?: {
+      /**
+       * events-backend-module-google-pubsub configuration.
+       */
+      googlePubSub?: {
+        /**
+         * Configuration for `GooglePubSubConsumingEventPublisher`, which
+         * consumes messages from a Google Pub/Sub subscription and forwards
+         * them into the Backstage events system.
+         */
+        googlePubSubConsumingEventPublisher?: {
+          /**
+           * Generally contains a record per subscription to consume.
+           */
+          subscriptions: {
+            /**
+             * The name can be anything, but it is recommended to use only
+             * letters, numbers, and hyphens for this identifier since it will
+             * appear in logs and metric names etc.
+             */
+            [name: string]: {
+              /**
+               * The complete name of the Pub/Sub subscription to be used, on the
+               * form
+               * `projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID`.
+               */
+              subscriptionName: string;
+
+              /**
+               * The name of the events backend topic to which the messages are
+               * to be forwarded.
+               *
+               * @remarks
+               *
+               * The value can contain placeholders on the form `{{ message.attributes.foo }}`,
+               * to mirror attribute `foo` as the whole or part of the topic name.
+               *
+               * @example
+               *
+               * This example expects the Pub/Sub topic to contain GitHub
+               * webhook events where the HTTP headers were mapped into
+               * message attributes. The outcome should be that messages
+               * end up on event topics such as `github.push`,
+               * `github.repository` etc which matches the [`@backstage/plugin-events-backend-module-github`](https://github.com/backstage/backstage/tree/master/plugins/events-backend-module-github) structure.
+               *
+               * ```yaml
+               * targetTopic: 'github.{{ message.attributes.x-github-event }}'
+               * ```
+               */
+              targetTopic: string;
+
+              /**
+               * Pub/Sub message attributes are by default copied to the event
+               * metadata field. This setting allows you to override or amend
+               * that metadata.
+               *
+               * @remarks
+               *
+               * The values can contain placeholders on the form `{{ message.attributes.foo }}`,
+               * to mirror attribute `foo` as the whole or part of a metadata value.
+               *
+               * @example
+               *
+               * ```yaml
+               * eventMetadata:
+               *   x-gitHub-event: '{{ message.attributes.event }}'
+               * ```
+               */
+              eventMetadata?: {
+                [key: string]: string;
+              };
+            };
+          };
+        };
+      };
+    };
+  };
+}
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import { createBackend } from '@backstage/backend-defaults';
+import {
+  coreServices,
+  createBackendPlugin,
+} from '@backstage/backend-plugin-api';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
+
+const backend = createBackend();
+
+backend.add(import('@backstage/plugin-events-backend'));
+backend.add(import('../src'));
+
+backend.add(
+  createBackendPlugin({
+    pluginId: 'example-event-consumer',
+    register(reg) {
+      reg.registerInit({
+        deps: {
+          events: eventsServiceRef,
+          logger: coreServices.logger,
+        },
+        async init({ events, logger }) {
+          events.subscribe({
+            id: 'example-subscription',
+            topics: ['example-topic'],
+            async onEvent(event) {
+              logger.info(
+                `Received event: ${JSON.stringify(
+                  {
+                    topic: event.topic,
+                    payload: event.eventPayload,
+                    metadata: event.metadata,
+                  },
+                  null,
+                  2,
+                )}`,
+              );
+            },
+          });
+        },
+      });
+    },
+  }),
+);
+
+backend.start();
@@ -0,0 +1,54 @@
+{
+  "name": "@backstage/plugin-events-backend-module-google-pubsub",
+  "version": "0.0.0",
+  "description": "The google-pubsub backend module for the events plugin.",
+  "backstage": {
+    "role": "backend-plugin-module",
+    "pluginId": "events",
+    "pluginPackage": "@backstage/plugin-events-backend"
+  },
+  "publishConfig": {
+    "access": "public",
+    "main": "dist/index.cjs.js",
+    "types": "dist/index.d.ts"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/backstage/backstage",
+    "directory": "plugins/events-backend-module-google-pubsub"
+  },
+  "license": "Apache-2.0",
+  "main": "src/index.ts",
+  "types": "src/index.ts",
+  "files": [
+    "config.d.ts",
+    "dist"
+  ],
+  "scripts": {
+    "build": "backstage-cli package build",
+    "clean": "backstage-cli package clean",
+    "lint": "backstage-cli package lint",
+    "prepack": "backstage-cli package prepack",
+    "postpack": "backstage-cli package postpack",
+    "start": "backstage-cli package start",
+    "test": "backstage-cli package test"
+  },
+  "dependencies": {
+    "@backstage/backend-plugin-api": "workspace:^",
+    "@backstage/config": "workspace:^",
+    "@backstage/errors": "workspace:^",
+    "@backstage/plugin-events-node": "workspace:^",
+    "@backstage/types": "workspace:^",
+    "@google-cloud/pubsub": "^4.10.0",
+    "@opentelemetry/api": "^1.9.0",
+    "lodash": "^4.17.21"
+  },
+  "devDependencies": {
+    "@backstage/backend-defaults": "workspace:^",
+    "@backstage/backend-test-utils": "workspace:^",
+    "@backstage/cli": "workspace:^",
+    "@backstage/plugin-events-backend": "workspace:^",
+    "wait-for-expect": "^3.0.2"
+  },
+  "configSchema": "config.d.ts"
+}
@@ -0,0 +1,11 @@
+## API Report File for "@backstage/plugin-events-backend-module-google-pubsub"
+
+> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
+
+```ts
+import { BackendFeature } from '@backstage/backend-plugin-api';
+
+// @public
+const eventsModuleGooglePubsubConsumingEventPublisher: BackendFeature;
+export default eventsModuleGooglePubsubConsumingEventPublisher;
+```
@@ -0,0 +1,120 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import { mockServices } from '@backstage/backend-test-utils';
+import { Message, PubSub } from '@google-cloud/pubsub';
+import { GooglePubSubConsumingEventPublisher } from './GooglePubSubConsumingEventPublisher';
+import { JsonObject } from '@backstage/types';
+import waitFor from 'wait-for-expect';
+
+function makeMessage(
+  data: JsonObject,
+  attributes: Record<string, string>,
+): Message {
+  const message: Partial<Message> = {
+    attributes,
+    data: Buffer.from(JSON.stringify(data), 'utf-8'),
+    ack: jest.fn(),
+    nack: jest.fn(),
+  };
+  return message as Message;
+}
+
+describe('GooglePubSubConsumingEventPublisher', () => {
+  const logger = mockServices.logger.mock();
+  const events = mockServices.events.mock();
+
+  let onMessage: undefined | ((message: Message) => void);
+  const subscription = {
+    close: jest.fn(),
+    on: jest.fn((event, fn) => {
+      if (event === 'message') {
+        onMessage = fn;
+      }
+    }),
+  };
+
+  const pubSub = {
+    close: jest.fn(),
+    subscription: jest.fn(() => subscription),
+  };
+  const pubSubFactory = jest.fn(() => pubSub as unknown as PubSub);
+
+  beforeEach(() => {
+    onMessage = undefined;
+    jest.clearAllMocks();
+  });
+
+  it('should go though the expected registration and flows', async () => {
+    const publisher = new GooglePubSubConsumingEventPublisher({
+      logger,
+      events,
+      tasks: [
+        {
+          id: 'my-id',
+          project: 'my-project',
+          subscription: 'my-subscription',
+          mapToTopic: () => 'my-topic',
+          mapToMetadata: m => ({ ...m.attributes, more: 'yes' }),
+        },
+      ],
+      pubSubFactory,
+    });
+
+    // Start up
+
+    await publisher.start();
+
+    expect(pubSubFactory).toHaveBeenCalledWith('my-project');
+    expect(pubSub.subscription).toHaveBeenCalledWith(
+      'my-subscription',
+      expect.anything(),
+    );
+
+    // Publish successfully
+
+    let message = makeMessage({ foo: 'bar' }, { extra: 'data' });
+    onMessage!(message);
+
+    await waitFor(() => {
+      expect(events.publish).toHaveBeenCalledWith({
+        topic: 'my-topic',
+        eventPayload: { foo: 'bar' },
+        metadata: { extra: 'data', more: 'yes' },
+      });
+      expect(message.ack).toHaveBeenCalled();
+    });
+
+    // Simulate a failed publish
+
+    events.publish.mockRejectedValueOnce(new Error('Failed to publish'));
+
+    message = makeMessage({ foo: 'bar' }, { extra: 'data' });
+    onMessage!(message);
+
+    await waitFor(() => {
+      expect(message.nack).toHaveBeenCalled();
+      expect(message.ack).not.toHaveBeenCalled();
+    });
+
+    // Shut down
+
+    await publisher.stop();
+
+    expect(subscription.close).toHaveBeenCalled();
+    expect(pubSub.close).toHaveBeenCalled();
+  });
+});
@@ -0,0 +1,210 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import {
+  LoggerService,
+  RootConfigService,
+  RootLifecycleService,
+} from '@backstage/backend-plugin-api';
+import { EventParams, EventsService } from '@backstage/plugin-events-node';
+import { Message, PubSub, Subscription } from '@google-cloud/pubsub';
+import { Counter, metrics } from '@opentelemetry/api';
+import { readSubscriptionTasksFromConfig } from './config';
+import { SubscriptionTask } from './types';
+
+/**
+ * Reads messages off of Google Pub/Sub subscriptions and forwards them into the
+ * Backstage events system.
+ */
+export class GooglePubSubConsumingEventPublisher {
+  readonly #logger: LoggerService;
+  readonly #events: EventsService;
+  readonly #tasks: SubscriptionTask[];
+  readonly #pubSubFactory: (projectId: string) => PubSub;
+  readonly #metrics: { messages: Counter };
+  #activeClientsByProjectId: Map<string, PubSub>;
+  #activeSubscriptions: Subscription[];
+
+  static create(options: {
+    config: RootConfigService;
+    logger: LoggerService;
+    rootLifecycle: RootLifecycleService;
+    events: EventsService;
+  }) {
+    const publisher = new GooglePubSubConsumingEventPublisher({
+      logger: options.logger,
+      events: options.events,
+      tasks: readSubscriptionTasksFromConfig(options.config),
+      pubSubFactory: projectId => new PubSub({ projectId }),
+    });
+
+    options.rootLifecycle.addStartupHook(async () => {
+      await publisher.start();
+    });
+
+    options.rootLifecycle.addBeforeShutdownHook(async () => {
+      await publisher.stop();
+    });
+
+    return publisher;
+  }
+
+  constructor(options: {
+    logger: LoggerService;
+    events: EventsService;
+    tasks: SubscriptionTask[];
+    pubSubFactory: (projectId: string) => PubSub;
+  }) {
+    this.#logger = options.logger;
+    this.#events = options.events;
+    this.#tasks = options.tasks;
+    this.#pubSubFactory = options.pubSubFactory;
+
+    const meter = metrics.getMeter('default');
+    this.#metrics = {
+      messages: meter.createCounter(
+        'events.google.pubsub.consumer.messages.total',
+        {
+          description:
+            'Number of Pub/Sub messages received by GooglePubSubConsumingEventPublisher',
+          unit: 'short',
+        },
+      ),
+    };
+
+    this.#activeClientsByProjectId = new Map();
+    this.#activeSubscriptions = [];
+  }
+
+  async start() {
+    for (const task of this.#tasks) {
+      this.#logger.info(
+        `Starting subscription: id=${task.id} project=${task.project} subscription=${task.subscription}`,
+      );
+
+      let pubsub = this.#activeClientsByProjectId.get(task.project);
+      if (!pubsub) {
+        pubsub = this.#pubSubFactory(task.project);
+        this.#activeClientsByProjectId.set(task.project, pubsub);
+      }
+
+      // You cannot control the actual batch size delivered to the client from
+      // pubsub, so these settings actually instead control the rate at which
+      // messages are released to our event handlers by the pubsub library. This
+      // means that there may be significantly more than maxMessages messages
+      // pending in memory before we see them. Thus, the settings here are rather
+      // chosen so as to limit the concurrency of hammering consumers (the catalog
+      // etc).
+      const subscription = pubsub.subscription(task.subscription, {
+        flowControl: {
+          maxMessages: 5,
+          allowExcessMessages: false,
+        },
+      });
+
+      this.#activeSubscriptions.push(subscription);
+
+      subscription.on('error', error => {
+        this.#logger.error(
+          `Error reading Google Pub/Sub subscription: ${task.id}`,
+          error,
+        );
+      });
+
+      subscription.on('message', async message => {
+        let event: EventParams;
+        try {
+          event = this.#messageToEvent(message, task)!;
+          if (!event) {
+            this.#metrics.messages.add(1, {
+              topic: 'unknown',
+              status: 'ignored',
+            });
+            return;
+          }
+        } catch (error) {
+          this.#logger.error('Error processing Google Pub/Sub message', error);
+          this.#metrics.messages.add(1, {
+            topic: 'unknown',
+            status: 'failed',
+          });
+          // We unconditionally ACK the message in this case, because if it's
+          // broken, it will still be broken next time around, so there is no
+          // point in re-delivering it.
+          message.ack();
+          return;
+        }
+
+        try {
+          await this.#events.publish(event);
+          this.#metrics.messages.add(1, {
+            topic: event.topic,
+            status: 'success',
+          });
+          message.ack();
+        } catch (error) {
+          this.#logger.error('Error processing Google Pub/Sub message', error);
+          this.#metrics.messages.add(1, {
+            topic: event.topic,
+            status: 'failed',
+          });
+          // We fast-NACK the message in this case because this may be a
+          // transient problem with the events backend.
+          message.nack();
+        }
+      });
+    }
+  }
+
+  async stop() {
+    const subscriptions = this.#activeSubscriptions;
+    const clients = Array.from(this.#activeClientsByProjectId.values());
+
+    this.#activeSubscriptions = [];
+    this.#activeClientsByProjectId = new Map();
+
+    await Promise.allSettled(
+      subscriptions.map(async subscription => {
+        this.#logger.info(
+          `Closing Google Pub/Sub subscription: ${subscription.name}`,
+        );
+        await subscription.close();
+      }),
+    );
+
+    await Promise.allSettled(
+      clients.map(async client => {
+        this.#logger.info(`Closing Google Pub/Sub client: ${client.projectId}`);
+        await client.close();
+      }),
+    );
+  }
+
+  #messageToEvent(
+    message: Message,
+    task: SubscriptionTask,
+  ): EventParams | undefined {
+    const topic = task.mapToTopic(message);
+    if (!topic) {
+      return undefined;
+    }
+    return {
+      topic,
+      eventPayload: JSON.parse(message.data.toString()),
+      metadata: task.mapToMetadata(message),
+    };
+  }
+}
@@ -0,0 +1,234 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import { mockServices } from '@backstage/backend-test-utils';
+import { JsonObject } from '@backstage/types';
+import { Message } from '@google-cloud/pubsub';
+import {
+  createPatternResolver,
+  readSubscriptionTasksFromConfig,
+} from './config';
+
+function makeMessage(
+  data: JsonObject,
+  attributes: Record<string, string>,
+): Message {
+  const message: Partial<Message> = {
+    attributes,
+    data: Buffer.from(JSON.stringify(data), 'utf-8'),
+    ack: jest.fn(),
+    nack: jest.fn(),
+  };
+  return message as Message;
+}
+
+describe('readSubscriptionTasksFromConfig', () => {
+  it('reads with basic targetTopic', () => {
+    const data = {
+      events: {
+        modules: {
+          googlePubSub: {
+            googlePubSubConsumingEventPublisher: {
+              subscriptions: {
+                subKey: {
+                  subscriptionName: 'projects/pid/subscriptions/sid',
+                  targetTopic: 'my-topic',
+                },
+              },
+            },
+          },
+        },
+      },
+    };
+
+    const result = readSubscriptionTasksFromConfig(
+      mockServices.rootConfig({ data }),
+    );
+
+    expect(result).toEqual([
+      {
+        id: 'subKey',
+        project: 'pid',
+        subscription: 'sid',
+        mapToTopic: expect.any(Function),
+        mapToMetadata: expect.any(Function),
+      },
+    ]);
+
+    expect(
+      result[0].mapToTopic(makeMessage({ foo: 'bar' }, { attr: 'yes' })),
+    ).toBe('my-topic');
+    expect(
+      result[0].mapToMetadata(makeMessage({ foo: 'bar' }, { attr: 'yes' })),
+    ).toEqual({ attr: 'yes' });
+  });
+
+  it('fills in placeholders', () => {
+    const data = {
+      events: {
+        modules: {
+          googlePubSub: {
+            googlePubSubConsumingEventPublisher: {
+              subscriptions: {
+                sub1: {
+                  subscriptionName: 'projects/pid/subscriptions/sid',
+                  targetTopic: 't.{{ message.attributes.exists }}',
+                  eventMetadata: {
+                    meta1: 'updated.{{ message.attributes.exists }}',
+                    meta2: 'updated.{{ message.attributes.missing }}',
+                  },
+                },
+                sub2: {
+                  subscriptionName: 'projects/pid/subscriptions/sid',
+                  targetTopic: 't.{{ message.attributes.missing }}',
+                  eventMetadata: {
+                    meta3: 'new',
+                  },
+                },
+              },
+            },
+          },
+        },
+      },
+    };
+
+    const result = readSubscriptionTasksFromConfig(
+      mockServices.rootConfig({ data }),
+    );
+
+    expect(result).toEqual([
+      {
+        id: 'sub1',
+        project: 'pid',
+        subscription: 'sid',
+        mapToTopic: expect.any(Function),
+        mapToMetadata: expect.any(Function),
+      },
+      {
+        id: 'sub2',
+        project: 'pid',
+        subscription: 'sid',
+        mapToTopic: expect.any(Function),
+        mapToMetadata: expect.any(Function),
+      },
+    ]);
+
+    expect(
+      result[0].mapToTopic(
+        makeMessage(
+          { foo: 'bar' },
+          { exists: 'exists', meta1: 'original1', meta2: 'original2' },
+        ),
+      ),
+    ).toBe('t.exists'); // Message attribute existed, successfully routed
+    expect(
+      result[0].mapToMetadata(
+        makeMessage(
+          { foo: 'bar' },
+          { exists: 'exists', meta1: 'original1', meta2: 'original2' },
+        ),
+      ),
+    ).toEqual({
+      exists: 'exists',
+      meta1: 'updated.exists', // message attribute existed, was replaced
+      meta2: 'original2', // message attribute did not exist, was not replaced
+    });
+
+    expect(
+      result[1].mapToTopic(
+        makeMessage(
+          { foo: 'bar' },
+          { exists: 'exists', meta1: 'original1', meta2: 'original2' },
+        ),
+      ),
+    ).toBeUndefined(); // Message attribute did not exist, could not be routed
+    expect(
+      result[1].mapToMetadata(
+        makeMessage(
+          { foo: 'bar' },
+          { exists: 'exists', meta1: 'original1', meta2: 'original2' },
+        ),
+      ),
+    ).toEqual({
+      exists: 'exists',
+      meta1: 'original1',
+      meta2: 'original2',
+      meta3: 'new',
+    });
+  });
+
+  it('rejects malformed subscription name', () => {
+    const data = {
+      events: {
+        modules: {
+          googlePubSub: {
+            googlePubSubConsumingEventPublisher: {
+              subscriptions: {
+                subKey: {
+                  subscriptionName: 'sid',
+                  targetTopic: 'foo',
+                },
+              },
+            },
+          },
+        },
+      },
+    };
+
+    expect(() =>
+      readSubscriptionTasksFromConfig(mockServices.rootConfig({ data })),
+    ).toThrowErrorMatchingInlineSnapshot(
+      `"Expected Googoe Pub/Sub 'subscriptionName' to be on the form 'projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID' but got 'sid'"`,
+    );
+  });
+});
+
+describe('createPatternResolver', () => {
+  it('resolves patterns as expected', () => {
+    expect(createPatternResolver('foo')({})).toEqual('foo');
+    expect(createPatternResolver('{{a.b}}')({ a: { b: 'test' } })).toEqual(
+      'test',
+    );
+    expect(createPatternResolver('{{a.b}}')({ a: { b: '7' } })).toEqual('7');
+    expect(
+      createPatternResolver('{{a.b-c}}')({ a: { 'b-c': 'test' } }),
+    ).toEqual('test');
+    expect(
+      createPatternResolver("{{a['b-c']}}")({ a: { 'b-c': 'test' } }),
+    ).toEqual('test');
+    expect(
+      createPatternResolver(' {{ a.b }} ')({ a: { b: ' test ' } }),
+    ).toEqual('  test  ');
+  });
+
+  it('throws on bad / missing context values', () => {
+    expect(() =>
+      createPatternResolver('{{a.b}}')({ a: { b: [7] } }),
+    ).toThrowErrorMatchingInlineSnapshot(
+      `"Expected string or number value for selector 'a.b', got object"`,
+    );
+    expect(() =>
+      createPatternResolver('{{a.b}}')({ a: {} }),
+    ).toThrowErrorMatchingInlineSnapshot(`"No value for selector 'a.b'"`);
+  });
+
+  it('just passes down broken parts of patterns', () => {
+    expect(createPatternResolver('-{{a}}-}}')({ a: 1 })).toEqual('-1-}}');
+    expect(createPatternResolver('{{-{{a}}-')({ a: 1 })).toEqual('{{-1-');
+    expect(createPatternResolver('{{-{{a}}-}}')({ a: 1 })).toEqual('{{-1-}}');
+    expect(createPatternResolver('{{-{{}}-}}')({ a: 1 })).toEqual('{{--}}');
+  });
+});
@@ -0,0 +1,178 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import { RootConfigService } from '@backstage/backend-plugin-api';
+import { Config } from '@backstage/config';
+import { InputError } from '@backstage/errors';
+import { Message } from '@google-cloud/pubsub';
+import { SubscriptionTask } from './types';
+import lodash from 'lodash';
+
+export function readSubscriptionTasksFromConfig(
+  rootConfig: RootConfigService,
+): SubscriptionTask[] {
+  const subscriptionsConfig = rootConfig.getOptionalConfig(
+    'events.modules.googlePubSub.googlePubSubConsumingEventPublisher.subscriptions',
+  );
+  if (!subscriptionsConfig) {
+    return [];
+  }
+
+  return subscriptionsConfig.keys().map(subscriptionId => {
+    if (!subscriptionId.match(/^[-_\w]+$/)) {
+      throw new InputError(
+        `Expected Googoe Pub/Sub subscription ID to consist of letters, numbers, dashes and lodashes, but got '${subscriptionId}'`,
+      );
+    }
+
+    const config = subscriptionsConfig.getConfig(subscriptionId);
+    const { project, subscription } = readSubscriptionName(config);
+    const mapToTopic = readTopicMapper(config);
+    const mapToMetadata = readMetadataMapper(config);
+
+    return {
+      id: subscriptionId,
+      project,
+      subscription,
+      mapToTopic,
+      mapToMetadata,
+    };
+  });
+}
+
+function readSubscriptionName(config: Config): {
+  project: string;
+  subscription: string;
+} {
+  const subscriptionName = config.getString('subscriptionName');
+  const parts = subscriptionName.match(
+    /^projects\/([^/]+)\/subscriptions\/(.+)$/,
+  );
+  if (!parts) {
+    throw new InputError(
+      `Expected Googoe Pub/Sub 'subscriptionName' to be on the form 'projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID' but got '${subscriptionName}'`,
+    );
+  }
+  return {
+    project: parts[1],
+    subscription: parts[2],
+  };
+}
+
+/**
+ * Handles the `targetTopic` configuration field.
+ */
+function readTopicMapper(
+  config: Config,
+): (message: Message) => string | undefined {
+  const targetTopicPattern = config.getString('targetTopic');
+  const patternResolver = createPatternResolver(targetTopicPattern);
+  return message => {
+    try {
+      return patternResolver({ message });
+    } catch {
+      // could not map to a topic
+      return undefined;
+    }
+  };
+}
+
+/**
+ * Handles the `eventMetadata` configuration field.
+ */
+function readMetadataMapper(
+  config: Config,
+): (message: Message) => Record<string, string> {
+  const setters = new Array<
+    (options: { message: Message; metadata: Record<string, string> }) => void
+  >();
+
+  const eventMetadata = config.getOptionalConfig('eventMetadata');
+  if (eventMetadata) {
+    for (const key of eventMetadata?.keys() ?? []) {
+      const valuePattern = eventMetadata.getString(key);
+      const patternResolver = createPatternResolver(valuePattern);
+      setters.push(({ message, metadata }) => {
+        try {
+          const value = patternResolver({ message });
+          if (value) {
+            metadata[key] = value;
+          }
+        } catch {
+          // ignore silently, keep original unchanged
+        }
+      });
+    }
+  }
+
+  return message => {
+    const result: Record<string, string> = {
+      ...message.attributes,
+    };
+    for (const setter of setters) {
+      setter({ message, metadata: result });
+    }
+    return result;
+  };
+}
+
+/**
+ * Takes a pattern string that may contain `{{ path.to.value }}` placeholders,
+ * and returns a function that accepts a context object and returns strings that
+ * have had its placeholders filled in by using `lodash.get` accordingly on the
+ * context.
+ */
+export function createPatternResolver(
+  pattern: string,
+): (context: unknown) => string {
+  // This split results in an array where even elements are static strings
+  // between placeholders, and odd elements are the contents inside
+  // placeholders.
+  //
+  // For example, the pattern:
+  //   "{{ foo }}-{{bar}}{{baz}}."
+  // will result in:
+  //   ['', 'foo', '-', 'bar', '', 'baz', .']
+  const patternParts = pattern.split(/\{{\s*([\w\[\]'_.-]*)\s*}}/g);
+
+  const resolvers = new Array<(context: unknown) => string>();
+
+  for (let i = 0; i < patternParts.length; i += 2) {
+    const staticPart = patternParts[i];
+    const placeholderPart = patternParts[i + 1];
+
+    if (staticPart) {
+      resolvers.push(() => staticPart);
+    }
+
+    if (placeholderPart) {
+      resolvers.push(context => {
+        const value = lodash.get(context, placeholderPart);
+        if (typeof value === 'string' || Number.isFinite(value)) {
+          return String(value);
+        } else if (!value) {
+          throw new InputError(`No value for selector '${placeholderPart}'`);
+        } else {
+          throw new InputError(
+            `Expected string or number value for selector '${placeholderPart}', got ${typeof value}`,
+          );
+        }
+      });
+    }
+  }
+
+  return context => resolvers.map(resolver => resolver(context)).join('');
+}
@@ -0,0 +1,17 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+export { eventsModuleGooglePubsubConsumingEventPublisher } from './module';
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import {
+  coreServices,
+  createBackendModule,
+} from '@backstage/backend-plugin-api';
+import { eventsServiceRef } from '@backstage/plugin-events-node';
+import { GooglePubSubConsumingEventPublisher } from './GooglePubSubConsumingEventPublisher';
+
+/**
+ * Reads messages off of Google Pub/Sub subscriptions and forwards them into the
+ * Backstage events system.
+ *
+ * @public
+ */
+export const eventsModuleGooglePubsubConsumingEventPublisher =
+  createBackendModule({
+    pluginId: 'events',
+    moduleId: 'google-pubsub-consuming-event-publisher',
+    register(reg) {
+      reg.registerInit({
+        deps: {
+          config: coreServices.rootConfig,
+          logger: coreServices.logger,
+          rootLifecycle: coreServices.rootLifecycle,
+          events: eventsServiceRef,
+        },
+        async init({ config, logger, rootLifecycle, events }) {
+          GooglePubSubConsumingEventPublisher.create({
+            config,
+            logger,
+            rootLifecycle,
+            events,
+          });
+        },
+      });
+    },
+  });
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+import { Message } from '@google-cloud/pubsub';
+
+/**
+ * A configured subscription task.
+ */
+export interface SubscriptionTask {
+  id: string;
+  project: string;
+  subscription: string;
+  mapToTopic: (message: Message) => string | undefined;
+  mapToMetadata: (message: Message) => Record<string, string>;
+}
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2025 The Backstage Authors
+ *
+ * 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.
+ */
+
+/**
+ * The google-pubsub backend module for the events plugin.
+ *
+ * @packageDocumentation
+ */
+
+export { eventsModuleGooglePubsubConsumingEventPublisher as default } from './GooglePubSubConsumingEventPublisher';
@@ -6384,6 +6384,26 @@ __metadata:
  languageName: unknown
  linkType: soft

+"@backstage/plugin-events-backend-module-google-pubsub@workspace:^, @backstage/plugin-events-backend-module-google-pubsub@workspace:plugins/events-backend-module-google-pubsub":
+  version: 0.0.0-use.local
+  resolution: "@backstage/plugin-events-backend-module-google-pubsub@workspace:plugins/events-backend-module-google-pubsub"
+  dependencies:
+    "@backstage/backend-defaults": "workspace:^"
+    "@backstage/backend-plugin-api": "workspace:^"
+    "@backstage/backend-test-utils": "workspace:^"
+    "@backstage/cli": "workspace:^"
+    "@backstage/config": "workspace:^"
+    "@backstage/errors": "workspace:^"
+    "@backstage/plugin-events-backend": "workspace:^"
+    "@backstage/plugin-events-node": "workspace:^"
+    "@backstage/types": "workspace:^"
+    "@google-cloud/pubsub": "npm:^4.10.0"
+    "@opentelemetry/api": "npm:^1.9.0"
+    lodash: "npm:^4.17.21"
+    wait-for-expect: "npm:^3.0.2"
+  languageName: unknown
+  linkType: soft
+
 "@backstage/plugin-events-backend-test-utils@workspace:^, @backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils":
  version: 0.0.0-use.local
  resolution: "@backstage/plugin-events-backend-test-utils@workspace:plugins/events-backend-test-utils"
@@ -9965,6 +9985,13 @@ __metadata:
  languageName: node
  linkType: hard

+"@google-cloud/precise-date@npm:^4.0.0":
+  version: 4.0.0
+  resolution: "@google-cloud/precise-date@npm:4.0.0"
+  checksum: 10/7c897bcad6a40efd77df165fca8a57bd3fbb84e6ace848085b16233c1086c32c06258caa88b9c9aaf7d4029ded13d01b57d1b5fd808dc5722cf57626ac4a386b
+  languageName: node
+  linkType: hard
+
 "@google-cloud/projectify@npm:^4.0.0":
  version: 4.0.0
  resolution: "@google-cloud/projectify@npm:4.0.0"
@@ -9972,13 +9999,35 @@ __metadata:
  languageName: node
  linkType: hard

-"@google-cloud/promisify@npm:^4.0.0":
+"@google-cloud/promisify@npm:^4.0.0, @google-cloud/promisify@npm:~4.0.0":
  version: 4.0.0
  resolution: "@google-cloud/promisify@npm:4.0.0"
  checksum: 10/c5de81321b3a5c567edcbe0b941fb32644611147f3ba22f20575918c225a979988a99bc2ebda05ac914fa8714b0a54c69be72c3f46c7a64c3b19db7d7fba8d04
  languageName: node
  linkType: hard

+"@google-cloud/pubsub@npm:^4.10.0":
+  version: 4.11.0
+  resolution: "@google-cloud/pubsub@npm:4.11.0"
+  dependencies:
+    "@google-cloud/paginator": "npm:^5.0.0"
+    "@google-cloud/precise-date": "npm:^4.0.0"
+    "@google-cloud/projectify": "npm:^4.0.0"
+    "@google-cloud/promisify": "npm:~4.0.0"
+    "@opentelemetry/api": "npm:~1.9.0"
+    "@opentelemetry/semantic-conventions": "npm:~1.30.0"
+    arrify: "npm:^2.0.0"
+    extend: "npm:^3.0.2"
+    google-auth-library: "npm:^9.3.0"
+    google-gax: "npm:^4.3.3"
+    heap-js: "npm:^2.2.0"
+    is-stream-ended: "npm:^0.1.4"
+    lodash.snakecase: "npm:^4.1.1"
+    p-defer: "npm:^3.0.0"
+  checksum: 10/bfaae6b0ca25d9adf2ba160850ae3db0481c7a45cd4107ec7b5afd5002845789e1effecaaeb6fe03ddf302060e7f2da55e9ff0c3dc3a590adbe73335369fae75
+  languageName: node
+  linkType: hard
+
 "@google-cloud/storage@npm:^7.0.0":
  version: 7.14.0
  resolution: "@google-cloud/storage@npm:7.14.0"
@@ -13402,7 +13451,7 @@ __metadata:
  languageName: node
  linkType: hard

-"@opentelemetry/api@npm:^1.3.0, @opentelemetry/api@npm:^1.4.0, @opentelemetry/api@npm:^1.9.0":
+"@opentelemetry/api@npm:^1.3.0, @opentelemetry/api@npm:^1.4.0, @opentelemetry/api@npm:^1.9.0, @opentelemetry/api@npm:~1.9.0":
  version: 1.9.0
  resolution: "@opentelemetry/api@npm:1.9.0"
  checksum: 10/a607f0eef971893c4f2ee2a4c2069aade6ec3e84e2a1f5c2aac19f65c5d9eeea41aa72db917c1029faafdd71789a1a040bdc18f40d63690e22ccae5d7070f194
@@ -14689,7 +14738,7 @@ __metadata:
  languageName: node
  linkType: hard

-"@opentelemetry/semantic-conventions@npm:^1.27.0":
+"@opentelemetry/semantic-conventions@npm:^1.27.0, @opentelemetry/semantic-conventions@npm:~1.30.0":
  version: 1.30.0
  resolution: "@opentelemetry/semantic-conventions@npm:1.30.0"
  checksum: 10/78df5976f5bcfd00acaea3e609cf06fdd34517ae8db994ae216aaac16c51af97ac22c534bfcbac5218e0086db83ec5ef6cc045b95626cc6ea807686bea549a41
@@ -29150,6 +29199,7 @@ __metadata:
    "@backstage/plugin-catalog-backend-module-unprocessed": "workspace:^"
    "@backstage/plugin-devtools-backend": "workspace:^"
    "@backstage/plugin-events-backend": "workspace:^"
+    "@backstage/plugin-events-backend-module-google-pubsub": "workspace:^"
    "@backstage/plugin-kubernetes-backend": "workspace:^"
    "@backstage/plugin-notifications-backend": "workspace:^"
    "@backstage/plugin-permission-backend": "workspace:^"
@@ -31435,6 +31485,13 @@ __metadata:
  languageName: node
  linkType: hard

+"heap-js@npm:^2.2.0":
+  version: 2.6.0
+  resolution: "heap-js@npm:2.6.0"
+  checksum: 10/33c4e9a82b8698856cf807845866e5b3e94e2c4898937a99a160aea698c68fbe121f0bffc514af8d71d32eae36ca9516063c7e13d3624457318d5cf246a623a9
+  languageName: node
+  linkType: hard
+
 "helmet@npm:^6.0.0":
  version: 6.0.1
  resolution: "helmet@npm:6.0.1"
@@ -32943,6 +33000,13 @@ __metadata:
  languageName: node
  linkType: hard

+"is-stream-ended@npm:^0.1.4":
+  version: 0.1.4
+  resolution: "is-stream-ended@npm:0.1.4"
+  checksum: 10/56cbc9cfa0a77877777a3df9e186abb5b0ca73dcbcaf0fd87ed573fb8f8e61283abec0fc072c9e3412336edc04449439b8a128d2bcc6c2797158de5465cfaf85
+  languageName: node
+  linkType: hard
+
 "is-stream@npm:^1.1.0":
  version: 1.1.0
  resolution: "is-stream@npm:1.1.0"
@@ -35493,6 +35557,13 @@ __metadata:
  languageName: node
  linkType: hard

+"lodash.snakecase@npm:^4.1.1":
+  version: 4.1.1
+  resolution: "lodash.snakecase@npm:4.1.1"
+  checksum: 10/82ed40935d840477ef8fee64f9f263f75989c6cde36b84aae817246d95826228e1b5a7f6093c51de324084f86433634c7af244cb89496633cacfe443071450d0
+  languageName: node
+  linkType: hard
+
 "lodash.sortby@npm:^4.7.0":
  version: 4.7.0
  resolution: "lodash.sortby@npm:4.7.0"
@@ -38767,6 +38838,13 @@ __metadata:
  languageName: node
  linkType: hard

+"p-defer@npm:^3.0.0":
+  version: 3.0.0
+  resolution: "p-defer@npm:3.0.0"
+  checksum: 10/ac3b0976a1c76b67cca1a34e00f7299b0cc230891f820749686aa84f8947326bbe0f8e3b7d9ca511578ee06f0c1a6e0ff68c8e9c325eac455f09d99f91697161
+  languageName: node
+  linkType: hard
+
 "p-filter@npm:^2.1.0":
  version: 2.1.0
  resolution: "p-filter@npm:2.1.0"
				`@@ -0,0 +1 @@`
				`module.exports = require('@backstage/cli/config/eslint-factory')(__dirname);`