flow like the river

VISUALIZACION/node_modules/@elastic/transport/lib/pool/ClusterConnectionPool.js

@@ -0,0 +1,259 @@
+"use strict";
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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.
+ */
+Object.defineProperty(exports, "__esModule", { value: true });
+const tslib_1 = require("tslib");
+const BaseConnectionPool_1 = tslib_1.__importDefault(require("./BaseConnectionPool"));
+const assert_1 = tslib_1.__importDefault(require("assert"));
+const debug_1 = tslib_1.__importDefault(require("debug"));
+const connection_1 = require("../connection");
+const debug = (0, debug_1.default)('elasticsearch');
+class ClusterConnectionPool extends BaseConnectionPool_1.default {
+    constructor(opts) {
+        var _a, _b;
+        super(opts);
+        Object.defineProperty(this, "dead", {
+            enumerable: true,
+            configurable: true,
+            writable: true,
+            value: void 0
+        });
+        Object.defineProperty(this, "resurrectTimeout", {
+            enumerable: true,
+            configurable: true,
+            writable: true,
+            value: void 0
+        });
+        Object.defineProperty(this, "resurrectTimeoutCutoff", {
+            enumerable: true,
+            configurable: true,
+            writable: true,
+            value: void 0
+        });
+        Object.defineProperty(this, "pingTimeout", {
+            enumerable: true,
+            configurable: true,
+            writable: true,
+            value: void 0
+        });
+        Object.defineProperty(this, "resurrectStrategy", {
+            enumerable: true,
+            configurable: true,
+            writable: true,
+            value: void 0
+        });
+        this.dead = [];
+        // the resurrect timeout is 60s
+        this.resurrectTimeout = 1000 * 60;
+        // number of consecutive failures after which
+        // the timeout doesn't increase
+        this.resurrectTimeoutCutoff = 5;
+        this.pingTimeout = (_a = opts.pingTimeout) !== null && _a !== void 0 ? _a : 3000;
+        const resurrectStrategy = (_b = opts.resurrectStrategy) !== null && _b !== void 0 ? _b : 'ping';
+        this.resurrectStrategy = ClusterConnectionPool.resurrectStrategies[resurrectStrategy];
+        (0, assert_1.default)(this.resurrectStrategy != null, `Invalid resurrection strategy: '${resurrectStrategy}'`);
+    }
+    /**
+     * Marks a connection as 'alive'.
+     * If needed removes the connection from the dead list
+     * and then resets the `deadCount`.
+     *
+     * @param {object} connection
+     */
+    markAlive(connection) {
+        const { id } = connection;
+        debug(`Marking as 'alive' connection '${id}'`);
+        const index = this.dead.indexOf(id);
+        if (index > -1)
+            this.dead.splice(index, 1);
+        connection.status = connection_1.BaseConnection.statuses.ALIVE;
+        connection.deadCount = 0;
+        connection.resurrectTimeout = 0;
+        return this;
+    }
+    /**
+     * Marks a connection as 'dead'.
+     * If needed adds the connection to the dead list
+     * and then increments the `deadCount`.
+     *
+     * @param {object} connection
+     */
+    markDead(connection) {
+        const { id } = connection;
+        debug(`Marking as 'dead' connection '${id}'`);
+        if (!this.dead.includes(id)) {
+            // It might happen that `markDead` is called jsut after
+            // a pool update, and in such case we will add to the dead
+            // list a node that no longer exist. The following check verify
+            // that the connection is still part of the pool before
+            // marking it as dead.
+            for (let i = 0; i < this.size; i++) {
+                if (this.connections[i].id === id) {
+                    this.dead.push(id);
+                    break;
+                }
+            }
+        }
+        connection.status = connection_1.BaseConnection.statuses.DEAD;
+        connection.deadCount++;
+        // resurrectTimeout formula:
+        // `resurrectTimeout * 2 ** min(deadCount - 1, resurrectTimeoutCutoff)`
+        connection.resurrectTimeout = Date.now() + this.resurrectTimeout * Math.pow(2, Math.min(connection.deadCount - 1, this.resurrectTimeoutCutoff));
+        // sort the dead list in ascending order
+        // based on the resurrectTimeout
+        this.dead.sort((a, b) => {
+            const conn1 = this.connections.find(c => c.id === a);
+            const conn2 = this.connections.find(c => c.id === b);
+            return conn1.resurrectTimeout - conn2.resurrectTimeout;
+        });
+        return this;
+    }
+    /**
+     * If enabled, tries to resurrect a connection with the given
+     * resurrect strategy ('ping', 'optimistic', 'none').
+     *
+     * @param {object} { now, requestId }
+     */
+    resurrect(opts) {
+        if (this.resurrectStrategy === 0 || this.dead.length === 0) {
+            debug('Nothing to resurrect');
+            return;
+        }
+        // the dead list is sorted in ascending order based on the timeout
+        // so the first element will always be the one with the smaller timeout
+        const connection = this.connections.find(c => c.id === this.dead[0]);
+        if (opts.now < connection.resurrectTimeout) {
+            debug('Nothing to resurrect');
+            return;
+        }
+        const { id } = connection;
+        // ping strategy
+        if (this.resurrectStrategy === 1) {
+            connection.request({ method: 'HEAD', path: '/' }, { timeout: this.pingTimeout, requestId: opts.requestId, name: opts.name, context: opts.context })
+                .then(({ statusCode }) => {
+                let isAlive = true;
+                if (statusCode === 502 || statusCode === 503 || statusCode === 504) {
+                    debug(`Resurrect: connection '${id}' is still dead`);
+                    this.markDead(connection);
+                    isAlive = false;
+                }
+                else {
+                    debug(`Resurrect: connection '${id}' is now alive`);
+                    this.markAlive(connection);
+                }
+                this.diagnostic.emit('resurrect', null, {
+                    strategy: 'ping',
+                    name: opts.name,
+                    request: { id: opts.requestId },
+                    isAlive,
+                    connection
+                });
+            })
+                .catch((err) => {
+                this.markDead(connection);
+                this.diagnostic.emit('resurrect', err, {
+                    strategy: 'ping',
+                    name: opts.name,
+                    request: { id: opts.requestId },
+                    isAlive: false,
+                    connection
+                });
+            });
+            // optimistic strategy
+        }
+        else {
+            debug(`Resurrect: optimistic resurrection for connection '${id}'`);
+            this.dead.splice(this.dead.indexOf(id), 1);
+            connection.status = connection_1.BaseConnection.statuses.ALIVE;
+            this.diagnostic.emit('resurrect', null, {
+                strategy: 'optimistic',
+                name: opts.name,
+                request: { id: opts.requestId },
+                isAlive: true,
+                connection
+            });
+        }
+    }
+    /**
+     * Returns an alive connection if present,
+     * otherwise returns a dead connection.
+     * By default it filters the `master` only nodes.
+     * It uses the selector to choose which
+     * connection return.
+     *
+     * @param {object} options (filter and selector)
+     * @returns {object|null} connection
+     */
+    getConnection(opts) {
+        const filter = opts.filter != null ? opts.filter : () => true;
+        const selector = opts.selector != null ? opts.selector : (c) => c[0];
+        this.resurrect({
+            now: opts.now,
+            requestId: opts.requestId,
+            name: opts.name,
+            context: opts.context
+        });
+        const noAliveConnections = this.size === this.dead.length;
+        // TODO: can we cache this?
+        const connections = [];
+        for (let i = 0; i < this.size; i++) {
+            const connection = this.connections[i];
+            if (noAliveConnections || connection.status === connection_1.BaseConnection.statuses.ALIVE) {
+                if (filter(connection)) {
+                    connections.push(connection);
+                }
+            }
+        }
+        if (connections.length === 0)
+            return null;
+        return selector(connections);
+    }
+    /**
+     * Empties the connection pool.
+     *
+     * @returns {ConnectionPool}
+     */
+    async empty() {
+        await super.empty();
+        this.dead = [];
+    }
+    /**
+     * Update the ConnectionPool with new connections.
+     *
+     * @param {array} array of connections
+     * @returns {ConnectionPool}
+     */
+    update(connections) {
+        super.update(connections);
+        this.dead = [];
+        return this;
+    }
+}
+exports.default = ClusterConnectionPool;
+Object.defineProperty(ClusterConnectionPool, "resurrectStrategies", {
+    enumerable: true,
+    configurable: true,
+    writable: true,
+    value: {
+        none: 0,
+        ping: 1,
+        optimistic: 2
+    }
+});
+//# sourceMappingURL=ClusterConnectionPool.js.map