@atomist/sdm-core/lib/pack/k8s/KubernetesGoalScheduler.js

"use strict";
/*
 * Copyright © 2019 Atomist, Inc.
 *
 * 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.
 */
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
Object.defineProperty(exports, "__esModule", { value: true });
const automation_client_1 = require("@atomist/automation-client");
const sdm_1 = require("@atomist/sdm");
const k8s = require("@kubernetes/client-node");
const cluster = require("cluster");
const _ = require("lodash");
const os = require("os");
const config_1 = require("./config");
const service_1 = require("./service");
/**
 * GoalScheduler implementation that schedules SDM goals inside k8s jobs.
 *
 * It reuses the podSpec of the deployed SDM to create a new jobSpec from.
 * Subclasses may change the spec and job creation behavior by overwriting beforeCreation
 * and/or afterCreation methods.
 */
class KubernetesGoalScheduler {
    constructor(options = { isolateAll: false }) {
        this.options = options;
    }
    supports(gi) {
        return __awaiter(this, void 0, void 0, function* () {
            return !process.env.ATOMIST_ISOLATED_GOAL &&
                (
                // Goal is marked as isolated and SDM is configured to use k8s jobs
                (gi.goal.definition.isolated && isConfiguredInEnv("kubernetes")) ||
                    // Force all goals to run isolated via env var
                    isConfiguredInEnv("kubernetes-all") ||
                    // Force all goals to run isolated via explicit option
                    (this.options.isolateAll && isConfiguredInEnv("kubernetes")) ||
                    // Force all goals to run isolated via explicit configuration
                    _.get(gi.configuration, "sdm.k8s.isolateAll", false) === true);
        });
    }
    schedule(gi) {
        return __awaiter(this, void 0, void 0, function* () {
            const { goalEvent } = gi;
            const podNs = process.env.ATOMIST_POD_NAMESPACE || process.env.ATOMIST_DEPLOYMENT_NAMESPACE || "default";
            const kc = config_1.loadKubeConfig();
            const batch = kc.makeApiClient(k8s.Batch_v1Api);
            const jobSpec = createJobSpec(_.cloneDeep(this.podSpec), podNs, gi);
            yield this.beforeCreation(gi, jobSpec);
            gi.progressLog.write(`/--`);
            gi.progressLog.write(`Scheduling k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' for goal '${goalEvent.name} (${goalEvent.uniqueName})'`);
            gi.progressLog.write("\\--");
            try {
                // Check if this job was previously launched
                yield batch.readNamespacedJob(jobSpec.metadata.name, jobSpec.metadata.namespace);
                automation_client_1.logger.debug(`k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' for goal '${goalEvent.uniqueName}' already exists. Deleting...`);
                try {
                    yield batch.deleteNamespacedJob(jobSpec.metadata.name, jobSpec.metadata.namespace, {});
                    automation_client_1.logger.debug(`k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' for goal '${goalEvent.uniqueName}' deleted`);
                }
                catch (e) {
                    automation_client_1.logger.error(`Failed to delete k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' for goal '${goalEvent.uniqueName}': ${JSON.stringify(e.body)}`);
                    return {
                        code: 1,
                        message: `Failed to delete k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' ` +
                            `for goal '${goalEvent.uniqueName}': ${prettyPrintError(e)}`,
                    };
                }
            }
            catch (e) {
                // This is ok to ignore as it just means the job doesn't exist
            }
            try {
                // Previous deletion might not have completed; hence the retry here
                const jobResult = (yield automation_client_1.doWithRetry(() => batch.createNamespacedJob(jobSpec.metadata.namespace, jobSpec), `Scheduling k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' for goal '${goalEvent.uniqueName}'`)).body;
                yield this.afterCreation(gi, jobResult);
                automation_client_1.logger.info(`Scheduled k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' for goal '${goalEvent.uniqueName}' with result: ${JSON.stringify(jobResult.status)}`);
                automation_client_1.logger.log("silly", JSON.stringify(jobResult));
            }
            catch (e) {
                automation_client_1.logger.error(`Failed to schedule k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' for goal '${goalEvent.uniqueName}': ${JSON.stringify(e.body)}`);
                return {
                    code: 1,
                    message: `Failed to schedule k8s job '${jobSpec.metadata.namespace}:${jobSpec.metadata.name}' ` +
                        `for goal '${goalEvent.uniqueName}': ${prettyPrintError(e)}`,
                };
            }
            yield gi.progressLog.flush();
        });
    }
    /**
     * Extension point for sub classes to modify k8s resources or provided jobSpec before the
     * Job gets created in k8s.
     * Note: A potentially existing job with the same name has already been deleted at this point.
     * @param gi
     * @param jobSpec
     */
    beforeCreation(gi, jobSpec) {
        return __awaiter(this, void 0, void 0, function* () {
            // Intentionally left empty
        });
    }
    /**
     * Extension point for sub classes to modify k8s resources after the job has been created.
     * The provided jobSpec contains the result of the job creation API call.
     * @param gi
     * @param jobSpec
     */
    afterCreation(gi, jobSpec) {
        return __awaiter(this, void 0, void 0, function* () {
            // Intentionally left empty
        });
    }
    initialize(configuration) {
        return __awaiter(this, void 0, void 0, function* () {
            const podName = process.env.ATOMIST_POD_NAME || os.hostname();
            const podNs = process.env.ATOMIST_POD_NAMESPACE || process.env.ATOMIST_DEPLOYMENT_NAMESPACE || "default";
            try {
                const kc = new k8s.KubeConfig();
                kc.loadFromCluster();
                const core = kc.makeApiClient(k8s.Core_v1Api);
                this.podSpec = (yield core.readNamespacedPod(podName, podNs)).body;
            }
            catch (e) {
                automation_client_1.logger.error(`Failed to obtain parent pod spec from k8s: ${prettyPrintError(e)}`);
                throw new Error(`Failed to obtain parent pod spec from k8s: ${prettyPrintError(e)}`);
            }
            if (configuration.cluster.enabled === false || cluster.isMaster) {
                setInterval(() => {
                    return this.cleanUp()
                        .then(() => {
                        automation_client_1.logger.debug("Finished cleaning scheduled goal jobs");
                    });
                }, automation_client_1.configurationValue("sdm.k8s.job.cleanupInterval", 1000 * 60 * 60 * 2)).unref();
            }
        });
    }
    /**
     * Extension point to allow for custom clean up logic.
     */
    cleanUp() {
        return __awaiter(this, void 0, void 0, function* () {
            return cleanCompletedJobs();
        });
    }
}
exports.KubernetesGoalScheduler = KubernetesGoalScheduler;
/**
 * Cleanup scheduled k8s goal jobs
 * @returns {Promise<void>}
 */
