From 82412894273addaac64689c72f2afc62f3a06d2e Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Thu, 19 Sep 2024 01:19:31 +0200 Subject: [PATCH] events-backend: track creator of events and subscriptions Signed-off-by: Patrik Oldsberg --- .../migrations/20240523100528_init.js | 8 +++ plugins/events-backend/src/migrations.test.ts | 4 ++ .../service/hub/DatabaseEventBusStore.test.ts | 55 +++++++++++++++---- .../src/service/hub/DatabaseEventBusStore.ts | 24 +++++++- .../service/hub/MemoryEventBusStore.test.ts | 5 ++ .../src/service/hub/MemoryEventBusStore.ts | 5 ++ .../src/service/hub/createEventBusRouter.ts | 3 +- .../events-backend/src/service/hub/types.ts | 11 +++- 8 files changed, 100 insertions(+), 15 deletions(-) diff --git a/plugins/events-backend/migrations/20240523100528_init.js b/plugins/events-backend/migrations/20240523100528_init.js index 7ec85d7917..c3b66658fd 100644 --- a/plugins/events-backend/migrations/20240523100528_init.js +++ b/plugins/events-backend/migrations/20240523100528_init.js @@ -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()) diff --git a/plugins/events-backend/src/migrations.test.ts b/plugins/events-backend/src/migrations.test.ts index 16853f6052..069b326075 100644 --- a/plugins/events-backend/src/migrations.test.ts +++ b/plugins/events-backend/src/migrations.test.ts @@ -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', diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts index 3c8d2cb377..bfc68b7663 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts @@ -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 `); diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts index 3a6a98d919..a25856e88d 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts @@ -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, +) { + 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; }): 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 { + async upsertSubscription( + id: string, + topics: string[], + credentials: BackstageCredentials, + ): Promise { const [{ max: maxId }] = await this.#db(TABLE_EVENTS).max('id'); const result = await this.#db(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) { diff --git a/plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts b/plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts index 6866f290b5..52d075951d 100644 --- a/plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts +++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.test.ts @@ -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) }); } diff --git a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts index 67d384b1fa..6c999dc390 100644 --- a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts @@ -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; }): Promise<{ eventId: string } | undefined> { const topic = options.event.topic; const notifiedSubscribers = new Set(options.notifiedSubscribers); diff --git a/plugins/events-backend/src/service/hub/createEventBusRouter.ts b/plugins/events-backend/src/service/hub/createEventBusRouter.ts index 485f1b0e58..4a332d5ce5 100644 --- a/plugins/events-backend/src/service/hub/createEventBusRouter.ts +++ b/plugins/events-backend/src/service/hub/createEventBusRouter.ts @@ -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("', '")}'`, diff --git a/plugins/events-backend/src/service/hub/types.ts b/plugins/events-backend/src/service/hub/types.ts index 0d363bec8b..316d9e561f 100644 --- a/plugins/events-backend/src/service/hub/types.ts +++ b/plugins/events-backend/src/service/hub/types.ts @@ -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; }): Promise<{ eventId: string } | undefined>; - upsertSubscription(subscriptionId: string, topics: string[]): Promise; + upsertSubscription( + subscriptionId: string, + topics: string[], + credentials: BackstageCredentials, + ): Promise; readSubscription(subscriptionId: string): Promise<{ events: EventParams[] }>;