index.js

/*
 * Copyright OpenSearch Contributors
 * SPDX-License-Identifier: Apache-2.0
 *
 * The OpenSearch Contributors require contributions made to
 * this file be licensed under the Apache-2.0 license or a
 * compatible open source license.
 *
 */

/*
 * 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.
 */

'use strict';

const { EventEmitter } = require('events');
const { URL } = require('url');
const debug = require('debug')('opensearch');
const Transport = require('./lib/Transport');
const Connection = require('./lib/Connection');
const { ConnectionPool, CloudConnectionPool } = require('./lib/pool');
const Helpers = require('./lib/Helpers');
const Serializer = require('./lib/Serializer');
const errors = require('./lib/errors');
const { ConfigurationError } = errors;
const { prepareHeaders } = Connection.internals;
let clientVersion = require('./package.json').version;
/* istanbul ignore next */
if (clientVersion.includes('-')) {
  // clean prerelease
  clientVersion = clientVersion.slice(0, clientVersion.indexOf('-')) + 'p';
}

const kInitialOptions = Symbol('opensearchjs-initial-options');
const kChild = Symbol('opensearchjs-child');
const kExtensions = Symbol('opensearchjs-extensions');
const kEventEmitter = Symbol('opensearchjs-event-emitter');

const OpenSearchAPI = require('./api');

class Client extends OpenSearchAPI {
  constructor(opts = {}) {
    super({ ConfigurationError });
    if (opts.cloud && opts[kChild] === undefined) {
      const { id, username, password } = opts.cloud;
      // the cloud id is `cluster-name:base64encodedurl`
      // the url is a string divided by two '$', the first is the cloud url
      // the second the opensearch instance, the third the opensearchDashboards instance
      const cloudUrls = Buffer.from(id.split(':')[1], 'base64').toString().split('$');

      // TODO: remove username and password here in 8
      if (username && password) {
        opts.auth = Object.assign({}, opts.auth, { username, password });
      }
      opts.node = `https://${cloudUrls[1]}.${cloudUrls[0]}`;

      // Cloud has better performances with compression enabled
      // see https://github.com/elastic/elasticsearch-py/pull/704.
      // So unless the user specifies otherwise, we enable compression.
      if (opts.compression == null) opts.compression = 'gzip';
      if (opts.suggestCompression == null) opts.suggestCompression = true;
      if (opts.ssl == null || (opts.ssl && opts.ssl.secureProtocol == null)) {
        opts.ssl = opts.ssl || {};
        opts.ssl.secureProtocol = 'TLSv1_2_method';
      }
    }

    if (!opts.node && !opts.nodes) {
      throw new ConfigurationError('Missing node(s) option');
    }

    if (opts[kChild] === undefined) {
      const checkAuth = getAuth(opts.node || opts.nodes);
      if (checkAuth && checkAuth.username && checkAuth.password) {
        opts.auth = Object.assign({}, opts.auth, {
          username: checkAuth.username,
          password: checkAuth.password,
        });
      }
    }

    const options =
      opts[kChild] !== undefined
        ? opts[kChild].initialOptions
        : Object.assign(
            {},
            {
              Connection,
              Transport,
              Serializer,
              ConnectionPool: opts.cloud ? CloudConnectionPool : ConnectionPool,
              maxRetries: 3,
              requestTimeout: 30000,
              pingTimeout: 3000,
              sniffInterval: false,
              sniffOnStart: false,
              sniffEndpoint: '_nodes/_all/http',
              sniffOnConnectionFault: false,
              resurrectStrategy: 'ping',
              suggestCompression: false,
              compression: false,
              ssl: null,
              agent: null,
              headers: {},
              nodeFilter: null,
              nodeSelector: 'round-robin',
              generateRequestId: null,
              name: 'opensearch-js',
              auth: null,
              opaqueIdPrefix: null,
              context: null,
              proxy: null,
              enableMetaHeader: true,
              disablePrototypePoisoningProtection: false,
              enableLongNumeralSupport: false,
            },
            opts
          );

    if (process.env.OPENSEARCH_CLIENT_APIVERSIONING === 'true') {
      options.headers = Object.assign(
        { accept: 'application/vnd.opensearch+json; compatible-with=7' },
        options.headers
      );
    }

    this[kInitialOptions] = options;
    this[kExtensions] = [];
    this.name = options.name;

    if (opts[kChild] !== undefined) {
      this.serializer = options[kChild].serializer;
      this.connectionPool = options[kChild].connectionPool;
      this[kEventEmitter] = options[kChild].eventEmitter;
    } else {
      this[kEventEmitter] = new EventEmitter();
      this.serializer = new options.Serializer({
        disablePrototypePoisoningProtection: options.disablePrototypePoisoningProtection,
        enableLongNumeralSupport: options.enableLongNumeralSupport,
      });
      this.connectionPool = new options.ConnectionPool({
        pingTimeout: options.pingTimeout,
        resurrectStrategy: options.resurrectStrategy,
        ssl: options.ssl,
        agent: options.agent,
        proxy: options.proxy,
        Connection: options.Connection,
        auth: options.auth,
        emit: this[kEventEmitter].emit.bind(this[kEventEmitter]),
        sniffEnabled:
          options.sniffInterval !== false ||
          options.sniffOnStart !== false ||
          options.sniffOnConnectionFault !== false,
      });
      // Add the connections before initialize the Transport
      this.connectionPool.addConnection(options.node || options.nodes);
    }

    this.transport = new options.Transport({
      emit: this[kEventEmitter].emit.bind(this[kEventEmitter]),
      connectionPool: this.connectionPool,
      serializer: this.serializer,
      maxRetries: options.maxRetries,
      requestTimeout: options.requestTimeout,
      sniffInterval: options.sniffInterval,
      sniffOnStart: options.sniffOnStart,
      sniffOnConnectionFault: options.sniffOnConnectionFault,
      sniffEndpoint: options.sniffEndpoint,
      suggestCompression: options.suggestCompression,
      compression: options.compression,
      headers: options.headers,
      nodeFilter: options.nodeFilter,
      nodeSelector: options.nodeSelector,
      generateRequestId: options.generateRequestId,
      name: options.name,
      opaqueIdPrefix: options.opaqueIdPrefix,
      context: options.context,
      memoryCircuitBreaker: options.memoryCircuitBreaker,
    });

    this.helpers = new Helpers({
      client: this,
      maxRetries: options.maxRetries,
    });
  }

