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

Added backend tests; Fixes following CR

Browse Source
This commit is contained in:
Nir Gazit
2021-01-12 19:45:48 +02:00
parent ba80eaa1f3
commit 44f841959b
15 changed files with 483 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/kafka-backend/package.json
+9 -4
View File
@@ -31,21 +31,26 @@
"clean": "backstage-cli clean"
},
"dependencies": {
"@backstage/backend-common": "^0.4.1",
"@backstage/catalog-model": "^0.6.0",
"@backstage/config": "^0.1.2",
"@types/express": "^4.17.6",
"compression": "^1.7.4",
"cors": "^2.8.5",
"express": "^4.17.1",
"express-promise-router": "^3.0.3",
"helmet": "^4.0.0",
"@backstage/backend-common": "^0.4.1",
"@backstage/catalog-model": "^0.6.0",
"@backstage/config": "^0.1.2",
"@types/express": "^4.17.6",
"kafkajs": "^1.16.0-beta.6",
"lodash": "^4.17.15",
"winston": "^3.2.1",
"yn": "^4.0.0"
},
"devDependencies": {
"@backstage/cli": "^0.4.3",
"@types/jest-when": "^2.7.2",
"@types/lodash": "^4.14.151",
"jest-extended": "^0.11.5",
"jest-when": "^3.1.0",
"supertest": "^4.0.2"
},
"files": [
plugins/kafka-backend/src/index.ts
+1 -2
View File
@@ -14,5 +14,4 @@
* limitations under the License.
*/
export * from './service/router';
export * from './types/types';
export { createRouter } from './service/router';
plugins/kafka-backend/src/service/KafkaApi.ts
+13 -6
View File
@@ -27,12 +27,17 @@ export type TopicOffset = {
partitions: PartitionOffset[];
};
export class KafkaApi {
export interface KafkaApi {
fetchTopicOffsets(topic: string): Promise<Array<PartitionOffset>>;
fetchGroupOffsets(groupId: string): Promise<Array<TopicOffset>>;
}
export class KafkaJsApiImpl implements KafkaApi {
private readonly kafka: Kafka;
private readonly logger: Logger;
constructor(clientId: string, brokers: string[], logger: Logger) {
logger.info(
logger.debug(
`creating kafka client with clientId=${clientId} and brokers=${brokers}`,
);
@@ -41,20 +46,22 @@ export class KafkaApi {
}
async fetchTopicOffsets(topic: string): Promise<Array<PartitionOffset>> {
this.logger.info(`fetching topic offsets for ${topic}`);
this.logger.debug(`fetching topic offsets for ${topic}`);
const admin = this.kafka.admin();
await admin.connect();
try {
return KafkaApi.toPartitionOffsets(await admin.fetchTopicOffsets(topic));
return KafkaJsApiImpl.toPartitionOffsets(
await admin.fetchTopicOffsets(topic),
);
} finally {
await admin.disconnect();
}
}
async fetchGroupOffsets(groupId: string): Promise<Array<TopicOffset>> {
this.logger.info(`fetching consumer group offsets for ${groupId}`);
this.logger.debug(`fetching consumer group offsets for ${groupId}`);
const admin = this.kafka.admin();
await admin.connect();
@@ -64,7 +71,7 @@ export class KafkaApi {
return groupOffsets.map(topicOffset => ({
topic: topicOffset.topic,
partitions: KafkaApi.toPartitionOffsets(topicOffset.partitions),
partitions: KafkaJsApiImpl.toPartitionOffsets(topicOffset.partitions),
}));
} finally {
await admin.disconnect();
plugins/kafka-backend/src/service/router.test.ts
+105
View File
@@ -0,0 +1,105 @@
/*
* Copyright 2020 Spotify AB
*
* 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 request from 'supertest';
import express from 'express';
import { makeRouter } from './router';
import { getVoidLogger } from '@backstage/backend-common';
import { KafkaApi } from './KafkaApi';
import { when } from 'jest-when';
describe('router', () => {
let app: express.Express;
let kafkaApi: jest.Mocked<KafkaApi>;
beforeAll(async () => {
kafkaApi = {
fetchTopicOffsets: jest.fn(),
fetchGroupOffsets: jest.fn(),
};
const router = makeRouter(getVoidLogger(), kafkaApi);
app = express().use(router);
});
beforeEach(() => {
jest.resetAllMocks();
});
describe('get /consumer/:consumerId/offsets', () => {
it('returns topic and group offsets', async () => {
const topic1Offsets = [
{ id: 1, offset: '500' },
{ id: 2, offset: '1000' },
];
const topic2Offsets = [{ id: 1, offset: '456' }];
const groupOffsets = [
{
topic: 'topic1',
partitions: [
{ id: 1, offset: '100' },
{ id: 2, offset: '213' },
],
},
{
topic: 'topic2',
partitions: [{ id: 1, offset: '456' }],
},
];
when(kafkaApi.fetchTopicOffsets)
.calledWith('topic1')
.mockResolvedValue(topic1Offsets);
when(kafkaApi.fetchTopicOffsets)
.calledWith('topic2')
.mockResolvedValue(topic2Offsets);
kafkaApi.fetchGroupOffsets.mockResolvedValue(groupOffsets);
const response = await request(app).get('/consumer/hey/offsets');
expect(response.status).toEqual(200);
expect(response.body.consumerId).toEqual('hey');
expect(response.body.offsets).toIncludeSameMembers([
{
topic: 'topic1',
partitionId: 1,
groupOffset: '100',
topicOffset: '500',
},
{
topic: 'topic1',
partitionId: 2,
groupOffset: '213',
topicOffset: '1000',
},
{
topic: 'topic2',
partitionId: 1,
groupOffset: '456',
topicOffset: '456',
},
]);
});
it('handles internal error correctly', async () => {
kafkaApi.fetchGroupOffsets.mockRejectedValue(Error('oh no'));
const response = await request(app).get('/consumer/hey/offsets');
expect(response.status).toEqual(500);
});
});
});
plugins/kafka-backend/src/service/router.ts
+22 -22
View File
@@ -18,7 +18,8 @@ import express from 'express';
import Router from 'express-promise-router';
import { Logger } from 'winston';
import { Config } from '@backstage/config';
import { KafkaApi } from './KafkaApi';
import { KafkaApi, KafkaJsApiImpl } from './KafkaApi';
import _ from 'lodash';
export interface RouterOptions {
logger: Logger;
@@ -32,28 +33,27 @@ export const makeRouter = (
const router = Router();
router.use(express.json());
router.get('/topic/:topicId/offsets', async (req, res) => {
const topicId = req.params.topicId;
try {
const response = await kafkaApi.fetchTopicOffsets(topicId);
res.send(response);
} catch (e) {
logger.error(`action=fetchTopicOffsets topicId=${topicId}, error=${e}`);
res.status(500).send({ error: e.message });
}
});
router.get('/consumer/:consumerId/offsets', async (req, res) => {
const consumerId = req.params.consumerId;
try {
const response = await kafkaApi.fetchGroupOffsets(consumerId);
res.send(response);
} catch (e) {
logger.error(
`action=fetchGroupOffsets consumerId=${consumerId}, error=${e}`,
);
res.status(500).send({ error: e.message });
}
const groupOffsets = await kafkaApi.fetchGroupOffsets(consumerId);
const groupWithTopicOffsets = await Promise.all(
groupOffsets.map(async ({ topic, partitions }) => {
const topicOffsets = _.keyBy(
await kafkaApi.fetchTopicOffsets(topic),
partition => partition.id,
);
return partitions.map(partition => ({
topic: topic,
partitionId: partition.id,
groupOffset: partition.offset,
topicOffset: topicOffsets[partition.id].offset,
}));
}),
);
res.send({ consumerId, offsets: groupWithTopicOffsets.flat() });
});
return router;
@@ -69,7 +69,7 @@ export async function createRouter(
const clientId = options.config.getString('kafka.clientId');
const brokers = options.config.getStringArray('kafka.brokers');
const kafkaApi = new KafkaApi(clientId, brokers, logger);
const kafkaApi = new KafkaJsApiImpl(clientId, brokers, logger);
return makeRouter(logger, kafkaApi);
}
plugins/kafka-backend/src/setupTests.ts
+8 -1
View File
@@ -14,4 +14,11 @@
* limitations under the License.
*/
export {};
import type { Config } from '@jest/types';
import 'jest-extended';
const config: Config.InitialOptions = {
setupFilesAfterEnv: ['jest-extended'],
};
export default config;
plugins/kafka-backend/src/types/types.ts
-20
View File
@@ -1,20 +0,0 @@
/*
* Copyright 2020 Spotify AB
*
* 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 { TopicOffset, PartitionOffset } from '../service/KafkaApi';
export type TopicOffsetsResponse = Array<PartitionOffset>;
export type ConsumerGroupOffsetsResponse = Array<TopicOffset>;