Add new elasticsearch client

package.json

@@ -125,6 +125,7 @@
     "@elastic/apm-rum": "^5.2.0",
     "@elastic/charts": "19.5.2",
     "@elastic/datemath": "5.0.3",
+    "@elastic/elasticsearch": "^7.7.1",
     "@elastic/ems-client": "7.9.3",
     "@elastic/eui": "24.1.0",
     "@elastic/filesaver": "1.1.2",
@@ -293,7 +294,6 @@
   "devDependencies": {
     "@babel/parser": "^7.10.2",
     "@babel/types": "^7.10.2",
-    "@elastic/elasticsearch": "^7.4.0",
     "@elastic/eslint-config-kibana": "0.15.0",
     "@elastic/eslint-plugin-eui": "0.0.2",
     "@elastic/github-checks-reporter": "0.0.20b3",

src/core/server/elasticsearch/client/client_config.ts

@@ -0,0 +1,145 @@
+/*
+ * 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.
+ */
+
+import { ConnectionOptions as TlsConnectionOptions } from 'tls';
+import { URL } from 'url';
+import { Duration } from 'moment';
+import { ClientOptions, NodeOptions } from '@elastic/elasticsearch';
+import { ElasticsearchConfig } from '../elasticsearch_config';
+
+/**
+ * @privateRemarks Config that consumers can pass to the Elasticsearch JS client is complex and includes
+ * not only entries from standard `elasticsearch.*` yaml config, but also some Elasticsearch JS
+ * client specific options like `keepAlive` or `plugins` (that eventually will be deprecated).
+ *
+ * @public
+ */
+export type ElasticsearchClientConfig = Pick<
+  ElasticsearchConfig,
+  | 'customHeaders'
+  | 'logQueries'
+  | 'sniffOnStart'
+  | 'sniffOnConnectionFault'
+  | 'requestHeadersWhitelist'
+  | 'sniffInterval'
+  | 'hosts'
+  | 'username'
+  | 'password'
+> & {
+  pingTimeout?: ElasticsearchConfig['pingTimeout'] | ClientOptions['pingTimeout'];
+  requestTimeout?: ElasticsearchConfig['requestTimeout'] | ClientOptions['requestTimeout'];
+  ssl?: Partial<ElasticsearchConfig['ssl']>;
+};
+
+export function parseClientOptions(
+  config: ElasticsearchClientConfig,
+  scoped: boolean
+): ClientOptions {
+  const clientOptions: ClientOptions = {
+    sniffOnStart: config.sniffOnStart,
+    sniffOnConnectionFault: config.sniffOnConnectionFault,
+    headers: config.customHeaders,
+  };
+
+  if (config.pingTimeout != null) {
+    clientOptions.pingTimeout = getDurationAsMs(config.pingTimeout);
+  }
+  if (config.requestTimeout != null) {
+    clientOptions.requestTimeout = getDurationAsMs(config.requestTimeout);
+  }
+  if (config.sniffInterval) {
+    clientOptions.sniffInterval = getDurationAsMs(config.sniffInterval);
+  }
+
+  // TODO: this can either be done globally here or by host in convertHost.
+  //       Not sure which option is the best.
+  if (config.username && config.password) {
+    clientOptions.auth = {
+      username: config.username,
+      password: config.password,
+    };
+  }
+
+  clientOptions.nodes = config.hosts.map((host) => convertHost(host, !scoped, config));
+
+  if (config.ssl) {
+    clientOptions.ssl = generateSslConfig(
+      config.ssl,
+      scoped && !config.ssl.alwaysPresentCertificate
+    );
+  }
+
+  return clientOptions;
+}
+
+const generateSslConfig = (
+  sslConfig: Required<ElasticsearchClientConfig>['ssl'],
+  ignoreCertAndKey: boolean
+): TlsConnectionOptions => {
+  const ssl: TlsConnectionOptions = {
+    ca: sslConfig.certificateAuthorities,
+  };
+
+  const verificationMode = sslConfig.verificationMode;
+  switch (verificationMode) {
+    case 'none':
+      ssl.rejectUnauthorized = false;
+      break;
+    case 'certificate':
+      ssl.rejectUnauthorized = true;
+      // by default, NodeJS is checking the server identify
+      ssl.checkServerIdentity = () => undefined;
+      break;
+    case 'full':
+      ssl.rejectUnauthorized = true;
+      break;
+    default:
+      throw new Error(`Unknown ssl verificationMode: ${verificationMode}`);
+  }
+
+  // Add client certificate and key if required by elasticsearch
+  if (!ignoreCertAndKey && sslConfig.certificate && sslConfig.key) {
+    ssl.cert = sslConfig.certificate;
+    ssl.key = sslConfig.key;
+    ssl.passphrase = sslConfig.keyPassphrase;
+  }
+
+  return ssl;
+};
+
+const convertHost = (
+  host: string,
+  needAuth: boolean,
+  { username, password }: ElasticsearchClientConfig
+): NodeOptions => {
+  const url = new URL(host);
+  const isHTTPS = url.protocol === 'https:';
+  url.port = url.port ?? isHTTPS ? '443' : '80';
+  if (needAuth && username && password) {
+    url.username = username;
+    url.password = password;
+  }
+
+  return {
+    url,
+  };
+};
+
+const getDurationAsMs = (duration: number | Duration) =>
+  typeof duration === 'number' ? duration : duration.asMilliseconds();

src/core/server/elasticsearch/client/client_facade.ts

@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+
+import { ApiResponse } from '@elastic/elasticsearch';
+import {
+  RequestBody,
+  RequestNDBody,
+  TransportRequestOptions,
+  TransportRequestPromise,
+} from '@elastic/elasticsearch/lib/Transport';
+import * as RequestParams from '@elastic/elasticsearch/api/requestParams';
+
+export interface ClientFacade {
+  bulk<
+    TResponse = Record<string, any>,
+    TRequestBody extends RequestNDBody = Array<Record<string, any>>,
+    TContext = unknown
+  >(
+    params?: RequestParams.Bulk<TRequestBody>,
+    options?: TransportRequestOptions
+  ): TransportRequestPromise<ApiResponse<TResponse, TContext>>;
+
+  asyncSearch: {
+    delete<TResponse = Record<string, any>, TContext = unknown>(
+      params?: RequestParams.AsyncSearchDelete,
+      options?: TransportRequestOptions
+    ): TransportRequestPromise<ApiResponse<TResponse, TContext>>;
+    get<TResponse = Record<string, any>, TContext = unknown>(
+      params?: RequestParams.AsyncSearchGet,
+      options?: TransportRequestOptions
+    ): TransportRequestPromise<ApiResponse<TResponse, TContext>>;
+    submit<
+      TResponse = Record<string, any>,
+      TRequestBody extends RequestBody = Record<string, any>,
+      TContext = unknown
+    >(
+      params?: RequestParams.AsyncSearchSubmit<TRequestBody>,
+      options?: TransportRequestOptions
+    ): TransportRequestPromise<ApiResponse<TResponse, TContext>>;
+  };
+}

src/core/server/elasticsearch/client/client_wrapper.ts

@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+
+import { Client } from '@elastic/elasticsearch';
+import { TransportRequestOptions } from '@elastic/elasticsearch/lib/Transport';
+import { Headers } from '../../http/router';
+import { ClientFacade } from './client_facade';
+
+export const getClientFacade = (client: Client, headers: Headers = {}): ClientFacade => {
+  const addHeaders = (options?: TransportRequestOptions): TransportRequestOptions => {
+    if (!options) {
+      return {
+        headers,
+      };
+    }
+    // TODO: do we need to throw in case of duplicates as it was done
+    //       in legacy? - src/core/server/elasticsearch/scoped_cluster_client.ts:L88
+    return {
+      ...options,
+      headers: {
+        ...options.headers,
+        ...headers,
+      },
+    };
+  };
+
+  return {
+    bulk: (params, options) => client.bulk(params, addHeaders(options)),
+    asyncSearch: {
+      delete: (params, options) => client.asyncSearch.delete(params, addHeaders(options)),
+      get: (params, options) => client.asyncSearch.get(params, addHeaders(options)),
+      submit: (params, options) => client.asyncSearch.submit(params, addHeaders(options)),
+    },
+  };
+};

src/core/server/elasticsearch/client/cluster_client.ts

@@ -0,0 +1,71 @@
+/*
+ * 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.
+ */
+
+import { Client } from '@elastic/elasticsearch';
+import { getClientFacade } from './client_wrapper';
+import { ClientFacade } from './client_facade';
+import { configureClient } from './configure_client';
+import { Logger } from '../../logging';
+import { GetAuthHeaders, isRealRequest } from '../../http';
+import { Headers } from '../../http/router';
+import { ElasticsearchClientConfig } from './client_config';
+import { ScopedClusterClient, IScopedClusterClient } from './scoped_cluster_client';
+import { ScopeableRequest } from './types';
+import { ensureRawRequest, filterHeaders } from '../../http/router';
+
+const noop = () => undefined;
+
+interface IClusterClient {
+  asInternalUser: () => ClientFacade;
+  asScoped: (request: ScopeableRequest) => IScopedClusterClient;
+}
+
+export class ClusterClient implements IClusterClient {
+  private readonly internalWrapper: ClientFacade;
+  private readonly scopedClient: Client;
+
+  constructor(
+    private readonly config: ElasticsearchClientConfig,
+    logger: Logger,
+    private readonly getAuthHeaders: GetAuthHeaders = noop
+  ) {
+    this.internalWrapper = getClientFacade(configureClient(config, { logger }));
+    this.scopedClient = configureClient(config, { logger, scoped: true });
+  }
+
+  asInternalUser() {
+    return this.internalWrapper;
+  }
+
+  asScoped(request: ScopeableRequest) {
+    const headers = this.getScopedHeaders(request);
+    const scopedWrapper = getClientFacade(this.scopedClient, headers);
+    return new ScopedClusterClient(this.internalWrapper, scopedWrapper);
+  }
+
+  private getScopedHeaders(request: ScopeableRequest): Headers {
+    if (!isRealRequest(request)) {
+      return request?.headers ?? {};
+    }
+    const authHeaders = this.getAuthHeaders(request);
+    const headers = ensureRawRequest(request).headers;
+
+    return filterHeaders({ ...headers, ...authHeaders }, this.config.requestHeadersWhitelist);
+  }
+}

src/core/server/elasticsearch/client/configure_client.ts

@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+
+import { Client } from '@elastic/elasticsearch';
+import { Logger } from '../../logging';
+import { parseClientOptions, ElasticsearchClientConfig } from './client_config';
+
+export const configureClient = (
+  config: ElasticsearchClientConfig,
+  { logger, scoped = false }: { logger: Logger; scoped?: boolean }
+): Client => {
+  const clientOptions = parseClientOptions(config, scoped);
+  const client = new Client(clientOptions);
+
+  client.on('response', (err, event) => {
+    if (err) {
+      logger.error(`${err.name}: ${err.message}`);
+    } else if (config.logQueries) {
+      const params = event.meta.request.params;
+      logger.debug(
+        `${event.statusCode}\n${params.method} ${params.path}\n${params.querystring?.trim() ?? ''}`,
+        {
+          tags: ['query'],
+        }
+      );
+    }
+  });
+
+  return client;
+};