Added a table for showing consumer group topic offsets

plugins/kafka/src/Router.tsx

@@ -0,0 +1,38 @@
+/*
+ * 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 { Entity } from '@backstage/catalog-model';
+import { MissingAnnotationEmptyState } from '@backstage/core';
+import React from 'react';
+import { Route, Routes } from 'react-router';
+import { KAFKA_CONSUMER_GROUP_ANNOTATION } from './constants';
+import { KafkaTopicsForConsumer } from './components/ConsumerGroupOffsets/ConsumerGroupOffsets';
+
+export const isPluginApplicableToEntity = (entity: Entity) =>
+  Boolean(entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION]);
+
+export const Router = ({ entity }: { entity: Entity }) => {
+  return !isPluginApplicableToEntity(entity) ? (
+    <MissingAnnotationEmptyState annotation={KAFKA_CONSUMER_GROUP_ANNOTATION} />
+  ) : (
+    <Routes>
+      <Route
+        path={`/$(rootRouteRef.path)`}
+        element={<KafkaTopicsForConsumer />}
+      />
+    </Routes>
+  );
+};

plugins/kafka/src/api/KafkaApi.ts

@@ -0,0 +1,112 @@
+/*
+ * 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 { ConfigApi, DiscoveryApi, createApiRef } from '@backstage/core';
+import { Kafka, SeekEntry } from 'kafkajs';
+
+export const kafkaApiRef = createApiRef<KafkaApi>({
+  id: 'plugin.kafka.service',
+  description: 'Used by the Kafka plugin to connect to the Kafka cluster',
+});
+
+export type TopicOffsets = {
+  topic: string;
+  partitions: {
+    id: number;
+    offset: string;
+  }[];
+};
+
+export class KafkaApi {
+  static fromConfig(configApi: ConfigApi, discoveryApi: DiscoveryApi) {
+    const clientId =
+      configApi.getOptionalString('kafka.clientId') ?? 'kafka-backstage';
+    const brokers = configApi.getOptionalStringArray('kafka.brokers') ?? [
+      '/kafka/broker',
+    ];
+
+    return new KafkaApi(discoveryApi, clientId, brokers);
+  }
+
+  constructor(
+    private readonly discoveryApi: DiscoveryApi,
+    private readonly clientId: string,
+    private readonly brokers: string[],
+  ) {}
+
+  async fetchTopicsOffsets(topics: string[]): Promise<Array<TopicOffsets>> {
+    const admin = await this.getAdmin();
+
+    await admin.connect();
+    try {
+      return await Promise.all(
+        topics.map(async topic =>
+          admin
+            .fetchTopicOffsets(topic)
+            .then(result => KafkaApi.toTopicOffsets(topic, result)),
+        ),
+      );
+    } finally {
+      await admin.disconnect();
+    }
+  }
+
+  async fetchGroupOffsets(groupId: string): Promise<Array<TopicOffsets>> {
+    const admin = await this.getAdmin();
+
+    await admin.connect();
+    try {
+      let groupOffsets = [
+        {
+          topic: 'shula',
+          partitions: await admin.fetchOffsets({ groupId, topic: 'shula' }),
+        },
+        {
+          topic: 'test',
+          partitions: await admin.fetchOffsets({ groupId, topic: 'test' }),
+        },
+      ];
+
+      return groupOffsets.map(topicOffset =>
+        KafkaApi.toTopicOffsets(topicOffset.topic, topicOffset.partitions),
+      );
+    } finally {
+      await admin.disconnect();
+    }
+  }
+
+  private async getAdmin() {
+    const kafka = await this.getKafka();
+    return kafka.admin();
+  }
+
+  private async getKafka() {
+    const proxyUrl = await this.discoveryApi.getBaseUrl('proxy');
+
+    return new Kafka({
+      clientId: this.clientId,
+      brokers: this.brokers.map(brokerUrl => proxyUrl + brokerUrl),
+    });
+  }
+
+  private static toTopicOffsets(topic: string, result: Array<SeekEntry>) {
+    const partitions = result.map(seekEntry => ({
+      id: seekEntry.partition,
+      offset: seekEntry.offset,
+    }));
+    return { topic, partitions };
+  }
+}

plugins/kafka/src/api/index.ts

@@ -13,4 +13,5 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-export { default } from './ExampleComponent';
+
+export { KafkaApi } from './KafkaApi';

plugins/kafka/src/components/ConsumerGroupOffsets/ConsumerGroupOffsets.tsx

@@ -0,0 +1,111 @@
+/*
+ * 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 { Table, TableColumn } from '@backstage/core';
+import { Box, Typography } from '@material-ui/core';
+import RetryIcon from '@material-ui/icons/Replay';
+import React from 'react';
+import { useConsumerGroupFromEntity } from './useConsumerGroupsFromEntity';
+import { useConsumerGroupOffsets } from './useConsumerGroupOffsets';
+
+export type TopicPartitionInfo = {
+  topic: string;
+  partitionId: number;
+  topicOffset: string;
+  groupOffset: string;
+};
+
+const generatedColumns: TableColumn[] = [
+  {
+    title: 'Topic',
+    field: 'topic',
+    highlight: true,
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.topic ?? ''}</>;
+    },
+  },
+  {
+    title: 'Partition',
+    field: 'partitionId',
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.partitionId ?? ''}</>;
+    },
+  },
+  {
+    title: 'Topic Offset',
+    field: 'topicOffset',
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.topicOffset ?? ''}</>;
+    },
+  },
+  {
+    title: 'Group Offset',
+    field: 'groupOffset',
+    render: (row: Partial<TopicPartitionInfo>) => {
+      return <>{row.groupOffset ?? ''}</>;
+    },
+  },
+];
+
+type Props = {
+  loading: boolean;
+  retry: () => void;
+  consumerGroup: string;
+  topics?: TopicPartitionInfo[];
+};
+
+export const ConsumerGroupOffsets = ({
+  loading,
+  topics,
+  consumerGroup,
+  retry,
+}: Props) => {
+  return (
+    <Table
+      isLoading={loading}
+      actions={[
+        {
+          icon: () => <RetryIcon />,
+          tooltip: 'Refresh Data',
+          isFreeAction: true,
+          onClick: () => retry(),
+        },
+      ]}
+      data={topics ?? []}
+      title={
+        <Box display="flex" alignItems="center">
+          <Typography variant="h6">
+            Consumed Topics for {consumerGroup}
+          </Typography>
+        </Box>
+      }
+      columns={generatedColumns}
+    />
+  );
+};
+
+export const KafkaTopicsForConsumer = () => {
+  const consumerGroup = useConsumerGroupFromEntity();
+  const [tableProps, { retry }] = useConsumerGroupOffsets(consumerGroup);
+
+  return (
+    <ConsumerGroupOffsets
+      {...tableProps}
+      consumerGroup={consumerGroup}
+      retry={retry}
+    />
+  );
+};

plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupOffsets.ts

@@ -0,0 +1,63 @@
+/*
+ * 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 { errorApiRef, useApi } from '@backstage/core';
+import { useAsyncRetry } from 'react-use';
+import { kafkaApiRef } from '../../api/KafkaApi';
+import _ from 'lodash';
+
+export function useConsumerGroupOffsets(groupId: string) {
+  const api = useApi(kafkaApiRef);
+  const errorApi = useApi(errorApiRef);
+
+  const { loading, value: topics, retry } = useAsyncRetry(async () => {
+    try {
+      const groupOnlyOffsets = await api.fetchGroupOffsets(groupId);
+      const topicOffsets = _.keyBy(
+        await api.fetchTopicsOffsets(
+          groupOnlyOffsets.map(value => value.topic),
+        ),
+        offsets => offsets.topic,
+      );
+
+      return groupOnlyOffsets.flatMap(value => {
+        let topicPartitionOffsets = _.keyBy(
+          topicOffsets[value.topic].partitions,
+          partition => partition.id,
+        );
+        return value.partitions.map(partition => ({
+          topic: value.topic,
+          partitionId: partition.id,
+          groupOffset: partition.offset,
+          topicOffset: topicPartitionOffsets[partition.id].offset,
+        }));
+      });
+    } catch (e) {
+      errorApi.post(e);
+      throw e;
+    }
+  }, [api, errorApi, groupId]);
+
+  return [
+    {
+      loading,
+      topics,
+    },
+    {
+      retry,
+    },
+  ] as const;
+}

plugins/kafka/src/components/ConsumerGroupOffsets/useConsumerGroupsFromEntity.ts

@@ -0,0 +1,24 @@
+/*
+ * 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 { useEntity } from '@backstage/plugin-catalog';
+import { KAFKA_CONSUMER_GROUP_ANNOTATION } from '../../constants';
+
+export const useConsumerGroupFromEntity = () => {
+  const { entity } = useEntity();
+
+  return entity.metadata.annotations?.[KAFKA_CONSUMER_GROUP_ANNOTATION] ?? '';
+};

plugins/kafka/src/components/ExampleComponent/ExampleComponent.test.tsx

@@ -1,45 +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 React from 'react';
-import { render } from '@testing-library/react';
-import ExampleComponent from './ExampleComponent';
-import { ThemeProvider } from '@material-ui/core';
-import { lightTheme } from '@backstage/theme';
-import { rest } from 'msw';
-import { setupServer } from 'msw/node';
-import { msw } from '@backstage/test-utils';
-
-describe('ExampleComponent', () => {
-  const server = setupServer();
-  // Enable sane handlers for network requests
-  msw.setupDefaultHandlers(server);
-
-  // setup mock response
-  beforeEach(() => {
-    server.use(
-      rest.get('/*', (_, res, ctx) => res(ctx.status(200), ctx.json({}))),
-    );
-  });
-
-  it('should render', () => {
-    const rendered = render(
-      <ThemeProvider theme={lightTheme}>
-        <ExampleComponent />
-      </ThemeProvider>,
-    );
-    expect(rendered.getByText('Welcome to kafka!')).toBeInTheDocument();
-  });
-});

plugins/kafka/src/components/ExampleComponent/ExampleComponent.tsx

@@ -1,55 +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 React from 'react';
-import { Typography, Grid } from '@material-ui/core';
-import {
-  InfoCard,
-  Header,
-  Page,
-  Content,
-  ContentHeader,
-  HeaderLabel,
-  SupportButton,
-} from '@backstage/core';
-import ExampleFetchComponent from '../ExampleFetchComponent';
-
-const ExampleComponent = () => (
-  <Page themeId="tool">
-    <Header title="Welcome to kafka!" subtitle="Optional subtitle">
-      <HeaderLabel label="Owner" value="Team X" />
-      <HeaderLabel label="Lifecycle" value="Alpha" />
-    </Header>
-    <Content>
-      <ContentHeader title="Plugin title">
-        <SupportButton>A description of your plugin goes here.</SupportButton>
-      </ContentHeader>
-      <Grid container spacing={3} direction="column">
-        <Grid item>
-          <InfoCard title="Information card">
-            <Typography variant="body1">
-              All content should be wrapped in a card like this.
-            </Typography>
-          </InfoCard>
-        </Grid>
-        <Grid item>
-          <ExampleFetchComponent />
-        </Grid>
-      </Grid>
-    </Content>
-  </Page>
-);
-
-export default ExampleComponent;

plugins/kafka/src/components/ExampleFetchComponent/ExampleFetchComponent.test.tsx

@@ -1,40 +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 React from 'react';
-import { render } from '@testing-library/react';
-import ExampleFetchComponent from './ExampleFetchComponent';
-import { rest } from 'msw';
-import { setupServer } from 'msw/node';
-import { msw } from '@backstage/test-utils';
-
-describe('ExampleFetchComponent', () => {
-  const server = setupServer();
-  // Enable sane handlers for network requests
-  msw.setupDefaultHandlers(server);
-
-  // setup mock response
-  beforeEach(() => {
-    server.use(
-      rest.get('https://randomuser.me/*', (_, res, ctx) =>
-        res(ctx.status(200), ctx.delay(2000), ctx.json({})),
-      ),
-    );
-  });
-  it('should render', async () => {
-    const rendered = render(<ExampleFetchComponent />);
-    expect(await rendered.findByTestId('progress')).toBeInTheDocument();
-  });
-});

plugins/kafka/src/components/ExampleFetchComponent/ExampleFetchComponent.tsx

@@ -1,107 +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 React from 'react';
-import { makeStyles } from '@material-ui/core/styles';
-import { Table, TableColumn, Progress } from '@backstage/core';
-import Alert from '@material-ui/lab/Alert';
-import { useAsync } from 'react-use';
-
-const useStyles = makeStyles({
-  avatar: {
-    height: 32,
-    width: 32,
-    borderRadius: '50%',
-  },
-});
-
-type User = {
-  gender: string; // "male"
-  name: {
-    title: string; // "Mr",
-    first: string; // "Duane",
-    last: string; // "Reed"
-  };
-  location: object; // {street: {number: 5060, name: "Hickory Creek Dr"}, city: "Albany", state: "New South Wales",…}
-  email: string; // "duane.reed@example.com"
-  login: object; // {uuid: "4b785022-9a23-4ab9-8a23-cb3fb43969a9", username: "blackdog796", password: "patch",…}
-  dob: object; // {date: "1983-06-22T12:30:23.016Z", age: 37}
-  registered: object; // {date: "2006-06-13T18:48:28.037Z", age: 14}
-  phone: string; // "07-2154-5651"
-  cell: string; // "0405-592-879"
-  id: {
-    name: string; // "TFN",
-    value: string; // "796260432"
-  };
-  picture: { medium: string }; // {medium: "https://randomuser.me/api/portraits/men/95.jpg",…}
-  nat: string; // "AU"
-};
-
-type DenseTableProps = {
-  users: User[];
-};
-
-export const DenseTable = ({ users }: DenseTableProps) => {
-  const classes = useStyles();
-
-  const columns: TableColumn[] = [
-    { title: 'Avatar', field: 'avatar' },
-    { title: 'Name', field: 'name' },
-    { title: 'Email', field: 'email' },
-    { title: 'Nationality', field: 'nationality' },
-  ];
-
-  const data = users.map(user => {
-    return {
-      avatar: (
-        <img
-          src={user.picture.medium}
-          className={classes.avatar}
-          alt={user.name.first}
-        />
-      ),
-      name: `${user.name.first} ${user.name.last}`,
-      email: user.email,
-      nationality: user.nat,
-    };
-  });
-
-  return (
-    <Table
-      title="Example User List (fetching data from randomuser.me)"
-      options={{ search: false, paging: false }}
-      columns={columns}
-      data={data}
-    />
-  );
-};
-
-const ExampleFetchComponent = () => {
-  const { value, loading, error } = useAsync(async (): Promise<User[]> => {
-    const response = await fetch('https://randomuser.me/api/?results=20');
-    const data = await response.json();
-    return data.results;
-  }, []);
-
-  if (loading) {
-    return <Progress />;
-  } else if (error) {
-    return <Alert severity="error">{error.message}</Alert>;
-  }
-
-  return <DenseTable users={value || []} />;
-};
-
-export default ExampleFetchComponent;

plugins/kafka/src/constants.ts

@@ -13,4 +13,5 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-export { default } from './ExampleFetchComponent';
+export const KAFKA_CONSUMER_GROUP_ANNOTATION =
+  'kafka.apache.org/consumer-groups';

plugins/kafka/src/index.ts

@@ -14,3 +14,6 @@
  * limitations under the License.
  */
 export { plugin } from './plugin';
+export { KAFKA_CONSUMER_GROUP_ANNOTATION } from './constants';
+export { Router, isPluginApplicableToEntity } from './Router';
+export * from './api';

plugins/kafka/src/plugin.ts

@@ -13,8 +13,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { createPlugin, createRouteRef } from '@backstage/core';
-import ExampleComponent from './components/ExampleComponent';
+import {
+  configApiRef,
+  createApiFactory,
+  createPlugin,
+  createRouteRef,
+  discoveryApiRef,
+} from '@backstage/core';
+import { KafkaApi, kafkaApiRef } from './api/KafkaApi';
 
 export const rootRouteRef = createRouteRef({
   path: '/kafka',
@@ -23,7 +29,12 @@ export const rootRouteRef = createRouteRef({
 
 export const plugin = createPlugin({
   id: 'kafka',
-  register({ router }) {
-    router.addRoute(rootRouteRef, ExampleComponent);
-  },
+  apis: [
+    createApiFactory({
+      api: kafkaApiRef,
+      deps: { discoveryApi: discoveryApiRef, configApi: configApiRef },
+      factory: ({ configApi, discoveryApi }) =>
+        KafkaApi.fromConfig(configApi, discoveryApi),
+    }),
+  ],
 });