[8.x] [ResponseOps][TaskManager] Discovery service running after shutdown (next one!) (#193478) (#194225)

# Backport This will backport the following commits from `main` to `8.x`: - [[ResponseOps][TaskManager] Discovery service running after shutdown (next one!) (#193478)](https://github.com/elastic/kibana/pull/193478)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: Alexi Doak <109488926+doakalexi@users.noreply.github.com>
2025-04-24 17:59:23 -04:00 · 2024-09-27 07:39:11 +10:00 · 2024-09-27 07:39:11 +10:00 · e1d68c81b0
commit e1d68c81b0
parent 81664a0bde
3 changed files with 96 additions and 22 deletions
--- a/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.test.ts
+++ b/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.test.ts
@ -5,7 +5,7 @@
 * 2.0.
 */
 import { savedObjectsRepositoryMock, loggingSystemMock } from '@kbn/core/server/mocks';
-import { KibanaDiscoveryService } from './kibana_discovery_service';
+import { DEFAULT_TIMEOUT, KibanaDiscoveryService } from './kibana_discovery_service';
 import { BACKGROUND_TASK_NODE_SO_NAME } from '../saved_objects';
 import { SavedObjectsBulkDeleteResponse, SavedObjectsUpdateResponse } from '@kbn/core/server';

@ -26,6 +26,7 @@ describe('KibanaDiscoveryService', () => {
  beforeEach(() => {
    jest.useFakeTimers();
    jest.spyOn(global, 'setTimeout');
+    jest.spyOn(global, 'clearTimeout');
    jest.setSystemTime(new Date(now));
  });

@ -110,6 +111,23 @@ describe('KibanaDiscoveryService', () => {
      expect(savedObjectsRepository.update).toHaveBeenCalledTimes(2);
    });

+    it('timeout should not be less than two seconds', async () => {
+      savedObjectsRepository.find.mockResolvedValueOnce(createFindResponse([]));
+      const kibanaDiscoveryService = new KibanaDiscoveryService({
+        savedObjectsRepository,
+        logger,
+        currentNode,
+        config: {
+          active_nodes_lookback: DEFAULT_ACTIVE_NODES_LOOK_BACK_DURATION,
+          interval: 500,
+        },
+      });
+      await kibanaDiscoveryService.start();
+
+      expect(setTimeout).toHaveBeenCalledTimes(1);
+      expect(setTimeout).toHaveBeenNthCalledWith(1, expect.any(Function), DEFAULT_TIMEOUT);
+    });
+
    it('reschedules when upsert fails on start', async () => {
      savedObjectsRepository.update.mockRejectedValueOnce(new Error('foo'));

@ -180,6 +198,47 @@ describe('KibanaDiscoveryService', () => {
        "Kibana Discovery Service couldn't update this node's last_seen timestamp. id: current-node-id, last_seen: 2024-08-10T10:00:10.000Z, error:foo"
      );
    });
+
+    it('does not schedule when Kibana is shutting down', async () => {
+      savedObjectsRepository.update.mockResolvedValueOnce(
+        {} as SavedObjectsUpdateResponse<unknown>
+      );
+
+      const kibanaDiscoveryService = new KibanaDiscoveryService({
+        savedObjectsRepository,
+        logger,
+        currentNode,
+        config: {
+          active_nodes_lookback: DEFAULT_ACTIVE_NODES_LOOK_BACK_DURATION,
+          interval: DEFAULT_DISCOVERY_INTERVAL_MS,
+        },
+      });
+      await kibanaDiscoveryService.start();
+
+      expect(savedObjectsRepository.update).toHaveBeenCalledTimes(1);
+      expect(logger.error).not.toHaveBeenCalled();
+      expect(logger.info).toHaveBeenCalledWith('Kibana Discovery Service has been started');
+      expect(kibanaDiscoveryService.isStarted()).toBe(true);
+      expect(setTimeout).toHaveBeenCalledTimes(1);
+      expect(setTimeout).toHaveBeenNthCalledWith(
+        1,
+        expect.any(Function),
+        DEFAULT_DISCOVERY_INTERVAL_MS
+      );
+
+      kibanaDiscoveryService.stop();
+
+      await jest.advanceTimersByTimeAsync(15000);
+
+      expect(savedObjectsRepository.update).toHaveBeenCalledTimes(1);
+      expect(setTimeout).toHaveBeenCalledTimes(1);
+      expect(setTimeout).toHaveBeenNthCalledWith(
+        1,
+        expect.any(Function),
+        DEFAULT_DISCOVERY_INTERVAL_MS
+      );
+      expect(clearTimeout).toHaveBeenCalledTimes(1);
+    });
  });

  describe('getActiveKibanaNodes', () => {
--- a/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.ts
+++ b/x-pack/plugins/task_manager/server/kibana_discovery_service/kibana_discovery_service.ts
@ -23,6 +23,8 @@ interface DiscoveryServiceUpsertParams {
  lastSeen: string;
 }

+export const DEFAULT_TIMEOUT = 2000;
+
 export class KibanaDiscoveryService {
  private readonly activeNodesLookBack: string;
  private readonly discoveryInterval: number;
@ -30,6 +32,8 @@ export class KibanaDiscoveryService {
  private started = false;
  private savedObjectsRepository: ISavedObjectsRepository;
  private logger: Logger;
+  private stopped = false;
+  private timer: NodeJS.Timeout | undefined;

  constructor({ config, currentNode, savedObjectsRepository, logger }: DiscoveryServiceParams) {
    this.activeNodesLookBack = config.active_nodes_lookback;
@ -52,29 +56,32 @@ export class KibanaDiscoveryService {
  }

  private async scheduleUpsertCurrentNode() {
-    const lastSeenDate = new Date();
-    const lastSeen = lastSeenDate.toISOString();
-    try {
-      await this.upsertCurrentNode({ id: this.currentNode, lastSeen });
-      if (!this.started) {
-        this.logger.info('Kibana Discovery Service has been started');
-        this.started = true;
-      }
-    } catch (e) {
-      if (!this.started) {
-        this.logger.error(
-          `Kibana Discovery Service couldn't be started and will be retried in ${this.discoveryInterval}ms, error:${e.message}`
-        );
-      } else {
-        this.logger.error(
-          `Kibana Discovery Service couldn't update this node's last_seen timestamp. id: ${this.currentNode}, last_seen: ${lastSeen}, error:${e.message}`
+    if (!this.stopped) {
+      const lastSeenDate = new Date();
+      const lastSeen = lastSeenDate.toISOString();
+      try {
+        await this.upsertCurrentNode({ id: this.currentNode, lastSeen });
+        if (!this.started) {
+          this.logger.info('Kibana Discovery Service has been started');
+          this.started = true;
+        }
+      } catch (e) {
+        if (!this.started) {
+          this.logger.error(
+            `Kibana Discovery Service couldn't be started and will be retried in ${this.discoveryInterval}ms, error:${e.message}`
+          );
+        } else {
+          this.logger.error(
+            `Kibana Discovery Service couldn't update this node's last_seen timestamp. id: ${this.currentNode}, last_seen: ${lastSeen}, error:${e.message}`
+          );
+        }
+      } finally {
+        this.timer = setTimeout(
+          async () => await this.scheduleUpsertCurrentNode(),
+          // The timeout should not be less than the default timeout of two seconds
+          Math.max(this.discoveryInterval - (Date.now() - lastSeenDate.getTime()), DEFAULT_TIMEOUT)
        );
      }
-    } finally {
-      setTimeout(
-        async () => await this.scheduleUpsertCurrentNode(),
-        this.discoveryInterval - (Date.now() - lastSeenDate.getTime())
-      );
    }
  }

@ -106,4 +113,11 @@ export class KibanaDiscoveryService {
    await this.savedObjectsRepository.delete(BACKGROUND_TASK_NODE_SO_NAME, this.currentNode);
    this.logger.info('Removed this node from the Kibana Discovery Service');
  }
+
+  public stop() {
+    this.stopped = true;
+    if (this.timer) {
+      clearTimeout(this.timer);
+    }
+  }
 }
--- a/x-pack/plugins/task_manager/server/plugin.ts
+++ b/x-pack/plugins/task_manager/server/plugin.ts
@ -404,6 +404,7 @@ export class TaskManagerPlugin

  public async stop() {
    if (this.kibanaDiscoveryService?.isStarted()) {
+      this.kibanaDiscoveryService.stop();
      try {
        await this.kibanaDiscoveryService.deleteCurrentNode();
      } catch (e) {