function cleanCompletedJobs() {
    return __awaiter(this, void 0, void 0, function* () {
        const selector = `creator=${sanitizeName(automation_client_1.configurationValue("name"))}`;
        const jobs = yield listJobs(selector);
        const completedJobs = jobs.filter(j => j.status && j.status.completionTime && j.status.succeeded && j.status.succeeded > 0);
        if (completedJobs.length > 0) {
            automation_client_1.logger.info(`Deleting the following k8s jobs: ${completedJobs.map(j => `${j.metadata.namespace}:${j.metadata.name}`).join(", ")}`);
            const kc = config_1.loadKubeConfig();
            const batch = kc.makeApiClient(k8s.Batch_v1Api);
            for (const completedSdmJob of completedJobs) {
                try {
                    yield batch.deleteNamespacedJob(completedSdmJob.metadata.name, completedSdmJob.metadata.namespace, 
                    // propagationPolicy is needed so that pods of the job are also getting deleted
                    { propagationPolicy: "Foreground" });
                }
                catch (e) {
                    automation_client_1.logger.warn(`Failed to delete k8s job '${completedSdmJob.metadata.namespace}:${completedSdmJob.metadata.name}': ` +
                        prettyPrintError(e));
                }
            }
        }
    });
}
exports.cleanCompletedJobs = cleanCompletedJobs;
/**
 * Create a jobSpec by modifying the provided podSpec
 * @param podSpec
 * @param podNs
 * @param gi
 * @param context
 */
