[Search Sessions] Use sync config (#90138)

* Search Sessions: Unskip Flaky Functional Test

* Save all search sessions and then manage them based on their persisted state

* Get default search session expiration from config

* randomize sleep time

* fix test

* fix test

* Make sure we poll, and dont persist, searches not in the context of a session

* Added keepalive unit tests

* fix ts

* code review @lukasolson

* ts

* More tests, rename onScreenTimeout to completedTimeout

* lint

* lint

* Delete async seaches

* Support saved object pagination
Fix get search status tests

* better PersistedSearchSessionSavedObjectAttributes ts

* test titles

* Remove runAt from monitoring task
Increase testing trackingInterval (caused bug)

* support workload histograms that take into account overdue tasks

* Update touched when changing session status to complete \ error

* removed test

* Updated management test data

* Rename configs

* delete tap first
add comments

* Use sync config in data-enhanced plugin

* fix merge

* fix merge

* ts

* code review

Co-authored-by: Timothy Sullivan <tsullivan@elastic.co>
Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>
Co-authored-by: Anton Dosov <anton.dosov@elastic.co>
Co-authored-by: Gidi Meir Morris <github@gidi.io>

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

@@ -6,7 +6,6 @@
  */
 
 import { CoreSetup, CoreStart, Logger, Plugin, PluginInitializerContext } from 'kibana/server';
-import { Observable } from 'rxjs';
 import { TaskManagerSetupContract, TaskManagerStartContract } from '../../task_manager/server';
 import {
   PluginSetup as DataPluginSetup,
@@ -40,11 +39,11 @@ export class EnhancedDataServerPlugin
   implements Plugin<void, void, SetupDependencies, StartDependencies> {
   private readonly logger: Logger;
   private sessionService!: SearchSessionService;
-  private config$: Observable<ConfigSchema>;
+  private config: ConfigSchema;
 
   constructor(private initializerContext: PluginInitializerContext<ConfigSchema>) {
     this.logger = initializerContext.logger.get('data_enhanced');
-    this.config$ = this.initializerContext.config.create();
+    this.config = this.initializerContext.config.get<ConfigSchema>();
   }
 
   public setup(core: CoreSetup<DataPluginStart>, deps: SetupDependencies) {
@@ -56,7 +55,7 @@ export class EnhancedDataServerPlugin
     deps.data.search.registerSearchStrategy(
       ENHANCED_ES_SEARCH_STRATEGY,
       enhancedEsSearchStrategyProvider(
-        this.config$,
+        this.config,
         this.initializerContext.config.legacy.globalConfig$,
         this.logger,
         usage
@@ -68,10 +67,7 @@ export class EnhancedDataServerPlugin
       eqlSearchStrategyProvider(this.logger)
     );
 
-    this.sessionService = new SearchSessionService(
-      this.logger,
-      this.initializerContext.config.create()
-    );
+    this.sessionService = new SearchSessionService(this.logger, this.config);
 
     deps.data.__enhance({
       search: {

x-pack/plugins/data_enhanced/server/search/es_search_strategy.test.ts

@@ -72,13 +72,13 @@ describe('ES search strategy', () => {
     },
   });
 
-  const mockConfig$ = new BehaviorSubject<any>({
+  const mockConfig: any = {
     search: {
       sessions: {
         defaultExpiration: moment.duration('1', 'm'),
       },
     },
-  });
+  };
 
   beforeEach(() => {
     mockApiCaller.mockClear();
@@ -89,7 +89,7 @@ describe('ES search strategy', () => {
 
   it('returns a strategy with `search and `cancel`', async () => {
     const esSearch = await enhancedEsSearchStrategyProvider(
-      mockConfig$,
+      mockConfig,
       mockLegacyConfig$,
       mockLogger
     );
@@ -104,7 +104,7 @@ describe('ES search strategy', () => {
 
         const params = { index: 'logstash-*', body: { query: {} } };
         const esSearch = await enhancedEsSearchStrategyProvider(
-          mockConfig$,
+          mockConfig,
           mockLegacyConfig$,
           mockLogger
         );
@@ -123,7 +123,7 @@ describe('ES search strategy', () => {
 
         const params = { index: 'logstash-*', body: { query: {} } };
         const esSearch = await enhancedEsSearchStrategyProvider(
-          mockConfig$,
+          mockConfig,
           mockLegacyConfig$,
           mockLogger
         );
@@ -142,7 +142,7 @@ describe('ES search strategy', () => {
 
         const params = { index: 'foo-*', body: {} };
         const esSearch = await enhancedEsSearchStrategyProvider(
-          mockConfig$,
+          mockConfig,
           mockLegacyConfig$,
           mockLogger
         );
@@ -160,7 +160,7 @@ describe('ES search strategy', () => {
 
         const params = { index: 'foo-程', body: {} };
         const esSearch = await enhancedEsSearchStrategyProvider(
-          mockConfig$,
+          mockConfig,
           mockLegacyConfig$,
           mockLogger
         );
@@ -189,7 +189,7 @@ describe('ES search strategy', () => {
 
         const params = { index: 'logstash-*', body: { query: {} } };
         const esSearch = await enhancedEsSearchStrategyProvider(
-          mockConfig$,
+          mockConfig,
           mockLegacyConfig$,
           mockLogger
         );
@@ -209,7 +209,7 @@ describe('ES search strategy', () => {
 
         const params = { index: 'logstash-*', body: { query: {} } };
         const esSearch = await enhancedEsSearchStrategyProvider(
-          mockConfig$,
+          mockConfig,
           mockLegacyConfig$,
           mockLogger
         );
@@ -237,7 +237,7 @@ describe('ES search strategy', () => {
 
       const params = { index: 'logstash-*', body: { query: {} } };
       const esSearch = await enhancedEsSearchStrategyProvider(
-        mockConfig$,
+        mockConfig,
         mockLegacyConfig$,
         mockLogger
       );
@@ -262,7 +262,7 @@ describe('ES search strategy', () => {
 
       const params = { index: 'logstash-*', body: { query: {} } };
       const esSearch = await enhancedEsSearchStrategyProvider(
-        mockConfig$,
+        mockConfig,
         mockLegacyConfig$,
         mockLogger
       );
@@ -287,7 +287,7 @@ describe('ES search strategy', () => {
 
       const id = 'some_id';
       const esSearch = await enhancedEsSearchStrategyProvider(
-        mockConfig$,
+        mockConfig,
         mockLegacyConfig$,
         mockLogger
       );
@@ -311,7 +311,7 @@ describe('ES search strategy', () => {
 
       const id = 'some_id';
       const esSearch = await enhancedEsSearchStrategyProvider(
-        mockConfig$,
+        mockConfig,
         mockLegacyConfig$,
         mockLogger
       );
@@ -338,7 +338,7 @@ describe('ES search strategy', () => {
       const id = 'some_other_id';
       const keepAlive = '1d';
       const esSearch = await enhancedEsSearchStrategyProvider(
-        mockConfig$,
+        mockConfig,
         mockLegacyConfig$,
         mockLogger
       );
@@ -357,7 +357,7 @@ describe('ES search strategy', () => {
       const id = 'some_other_id';
       const keepAlive = '1d';
       const esSearch = await enhancedEsSearchStrategyProvider(
-        mockConfig$,
+        mockConfig,
         mockLegacyConfig$,
         mockLogger
       );

x-pack/plugins/data_enhanced/server/search/es_search_strategy.ts

@@ -39,7 +39,7 @@ import { ConfigSchema } from '../../config';
 import { getKbnServerError, KbnServerError } from '../../../../../src/plugins/kibana_utils/server';
 
 export const enhancedEsSearchStrategyProvider = (
-  config$: Observable<ConfigSchema>,
+  config: ConfigSchema,
   legacyConfig$: Observable<SharedGlobalConfig>,
   logger: Logger,
   usage?: SearchUsage
@@ -60,7 +60,6 @@ export const enhancedEsSearchStrategyProvider = (
     const client = esClient.asCurrentUser.asyncSearch;
 
     const search = async () => {
-      const config = await config$.pipe(first()).toPromise();
       const params = id
         ? getDefaultAsyncGetParams(options)
         : {

x-pack/plugins/data_enhanced/server/search/session/monitoring_task.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { Observable } from 'rxjs';
-import { first } from 'rxjs/operators';
 import { Duration } from 'moment';
 import {
   TaskManagerSetupContract,
@@ -24,14 +22,13 @@ export const SEARCH_SESSIONS_TASK_ID = `data_enhanced_${SEARCH_SESSIONS_TASK_TYP
 interface SearchSessionTaskDeps {
   taskManager: TaskManagerSetupContract;
   logger: Logger;
-  config$: Observable<ConfigSchema>;
+  config: ConfigSchema;
 }
 
-function searchSessionRunner(core: CoreSetup, { logger, config$ }: SearchSessionTaskDeps) {
+function searchSessionRunner(core: CoreSetup, { logger, config }: SearchSessionTaskDeps) {
   return ({ taskInstance }: RunContext) => {
     return {
       async run() {
-        const config = await config$.pipe(first()).toPromise();
         const sessionConfig = config.search.sessions;
         const [coreStart] = await core.getStartServices();
         const internalRepo = coreStart.savedObjects.createInternalRepository([SEARCH_SESSION_TYPE]);

x-pack/plugins/data_enhanced/server/search/session/session_service.test.ts

@@ -5,7 +5,6 @@
  * 2.0.
  */
 
-import { BehaviorSubject } from 'rxjs';
 import {
   SavedObject,
   SavedObjectsClientContract,
@@ -46,7 +45,7 @@ describe('SearchSessionService', () => {
 
   beforeEach(async () => {
     savedObjectsClient = savedObjectsClientMock.create();
-    const config$ = new BehaviorSubject<ConfigSchema>({
+    const config: ConfigSchema = {
       search: {
         sessions: {
           enabled: true,
@@ -59,13 +58,13 @@ describe('SearchSessionService', () => {
           management: {} as any,
         },
       },
-    });
+    };
     const mockLogger: any = {
       debug: jest.fn(),
       warn: jest.fn(),
       error: jest.fn(),
     };
-    service = new SearchSessionService(mockLogger, config$);
+    service = new SearchSessionService(mockLogger, config);
     const coreStart = coreMock.createStart();
     const mockTaskManager = taskManagerMock.createStart();
     await flushPromises();

x-pack/plugins/data_enhanced/server/search/session/session_service.ts

@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { Observable } from 'rxjs';
-import { first } from 'rxjs/operators';
 import {
   CoreSetup,
   CoreStart,
@@ -50,32 +48,33 @@ function sleep(ms: number) {
 }
 export class SearchSessionService
   implements ISearchSessionService<SearchSessionSavedObjectAttributes> {
-  private config!: SearchSessionsConfig;
+  private sessionConfig: SearchSessionsConfig;
 
-  constructor(
-    private readonly logger: Logger,
-    private readonly config$: Observable<ConfigSchema>
-  ) {}
+  constructor(private readonly logger: Logger, private readonly config: ConfigSchema) {
+    this.sessionConfig = this.config.search.sessions;
+  }
 
   public setup(core: CoreSetup, deps: SetupDependencies) {
     registerSearchSessionsTask(core, {
-      config$: this.config$,
+      config: this.config,
       taskManager: deps.taskManager,
       logger: this.logger,
     });
   }
 
   public async start(core: CoreStart, deps: StartDependencies) {
-    const configPromise = await this.config$.pipe(first()).toPromise();
-    this.config = (await configPromise).search.sessions;
     return this.setupMonitoring(core, deps);
   }
 
   public stop() {}
 
   private setupMonitoring = async (core: CoreStart, deps: StartDependencies) => {
-    if (this.config.enabled) {
-      scheduleSearchSessionsTasks(deps.taskManager, this.logger, this.config.trackingInterval);
+    if (this.sessionConfig.enabled) {
+      scheduleSearchSessionsTasks(
+        deps.taskManager,
+        this.logger,
+        this.sessionConfig.trackingInterval
+      );
     }
   };
 
@@ -107,7 +106,7 @@ export class SearchSessionService
         } catch (createError) {
           if (
             SavedObjectsErrorHelpers.isConflictError(createError) &&
-            retry < this.config.maxUpdateRetries
+            retry < this.sessionConfig.maxUpdateRetries
           ) {
             return await retryOnConflict(createError);
           } else {
@@ -116,7 +115,7 @@ export class SearchSessionService
         }
       } else if (
         SavedObjectsErrorHelpers.isConflictError(e) &&
-        retry < this.config.maxUpdateRetries
+        retry < this.sessionConfig.maxUpdateRetries
       ) {
         return await retryOnConflict(e);
       } else {
@@ -164,7 +163,7 @@ export class SearchSessionService
         sessionId,
         status: SearchSessionStatus.IN_PROGRESS,
         expires: new Date(
-          Date.now() + this.config.defaultExpiration.asMilliseconds()
+          Date.now() + this.sessionConfig.defaultExpiration.asMilliseconds()
         ).toISOString(),
         created: new Date().toISOString(),
         touched: new Date().toISOString(),