[Search] [WIP] Add shard delay aggregation

src/plugins/data/common/search/aggs/buckets/shard_delay.ts

@@ -0,0 +1,43 @@
+/*
+ * 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 { BucketAggType } from './bucket_agg_type';
+
+export const SHARD_DELAY_AGG_NAME = 'shard_delay';
+
+export const getShardDelayBucketAgg = () =>
+  new BucketAggType({
+    name: SHARD_DELAY_AGG_NAME,
+    title: 'Shard Delay',
+    createFilter: () => ({ match_all: {} }),
+    customLabels: false,
+    params: [
+      {
+        name: 'delay',
+        type: 'string',
+        default: '5s',
+        write(aggConfig, output) {
+          output.params = {
+            ...output.params,
+            value: aggConfig.params.delay,
+          };
+        },
+      },
+    ],
+  });

src/plugins/data/config.ts

@@ -28,6 +28,13 @@ export const configSchema = schema.object({
       enabled: schema.boolean({ defaultValue: true }),
     }),
   }),
+  search: schema.object({
+    aggs: schema.object({
+      shardDelay: schema.object({
+        enabled: schema.boolean({ defaultValue: false }),
+      }),
+    }),
+  }),
 });
 
 export type ConfigSchema = TypeOf<typeof configSchema>;

src/plugins/data/public/plugin.ts

@@ -96,7 +96,7 @@ export class DataPublicPlugin
   private readonly storage: IStorageWrapper;
 
   constructor(initializerContext: PluginInitializerContext<ConfigSchema>) {
-    this.searchService = new SearchService();
+    this.searchService = new SearchService(initializerContext);
     this.queryService = new QueryService();
     this.fieldFormatsService = new FieldFormatsService();
     this.autocomplete = new AutocompleteService(initializerContext);

src/plugins/data/public/search/search_service.ts

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { Plugin, CoreSetup, CoreStart } from 'src/core/public';
+import { Plugin, CoreSetup, CoreStart, PluginInitializerContext } from 'src/core/public';
 import { BehaviorSubject } from 'rxjs';
 import { ISearchSetup, ISearchStart, SearchEnhancements } from './types';
 
@@ -30,6 +30,11 @@ import { SearchUsageCollector, createUsageCollector } from './collectors';
 import { UsageCollectionSetup } from '../../../usage_collection/public';
 import { esdsl, esRawResponse } from './expressions';
 import { ExpressionsSetup } from '../../../expressions/public';
+import { ConfigSchema } from '../../config';
+import {
+  SHARD_DELAY_AGG_NAME,
+  getShardDelayBucketAgg,
+} from '../../common/search/aggs/buckets/shard_delay';
 
 /** @internal */
 export interface SearchServiceSetupDependencies {
@@ -48,6 +53,8 @@ export class SearchService implements Plugin<ISearchSetup, ISearchStart> {
   private searchInterceptor!: ISearchInterceptor;
   private usageCollector?: SearchUsageCollector;
 
+  constructor(private initializerContext: PluginInitializerContext<ConfigSchema>) {}
+
   public setup(
     { http, getStartServices, injectedMetadata, notifications, uiSettings }: CoreSetup,
     { expressions, usageCollection }: SearchServiceSetupDependencies
@@ -69,11 +76,17 @@ export class SearchService implements Plugin<ISearchSetup, ISearchStart> {
     expressions.registerFunction(esdsl);
     expressions.registerType(esRawResponse);
 
+    const aggs = this.aggsService.setup({
+      registerFunction: expressions.registerFunction,
+      uiSettings,
+    });
+
+    if (this.initializerContext.config.get().search.aggs.shardDelay.enabled) {
+      aggs.types.registerBucket(SHARD_DELAY_AGG_NAME, getShardDelayBucketAgg);
+    }
+
     return {
-      aggs: this.aggsService.setup({
-        registerFunction: expressions.registerFunction,
-        uiSettings,
-      }),
+      aggs,
       usageCollector: this.usageCollector!,
       __enhance: (enhancements: SearchEnhancements) => {
         this.searchInterceptor = enhancements.searchInterceptor;

src/plugins/data/server/index.ts

@@ -286,6 +286,7 @@ export {
 export const config: PluginConfigDescriptor<ConfigSchema> = {
   exposeToBrowser: {
     autocomplete: true,
+    search: true,
   },
   schema: configSchema,
 };

src/plugins/data/server/search/search_service.ts

@@ -28,6 +28,7 @@ import {
   SharedGlobalConfig,
   StartServicesAccessor,
 } from 'src/core/server';
+import { first } from 'rxjs/operators';
 import { ISearchSetup, ISearchStart, ISearchStrategy, SearchEnhancements } from './types';
 
 import { AggsService, AggsSetupDependencies } from './aggs';
@@ -41,6 +42,11 @@ import { registerUsageCollector } from './collectors/register';
 import { usageProvider } from './collectors/usage';
 import { searchTelemetry } from '../saved_objects';
 import { IEsSearchRequest, IEsSearchResponse, ISearchOptions } from '../../common';
+import {
+  getShardDelayBucketAgg,
+  SHARD_DELAY_AGG_NAME,
+} from '../../common/search/aggs/buckets/shard_delay';
+import { ConfigSchema } from '../../config';
 
 type StrategyMap<
   SearchStrategyRequest extends IEsSearchRequest = IEsSearchRequest,
@@ -70,7 +76,7 @@ export class SearchService implements Plugin<ISearchSetup, ISearchStart> {
   private searchStrategies: StrategyMap<any, any> = {};
 
   constructor(
-    private initializerContext: PluginInitializerContext,
+    private initializerContext: PluginInitializerContext<ConfigSchema>,
     private readonly logger: Logger
   ) {}
 
@@ -102,13 +108,25 @@ export class SearchService implements Plugin<ISearchSetup, ISearchStart> {
       registerUsageCollector(usageCollection, this.initializerContext);
     }
 
+    const aggs = this.aggsService.setup({ registerFunction });
+
+    this.initializerContext.config
+      .create<ConfigSchema>()
+      .pipe(first())
+      .toPromise()
+      .then((value) => {
+        if (value.search.aggs.shardDelay.enabled) {
+          aggs.types.registerBucket(SHARD_DELAY_AGG_NAME, getShardDelayBucketAgg);
+        }
+      });
+
     return {
       __enhance: (enhancements: SearchEnhancements) => {
         if (this.searchStrategies.hasOwnProperty(enhancements.defaultStrategy)) {
           this.defaultSearchStrategyName = enhancements.defaultStrategy;
         }
       },
-      aggs: this.aggsService.setup({ registerFunction }),
+      aggs,
       registerSearchStrategy: this.registerSearchStrategy,
       usage,
     };