[8.16] implements backoff delay in usage counters service (#206363) (#206850)

# Backport This will backport the following commits from `main` to `8.16`: - [implements backoff delay in usage counters service (#206363)](https://github.com/elastic/kibana/pull/206363)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: Christiane (Tina) Heiligers <christiane.heiligers@elastic.co>
2025-04-24 09:48:58 -04:00 · 2025-01-16 07:36:13 +11:00 · 2025-01-16 07:36:13 +11:00 · 6f206300e1
commit 6f206300e1
parent e9944f8a3f
5 changed files with 67 additions and 45 deletions
--- a/src/plugins/usage_collection/common/constants.ts
+++ b/src/plugins/usage_collection/common/constants.ts
@ -10,3 +10,4 @@
 export const KIBANA_STATS_TYPE = 'kibana_stats';
 export const DEFAULT_MAXIMUM_WAIT_TIME_FOR_ALL_COLLECTORS_IN_S = 1;
 export const MAIN_APP_DEFAULT_VIEW_ID = 'main';
+export const USAGE_COUNTERS_BACKOFF_DELAY_IN_MS = 500;
--- a/src/plugins/usage_collection/server/config.ts
+++ b/src/plugins/usage_collection/server/config.ts
@ -14,7 +14,7 @@ import { DEFAULT_MAXIMUM_WAIT_TIME_FOR_ALL_COLLECTORS_IN_S } from '../common/con
 export const configSchema = schema.object({
  usageCounters: schema.object({
    enabled: schema.boolean({ defaultValue: true }),
-    retryCount: schema.number({ defaultValue: 1 }),
+    retryCount: schema.number({ defaultValue: 3 }),
    bufferDuration: schema.duration({ defaultValue: '5s' }),
  }),
  uiCounters: schema.object({
--- a/src/plugins/usage_collection/server/usage_counters/usage_counter.ts
+++ b/src/plugins/usage_collection/server/usage_counters/usage_counter.ts
@ -58,7 +58,6 @@ export class UsageCounter implements IUsageCounter {
      incrementBy = 1,
      namespace,
    } = params;
-
    this.counter$.next({
      domainId: this.domainId,
      counterName,
--- a/src/plugins/usage_collection/server/usage_counters/usage_counters_service.test.ts
+++ b/src/plugins/usage_collection/server/usage_counters/usage_counters_service.test.ts
@ -25,9 +25,10 @@ const registerUsageCountersRollupsMock = registerUsageCountersRollups as jest.Mo
  typeof registerUsageCountersRollups
 >;

-const tick = () => {
+// optionally advance test timers after a delay
+const tickWithDelay = (delay = 1) => {
  jest.useRealTimers();
-  return new Promise((resolve) => setTimeout(resolve, 1));
+  return new Promise((resolve) => setTimeout(resolve, delay));
 };

 describe('UsageCountersService', () => {
@ -188,51 +189,62 @@ describe('UsageCountersService', () => {
      ]
    `);
  });
+  //  requires extended test runtime because of exponential backoff delay for retries
+  it(
+    'retries errors by `retryCount` times before failing to store',
+    async () => {
+      const retryConst = 2;
+      const usageCountersService = new UsageCountersService({
+        logger,
+        retryCount: retryConst,
+        bufferDurationMs: 50000,
+      });

-  it('retries errors by `retryCount` times before failing to store', async () => {
-    const usageCountersService = new UsageCountersService({
-      logger,
-      retryCount: 1,
-      bufferDurationMs,
-    });
+      const mockRepository = coreStart.savedObjects.createInternalRepository();
+      const mockError = new Error('failed');
+      const mockIncrementCounter = jest.fn().mockImplementation((_, key) => {
+        switch (key) {
+          case 'test-counter:counterA:count:server:20210409':
+            throw mockError;
+          case 'test-counter:counterB:count:server:20210409':
+            return 'pass';
+          default:
+            throw new Error(`unknown key ${key}`);
+        }
+      });

-    const mockRepository = coreStart.savedObjects.createInternalRepository();
-    const mockError = new Error('failed.');
-    const mockIncrementCounter = jest.fn().mockImplementation((_, key) => {
-      switch (key) {
-        case 'test-counter:counterA:count:server:20210409':
-          throw mockError;
-        case 'test-counter:counterB:count:server:20210409':
-          return 'pass';
-        default:
-          throw new Error(`unknown key ${key}`);
-      }
-    });
+      mockRepository.incrementCounter = mockIncrementCounter;

-    mockRepository.incrementCounter = mockIncrementCounter;
+      coreStart.savedObjects.createInternalRepository.mockReturnValue(mockRepository);
+      const { createUsageCounter } = usageCountersService.setup(coreSetup);
+      jest.useFakeTimers();
+      const usageCounter = createUsageCounter('test-counter');

-    coreStart.savedObjects.createInternalRepository.mockReturnValue(mockRepository);
-    const { createUsageCounter } = usageCountersService.setup(coreSetup);
-    jest.useFakeTimers();
-    const usageCounter = createUsageCounter('test-counter');
+      usageCountersService.start(coreStart);
+      usageCounter.incrementCounter({ counterName: 'counterA' });
+      usageCounter.incrementCounter({ counterName: 'counterB' });
+      jest.runOnlyPendingTimers();

-    usageCountersService.start(coreStart);
-    usageCounter.incrementCounter({ counterName: 'counterA' });
-    usageCounter.incrementCounter({ counterName: 'counterB' });
-    jest.runOnlyPendingTimers();
-
-    // wait for retries to kick in on next scheduler call
-    await tick();
-    // number of incrementCounter calls + number of retries
-    expect(mockIncrementCounter).toBeCalledTimes(2 + 1);
-    expect(logger.debug).toHaveBeenNthCalledWith(1, 'Store counters into savedObjects', {
-      kibana: {
-        usageCounters: {
-          results: [mockError, 'pass'],
+      // wait for retries to kick in on next scheduler call
+      await tickWithDelay(5000);
+      // number of incrementCounter calls + number of retries
+      expect(mockIncrementCounter).toBeCalledTimes(2 + retryConst);
+      // assert counterA increment error warning logs
+      expect(logger.warn).toHaveBeenNthCalledWith(
+        2,
+        `${mockError}, retrying attempt ${retryConst}`
+      );
+      expect(logger.warn).toHaveBeenNthCalledWith(3, mockError);
+      expect(logger.debug).toHaveBeenNthCalledWith(1, 'Store counters into savedObjects', {
+        kibana: {
+          usageCounters: {
+            results: [mockError, 'pass'],
+          },
        },
-      },
-    });
-  });
+      });
+    },
+    10 * 1000
+  );

  it('buffers counters within `bufferDurationMs` time', async () => {
    const usageCountersService = new UsageCountersService({
@ -264,7 +276,7 @@ describe('UsageCountersService', () => {
    jest.runOnlyPendingTimers();

    // wait for debounce to kick in on next scheduler call
-    await tick();
+    await tickWithDelay();
    expect(mockIncrementCounter).toBeCalledTimes(2);
    expect(mockIncrementCounter.mock.results.map(({ value }) => value)).toMatchInlineSnapshot(`
      Array [
--- a/src/plugins/usage_collection/server/usage_counters/usage_counters_service.ts
+++ b/src/plugins/usage_collection/server/usage_counters/usage_counters_service.ts
@ -33,6 +33,7 @@ import {
 } from './saved_objects';
 import { registerUsageCountersRollups } from './rollups';
 import { searchUsageCounters } from './search';
+import { USAGE_COUNTERS_BACKOFF_DELAY_IN_MS } from '../../common/constants';

 interface UsageCountersLogMeta extends LogMeta {
  kibana: { usageCounters: { results: unknown[] } };
@ -159,6 +160,9 @@ export class UsageCountersService {
    };
  };

+  private backoffDelay = (attempt: number) =>
+    Math.pow(2, attempt) * USAGE_COUNTERS_BACKOFF_DELAY_IN_MS; // exponential backoff: 500ms, 1000ms, 2000ms
+
  private storeDate$(
    counters: UsageCounters.v1.CounterMetric[],
    soRepository: Pick<SavedObjectsRepository, 'incrementCounter'>
@ -166,7 +170,13 @@ export class UsageCountersService {
    return Rx.forkJoin(
      counters.map((metric) =>
        Rx.defer(() => storeCounter({ metric, soRepository })).pipe(
-          Rx.retry(this.retryCount),
+          Rx.retry({
+            count: this.retryCount,
+            delay: (error, retryIndex) => {
+              this.logger.warn(`Error: ${error.message}, retrying attempt ${retryIndex}`); // extra warning logger
+              return Rx.timer(this.backoffDelay(retryIndex));
+            },
+          }),
          Rx.catchError((error) => {
            this.logger.warn(error);
            return Rx.of(error);