Create HTTP Agent manager (#137748)

* Create HTTP Agent factory

* Properly extract agent options

* Use independent Agent for preboot

* Create AgentManager to obtain factories

* Make client type mandatory, fix outdated mocks

* Temporarily force new Agent creation

* Revert changes in utils

* Add correct defaults for Agent Options, support proxy agent.

* Forgot to push package.json

* Add hpagent dependency in BUILD.bazel

* Get rid of hpagent (proxy param is not exposed in kibana.yml)

* Remove hpagent from BUILD.bazel

* Use different agents for normal Vs scoped client

* Fix Agent constructor params

* Fix incorrect access to err.message

* Use separate Agent for scoped client

* Create different agents for std vs scoped

* Provide different Agent instances if config differs

* Create a new Agent for each ES Client

* Restructure agent store. Add UTs

* Remove obsolete comment

* Simplify AgentManager store structure (no type needed)

* Fine tune client_config return type

* Misc enhancements following PR comments

* Fix missing param in cli_setup/utils

packages/core/elasticsearch/core-elasticsearch-client-server-internal/index.ts

@@ -9,6 +9,7 @@
 export { ScopedClusterClient } from './src/scoped_cluster_client';
 export { ClusterClient } from './src/cluster_client';
 export { configureClient } from './src/configure_client';
+export { AgentManager } from './src/agent_manager';
 export { getRequestDebugMeta, getErrorMessage } from './src/log_query_and_deprecation';
 export {
   PRODUCT_RESPONSE_HEADER,

packages/core/elasticsearch/core-elasticsearch-client-server-internal/src/agent_manager.test.ts

@@ -0,0 +1,125 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { AgentManager } from './agent_manager';
+import { Agent as HttpAgent } from 'http';
+import { Agent as HttpsAgent } from 'https';
+
+jest.mock('http');
+jest.mock('https');
+
+const HttpAgentMock = HttpAgent as jest.Mock<HttpAgent>;
+const HttpsAgentMock = HttpsAgent as jest.Mock<HttpsAgent>;
+
+describe('AgentManager', () => {
+  afterEach(() => {
+    HttpAgentMock.mockClear();
+    HttpsAgentMock.mockClear();
+  });
+
+  describe('#getAgentFactory()', () => {
+    it('provides factories which are different at each call', () => {
+      const agentManager = new AgentManager();
+      const agentFactory1 = agentManager.getAgentFactory();
+      const agentFactory2 = agentManager.getAgentFactory();
+      expect(agentFactory1).not.toEqual(agentFactory2);
+    });
+
+    describe('one agent factory', () => {
+      it('provides instances of the http and https Agent classes', () => {
+        const mockedHttpAgent = new HttpAgent();
+        HttpAgentMock.mockImplementationOnce(() => mockedHttpAgent);
+        const mockedHttpsAgent = new HttpsAgent();
+        HttpsAgentMock.mockImplementationOnce(() => mockedHttpsAgent);
+        const agentManager = new AgentManager();
+        const agentFactory = agentManager.getAgentFactory();
+        const httpAgent = agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        const httpsAgent = agentFactory({ url: new URL('https://elastic-node-1:9200') });
+        expect(httpAgent).toEqual(mockedHttpAgent);
+        expect(httpsAgent).toEqual(mockedHttpsAgent);
+      });
+
+      it('provides Agents with a valid default configuration', () => {
+        const agentManager = new AgentManager();
+        const agentFactory = agentManager.getAgentFactory();
+        agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        expect(HttpAgent).toBeCalledTimes(1);
+        expect(HttpAgent).toBeCalledWith({
+          keepAlive: true,
+          keepAliveMsecs: 50000,
+          maxFreeSockets: 256,
+          maxSockets: 256,
+          scheduling: 'lifo',
+        });
+      });
+
+      it('takes into account the provided configurations', () => {
+        const agentManager = new AgentManager({ maxFreeSockets: 32, maxSockets: 2048 });
+        const agentFactory = agentManager.getAgentFactory({
+          maxSockets: 1024,
+          scheduling: 'fifo',
+        });
+        agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        expect(HttpAgent).toBeCalledTimes(1);
+        expect(HttpAgent).toBeCalledWith({
+          keepAlive: true,
+          keepAliveMsecs: 50000,
+          maxFreeSockets: 32,
+          maxSockets: 1024,
+          scheduling: 'fifo',
+        });
+      });
+
+      it('provides Agents that match the URLs protocol', () => {
+        const agentManager = new AgentManager();
+        const agentFactory = agentManager.getAgentFactory();
+        agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        expect(HttpAgent).toHaveBeenCalledTimes(1);
+        expect(HttpsAgent).toHaveBeenCalledTimes(0);
+        agentFactory({ url: new URL('https://elastic-node-3:9200') });
+        expect(HttpAgent).toHaveBeenCalledTimes(1);
+        expect(HttpsAgent).toHaveBeenCalledTimes(1);
+      });
+
+      it('provides the same Agent iif URLs use the same protocol', () => {
+        const agentManager = new AgentManager();
+        const agentFactory = agentManager.getAgentFactory();
+        const agent1 = agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        const agent2 = agentFactory({ url: new URL('http://elastic-node-2:9200') });
+        const agent3 = agentFactory({ url: new URL('https://elastic-node-3:9200') });
+        const agent4 = agentFactory({ url: new URL('https://elastic-node-4:9200') });
+
+        expect(agent1).toEqual(agent2);
+        expect(agent1).not.toEqual(agent3);
+        expect(agent3).toEqual(agent4);
+      });
+
+      it('dereferences an agent instance when the agent is closed', () => {
+        const agentManager = new AgentManager();
+        const agentFactory = agentManager.getAgentFactory();
+        const agent = agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        // eslint-disable-next-line dot-notation
+        expect(agentManager['httpStore'].has(agent)).toEqual(true);
+        agent.destroy();
+        // eslint-disable-next-line dot-notation
+        expect(agentManager['httpStore'].has(agent)).toEqual(false);
+      });
+    });
+
+    describe('two agent factories', () => {
+      it('never provide the same Agent instance even if they use the same type', () => {
+        const agentManager = new AgentManager();
+        const agentFactory1 = agentManager.getAgentFactory();
+        const agentFactory2 = agentManager.getAgentFactory();
+        const agent1 = agentFactory1({ url: new URL('http://elastic-node-1:9200') });
+        const agent2 = agentFactory2({ url: new URL('http://elastic-node-1:9200') });
+        expect(agent1).not.toEqual(agent2);
+      });
+    });
+  });
+});

packages/core/elasticsearch/core-elasticsearch-client-server-internal/src/agent_manager.ts

@@ -0,0 +1,89 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Agent as HttpAgent } from 'http';
+import { Agent as HttpsAgent } from 'https';
+import { ConnectionOptions, HttpAgentOptions } from '@elastic/elasticsearch';
+
+const HTTPS = 'https:';
+const DEFAULT_CONFIG: HttpAgentOptions = {
+  keepAlive: true,
+  keepAliveMsecs: 50000,
+  maxSockets: 256,
+  maxFreeSockets: 256,
+  scheduling: 'lifo',
+};
+
+export type NetworkAgent = HttpAgent | HttpsAgent;
+export type AgentFactory = (connectionOpts: ConnectionOptions) => NetworkAgent;
+
+/**
+ * Allows obtaining Agent factories, which can then be fed into elasticsearch-js's Client class.
+ * Ideally, we should obtain one Agent factory for each ES Client class.
+ * This allows using the same Agent across all the Pools and Connections of the Client (one per ES node).
+ *
+ * Agent instances are stored internally to allow collecting metrics (nbr of active/idle connections to ES).
+ *
+ * Using the same Agent factory across multiple ES Client instances is strongly discouraged, cause ES Client
+ * exposes methods that can modify the underlying pools, effectively impacting the connections of other Clients.
+ * @internal
+ **/
+export class AgentManager {
+  // Stores Https Agent instances
+  private httpsStore: Set<HttpsAgent>;
+  // Stores Http Agent instances
+  private httpStore: Set<HttpAgent>;
+
+  constructor(private agentOptions: HttpAgentOptions = DEFAULT_CONFIG) {
+    this.httpsStore = new Set();
+    this.httpStore = new Set();
+  }
+
+  public getAgentFactory(agentOptions?: HttpAgentOptions): AgentFactory {
+    // a given agent factory always provides the same Agent instances (for the same protocol)
+    // we keep references to the instances at factory level, to be able to reuse them
+    let httpAgent: HttpAgent;
+    let httpsAgent: HttpsAgent;
+
+    return (connectionOpts: ConnectionOptions): NetworkAgent => {
+      if (connectionOpts.url.protocol === HTTPS) {
+        if (!httpsAgent) {
+          const config = Object.assign(
+            {},
+            DEFAULT_CONFIG,
+            this.agentOptions,
+            agentOptions,
+            connectionOpts.tls
+          );
+          httpsAgent = new HttpsAgent(config);
+          this.httpsStore.add(httpsAgent);
+          dereferenceOnDestroy(this.httpsStore, httpsAgent);
+        }
+
+        return httpsAgent;
+      }
+
+      if (!httpAgent) {
+        const config = Object.assign({}, DEFAULT_CONFIG, this.agentOptions, agentOptions);
+        httpAgent = new HttpAgent(config);
+        this.httpStore.add(httpAgent);
+        dereferenceOnDestroy(this.httpStore, httpAgent);
+      }
+
+      return httpAgent;
+    };
+  }
+}
+
+const dereferenceOnDestroy = (protocolStore: Set<NetworkAgent>, agent: NetworkAgent) => {
+  const doDestroy = agent.destroy.bind(agent);
+  agent.destroy = () => {
+    protocolStore.delete(agent);
+    doDestroy();
+  };
+};

packages/core/elasticsearch/core-elasticsearch-client-server-internal/src/client_config.ts

@@ -9,10 +9,12 @@
 import { ConnectionOptions as TlsConnectionOptions } from 'tls';
 import { URL } from 'url';
 import { Duration } from 'moment';
-import type { ClientOptions } from '@elastic/elasticsearch/lib/client';
+import type { ClientOptions, HttpAgentOptions } from '@elastic/elasticsearch';
 import type { ElasticsearchClientConfig } from '@kbn/core-elasticsearch-server';
 import { DEFAULT_HEADERS } from './headers';
 
+export type ParsedClientOptions = Omit<ClientOptions, 'agent'> & { agent: HttpAgentOptions };
+
 /**
  * Parse the client options from given client config and `scoped` flag.
  *
@@ -23,8 +25,8 @@ import { DEFAULT_HEADERS } from './headers';
 export function parseClientOptions(
   config: ElasticsearchClientConfig,
   scoped: boolean
-): ClientOptions {
-  const clientOptions: ClientOptions = {
+): ParsedClientOptions {
+  const clientOptions: ParsedClientOptions = {
     sniffOnStart: config.sniffOnStart,
     sniffOnConnectionFault: config.sniffOnConnectionFault,
     headers: {