Create HTTP Agent manager

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,138 @@
+/*
+ * 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 { loggingSystemMock } from '@kbn/core-logging-server-mocks';
+import { AgentManager } from './agent_manager';
+import { Agent as HttpAgent } from 'http';
+import { Agent as HttpsAgent } from 'https';
+
+jest.mock('http');
+jest.mock('https');
+
+describe('AgentManager', () => {
+  let logger: ReturnType<typeof loggingSystemMock.createLogger>;
+
+  beforeEach(() => {
+    logger = loggingSystemMock.createLogger();
+  });
+
+  afterEach(() => {
+    const HttpAgentMock = HttpAgent as jest.Mock<HttpAgent>;
+    const HttpsAgentMock = HttpsAgent as jest.Mock<HttpsAgent>;
+    HttpAgentMock.mockClear();
+    HttpsAgentMock.mockClear();
+  });
+
+  describe('provides agent factories', () => {
+    it('with a valid default configuration', () => {
+      const agentManager = new AgentManager(logger);
+      const agentFactory = agentManager.getAgentFactory('test');
+      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 configuration provided in the constructor', () => {
+      const agentManager = new AgentManager(logger);
+      const agentFactory = agentManager.getAgentFactory('test', {
+        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: 256,
+        maxSockets: 1024,
+        scheduling: 'fifo',
+      });
+    });
+
+    it('which are different at each call', () => {
+      const agentManager = new AgentManager(logger);
+      const agentFactory1 = agentManager.getAgentFactory('test');
+      const agentFactory2 = agentManager.getAgentFactory('test');
+      expect(agentFactory1).not.toEqual(agentFactory2);
+    });
+
+    it('throws an error when an Agent factory is requested using undici params', () => {
+      const agentManager = new AgentManager(logger);
+      expect(() => {
+        agentManager.getAgentFactory('anotherTest', { keepAliveTimeout: 2000 });
+      }).toThrowError();
+    });
+
+    describe('when configured with a custom factory', () => {
+      it('uses the provided factory', () => {
+        const customAgent = new HttpsAgent({ maxSockets: 32 });
+        expect(HttpsAgent).toBeCalledTimes(1);
+        const factory = () => customAgent;
+        const agentManager = new AgentManager(logger);
+        const agentFactory = agentManager.getAgentFactory('test', factory);
+        const agent = agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        expect(HttpsAgent).toBeCalledTimes(1);
+        expect(agent).toEqual(customAgent);
+      });
+    });
+
+    describe('one agent factory', () => {
+      it('provides Agents that match the URLs protocol', () => {
+        const agentManager = new AgentManager(logger);
+        const agentFactory = agentManager.getAgentFactory('test');
+        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(logger);
+        const agentFactory = agentManager.getAgentFactory('test');
+        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(logger);
+        const agentFactory = agentManager.getAgentFactory('test');
+        const agent = agentFactory({ url: new URL('http://elastic-node-1:9200') });
+        // eslint-disable-next-line dot-notation
+        expect(agentManager['agentStore']['test']['http:'][0]).toEqual(agent);
+        agent.destroy();
+        // eslint-disable-next-line dot-notation
+        expect(agentManager['agentStore']['test']['http:'][0]).toBeUndefined();
+      });
+    });
+
+    describe('two agent factories', () => {
+      it('never provide the same Agent instance even if they use the same type', () => {
+        const agentManager = new AgentManager(logger);
+        const agentFactory1 = agentManager.getAgentFactory('test');
+        const agentFactory2 = agentManager.getAgentFactory('test');
+        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,188 @@
+/*
+ * 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, UndiciAgentOptions } from '@elastic/elasticsearch';
+import { Logger } from '@kbn/logging';
+
+const HTTP = 'http:';
+const HTTPS = 'https:';
+
+export type Protocol = typeof HTTP | typeof HTTPS;
+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 {
+  /**
+   * Store Agent instances by type and protocol, e.g.:
+   *
+   * data: {
+   *   'http:': [agentInstance1, agentInstance2],
+   *   'https:': [agentInstance3, agentInstance4]
+   * },
+   * monitoring: {
+   *   'http:': [],
+   *   'https:': [agentInstance5]
+   * }
+   */
+  private agentStore: Record<string, Record<Protocol, Array<NetworkAgent | undefined>>>;
+  private logger: Logger;
+
+  constructor(
+    logger: Logger,
+    private defaultConfig: HttpAgentOptions = {
+      keepAlive: true,
+      keepAliveMsecs: 50000,
+      maxSockets: 256,
+      maxFreeSockets: 256,
+      scheduling: 'lifo',
+    }
+  ) {
+    this.logger = logger;
+    this.agentStore = {};
+  }
+
+  public getAgentFactory(
+    type: string,
+    agentOptions?: HttpAgentOptions | UndiciAgentOptions | AgentFactory | false
+  ): AgentFactory {
+    if (isAgentFactory(agentOptions)) {
+      // use the user-provided factory directly
+      return agentOptions;
+    }
+
+    const agentConfig = assertValidAgentConfig(agentOptions);
+
+    // a given agent factory (of a given type) always provides the same Agent instances (for the same protocol)
+    // we keep the indexes for each protocol, so that we can access them later on, when the factory is invoked
+    const { httpInstanceIndex, httpsInstanceIndex } = this.initTypeStore(type);
+
+    return (connectionOpts: ConnectionOptions): NetworkAgent => {
+      const agents = this.getAgentInstances(type, connectionOpts);
+
+      // when the factory is called for a given connection (and protocol), we use the appropriate index
+      const agentIndex = isHttps(connectionOpts) ? httpsInstanceIndex : httpInstanceIndex;
+
+      let agent = agents[agentIndex];
+
+      if (!agent) {
+        agent = this.createAgent(agentConfig, connectionOpts);
+
+        // Allow GC'ing the Agent after it has been terminated
+        const doDestroy = agent.destroy.bind(agent);
+        agent.destroy = () => {
+          doDestroy();
+          agents[agentIndex] = undefined;
+        };
+
+        // add the new Agent instance to the list of instances of that type
+        agents[agentIndex] = agent;
+      }
+
+      return agent;
+    };
+  }
+
+  private initTypeStore(type: string): { httpInstanceIndex: number; httpsInstanceIndex: number } {
+    let agentsOfType = this.agentStore[type];
+
+    if (!agentsOfType) {
+      agentsOfType = {
+        [HTTP]: [],
+        [HTTPS]: [],
+      };
+
+      this.agentStore[type] = agentsOfType;
+    } else {
+      this.logger.warn(
+        `An Agent factory for type ${type} has already been retrieved.
+        This is not optimal since the existing instance cannot be reused.
+        Please consider reusing the previously retrieved factory instead`
+      );
+    }
+
+    const httpInstanceIndex = agentsOfType[HTTP].length;
+    const httpsInstanceIndex = agentsOfType[HTTPS].length;
+
+    agentsOfType[HTTP].push(undefined);
+    agentsOfType[HTTPS].push(undefined);
+    return { httpInstanceIndex, httpsInstanceIndex };
+  }
+
+  private getAgentInstances(
+    type: string,
+    connectionOpts: ConnectionOptions
+  ): Array<NetworkAgent | undefined> {
+    let protocol: Protocol;
+
+    switch (connectionOpts.url.protocol) {
+      case 'http:':
+        protocol = HTTP;
+        break;
+      case 'https:':
+        protocol = HTTPS;
+        break;
+      default:
+        throw new Error(`Unsupported protocol: '${connectionOpts.url.protocol}'`);
+    }
+
+    return this.agentStore[type][protocol];
+  }
+
+  private createAgent(agentConfig: HttpAgentOptions, connectionOpts: ConnectionOptions) {
+    const config = Object.assign({}, this.defaultConfig, agentConfig, connectionOpts.tls);
+    return isHttps(connectionOpts) ? new HttpsAgent(config) : new HttpAgent(config);
+  }
+}
+
+const isHttps = (connectionOpts: ConnectionOptions): boolean => {
+  return connectionOpts.url.protocol === HTTPS;
+};
+
+const assertValidAgentConfig = (
+  agentOptions?: HttpAgentOptions | UndiciAgentOptions | false
+): HttpAgentOptions => {
+  if (!agentOptions) {
+    return {};
+  } else if (isHttpAgentOptions(agentOptions)) {
+    return agentOptions;
+  }
+
+  throw new Error('Unsupported agent options: UndiciAgentOptions');
+};
+
+const isAgentFactory = (
+  agentOptions?: HttpAgentOptions | UndiciAgentOptions | AgentFactory | false
+): agentOptions is AgentFactory => {
+  return typeof agentOptions === 'function';
+};
+
+const isHttpAgentOptions = (
+  opts: HttpAgentOptions | UndiciAgentOptions
+): opts is HttpAgentOptions => {
+  return (
+    !('keepAliveTimeout' in opts) &&
+    !('keepAliveMaxTimeout' in opts) &&
+    !('keepAliveTimeoutThreshold' in opts) &&
+    !('pipelining' in opts) &&
+    !('maxHeaderSize' in opts) &&
+    !('connections' in opts)
+  );
+};

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

