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

events-backend: track creator of events and subscriptions

Browse Source 
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Patrik Oldsberg
2024-09-19 01:19:31 +02:00
parent e7a0b6572c
commit 8241289427
8 changed files with 100 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/events-backend/migrations/20240523100528_init.js
+8
View File
@@ -29,6 +29,10 @@ exports.up = async function up(knex) {
.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())
@@ -53,6 +57,10 @@ exports.up = async function up(knex) {
.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())
plugins/events-backend/src/migrations.test.ts
+4
View File
@@ -56,11 +56,13 @@ describe('migrations', () => {
await knex('event_bus_events').insert({
topic: 'test',
created_by: 'abc',
data_json: JSON.stringify({ message: 'hello' }),
consumed_by: ['tester'],
});
await knex('event_bus_subscriptions').insert({
id: 'tester',
created_by: 'abc',
read_until: '5',
topics: ['test', 'test2'],
});
@@ -68,6 +70,7 @@ describe('migrations', () => {
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(),
@@ -77,6 +80,7 @@ describe('migrations', () => {
await expect(knex('event_bus_subscriptions')).resolves.toEqual([
{
id: 'tester',
created_by: 'abc',
created_at: expect.anything(),
updated_at: expect.anything(),
read_until: '5',
plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts
+45 -10
View File
@@ -14,7 +14,11 @@
* limitations under the License.
*/
import { TestDatabases, mockServices } from '@backstage/backend-test-utils';
import {
TestDatabases,
mockCredentials,
mockServices,
} from '@backstage/backend-test-utils';
import { DatabaseEventBusStore } from './DatabaseEventBusStore';
const logger = mockServices.logger.mock();
@@ -30,12 +34,21 @@ describe('DatabaseEventBusStore', () => {
const db = await databases.init(databaseId);
const store = await DatabaseEventBusStore.forTest({ logger, db });
await store.upsertSubscription('tester-1', ['test']);
await store.upsertSubscription('tester-2', ['test']);
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(),
});
}
@@ -48,7 +61,11 @@ describe('DatabaseEventBusStore', () => {
"Subscription with ID 'tester-2' not found",
);
await store.upsertSubscription('tester-3', ['test']);
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({
@@ -70,12 +87,21 @@ describe('DatabaseEventBusStore', () => {
maxAge: 0,
});
await store.upsertSubscription('tester-1', ['test']);
await store.upsertSubscription('tester-2', ['test']);
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(),
});
}
@@ -88,7 +114,11 @@ describe('DatabaseEventBusStore', () => {
"Subscription with ID 'tester-2' not found",
);
await store.upsertSubscription('tester-3', ['test']);
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({
@@ -110,11 +140,16 @@ describe('DatabaseEventBusStore', () => {
minAge: 1000,
});
await store.upsertSubscription('tester-1', ['test']);
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(),
});
}
@@ -137,8 +172,8 @@ describe('DatabaseEventBusStore', () => {
const COUNT = '100000';
await db.raw(`
INSERT INTO event_bus_events (id, topic, data_json)
SELECT id, 'test', '{}'
INSERT INTO event_bus_events (id, created_by, topic, data_json)
SELECT id, 'abc', 'test', '{}'
FROM generate_series(1, ${COUNT}) AS id
`);
plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
+21 -3
View File
@@ -17,6 +17,8 @@ import { EventParams } from '@backstage/plugin-events-node';
import { EventBusStore } from './types';
import { Knex } from 'knex';
import {
BackstageCredentials,
BackstageServicePrincipal,
DatabaseService,
LifecycleService,
LoggerService,
@@ -40,6 +42,7 @@ const TOPIC_PUBLISH = 'event_bus_publish';
type EventsRow = {
id: string;
created_by: string;
created_at: Date;
topic: string;
data_json: string;
@@ -48,12 +51,19 @@ type EventsRow = {
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',
@@ -366,6 +376,7 @@ export class DatabaseEventBusStore implements EventBusStore {
async publish(options: {
event: EventParams;
notifiedSubscribers?: string[];
credentials: BackstageCredentials<BackstageServicePrincipal>;
}): Promise<{ eventId: string } | undefined> {
const topic = options.event.topic;
const notifiedSubscribers = options.notifiedSubscribers ?? [];
@@ -374,9 +385,10 @@ export class DatabaseEventBusStore implements EventBusStore {
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('?? (??, ??, ??)', [
this.#db.raw('?? (??, ??, ??, ??)', [
TABLE_EVENTS,
// These are the rows that we insert, and should match the SELECT below
'created_by',
'topic',
'data_json',
'consumed_by',
@@ -387,6 +399,7 @@ export class DatabaseEventBusStore implements EventBusStore {
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({
@@ -429,17 +442,22 @@ export class DatabaseEventBusStore implements EventBusStore {
return { eventId: id };
}
async upsertSubscription(id: string, topics: string[]): Promise<void> {
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(['topics', 'updated_at'])
.merge(['created_by', 'topics', 'updated_at'])
.returning('*');
if (result.length !== 1) {
plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts
+5
View File
@@ -15,6 +15,7 @@
*/
import { EventParams } from '@backstage/plugin-events-node';
import { MemoryEventBusStore } from './MemoryEventBusStore';
import { mockCredentials } from '@backstage/backend-test-utils';
function mkEvent(message: string): EventParams {
return {
@@ -31,6 +32,7 @@ describe('MemoryEventBusStore', () => {
store.publish({
event: mkEvent('hello'),
notifiedSubscribers: [],
credentials: mockCredentials.service(),
}),
).resolves.toEqual(undefined);
@@ -44,6 +46,7 @@ describe('MemoryEventBusStore', () => {
store.publish({
event: mkEvent('hello'),
notifiedSubscribers: [],
credentials: mockCredentials.service(),
}),
).resolves.toEqual({ eventId: '1' });
@@ -51,6 +54,7 @@ describe('MemoryEventBusStore', () => {
store.publish({
event: mkEvent('ignored'),
notifiedSubscribers: ['tester'],
credentials: mockCredentials.service(),
}),
).resolves.toEqual(undefined);
@@ -69,6 +73,7 @@ describe('MemoryEventBusStore', () => {
store.publish({
event: mkEvent(`hello ${i}`),
notifiedSubscribers: [],
credentials: mockCredentials.service(),
}),
).resolves.toEqual({ eventId: String(i + 1) });
}
plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
+5
View File
@@ -16,6 +16,10 @@
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;
@@ -41,6 +45,7 @@ export class MemoryEventBusStore implements EventBusStore {
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);
plugins/events-backend/src/service/hub/createEventBusRouter.ts
+2 -1
View File
@@ -161,6 +161,7 @@ export async function createEventBusRouter(options: {
eventPayload: req.body.event.payload,
} as EventParams,
notifiedSubscribers,
credentials,
});
if (result) {
logger.debug(
@@ -263,7 +264,7 @@ export async function createEventBusRouter(options: {
});
const id = req.params.subscriptionId;
await store.upsertSubscription(id, req.body.topics);
await store.upsertSubscription(id, req.body.topics, credentials);
logger.debug(
`New subscription '${id}' for topics '${req.body.topics.join("', '")}'`,
plugins/events-backend/src/service/hub/types.ts
+10 -1
View File
@@ -14,15 +14,24 @@
* 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[]): Promise<void>;
upsertSubscription(
subscriptionId: string,
topics: string[],
credentials: BackstageCredentials<BackstageServicePrincipal>,
): Promise<void>;
readSubscription(subscriptionId: string): Promise<{ events: EventParams[] }>;