function createJobSpec(podSpec, podNs, gi) {
    const { goalEvent, context } = gi;
    const goalName = goalEvent.uniqueName.split("#")[0].toLowerCase();
    const jobSpec = createJobSpecWithAffinity(podSpec, gi);
    jobSpec.metadata.name = `${podSpec.spec.containers[0].name}-job-${goalEvent.goalSetId.slice(0, 7)}-${goalName}`;
    jobSpec.metadata.namespace = podNs;
    jobSpec.spec.template.spec.restartPolicy = "Never";
    jobSpec.spec.template.spec.containers[0].name = jobSpec.metadata.name;
    jobSpec.spec.template.spec.containers[0].env.push({
        name: "ATOMIST_JOB_NAME",
        value: jobSpec.metadata.name,
    }, {
        name: "ATOMIST_REGISTRATION_NAME",
        value: `${automation_client_1.automationClientInstance().configuration.name}-job-${goalEvent.goalSetId.slice(0, 7)}-${goalName}`,
    }, {
        name: "ATOMIST_GOAL_TEAM",
        value: context.workspaceId,
    }, {
        name: "ATOMIST_GOAL_TEAM_NAME",
        value: context.context.workspaceName,
    }, {
        name: "ATOMIST_GOAL_ID",
        value: goalEvent.id,
    }, {
        name: "ATOMIST_GOAL_SET_ID",
        value: goalEvent.goalSetId,
    }, {
        name: "ATOMIST_GOAL_UNIQUE_NAME",
        value: goalEvent.uniqueName,
    }, {
        name: "ATOMIST_CORRELATION_ID",
        value: context.correlationId,
    }, {
        name: "ATOMIST_ISOLATED_GOAL",
        value: "true",
    });
    rewriteCachePath(jobSpec, context.workspaceId);
    // Add additional specs from registered services to the job spec
    if (_.get(gi.configuration, "sdm.k8s.service.enabled", true)) {
        if (!!goalEvent.data) {
            let data = {};
            try {
                data = JSON.parse(goalEvent.data);
            }
            catch (e) {
                automation_client_1.logger.warn(`Failed to parse goal data on '${goalEvent.uniqueName}'`);
            }
            if (!!data[sdm_1.ServiceRegistrationGoalDataKey]) {
                _.forEach(data[sdm_1.ServiceRegistrationGoalDataKey], (v, k) => {
                    automation_client_1.logger.debug(`Service with name '${k}' and type '${v.type}' found for goal '${goalEvent.uniqueName}'`);
                    if (v.type === service_1.K8sServiceRegistrationType.K8sService) {
                        const spec = v.spec;
                        if (!!spec.container) {
                            const c = Array.isArray(spec.container) ? spec.container : [spec.container];
                            jobSpec.spec.template.spec.containers.push(...c);
                        }
                        if (!!spec.volume) {
                            const vo = Array.isArray(spec.volume) ? spec.volume : [spec.volume];
                            jobSpec.spec.template.spec.volumes.push(...vo);
                        }
                        if (!!spec.volumeMount) {
                            const vm = Array.isArray(spec.volumeMount) ? spec.volumeMount : [spec.volumeMount];
                            jobSpec.spec.template.spec.containers.forEach(c => c.volumeMounts.push(...vm));
                        }
                    }
                });
            }
        }
    }
    return jobSpec;
}
exports.createJobSpec = createJobSpec;
/**
 * Create a k8s Job spec with affinity to jobs for the same goal set
 * @param goalSetId
 */
