feat(events/sqs): add a new AWS SQS event publisher
This change introduces a new plugin `@backstage/plugin-events-backend-module-aws-sqs`. This plugin provides an event publisher which receives events from (an) AWS SQS queue(s) and publishes them to the event broker. The plugin supports the new backend-plugin-api and connects with the other plugins. Signed-off-by: Patrick Jungermann <Patrick.Jungermann@gmail.com>
This commit is contained in:
Patrick Jungermann
@@ -0,0 +1 @@
|
||||
module.exports = require('@backstage/cli/config/eslint-factory')(__dirname);
|
||||
@@ -0,0 +1,42 @@
|
||||
# events-backend-module-aws-sqs
|
||||
|
||||
Welcome to the `events-backend-module-aws-sqs` backend plugin!
|
||||
|
||||
This plugin is a module for the `events-backend` backend plugin
|
||||
and extends it with an `AwsSqsConsumingEventPublisher`.
|
||||
|
||||
This event publisher will allow you to receive events from
|
||||
an AWS SQS queue and will publish these to the used event broker.
|
||||
|
||||
## Configuration
|
||||
|
||||
The polled AWS SQS queues depend on your configuration:
|
||||
|
||||
```yaml
|
||||
events:
|
||||
modules:
|
||||
awsSqs:
|
||||
awsSqsConsumingEventPublisher:
|
||||
topics:
|
||||
topicName1: # replace with actual topic name as expected by subscribers
|
||||
queue:
|
||||
url: 'https://sqs.us-east-2.amazonaws.com/123456789012/MyQueue'
|
||||
region: us-east-2
|
||||
# visibilityTimeout - as HumanDuration; defaults to queue-based config
|
||||
# waitTime - as HumanDuration; defaults to max of 20 seconds (long polling)
|
||||
# timeout - as HumanDuration; timeout for the task execution
|
||||
# waitTimeAfterEmptyReceive - as HumanDuration; time to wait before a retry when there was no message.
|
||||
topicName2:
|
||||
# [...]
|
||||
```
|
||||
|
||||
## Installation
|
||||
|
||||
1. Install the [`events-backend` plugin](../events-backend/README.md).
|
||||
2. Install this module
|
||||
3. Add your configuration.
|
||||
|
||||
```bash
|
||||
# From your Backstage root directory
|
||||
yarn add --cwd packages/backend @backstage/plugin-events-backend-module-aws-sqs
|
||||
```
|
||||
@@ -0,0 +1,29 @@
|
||||
## API Report File for "@backstage/plugin-events-backend-module-aws-sqs"
|
||||
|
||||
> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
|
||||
|
||||
```ts
|
||||
import { BackendFeature } from '@backstage/backend-plugin-api';
|
||||
import { Config } from '@backstage/config';
|
||||
import { EventBroker } from '@backstage/plugin-events-node';
|
||||
import { EventPublisher } from '@backstage/plugin-events-node';
|
||||
import { Logger } from 'winston';
|
||||
import { PluginTaskScheduler } from '@backstage/backend-tasks';
|
||||
|
||||
// @public
|
||||
export class AwsSqsConsumingEventPublisher implements EventPublisher {
|
||||
// (undocumented)
|
||||
static fromConfig(env: {
|
||||
config: Config;
|
||||
logger: Logger;
|
||||
scheduler: PluginTaskScheduler;
|
||||
}): AwsSqsConsumingEventPublisher[];
|
||||
// (undocumented)
|
||||
setEventBroker(eventBroker: EventBroker): Promise<void>;
|
||||
}
|
||||
|
||||
// @alpha
|
||||
export const awsSqsConsumingEventPublisherEventsModule: (
|
||||
options?: undefined,
|
||||
) => BackendFeature;
|
||||
```
|
||||
@@ -0,0 +1,78 @@
|
||||
/*
|
||||
* 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 { HumanDuration } from '@backstage/types';
|
||||
|
||||
export interface Config {
|
||||
events?: {
|
||||
modules?: {
|
||||
/**
|
||||
* events-backend-module-aws-sqs plugin configuration.
|
||||
*/
|
||||
awsSqs?: {
|
||||
/**
|
||||
* Configuration for AwsSqsConsumingEventPublisher.
|
||||
*/
|
||||
awsSqsConsumingEventPublisher?: {
|
||||
/**
|
||||
* Contains a record per topic for which an AWS SQS queue
|
||||
* should be used as source of events.
|
||||
*/
|
||||
topics: Record<
|
||||
string,
|
||||
{
|
||||
/**
|
||||
* (Required) Queue-related configuration.
|
||||
*/
|
||||
queue: {
|
||||
/**
|
||||
* (Required) The region of the AWS SQS queue.
|
||||
*/
|
||||
region: string;
|
||||
/**
|
||||
* (Required) The absolute URL for the AWS SQS queue to be used.
|
||||
*/
|
||||
url: string;
|
||||
/**
|
||||
* (Optional) Visibility timeout for messages in flight.
|
||||
*/
|
||||
visibilityTimeout: HumanDuration;
|
||||
/**
|
||||
* (Optional) Wait time when polling for available messages.
|
||||
* Default: 20 seconds.
|
||||
*/
|
||||
waitTime: HumanDuration;
|
||||
};
|
||||
/**
|
||||
* (Optional) Timeout for the task execution which includes polling for messages
|
||||
* and publishing the events to the event broker
|
||||
* and the wait time after empty receives.
|
||||
*
|
||||
* Must be greater than `queue.waitTime` + `waitTimeAfterEmptyReceive`.
|
||||
*/
|
||||
timeout: HumanDuration;
|
||||
/**
|
||||
* (Optional) Wait time before polling again if no message was received.
|
||||
* Default: 1 minute.
|
||||
*/
|
||||
waitTimeAfterEmptyReceive: HumanDuration;
|
||||
}
|
||||
>;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
{
|
||||
"name": "@backstage/plugin-events-backend-module-aws-sqs",
|
||||
"version": "0.0.0",
|
||||
"main": "src/index.ts",
|
||||
"types": "src/index.ts",
|
||||
"license": "Apache-2.0",
|
||||
"publishConfig": {
|
||||
"access": "public",
|
||||
"alphaTypes": "dist/index.alpha.d.ts",
|
||||
"main": "dist/index.cjs.js",
|
||||
"types": "dist/index.d.ts"
|
||||
},
|
||||
"backstage": {
|
||||
"role": "backend-plugin-module"
|
||||
},
|
||||
"scripts": {
|
||||
"start": "backstage-cli package start",
|
||||
"build": "backstage-cli package build --experimental-type-build",
|
||||
"lint": "backstage-cli package lint",
|
||||
"test": "backstage-cli package test",
|
||||
"clean": "backstage-cli package clean",
|
||||
"prepack": "backstage-cli package prepack",
|
||||
"postpack": "backstage-cli package postpack"
|
||||
},
|
||||
"dependencies": {
|
||||
"@aws-sdk/client-sqs": "^3.0.0",
|
||||
"@backstage/backend-plugin-api": "workspace:^",
|
||||
"@backstage/backend-tasks": "workspace:^",
|
||||
"@backstage/config": "workspace:^",
|
||||
"@backstage/plugin-events-node": "workspace:^",
|
||||
"@backstage/types": "workspace:^",
|
||||
"luxon": "^3.0.0",
|
||||
"winston": "^3.2.1"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@backstage/backend-common": "workspace:^",
|
||||
"@backstage/backend-test-utils": "workspace:^",
|
||||
"@backstage/cli": "workspace:^",
|
||||
"@backstage/plugin-events-backend-test-utils": "workspace:^",
|
||||
"aws-sdk-client-mock": "^2.0.0"
|
||||
},
|
||||
"files": [
|
||||
"alpha",
|
||||
"config.d.ts",
|
||||
"dist"
|
||||
],
|
||||
"configSchema": "config.d.ts"
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
/**
|
||||
* The module "sqs" for the Backstage backend plugin "events"
|
||||
* adding an AWS SQS-based publisher,
|
||||
* receiving events from an AWS SQS queue and passing it to the
|
||||
* internal event broker.
|
||||
*
|
||||
* @packageDocumentation
|
||||
*/
|
||||
|
||||
export { AwsSqsConsumingEventPublisher } from './publisher/AwsSqsConsumingEventPublisher';
|
||||
export { awsSqsConsumingEventPublisherEventsModule } from './service/AwsSqsConsumingEventPublisherEventsModule';
|
||||
+227
@@ -0,0 +1,227 @@
|
||||
/*
|
||||
* 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 {
|
||||
DeleteMessageBatchCommand,
|
||||
ReceiveMessageCommand,
|
||||
SQSClient,
|
||||
} from '@aws-sdk/client-sqs';
|
||||
import { getVoidLogger } from '@backstage/backend-common';
|
||||
import { PluginTaskScheduler } from '@backstage/backend-tasks';
|
||||
import { ConfigReader } from '@backstage/config';
|
||||
import { TestEventBroker } from '@backstage/plugin-events-backend-test-utils';
|
||||
import { mockClient } from 'aws-sdk-client-mock';
|
||||
import { AwsSqsConsumingEventPublisher } from './AwsSqsConsumingEventPublisher';
|
||||
|
||||
describe('AwsSqsConsumingEventPublisher', () => {
|
||||
it('creates one publisher instance per configured topic', async () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
},
|
||||
},
|
||||
fake2: {
|
||||
queue: {
|
||||
region: 'us-east-1',
|
||||
url: 'https://fake2.queue.url',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
const logger = getVoidLogger();
|
||||
const scheduler = {
|
||||
scheduleTask: jest.fn(),
|
||||
} as unknown as PluginTaskScheduler;
|
||||
|
||||
const publishers = AwsSqsConsumingEventPublisher.fromConfig({
|
||||
config,
|
||||
logger,
|
||||
scheduler,
|
||||
});
|
||||
expect(publishers.length).toEqual(2);
|
||||
});
|
||||
|
||||
it('polling will be scheduled after connecting to the EventBroker', async () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
const logger = getVoidLogger();
|
||||
const scheduler = {
|
||||
scheduleTask: jest.fn(),
|
||||
} as unknown as PluginTaskScheduler;
|
||||
|
||||
const publishers = AwsSqsConsumingEventPublisher.fromConfig({
|
||||
config,
|
||||
logger,
|
||||
scheduler,
|
||||
});
|
||||
expect(publishers.length).toEqual(1);
|
||||
|
||||
const publisher = publishers[0];
|
||||
|
||||
const eventBroker = new TestEventBroker();
|
||||
await publisher.setEventBroker(eventBroker);
|
||||
|
||||
// publisher.connect(..) was causing the polling for events to be scheduled
|
||||
expect(scheduler.scheduleTask).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
id: 'events.awsSqs.publisher:fake1',
|
||||
frequency: { seconds: 0 },
|
||||
timeout: { seconds: 260 },
|
||||
scope: 'local',
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('publishes events for received messages and deletes them in bulk', async () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
},
|
||||
waitTimeAfterEmptyReceive: { seconds: 1 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
const logger = getVoidLogger();
|
||||
let taskFn: (() => Promise<void>) | undefined = undefined;
|
||||
const scheduler = {
|
||||
scheduleTask: (spec: { fn: () => Promise<void> }) => {
|
||||
taskFn = spec.fn;
|
||||
},
|
||||
} as unknown as PluginTaskScheduler;
|
||||
|
||||
// on the first attempt, we will return 1 message and 0 messages afterwards
|
||||
const sqsMock = mockClient(SQSClient);
|
||||
sqsMock
|
||||
.on(ReceiveMessageCommand, {
|
||||
MaxNumberOfMessages: 10,
|
||||
QueueUrl: 'https://fake1.queue.url',
|
||||
WaitTimeSeconds: 20,
|
||||
})
|
||||
.resolvesOnce({
|
||||
Messages: [],
|
||||
})
|
||||
.resolvesOnce({
|
||||
Messages: [
|
||||
{
|
||||
Body: '{"event":"payload1"}',
|
||||
ReceiptHandle: 'fake-handle1',
|
||||
MessageAttributes: {
|
||||
'X-Custom-Attr': {
|
||||
DataType: 'String',
|
||||
StringValue: 'value',
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
Body: '{"event":"payload2"}',
|
||||
ReceiptHandle: 'fake-handle2',
|
||||
},
|
||||
],
|
||||
})
|
||||
.on(DeleteMessageBatchCommand, {
|
||||
Entries: [
|
||||
{
|
||||
Id: 'message-0',
|
||||
ReceiptHandle: 'fake-handle1',
|
||||
},
|
||||
{
|
||||
Id: 'message-1',
|
||||
ReceiptHandle: 'fake-handle2',
|
||||
},
|
||||
],
|
||||
QueueUrl: 'https://fake1.queue.url',
|
||||
})
|
||||
.resolvesOnce({
|
||||
Failed: [
|
||||
{
|
||||
Id: 'message-1',
|
||||
Message: 'test failure',
|
||||
SenderFault: true,
|
||||
Code: '400',
|
||||
},
|
||||
],
|
||||
Successful: [{ Id: 'message-0' }],
|
||||
});
|
||||
|
||||
const publishers = AwsSqsConsumingEventPublisher.fromConfig({
|
||||
config,
|
||||
logger,
|
||||
scheduler,
|
||||
});
|
||||
expect(publishers.length).toEqual(1);
|
||||
const publisher = publishers[0];
|
||||
|
||||
const eventBroker = new TestEventBroker();
|
||||
await publisher.setEventBroker(eventBroker);
|
||||
|
||||
await taskFn!();
|
||||
await taskFn!();
|
||||
await taskFn!();
|
||||
|
||||
expect(eventBroker.published.length).toEqual(2);
|
||||
expect(eventBroker.published[0].topic).toEqual('fake1');
|
||||
expect(eventBroker.published[0].eventPayload).toEqual({
|
||||
event: 'payload1',
|
||||
});
|
||||
expect(eventBroker.published[0].metadata).toEqual({
|
||||
'X-Custom-Attr': 'value',
|
||||
});
|
||||
|
||||
expect(eventBroker.published[1].topic).toEqual('fake1');
|
||||
expect(eventBroker.published[1].eventPayload).toEqual({
|
||||
event: 'payload2',
|
||||
});
|
||||
expect(eventBroker.published[1].metadata).toEqual({});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,191 @@
|
||||
/*
|
||||
* 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 {
|
||||
DeleteMessageBatchCommand,
|
||||
Message,
|
||||
ReceiveMessageCommand,
|
||||
ReceiveMessageCommandInput,
|
||||
SQSClient,
|
||||
} from '@aws-sdk/client-sqs';
|
||||
import { PluginTaskScheduler } from '@backstage/backend-tasks';
|
||||
import { Config } from '@backstage/config';
|
||||
import { EventBroker, EventPublisher } from '@backstage/plugin-events-node';
|
||||
import { Logger } from 'winston';
|
||||
import { AwsSqsEventSourceConfig, readConfig } from './config';
|
||||
|
||||
/**
|
||||
* Publishes events received from an AWS SQS queue.
|
||||
* The message payload will be used as event payload and passed to registered subscribers.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
// TODO(pjungermann): add prom metrics? (see plugins/catalog-backend/src/util/metrics.ts, etc.)
|
||||
export class AwsSqsConsumingEventPublisher implements EventPublisher {
|
||||
private readonly topic: string;
|
||||
private readonly receiveParams: ReceiveMessageCommandInput;
|
||||
private readonly sqs: SQSClient;
|
||||
private readonly queueUrl: string;
|
||||
private readonly taskTimeoutSeconds: number;
|
||||
private readonly waitTimeAfterEmptyReceiveMs;
|
||||
private eventBroker?: EventBroker;
|
||||
|
||||
static fromConfig(env: {
|
||||
config: Config;
|
||||
logger: Logger;
|
||||
scheduler: PluginTaskScheduler;
|
||||
}): AwsSqsConsumingEventPublisher[] {
|
||||
return readConfig(env.config).map(
|
||||
config =>
|
||||
new AwsSqsConsumingEventPublisher(env.logger, env.scheduler, config),
|
||||
);
|
||||
}
|
||||
|
||||
private constructor(
|
||||
private readonly logger: Logger,
|
||||
private readonly scheduler: PluginTaskScheduler,
|
||||
config: AwsSqsEventSourceConfig,
|
||||
) {
|
||||
this.topic = config.topic;
|
||||
|
||||
this.receiveParams = {
|
||||
MaxNumberOfMessages: 10,
|
||||
MessageAttributeNames: ['All'],
|
||||
QueueUrl: config.queueUrl,
|
||||
VisibilityTimeout: config.visibilityTimeout?.as('seconds'),
|
||||
WaitTimeSeconds: config.pollingWaitTime.as('seconds'),
|
||||
};
|
||||
this.sqs = new SQSClient({ region: config.region });
|
||||
this.queueUrl = config.queueUrl;
|
||||
|
||||
this.taskTimeoutSeconds = config.timeout.as('seconds');
|
||||
this.waitTimeAfterEmptyReceiveMs =
|
||||
config.waitTimeAfterEmptyReceive.as('milliseconds');
|
||||
}
|
||||
|
||||
async setEventBroker(eventBroker: EventBroker): Promise<void> {
|
||||
this.eventBroker = eventBroker;
|
||||
return this.start();
|
||||
}
|
||||
|
||||
private async start(): Promise<void> {
|
||||
const id = `events.awsSqs.publisher:${this.topic}`;
|
||||
const logger = this.logger.child({
|
||||
class: AwsSqsConsumingEventPublisher.prototype.constructor.name,
|
||||
taskId: id,
|
||||
});
|
||||
|
||||
await this.scheduler.scheduleTask({
|
||||
id: id,
|
||||
frequency: { seconds: 0 },
|
||||
timeout: { seconds: this.taskTimeoutSeconds },
|
||||
scope: 'local',
|
||||
fn: async () => {
|
||||
try {
|
||||
const numMessages = await this.consumeMessages();
|
||||
if (numMessages === 0) {
|
||||
await this.sleep(this.waitTimeAfterEmptyReceiveMs);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(error);
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private async deleteMessages(messages?: Message[]): Promise<void> {
|
||||
if (!messages) {
|
||||
return;
|
||||
}
|
||||
|
||||
const deleteParams = {
|
||||
QueueUrl: this.queueUrl,
|
||||
Entries: messages.map((message, index) => {
|
||||
return {
|
||||
Id: message.MessageId ?? `message-${index}`,
|
||||
ReceiptHandle: message.ReceiptHandle,
|
||||
};
|
||||
}),
|
||||
};
|
||||
|
||||
try {
|
||||
const result = await this.sqs.send(
|
||||
new DeleteMessageBatchCommand(deleteParams),
|
||||
);
|
||||
if (result.Failed) {
|
||||
this.logger.error(
|
||||
`Failed to delete ${result.Failed!.length} of ${
|
||||
messages.length
|
||||
} messages from AWS SQS ${this.queueUrl}. First: ${
|
||||
result.Failed[0].Message
|
||||
}`,
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(
|
||||
`Failed to delete message from AWS SQS ${this.queueUrl}`,
|
||||
error,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private async consumeMessages(): Promise<number> {
|
||||
try {
|
||||
const data = await this.sqs.send(
|
||||
new ReceiveMessageCommand(this.receiveParams),
|
||||
);
|
||||
|
||||
data.Messages?.forEach(message => {
|
||||
const eventPayload = JSON.parse(message.Body!);
|
||||
|
||||
const metadata: Record<string, string | string[]> = {};
|
||||
Object.keys(message.MessageAttributes ?? {}).forEach(key => {
|
||||
const attrValue = message.MessageAttributes![key];
|
||||
if (
|
||||
!attrValue ||
|
||||
!attrValue.DataType ||
|
||||
!['String', 'Number'].includes(attrValue.DataType)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
const value = attrValue.StringListValues ?? attrValue.StringValue;
|
||||
if (value !== undefined) {
|
||||
metadata[key] = value;
|
||||
}
|
||||
});
|
||||
|
||||
this.eventBroker!.publish({
|
||||
topic: this.topic,
|
||||
eventPayload,
|
||||
metadata,
|
||||
});
|
||||
});
|
||||
await this.deleteMessages(data.Messages);
|
||||
return data.Messages?.length ?? 0;
|
||||
} catch (error) {
|
||||
this.logger.error(
|
||||
`Failed to receive events from AWS SQS ${this.queueUrl}`,
|
||||
error,
|
||||
);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
private sleep(ms: number): Promise<void> {
|
||||
return new Promise<void>(resolve => setTimeout(resolve, ms));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,222 @@
|
||||
/*
|
||||
* 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 { ConfigReader } from '@backstage/config';
|
||||
import { readConfig } from './config';
|
||||
|
||||
describe('readConfig', () => {
|
||||
it('not configured', () => {
|
||||
const config = new ConfigReader({});
|
||||
|
||||
const publisherConfigs = readConfig(config);
|
||||
|
||||
expect(publisherConfigs.length).toBe(0);
|
||||
});
|
||||
|
||||
it('only required fields configured', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
},
|
||||
},
|
||||
fake2: {
|
||||
queue: {
|
||||
region: 'us-east-1',
|
||||
url: 'https://fake2.queue.url',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readConfig(config);
|
||||
|
||||
expect(publisherConfigs.length).toBe(2);
|
||||
|
||||
expect(publisherConfigs[0].topic).toEqual('fake1');
|
||||
expect(publisherConfigs[0].region).toEqual('eu-west-1');
|
||||
expect(publisherConfigs[0].queueUrl).toEqual('https://fake1.queue.url');
|
||||
expect(publisherConfigs[0].pollingWaitTime.as('seconds')).toBe(20);
|
||||
expect(publisherConfigs[0].timeout.as('seconds')).toBe(260);
|
||||
expect(publisherConfigs[0].waitTimeAfterEmptyReceive.as('seconds')).toBe(
|
||||
60,
|
||||
);
|
||||
});
|
||||
|
||||
it('all fields configured', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
visibilityTimeout: { minutes: 5 },
|
||||
waitTime: { seconds: 10 },
|
||||
},
|
||||
timeout: { minutes: 5 },
|
||||
waitTimeAfterEmptyReceive: { seconds: 30 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const publisherConfigs = readConfig(config);
|
||||
|
||||
expect(publisherConfigs.length).toBe(1);
|
||||
|
||||
expect(publisherConfigs[0].topic).toEqual('fake1');
|
||||
expect(publisherConfigs[0].region).toEqual('eu-west-1');
|
||||
expect(publisherConfigs[0].queueUrl).toEqual('https://fake1.queue.url');
|
||||
expect(publisherConfigs[0].pollingWaitTime.as('seconds')).toBe(10);
|
||||
expect(publisherConfigs[0].timeout.as('seconds')).toBe(300);
|
||||
expect(publisherConfigs[0].waitTimeAfterEmptyReceive.as('seconds')).toBe(
|
||||
30,
|
||||
);
|
||||
});
|
||||
|
||||
it('fail on negative queue.waitTime', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
visibilityTimeout: { minutes: 5 },
|
||||
waitTime: { seconds: -10 },
|
||||
},
|
||||
timeout: { minutes: 5 },
|
||||
waitTimeAfterEmptyReceive: { seconds: 30 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(() => readConfig(config)).toThrow(
|
||||
'events.modules.awsSqs.awsSqsConsumingEventPublisher.topics.fake1.queue.waitTime must be within 0..20 seconds',
|
||||
);
|
||||
});
|
||||
|
||||
it('fail on too high queue.waitTime', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
visibilityTimeout: { minutes: 5 },
|
||||
waitTime: { seconds: 30 },
|
||||
},
|
||||
timeout: { minutes: 5 },
|
||||
waitTimeAfterEmptyReceive: { seconds: 30 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(() => readConfig(config)).toThrow(
|
||||
'events.modules.awsSqs.awsSqsConsumingEventPublisher.topics.fake1.queue.waitTime must be within 0..20 seconds',
|
||||
);
|
||||
});
|
||||
|
||||
it('fail on too low timeout', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
visibilityTimeout: { minutes: 5 },
|
||||
waitTime: { seconds: 10 },
|
||||
},
|
||||
timeout: { seconds: 10 },
|
||||
waitTimeAfterEmptyReceive: { seconds: 30 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(() => readConfig(config)).toThrow(
|
||||
'The events.modules.awsSqs.awsSqsConsumingEventPublisher.topics.fake1.timeout must be greater than events.modules.awsSqs.awsSqsConsumingEventPublisher.topics.fake1.queue.waitTime',
|
||||
);
|
||||
});
|
||||
|
||||
it('fail on negative waitTimeAfterEmptyReceive', () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
visibilityTimeout: { minutes: 5 },
|
||||
waitTime: { seconds: 10 },
|
||||
},
|
||||
timeout: { minutes: 5 },
|
||||
waitTimeAfterEmptyReceive: { seconds: -30 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(() => readConfig(config)).toThrow(
|
||||
'The events.modules.awsSqs.awsSqsConsumingEventPublisher.topics.fake1.waitTimeAfterEmptyReceive must not be negative',
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,117 @@
|
||||
/*
|
||||
* 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 { Config } from '@backstage/config';
|
||||
import { HumanDuration, JsonObject } from '@backstage/types';
|
||||
import { Duration } from 'luxon';
|
||||
|
||||
const CONFIG_PREFIX_MODULE = 'events.modules.awsSqs.';
|
||||
const CONFIG_PREFIX_PUBLISHER = `${CONFIG_PREFIX_MODULE}awsSqsConsumingEventPublisher.`;
|
||||
const DEFAULT_WAIT_TIME_AFTER_EMPTY_RECEIVE = { minutes: 1 };
|
||||
const MAX_WAIT_SECONDS = 20;
|
||||
|
||||
export interface AwsSqsEventSourceConfig {
|
||||
pollingWaitTime: Duration;
|
||||
queueUrl: string;
|
||||
region: string;
|
||||
timeout: Duration;
|
||||
topic: string;
|
||||
visibilityTimeout?: Duration;
|
||||
waitTimeAfterEmptyReceive: Duration;
|
||||
}
|
||||
|
||||
// TODO(pjungermann): validation could be improved similar to `convertToHumanDuration` at @backstage/backend-tasks
|
||||
function readOptionalHumanDuration(
|
||||
config: Config,
|
||||
key: string,
|
||||
): HumanDuration | undefined {
|
||||
return config.getOptional<JsonObject>(key) as HumanDuration;
|
||||
}
|
||||
|
||||
function readOptionalDuration(
|
||||
config: Config,
|
||||
key: string,
|
||||
): Duration | undefined {
|
||||
const duration = readOptionalHumanDuration(config, key);
|
||||
return duration ? Duration.fromObject(duration) : undefined;
|
||||
}
|
||||
|
||||
export function readConfig(config: Config): AwsSqsEventSourceConfig[] {
|
||||
const key = `${CONFIG_PREFIX_PUBLISHER}topics`;
|
||||
const topics = config.getOptionalConfig(key);
|
||||
|
||||
return (
|
||||
topics?.keys()?.map(topic => {
|
||||
const topicConfig = topics.getConfig(topic);
|
||||
const keyPrefix = `${key}.${topic}.`;
|
||||
|
||||
// queue config:
|
||||
const pollingWaitTime = Duration.fromObject(
|
||||
readOptionalHumanDuration(topicConfig, 'queue.waitTime') ?? {
|
||||
seconds: MAX_WAIT_SECONDS,
|
||||
},
|
||||
);
|
||||
if (
|
||||
pollingWaitTime.valueOf() < 0 ||
|
||||
pollingWaitTime.as('seconds') > MAX_WAIT_SECONDS
|
||||
) {
|
||||
throw new Error(
|
||||
`${keyPrefix}queue.waitTime must be within 0..${MAX_WAIT_SECONDS} seconds.`,
|
||||
);
|
||||
}
|
||||
const queueUrl = topicConfig.getString('queue.url');
|
||||
const region = topicConfig.getString('queue.region');
|
||||
const visibilityTimeout = readOptionalDuration(
|
||||
topicConfig,
|
||||
'queue.visibilityTimeout',
|
||||
);
|
||||
|
||||
// task:
|
||||
const waitTimeAfterEmptyReceive = Duration.fromObject(
|
||||
readOptionalHumanDuration(topicConfig, 'waitTimeAfterEmptyReceive') ??
|
||||
DEFAULT_WAIT_TIME_AFTER_EMPTY_RECEIVE,
|
||||
);
|
||||
if (waitTimeAfterEmptyReceive.valueOf() < 0) {
|
||||
throw new Error(
|
||||
`The ${keyPrefix}waitTimeAfterEmptyReceive must not be negative.`,
|
||||
);
|
||||
}
|
||||
const timeout =
|
||||
readOptionalDuration(topicConfig, 'timeout') ??
|
||||
pollingWaitTime
|
||||
.plus(waitTimeAfterEmptyReceive)
|
||||
.plus(Duration.fromObject({ seconds: 180 }));
|
||||
if (
|
||||
timeout.valueOf() <=
|
||||
pollingWaitTime.valueOf() + waitTimeAfterEmptyReceive.valueOf()
|
||||
) {
|
||||
throw new Error(
|
||||
`The ${keyPrefix}timeout must be greater than ${keyPrefix}queue.waitTime + ${keyPrefix}waitTimeAfterEmptyReceive.`,
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
pollingWaitTime,
|
||||
queueUrl,
|
||||
region,
|
||||
timeout,
|
||||
topic,
|
||||
visibilityTimeout,
|
||||
waitTimeAfterEmptyReceive,
|
||||
};
|
||||
}) ?? []
|
||||
);
|
||||
}
|
||||
+94
@@ -0,0 +1,94 @@
|
||||
/*
|
||||
* 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 { getVoidLogger } from '@backstage/backend-common';
|
||||
import {
|
||||
configServiceRef,
|
||||
loggerServiceRef,
|
||||
schedulerServiceRef,
|
||||
} from '@backstage/backend-plugin-api';
|
||||
import { startTestBackend } from '@backstage/backend-test-utils';
|
||||
import { ConfigReader } from '@backstage/config';
|
||||
import { eventsExtensionPoint } from '@backstage/plugin-events-node';
|
||||
import { TestEventBroker } from '@backstage/plugin-events-backend-test-utils';
|
||||
import { awsSqsConsumingEventPublisherEventsModule } from './AwsSqsConsumingEventPublisherEventsModule';
|
||||
import { AwsSqsConsumingEventPublisher } from '../publisher/AwsSqsConsumingEventPublisher';
|
||||
|
||||
describe('awsSqsEventsModule', () => {
|
||||
it('should be correctly wired and set up', async () => {
|
||||
const config = new ConfigReader({
|
||||
events: {
|
||||
modules: {
|
||||
awsSqs: {
|
||||
awsSqsConsumingEventPublisher: {
|
||||
topics: {
|
||||
fake1: {
|
||||
queue: {
|
||||
region: 'eu-west-1',
|
||||
url: 'https://fake1.queue.url',
|
||||
},
|
||||
},
|
||||
fake2: {
|
||||
queue: {
|
||||
region: 'us-east-1',
|
||||
url: 'https://fake2.queue.url',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
let addedPublishers: AwsSqsConsumingEventPublisher[] | undefined;
|
||||
const extensionPoint = {
|
||||
addPublishers: (publishers: any) => {
|
||||
addedPublishers = publishers;
|
||||
},
|
||||
};
|
||||
|
||||
const scheduler = {
|
||||
scheduleTask: jest.fn(),
|
||||
};
|
||||
|
||||
await startTestBackend({
|
||||
extensionPoints: [[eventsExtensionPoint, extensionPoint]],
|
||||
services: [
|
||||
[configServiceRef, config],
|
||||
[loggerServiceRef, getVoidLogger()],
|
||||
[schedulerServiceRef, scheduler],
|
||||
],
|
||||
features: [awsSqsConsumingEventPublisherEventsModule()],
|
||||
});
|
||||
|
||||
expect(addedPublishers).not.toBeUndefined();
|
||||
expect(addedPublishers!.length).toEqual(2);
|
||||
|
||||
const eventBroker = new TestEventBroker();
|
||||
await Promise.all(
|
||||
addedPublishers!.map(publisher => publisher.setEventBroker(eventBroker)),
|
||||
);
|
||||
|
||||
// publisher.connect(..) was causing the polling for events to be scheduled
|
||||
expect(scheduler.scheduleTask).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ id: 'events.awsSqs.publisher:fake1' }),
|
||||
);
|
||||
expect(scheduler.scheduleTask).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ id: 'events.awsSqs.publisher:fake2' }),
|
||||
);
|
||||
});
|
||||
});
|
||||
+55
@@ -0,0 +1,55 @@
|
||||
/*
|
||||
* 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 {
|
||||
configServiceRef,
|
||||
createBackendModule,
|
||||
loggerServiceRef,
|
||||
loggerToWinstonLogger,
|
||||
schedulerServiceRef,
|
||||
} from '@backstage/backend-plugin-api';
|
||||
import { eventsExtensionPoint } from '@backstage/plugin-events-node';
|
||||
import { AwsSqsConsumingEventPublisher } from '../publisher/AwsSqsConsumingEventPublisher';
|
||||
|
||||
/**
|
||||
* AWS SQS module for the Events plugin.
|
||||
*
|
||||
* @alpha
|
||||
*/
|
||||
export const awsSqsConsumingEventPublisherEventsModule = createBackendModule({
|
||||
pluginId: 'events',
|
||||
moduleId: 'awsSqsConsumingEventPublisherEventsModule',
|
||||
register(env) {
|
||||
env.registerInit({
|
||||
deps: {
|
||||
config: configServiceRef,
|
||||
events: eventsExtensionPoint,
|
||||
logger: loggerServiceRef,
|
||||
scheduler: schedulerServiceRef,
|
||||
},
|
||||
async init({ config, events, logger, scheduler }) {
|
||||
const winstonLogger = loggerToWinstonLogger(logger);
|
||||
const sqs = AwsSqsConsumingEventPublisher.fromConfig({
|
||||
config: config,
|
||||
logger: winstonLogger,
|
||||
scheduler: scheduler,
|
||||
});
|
||||
|
||||
events.addPublishers(sqs);
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
@@ -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 {};
|
||||
Reference in New Issue
Block a user