switch to throttleInterval and concurrencyLimit

Signed-off-by: Kai Dubauskas <kai.dubauskas@doordash.com>
This commit is contained in:
Kai Dubauskas
2025-12-01 11:05:34 -05:00
parent 061817fb23
commit 08d6456b4a
4 changed files with 71 additions and 23 deletions
+1 -1
View File
@@ -2,4 +2,4 @@
'@backstage/plugin-notifications-backend-module-slack': patch
---
The rate limit is now a config variable
The throttle limit and interval is now a config variable
+2 -1
View File
@@ -149,7 +149,8 @@ notifications:
broadcastChannels: # Optional, if you wish to support broadcast notifications.
- C12345678
username: 'Backstage Bot' # Optional, defaults to the name of the Slack App.
rateLimit: 40 # Optional, number of messages per minute. Defaults to 10.
concurrencyLimit: 20 # Optional, number of messages allowed per interval. Defaults to 10.
throttleInterval: 1m # Optional, ISO 8601 duration (or ms value). Defaults to 1 minute.
```
Multiple instances can be added in the `slack` array, allowing you to have multiple configurations if you need to send
@@ -19,6 +19,22 @@ import { SlackNotificationProcessor } from './SlackNotificationProcessor';
import { catalogServiceMock } from '@backstage/plugin-catalog-node/testUtils';
import { WebClient } from '@slack/web-api';
import { Entity } from '@backstage/catalog-model';
import pThrottle from 'p-throttle';
import { durationToMilliseconds } from '@backstage/types';
const throttleConfigs: Array<{ limit: number; interval: number }> = [];
jest.mock('p-throttle', () => ({
__esModule: true,
default: jest.fn((config: { limit: number; interval: number }) => {
throttleConfigs.push(config);
return <T extends (...args: any[]) => Promise<any> | any>(fn: T) =>
(...args: Parameters<T>) =>
Promise.resolve(fn(...args));
}),
}));
const mockedPThrottle = pThrottle as jest.MockedFunction<typeof pThrottle>;
jest.mock('@slack/web-api', () => {
const mockSlack = {
@@ -128,6 +144,8 @@ describe('SlackNotificationProcessor', () => {
beforeEach(() => {
jest.clearAllMocks();
throttleConfigs.length = 0;
mockedPThrottle.mockClear();
});
it('should send a notification to a group', async () => {
@@ -954,8 +972,8 @@ describe('SlackNotificationProcessor', () => {
});
});
describe('when rate limit is not configured', () => {
it('should use default rate limit of 10 messages per minute', async () => {
describe('when throttling is not configured', () => {
it('should use default concurrency limit of 10 per minute', async () => {
const slack = new WebClient();
const processor = SlackNotificationProcessor.fromConfig(config, {
@@ -973,13 +991,19 @@ describe('SlackNotificationProcessor', () => {
});
expect(slack.chat.postMessage).toHaveBeenCalled();
expect(throttleConfigs).toEqual([
{
limit: 10,
interval: durationToMilliseconds({ minutes: 1 }),
},
]);
});
});
describe('when rate limit is configured', () => {
it('should use custom rate limit value', async () => {
describe('when throttling is configured', () => {
it('should use custom concurrency limit and interval values', async () => {
const slack = new WebClient();
const rateLimitConfig = mockServices.rootConfig({
const throttlingConfig = mockServices.rootConfig({
data: {
app: {
baseUrl: 'https://example.org',
@@ -989,7 +1013,8 @@ describe('SlackNotificationProcessor', () => {
slack: [
{
token: 'mock-token',
rateLimit: 5,
concurrencyLimit: 5,
throttleInterval: 'PT30S',
},
],
},
@@ -997,14 +1022,17 @@ describe('SlackNotificationProcessor', () => {
},
});
const processor = SlackNotificationProcessor.fromConfig(rateLimitConfig, {
auth,
logger,
catalog: catalogServiceMock({
entities: DEFAULT_ENTITIES_RESPONSE.items,
}),
slack,
})[0];
const processor = SlackNotificationProcessor.fromConfig(
throttlingConfig,
{
auth,
logger,
catalog: catalogServiceMock({
entities: DEFAULT_ENTITIES_RESPONSE.items,
}),
slack,
},
)[0];
await processor.processOptions({
recipients: { type: 'entity', entityRef: 'group:default/mock' },
@@ -1012,6 +1040,12 @@ describe('SlackNotificationProcessor', () => {
});
expect(slack.chat.postMessage).toHaveBeenCalled();
expect(throttleConfigs).toEqual([
{
limit: 5,
interval: durationToMilliseconds({ seconds: 30 }),
},
]);
});
});
});
@@ -21,7 +21,7 @@ import {
parseEntityRef,
UserEntity,
} from '@backstage/catalog-model';
import { Config } from '@backstage/config';
import { Config, readDurationFromConfig } from '@backstage/config';
import { NotFoundError } from '@backstage/errors';
import { Notification } from '@backstage/plugin-notifications-common';
import {
@@ -50,6 +50,8 @@ export class SlackNotificationProcessor implements NotificationProcessor {
private readonly broadcastChannels?: string[];
private readonly entityLoader: DataLoader<string, Entity | undefined>;
private readonly username?: string;
private readonly concurrencyLimit: number;
private readonly throttleInterval: number;
static fromConfig(
config: Config,
@@ -68,12 +70,18 @@ export class SlackNotificationProcessor implements NotificationProcessor {
const slack = options.slack ?? new WebClient(token);
const broadcastChannels = c.getOptionalStringArray('broadcastChannels');
const username = c.getOptionalString('username');
const rateLimit = c.getOptionalNumber('rateLimit');
const concurrencyLimit = c.getOptionalNumber('concurrencyLimit') ?? 10;
const throttleInterval = c.has('throttleInterval')
? durationToMilliseconds(
readDurationFromConfig(c, { key: 'throttleInterval' }),
)
: durationToMilliseconds({ minutes: 1 });
return new SlackNotificationProcessor({
slack,
broadcastChannels,
username,
rateLimit,
concurrencyLimit,
throttleInterval,
...options,
});
});
@@ -86,7 +94,8 @@ export class SlackNotificationProcessor implements NotificationProcessor {
catalog: CatalogService;
broadcastChannels?: string[];
username?: string;
rateLimit?: number;
concurrencyLimit?: number;
throttleInterval?: number;
}) {
const {
auth,
@@ -95,7 +104,8 @@ export class SlackNotificationProcessor implements NotificationProcessor {
slack,
broadcastChannels,
username,
rateLimit,
concurrencyLimit,
throttleInterval,
} = options;
this.logger = logger;
this.catalog = catalog;
@@ -103,6 +113,9 @@ export class SlackNotificationProcessor implements NotificationProcessor {
this.slack = slack;
this.broadcastChannels = broadcastChannels;
this.username = username;
this.concurrencyLimit = concurrencyLimit ?? 10;
this.throttleInterval =
throttleInterval ?? durationToMilliseconds({ minutes: 1 });
this.entityLoader = new DataLoader<string, Entity | undefined>(
async entityRefs => {
@@ -144,8 +157,8 @@ export class SlackNotificationProcessor implements NotificationProcessor {
);
const throttle = pThrottle({
limit: rateLimit ?? 10,
interval: durationToMilliseconds({ minutes: 1 }),
limit: this.concurrencyLimit,
interval: this.throttleInterval,
});
const throttled = throttle((opts: ChatPostMessageArguments) =>
this.sendNotification(opts),