@mittwald/kubernetes/client.js

"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
const request_1 = __importDefault(require("request"));
const label_1 = require("./label");
const meta_1 = require("./types/meta");
const debug = require("debug")("kubernetes:client");
exports.patchKindStrategicMergePatch = "application/stategic-merge-patch+json";
exports.patchKindMergePatch = "application/merge-patch+json";
exports.patchKindJSONPatch = "application/json-patch+json";
const joinURL = (left, right) => (left + "/" + right).replace(/([^:])(\/\/)/g, "$1/");
class KubernetesRESTClient {
    constructor(config) {
        this.config = config;
    }
    request(url, body, method = "POST", additionalOptions = {}) {
        const absoluteURL = joinURL(this.config.apiServerURL, url);
        return new Promise((res, rej) => {
            let opts = {
                method,
                url: absoluteURL,
                json: true,
            };
            if (body) {
                opts.json = body;
            }
            opts = this.config.mapRequestOptions(opts);
            if (additionalOptions.headers) {
                additionalOptions.headers = Object.assign(Object.assign({}, (opts.headers || {})), additionalOptions.headers);
            }
            opts = Object.assign(Object.assign({}, opts), additionalOptions);
            debug(`executing ${method} request on ${opts.url}`);
            request_1.default(opts, (err, response, responseBody) => {
                if (err) {
                    rej(err);
                    return;
                }
                if (meta_1.isStatus(responseBody) && responseBody.status === "Failure") {
                    rej(new Error(responseBody.message));
                    return;
                }
                debug(`${method} request on ${opts.url} succeeded with status ${response.statusCode}: ${responseBody}`);
                res(responseBody);
            });
        });
    }
    post(url, body) {
        return this.request(url, body, "POST");
    }
    put(url, body) {
        return this.request(url, body, "PUT");
    }
    patch(url, body, patchKind) {
        return this.request(url, body, "PATCH", { headers: {
                "Content-Type": patchKind,
            } });
    }
    delete(url, deleteOptions, queryParams = {}, body) {
        const opts = {};
        opts.qs = queryParams;
        if (deleteOptions && deleteOptions.labelSelector) {
            opts.qs.labelSelector = label_1.selectorToQueryString(deleteOptions.labelSelector);
        }
        if (deleteOptions && deleteOptions.fieldSelector) {
            opts.qs.fieldSelector = label_1.selectorToQueryString(deleteOptions.fieldSelector);
        }
        return this.request(url, body, "DELETE", opts);
    }
    watch(url, onUpdate, onError, watchOpts = {}) {
        const absoluteURL = joinURL(this.config.apiServerURL, url);
        let opts = {
            url: absoluteURL,
            qs: { watch: "true" },
        };
        if (watchOpts.labelSelector) {
            opts.qs.labelSelector = label_1.selectorToQueryString(watchOpts.labelSelector);
        }
        if (watchOpts.fieldSelector) {
            opts.qs.fieldSelector = label_1.selectorToQueryString(watchOpts.fieldSelector);
        }
        if (watchOpts.resourceVersion) {
            opts.qs.resourceVersion = watchOpts.resourceVersion;
        }
        opts = this.config.mapRequestOptions(opts);
        let lastVersion = watchOpts.resourceVersion || 0;
        debug(`executing WATCH request on ${absoluteURL} (starting revision ${lastVersion})`);
        return new Promise((res, rej) => {
            const req = request_1.default(opts, (err, response, bodyString) => {
                if (err) {
                    rej(err);
                    return;
                }
                debug(`%o request on %o completed with status %o: %O`, "WATCH", opts.url, response.statusCode, bodyString);
                if (response.statusCode && response.statusCode >= 400) {
                    if (response.statusCode === 410) {
                        debug(`last known resource has expired -- resync required`);
                        res({ resourceVersion: lastVersion, resyncRequired: true });
                        return;
                    }
                    rej(new Error("Unexpected status code: " + response.statusCode));
                    return;
                }
                if (bodyString.length === 0) {
                    debug(`WATCH request on ${url} returned empty response`);
                    res({ resourceVersion: lastVersion });
                    return;
                }
                let body;
                try {
                    body = JSON.parse(bodyString);
                }
                catch (err) {
                    const bodyLines = bodyString.split("\n");
                    for (const line of bodyLines) {
                        if (line === "") {
                            continue;
                        }
                        try {
                            const parsedLine = JSON.parse(line);
                            if (parsedLine.type === "ADDED" || parsedLine.type === "MODIFIED" || parsedLine.type === "DELETED") {
                                const resourceVersion = parseInt(parsedLine.object.metadata.resourceVersion || "0", 10);
                                if (resourceVersion > lastVersion) {
                                    debug(`watch: emitting missed ${parsedLine.type} event for ${parsedLine.object.metadata.name}`);
                                    lastVersion = resourceVersion;
                                    onUpdate(parsedLine);
                                }
                            }
                        }
                        catch (err) {
                            debug(`watch: could not parse JSON line '${line}'`);
                            rej(err);
                            return;
                        }
                    }
                    res({ resourceVersion: lastVersion });
                    return;
                }
                if (meta_1.isStatus(body) && body.status === "Failure") {
                    debug(`watch: failed with status %O`, body);
                    rej(body.message);
                    return;
                }
                res({ resourceVersion: lastVersion });
            });
            let buffer = "";
            req.on("data", chunk => {
                if (chunk instanceof Buffer) {
                    chunk = chunk.toString("utf-8");
                }
                debug("WATCH request on %o received %d bytes of data", opts.url, chunk.length);
                buffer += chunk;
                try {
                    const obj = JSON.parse(buffer);
                    buffer = "";
                    const resourceVersion = obj.object.metadata.resourceVersion ? parseInt(obj.object.metadata.resourceVersion, 10) : -1;
                    if (resourceVersion > lastVersion) {
                        debug(`watch: emitting ${obj.type} event for ${obj.object.metadata.name}`);
                        lastVersion = resourceVersion;
                        onUpdate(obj);
                    }
                }
                catch (err) {
                    onError(err);
                }
            });
        });
    }
    get(url, listOptions = {}) {
        const absoluteURL = joinURL(this.config.apiServerURL, url);
        const { labelSelector, fieldSelector } = listOptions;
        return new Promise((res, rej) => {
            let opts = {
                url: absoluteURL,
                qs: {},
            };
            if (labelSelector) {
                opts.qs.labelSelector = label_1.selectorToQueryString(labelSelector);
            }
            if (fieldSelector) {
                opts.qs.fieldSelector = label_1.selectorToQueryString(fieldSelector);
            }
            opts = this.config.mapRequestOptions(opts);
            debug(`executing GET request on ${opts.url}`);
            request_1.default(opts, (err, response, body) => {
                if (err) {
                    rej(err);
                    return;
                }
                if (response.statusCode === 404) {
                    debug(`GET request on %o failed with status %o`, opts.url, response.statusCode);
                    res(undefined);
                    return;
                }
                try {
                    body = JSON.parse(body);
                }
                catch (err) {
                    rej(err);
                    return;
                }
                if (meta_1.isStatus(body) && body.status === "Failure") {
                    if (body.code === 404) {
                        res(undefined);
                        return;
                    }
                    debug(`executing GET request on %o failed. response body: %O`, response.statusCode, body);
                    rej(new Error(body.message));
                    return;
                }
                debug(`GET request on %o succeeded with status %o: %O`, opts.url, response.statusCode, body);
                res(body);
            });
        });
    }
}
exports.KubernetesRESTClient = KubernetesRESTClient;
//# sourceMappingURL=client.js.map