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

feat(events/http): add HTTP endpoint-based event publisher

Browse Source 
This plugin adds an event publisher which receives events via (an) HTTP endpoint(s)
and can be used as destination at webhook subscriptions.

Relates-to: #11082
Signed-off-by: Patrick Jungermann <Patrick.Jungermann@gmail.com>
This commit is contained in:
Patrick Jungermann
2022-10-04 16:41:08 +02:00
parent 7bbd2403a1
commit dc9da28abd
24 changed files with 918 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/events-backend/src/index.ts
+1
View File
@@ -22,3 +22,4 @@
export { EventsBackend } from './service/EventsBackend';
export { eventsPlugin } from './service/EventsPlugin';
export { HttpPostIngressEventPublisher } from './service/http';
plugins/events-backend/src/service/EventsPlugin.test.ts
+36 -3
View File
@@ -14,9 +14,12 @@
* limitations under the License.
*/
import { getVoidLogger } from '@backstage/backend-common';
import { errorHandler, getVoidLogger } from '@backstage/backend-common';
import { ConfigReader } from '@backstage/config';
import {
configServiceRef,
createBackendModule,
httpRouterServiceRef,
loggerServiceRef,
} from '@backstage/backend-plugin-api';
import { startTestBackend } from '@backstage/backend-test-utils';
@@ -26,13 +29,29 @@ import {
TestEventPublisher,
TestEventSubscriber,
} from '@backstage/plugin-events-backend-test-utils';
import express from 'express';
import Router from 'express-promise-router';
import request from 'supertest';
import { eventsPlugin } from './EventsPlugin';
describe('eventPlugin', () => {
it('should be initialized properly', async () => {
const eventBroker = new TestEventBroker();
const publisher = new TestEventPublisher();
const subscriber = new TestEventSubscriber('sub', ['topicA']);
const subscriber = new TestEventSubscriber('sub', ['fake']);
const config = new ConfigReader({
events: {
http: {
topics: ['fake'],
},
},
});
const httpRouter = Router();
httpRouter.use(express.json());
httpRouter.use(errorHandler());
const app = express().use(httpRouter);
const testModule = createBackendModule({
pluginId: 'events',
@@ -53,12 +72,26 @@ describe('eventPlugin', () => {
await startTestBackend({
extensionPoints: [],
services: [[loggerServiceRef, getVoidLogger()]],
services: [
[configServiceRef, config],
[httpRouterServiceRef, httpRouter],
[loggerServiceRef, getVoidLogger()],
],
features: [eventsPlugin(), testModule()],
});
expect(publisher.eventBroker).toBe(eventBroker);
expect(eventBroker.subscribed.length).toEqual(1);
expect(eventBroker.subscribed[0]).toBe(subscriber);
const response = await request(app)
.post('/http/fake')
.timeout(100)
.send({ test: 'fake' });
expect(response.status).toBe(202);
expect(eventBroker.published.length).toEqual(1);
expect(eventBroker.published[0].topic).toEqual('fake');
expect(eventBroker.published[0].eventPayload).toEqual({ test: 'fake' });
});
});
plugins/events-backend/src/service/EventsPlugin.ts
+43 -5
View File
@@ -15,7 +15,9 @@
*/
import {
configServiceRef,
createBackendPlugin,
httpRouterServiceRef,
loggerServiceRef,
loggerToWinstonLogger,
} from '@backstage/backend-plugin-api';
@@ -25,11 +27,15 @@ import {
EventSubscriber,
eventsExtensionPoint,
EventsExtensionPoint,
HttpPostIngressOptions,
} from '@backstage/plugin-events-node';
import { InMemoryEventBroker } from './InMemoryEventBroker';
import Router from 'express-promise-router';
import { HttpPostIngressEventPublisher } from './http';
class EventsExtensionPointImpl implements EventsExtensionPoint {
#eventBroker: EventBroker | undefined;
#httpPostIngresses: HttpPostIngressOptions[] = [];
#publishers: EventPublisher[] = [];
#subscribers: EventSubscriber[] = [];
@@ -49,6 +55,10 @@ class EventsExtensionPointImpl implements EventsExtensionPoint {
this.#subscribers.push(...subscribers.flat());
}
addHttpPostIngress(options: HttpPostIngressOptions) {
this.#httpPostIngresses.push(options);
}
get eventBroker() {
return this.#eventBroker;
}
@@ -60,6 +70,10 @@ class EventsExtensionPointImpl implements EventsExtensionPoint {
get subscribers() {
return this.#subscribers;
}
get httpPostIngresses() {
return this.#httpPostIngresses;
}
}
/**
@@ -75,18 +89,42 @@ export const eventsPlugin = createBackendPlugin({
env.registerInit({
deps: {
config: configServiceRef,
httpRouter: httpRouterServiceRef,
logger: loggerServiceRef,
},
async init({ logger }) {
async init({ config, httpRouter, logger }) {
const winstonLogger = loggerToWinstonLogger(logger);
const eventsRouter = Router();
const router = Router();
eventsRouter.use('/http', router);
const ingresses = Object.fromEntries(
extensionPoint.httpPostIngresses.map(ingress => [
ingress.topic,
ingress as Omit<HttpPostIngressOptions, 'topic'>,
]),
);
const http = HttpPostIngressEventPublisher.fromConfig({
config,
logger: winstonLogger,
router,
ingresses,
});
if (!extensionPoint.eventBroker) {
const winstonLogger = loggerToWinstonLogger(logger);
extensionPoint.setEventBroker(new InMemoryEventBroker(winstonLogger));
}
extensionPoint.eventBroker!.subscribe(extensionPoint.subscribers);
extensionPoint.publishers.forEach(publisher =>
publisher.setEventBroker(extensionPoint.eventBroker!),
);
[extensionPoint.publishers, http]
.flat()
.forEach(publisher =>
publisher.setEventBroker(extensionPoint.eventBroker!),
);
httpRouter.use(eventsRouter);
},
});
},
plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.test.ts
+225
View File
@@ -0,0 +1,225 @@
/*
* 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 { errorHandler, getVoidLogger } from '@backstage/backend-common';
import { ConfigReader } from '@backstage/config';
import { TestEventBroker } from '@backstage/plugin-events-backend-test-utils';
import express from 'express';
import Router from 'express-promise-router';
import request from 'supertest';
import { HttpPostIngressEventPublisher } from './HttpPostIngressEventPublisher';
describe('HttpPostIngressEventPublisher', () => {
const logger = getVoidLogger();
it('should set up routes correctly', async () => {
const config = new ConfigReader({
events: {
http: {
topics: ['testA'],
},
},
});
const router = Router();
router.use(express.json());
router.use(errorHandler());
const app = express().use(router);
const publisher = HttpPostIngressEventPublisher.fromConfig({
config,
logger,
router,
ingresses: {
testB: {},
},
});
const eventBroker = new TestEventBroker();
await publisher.setEventBroker(eventBroker);
const notFoundResponse = await request(app)
.post('/unknown')
.timeout(100)
.send({ test: 'data' });
expect(notFoundResponse.status).toBe(404);
const response1 = await request(app)
.post('/testA')
.set('X-Custom-Header', 'test-value')
.timeout(100)
.send({ testA: 'data' });
expect(response1.status).toBe(202);
const response2 = await request(app)
.post('/testB')
.set('X-Custom-Header', 'test-value')
.timeout(100)
.send({ testB: 'data' });
expect(response2.status).toBe(202);
expect(eventBroker.published.length).toEqual(2);
expect(eventBroker.published[0].topic).toEqual('testA');
expect(eventBroker.published[0].eventPayload).toEqual({ testA: 'data' });
expect(eventBroker.published[0].metadata).toEqual(
expect.objectContaining({
'content-type': 'application/json',
'x-custom-header': 'test-value',
}),
);
expect(eventBroker.published[1].topic).toEqual('testB');
expect(eventBroker.published[1].eventPayload).toEqual({ testB: 'data' });
expect(eventBroker.published[1].metadata).toEqual(
expect.objectContaining({
'content-type': 'application/json',
'x-custom-header': 'test-value',
}),
);
});
it('with validator', async () => {
const config = new ConfigReader({
events: {
http: {
topics: ['testA'],
},
},
});
const router = Router();
router.use(express.json());
router.use(errorHandler());
const app = express().use(router);
const publisher = HttpPostIngressEventPublisher.fromConfig({
config,
logger,
router,
ingresses: {
testB: {
validator: async (req, context) => {
if (req.headers['x-test-signature'] === 'testB-signature') {
return;
}
context.reject({
status: 400,
payload: {
message: 'wrong signature',
},
});
},
},
testC: {
validator: async (req, context) => {
if (req.headers['x-test-signature'] === 'testC-signature') {
return;
}
context.reject({
status: 404,
// payload: {},
});
},
},
testD: {
validator: async (req, context) => {
if (req.headers['x-test-signature'] === 'testD-signature') {
return;
}
context.reject({
// status: 403,
// payload: {},
});
},
},
},
});
const eventBroker = new TestEventBroker();
await publisher.setEventBroker(eventBroker);
const response1 = await request(app)
.post('/testA')
.timeout(100)
.send({ test: 'data' });
expect(response1.status).toBe(202);
const response2 = await request(app)
.post('/testB')
.timeout(100)
.send({ test: 'data' });
expect(response2.status).toBe(400);
expect(response2.body).toEqual({ message: 'wrong signature' });
const response3 = await request(app)
.post('/testB')
.set('X-Test-Signature', 'wrong')
.timeout(100)
.send({ test: 'data' });
expect(response3.status).toBe(400);
expect(response3.body).toEqual({ message: 'wrong signature' });
const response4 = await request(app)
.post('/testB')
.set('X-Test-Signature', 'testB-signature')
.timeout(100)
.send({ test: 'data' });
expect(response4.status).toBe(202);
const response5 = await request(app)
.post('/testC')
.timeout(100)
.send({ test: 'data' });
expect(response5.status).toBe(404);
expect(response5.body).toEqual({});
const response6 = await request(app)
.post('/testD')
.timeout(100)
.send({ test: 'data' });
expect(response6.status).toBe(403);
expect(response6.body).toEqual({});
expect(eventBroker.published.length).toEqual(2);
expect(eventBroker.published[0].topic).toEqual('testA');
expect(eventBroker.published[0].eventPayload).toEqual({ test: 'data' });
expect(eventBroker.published[1].topic).toEqual('testB');
expect(eventBroker.published[1].eventPayload).toEqual({ test: 'data' });
expect(eventBroker.published[1].metadata).toEqual(
expect.objectContaining({
'x-test-signature': 'testB-signature',
}),
);
});
it('without configuration', async () => {
const config = new ConfigReader({});
const router = Router();
router.use(express.json());
router.use(errorHandler());
expect(() =>
HttpPostIngressEventPublisher.fromConfig({
config,
logger,
router,
}),
).not.toThrow();
});
});
plugins/events-backend/src/service/http/HttpPostIngressEventPublisher.ts
+118
View File
@@ -0,0 +1,118 @@
/*
* 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 { errorHandler } from '@backstage/backend-common';
import { Config } from '@backstage/config';
import {
EventBroker,
EventPublisher,
HttpPostIngressOptions,
RequestValidator,
} from '@backstage/plugin-events-node';
import express from 'express';
import Router from 'express-promise-router';
import { Logger } from 'winston';
import { RequestValidationContextImpl } from './validation';
/**
* Publishes events received from their origin (e.g., webhook events from an SCM system)
* via HTTP POST endpoint and passes the request body as event payload to the registered subscribers.
*
* @public
*/
// TODO(pjungermann): add prom metrics? (see plugins/catalog-backend/src/util/metrics.ts, etc.)
export class HttpPostIngressEventPublisher implements EventPublisher {
private eventBroker?: EventBroker;
static fromConfig(env: {
config: Config;
ingresses?: { [topic: string]: Omit<HttpPostIngressOptions, 'topic'> };
logger: Logger;
router: express.Router;
}): HttpPostIngressEventPublisher {
const topics =
env.config.getOptionalStringArray('events.http.topics') ?? [];
const ingresses = env.ingresses ?? {};
topics.forEach(topic => {
// don't overwrite topic settings
// (e.g., added at the config as well as argument)
if (!ingresses[topic]) {
ingresses[topic] = {};
}
});
return new HttpPostIngressEventPublisher(env.logger, env.router, ingresses);
}
private constructor(
private logger: Logger,
router: express.Router,
ingresses: { [topic: string]: Omit<HttpPostIngressOptions, 'topic'> },
) {
router.use(this.createRouter(ingresses));
}
async setEventBroker(eventBroker: EventBroker): Promise<void> {
this.eventBroker = eventBroker;
}
private createRouter(ingresses: {
[topic: string]: Omit<HttpPostIngressOptions, 'topic'>;
}): express.Router {
const router = Router();
router.use(express.json());
Object.keys(ingresses).forEach(topic =>
this.addRouteForTopic(router, topic, ingresses[topic].validator),
);
router.use(errorHandler());
return router;
}
private addRouteForTopic(
router: express.Router,
topic: string,
validator?: RequestValidator,
): void {
const path = `/${topic}`;
router.post(path, async (request, response) => {
const context = new RequestValidationContextImpl();
await validator?.(request, context);
if (context.wasRejected()) {
response
.status(context.rejectionDetails!.status)
.json(context.rejectionDetails!.payload);
return;
}
const eventPayload = request.body;
await this.eventBroker!.publish({
topic,
eventPayload,
metadata: request.headers,
});
response.status(202).json({ status: 'accepted' });
});
// TODO(pjungermann): We don't really know the externally defined path prefix here,
// however it is more useful for users to have it. Is there a better way?
this.logger.info(`Registered /api/events/http${path} to receive events`);
}
}
plugins/events-backend/src/service/http/index.ts
+18
View File
@@ -0,0 +1,18 @@
/*
* Copyright 2020 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 { HttpPostIngressEventPublisher } from './HttpPostIngressEventPublisher';
export * from './validation';
plugins/events-backend/src/service/http/validation/RequestValidationContextImpl.test.ts
+64
View File
@@ -0,0 +1,64 @@
/*
* 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 { RequestValidationContextImpl } from './RequestValidationContextImpl';
describe('RequestValidationContextImpl', () => {
it('not rejected', () => {
const context = new RequestValidationContextImpl();
expect(context.wasRejected()).toBe(false);
expect(context.rejectionDetails).toBeUndefined();
});
it('reject without details', () => {
const context = new RequestValidationContextImpl();
context.reject();
expect(context.wasRejected()).toBe(true);
expect(context.rejectionDetails).not.toBeUndefined();
expect(context.rejectionDetails!.status).toBe(403);
expect(context.rejectionDetails!.payload).toEqual({});
});
it('reject with partial details', () => {
const context = new RequestValidationContextImpl();
context.reject({ status: 404 });
expect(context.wasRejected()).toBe(true);
expect(context.rejectionDetails).not.toBeUndefined();
expect(context.rejectionDetails!.status).toBe(404);
expect(context.rejectionDetails!.payload).toEqual({});
});
it('reject with details', () => {
const context = new RequestValidationContextImpl();
context.reject({
status: 403,
payload: { message: 'invalid signature' },
});
expect(context.wasRejected()).toBe(true);
expect(context.rejectionDetails).not.toBeUndefined();
expect(context.rejectionDetails!.status).toBe(403);
expect(context.rejectionDetails!.payload).toEqual({
message: 'invalid signature',
});
});
});
plugins/events-backend/src/service/http/validation/RequestValidationContextImpl.ts
+39
View File
@@ -0,0 +1,39 @@
/*
* 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 {
RequestRejectionDetails,
RequestValidationContext,
} from '@backstage/plugin-events-node';
export class RequestValidationContextImpl implements RequestValidationContext {
#rejectionDetails: RequestRejectionDetails | undefined;
reject(details?: Partial<RequestRejectionDetails>): void {
this.#rejectionDetails = {
status: details?.status ?? 403,
payload: details?.payload ?? {},
};
}
wasRejected(): boolean {
return this.#rejectionDetails !== undefined;
}
get rejectionDetails() {
return this.#rejectionDetails;
}
}
plugins/events-backend/src/service/http/validation/index.ts
+17
View File
@@ -0,0 +1,17 @@
/*
* Copyright 2020 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 { RequestValidationContextImpl } from './RequestValidationContextImpl';