Split to kafka backend plugin

plugins/kafka/src/api/KafkaApi.ts

@@ -1,112 +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 { 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/KafkaBackendClient.ts

@@ -0,0 +1,58 @@
+/*
+ * 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 { DiscoveryApi } from '@backstage/core';
+import { KafkaApi } from './types';
+import {
+  ConsumerGroupOffsetsResponse,
+  TopicOffsetsResponse,
+} from '@backstage/plugin-kafka-backend';
+
+export class KafkaBackendClient implements KafkaApi {
+  private readonly discoveryApi: DiscoveryApi;
+
+  constructor(options: { discoveryApi: DiscoveryApi }) {
+    this.discoveryApi = options.discoveryApi;
+  }
+
+  private async getRequired(path: string): Promise<any> {
+    const url = `${await this.discoveryApi.getBaseUrl('kafka')}${path}`;
+    const response = await fetch(url, {
+      method: 'GET',
+      headers: {
+        'Content-Type': 'application/json',
+      },
+    });
+
+    if (!response.ok) {
+      const payload = await response.text();
+      const message = `Request failed with ${response.status} ${response.statusText}, ${payload}`;
+      throw new Error(message);
+    }
+
+    return await response.json();
+  }
+
+  async getTopicOffsets(topic: string): Promise<TopicOffsetsResponse> {
+    return await this.getRequired(`/topic/${topic}/offsets`);
+  }
+
+  async getConsumerGroupOffsets(
+    consumerGroup: string,
+  ): Promise<ConsumerGroupOffsetsResponse> {
+    return await this.getRequired(`/consumer/${consumerGroup}/offsets`);
+  }
+}

plugins/kafka/src/api/types.ts

@@ -14,4 +14,19 @@
  * limitations under the License.
  */
 
-export { KafkaApi } from './KafkaApi';
+import { createApiRef } from '@backstage/core';
+import {
+  ConsumerGroupOffsetsResponse,
+  TopicOffsetsResponse,
+} from '@backstage/plugin-kafka-backend';
+
+export const kafkaApiRef = createApiRef<KafkaApi>({
+  id: 'plugin.kafka.service',
+  description:
+    'Used by the Kafka plugin to make requests to accompanying backend',
+});
+
+export interface KafkaApi {
+  getTopicOffsets(topic: string): Promise<TopicOffsetsResponse>;
+  getConsumerGroupOffsets(consumerGroup: string): Promise<ConsumerGroupOffsetsResponse>;
+}

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

@@ -16,7 +16,7 @@
 
 import { errorApiRef, useApi } from '@backstage/core';
 import { useAsyncRetry } from 'react-use';
-import { kafkaApiRef } from '../../api/KafkaApi';
+import { kafkaApiRef } from '../../api/types';
 import _ from 'lodash';
 
 export function useConsumerGroupOffsets(groupId: string) {
@@ -25,32 +25,32 @@ export function useConsumerGroupOffsets(groupId: string) {
 
   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,
-      );
+      const groupOffsets = await api.getConsumerGroupOffsets(groupId);
+      const groupWithTopicOffsets = await Promise.all(
+        groupOffsets.map(async ({ topic, partitions }) => {
+          debugger;
+          const topicOffsets = _.keyBy(
+            await api.getTopicOffsets(topic),
+            partition => partition.id,
+          );
 
-      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,
-        }));
-      });
+          return partitions.map(partition => ({
+            topic: topic,
+            partitionId: partition.id,
+            groupOffset: partition.offset,
+            topicOffset: topicOffsets[partition.id].offset,
+          }));
+        }),
+      );
+      return groupWithTopicOffsets.flat();
     } catch (e) {
       errorApi.post(e);
       throw e;
     }
   }, [api, errorApi, groupId]);
 
+  debugger;
+
   return [
     {
       loading,

plugins/kafka/src/index.ts

@@ -16,4 +16,3 @@
 export { plugin } from './plugin';
 export { KAFKA_CONSUMER_GROUP_ANNOTATION } from './constants';
 export { Router, isPluginApplicableToEntity } from './Router';
-export * from './api';

plugins/kafka/src/plugin.ts

@@ -20,7 +20,8 @@ import {
   createRouteRef,
   discoveryApiRef,
 } from '@backstage/core';
-import { KafkaApi, kafkaApiRef } from './api/KafkaApi';
+import { KafkaBackendClient } from './api/KafkaBackendClient';
+import { kafkaApiRef } from './api/types';
 
 export const rootCatalogKafkaRouteRef = createRouteRef({
   path: '*',
@@ -33,8 +34,7 @@ export const plugin = createPlugin({
     createApiFactory({
       api: kafkaApiRef,
       deps: { discoveryApi: discoveryApiRef, configApi: configApiRef },
-      factory: ({ configApi, discoveryApi }) =>
-        KafkaApi.fromConfig(configApi, discoveryApi),
+      factory: ({ discoveryApi }) => new KafkaBackendClient({ discoveryApi }),
     }),
   ],
 });