From 560e2c370cb8a29c2b1b303caf20a52b815f5b21 Mon Sep 17 00:00:00 2001
From: Gidi Meir Morris <github@gidi.io>
Date: Fri, 20 Nov 2020 12:04:05 +0000
Subject: [PATCH] ensure workload agg doesnt run until next interval when it
 fails (#83632) (#83894)

Ensures the WorkloadAggregator doesn't retry immediately after errors, and instead retries on the next interval.
---
 .../monitoring/workload_statistics.test.ts    | 35 +++++++++++++++++++
 .../server/monitoring/workload_statistics.ts  |  6 ++--
 2 files changed, 38 insertions(+), 3 deletions(-)

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
index c2e62b6e1898b..3470ee4d76486 100644
--- a/x-pack/plugins/task_manager/server/monitoring/workload_statistics.test.ts
+++ b/x-pack/plugins/task_manager/server/monitoring/workload_statistics.test.ts
@@ -477,6 +477,41 @@ describe('Workload Statistics Aggregator', () => {
       }, reject);
     });
   });
+
+  test('recovery after errors occurrs at the next interval', async () => {
+    const refreshInterval = 1000;
+
+    const taskStore = taskStoreMock.create({});
+    const logger = loggingSystemMock.create().get();
+    const workloadAggregator = createWorkloadAggregator(
+      taskStore,
+      of(true),
+      refreshInterval,
+      3000,
+      logger
+    );
+
+    return new Promise((resolve, reject) => {
+      let errorWasThrowAt = 0;
+      taskStore.aggregate.mockImplementation(async () => {
+        if (errorWasThrowAt === 0) {
+          errorWasThrowAt = Date.now();
+          throw new Error(`Elasticsearch has gone poof`);
+        } else if (Date.now() - errorWasThrowAt < refreshInterval) {
+          reject(new Error(`Elasticsearch is still poof`));
+        }
+
+        return setTaskTypeCount(mockAggregatedResult(), 'alerting_telemetry', {
+          idle: 2,
+        });
+      });
+
+      workloadAggregator.pipe(take(2), bufferCount(2)).subscribe((results) => {
+        expect(results.length).toEqual(2);
+        resolve();
+      }, reject);
+    });
+  });
 });
 
 describe('estimateRecurringTaskScheduling', () => {
diff --git a/x-pack/plugins/task_manager/server/monitoring/workload_statistics.ts b/x-pack/plugins/task_manager/server/monitoring/workload_statistics.ts
index a27b5e2282e32..8002ee44d01ff 100644
--- a/x-pack/plugins/task_manager/server/monitoring/workload_statistics.ts
+++ b/x-pack/plugins/task_manager/server/monitoring/workload_statistics.ts
@@ -5,7 +5,7 @@
  */
 
 import { combineLatest, Observable, timer } from 'rxjs';
-import { mergeMap, map, filter, catchError } from 'rxjs/operators';
+import { mergeMap, map, filter, switchMap, catchError } from 'rxjs/operators';
 import { Logger } from 'src/core/server';
 import { JsonObject } from 'src/plugins/kibana_utils/common';
 import { keyBy, mapValues } from 'lodash';
@@ -222,8 +222,8 @@ export function createWorkloadAggregator(
     }),
     catchError((ex: Error, caught) => {
       logger.error(`[WorkloadAggregator]: ${ex}`);
-      // continue to pull values from the same observable
-      return caught;
+      // continue to pull values from the same observable but only on the next refreshInterval
+      return timer(refreshInterval).pipe(switchMap(() => caught));
     })
   );
 }