paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 26 Packages Projects Releases Wiki Activity

Merge pull request #24916 from backstage/rugvip/events

Browse Source 
events-backend: added scaleable events service implementation
This commit is contained in:
Patrik Oldsberg
2024-09-23 16:45:43 +02:00
committed by GitHub
parent efe0be085c 6d28caf285
commit 8caad3d5bf
39 changed files with 3547 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/slow-gorillas-thank.md
+9
View File
@@ -0,0 +1,9 @@
---
'@backstage/plugin-events-backend': patch
---
The events backend now has its own built-in event bus for distributing events across multiple backend instances. It exposes a new HTTP API under `/bus/v1/` for publishing and reading events from the bus, as well as its own storage and notification mechanism for events.
The backing event store for the bus only supports scaled deployment if PostgreSQL is used as the DBMS. If SQLite or MySQL is used, the event bus will fall back to an in-memory store that does not support multiple backend instances.
The default `EventsService` implementation from `@backstage/plugin-events-node` has also been updated to use the new events bus.
.changeset/stale-roses-serve.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-events-node': patch
---
The default implementation of the `EventsService` now uses the new event bus for distributing events across multiple backend instances if the events backend plugin is installed.
packages/backend/package.json
+1
View File
@@ -44,6 +44,7 @@
"@backstage/plugin-catalog-backend-module-scaffolder-entity-model": "workspace:^",
"@backstage/plugin-catalog-backend-module-unprocessed": "workspace:^",
"@backstage/plugin-devtools-backend": "workspace:^",
"@backstage/plugin-events-backend": "workspace:^",
"@backstage/plugin-kubernetes-backend": "workspace:^",
"@backstage/plugin-notifications-backend": "workspace:^",
"@backstage/plugin-permission-backend": "workspace:^",
packages/backend/src/index.ts
+1
View File
@@ -39,6 +39,7 @@ backend.add(
import('@backstage/plugin-catalog-backend-module-scaffolder-entity-model'),
);
backend.add(import('@backstage/plugin-catalog-backend/alpha'));
backend.add(import('@backstage/plugin-events-backend/alpha'));
backend.add(import('@backstage/plugin-devtools-backend'));
backend.add(import('@backstage/plugin-kubernetes-backend/alpha'));
backend.add(
plugins/events-backend/dev/index.ts
+151
View File
@@ -0,0 +1,151 @@
/*
* Copyright 2024 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 { WinstonLogger } from '@backstage/backend-defaults/rootLogger';
import { createBackend } from '@backstage/backend-defaults';
import {
coreServices,
createBackendPlugin,
createServiceFactory,
} from '@backstage/backend-plugin-api';
import { mockServices } from '@backstage/backend-test-utils';
import { eventsServiceRef } from '@backstage/plugin-events-node';
function makeBackend(port: number) {
const backend = createBackend();
backend.add(
mockServices.rootConfig.factory({
data: {
backend: {
baseUrl: `http://localhost:${port}`,
listen: { port },
database: {
client: 'pg',
connection: {
host: 'localhost',
port: 5432,
user: process.env.USER || 'postgres',
},
},
},
},
}),
);
backend.add(
createServiceFactory({
service: coreServices.rootLogger,
deps: {},
async factory() {
return WinstonLogger.create({
meta: {
service: 'backstage',
port,
},
// level: 'debug',
format: WinstonLogger.colorFormat(),
});
},
}),
);
backend.add(import('../src/alpha'));
return backend;
}
const backend7008 = makeBackend(7008);
backend7008.add(
createBackendPlugin({
pluginId: 'producer',
register(reg) {
reg.registerInit({
deps: {
events: eventsServiceRef,
logger: coreServices.logger,
rootLifecycle: coreServices.rootLifecycle,
},
async init({ events, logger, rootLifecycle }) {
rootLifecycle.addStartupHook(async () => {
const publish = () => {
logger.info(`Publishing event to topic 'test'`);
events
.publish({
eventPayload: { foo: 'bar' },
topic: 'test',
})
.catch(error => {
logger.error(`Failed to publish event from producer`, error);
});
};
publish();
setInterval(publish, 10000);
});
},
});
},
}),
);
backend7008.start();
const backend7009 = makeBackend(7009);
backend7009.add(
createBackendPlugin({
pluginId: 'consumer',
register(reg) {
reg.registerInit({
deps: {
events: eventsServiceRef,
logger: coreServices.logger,
},
async init({ events, logger }) {
await events.subscribe({
id: 'test',
topics: ['test'],
async onEvent(event) {
logger.info(`Received event: ${JSON.stringify(event)}`);
},
});
},
});
},
}),
);
backend7009.start();
const backend7010 = makeBackend(7010);
backend7010.add(
createBackendPlugin({
pluginId: 'consumer',
register(reg) {
reg.registerInit({
deps: {
events: eventsServiceRef,
logger: coreServices.logger,
},
async init({ events, logger }) {
await events.subscribe({
id: 'test',
topics: ['test'],
async onEvent(event) {
logger.info(`Received event: ${JSON.stringify(event)}`);
},
});
},
});
},
}),
);
backend7010.start();
plugins/events-backend/migrations/20240523100528_init.js
+98
View File
@@ -0,0 +1,98 @@
/*
* Copyright 2024 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.
*/
// @ts-check
/**
* @param { import("knex").Knex } knex
* @returns { Promise<void> }
*/
exports.up = async function up(knex) {
// The event bus only supports PostgresSQL
if (knex.client.config.client === 'pg') {
await knex.schema.createTable('event_bus_events', table => {
table.comment('Events published to the events bus');
table
.bigIncrements('id')
.primary()
.comment('The unique ID of this event');
table
.text('created_by')
.notNullable()
.comment('The principal that published the event');
table
.dateTime('created_at')
.defaultTo(knex.fn.now())
.notNullable()
.comment('The time that the event was created');
table.text('topic').notNullable().comment('The topic of the event');
table
.text('data_json')
.notNullable()
.comment('The payload data of this event');
table
.specificType('notified_subscribers', 'text ARRAY')
.comment(
'The IDs of the subscribers that have already consumed this event',
);
table.index('topic', 'event_bus_events_topic_idx');
});
await knex.schema.createTable('event_bus_subscriptions', table => {
table.comment('Subscriptions to the event bus');
table
.string('id')
.primary()
.notNullable()
.comment('The unique ID of this particular subscription');
table
.text('created_by')
.notNullable()
.comment('The principal that created the subscription');
table
.dateTime('created_at')
.defaultTo(knex.fn.now())
.notNullable()
.comment('The time that the subscription was created');
table
.dateTime('updated_at')
.defaultTo(knex.fn.now())
.notNullable()
.comment('The time that the subscription was last updated');
table
.bigInteger('read_until')
.notNullable()
.comment(
'The sequence counter until which the subscription has read events',
);
table
.specificType('topics', 'text ARRAY')
.comment('The topics that this subscriber is interested in');
});
}
};
/**
* @param { import("knex").Knex } knex
* @returns { Promise<void> }
*/
exports.down = async function down(knex) {
if (knex.client.config.client === 'pg') {
await knex.schema.dropTable('event_bus_subscriptions');
await knex.schema.dropTable('event_bus_events');
}
};
plugins/events-backend/package.json
+10 -1
View File
@@ -39,11 +39,13 @@
},
"files": [
"config.d.ts",
"dist"
"dist",
"migrations"
],
"scripts": {
"build": "backstage-cli package build",
"clean": "backstage-cli package clean",
"generate": "backstage-repo-tools package schema openapi generate --server --client-package plugins/events-node",
"lint": "backstage-cli package lint",
"prepack": "backstage-cli package prepack",
"postpack": "backstage-cli package postpack",
@@ -52,18 +54,25 @@
},
"dependencies": {
"@backstage/backend-common": "^0.25.0",
"@backstage/backend-openapi-utils": "workspace:^",
"@backstage/backend-plugin-api": "workspace:^",
"@backstage/config": "workspace:^",
"@backstage/errors": "workspace:^",
"@backstage/plugin-events-node": "workspace:^",
"@backstage/types": "workspace:^",
"@types/express": "^4.17.6",
"express": "^4.17.1",
"express-promise-router": "^4.1.0",
"knex": "^3.0.0",
"winston": "^3.2.1"
},
"devDependencies": {
"@backstage/backend-app-api": "workspace:^",
"@backstage/backend-defaults": "workspace:^",
"@backstage/backend-test-utils": "workspace:^",
"@backstage/cli": "workspace:^",
"@backstage/plugin-events-backend-test-utils": "workspace:^",
"@backstage/repo-tools": "workspace:^",
"supertest": "^7.0.0"
},
"configSchema": "config.d.ts"
plugins/events-backend/src/migrations.test.ts
+101
View File
@@ -0,0 +1,101 @@
/*
* 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.
*/
import { Knex } from 'knex';
import { TestDatabases } from '@backstage/backend-test-utils';
import fs from 'fs';
const migrationsDir = `${__dirname}/../migrations`;
const migrationsFiles = fs.readdirSync(migrationsDir).sort();
async function migrateUpOnce(knex: Knex): Promise<void> {
await knex.migrate.up({ directory: migrationsDir });
}
async function migrateDownOnce(knex: Knex): Promise<void> {
await knex.migrate.down({ directory: migrationsDir });
}
async function migrateUntilBefore(knex: Knex, target: string): Promise<void> {
const index = migrationsFiles.indexOf(target);
if (index === -1) {
throw new Error(`Migration ${target} not found`);
}
for (let i = 0; i < index; i++) {
await migrateUpOnce(knex);
}
}
jest.setTimeout(60_000);
describe('migrations', () => {
const databases = TestDatabases.create({
ids: ['POSTGRES_9', 'POSTGRES_13', 'POSTGRES_16'],
});
it.each(databases.eachSupportedId())(
'20240523100528_init.js, %p',
async databaseId => {
const knex = await databases.init(databaseId);
await migrateUntilBefore(knex, '20240523100528_init.js');
await migrateUpOnce(knex);
await knex('event_bus_events').insert({
topic: 'test',
created_by: 'abc',
data_json: JSON.stringify({ message: 'hello' }),
notified_subscribers: ['tester'],
});
await knex('event_bus_subscriptions').insert({
id: 'tester',
created_by: 'abc',
read_until: '5',
topics: ['test', 'test2'],
});
await expect(knex('event_bus_events')).resolves.toEqual([
{
id: '1',
created_by: 'abc',
topic: 'test',
data_json: JSON.stringify({ message: 'hello' }),
created_at: expect.anything(),
notified_subscribers: ['tester'],
},
]);
await expect(knex('event_bus_subscriptions')).resolves.toEqual([
{
id: 'tester',
created_by: 'abc',
created_at: expect.anything(),
updated_at: expect.anything(),
read_until: '5',
topics: ['test', 'test2'],
},
]);
await migrateDownOnce(knex);
// This looks odd - you might expect a .toThrow at the end but that
// actually is flaky for some reason specifically on sqlite when
// performing multiple runs in sequence
await expect(knex('event_bus_events')).rejects.toEqual(expect.anything());
await knex.destroy();
},
);
});
plugins/events-backend/src/schema/openapi.generated.ts
+291
View File
@@ -0,0 +1,291 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
import { createValidatedOpenApiRouter } from '@backstage/backend-openapi-utils';
export const spec = {
openapi: '3.0.3',
info: {
title: 'events',
version: '1',
description: 'The Backstage backend plugin that powers the events system.',
license: {
name: 'Apache-2.0',
url: 'http://www.apache.org/licenses/LICENSE-2.0.html',
},
contact: {},
},
servers: [
{
url: '/',
},
],
components: {
examples: {},
headers: {},
parameters: {
subscriptionId: {
name: 'subscriptionId',
in: 'path',
required: true,
allowReserved: true,
schema: {
type: 'string',
},
},
},
requestBodies: {},
responses: {
ErrorResponse: {
description: 'An error response from the backend.',
content: {
'application/json; charset=utf-8': {
schema: {
$ref: '#/components/schemas/Error',
},
},
},
},
},
schemas: {
Event: {
type: 'object',
required: ['topic', 'payload'],
properties: {
topic: {
type: 'string',
description: 'The topic that the event is published on',
},
payload: {
description: 'The event payload',
},
},
},
Error: {
type: 'object',
properties: {
error: {
type: 'object',
properties: {
name: {
type: 'string',
},
message: {
type: 'string',
},
},
required: ['name', 'message'],
},
request: {
type: 'object',
properties: {
method: {
type: 'string',
},
url: {
type: 'string',
},
},
required: ['method', 'url'],
},
response: {
type: 'object',
properties: {
statusCode: {
type: 'number',
},
},
required: ['statusCode'],
},
},
required: ['error', 'request', 'response'],
},
},
securitySchemes: {
JWT: {
type: 'http',
scheme: 'bearer',
bearerFormat: 'JWT',
},
},
},
paths: {
'/bus/v1/events': {
post: {
operationId: 'PostEvent',
description: 'Publish a new event',
responses: {
'201': {
description: 'The event was published successfully',
},
'204': {
description:
'The event did not need to be published as all subscribers have already been notified',
},
default: {
$ref: '#/components/responses/ErrorResponse',
},
},
security: [
{},
{
JWT: [],
},
],
requestBody: {
required: true,
content: {
'application/json': {
schema: {
type: 'object',
required: ['event'],
properties: {
event: {
$ref: '#/components/schemas/Event',
},
notifiedSubscribers: {
type: 'array',
description:
'The IDs of subscriptions that have already received this event',
items: {
type: 'string',
},
},
},
},
examples: {
'Publishing a simple Event': {
value: {
event: {
topic: 'test-topic',
payload: {
myData: 'foo',
},
},
},
},
},
},
},
},
},
},
'/bus/v1/subscriptions/{subscriptionId}': {
put: {
operationId: 'PutSubscription',
description:
'Ensures that the subscription exists with the provided configuration',
responses: {
'201': {
description: 'The subscription exists or was created successfully',
},
default: {
$ref: '#/components/responses/ErrorResponse',
},
},
security: [
{},
{
JWT: [],
},
],
parameters: [
{
$ref: '#/components/parameters/subscriptionId',
},
],
requestBody: {
required: true,
content: {
'application/json': {
schema: {
type: 'object',
required: ['topics'],
properties: {
topics: {
type: 'array',
description: 'The topics to subscribe to',
items: {
type: 'string',
},
},
},
},
examples: {
'Subscribing to a single topic': {
value: {
topics: ['test-topic'],
},
},
},
},
},
},
},
},
'/bus/v1/subscriptions/{subscriptionId}/events': {
get: {
operationId: 'GetSubscriptionEvents',
description: 'Get new events for the provided subscription',
responses: {
'200': {
description: 'New events',
content: {
'application/json': {
schema: {
type: 'object',
required: ['events'],
properties: {
events: {
type: 'array',
items: {
$ref: '#/components/schemas/Event',
},
},
},
},
},
},
},
'202': {
description:
'No new events are available. Response will block until the client should try again.',
},
default: {
$ref: '#/components/responses/ErrorResponse',
},
},
security: [
{},
{
JWT: [],
},
],
parameters: [
{
$ref: '#/components/parameters/subscriptionId',
},
],
},
},
},
} as const;
export const createOpenApiRouter = async (
options?: Parameters<typeof createValidatedOpenApiRouter>['1'],
) => createValidatedOpenApiRouter<typeof spec>(spec, options);
plugins/events-backend/src/schema/openapi.yaml
+182
View File
@@ -0,0 +1,182 @@
openapi: 3.0.3
info:
title: events
version: '1'
description: The Backstage backend plugin that powers the events system.
license:
name: Apache-2.0
url: http://www.apache.org/licenses/LICENSE-2.0.html
contact: {}
servers:
- url: /
components:
examples: {}
headers: {}
parameters:
subscriptionId:
name: subscriptionId
in: path
required: true
allowReserved: true
schema:
type: string
requestBodies: {}
responses:
ErrorResponse:
description: An error response from the backend.
content:
application/json; charset=utf-8:
schema:
$ref: '#/components/schemas/Error'
schemas:
Event:
type: object
required:
- topic
- payload
properties:
topic:
type: string
description: The topic that the event is published on
payload:
description: The event payload
Error:
type: object
properties:
error:
type: object
properties:
name:
type: string
message:
type: string
required:
- name
- message
request:
type: object
properties:
method:
type: string
url:
type: string
required:
- method
- url
response:
type: object
properties:
statusCode:
type: number
required:
- statusCode
required:
- error
- request
- response
securitySchemes:
JWT:
type: http
scheme: bearer
bearerFormat: JWT
paths:
/bus/v1/events:
post:
operationId: PostEvent
description: Publish a new event
responses:
'201':
description: The event was published successfully
'204':
description: The event did not need to be published as all subscribers have already been notified
default:
$ref: '#/components/responses/ErrorResponse'
security:
- {}
- JWT: []
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- event
properties:
event:
$ref: '#/components/schemas/Event'
notifiedSubscribers:
type: array
description: The IDs of subscriptions that have already received this event
items:
type: string
examples:
Publishing a simple Event:
value:
event:
topic: test-topic
payload:
myData: foo
/bus/v1/subscriptions/{subscriptionId}:
put:
operationId: PutSubscription
description: Ensures that the subscription exists with the provided configuration
responses:
'201':
description: The subscription exists or was created successfully
default:
$ref: '#/components/responses/ErrorResponse'
security:
- {}
- JWT: []
parameters:
- $ref: '#/components/parameters/subscriptionId'
requestBody:
required: true
content:
application/json:
schema:
type: object
required:
- topics
properties:
topics:
type: array
description: The topics to subscribe to
items:
type: string
examples:
Subscribing to a single topic:
value:
topics:
- test-topic
/bus/v1/subscriptions/{subscriptionId}/events:
get:
operationId: GetSubscriptionEvents
description: Get new events for the provided subscription
responses:
'200':
description: New events
content:
application/json:
schema:
type: object
required:
- events
properties:
events:
type: array
items:
$ref: '#/components/schemas/Event'
'202':
description: No new events are available. Response will block until the client should try again.
default:
$ref: '#/components/responses/ErrorResponse'
security:
- {}
- JWT: []
parameters:
- $ref: '#/components/parameters/subscriptionId'
plugins/events-backend/src/service/EventsPlugin.test.ts
+340 -1
View File
@@ -14,11 +14,20 @@
* limitations under the License.
*/
/* eslint-disable jest/expect-expect */
import {
createBackendModule,
createServiceFactory,
} from '@backstage/backend-plugin-api';
import { mockServices, startTestBackend } from '@backstage/backend-test-utils';
import {
TestBackend,
TestDatabaseId,
TestDatabases,
mockCredentials,
mockServices,
startTestBackend,
} from '@backstage/backend-test-utils';
import { eventsServiceRef } from '@backstage/plugin-events-node';
import { eventsExtensionPoint } from '@backstage/plugin-events-node/alpha';
import { TestEventsService } from '@backstage/plugin-events-backend-test-utils';
@@ -92,4 +101,334 @@ describe('eventsPlugin', () => {
test: 'fake-ext',
});
});
describe('event bus', () => {
class ReqHelper {
constructor(private readonly backend: TestBackend) {}
subscribe(id: string, topics: string[], options?: { auth?: string }) {
return request(this.backend.server)
.put(`/api/events/bus/v1/subscriptions/${id}`)
.set(
'authorization',
options?.auth ?? mockCredentials.service.header(),
)
.send({ topics });
}
publish(
topic: string,
payload: unknown,
options?: { notifiedSubscribers?: string[] },
) {
return request(this.backend.server)
.post('/api/events/bus/v1/events')
.set('authorization', mockCredentials.service.header())
.send({
event: { topic, payload },
notifiedSubscribers: options?.notifiedSubscribers,
});
}
readEvents(id: string) {
return request(this.backend.server)
.get(`/api/events/bus/v1/subscriptions/${id}/events`)
.set('authorization', mockCredentials.service.header())
.send();
}
}
const databases = TestDatabases.create({
ids: ['SQLITE_3', 'MYSQL_8', 'POSTGRES_9', 'POSTGRES_13', 'POSTGRES_16'],
});
async function mockKnexFactory(databaseId: TestDatabaseId) {
const knex = await databases.init(databaseId);
return mockServices.database.mock({
getClient: async () => knex,
}).factory;
}
let backend: TestBackend | undefined = undefined;
afterEach(async () => {
if (backend) {
await backend.stop();
backend = undefined;
}
});
it.each(databases.eachSupportedId())(
'should be possible to publish events as a service, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
await helper
.publish('test', { n: 1 })
.set('authorization', mockCredentials.none.header())
.expect(401);
await helper
.publish('test', { n: 1 })
.set('authorization', mockCredentials.user.header())
.expect(403);
await helper
.publish('test', { n: 1 })
.set('authorization', mockCredentials.service.header())
.expect(204); // 204, since there are no subscribers
},
);
it.each(databases.eachSupportedId())(
'should be possible to subscribe as a service and receive an event, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
await helper
.subscribe('tester', ['test'])
.set('authorization', mockCredentials.none.header())
.expect(401);
await helper
.subscribe('tester', ['test'])
.set('authorization', mockCredentials.user.header())
.expect(403);
await helper.subscribe('tester', ['test']).expect(201);
await helper
.readEvents('tester')
.set('authorization', mockCredentials.none.header())
.expect(401);
await helper
.readEvents('tester')
.set('authorization', mockCredentials.user.header())
.expect(403);
await helper.publish('test', { n: 1 }).expect(201); // 201, since there is a subscriber
await helper.readEvents('tester').expect(200, {
events: [{ topic: 'test', payload: { n: 1 } }],
});
},
);
it.each(databases.eachSupportedId())(
'should only send an event for each subscriber once, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
// 2 subscribers
await helper.subscribe('tester-1', ['test']).expect(201);
await helper.subscribe('tester-2', ['test']).expect(201);
// A single event
await helper.publish('test', { n: 1 }).expect(201);
// Single client for subscriber 1 gets the event
await helper.readEvents('tester-1').expect(200, {
events: [{ topic: 'test', payload: { n: 1 } }],
});
// Two clients for subscriber 2, only one gets the event
const res1 = helper.readEvents('tester-2');
const res2 = helper.readEvents('tester-2');
const res = await Promise.race([res1, res2]);
expect(res.status).toBe(200);
expect(res.body).toEqual({
events: [{ topic: 'test', payload: { n: 1 } }],
});
// Post another event, which triggers the other client to return
await helper.publish('test', { n: 2 }).expect(201);
const otherRes = await Promise.all([res1, res2]).then(rs =>
rs.find(r => r !== res),
);
expect(otherRes?.status).toBe(202);
// Reading subscriber 2 should now return the second event only
await helper.readEvents('tester-2').expect(200, {
events: [{ topic: 'test', payload: { n: 2 } }],
});
},
);
it.each(databases.eachSupportedId())(
'should not notify subscribers that have already consumed the event, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
// 2 subscribers
await helper.subscribe('tester-1', ['test']).expect(201);
await helper.subscribe('tester-2', ['test']).expect(201);
// A single event for each subscriber, that should not be sent to the other one
await helper
.publish(
'test',
{ for: 'tester-2' },
{
notifiedSubscribers: ['tester-1'],
},
)
.expect(201);
await helper
.publish(
'test',
{ for: 'tester-1' },
{
notifiedSubscribers: ['tester-2'],
},
)
.expect(201);
// Single client for subscriber 1 gets the event
await helper.readEvents('tester-1').expect(200, {
events: [{ topic: 'test', payload: { for: 'tester-1' } }],
});
// Single client for subscriber 2 gets the event
await helper.readEvents('tester-2').expect(200, {
events: [{ topic: 'test', payload: { for: 'tester-2' } }],
});
},
);
it.each(databases.eachSupportedId())(
'should return multiple events in order, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
// 2 subscribers
await helper.subscribe('tester', ['test']).expect(201);
// A sequence of events published one at a time
for (let n = 0; n < 15; ++n) {
await helper.publish('test', { n }).expect(201);
}
// Batch size it 10
await helper.readEvents('tester').expect(200, {
events: [
{ topic: 'test', payload: { n: 0 } },
{ topic: 'test', payload: { n: 1 } },
{ topic: 'test', payload: { n: 2 } },
{ topic: 'test', payload: { n: 3 } },
{ topic: 'test', payload: { n: 4 } },
{ topic: 'test', payload: { n: 5 } },
{ topic: 'test', payload: { n: 6 } },
{ topic: 'test', payload: { n: 7 } },
{ topic: 'test', payload: { n: 8 } },
{ topic: 'test', payload: { n: 9 } },
],
});
await helper.readEvents('tester').expect(200, {
events: [
{ topic: 'test', payload: { n: 10 } },
{ topic: 'test', payload: { n: 11 } },
{ topic: 'test', payload: { n: 12 } },
{ topic: 'test', payload: { n: 13 } },
{ topic: 'test', payload: { n: 14 } },
],
});
},
);
it.each(databases.eachSupportedId())(
'should skip publishing if all subscribers have already consumed the event, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
await helper.subscribe('tester', ['test']).expect(201);
await helper
.publish(
'test',
{ for: 'tester-2' },
{ notifiedSubscribers: ['tester'] },
)
.expect(204);
},
);
it.each(databases.eachSupportedId())(
'should time out when no events are available, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
await helper.subscribe('tester', ['test']).expect(201);
jest.useFakeTimers({
advanceTimers: true,
});
try {
// Can't use supertest for this one because it can't handle the partially blocking response
const res = await fetch(
`http://localhost:${backend.server.port()}/api/events/bus/v1/subscriptions/tester/events`,
{
headers: {
authorization: mockCredentials.service.header(),
},
},
);
expect(res.status).toBe(202);
const reader = res.body!.getReader();
const isClosed = () =>
Promise.race([
reader
.read()
.then(() => reader.closed)
.then(() => true),
new Promise<boolean>(r => setTimeout(r, 100, false)),
]);
await expect(isClosed()).resolves.toBe(false);
await jest.advanceTimersByTimeAsync(30000);
await expect(isClosed()).resolves.toBe(false);
await jest.advanceTimersByTimeAsync(30000);
await expect(isClosed()).resolves.toBe(true);
} finally {
jest.useRealTimers();
}
},
);
it.each(databases.eachSupportedId())(
'should refuse listen without a subscription, %p',
async databaseId => {
backend = await startTestBackend({
features: [eventsPlugin, await mockKnexFactory(databaseId)],
});
const helper = new ReqHelper(backend);
await helper.readEvents('nonexistent').expect(404);
},
);
});
});
plugins/events-backend/src/service/EventsPlugin.ts
+26 -1
View File
@@ -28,6 +28,7 @@ import {
} from '@backstage/plugin-events-node';
import Router from 'express-promise-router';
import { HttpPostIngressEventPublisher } from './http';
import { createEventBusRouter } from './hub';
class EventsExtensionPointImpl implements EventsExtensionPoint {
#httpPostIngresses: HttpPostIngressOptions[] = [];
@@ -74,10 +75,23 @@ export const eventsPlugin = createBackendPlugin({
deps: {
config: coreServices.rootConfig,
events: eventsServiceRef,
database: coreServices.database,
logger: coreServices.logger,
scheduler: coreServices.scheduler,
lifecycle: coreServices.lifecycle,
httpAuth: coreServices.httpAuth,
router: coreServices.httpRouter,
},
async init({ config, events, logger, router }) {
async init({
config,
events,
database,
logger,
scheduler,
lifecycle,
httpAuth,
router,
}) {
const ingresses = Object.fromEntries(
extensionPoint.httpPostIngresses.map(ingress => [
ingress.topic,
@@ -93,6 +107,17 @@ export const eventsPlugin = createBackendPlugin({
});
const eventsRouter = Router();
http.bind(eventsRouter);
router.use(
await createEventBusRouter({
database,
logger,
httpAuth,
scheduler,
lifecycle,
}),
);
router.use(eventsRouter);
router.addAuthPolicy({
allow: 'unauthenticated',
plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts
+245
View File
@@ -0,0 +1,245 @@
/*
* Copyright 2024 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 {
TestDatabases,
mockCredentials,
mockServices,
} from '@backstage/backend-test-utils';
import { DatabaseEventBusStore } from './DatabaseEventBusStore';
const logger = mockServices.logger.mock();
const databases = TestDatabases.create({
ids: ['POSTGRES_9', 'POSTGRES_13', 'POSTGRES_16'],
});
describe('DatabaseEventBusStore', () => {
it.each(databases.eachSupportedId())(
'should clean up old events, %p',
async databaseId => {
const db = await databases.init(databaseId);
const store = await DatabaseEventBusStore.forTest({ logger, db });
await store.upsertSubscription(
'tester-1',
['test'],
mockCredentials.service(),
);
await store.upsertSubscription(
'tester-2',
['test'],
mockCredentials.service(),
);
for (let i = 0; i < 10; ++i) {
await store.publish({
event: { topic: 'test', eventPayload: { n: i } },
credentials: mockCredentials.service(),
});
}
const { events: events1 } = await store.readSubscription('tester-1');
expect(events1.length).toBe(10);
await store.clean();
await expect(store.readSubscription('tester-2')).rejects.toThrow(
"Subscription with ID 'tester-2' not found",
);
await store.upsertSubscription(
'tester-3',
['test'],
mockCredentials.service(),
);
// Reset read pointer to read form the beginning
await db('event_bus_subscriptions').select({ id: 'tester-3' }).update({
read_until: 0,
});
const { events: events3 } = await store.readSubscription('tester-3');
expect(events3.length).toBe(5);
},
);
it.each(databases.eachSupportedId())(
'should always clean up events outside the max age window, %p',
async databaseId => {
const db = await databases.init(databaseId);
const store = await DatabaseEventBusStore.forTest({
logger,
db,
maxAge: 0,
});
await store.upsertSubscription(
'tester-1',
['test'],
mockCredentials.service(),
);
await store.upsertSubscription(
'tester-2',
['test'],
mockCredentials.service(),
);
for (let i = 0; i < 10; ++i) {
await store.publish({
event: { topic: 'test', eventPayload: { n: i } },
credentials: mockCredentials.service(),
});
}
const { events: events1 } = await store.readSubscription('tester-1');
expect(events1.length).toBe(10);
await store.clean();
await expect(store.readSubscription('tester-2')).rejects.toThrow(
"Subscription with ID 'tester-2' not found",
);
await store.upsertSubscription(
'tester-3',
['test'],
mockCredentials.service(),
);
// Reset read pointer to read form the beginning
await db('event_bus_subscriptions').select({ id: 'tester-3' }).update({
read_until: 0,
});
const { events: events3 } = await store.readSubscription('tester-3');
expect(events3.length).toBe(0);
},
);
it.each(databases.eachSupportedId())(
'should not clean up events within the min age window, %p',
async databaseId => {
const db = await databases.init(databaseId);
const store = await DatabaseEventBusStore.forTest({
logger,
db,
minAge: 1000,
});
await store.upsertSubscription(
'tester-1',
['test'],
mockCredentials.service(),
);
for (let i = 0; i < 10; ++i) {
await store.publish({
event: { topic: 'test', eventPayload: { n: i } },
credentials: mockCredentials.service(),
});
}
await store.clean();
const { events: events1 } = await store.readSubscription('tester-1');
expect(events1.length).toBe(10);
},
);
it.each(databases.eachSupportedId())(
'should clean up a large number of events, %p',
async databaseId => {
const db = await databases.init(databaseId);
const store = await DatabaseEventBusStore.forTest({
logger,
db,
});
const COUNT = '100000';
await db.raw(`
INSERT INTO event_bus_events (id, created_by, topic, data_json)
SELECT id, 'abc', 'test', '{}'
FROM generate_series(1, ${COUNT}) AS id
`);
await expect(db('event_bus_events').count()).resolves.toEqual([
{ count: COUNT },
]);
const start = Date.now();
await store.clean();
// Local testing shows this takes about 80ms, but if this is flaky we can
// reduce the count down to 10_000.
expect(Date.now() - start).toBeLessThan(500);
await expect(db('event_bus_events').count()).resolves.toEqual([
{ count: '5' },
]);
},
);
it.each(databases.eachSupportedId())(
'should perform well when looking up events by topic, %p',
async databaseId => {
const db = await databases.init(databaseId);
const store = await DatabaseEventBusStore.forTest({
logger,
db,
});
const COUNT = '100000';
// Insert 100,000 events, a lot more than we'd expect to ever have
// in a real-world scenario given our count window size is 10,000.
await db.raw(`
INSERT INTO event_bus_events (id, created_by, topic, data_json, notified_subscribers)
SELECT id, 'abc', CONCAT('test-', MOD(id, 10)), CONCAT('{"payload":{"id":"', id, '"}}'), '{"${String(
Math.random(),
).slice(2, 6)}"}'
FROM generate_series(1, ${COUNT}) AS id
`);
await db('event_bus_subscriptions').insert({
id: 'tester',
created_by: 'abc',
read_until: 0,
topics: ['test-5'],
});
const start = Date.now();
const { events } = await store.readSubscription('tester');
const duration = Date.now() - start;
expect(events).toEqual([
{ topic: 'test-5', eventPayload: { id: '5' } },
{ topic: 'test-5', eventPayload: { id: '15' } },
{ topic: 'test-5', eventPayload: { id: '25' } },
{ topic: 'test-5', eventPayload: { id: '35' } },
{ topic: 'test-5', eventPayload: { id: '45' } },
{ topic: 'test-5', eventPayload: { id: '55' } },
{ topic: 'test-5', eventPayload: { id: '65' } },
{ topic: 'test-5', eventPayload: { id: '75' } },
{ topic: 'test-5', eventPayload: { id: '85' } },
{ topic: 'test-5', eventPayload: { id: '95' } },
]);
expect(duration).toBeLessThan(20);
},
);
});
plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
+622
View File
@@ -0,0 +1,622 @@
/*
* Copyright 2024 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 { EventParams } from '@backstage/plugin-events-node';
import { EventBusStore } from './types';
import { Knex } from 'knex';
import {
BackstageCredentials,
BackstageServicePrincipal,
DatabaseService,
LifecycleService,
LoggerService,
SchedulerService,
resolvePackagePath,
} from '@backstage/backend-plugin-api';
import { ForwardedError, NotFoundError } from '@backstage/errors';
import { HumanDuration, durationToMilliseconds } from '@backstage/types';
const WINDOW_MAX_COUNT_DEFAULT = 10_000;
const WINDOW_MIN_AGE_DEFAULT = { minutes: 10 };
const WINDOW_MAX_AGE_DEFAULT = { days: 1 };
const MAX_BATCH_SIZE = 10;
const LISTENER_CONNECTION_TIMEOUT_MS = 60_000;
const KEEPALIVE_INTERVAL_MS = 60_000;
const TABLE_EVENTS = 'event_bus_events';
const TABLE_SUBSCRIPTIONS = 'event_bus_subscriptions';
const TOPIC_PUBLISH = 'event_bus_publish';
type EventsRow = {
id: string;
created_by: string;
created_at: Date;
topic: string;
data_json: string;
notified_subscribers: string[];
};
type SubscriptionsRow = {
id: string;
created_by: string;
created_at: Date;
updated_at: Date;
read_until: string;
topics: string[];
};
function creatorId(
credentials: BackstageCredentials<BackstageServicePrincipal>,
) {
return `service=${credentials.principal.subject}`;
}
const migrationsDir = resolvePackagePath(
'@backstage/plugin-events-backend',
'migrations',
);
interface InternalDbClient {
acquireRawConnection(): Promise<InternalDbConnection>;
destroyRawConnection(conn: InternalDbConnection): Promise<void>;
}
interface InternalDbConnection {
query(sql: string): Promise<void>;
end(): Promise<void>;
on(
event: 'notification',
listener: (event: { channel: string; payload: string }) => void,
): void;
on(event: 'error', listener: (error: Error) => void): void;
on(event: 'end', listener: (error?: Error) => void): void;
removeAllListeners(): void;
}
// This internal class manages a single connection to the database that all listeners share
class DatabaseEventBusListener {
readonly #client: InternalDbClient;
readonly #logger: LoggerService;
readonly #listeners = new Set<{
topics: Set<string>;
resolve: (result: { topic: string }) => void;
reject: (error: Error) => void;
}>();
#isShuttingDown = false;
#connPromise?: Promise<InternalDbConnection>;
#connTimeout?: NodeJS.Timeout;
#keepaliveInterval?: NodeJS.Timeout;
constructor(client: InternalDbClient, logger: LoggerService) {
this.#client = client;
this.#logger = logger.child({ type: 'DatabaseEventBusListener' });
}
async setupListener(
topics: Set<string>,
signal: AbortSignal,
): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
if (this.#connTimeout) {
clearTimeout(this.#connTimeout);
this.#connTimeout = undefined;
}
await this.#ensureConnection();
const updatePromise = new Promise<{ topic: string }>((resolve, reject) => {
const listener = {
topics,
resolve(result: { topic: string }) {
resolve(result);
cleanup();
},
reject(err: Error) {
reject(err);
cleanup();
},
};
this.#listeners.add(listener);
const onAbort = () => {
this.#listeners.delete(listener);
this.#maybeTimeoutConnection();
reject(signal.reason);
cleanup();
};
function cleanup() {
signal.removeEventListener('abort', onAbort);
}
signal.addEventListener('abort', onAbort);
});
// Ignore unhandled rejections
updatePromise.catch(() => {});
return { waitForUpdate: () => updatePromise };
}
async shutdown() {
if (this.#isShuttingDown) {
return;
}
this.#isShuttingDown = true;
const conn = await this.#connPromise?.catch(() => undefined);
if (conn) {
this.#destroyConnection(conn);
}
}
#handleNotify(topic: string) {
this.#logger.debug(`Listener received notification for topic '${topic}'`);
for (const l of this.#listeners) {
if (l.topics.has(topic)) {
l.resolve({ topic });
this.#listeners.delete(l);
}
}
this.#maybeTimeoutConnection();
}
// We don't try to reconnect on error, instead we notify all listeners and let
// them try to establish a new connection
#handleError(error: Error) {
this.#logger.error(
`Listener connection failed, notifying all listeners`,
error,
);
for (const l of this.#listeners) {
l.reject(new Error('Listener connection failed'));
}
this.#listeners.clear();
this.#maybeTimeoutConnection();
}
#maybeTimeoutConnection() {
// If we don't have any listeners, destroy the connection after a timeout
if (this.#listeners.size === 0 && !this.#connTimeout) {
this.#connTimeout = setTimeout(() => {
this.#connTimeout = undefined;
this.#connPromise?.then(conn => {
this.#logger.info('Listener connection timed out, destroying');
this.#connPromise = undefined;
this.#destroyConnection(conn);
});
}, LISTENER_CONNECTION_TIMEOUT_MS);
}
}
#destroyConnection(conn: InternalDbConnection) {
if (this.#keepaliveInterval) {
clearInterval(this.#keepaliveInterval);
this.#keepaliveInterval = undefined;
}
this.#client.destroyRawConnection(conn).catch(error => {
this.#logger.error(`Listener failed to destroy connection`, error);
});
conn.removeAllListeners();
}
async #ensureConnection() {
if (this.#isShuttingDown) {
throw new Error('Listener is shutting down');
}
if (this.#connPromise) {
await this.#connPromise;
return;
}
this.#connPromise = Promise.resolve().then(async () => {
const conn = await this.#client.acquireRawConnection();
try {
await conn.query(`LISTEN ${TOPIC_PUBLISH}`);
// Set up a keepalive interval to make sure the connection stays alive
if (this.#keepaliveInterval) {
clearInterval(this.#keepaliveInterval);
}
this.#keepaliveInterval = setInterval(() => {
conn.query('select 1').catch(error => {
this.#connPromise = undefined;
this.#destroyConnection(conn);
this.#handleError(new ForwardedError('Keepalive failed', error));
});
}, KEEPALIVE_INTERVAL_MS);
conn.on('notification', event => {
this.#handleNotify(event.payload);
});
conn.on('error', error => {
this.#connPromise = undefined;
this.#destroyConnection(conn);
this.#handleError(error);
});
conn.on('end', error => {
this.#connPromise = undefined;
this.#destroyConnection(conn);
this.#handleError(
error ?? new Error('Connection ended unexpectedly'),
);
});
return conn;
} catch (error) {
this.#destroyConnection(conn);
throw error;
}
});
try {
await this.#connPromise;
} catch (error) {
this.#connPromise = undefined;
throw error;
}
}
}
export class DatabaseEventBusStore implements EventBusStore {
static async create(options: {
database: DatabaseService;
logger: LoggerService;
scheduler: SchedulerService;
lifecycle: LifecycleService;
window?: {
/** Events within this range will never be deleted */
minAge?: HumanDuration;
/** Events outside of this age will always be deleted */
maxAge?: HumanDuration;
/** Events outside of this count will be deleted if they are outside the minAge window */
maxCount?: number;
};
}): Promise<DatabaseEventBusStore> {
const db = await options.database.getClient();
if (db.client.config.client !== 'pg') {
throw new Error(
`DatabaseEventBusStore only supports PostgreSQL, got '${db.client.config.client}'`,
);
}
if (!options.database.migrations?.skip) {
await db.migrate.latest({
directory: migrationsDir,
});
}
const listener = new DatabaseEventBusListener(db.client, options.logger);
const store = new DatabaseEventBusStore(
db,
options.logger,
listener,
options.window?.maxCount ?? WINDOW_MAX_COUNT_DEFAULT,
durationToMilliseconds(options.window?.minAge ?? WINDOW_MIN_AGE_DEFAULT),
durationToMilliseconds(options.window?.maxAge ?? WINDOW_MAX_AGE_DEFAULT),
);
await options.scheduler.scheduleTask({
id: 'event-bus-cleanup',
frequency: { seconds: 10 },
timeout: { minutes: 1 },
initialDelay: { seconds: 10 },
fn: () => store.#cleanup(),
});
options.lifecycle.addShutdownHook(async () => {
await listener.shutdown();
});
return store;
}
/** @internal */
static async forTest({
db,
logger,
minAge = 0,
maxAge = 10_000,
}: {
db: Knex;
logger: LoggerService;
minAge?: number;
maxAge?: number;
}) {
await db.migrate.latest({ directory: migrationsDir });
const store = new DatabaseEventBusStore(
db,
logger,
new DatabaseEventBusListener(db.client, logger),
5,
minAge,
maxAge,
);
return Object.assign(store, { clean: () => store.#cleanup() });
}
readonly #db: Knex;
readonly #logger: LoggerService;
readonly #listener: DatabaseEventBusListener;
readonly #windowMaxCount: number;
readonly #windowMinAge: number;
readonly #windowMaxAge: number;
private constructor(
db: Knex,
logger: LoggerService,
listener: DatabaseEventBusListener,
windowMaxCount: number,
windowMinAge: number,
windowMaxAge: number,
) {
this.#db = db;
this.#logger = logger;
this.#listener = listener;
this.#windowMaxCount = windowMaxCount;
this.#windowMinAge = windowMinAge;
this.#windowMaxAge = windowMaxAge;
}
async publish(options: {
event: EventParams;
notifiedSubscribers?: string[];
credentials: BackstageCredentials<BackstageServicePrincipal>;
}): Promise<{ eventId: string } | undefined> {
const topic = options.event.topic;
const notifiedSubscribers = options.notifiedSubscribers ?? [];
// This query inserts a new event into the database, but only if there are
// subscribers to the topic that have not already been notified
const result = await this.#db
// There's no clean way to create a INSERT INTO .. SELECT with knex, so we end up with quite a lot of .raw(...)
.into(
this.#db.raw('?? (??, ??, ??, ??)', [
TABLE_EVENTS,
// These are the rows that we insert, and should match the SELECT below
'created_by',
'topic',
'data_json',
'notified_subscribers',
]),
)
.insert<EventsRow>(
(q: Knex.QueryBuilder) =>
q
// We're not reading data to insert from anywhere else, just raw data
.select(
this.#db.raw('?', [creatorId(options.credentials)]),
this.#db.raw('?', [topic]),
this.#db.raw('?', [
JSON.stringify({
payload: options.event.eventPayload,
metadata: options.event.metadata,
}),
]),
this.#db.raw('?', [notifiedSubscribers]),
)
// The rest of this query is to check whether there are any
// subscribers that have not been notified yet
.from(TABLE_SUBSCRIPTIONS)
.whereNotIn('id', notifiedSubscribers) // Skip notified subscribers
.andWhere(this.#db.raw('? = ANY(topics)', [topic])) // Match topic
.having(this.#db.raw('count(*)'), '>', 0), // Check if there are any results
)
.returning<{ id: string }[]>('id');
if (result.length === 0) {
return undefined;
}
if (result.length > 1) {
throw new Error(
`Failed to insert event, unexpectedly updated ${result.length} rows`,
);
}
const [{ id }] = result;
// Notify other event bus instances that an event is available on the topic
const notifyResult = await this.#db.select(
this.#db.raw(`pg_notify(?, ?)`, [TOPIC_PUBLISH, topic]),
);
if (notifyResult?.length !== 1) {
this.#logger.warn(
`Failed to notify subscribers of event with ID '${id}' on topic '${topic}'`,
);
}
return { eventId: id };
}
async upsertSubscription(
id: string,
topics: string[],
credentials: BackstageCredentials<BackstageServicePrincipal>,
): Promise<void> {
const [{ max: maxId }] = await this.#db(TABLE_EVENTS).max('id');
const result = await this.#db<SubscriptionsRow>(TABLE_SUBSCRIPTIONS)
.insert({
id,
created_by: creatorId(credentials),
updated_at: this.#db.fn.now(),
topics,
read_until: maxId || 0,
})
.onConflict('id')
.merge(['created_by', 'topics', 'updated_at'])
.returning('*');
if (result.length !== 1) {
throw new Error(
`Failed to upsert subscription, updated ${result.length} rows`,
);
}
}
async readSubscription(id: string): Promise<{ events: EventParams[] }> {
// The below query selects the subscription we're reading from, locks it for
// an update, reads events for the subscription up to the limit, and then
// updates the pointer to the last read event.
//
// This is written as a plain SQL query to spare us all the horrors of
// expressing this in knex.
const { rows: result } = await this.#db.raw<{
rows: [] | [{ events: EventsRow[] }];
}>(
`
WITH subscription AS (
SELECT topics, read_until
FROM event_bus_subscriptions
WHERE id = :id
FOR UPDATE
),
selected_events AS (
SELECT event_bus_events.*
FROM event_bus_events
INNER JOIN subscription
ON event_bus_events.topic = ANY(subscription.topics)
WHERE event_bus_events.id > subscription.read_until
AND NOT :id = ANY(event_bus_events.notified_subscribers)
ORDER BY event_bus_events.id ASC LIMIT :limit
),
last_event_id AS (
SELECT max(id) AS last_event_id
FROM selected_events
),
events_array AS (
SELECT json_agg(row_to_json(selected_events)) AS events
FROM selected_events
)
UPDATE event_bus_subscriptions
SET read_until = COALESCE(last_event_id, (SELECT MAX(id) FROM event_bus_events), 0)
FROM events_array, last_event_id
WHERE event_bus_subscriptions.id = :id
RETURNING events_array.events
`,
{ id, limit: MAX_BATCH_SIZE },
);
if (result.length === 0) {
throw new NotFoundError(`Subscription with ID '${id}' not found`);
} else if (result.length > 1) {
throw new Error(
`Failed to read subscription, unexpectedly updated ${result.length} rows`,
);
}
const rows = result[0].events;
if (!rows || rows.length === 0) {
return { events: [] };
}
return {
events: rows.map(row => {
const { payload, metadata } = JSON.parse(row.data_json);
return {
topic: row.topic,
eventPayload: payload,
metadata,
};
}),
};
}
async setupListener(
subscriptionId: string,
options: {
signal: AbortSignal;
},
): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
const result = await this.#db<SubscriptionsRow>(TABLE_SUBSCRIPTIONS)
.select('topics')
.where({ id: subscriptionId })
.first();
if (!result) {
throw new NotFoundError(
`Subscription with ID '${subscriptionId}' not found`,
);
}
options.signal.throwIfAborted();
return this.#listener.setupListener(
new Set(result.topics ?? []),
options.signal,
);
}
async #cleanup() {
try {
const eventCount = await this.#db(TABLE_EVENTS)
.delete()
// Delete any events that are outside both the min age and size window
.orWhere(inner =>
inner
.whereIn(
'id',
this.#db
.select('id')
.from(TABLE_EVENTS)
.orderBy('id', 'desc')
.offset(this.#windowMaxCount),
)
.andWhere(
'created_at',
'<',
new Date(Date.now() - this.#windowMinAge),
),
)
// If events are outside the max age they will always be deleted
.orWhere('created_at', '<', new Date(Date.now() - this.#windowMaxAge));
if (eventCount > 0) {
this.#logger.info(
`Event cleanup resulted in ${eventCount} old events being deleted`,
);
}
} catch (error) {
this.#logger.error('Event cleanup failed', error);
}
try {
// Delete any subscribers that aren't keeping up with current events
const [{ min: minId }] = await this.#db(TABLE_EVENTS).min('id');
let subscriberCount;
if (minId === null) {
// No events left, remove all subscribers. This can happen if no events
// are published within the max age window.
subscriberCount = await this.#db(TABLE_SUBSCRIPTIONS).delete();
} else {
subscriberCount = await this.#db(TABLE_SUBSCRIPTIONS)
.delete()
// Read pointer points to the ID that has been read, so we need an additional offset
.where('read_until', '<', minId - 1);
}
if (subscriberCount > 0) {
this.#logger.info(
`Subscription cleanup resulted in ${subscriberCount} stale subscribers being deleted`,
);
}
} catch (error) {
this.#logger.error('Subscription cleanup failed', error);
}
}
}
plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts
+91
View File
@@ -0,0 +1,91 @@
/*
* Copyright 2024 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 { EventParams } from '@backstage/plugin-events-node';
import { MemoryEventBusStore } from './MemoryEventBusStore';
import { mockCredentials } from '@backstage/backend-test-utils';
function mkEvent(message: string): EventParams {
return {
topic: 'test',
eventPayload: { message },
};
}
describe('MemoryEventBusStore', () => {
it('should publish to subscribers', async () => {
const store = new MemoryEventBusStore();
await expect(
store.publish({
event: mkEvent('hello'),
notifiedSubscribers: [],
credentials: mockCredentials.service(),
}),
).resolves.toEqual(undefined);
await expect(store.readSubscription('test')).rejects.toThrow(
'Subscription not found',
);
await store.upsertSubscription('tester', ['test']);
await expect(
store.publish({
event: mkEvent('hello'),
notifiedSubscribers: [],
credentials: mockCredentials.service(),
}),
).resolves.toEqual({ eventId: '1' });
await expect(
store.publish({
event: mkEvent('ignored'),
notifiedSubscribers: ['tester'],
credentials: mockCredentials.service(),
}),
).resolves.toEqual(undefined);
await expect(store.readSubscription('tester')).resolves.toEqual({
events: [mkEvent('hello')],
});
});
it('should clean up old events', async () => {
const store = new MemoryEventBusStore({ maxEvents: 5 });
await store.upsertSubscription('tester', ['test']);
for (let i = 0; i < 20; ++i) {
await expect(
store.publish({
event: mkEvent(`hello ${i}`),
notifiedSubscribers: [],
credentials: mockCredentials.service(),
}),
).resolves.toEqual({ eventId: String(i + 1) });
}
await expect(store.readSubscription('tester')).resolves.toEqual({
events: [
mkEvent('hello 15'),
mkEvent('hello 16'),
mkEvent('hello 17'),
mkEvent('hello 18'),
mkEvent('hello 19'),
],
});
});
});
plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
+164
View File
@@ -0,0 +1,164 @@
/*
* Copyright 2024 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 { EventParams } from '@backstage/plugin-events-node';
import { EventBusStore } from './types';
import { NotFoundError } from '@backstage/errors';
import {
BackstageCredentials,
BackstageServicePrincipal,
} from '@backstage/backend-plugin-api';
const MAX_BATCH_SIZE = 10;
const MAX_EVENTS_DEFAULT = 1_000;
export class MemoryEventBusStore implements EventBusStore {
#maxEvents: number;
#events = new Array<
EventParams & { seq: number; notifiedSubscribers: Set<string> }
>();
#subscribers = new Map<
string,
{ id: string; seq: number; topics: Set<string> }
>();
#listeners = new Set<{
topics: Set<string>;
resolve(result: { topic: string }): void;
}>();
constructor(options: { maxEvents?: number } = {}) {
this.#maxEvents = options.maxEvents ?? MAX_EVENTS_DEFAULT;
}
async publish(options: {
event: EventParams;
notifiedSubscribers: string[];
credentials: BackstageCredentials<BackstageServicePrincipal>;
}): Promise<{ eventId: string } | undefined> {
const topic = options.event.topic;
const notifiedSubscribers = new Set(options.notifiedSubscribers);
let hasOtherSubscribers = false;
for (const sub of this.#subscribers.values()) {
if (sub.topics.has(topic) && !notifiedSubscribers.has(sub.id)) {
hasOtherSubscribers = true;
break;
}
}
if (!hasOtherSubscribers) {
return undefined;
}
const nextSeq = this.#getMaxSeq() + 1;
this.#events.push({ ...options.event, notifiedSubscribers, seq: nextSeq });
for (const listener of this.#listeners) {
if (listener.topics.has(topic)) {
listener.resolve({ topic });
this.#listeners.delete(listener);
}
}
// Trim old events
if (this.#events.length > this.#maxEvents) {
this.#events.shift();
}
return { eventId: String(nextSeq) };
}
#getMaxSeq() {
return this.#events[this.#events.length - 1]?.seq ?? 0;
}
async upsertSubscription(id: string, topics: string[]): Promise<void> {
const existing = this.#subscribers.get(id);
if (existing) {
existing.topics = new Set(topics);
return;
}
const sub = {
id: id,
seq: this.#getMaxSeq(),
topics: new Set(topics),
};
this.#subscribers.set(id, sub);
}
async readSubscription(id: string): Promise<{ events: EventParams[] }> {
const sub = this.#subscribers.get(id);
if (!sub) {
throw new NotFoundError(`Subscription not found`);
}
const events = this.#events
.filter(
event =>
event.seq > sub.seq &&
sub.topics.has(event.topic) &&
!event.notifiedSubscribers.has(id),
)
.slice(0, MAX_BATCH_SIZE);
sub.seq = events[events.length - 1]?.seq ?? sub.seq;
return {
events: events.map(({ topic, eventPayload }) => ({
topic,
eventPayload,
})),
};
}
async setupListener(
subscriptionId: string,
options: {
signal: AbortSignal;
},
): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
return {
waitForUpdate: async () => {
options.signal.throwIfAborted();
const sub = this.#subscribers.get(subscriptionId);
if (!sub) {
throw new NotFoundError(`Subscription not found`);
}
return new Promise<{ topic: string }>((resolve, reject) => {
const listener = {
topics: sub.topics,
resolve(result: { topic: string }) {
resolve(result);
cleanup();
},
};
this.#listeners.add(listener);
const onAbort = () => {
this.#listeners.delete(listener);
reject(options.signal.reason);
cleanup();
};
function cleanup() {
options.signal.removeEventListener('abort', onAbort);
}
options.signal.addEventListener('abort', onAbort);
});
},
};
}
}
plugins/events-backend/src/service/hub/createEventBusRouter.ts
+278
View File
@@ -0,0 +1,278 @@
/*
* Copyright 2024 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 {
DatabaseService,
HttpAuthService,
LifecycleService,
LoggerService,
SchedulerService,
} from '@backstage/backend-plugin-api';
import { Handler } from 'express';
import { createOpenApiRouter } from '../../schema/openapi.generated';
import { MemoryEventBusStore } from './MemoryEventBusStore';
import { DatabaseEventBusStore } from './DatabaseEventBusStore';
import { EventBusStore } from './types';
import { EventParams } from '@backstage/plugin-events-node';
const DEFAULT_NOTIFY_TIMEOUT_MS = 55_000; // Just below 60s, which is a common HTTP timeout
/*
# Event Bus
This comment describes the event bus that is implemented here in the events
backend, and by default used by the events service.
## Overview
The events bus implements a subscription mechanism where subscribers must exist
upfront for events to be stored. It uses a single inbox for all events, with
each subscriber having its own pointer for how far into the inbox it has read.
In order to avoid busy polling, the API uses a long-polling mechanism where a
request is left hanging until the client should try to read again.
The event bus is not implemented with any guarantees of events being consumed,
but it does aim to make it unlikely that events are dropped
## API
### POST /bus/v1/events
This endpoint is used to publish new events to the event bus on a specific
topic. It can optionally include a set of subscription IDs for subscribers that
have already been notified of the event. This is to enable an optimization where
we notify subscribers locally if possible, and avoid the need for events to be
relayed through the events bus at all of possible.
For an event to be published and stored there must already exist a subscriber
that is subscribed to the event's topic, and that hasn't already been notified
of the event. If no such subscriber is found, the event will be discarded.
### PUT /bus/v1/subscriptions/:subscriptionId
This endpoint is used to create or update a subscriptions. Subscriptions are
shared across the entire bus and divided by subscription ID. Multiple clients
can be reading events from the same subscription at the same time, but only one
of those clients will receive each event. This enables division of work by using
the same subscriber ID across multiple instances, as well as broadcasting by
ensuring separate subscribers IDs.
### GET /bus/v1/subscriptions/:subscriptionId/events
This endpoint is used to read events from a subscription. It will return a batch
of events for the subscribed topics that have not yet been read by the
subscription. If no such events are available, the endpoint will return a 202
response and then hang end response until an event is available or a timeout is
reached. This allows clients to call this endpoint in a loop but will keep
traffic overhead fairly low.
## Delivery guarantees
When reading events from the event bus, clients should always implement a
graceful shutdown where they process any events that are received from the
events endpoint before shutting down. This is also the reason that the events
endpoint does not return any events when responding with a 202 blocking the
response, because there would otherwise be a race condition where the events
might be lost in transit if the client shuts down. By always sending an empty
response and requiring the client to send another request, we ensure that the
client is prepared to halt shutdown until the request had been fully processed.
## Local processing optimization
When possible, events will be processed locally before sent to the event bus.
The client will also inform the bus of which subscriptions have already been
notified of the event, so that the bus can completely avoid storing an event if
it has already been fully consumed by all subscribers.
## Automated cleanup & event window
Events are deleted once they are outside the guaranteed storage window. By
default the window 10 minutes for all events, and 24 hours for the last 10000
events. This ensures that the event log doesn't grow indefinitely, while still
allowing subscribers with restarts or outages to catch up to past events,
ensuring that events are likely not lost.
Subscriptions are also cleaned up if their read pointer falls outside of the
current event window. This ensures that stale subscribers don't accumulate and
cause unnecessary storage of events.
*/
async function createEventBusStore(deps: {
logger: LoggerService;
database: DatabaseService;
scheduler: SchedulerService;
lifecycle: LifecycleService;
httpAuth: HttpAuthService;
}): Promise<EventBusStore> {
const db = await deps.database.getClient();
if (db.client.config.client === 'pg') {
deps.logger.info('Database is PostgreSQL, using database store');
return await DatabaseEventBusStore.create(deps);
}
deps.logger.info('Database is not PostgreSQL, using memory store');
return new MemoryEventBusStore();
}
/**
* Creates a new event bus router
* @internal
*/
export async function createEventBusRouter(options: {
logger: LoggerService;
database: DatabaseService;
scheduler: SchedulerService;
lifecycle: LifecycleService;
httpAuth: HttpAuthService;
notifyTimeoutMs?: number; // for testing
}): Promise<Handler> {
const { httpAuth, notifyTimeoutMs = DEFAULT_NOTIFY_TIMEOUT_MS } = options;
const logger = options.logger.child({ type: 'EventBus' });
const store = await createEventBusStore(options);
const apiRouter = await createOpenApiRouter();
apiRouter.post('/bus/v1/events', async (req, res) => {
const credentials = await httpAuth.credentials(req, {
allow: ['service'],
});
const topic = req.body.event.topic;
const notifiedSubscribers = req.body.notifiedSubscribers;
const result = await store.publish({
event: {
topic,
eventPayload: req.body.event.payload,
} as EventParams,
notifiedSubscribers,
credentials,
});
if (result) {
logger.debug(
`Published event to '${topic}' with ID '${result.eventId}'`,
{
subject: credentials.principal.subject,
},
);
res.status(201).end();
} else {
if (notifiedSubscribers) {
const notified = `'${notifiedSubscribers.join("', '")}'`;
logger.debug(
`Skipped publishing of event to '${topic}', subscribers have already been notified: ${notified}`,
{ subject: credentials.principal.subject },
);
} else {
logger.debug(
`Skipped publishing of event to '${topic}', no subscribers present`,
{ subject: credentials.principal.subject },
);
}
res.status(204).end();
}
});
apiRouter.get(
'/bus/v1/subscriptions/:subscriptionId/events',
async (req, res) => {
const credentials = await httpAuth.credentials(req, {
allow: ['service'],
});
const id = req.params.subscriptionId;
const controller = new AbortController();
req.on('end', () => controller.abort());
// By setting up the listener first we make sure we don't miss any events
// that are published while reading. If an event is published we'll receive
// a notification, which we may ignore depending on the outcome of the read
const listener = await store.setupListener(id, {
signal: controller.signal,
});
// By timing out requests we make sure they don't stall or that events get stuck.
// For the caller there's no difference between a timeout and a
// notification, either way they should try reading again.
const timeout = setTimeout(() => {
controller.abort();
}, notifyTimeoutMs);
try {
const { events } = await store.readSubscription(id);
logger.debug(
`Reading subscription '${id}' resulted in ${events.length} events`,
{ subject: credentials.principal.subject },
);
if (events.length > 0) {
res.json({
events: events.map(event => ({
topic: event.topic,
payload: event.eventPayload,
})),
});
} else {
res.status(202);
res.flushHeaders();
try {
const { topic } = await listener.waitForUpdate();
logger.debug(
`Received notification for subscription '${id}' for topic '${topic}'`,
{ subject: credentials.principal.subject },
);
} catch (error) {
if (error !== controller.signal.reason) {
logger.error(`Error listening for subscription '${id}'`, error);
}
} finally {
// A small extra delay ensures a more even spread of events across
// consumers in case some consumers are faster than others
await new Promise(resolve =>
setTimeout(resolve, 1 + Math.random() * 9),
);
res.end();
}
}
} finally {
controller.abort();
clearTimeout(timeout);
}
},
);
apiRouter.put('/bus/v1/subscriptions/:subscriptionId', async (req, res) => {
const credentials = await httpAuth.credentials(req, {
allow: ['service'],
});
const id = req.params.subscriptionId;
await store.upsertSubscription(id, req.body.topics, credentials);
logger.debug(
`New subscription '${id}' for topics '${req.body.topics.join("', '")}'`,
{ subject: credentials.principal.subject },
);
res.status(201).end();
});
return apiRouter;
}
plugins/events-backend/src/service/hub/index.ts
+17
View File
@@ -0,0 +1,17 @@
/*
* Copyright 2024 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 { createEventBusRouter } from './createEventBusRouter';
plugins/events-backend/src/service/hub/types.ts
+44
View File
@@ -0,0 +1,44 @@
/*
* Copyright 2024 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 {
BackstageCredentials,
BackstageServicePrincipal,
} from '@backstage/backend-plugin-api';
import { EventParams } from '@backstage/plugin-events-node';
export type EventBusStore = {
publish(options: {
event: EventParams;
notifiedSubscribers?: string[];
credentials: BackstageCredentials<BackstageServicePrincipal>;
}): Promise<{ eventId: string } | undefined>;
upsertSubscription(
subscriptionId: string,
topics: string[],
credentials: BackstageCredentials<BackstageServicePrincipal>,
): Promise<void>;
readSubscription(subscriptionId: string): Promise<{ events: EventParams[] }>;
setupListener(
subscriptionId: string,
options: {
signal: AbortSignal;
},
): Promise<{ waitForUpdate(): Promise<{ topic: string }> }>;
};
plugins/events-node/api-report.md
+12 -1
View File
@@ -3,6 +3,9 @@
> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
```ts
import { AuthService } from '@backstage/backend-plugin-api';
import { DiscoveryService } from '@backstage/backend-plugin-api';
import { LifecycleService } from '@backstage/backend-plugin-api';
import { LoggerService } from '@backstage/backend-plugin-api';
import { ServiceFactory } from '@backstage/backend-plugin-api';
import { ServiceRef } from '@backstage/backend-plugin-api';
@@ -11,7 +14,15 @@ import { ServiceRef } from '@backstage/backend-plugin-api';
export class DefaultEventsService implements EventsService {
// (undocumented)
static create(options: { logger: LoggerService }): DefaultEventsService;
forPlugin(pluginId: string): EventsService;
forPlugin(
pluginId: string,
options?: {
discovery: DiscoveryService;
logger: LoggerService;
auth: AuthService;
lifecycle: LifecycleService;
},
): EventsService;
// (undocumented)
publish(params: EventParams): Promise<void>;
// (undocumented)
plugins/events-node/package.json
+5 -1
View File
@@ -51,7 +51,11 @@
"test": "backstage-cli package test"
},
"dependencies": {
"@backstage/backend-plugin-api": "workspace:^"
"@backstage/backend-plugin-api": "workspace:^",
"@backstage/errors": "workspace:^",
"@backstage/types": "workspace:^",
"cross-fetch": "^4.0.0",
"uri-template": "^2.0.0"
},
"devDependencies": {
"@backstage/backend-common": "^0.25.0",
plugins/events-node/src/api/DefaultEventsService.ts
+349 -57
View File
@@ -14,9 +14,320 @@
* limitations under the License.
*/
import { LoggerService } from '@backstage/backend-plugin-api';
import {
AuthService,
DiscoveryService,
LifecycleService,
LoggerService,
} from '@backstage/backend-plugin-api';
import { EventParams } from './EventParams';
import { EventsService, EventsServiceSubscribeOptions } from './EventsService';
import { DefaultApiClient } from '../generated';
import { ResponseError } from '@backstage/errors';
const POLL_BACKOFF_START_MS = 1_000;
const POLL_BACKOFF_MAX_MS = 60_000;
const POLL_BACKOFF_FACTOR = 2;
/**
* Local event bus for subscribers within the same process.
*
* When publishing events we'll keep track of which subscribers we managed to
* reach locally, and forward those subscriber IDs to the events backend if it
* is in use. The events backend will then both avoid forwarding the same events
* to those subscribers again, but also avoid storing the event altogether if
* there are no other subscribers.
* @internal
*/
export class LocalEventBus {
readonly #logger: LoggerService;
readonly #subscribers = new Map<
string,
Omit<EventsServiceSubscribeOptions, 'topics'>[]
>();
constructor(logger: LoggerService) {
this.#logger = logger;
}
async publish(
params: EventParams,
): Promise<{ notifiedSubscribers: string[] }> {
this.#logger.debug(
`Event received: topic=${params.topic}, metadata=${JSON.stringify(
params.metadata,
)}, payload=${JSON.stringify(params.eventPayload)}`,
);
if (!this.#subscribers.has(params.topic)) {
return { notifiedSubscribers: [] };
}
const onEventPromises: Promise<string>[] = [];
this.#subscribers.get(params.topic)?.forEach(subscription => {
onEventPromises.push(
(async () => {
try {
await subscription.onEvent(params);
} catch (error) {
this.#logger.warn(
`Subscriber "${subscription.id}" failed to process event for topic "${params.topic}"`,
error,
);
}
return subscription.id;
})(),
);
});
return { notifiedSubscribers: await Promise.all(onEventPromises) };
}
async subscribe(options: EventsServiceSubscribeOptions): Promise<void> {
options.topics.forEach(topic => {
if (!this.#subscribers.has(topic)) {
this.#subscribers.set(topic, []);
}
this.#subscribers.get(topic)!.push({
id: options.id,
onEvent: options.onEvent,
});
});
}
}
/**
* Plugin specific events bus that delegates to the local bus, as well as the
* events backend if it is available.
*/
class PluginEventsService implements EventsService {
constructor(
private readonly pluginId: string,
private readonly localBus: LocalEventBus,
private readonly logger: LoggerService,
private client?: DefaultApiClient,
private readonly auth?: AuthService,
) {}
async publish(params: EventParams): Promise<void> {
const lock = this.#getShutdownLock();
try {
const { notifiedSubscribers } = await this.localBus.publish(params);
if (!this.client) {
return;
}
const token = await this.#getToken();
if (!token) {
return;
}
const res = await this.client.postEvent(
{
body: {
event: { payload: params.eventPayload, topic: params.topic },
notifiedSubscribers,
},
},
{ token },
);
if (!res.ok) {
if (res.status === 404) {
this.logger.warn(
`Event publish request failed with status 404, events backend not found. Future events will not be persisted.`,
);
delete this.client;
return;
}
throw await ResponseError.fromResponse(res);
}
} finally {
lock.release();
}
}
async subscribe(options: EventsServiceSubscribeOptions): Promise<void> {
const subscriptionId = `${this.pluginId}.${options.id}`;
await this.localBus.subscribe({
id: subscriptionId,
topics: options.topics,
onEvent: options.onEvent,
});
if (!this.client) {
return;
}
const token = await this.#getToken();
if (!token) {
return;
}
const res = await this.client.putSubscription(
{
path: { subscriptionId },
body: { topics: options.topics },
},
{ token },
);
if (!res.ok) {
if (res.status === 404) {
this.logger.warn(
`Event subscribe request failed with status 404, events backend not found. Will only receive events that were sent locally on this process.`,
);
delete this.client;
return;
}
throw await ResponseError.fromResponse(res);
}
this.#startPolling(subscriptionId, options.topics, options.onEvent);
}
#startPolling(
subscriptionId: string,
topics: string[],
onEvent: EventsServiceSubscribeOptions['onEvent'],
) {
let backoffMs = POLL_BACKOFF_START_MS;
const poll = async () => {
if (!this.client) {
return;
}
const lock = this.#getShutdownLock();
try {
const token = await this.#getToken();
if (!token) {
return;
}
const res = await this.client.getSubscriptionEvents(
{
path: { subscriptionId },
},
{ token },
);
if (!res.ok) {
if (res.status === 404) {
this.logger.info(
`Polling event subscription resulted in a 404, recreating subscription`,
);
const putRes = await this.client.putSubscription(
{
path: { subscriptionId },
body: { topics },
},
{ token },
);
if (!putRes.ok) {
throw await ResponseError.fromResponse(res);
}
}
throw await ResponseError.fromResponse(res);
}
backoffMs = POLL_BACKOFF_START_MS;
// 202 means there were no immediately available events, but the
// response will block until either new events are available or the
// request times out. In both cases we should should try to read events
// immediately again
if (res.status === 202) {
lock.release();
await res.body?.getReader()?.closed;
process.nextTick(poll);
} else if (res.status === 200) {
const data = await res.json();
if (data) {
for (const event of data.events ?? []) {
try {
await onEvent({
topic: event.topic,
eventPayload: event.payload,
});
} catch (error) {
this.logger.warn(
`Subscriber "${subscriptionId}" failed to process event for topic "${event.topic}"`,
error,
);
}
}
}
process.nextTick(poll);
} else {
this.logger.warn(
`Unexpected response status ${res.status} from events backend for subscription "${subscriptionId}"`,
);
}
} catch (error) {
this.logger.warn(
`Poll failed for subscription "${subscriptionId}", retrying in ${backoffMs.toFixed(
0,
)}ms`,
error,
);
setTimeout(poll, backoffMs);
backoffMs = Math.min(
backoffMs * POLL_BACKOFF_FACTOR,
POLL_BACKOFF_MAX_MS,
);
} finally {
lock.release();
}
};
poll();
}
async #getToken() {
if (!this.auth) {
throw new Error('Auth service not available');
}
try {
const { token } = await this.auth.getPluginRequestToken({
onBehalfOf: await this.auth.getOwnServiceCredentials(),
targetPluginId: 'events',
});
return token;
} catch (error) {
// This is a bit hacky, but handles the case where new auth is used
// without legacy auth fallback, and the events backend is not installed
if (String(error).includes('Unable to generate legacy token')) {
this.logger.warn(
`The events backend is not available and neither is legacy auth. Future events will not be persisted.`,
);
delete this.client;
return undefined;
}
throw error;
}
}
async shutdown() {
this.#isShuttingDown = true;
await Promise.all(this.#shutdownLocks);
}
#isShuttingDown = false;
#shutdownLocks: Promise<void>[] = [];
// This locking mechanism helps ensure that we are either idle or waiting for
// a blocked events call before shutting down. It increases out changes of
// never dropping any events on shutdown.
#getShutdownLock(): { release(): void } {
if (this.#isShuttingDown) {
throw new Error('Service is shutting down');
}
let release: () => void;
this.#shutdownLocks.push(
new Promise<void>(resolve => {
release = resolve;
}),
);
return { release: release! };
}
}
/**
* In-process event broker which will pass the event to all registered subscribers
@@ -28,15 +339,16 @@ import { EventsService, EventsServiceSubscribeOptions } from './EventsService';
*/
// TODO(pjungermann): add opentelemetry? (see plugins/catalog-backend/src/util/opentelemetry.ts, etc.)
export class DefaultEventsService implements EventsService {
private readonly subscribers = new Map<
string,
Omit<EventsServiceSubscribeOptions, 'topics'>[]
>();
private constructor(private readonly logger: LoggerService) {}
private constructor(
private readonly logger: LoggerService,
private readonly localBus: LocalEventBus,
) {}
static create(options: { logger: LoggerService }): DefaultEventsService {
return new DefaultEventsService(options.logger);
return new DefaultEventsService(
options.logger,
new LocalEventBus(options.logger),
);
}
/**
@@ -45,60 +357,40 @@ export class DefaultEventsService implements EventsService {
*
* @param pluginId - The plugin that the `EventService` should be created for.
*/
forPlugin(pluginId: string): EventsService {
return {
publish: (params: EventParams): Promise<void> => {
return this.publish(params);
},
subscribe: (options: EventsServiceSubscribeOptions): Promise<void> => {
return this.subscribe({
...options,
id: `${pluginId}.${options.id}`,
});
},
};
forPlugin(
pluginId: string,
options?: {
discovery: DiscoveryService;
logger: LoggerService;
auth: AuthService;
lifecycle: LifecycleService;
},
): EventsService {
const client =
options &&
new DefaultApiClient({
discoveryApi: options.discovery,
fetchApi: { fetch }, // use native node fetch
});
const logger = options?.logger ?? this.logger;
const service = new PluginEventsService(
pluginId,
this.localBus,
logger,
client,
options?.auth,
);
options?.lifecycle.addShutdownHook(async () => {
await service.shutdown();
});
return service;
}
async publish(params: EventParams): Promise<void> {
this.logger.debug(
`Event received: topic=${params.topic}, metadata=${JSON.stringify(
params.metadata,
)}, payload=${JSON.stringify(params.eventPayload)}`,
);
if (!this.subscribers.has(params.topic)) {
return;
}
const onEventPromises: Promise<void>[] = [];
this.subscribers.get(params.topic)?.forEach(subscription => {
onEventPromises.push(
(async () => {
try {
await subscription.onEvent(params);
} catch (error) {
this.logger.warn(
`Subscriber "${subscription.id}" failed to process event for topic "${params.topic}"`,
error,
);
}
})(),
);
});
await Promise.all(onEventPromises);
await this.localBus.publish(params);
}
async subscribe(options: EventsServiceSubscribeOptions): Promise<void> {
options.topics.forEach(topic => {
if (!this.subscribers.has(topic)) {
this.subscribers.set(topic, []);
}
this.subscribers.get(topic)!.push({
id: options.id,
onEvent: options.onEvent,
});
});
this.localBus.subscribe(options);
}
}
plugins/events-node/src/generated/apis/DefaultApi.client.ts
+152
View File
@@ -0,0 +1,152 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
import { DiscoveryApi } from '../types/discovery';
import { FetchApi } from '../types/fetch';
import crossFetch from 'cross-fetch';
import { pluginId } from '../pluginId';
import * as parser from 'uri-template';
import { GetSubscriptionEvents200Response } from '../models/GetSubscriptionEvents200Response.model';
import { PostEventRequest } from '../models/PostEventRequest.model';
import { PutSubscriptionRequest } from '../models/PutSubscriptionRequest.model';
/**
* Wraps the Response type to convey a type on the json call.
*
* @public
*/
export type TypedResponse<T> = Omit<Response, 'json'> & {
json: () => Promise<T>;
};
/**
* Options you can pass into a request for additional information.
*
* @public
*/
export interface RequestOptions {
token?: string;
}
/**
* no description
*/
export class DefaultApiClient {
private readonly discoveryApi: DiscoveryApi;
private readonly fetchApi: FetchApi;
constructor(options: {
discoveryApi: { getBaseUrl(pluginId: string): Promise<string> };
fetchApi?: { fetch: typeof fetch };
}) {
this.discoveryApi = options.discoveryApi;
this.fetchApi = options.fetchApi || { fetch: crossFetch };
}
/**
* Get new events for the provided subscription
* @param subscriptionId
*/
public async getSubscriptionEvents(
// @ts-ignore
request: {
path: {
subscriptionId: string;
};
},
options?: RequestOptions,
): Promise<TypedResponse<void | GetSubscriptionEvents200Response>> {
const baseUrl = await this.discoveryApi.getBaseUrl(pluginId);
const uriTemplate = `/bus/v1/subscriptions/{subscriptionId}/events`;
const uri = parser.parse(uriTemplate).expand({
subscriptionId: request.path.subscriptionId,
});
return await this.fetchApi.fetch(`${baseUrl}${uri}`, {
headers: {
'Content-Type': 'application/json',
...(options?.token && { Authorization: `Bearer ${options?.token}` }),
},
method: 'GET',
});
}
/**
* Publish a new event
* @param postEventRequest
*/
public async postEvent(
// @ts-ignore
request: {
body: PostEventRequest;
},
options?: RequestOptions,
): Promise<TypedResponse<void>> {
const baseUrl = await this.discoveryApi.getBaseUrl(pluginId);
const uriTemplate = `/bus/v1/events`;
const uri = parser.parse(uriTemplate).expand({});
return await this.fetchApi.fetch(`${baseUrl}${uri}`, {
headers: {
'Content-Type': 'application/json',
...(options?.token && { Authorization: `Bearer ${options?.token}` }),
},
method: 'POST',
body: JSON.stringify(request.body),
});
}
/**
* Ensures that the subscription exists with the provided configuration
* @param subscriptionId
* @param putSubscriptionRequest
*/
public async putSubscription(
// @ts-ignore
request: {
path: {
subscriptionId: string;
};
body: PutSubscriptionRequest;
},
options?: RequestOptions,
): Promise<TypedResponse<void>> {
const baseUrl = await this.discoveryApi.getBaseUrl(pluginId);
const uriTemplate = `/bus/v1/subscriptions/{subscriptionId}`;
const uri = parser.parse(uriTemplate).expand({
subscriptionId: request.path.subscriptionId,
});
return await this.fetchApi.fetch(`${baseUrl}${uri}`, {
headers: {
'Content-Type': 'application/json',
...(options?.token && { Authorization: `Bearer ${options?.token}` }),
},
method: 'PUT',
body: JSON.stringify(request.body),
});
}
}
plugins/events-node/src/generated/apis/index.ts
+17
View File
@@ -0,0 +1,17 @@
/*
* Copyright 2024 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 * from './DefaultApi.client';
plugins/events-node/src/generated/index.ts
+18
View File
@@ -0,0 +1,18 @@
/*
* Copyright 2024 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 * from './apis';
export * from './models';
plugins/events-node/src/generated/models/ErrorError.model.ts
+24
View File
@@ -0,0 +1,24 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
export interface ErrorError {
name: string;
message: string;
}
plugins/events-node/src/generated/models/ErrorRequest.model.ts
+24
View File
@@ -0,0 +1,24 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
export interface ErrorRequest {
method: string;
url: string;
}
plugins/events-node/src/generated/models/ErrorResponse.model.ts
+23
View File
@@ -0,0 +1,23 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
export interface ErrorResponse {
statusCode: number;
}
plugins/events-node/src/generated/models/Event.model.ts
+30
View File
@@ -0,0 +1,30 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
export interface Event {
/**
* The topic that the event is published on
*/
topic: string;
/**
* The event payload
*/
payload: any | null;
}
plugins/events-node/src/generated/models/GetSubscriptionEvents200Response.model.ts
+24
View File
@@ -0,0 +1,24 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
import { Event } from '../models/Event.model';
export interface GetSubscriptionEvents200Response {
events: Array<Event>;
}
plugins/events-node/src/generated/models/ModelError.model.ts
+28
View File
@@ -0,0 +1,28 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
import { ErrorError } from '../models/ErrorError.model';
import { ErrorRequest } from '../models/ErrorRequest.model';
import { ErrorResponse } from '../models/ErrorResponse.model';
export interface ModelError {
error: ErrorError;
request: ErrorRequest;
response: ErrorResponse;
}
plugins/events-node/src/generated/models/PostEventRequest.model.ts
+28
View File
@@ -0,0 +1,28 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
import { Event } from '../models/Event.model';
export interface PostEventRequest {
event: Event;
/**
* The IDs of subscriptions that have already received this event
*/
notifiedSubscribers?: Array<string>;
}
plugins/events-node/src/generated/models/PutSubscriptionRequest.model.ts
+26
View File
@@ -0,0 +1,26 @@
/*
* Copyright 2024 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.
*/
// ******************************************************************
// * THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY. *
// ******************************************************************
export interface PutSubscriptionRequest {
/**
* The topics to subscribe to
*/
topics: Array<string>;
}
plugins/events-node/src/generated/models/index.ts
+24
View File
@@ -0,0 +1,24 @@
/*
* Copyright 2024 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 * from '../models/ErrorError.model';
export * from '../models/ErrorRequest.model';
export * from '../models/ErrorResponse.model';
export * from '../models/Event.model';
export * from '../models/GetSubscriptionEvents200Response.model';
export * from '../models/ModelError.model';
export * from '../models/PostEventRequest.model';
export * from '../models/PutSubscriptionRequest.model';
plugins/events-node/src/generated/pluginId.ts
+17
View File
@@ -0,0 +1,17 @@
/*
* Copyright 2024 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 const pluginId = 'events';
plugins/events-node/src/generated/types/discovery.ts
+22
View File
@@ -0,0 +1,22 @@
/*
* Copyright 2023 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.
*/
/**
* This is a copy of the DiscoveryApi, to avoid importing core-plugin-api.
*/
export type DiscoveryApi = {
getBaseUrl(pluginId: string): Promise<string>;
};
plugins/events-node/src/generated/types/fetch.ts
+22
View File
@@ -0,0 +1,22 @@
/*
* Copyright 2023 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.
*/
/**
* This is a copy of FetchApi, to avoid importing core-plugin-api.
*/
export type FetchApi = {
fetch: typeof fetch;
};
plugins/events-node/src/service.ts
+14 -2
View File
@@ -38,11 +38,23 @@ export const eventsServiceFactory = createServiceFactory({
deps: {
pluginMetadata: coreServices.pluginMetadata,
rootLogger: coreServices.rootLogger,
discovery: coreServices.discovery,
logger: coreServices.logger,
lifecycle: coreServices.lifecycle,
auth: coreServices.auth,
},
async createRootContext({ rootLogger }) {
return DefaultEventsService.create({ logger: rootLogger });
},
async factory({ pluginMetadata }, eventsService) {
return eventsService.forPlugin(pluginMetadata.getId());
async factory(
{ pluginMetadata, discovery, logger, lifecycle, auth },
eventsService,
) {
return eventsService.forPlugin(pluginMetadata.getId(), {
discovery,
logger,
lifecycle,
auth,
});
},
});
yarn.lock
+12
View File
@@ -6141,16 +6141,23 @@ __metadata:
version: 0.0.0-use.local
resolution: "@backstage/plugin-events-backend@workspace:plugins/events-backend"
dependencies:
"@backstage/backend-app-api": "workspace:^"
"@backstage/backend-common": ^0.25.0
"@backstage/backend-defaults": "workspace:^"
"@backstage/backend-openapi-utils": "workspace:^"
"@backstage/backend-plugin-api": "workspace:^"
"@backstage/backend-test-utils": "workspace:^"
"@backstage/cli": "workspace:^"
"@backstage/config": "workspace:^"
"@backstage/errors": "workspace:^"
"@backstage/plugin-events-backend-test-utils": "workspace:^"
"@backstage/plugin-events-node": "workspace:^"
"@backstage/repo-tools": "workspace:^"
"@backstage/types": "workspace:^"
"@types/express": ^4.17.6
express: ^4.17.1
express-promise-router: ^4.1.0
knex: ^3.0.0
supertest: ^7.0.0
winston: ^3.2.1
languageName: unknown
@@ -6164,6 +6171,10 @@ __metadata:
"@backstage/backend-plugin-api": "workspace:^"
"@backstage/backend-test-utils": "workspace:^"
"@backstage/cli": "workspace:^"
"@backstage/errors": "workspace:^"
"@backstage/types": "workspace:^"
cross-fetch: ^4.0.0
uri-template: ^2.0.0
languageName: unknown
linkType: soft
@@ -26388,6 +26399,7 @@ __metadata:
"@backstage/plugin-catalog-backend-module-scaffolder-entity-model": "workspace:^"
"@backstage/plugin-catalog-backend-module-unprocessed": "workspace:^"
"@backstage/plugin-devtools-backend": "workspace:^"
"@backstage/plugin-events-backend": "workspace:^"
"@backstage/plugin-kubernetes-backend": "workspace:^"
"@backstage/plugin-notifications-backend": "workspace:^"
"@backstage/plugin-permission-backend": "workspace:^"