Merge pull request #3985 from nirga/master

Initial version of new Kafka plugin

packages/app/package.json

@@ -20,6 +20,7 @@
     "@backstage/plugin-graphiql": "^0.2.3",
     "@backstage/plugin-org": "^0.3.2",
     "@backstage/plugin-jenkins": "^0.3.4",
+    "@backstage/plugin-kafka": "^0.1.0",
     "@backstage/plugin-kubernetes": "^0.3.3",
     "@backstage/plugin-lighthouse": "^0.2.6",
     "@backstage/plugin-newrelic": "^0.2.2",

packages/app/src/components/catalog/EntityPage.tsx

@@ -63,6 +63,7 @@ import {
   UserProfileCard,
 } from '@backstage/plugin-org';
 import { Router as SentryRouter } from '@backstage/plugin-sentry';
+import { Router as KafkaRouter } from '@backstage/plugin-kafka';
 import { EmbeddedDocsRouter as DocsRouter } from '@backstage/plugin-techdocs';
 import { Button, Grid } from '@material-ui/core';
 import {
@@ -243,6 +244,11 @@ const ServiceEntityPage = ({ entity }: { entity: Entity }) => (
       title="Code Insights"
       element={<GitHubInsightsRouter entity={entity} />}
     />
+    <EntityPageLayout.Content
+      path="/kafka/*"
+      title="Kafka"
+      element={<KafkaRouter entity={entity} />}
+    />
   </EntityPageLayout>
 );
 

packages/app/src/plugins.ts

@@ -43,3 +43,4 @@ export { plugin as PagerDuty } from '@backstage/plugin-pagerduty';
 export { plugin as Buildkite } from '@roadiehq/backstage-plugin-buildkite';
 export { plugin as Search } from '@backstage/plugin-search';
 export { plugin as Org } from '@backstage/plugin-org';
+export { plugin as Kafka } from '@backstage/plugin-kafka';

packages/backend/package.json

@@ -35,6 +35,7 @@
     "@backstage/plugin-catalog-backend": "^0.5.3",
     "@backstage/plugin-graphql-backend": "^0.1.4",
     "@backstage/plugin-kubernetes-backend": "^0.2.4",
+    "@backstage/plugin-kafka-backend": "^0.1.0",
     "@backstage/plugin-proxy-backend": "^0.2.3",
     "@backstage/plugin-rollbar-backend": "^0.1.5",
     "@backstage/plugin-scaffolder-backend": "^0.4.1",

packages/backend/src/index.ts

@@ -38,6 +38,7 @@ import healthcheck from './plugins/healthcheck';
 import auth from './plugins/auth';
 import catalog from './plugins/catalog';
 import kubernetes from './plugins/kubernetes';
+import kafka from './plugins/kafka';
 import rollbar from './plugins/rollbar';
 import scaffolder from './plugins/scaffolder';
 import proxy from './plugins/proxy';
@@ -77,6 +78,7 @@ async function main() {
   const rollbarEnv = useHotMemoize(module, () => createEnv('rollbar'));
   const techdocsEnv = useHotMemoize(module, () => createEnv('techdocs'));
   const kubernetesEnv = useHotMemoize(module, () => createEnv('kubernetes'));
+  const kafkaEnv = useHotMemoize(module, () => createEnv('kafka'));
   const graphqlEnv = useHotMemoize(module, () => createEnv('graphql'));
   const appEnv = useHotMemoize(module, () => createEnv('app'));
 
@@ -87,6 +89,7 @@ async function main() {
   apiRouter.use('/auth', await auth(authEnv));
   apiRouter.use('/techdocs', await techdocs(techdocsEnv));
   apiRouter.use('/kubernetes', await kubernetes(kubernetesEnv));
+  apiRouter.use('/kafka', await kafka(kafkaEnv));
   apiRouter.use('/proxy', await proxy(proxyEnv));
   apiRouter.use('/graphql', await graphql(graphqlEnv));
   apiRouter.use(notFoundHandler());

packages/backend/src/plugins/kafka.ts

@@ -0,0 +1,25 @@
+/*
+ * 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 { createRouter } from '@backstage/plugin-kafka-backend';
+import { PluginEnvironment } from '../types';
+
+export default async function createPlugin({
+  logger,
+  config,
+}: PluginEnvironment) {
+  return await createRouter({ logger, config });
+}

plugins/kafka-backend/.eslintrc.js

@@ -0,0 +1,3 @@
+module.exports = {
+  extends: [require.resolve('@backstage/cli/config/eslint.backend')],
+};

plugins/kafka-backend/README.md

@@ -0,0 +1,28 @@
+# Kafka Backend
+
+This is the backend part of the Kafka plugin. It responds to Kafka requests from the frontend.
+
+## Configuration
+
+This configures how to connect to the brokers in your Kafka cluster.
+
+### clientId
+
+The name of the client to use when connecting to the cluster.
+
+### brokers
+
+A list of the brokers' host names and ports to connect to.
+
+### SSL (optional)
+
+Configure TLS connection to the Kafka cluster. The options are passed directly to [tls.connect] and used to create the TLS secure context. Normally these would include `key` and `cert`.
+
+Example:
+
+```yaml
+kafka:
+  clientId: backstage
+  brokers:
+    - localhost:9092
+```

plugins/kafka-backend/config.d.ts

@@ -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.
+ */
+export interface Config {
+  kafka?: {
+    /**
+     * Client ID used to Backstage uses to identify when connecting to the Kafka cluster.
+     */
+    clientId: string;
+    /**
+     * List of brokers in the Kafka cluster to connect to.
+     */
+    brokers: string[];
+
+    /**
+     * Optional SSL connection parameters to connect to the cluster. Passed directly to Node tls.connect.
+     * See https://nodejs.org/dist/latest-v8.x/docs/api/tls.html#tls_tls_createsecurecontext_options
+     */
+    ssl?: {
+      ca: string[];
+      /** @visibility secret */
+      key: string;
+      cert: string;
+    };
+  };
+}

plugins/kafka-backend/package.json

@@ -0,0 +1,55 @@
+{
+  "name": "@backstage/plugin-kafka-backend",
+  "version": "0.1.0",
+  "main": "src/index.ts",
+  "types": "src/index.ts",
+  "license": "Apache-2.0",
+  "private": false,
+  "publishConfig": {
+    "access": "public",
+    "main": "dist/index.cjs.js",
+    "types": "dist/index.d.ts"
+  },
+  "homepage": "https://backstage.io",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/backstage/backstage",
+    "directory": "plugins/kafka-backend"
+  },
+  "keywords": [
+    "backstage",
+    "kafka"
+  ],
+  "configSchema": "config.d.ts",
+  "scripts": {
+    "start": "backstage-cli backend:dev",
+    "build": "backstage-cli backend:build",
+    "lint": "backstage-cli lint",
+    "test": "backstage-cli test",
+    "prepack": "backstage-cli prepack",
+    "postpack": "backstage-cli postpack",
+    "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",
+    "express": "^4.17.1",
+    "express-promise-router": "^3.0.3",
+    "kafkajs": "^1.16.0-beta.6",
+    "lodash": "^4.17.15",
+    "winston": "^3.2.1"
+  },
+  "devDependencies": {
+    "@backstage/cli": "^0.4.3",
+    "@types/jest-when": "^2.7.2",
+    "@types/lodash": "^4.14.151",
+    "jest-when": "^3.1.0",
+    "supertest": "^4.0.2"
+  },
+  "files": [
+    "dist",
+    "schema.d.ts"
+  ]
+}

plugins/kafka-backend/src/index.ts

@@ -0,0 +1,17 @@
+/*
+ * 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.
+ */
+
+export { createRouter } from './service/router';

plugins/kafka-backend/src/service/KafkaApi.ts

@@ -0,0 +1,97 @@
+/*
+ * 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 { Kafka, SeekEntry } from 'kafkajs';
+import { Logger } from 'winston';
+import { ConnectionOptions } from 'tls';
+
+export type PartitionOffset = {
+  id: number;
+  offset: string;
+};
+
+export type TopicOffset = {
+  topic: string;
+  partitions: PartitionOffset[];
+};
+
+export type Options = {
+  clientId: string;
+  brokers: string[];
+  ssl?: ConnectionOptions;
+  logger: Logger;
+};
+
+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(options: Options) {
+    options.logger.debug(
+      `creating kafka client with clientId=${options.clientId} and brokers=${options.brokers}`,
+    );
+
+    this.kafka = new Kafka(options);
+    this.logger = options.logger;
+  }
+
+  async fetchTopicOffsets(topic: string): Promise<Array<PartitionOffset>> {
+    this.logger.debug(`fetching topic offsets for ${topic}`);
+
+    const admin = this.kafka.admin();
+    await admin.connect();
+
+    try {
+      return KafkaJsApiImpl.toPartitionOffsets(
+        await admin.fetchTopicOffsets(topic),
+      );
+    } finally {
+      await admin.disconnect();
+    }
+  }
+
+  async fetchGroupOffsets(groupId: string): Promise<Array<TopicOffset>> {
+    this.logger.debug(`fetching consumer group offsets for ${groupId}`);
+
+    const admin = this.kafka.admin();
+    await admin.connect();
+
+    try {
+      const groupOffsets = await admin.fetchOffsets({ groupId });
+
+      return groupOffsets.map(topicOffset => ({
+        topic: topicOffset.topic,
+        partitions: KafkaJsApiImpl.toPartitionOffsets(topicOffset.partitions),
+      }));
+    } finally {
+      await admin.disconnect();
+    }
+  }
+
+  private static toPartitionOffsets(
+    result: Array<SeekEntry>,
+  ): Array<PartitionOffset> {
+    return result.map(seekEntry => ({
+      id: seekEntry.partition,
+      offset: seekEntry.offset,
+    }));
+  }
+}