Add config flag. Increase default value.

x-pack/plugins/ingest_manager/common/types/index.ts

@@ -13,6 +13,7 @@ export interface IngestManagerConfigType {
     enabled: boolean;
     tlsCheckDisabled: boolean;
     pollingRequestTimeout: number;
+    maxConcurrentConnections: number;
     kibana: {
       host?: string;
       ca_sha256?: string;

x-pack/plugins/ingest_manager/server/index.ts

@@ -26,6 +26,7 @@ export const config = {
       enabled: schema.boolean({ defaultValue: true }),
       tlsCheckDisabled: schema.boolean({ defaultValue: false }),
       pollingRequestTimeout: schema.number({ defaultValue: 60000 }),
+      maxConcurrentConnections: schema.number({ defaultValue: 750 }),
       kibana: schema.object({
         host: schema.maybe(schema.string()),
         ca_sha256: schema.maybe(schema.string()),

x-pack/plugins/ingest_manager/server/plugin.ts

@@ -34,7 +34,7 @@ import {
 } from './constants';
 import { registerSavedObjects, registerEncryptedSavedObjects } from './saved_objects';
 import {
-  preAuthHandler,
+  registerLimitedConcurrencyRoutes,
   registerEPMRoutes,
   registerPackageConfigRoutes,
   registerDataStreamRoutes,
@@ -231,7 +231,7 @@ export class IngestManagerPlugin
         } else {
           // we currently only use this global interceptor if fleet is enabled
           // since it would run this func on *every* req (other plugins, CSS, etc)
-          this.httpSetup.registerOnPreAuth(preAuthHandler);
+          registerLimitedConcurrencyRoutes(core, config);
           registerAgentRoutes(router);
           registerEnrollmentApiKeyRoutes(router);
           registerInstallScriptRoutes({

x-pack/plugins/ingest_manager/server/routes/index.ts

@@ -14,4 +14,4 @@ export { registerRoutes as registerInstallScriptRoutes } from './install_script'
 export { registerRoutes as registerOutputRoutes } from './output';
 export { registerRoutes as registerSettingsRoutes } from './settings';
 export { registerRoutes as registerAppRoutes } from './app';
-export { preAuthHandler } from './global_interceptors';
+export { registerLimitedConcurrencyRoutes } from './limited_concurrency';

x-pack/plugins/ingest_manager/server/routes/limited_concurrency.ts

@@ -0,0 +1,72 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import {
+  CoreSetup,
+  KibanaRequest,
+  LifecycleResponseFactory,
+  OnPreAuthToolkit,
+} from 'kibana/server';
+import { LIMITED_CONCURRENCY_ROUTE_TAG } from '../../common';
+import { IngestManagerConfigType } from '../index';
+class MaxCounter {
+  constructor(private readonly max: number = 1) {}
+  private counter = 0;
+  valueOf() {
+    return this.counter;
+  }
+  increase() {
+    if (this.counter < this.max) {
+      this.counter += 1;
+    }
+  }
+  decrease() {
+    if (this.counter > 0) {
+      this.counter -= 1;
+    }
+  }
+  lessThanMax() {
+    return this.counter < this.max;
+  }
+}
+
+function shouldHandleRequest(request: KibanaRequest) {
+  const tags = request.route.options.tags;
+  return tags.includes(LIMITED_CONCURRENCY_ROUTE_TAG);
+}
+
+export function registerLimitedConcurrencyRoutes(core: CoreSetup, config: IngestManagerConfigType) {
+  const max = config.fleet.maxConcurrentConnections;
+  if (!max) return;
+
+  const counter = new MaxCounter(max);
+  core.http.registerOnPreAuth(function preAuthHandler(
+    request: KibanaRequest,
+    response: LifecycleResponseFactory,
+    toolkit: OnPreAuthToolkit
+  ) {
+    if (!shouldHandleRequest(request)) {
+      return toolkit.next();
+    }
+
+    if (!counter.lessThanMax()) {
+      return response.customError({
+        body: 'Too Many Requests',
+        statusCode: 429,
+      });
+    }
+
+    counter.increase();
+
+    // requests.events.aborted$ has a bug (but has test which explicitly verifies) where it's fired even when the request completes
+    // https://github.com/elastic/kibana/pull/70495#issuecomment-656288766
+    request.events.aborted$.toPromise().then(() => {
+      counter.decrease();
+    });
+
+    return toolkit.next();
+  });
+}