Expose max_concurrent_shard_requests for _msearch requests (#22379)…

… (#22930)

Allows Kibana users to configure the max_concurrent_shard_requests param used by Kibana when sending _msearch requests. Exposes the config as an advanced setting. By default we won't send the param at all, relying on the ES default instead.

docs/management/advanced-options.asciidoc

@@ -43,6 +43,7 @@ working on big documents. Set this property to `false` to disable highlighting.
 the Elasticsearch cluster. This setting constrains the length of the segment list. Long segment lists can significantly
 increase request processing time.
 `courier:ignoreFilterIfFieldNotInIndex`:: Set this property to `true` to skip filters that apply to fields that don't exist in a visualization's index. Useful when dashboards consist of visualizations from multiple index patterns.
+`courier:maxConcurrentShardRequests`:: Controls the {ref}/search-multi-search.html[max_concurrent_shard_requests] setting used for _msearch requests sent by Kibana. Set to 0 to disable this config and use the Elasticsearch default.
 `fields:popularLimit`:: This setting governs how many of the top most popular fields are shown.
 `histogram:barTarget`:: When date histograms use the `auto` interval, Kibana attempts to generate this number of bars.
 `histogram:maxBars`:: Date histograms are not generated with more bars than the value of this property, scaling values

src/core_plugins/kibana/ui_setting_defaults.js

@@ -186,6 +186,15 @@ export function getUiSettingDefaults() {
         used when <strong>courier:setRequestPreference</strong> is set to "custom".`,
       category: ['search'],
     },
+    'courier:maxConcurrentShardRequests': {
+      name: 'Max Concurrent Shard Requests',
+      value: 0,
+      type: 'number',
+      description: `Controls the <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/search-multi-search.html"
+        target="_blank" rel="noopener noreferrer">max_concurrent_shard_requests</a>
+        setting used for _msearch requests sent by Kibana. Set to 0 to disable this config and use the Elasticsearch default.`,
+      category: ['search'],
+    },
     'fields:popularLimit': {
       name: 'Popular fields limit',
       value: 10,

src/ui/public/courier/fetch/call_client.js

@@ -24,7 +24,7 @@ import { MergeDuplicatesRequestProvider } from './merge_duplicate_requests';
 import { RequestStatus } from './req_status';
 import { SerializeFetchParamsProvider } from './request/serialize_fetch_params';
 
-export function CallClientProvider(Private, Promise, es) {
+export function CallClientProvider(Private, Promise, es, config) {
   const errorAllowExplicitIndex = Private(ErrorAllowExplicitIndexProvider);
   const isRequest = Private(IsRequestProvider);
   const mergeDuplicateRequests = Private(MergeDuplicatesRequestProvider);
@@ -34,6 +34,8 @@ export function CallClientProvider(Private, Promise, es) {
   const DUPLICATE = RequestStatus.DUPLICATE;
 
   function callClient(searchRequests) {
+    const maxConcurrentShardRequests = config.get('courier:maxConcurrentShardRequests');
+
     // merging docs can change status to DUPLICATE, capture new statuses
     const searchRequestsAndStatuses = mergeDuplicateRequests(searchRequests);
 
@@ -136,7 +138,7 @@ export function CallClientProvider(Private, Promise, es) {
           searching,
           abort,
           failedSearchRequests,
-        } = await searchStrategy.search({ searchRequests, es, Promise, serializeFetchParams });
+        } = await searchStrategy.search({ searchRequests, es, Promise, serializeFetchParams, maxConcurrentShardRequests });
 
         // Collect searchRequests which have successfully been sent.
         searchRequests.forEach(searchRequest => {

src/ui/public/courier/search_strategy/default_search_strategy.js

@@ -57,7 +57,7 @@ async function serializeAllFetchParams(fetchParams, searchRequests, serializeFet
 export const defaultSearchStrategy = {
   id: 'default',
 
-  search: async ({ searchRequests, es, Promise, serializeFetchParams }) => {
+  search: async ({ searchRequests, es, Promise, serializeFetchParams, maxConcurrentShardRequests = 0 }) => {
     // Flatten the searchSource within each searchRequest to get the fetch params,
     // e.g. body, filters, index pattern, query.
     const allFetchParams = await getAllFetchParams(searchRequests, Promise);
@@ -68,7 +68,15 @@ export const defaultSearchStrategy = {
       failedSearchRequests,
     } = await serializeAllFetchParams(allFetchParams, searchRequests, serializeFetchParams);
 
-    const searching = es.msearch({ body: serializedFetchParams });
+    const msearchParams = {
+      body: serializedFetchParams,
+    };
+
+    if (maxConcurrentShardRequests !== 0) {
+      msearchParams.max_concurrent_shard_requests = maxConcurrentShardRequests;
+    }
+
+    const searching = es.msearch(msearchParams);
 
     return {
       // Unwrap the responses object returned by the es client.

src/ui/public/courier/search_strategy/default_search_strategy.test.js

@@ -0,0 +1,55 @@
+/*
+ * 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 { defaultSearchStrategy } from './default_search_strategy';
+import { Promise } from 'bluebird';
+
+const { search } = defaultSearchStrategy;
+
+describe('defaultSearchStrategy', function () {
+
+  describe('search', function () {
+
+    let searchArgs;
+
+    beforeEach(() => {
+      const msearchMock = jest.fn().mockReturnValue(Promise.resolve([]));
+
+      searchArgs = {
+        searchRequests: [],
+        es: { msearch: msearchMock },
+        Promise,
+        serializeFetchParams: () => Promise.resolve([]),
+      };
+    });
+
+    test('does not send max_concurrent_shard_requests by default', async () => {
+      await search(searchArgs);
+      expect(searchArgs.es.msearch.mock.calls[0][0]).not.toHaveProperty('max_concurrent_shard_requests');
+    });
+
+    test('allows configuration of max_concurrent_shard_requests', async () => {
+      searchArgs.maxConcurrentShardRequests = 42;
+      await search(searchArgs);
+      expect(searchArgs.es.msearch.mock.calls[0][0].max_concurrent_shard_requests).toBe(42);
+    });
+
+  });
+
+});