  get emit() {
    return this[kEventEmitter].emit.bind(this[kEventEmitter]);
  }

  get on() {
    return this[kEventEmitter].on.bind(this[kEventEmitter]);
  }

  get once() {
    return this[kEventEmitter].once.bind(this[kEventEmitter]);
  }

  get off() {
    return this[kEventEmitter].off.bind(this[kEventEmitter]);
  }

  extend(name, opts, fn) {
    if (typeof opts === 'function') {
      fn = opts;
      opts = {};
    }

    let [namespace, method] = name.split('.');
    if (method == null) {
      method = namespace;
      namespace = null;
    }

    if (namespace != null) {
      if (this[namespace] != null && this[namespace][method] != null && opts.force !== true) {
        throw new Error(`The method "${method}" already exists on namespace "${namespace}"`);
      }

      if (this[namespace] == null) this[namespace] = {};
      this[namespace][method] = fn({
        makeRequest: this.transport.request.bind(this.transport),
        result: { body: null, statusCode: null, headers: null, warnings: null },
        ConfigurationError,
      });
    } else {
      if (this[method] != null && opts.force !== true) {
        throw new Error(`The method "${method}" already exists`);
      }

      this[method] = fn({
        makeRequest: this.transport.request.bind(this.transport),
        result: { body: null, statusCode: null, headers: null, warnings: null },
        ConfigurationError,
      });
    }

    this[kExtensions].push({ name, opts, fn });
  }

  child(opts) {
    // Merge the new options with the initial ones
    const options = Object.assign({}, this[kInitialOptions], opts);
    // Pass to the child client the parent instances that cannot be overriden
    options[kChild] = {
      connectionPool: this.connectionPool,
      serializer: this.serializer,
      eventEmitter: this[kEventEmitter],
      initialOptions: options,
    };

    /* istanbul ignore else */
    if (options.auth !== undefined) {
      options.headers = prepareHeaders(options.headers, options.auth);
    }

    const client = new Client(options);
    // sync compatible check
    const tSymbol = Object.getOwnPropertySymbols(this.transport).filter(
      (symbol) => symbol.description === 'compatible check'
    )[0];
    client.transport[tSymbol] = this.transport[tSymbol];
    // Add parent extensions
    if (this[kExtensions].length > 0) {
      this[kExtensions].forEach(({ name, opts, fn }) => {
        client.extend(name, opts, fn);
      });
    }
    return client;
  }

  close(callback) {
    if (callback == null) {
      return new Promise((resolve) => {
        this.close(resolve);
      });
    }
    debug('Closing the client');
    this.connectionPool.empty(callback);
  }
}

function getAuth(node) {
  if (Array.isArray(node)) {
    for (const url of node) {
      const auth = getUsernameAndPassword(url);
      if (auth.username !== '' && auth.password !== '') {
        return auth;
      }
    }

    return null;
  }

  const auth = getUsernameAndPassword(node);
  if (auth.username !== '' && auth.password !== '') {
    return auth;
  }

  return null;

  function getUsernameAndPassword(node) {
    /* istanbul ignore else */
    if (typeof node === 'string') {
      const { username, password } = new URL(node);
      return {
        username: decodeURIComponent(username),
        password: decodeURIComponent(password),
      };
    } else if (node.url instanceof URL) {
      return {
        username: decodeURIComponent(node.url.username),
        password: decodeURIComponent(node.url.password),
      };
    }
  }
}

const events = {
  RESPONSE: 'response',
  REQUEST: 'request',
  SNIFF: 'sniff',
  RESURRECT: 'resurrect',
  SERIALIZATION: 'serialization',
  DESERIALIZATION: 'deserialization',
};

module.exports = {
  Client,
  Transport,
  ConnectionPool,
  Connection,
  Serializer,
  events,
  errors,
};