@@ -83,11 +83,13 @@ describe('ClusterClient', () => {
     expect(configureClientMock).toHaveBeenCalledTimes(2);
     expect(configureClientMock).toHaveBeenCalledWith(config, {
       logger,
+      agentManager: undefined,
       type: 'custom-type',
       getExecutionContext: getExecutionContextMock,
     });
     expect(configureClientMock).toHaveBeenCalledWith(config, {
       logger,
+      agentManager: undefined,
       type: 'custom-type',
       getExecutionContext: getExecutionContextMock,
       scoped: true,

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

@@ -26,6 +26,7 @@ import { ScopedClusterClient } from './scoped_cluster_client';
 import { DEFAULT_HEADERS } from './headers';
 import { createInternalErrorHandler, InternalUnauthorizedErrorHandler } from './retry_unauthorized';
 import { createTransport } from './create_transport';
+import { AgentManager } from './agent_manager';
 
 const noop = () => undefined;
 
@@ -47,25 +48,33 @@ export class ClusterClient implements ICustomClusterClient {
     authHeaders,
     getExecutionContext = noop,
     getUnauthorizedErrorHandler = noop,
+    agentManager,
   }: {
     config: ElasticsearchClientConfig;
     logger: Logger;
     type: string;
     authHeaders?: IAuthHeadersStorage;
     getExecutionContext?: () => string | undefined;
     getUnauthorizedErrorHandler?: () => UnauthorizedErrorHandler | undefined;
+    agentManager?: AgentManager;
   }) {
     this.config = config;
     this.authHeaders = authHeaders;
     this.getExecutionContext = getExecutionContext;
     this.getUnauthorizedErrorHandler = getUnauthorizedErrorHandler;
 
-    this.asInternalUser = configureClient(config, { logger, type, getExecutionContext });
+    this.asInternalUser = configureClient(config, {
+      logger,
+      type,
+      getExecutionContext,
+      agentManager,
+    });
     this.rootScopedClient = configureClient(config, {
       logger,
       type,
       getExecutionContext,
       scoped: true,
+      agentManager,
     });
   }