first draft KubernetesContainerRunner

Signed-off-by: Matteo Silvestri <matteosilv@gmail.com>
This commit is contained in:
Matteo Silvestri
2022-07-26 18:18:23 +02:00
parent 86fafd4058
commit 4e662da44a
4 changed files with 629 additions and 4 deletions
+4 -1
View File
@@ -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",
@@ -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',
);
});
});
@@ -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<Request> {
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<void>; close: () => Promise<void> } {
let log: Promise<Request>;
let req: Promise<Request>;
const watchPromise = new Promise<void>((_, 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<void>((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<void>;
close: () => Promise<void>;
} {
let req: Promise<Request>;
const promise = new Promise<void>(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<any> {
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<any> {
let timeout: NodeJS.Timeout;
const timeoutPromise = new Promise<void>((_, 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}`);
}
+3 -3
View File
@@ -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==