diff --git a/packages/backend-common/package.json b/packages/backend-common/package.json index 7a0d492638..8f0e9ad412 100644 --- a/packages/backend-common/package.json +++ b/packages/backend-common/package.json @@ -42,6 +42,7 @@ "@backstage/types": "^1.0.0", "@google-cloud/storage": "^6.0.0", "@keyv/redis": "^2.2.3", + "@kubernetes/client-node": "^0.17.0", "@manypkg/get-packages": "^1.1.3", "@octokit/rest": "^19.0.3", "@types/cors": "^2.8.6", @@ -75,9 +76,11 @@ "node-abort-controller": "^3.0.1", "node-fetch": "^2.6.7", "raw-body": "^2.4.1", + "request": "^2.88.2", "selfsigned": "^2.0.0", "stoppable": "^1.1.0", "tar": "^6.1.2", + "uuid": "^8.3.2", "winston": "^3.2.1", "yauzl": "^2.10.0", "yn": "^4.0.0" @@ -106,8 +109,8 @@ "@types/stoppable": "^1.1.0", "@types/supertest": "^2.0.8", "@types/tar": "^6.1.1", - "@types/yauzl": "^2.10.0", "@types/webpack-env": "^1.15.2", + "@types/yauzl": "^2.10.0", "aws-sdk-mock": "^5.2.1", "better-sqlite3": "^7.5.0", "http-errors": "^2.0.0", diff --git a/packages/backend-common/src/util/KubernetesContainerRunner.test.ts b/packages/backend-common/src/util/KubernetesContainerRunner.test.ts new file mode 100644 index 0000000000..52af488d66 --- /dev/null +++ b/packages/backend-common/src/util/KubernetesContainerRunner.test.ts @@ -0,0 +1,266 @@ +/* + * Copyright 2020 The Backstage Authors + * + * 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 { + CoreV1Api, + KubeConfig, + RbacAuthorizationV1Api, +} from '@kubernetes/client-node'; +import { + KubernetesContainerRunner, + KubernetesContainerRunnerOptions, +} from './KubernetesContainerRunner'; +import { RunContainerOptions } from './ContainerRunner'; +import { PassThrough } from 'stream'; + +jest.setTimeout(10 * 1000); + +describe('KubernetesContainerRunner', () => { + const kubeConfig = new KubeConfig(); + kubeConfig.loadFromDefault(); + const name = 'kube-runner'; + const api = kubeConfig.makeApiClient(CoreV1Api); + const authApi = kubeConfig.makeApiClient(RbacAuthorizationV1Api); + + beforeAll(async () => { + await api.createNamespace({ + metadata: { + name: 'test', + }, + }); + }); + + afterAll(async () => { + await api.deleteNamespace('test'); + }); + + it('should throw error when no namespace is configured', () => { + const testConfig = new KubeConfig(); + testConfig.loadFromDefault(); + testConfig.addContext({ + name: 'test', + cluster: kubeConfig.getCurrentCluster()!.name, + user: kubeConfig.getCurrentUser()!.name, + }); + testConfig.setCurrentContext('test'); + const test = () => + new KubernetesContainerRunner({ + kubeConfig: testConfig, + name, + }); + expect(test).toThrow( + /Cannot read current namespace from Kubernetes cluster/, + ); + }); + + it('should throw error when mountBase is provided and podTemplate is invalid', () => { + const error = /A Pod template containing the volume .+ is required/; + const options: KubernetesContainerRunnerOptions = { + kubeConfig, + name, + namespace: 'default', + mountBase: { + basePath: '/workdir', + volumeName: 'workdir', + }, + }; + const test = () => { + return new KubernetesContainerRunner(options); + }; + expect(test).toThrow(error); + options.podTemplate = {}; + expect(test).toThrow(error); + options.podTemplate.spec = { containers: [] }; + expect(test).toThrow(error); + options.podTemplate.spec.volumes = []; + expect(test).toThrow(error); + }); + + it('should not run the container when the mounts are not subdirectories of the basePath', async () => { + const options: KubernetesContainerRunnerOptions = { + kubeConfig, + name, + namespace: 'default', + mountBase: { + basePath: '/workdir', + volumeName: 'workdir', + }, + podTemplate: { + spec: { + containers: [], + volumes: [ + { + name: 'workdir', + }, + ], + }, + }, + }; + const containerRunner = new KubernetesContainerRunner(options); + const hostDir = '/notWorkdir/app'; + const logStream = new PassThrough(); + const runOptions: RunContainerOptions = { + imageName: 'golang:1.17', + args: ['echo', 'hello world'], + logStream, + mountDirs: { + '/notWorkdir/app': '/app', + }, + }; + await expect(containerRunner.runContainer(runOptions)).rejects.toThrowError( + `Mounted '${hostDir}' dir should be subdirectories of '${ + options!.mountBase!.basePath + }`, + ); + }); + + it('should succeed when the container command returns a 0 exit code', async () => { + const options: KubernetesContainerRunnerOptions = { + kubeConfig, + name, + namespace: 'default', + }; + const containerRunner = new KubernetesContainerRunner(options); + const logStream = new PassThrough(); + const chunks: any[] = []; + logStream.on('data', chunk => chunks.push(Buffer.from(chunk))); + const runOptions: RunContainerOptions = { + imageName: 'alpine', + args: ['echo', 'hello world'], + logStream, + }; + await containerRunner.runContainer(runOptions); + const result = Buffer.concat(chunks).toString('utf8'); + expect(result).toBe('hello world\n'); + }); + + it('should fail when container run time exceeds the timeout', async () => { + const options: KubernetesContainerRunnerOptions = { + kubeConfig, + name, + namespace: 'default', + timeoutMs: 5000, + }; + const containerRunner = new KubernetesContainerRunner(options); + const runOptions: RunContainerOptions = { + imageName: 'alpine', + args: ['sleep', '10'], + }; + await expect(containerRunner.runContainer(runOptions)).rejects.toThrowError( + `Failed to complete in ${options.timeoutMs} ms`, + ); + }); + + it('should fail when container command returns a non 0 exit code', async () => { + const options: KubernetesContainerRunnerOptions = { + kubeConfig, + name, + namespace: 'default', + }; + const containerRunner = new KubernetesContainerRunner(options); + const runOptions: RunContainerOptions = { + imageName: 'alpine', + args: ['fake'], + }; + await expect(containerRunner.runContainer(runOptions)).rejects.toThrowError( + `Container execution failed`, + ); + }); + + it('should fail when job creation fails', async () => { + const options: KubernetesContainerRunnerOptions = { + kubeConfig, + name, + namespace: 'fake', + }; + const containerRunner = new KubernetesContainerRunner(options); + const runOptions: RunContainerOptions = { + imageName: 'golang:1.17', + args: ['echo', 'hello world'], + }; + await expect(containerRunner.runContainer(runOptions)).rejects.toThrowError( + 'Kubernetes Job creation failed with the following error message: namespaces "fake" not found', + ); + }); + + it('should fail when watch fails', async () => { + await api.createNamespacedServiceAccount('test', { + metadata: { + name: 'test', + }, + }); + await authApi.createNamespacedRole('test', { + metadata: { + name: 'test', + }, + rules: [ + { + apiGroups: ['batch'], + verbs: ['create'], + resources: ['jobs'], + }, + ], + }); + await authApi.createNamespacedRoleBinding('test', { + metadata: { + name: 'test', + }, + subjects: [ + { + kind: 'ServiceAccount', + name: 'test', + }, + ], + roleRef: { + apiGroup: 'rbac.authorization.k8s.io', + kind: 'Role', + name: 'test', + }, + }); + const token = ( + await api.createNamespacedServiceAccountToken('test', 'test', { + spec: { + audiences: [], + }, + }) + ).body.status?.token; + const testConfig = new KubeConfig(); + testConfig.loadFromDefault(); + testConfig.addUser({ + name: 'test', + token, + }); + testConfig.addContext({ + name: 'test', + cluster: kubeConfig.getCurrentCluster()!.name, + user: 'test', + }); + testConfig.setCurrentContext('test'); + const options: KubernetesContainerRunnerOptions = { + kubeConfig: testConfig, + name, + namespace: 'test', + }; + const containerRunner = new KubernetesContainerRunner(options); + const runOptions: RunContainerOptions = { + imageName: 'golang:1.17', + args: ['echo', 'hello world'], + }; + await expect(containerRunner.runContainer(runOptions)).rejects.toThrowError( + 'Kubernetes watch request failed with the following error message: Error: Forbidden', + ); + }); +}); diff --git a/packages/backend-common/src/util/KubernetesContainerRunner.ts b/packages/backend-common/src/util/KubernetesContainerRunner.ts new file mode 100644 index 0000000000..e4393c1bd9 --- /dev/null +++ b/packages/backend-common/src/util/KubernetesContainerRunner.ts @@ -0,0 +1,356 @@ +/* + * Copyright 2020 The Backstage Authors + * + * 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 { PassThrough, Writable } from 'stream'; +import { ContainerRunner, RunContainerOptions } from './ContainerRunner'; +import { + KubeConfig, + BatchV1Api, + V1Job, + V1EnvVar, + Log, + HttpError, + V1Status, + V1VolumeMount, + V1PodTemplateSpec, + V1Pod, + Watch, +} from '@kubernetes/client-node'; +import { v4 as uuid } from 'uuid'; +import { Request } from 'request'; + +export type MountBaseOptions = { + volumeName: string; + basePath: string; +}; + +export type KubernetesContainerRunnerOptions = { + kubeConfig: KubeConfig; + name: string; + namespace?: string; + mountBase?: MountBaseOptions; + podTemplate?: V1PodTemplateSpec; + timeoutMs?: number; +}; + +/** + * A {@link ContainerRunner} for Kubernetes. + * + * @public + */ +export class KubernetesContainerRunner implements ContainerRunner { + private readonly kubeConfig: KubeConfig; + private readonly batchV1Api: BatchV1Api; + private readonly log: Log; + private readonly name: string; + private readonly namespace: string; + private readonly mountBase?: MountBaseOptions; + private readonly podTemplate?: V1PodTemplateSpec; + private readonly timeoutMs: number; + private readonly containerName = 'executor'; + + getNamespace(kubeConfig: KubeConfig, namespace?: string): string { + let _namespace = namespace; + if (!_namespace) { + _namespace = kubeConfig.getContextObject( + kubeConfig.currentContext, + )?.namespace; + } + if (!_namespace) { + throw new Error('Cannot read current namespace from Kubernetes cluster'); + } + return _namespace; + } + + validateMountBase( + mountBase: MountBaseOptions, + podTemplate?: V1PodTemplateSpec, + ): MountBaseOptions { + if ( + !podTemplate?.spec?.volumes?.filter(v => v.name === mountBase.volumeName) + .length + ) { + throw new Error( + `A Pod template containing the volume ${mountBase.volumeName} is required`, + ); + } + if (!mountBase.basePath.endsWith('/')) { + mountBase.basePath += '/'; + } + return mountBase; + } + + constructor(options: KubernetesContainerRunnerOptions) { + const { kubeConfig, name, namespace, mountBase, podTemplate, timeoutMs } = + options; + this.kubeConfig = kubeConfig; + this.batchV1Api = kubeConfig.makeApiClient(BatchV1Api); + this.log = new Log(kubeConfig); + this.name = name; + this.namespace = this.getNamespace(kubeConfig, namespace); + if (mountBase) { + this.mountBase = this.validateMountBase(mountBase, podTemplate); + } + this.podTemplate = podTemplate; + this.timeoutMs = timeoutMs || 120 * 1000; + } + + async runContainer(options: RunContainerOptions) { + const { + imageName, + command, + args, + logStream = new PassThrough(), + mountDirs = {}, + workingDir, + envVars = {}, + } = options; + + const commandArr = typeof command === 'string' ? [command] : command; + + const volumeMounts: V1VolumeMount[] = []; + for (const [hostDir, containerDir] of Object.entries(mountDirs)) { + if (!this.mountBase) { + throw new Error( + 'A volumeName and a basePath must be configured to bind mount directories', + ); + } + if (!hostDir.startsWith(this.mountBase.basePath)) { + throw new Error( + `Mounted '${hostDir}' dir should be subdirectories of '${this.mountBase.basePath}'`, + ); + } + volumeMounts.push({ + name: this.mountBase.volumeName, + mountPath: containerDir, + subPath: hostDir.slice(this.mountBase.basePath.length), + }); + } + + const env = []; + for (const [key, value] of Object.entries(envVars)) { + env.push({ + name: key, + value: value, + } as V1EnvVar); + } + + const taskId = uuid(); + + const mergedPodTemplate: V1PodTemplateSpec = { + metadata: { + ...{ + labels: { + task: taskId, + }, + }, + ...this.podTemplate?.metadata, + }, + spec: { + ...{ + containers: [ + { + name: this.containerName, + image: imageName, + command: commandArr, + args: args, + env: env, + workingDir: workingDir, + volumeMounts: volumeMounts, + }, + ], + restartPolicy: 'Never', + }, + ...this.podTemplate?.spec, + }, + }; + + const jobSpec: V1Job = { + metadata: { + generateName: `${this.name}-`, + }, + spec: { + backoffLimit: 0, + ttlSecondsAfterFinished: 60, + template: mergedPodTemplate, + }, + }; + + await this.runJob(jobSpec, taskId, logStream); + } + + handleError(err: any, errorCallback: (reason: any) => void) { + if (err.code !== 'ECONNRESET' && err.message !== 'aborted') { + errorCallback( + handleKubernetesError( + 'Kubernetes watch request failed with the following error message:', + err, + ), + ); + } + } + + watchPod( + taskId: string, + callback: (pod: V1Pod) => void, + errorCallback: (reason: any) => void, + ): Promise { + const watch = new Watch(this.kubeConfig); + const labelSelector = `task=${taskId}`; + return watch.watch( + `/api/v1/namespaces/${this.namespace}/pods`, + { + labelSelector, + }, + (_, pod) => { + callback(pod); + }, + err => { + this.handleError(err, errorCallback); + }, + ); + } + + tailLogs( + taskId: string, + logStream: Writable, + ): { promise: Promise; close: () => Promise } { + let log: Promise; + let req: Promise; + const watchPromise = new Promise((_, reject) => { + req = this.watchPod( + taskId, + pod => { + if ( + log === undefined && + (pod.status?.phase === 'Running' || + pod.status?.phase === 'Succeeded' || + pod.status?.phase === 'Failed') + ) { + log = this.log.log( + this.namespace, + pod.metadata?.name!, + this.containerName, + logStream, + { follow: true }, + ); + } + }, + reject, + ); + }); + const logPromise = new Promise((resolve, _) => { + if (!logStream.writableFinished) { + logStream.on('finish', () => { + resolve(); + }); + } else { + resolve(); + } + }); + const close = async () => { + if (req) { + (await req).abort(); + } + if (log) { + (await log).abort(); + } + }; + return { promise: Promise.race([watchPromise, logPromise]), close }; + } + + waitPod(taskId: string): { + promise: Promise; + close: () => Promise; + } { + let req: Promise; + const promise = new Promise(async (resolve, reject) => { + req = this.watchPod( + taskId, + pod => { + if (pod.status?.phase === 'Succeeded') { + resolve(); + } + if (pod.status?.phase === 'Failed') { + reject(new Error('Container execution failed')); + } + }, + reject, + ); + }); + const close = async () => { + if (req) { + (await req).abort(); + } + }; + return { promise, close }; + } + + async createJob(jobSpec: V1Job): Promise { + return this.batchV1Api + .createNamespacedJob(this.namespace, jobSpec) + .catch(err => { + throw handleKubernetesError( + 'Kubernetes Job creation failed with the following error message:', + err, + ); + }); + } + + async runJob( + jobSpec: V1Job, + taskId: string, + logStream: Writable, + ): Promise { + let timeout: NodeJS.Timeout; + const timeoutPromise = new Promise((_, reject) => { + timeout = setTimeout( + reject, + this.timeoutMs, + new Error(`Failed to complete in ${this.timeoutMs} ms`), + ); + }); + + const { promise: waitPromise, close: waitClose } = this.waitPod(taskId); + const { promise: tailPromise, close: tailClose } = this.tailLogs( + taskId, + logStream, + ); + + const taskPromise = Promise.all([ + waitPromise, + tailPromise, + this.createJob(jobSpec), + ]).finally(() => { + clearTimeout(timeout); + }); + + return Promise.race([timeoutPromise, taskPromise]) + .finally(() => { + return waitClose(); + }) + .finally(() => { + return tailClose(); + }); + } +} + +function handleKubernetesError(message: string, err: Error): Error { + if (err instanceof HttpError) { + return new Error(`${message} ${(err.body as V1Status).message}`); + } + return new Error(`${message} ${err}`); +} diff --git a/yarn.lock b/yarn.lock index a1e5c018aa..6309737e43 100644 --- a/yarn.lock +++ b/yarn.lock @@ -7365,7 +7365,7 @@ resolved "https://registry.npmjs.org/@types/range-parser/-/range-parser-1.2.3.tgz#7ee330ba7caafb98090bece86a5ee44115904c2c" integrity sha512-ewFXqrQHlFsgc09MK5jP5iR7vumV/BYayNC6PgJO2LPe8vrnNFyjQjSppfEngITi0qvfKtzFvgKymGheFM9UOA== -"@types/react-dom@*", "@types/react-dom@<18.0.0", "@types/react-dom@^17": +"@types/react-dom@*", "@types/react-dom@<18.0.0": version "17.0.17" resolved "https://registry.npmjs.org/@types/react-dom/-/react-dom-17.0.17.tgz#2e3743277a793a96a99f1bf87614598289da68a1" integrity sha512-VjnqEmqGnasQKV0CWLevqMTXBYG9GbwuE6x3VetERLh0cq2LTptFE73MrQi2S7GkKXCf2GgwItB/melLnxfnsg== @@ -22861,7 +22861,7 @@ request-progress@^3.0.0: dependencies: throttleit "^1.0.0" -request@^2.88.0: +request@^2.88.0, request@^2.88.2: version "2.88.2" resolved "https://registry.npmjs.org/request/-/request-2.88.2.tgz#d73c918731cb5a87da047e207234146f664d12b3" integrity sha512-MsvtOrfG9ZcrOwAW+Qi+F6HbD0CWXEh9ou77uOb7FM2WPhwT7smM833PzanhJLsgXjN89Ir6V2PczXNnMpwKhw== @@ -26588,7 +26588,7 @@ ws@^7.3.1, ws@^7.4.6: resolved "https://registry.npmjs.org/ws/-/ws-7.5.7.tgz#9e0ac77ee50af70d58326ecff7e85eb3fa375e67" integrity sha512-KMvVuFzpKBuiIXW3E4u3mySRO2/mCHSyZDJQM5NQ9Q9KHWHWh0NHgfbRMLLrceUK5qAL4ytALJbpRMjixFZh8A== -ws@^8.0.0, ws@^8.3.0: +ws@^8.3.0: version "8.8.1" resolved "https://registry.npmjs.org/ws/-/ws-8.8.1.tgz#5dbad0feb7ade8ecc99b830c1d77c913d4955ff0" integrity sha512-bGy2JzvzkPowEJV++hF07hAD6niYSr0JzBNo/J29WsB57A2r7Wlc1UFcTR9IzrPvuNVO4B8LGqF8qcpsVOhJCA==