function createJobSpecWithAffinity(podSpec, gi) {
    const { goalEvent, configuration, context } = gi;
    podSpec.spec.affinity = {
        podAffinity: {
            preferredDuringSchedulingIgnoredDuringExecution: [
                {
                    weight: 100,
                    podAffinityTerm: {
                        labelSelector: {
                            matchExpressions: [
                                {
                                    key: "atomist.com/goal-set-id",
                                    operator: "In",
                                    values: [
                                        goalEvent.goalSetId,
                                    ],
                                },
                            ],
                        },
                        topologyKey: "kubernetes.io/hostname",
                    },
                },
            ],
        },
    };
    // Clean up podSpec
    // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.13/#pod-v1-core note on nodeName
    delete podSpec.spec.nodeName;
    const labels = {
        "atomist.com/goal-set-id": goalEvent.goalSetId,
        "atomist.com/goal-id": goalEvent.id,
        "atomist.com/creator": sanitizeName(configuration.name),
        "atomist.com/workspace-id": context.workspaceId,
    };
    const detail = {
        sdm: {
            name: configuration.name,
            version: configuration.version,
        },
        goal: {
            goalId: goalEvent.id,
            goalSetId: goalEvent.goalSetId,
            uniqueName: goalEvent.uniqueName,
        },
    };
    const annotations = {
        "atomist.com/sdm": JSON.stringify(detail),
    };
    return {
        kind: "Job",
        apiVersion: "batch/v1",
        metadata: {
            labels,
            annotations,
        },
        spec: {
            template: {
                metadata: {
                    labels,
                },
                spec: podSpec.spec,
            },
        },
    };
}
/**
 * Rewrite the volume host path to include the workspace id to prevent cross workspace content ending
 * up in the same directory.
 * @param jobSpec
 * @param workspaceId
 */
function rewriteCachePath(jobSpec, workspaceId) {
    const cachePath = automation_client_1.configurationValue("sdm.cache.path", "/opt/data");
    const containers = _.get(jobSpec, "spec.template.spec.containers", []);
    const cacheVolumeNames = [];
    containers.forEach(c => {
        cacheVolumeNames.push(...c.volumeMounts.filter(vm => vm.mountPath === cachePath).map(cm => cm.name));
    });
    _.uniq(cacheVolumeNames).forEach(vn => {
        const volume = _.get(jobSpec, "spec.template.spec.volumes", []).find(v => v.name === vn);
        if (!!volume && !!volume.hostPath && !!volume.hostPath.path) {
            const path = volume.hostPath.path;
            if (!path.endsWith(workspaceId) || !path.endsWith(`${workspaceId}/`)) {
                if (path.endsWith("/")) {
                    volume.hostPath.path = `${path}${workspaceId}`;
                }
                else {
                    volume.hostPath.path = `${path}/${workspaceId}`;
                }
            }
        }
    });
}
/**
 * Checks if one of the provided values is configured in ATOMIST_GOAL_SCHEDULER or -
 * for backwards compatibility reasons - ATOMIST_GOAL_LAUNCHER.
 * @param values
 */
function isConfiguredInEnv(...values) {
    const value = process.env.ATOMIST_GOAL_SCHEDULER || process.env.ATOMIST_GOAL_LAUNCHER;
    if (!!value) {
        try {
            const json = JSON.parse(value);
            if (Array.isArray(json)) {
                return json.some(v => values.includes(v));
            }
            else {
                return values.includes(json);
            }
        }
        catch (e) {
            if (typeof value === "string") {
                return values.includes(value);
            }
        }
    }
    else {
        return false;
    }
}
exports.isConfiguredInEnv = isConfiguredInEnv;
/**
 * Strip out any characters that aren't allowed a k8s label value
 * @param name
 */
function sanitizeName(name) {
    return name.replace(/@/g, "").replace(/\//g, ".");
}
exports.sanitizeName = sanitizeName;
/**
 * List k8s jobs for a single namespace or cluster-wide depending on evn configuration
 * @param labelSelector
 */
function listJobs(labelSelector) {
    return __awaiter(this, void 0, void 0, function* () {
        const kc = config_1.loadKubeConfig();
        const batch = kc.makeApiClient(k8s.Batch_v1Api);
        if (automation_client_1.configurationValue("sdm.k8s.job.singleNamespace", true)) {
            const podNs = process.env.ATOMIST_POD_NAMESPACE || process.env.ATOMIST_DEPLOYMENT_NAMESPACE || "default";
            return (yield batch.listNamespacedJob(podNs, undefined, undefined, undefined, undefined, labelSelector)).body.items;
        }
        else {
            return (yield batch.listJobForAllNamespaces(undefined, undefined, undefined, labelSelector)).body.items;
        }
    });
}
exports.listJobs = listJobs;
function prettyPrintError(e) {
    if (!!e.body) {
        return e.body.message;
    }
    else {
        return e.message;
    }
}
exports.prettyPrintError = prettyPrintError;
//# sourceMappingURL=KubernetesGoalScheduler.js.map