diff --git a/app-config.yaml b/app-config.yaml index 02afb8ac41..7017b0eee8 100644 --- a/app-config.yaml +++ b/app-config.yaml @@ -99,6 +99,13 @@ kubernetes: - 'config' clusters: [] +kafka: + clientId: backstage + clusters: + - name: cluster + brokers: + - localhost:9092 + integrations: github: - host: github.com @@ -225,6 +232,8 @@ catalog: # Backstage example groups and users - type: file target: ../catalog-model/examples/acme-corp.yaml + - type: file + target: ../../local.yaml scaffolder: github: @@ -372,8 +381,3 @@ homepage: timezone: 'Asia/Tokyo' pagerduty: eventsBaseUrl: 'https://events.pagerduty.com/v2' - -kafka: - clientId: backstage - brokers: - - localhost:9092 diff --git a/plugins/kafka-backend/src/service/router.test.ts b/plugins/kafka-backend/src/service/router.test.ts index 400729627b..42cd92ddfb 100644 --- a/plugins/kafka-backend/src/service/router.test.ts +++ b/plugins/kafka-backend/src/service/router.test.ts @@ -51,7 +51,7 @@ describe('router', () => { jest.resetAllMocks(); }); - describe('get /:clusterId/consumer/:consumerId/offsets', () => { + describe('get /consumers/clusterId/:consumerId/offsets', () => { it('returns topic and group offsets', async () => { const topic1Offsets = [ { id: 1, offset: '500' }, @@ -82,7 +82,7 @@ describe('router', () => { .calledWith('hey') .mockResolvedValue(groupOffsets); - const response = await request(app).get('/prod/consumer/hey/offsets'); + const response = await request(app).get('/consumers/prod/hey/offsets'); expect(response.status).toEqual(200); expect(response.body.consumerId).toEqual('hey'); @@ -114,7 +114,7 @@ describe('router', () => { it('handles internal error correctly', async () => { prodKafkaApi.fetchGroupOffsets.mockRejectedValue(Error('oh no')); - const response = await request(app).get('/prod/consumer/hey/offsets'); + const response = await request(app).get('/consumers/prod/hey/offsets'); expect(response.status).toEqual(500); }); @@ -147,29 +147,35 @@ describe('router', () => { .calledWith('hey') .mockResolvedValue(groupDevOffsets); - const prodResponse = await request(app).get('/prod/consumer/hey/offsets'); - const devResponse = await request(app).get('/dev/consumer/hey/offsets'); + const prodResponse = await request(app).get( + '/consumers/prod/hey/offsets', + ); + const devResponse = await request(app).get('/consumers/dev/hey/offsets'); expect(prodResponse.status).toEqual(200); expect(prodResponse.body.consumerId).toEqual('hey'); - expect(prodResponse.body.offsets).toIncludeSameMembers([ - { - topic: 'topic1', - partitionId: 1, - groupOffset: '100', - topicOffset: '500', - }, - ]); + expect(new Set(prodResponse.body.offsets)).toStrictEqual( + new Set([ + { + topic: 'topic1', + partitionId: 1, + groupOffset: '100', + topicOffset: '500', + }, + ]), + ); expect(devResponse.status).toEqual(200); expect(devResponse.body.consumerId).toEqual('hey'); - expect(devResponse.body.offsets).toIncludeSameMembers([ - { - topic: 'topic1', - partitionId: 1, - groupOffset: '567', - topicOffset: '1234', - }, - ]); + expect(new Set(devResponse.body.offsets)).toStrictEqual( + new Set([ + { + topic: 'topic1', + partitionId: 1, + groupOffset: '567', + topicOffset: '1234', + }, + ]), + ); }); }); }); diff --git a/plugins/kafka/src/components/ConsumerGroupOffsets/ConsumerGroupOffsets.tsx b/plugins/kafka/src/components/ConsumerGroupOffsets/ConsumerGroupOffsets.tsx index 448d420bc0..7928393526 100644 --- a/plugins/kafka/src/components/ConsumerGroupOffsets/ConsumerGroupOffsets.tsx +++ b/plugins/kafka/src/components/ConsumerGroupOffsets/ConsumerGroupOffsets.tsx @@ -15,7 +15,7 @@ */ import { Table, TableColumn } from '@backstage/core'; -import { Box, Typography } from '@material-ui/core'; +import { Box, Grid, Typography } from '@material-ui/core'; import RetryIcon from '@material-ui/icons/Replay'; import React from 'react'; import { useConsumerGroupsOffsetsForEntity } from './useConsumerGroupsOffsetsForEntity'; @@ -62,6 +62,7 @@ const generatedColumns: TableColumn[] = [ type Props = { loading: boolean; retry: () => void; + clusterId: string; consumerGroup: string; topics?: TopicPartitionInfo[]; }; @@ -69,6 +70,7 @@ type Props = { export const ConsumerGroupOffsets = ({ loading, topics, + clusterId, consumerGroup, retry, }: Props) => { @@ -87,7 +89,7 @@ export const ConsumerGroupOffsets = ({ title={ - Consumed Topics for {consumerGroup} + Consumed Topics for {consumerGroup} ({clusterId}) } @@ -98,6 +100,15 @@ export const ConsumerGroupOffsets = ({ export const KafkaTopicsForConsumer = () => { const [tableProps, { retry }] = useConsumerGroupsOffsetsForEntity(); - - return ; + return ( + + {tableProps.consumerGroupsTopics?.map(consumerGroup => ( + + ))} + + ); }; diff --git a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.test.tsx b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.test.tsx index 483986d4ec..3bee4e2e88 100644 --- a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.test.tsx +++ b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.test.tsx @@ -49,10 +49,66 @@ describe('useConsumerGroupOffsets', () => { }, }; const { result } = subject(); - expect(result.current).toStrictEqual({ - clusterId: 'prod', - consumerGroup: 'consumer', - }); + expect(result.current).toStrictEqual([ + { + clusterId: 'prod', + consumerGroup: 'consumer', + }, + ]); + }); + + it('returns correct cluster and consumer group for multiple consumers', async () => { + entity = { + apiVersion: 'v1', + kind: 'Component', + metadata: { + name: 'test', + annotations: { + 'kafka.apache.org/consumer-groups': + 'prod/consumer,dev/another-consumer', + }, + }, + spec: { + owner: 'guest', + type: 'Website', + lifecycle: 'development', + }, + }; + const { result } = subject(); + expect(result.current).toStrictEqual([ + { clusterId: 'prod', consumerGroup: 'consumer' }, + { + clusterId: 'dev', + consumerGroup: 'another-consumer', + }, + ]); + }); + + it('returns correct cluster and consumer group for annotation with extra spaces', async () => { + entity = { + apiVersion: 'v1', + kind: 'Component', + metadata: { + name: 'test', + annotations: { + 'kafka.apache.org/consumer-groups': + ' prod/consumer , dev/another-consumer ', + }, + }, + spec: { + owner: 'guest', + type: 'Website', + lifecycle: 'development', + }, + }; + const { result } = subject(); + expect(result.current).toStrictEqual([ + { clusterId: 'prod', consumerGroup: 'consumer' }, + { + clusterId: 'dev', + consumerGroup: 'another-consumer', + }, + ]); }); it('fails on missing cluster', async () => { @@ -62,7 +118,7 @@ describe('useConsumerGroupOffsets', () => { metadata: { name: 'test', annotations: { - 'kafka.apache.org/consumer-groups': 'consumer', + 'kafka.apache.org/consumer-groups': 'dev/another,consumer', }, }, spec: { @@ -75,7 +131,7 @@ describe('useConsumerGroupOffsets', () => { expect(() => result.current).toThrowError(); expect(result.error).toStrictEqual( new Error( - `Failed to parse kafka consumer group annotation: got "consumer"`, + `Failed to parse kafka consumer group annotation: got "dev/another,consumer"`, ), ); }); diff --git a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.ts b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.ts index 0afec42870..522a6273fe 100644 --- a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.ts +++ b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsForEntity.ts @@ -15,19 +15,29 @@ */ import { useEntity } from '@backstage/plugin-catalog'; +import { useMemo } from 'react'; import { KAFKA_CONSUMER_GROUP_ANNOTATION } from '../../constants'; export const useConsumerGroupsForEntity = () => { const { entity } = useEntity(); const annotation = entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION] ?? ''; - const [clusterId, consumerGroup] = annotation.split('/'); - if (!clusterId || !consumerGroup) { - throw new Error( - `Failed to parse kafka consumer group annotation: got "${annotation}"`, - ); - } + const consumerList = useMemo(() => { + return annotation.split(',').map(consumer => { + const [clusterId, consumerGroup] = consumer.split('/'); - return { clusterId, consumerGroup }; + if (!clusterId || !consumerGroup) { + throw new Error( + `Failed to parse kafka consumer group annotation: got "${annotation}"`, + ); + } + return { + clusterId: clusterId.trim(), + consumerGroup: consumerGroup.trim(), + }; + }); + }, [annotation]); + + return consumerList; }; diff --git a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.test.tsx b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.test.tsx index 16bf2f7665..b7e3a29e80 100644 --- a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.test.tsx +++ b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.test.tsx @@ -45,7 +45,7 @@ describe('useConsumerGroupOffsets', () => { metadata: { name: 'test', annotations: { - 'kafka.apache.org/consumer-groups': `cluster/${consumerGroupOffsets.consumerId}`, + 'kafka.apache.org/consumer-groups': `prod/${consumerGroupOffsets.consumerId}`, }, }, spec: { @@ -74,16 +74,24 @@ describe('useConsumerGroupOffsets', () => { renderHook(useConsumerGroupsOffsetsForEntity, { wrapper }); it('returns correct consumer group for annotation', async () => { + mockKafkaApi.getConsumerGroupOffsets.mockResolvedValue( + consumerGroupOffsets, + ); when(mockKafkaApi.getConsumerGroupOffsets) - .calledWith('cluster', consumerGroupOffsets.consumerId) + .calledWith('prod', consumerGroupOffsets.consumerId) .mockResolvedValue(consumerGroupOffsets); const { result, waitForNextUpdate } = subject(); await waitForNextUpdate(); const [tableProps] = result.current; - expect(tableProps.consumerGroup).toBe(consumerGroupOffsets.consumerId); - expect(tableProps.topics).toBe(consumerGroupOffsets.offsets); + expect(tableProps.consumerGroupsTopics).toStrictEqual([ + { + clusterId: 'prod', + consumerGroup: consumerGroupOffsets.consumerId, + topics: consumerGroupOffsets.offsets, + }, + ]); }); it('posts an error to the error api', async () => { diff --git a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.ts b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.ts index 4081b59796..8cb30040e9 100644 --- a/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.ts +++ b/plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsOffsetsForEntity.ts @@ -20,28 +20,35 @@ import { kafkaApiRef } from '../../api/types'; import { useConsumerGroupsForEntity } from './useConsumerGroupsForEntity'; export const useConsumerGroupsOffsetsForEntity = () => { - const { clusterId, consumerGroup } = useConsumerGroupsForEntity(); + const consumers = useConsumerGroupsForEntity(); const api = useApi(kafkaApiRef); const errorApi = useApi(errorApiRef); - const { loading, value: topics, retry } = useAsyncRetry(async () => { + const { + loading, + value: consumerGroupsTopics, + retry, + } = useAsyncRetry(async () => { try { - const response = await api.getConsumerGroupOffsets( - clusterId, - consumerGroup, + return await Promise.all( + consumers.map(async ({ clusterId, consumerGroup }) => { + const response = await api.getConsumerGroupOffsets( + clusterId, + consumerGroup, + ); + return { clusterId, consumerGroup, topics: response.offsets }; + }), ); - return response.offsets; } catch (e) { errorApi.post(e); throw e; } - }, [api, errorApi, consumerGroup]); + }, [consumers, api, errorApi]); return [ { loading, - consumerGroup, - topics, + consumerGroupsTopics, }, { retry,