Reactively disable Task Manager lifecycle when core services become unavailable (#81779) (#81991)

gmmorris · web-flow · commit d46b60422462 · 2020-10-29T13:28:09.000Z
Plugs the Task Manager polling lifecycle into the Kibana Services Status streams in order to ensure we reactively start and stop polling whenever the Elasticsearch or SavedObjects service switch between `available` and `unavailable`.

This will prevent Task Manager from polling whenever these services switch to an `unavailable` state.
diff --git a/x-pack/plugins/task_manager/server/monitoring/index.ts b/x-pack/plugins/task_manager/server/monitoring/index.ts
@@ -28,12 +28,20 @@ export {
 export function createMonitoringStats(
   taskPollingLifecycle: TaskPollingLifecycle,
   taskStore: TaskStore,
+  elasticsearchAndSOAvailability$: Observable<boolean>,
   config: TaskManagerConfig,
   managedConfig: ManagedConfiguration,
   logger: Logger
 ): Observable<MonitoringStats> {
   return createMonitoringStatsStream(
-    createAggregators(taskPollingLifecycle, taskStore, config, managedConfig, logger),
+    createAggregators(
+      taskPollingLifecycle,
+      taskStore,
+      elasticsearchAndSOAvailability$,
+      config,
+      managedConfig,
+      logger
+    ),
     config
   );
 }
diff --git a/x-pack/plugins/task_manager/server/monitoring/monitoring_stats_stream.ts b/x-pack/plugins/task_manager/server/monitoring/monitoring_stats_stream.ts
@@ -63,6 +63,7 @@ export interface RawMonitoringStats {
 export function createAggregators(
   taskPollingLifecycle: TaskPollingLifecycle,
   taskStore: TaskStore,
+  elasticsearchAndSOAvailability$: Observable<boolean>,
   config: TaskManagerConfig,
   managedConfig: ManagedConfiguration,
   logger: Logger
@@ -72,6 +73,7 @@ export function createAggregators(
     createTaskRunAggregator(taskPollingLifecycle, config.monitored_stats_running_average_window),
     createWorkloadAggregator(
       taskStore,
+      elasticsearchAndSOAvailability$,
       config.monitored_aggregated_stats_refresh_rate,
       config.poll_interval,
       logger
diff --git a/x-pack/plugins/task_manager/server/monitoring/workload_statistics.test.ts b/x-pack/plugins/task_manager/server/monitoring/workload_statistics.test.ts
@@ -17,6 +17,8 @@ import { ESSearchResponse } from '../../../apm/typings/elasticsearch';
 import { AggregationResultOf } from '../../../apm/typings/elasticsearch/aggregations';
 import { times } from 'lodash';
 import { taskStoreMock } from '../task_store.mock';
+import { of, Subject } from 'rxjs';
+import { sleep } from '../test_utils';
 
 type MockESResult = ESSearchResponse<
   ConcreteTaskInstance,
@@ -75,6 +77,7 @@ describe('Workload Statistics Aggregator', () => {
 
     const workloadAggregator = createWorkloadAggregator(
       taskStore,
+      of(true),
       10,
       3000,
       loggingSystemMock.create().get()
@@ -231,6 +234,7 @@ describe('Workload Statistics Aggregator', () => {
 
     const workloadAggregator = createWorkloadAggregator(
       taskStore,
+      of(true),
       10,
       3000,
       loggingSystemMock.create().get()
@@ -252,12 +256,51 @@ describe('Workload Statistics Aggregator', () => {
     });
   });
 
+  test('skips summary of the workload when services are unavailable', async () => {
+    const taskStore = taskStoreMock.create({});
+    taskStore.aggregate.mockResolvedValue(mockAggregatedResult());
+
+    const availability$ = new Subject<boolean>();
+
+    const workloadAggregator = createWorkloadAggregator(
+      taskStore,
+      availability$,
+      10,
+      3000,
+      loggingSystemMock.create().get()
+    );
+
+    return new Promise(async (resolve) => {
+      workloadAggregator.pipe(first()).subscribe((result) => {
+        expect(result.key).toEqual('workload');
+        expect(result.value).toMatchObject({
+          count: 4,
+          task_types: {
+            actions_telemetry: { count: 2, status: { idle: 2 } },
+            alerting_telemetry: { count: 1, status: { idle: 1 } },
+            session_cleanup: { count: 1, status: { idle: 1 } },
+          },
+        });
+        resolve();
+      });
+
+      availability$.next(false);
+
+      await sleep(10);
+      expect(taskStore.aggregate).not.toHaveBeenCalled();
+      await sleep(10);
+      expect(taskStore.aggregate).not.toHaveBeenCalled();
+      availability$.next(true);
+    });
+  });
+
   test('returns a count of the overdue workload', async () => {
     const taskStore = taskStoreMock.create({});
     taskStore.aggregate.mockResolvedValue(mockAggregatedResult());
 
     const workloadAggregator = createWorkloadAggregator(
       taskStore,
+      of(true),
       10,
       3000,
       loggingSystemMock.create().get()
@@ -280,6 +323,7 @@ describe('Workload Statistics Aggregator', () => {
 
     const workloadAggregator = createWorkloadAggregator(
       taskStore,
+      of(true),
       10,
       3000,
       loggingSystemMock.create().get()
@@ -307,6 +351,7 @@ describe('Workload Statistics Aggregator', () => {
 
     const workloadAggregator = createWorkloadAggregator(
       taskStore,
+      of(true),
       60 * 1000,
       3000,
       loggingSystemMock.create().get()
@@ -344,6 +389,7 @@ describe('Workload Statistics Aggregator', () => {
 
     const workloadAggregator = createWorkloadAggregator(
       taskStore,
+      of(true),
       15 * 60 * 1000,
       3000,
       loggingSystemMock.create().get()
@@ -392,7 +438,7 @@ describe('Workload Statistics Aggregator', () => {
         })
       );
     const logger = loggingSystemMock.create().get();
-    const workloadAggregator = createWorkloadAggregator(taskStore, 10, 3000, logger);
+    const workloadAggregator = createWorkloadAggregator(taskStore, of(true), 10, 3000, logger);
 
     return new Promise((resolve, reject) => {
       workloadAggregator.pipe(take(2), bufferCount(2)).subscribe((results) => {
diff --git a/x-pack/plugins/task_manager/server/monitoring/workload_statistics.ts b/x-pack/plugins/task_manager/server/monitoring/workload_statistics.ts
@@ -4,8 +4,8 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { timer } from 'rxjs';
-import { mergeMap, map, catchError } from 'rxjs/operators';
+import { combineLatest, Observable, timer } from 'rxjs';
+import { mergeMap, map, filter, catchError } from 'rxjs/operators';
 import { Logger } from 'src/core/server';
 import { JsonObject } from 'src/plugins/kibana_utils/common';
 import { keyBy, mapValues } from 'lodash';
@@ -94,6 +94,7 @@ const MAX_SHCEDULE_DENSITY_BUCKETS = 50;
 
 export function createWorkloadAggregator(
   taskStore: TaskStore,
+  elasticsearchAndSOAvailability$: Observable<boolean>,
   refreshInterval: number,
   pollInterval: number,
   logger: Logger
@@ -105,7 +106,8 @@ export function createWorkloadAggregator(
     MAX_SHCEDULE_DENSITY_BUCKETS
   );
 
-  return timer(0, refreshInterval).pipe(
+  return combineLatest([timer(0, refreshInterval), elasticsearchAndSOAvailability$]).pipe(
+    filter(([, areElasticsearchAndSOAvailable]) => areElasticsearchAndSOAvailable),
     mergeMap(() =>
       taskStore.aggregate({
         aggs: {
diff --git a/x-pack/plugins/task_manager/server/plugin.test.ts b/x-pack/plugins/task_manager/server/plugin.test.ts
@@ -4,9 +4,12 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { TaskManagerPlugin } from './plugin';
+import { TaskManagerPlugin, getElasticsearchAndSOAvailability } from './plugin';
 import { coreMock } from '../../../../src/core/server/mocks';
 import { TaskManagerConfig } from './config';
+import { Subject } from 'rxjs';
+import { bufferCount, take } from 'rxjs/operators';
+import { CoreStatus, ServiceStatusLevels } from 'src/core/server';
 
 describe('TaskManagerPlugin', () => {
   describe('setup', () => {
@@ -88,4 +91,99 @@ describe('TaskManagerPlugin', () => {
       );
     });
   });
+
+  describe('getElasticsearchAndSOAvailability', () => {
+    test('returns true when both services are available', async () => {
+      const core$ = new Subject<CoreStatus>();
+
+      const availability = getElasticsearchAndSOAvailability(core$)
+        .pipe(take(1), bufferCount(1))
+        .toPromise();
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: true, savedObjects: true }));
+
+      expect(await availability).toEqual([true]);
+    });
+
+    test('returns false when both services are unavailable', async () => {
+      const core$ = new Subject<CoreStatus>();
+
+      const availability = getElasticsearchAndSOAvailability(core$)
+        .pipe(take(1), bufferCount(1))
+        .toPromise();
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: false, savedObjects: false }));
+
+      expect(await availability).toEqual([false]);
+    });
+
+    test('returns false when one service is unavailable but the other is available', async () => {
+      const core$ = new Subject<CoreStatus>();
+
+      const availability = getElasticsearchAndSOAvailability(core$)
+        .pipe(take(1), bufferCount(1))
+        .toPromise();
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: true, savedObjects: false }));
+
+      expect(await availability).toEqual([false]);
+    });
+
+    test('shift back and forth between values as status changes', async () => {
+      const core$ = new Subject<CoreStatus>();
+
+      const availability = getElasticsearchAndSOAvailability(core$)
+        .pipe(take(3), bufferCount(3))
+        .toPromise();
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: true, savedObjects: false }));
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: true, savedObjects: true }));
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: false, savedObjects: false }));
+
+      expect(await availability).toEqual([false, true, false]);
+    });
+
+    test(`skips values when the status hasn't changed`, async () => {
+      const core$ = new Subject<CoreStatus>();
+
+      const availability = getElasticsearchAndSOAvailability(core$)
+        .pipe(take(3), bufferCount(3))
+        .toPromise();
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: true, savedObjects: false }));
+
+      // still false, so shouldn't emit a second time
+      core$.next(mockCoreStatusAvailability({ elasticsearch: false, savedObjects: true }));
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: true, savedObjects: true }));
+
+      // shouldn't emit as already true
+      core$.next(mockCoreStatusAvailability({ elasticsearch: true, savedObjects: true }));
+
+      core$.next(mockCoreStatusAvailability({ elasticsearch: false, savedObjects: false }));
+
+      expect(await availability).toEqual([false, true, false]);
+    });
+  });
 });
+
+function mockCoreStatusAvailability({
+  elasticsearch,
+  savedObjects,
+}: {
+  elasticsearch: boolean;
+  savedObjects: boolean;
+}) {
+  return {
+    elasticsearch: {
+      level: elasticsearch ? ServiceStatusLevels.available : ServiceStatusLevels.unavailable,
+      summary: '',
+    },
+    savedObjects: {
+      level: savedObjects ? ServiceStatusLevels.available : ServiceStatusLevels.unavailable,
+      summary: '',
+    },
+  };
+}
diff --git a/x-pack/plugins/task_manager/server/plugin.ts b/x-pack/plugins/task_manager/server/plugin.ts
@@ -3,9 +3,17 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-import { PluginInitializerContext, Plugin, CoreSetup, Logger, CoreStart } from 'src/core/server';
-import { combineLatest, Subject } from 'rxjs';
-import { first, map } from 'rxjs/operators';
+import { combineLatest, Observable, Subject } from 'rxjs';
+import { first, map, distinctUntilChanged } from 'rxjs/operators';
+import {
+  PluginInitializerContext,
+  Plugin,
+  CoreSetup,
+  Logger,
+  CoreStart,
+  ServiceStatusLevels,
+  CoreStatus,
+} from '../../../../src/core/server';
 import { TaskDefinition } from './task';
 import { TaskPollingLifecycle } from './polling_lifecycle';
 import { TaskManagerConfig } from './config';
@@ -37,6 +45,7 @@ export class TaskManagerPlugin
   private logger: Logger;
   private definitions: TaskTypeDictionary;
   private middleware: Middleware = createInitialMiddleware();
+  private elasticsearchAndSOAvailability$?: Observable<boolean>;
   private monitoringStats$ = new Subject<MonitoringStats>();
 
   constructor(private readonly initContext: PluginInitializerContext) {
@@ -51,6 +60,8 @@ export class TaskManagerPlugin
       .pipe(first())
       .toPromise();
 
+    this.elasticsearchAndSOAvailability$ = getElasticsearchAndSOAvailability(core.status.core$);
+
     setupSavedObjects(core.savedObjects, this.config);
     this.taskManagerId = this.initContext.env.instanceUuid;
 
@@ -115,19 +126,20 @@ export class TaskManagerPlugin
       startingPollInterval: this.config!.poll_interval,
     });
 
-    const taskPollingLifecycle = new TaskPollingLifecycle({
+    this.taskPollingLifecycle = new TaskPollingLifecycle({
       config: this.config!,
       definitions: this.definitions,
       logger: this.logger,
       taskStore,
       middleware: this.middleware,
+      elasticsearchAndSOAvailability$: this.elasticsearchAndSOAvailability$!,
       ...managedConfiguration,
     });
-    this.taskPollingLifecycle = taskPollingLifecycle;
 
     createMonitoringStats(
-      taskPollingLifecycle,
+      this.taskPollingLifecycle,
       taskStore,
+      this.elasticsearchAndSOAvailability$!,
       this.config!,
       managedConfiguration,
       this.logger
@@ -137,12 +149,9 @@ export class TaskManagerPlugin
       logger: this.logger,
       taskStore,
       middleware: this.middleware,
-      taskPollingLifecycle,
+      taskPollingLifecycle: this.taskPollingLifecycle,
     });
 
-    // start polling for work
-    taskPollingLifecycle.start();
-
     return {
       fetch: (opts: SearchOpts): Promise<FetchResult> => taskStore.fetch(opts),
       get: (id: string) => taskStore.get(id),
@@ -153,12 +162,6 @@ export class TaskManagerPlugin
     };
   }
 
-  public stop() {
-    if (this.taskPollingLifecycle) {
-      this.taskPollingLifecycle.stop();
-    }
-  }
-
   /**
    * Ensures task manager hasn't started
    *
@@ -171,3 +174,16 @@ export class TaskManagerPlugin
     }
   }
 }
+
+export function getElasticsearchAndSOAvailability(
+  core$: Observable<CoreStatus>
+): Observable<boolean> {
+  return core$.pipe(
+    map(
+      ({ elasticsearch, savedObjects }) =>
+        elasticsearch.level === ServiceStatusLevels.available &&
+        savedObjects.level === ServiceStatusLevels.available
+    ),
+    distinctUntilChanged()
+  );
+}
diff --git a/x-pack/plugins/task_manager/server/polling_lifecycle.mock.ts b/x-pack/plugins/task_manager/server/polling_lifecycle.mock.ts
diff --git a/x-pack/plugins/task_manager/server/polling_lifecycle.test.ts b/x-pack/plugins/task_manager/server/polling_lifecycle.test.ts
diff --git a/x-pack/plugins/task_manager/server/polling_lifecycle.ts b/x-pack/plugins/task_manager/server/polling_lifecycle.ts
