[upgrade assistant] Stop rollup jobs before reindexing (elastic#212815)

This PR improve support for rollup indices. Rollup indices can be
handled like normal indices but jobs should be stopped before reindexing
begins or index is marked read only. Also handles case where the rollup
job is already stopped.

To review: Mark the following read only and make sure rollup jobs are
handled as appropriate: Rollup index with and without job running,
normal index.

Follow up to elastic#212592 and
elastic#214656

Closes: elastic#211850

---------

Co-authored-by: Ignacio Rivas <[email protected]>

Author: mattkime

x-pack/platform/plugins/private/upgrade_assistant/common/types.ts

@@ -115,6 +115,7 @@ export interface ReindexOperation {
   errorMessage: string | null;
   // This field is only used for the singleton IndexConsumerType documents.
   runningReindexCount: number | null;
+  rollupJob?: string;
 
   /**
    * The original index settings to set after reindex is completed.

x-pack/platform/plugins/private/upgrade_assistant/server/lib/reindexing/reindex_actions.test.ts

@@ -6,7 +6,7 @@
  */
 
 import { SavedObjectsErrorHelpers } from '@kbn/core/server';
-import { elasticsearchServiceMock } from '@kbn/core/server/mocks';
+import { elasticsearchServiceMock, loggingSystemMock } from '@kbn/core/server/mocks';
 import type { ScopedClusterClientMock } from '@kbn/core-elasticsearch-client-server-mocks';
 import moment from 'moment';
 
@@ -22,10 +22,15 @@ import { getMockVersionInfo } from '../__fixtures__/version';
 
 const { currentMajor, prevMajor } = getMockVersionInfo();
 
+jest.mock('../rollup_job', () => ({
+  getRollupJobByIndexName: jest.fn(),
+}));
+
 describe('ReindexActions', () => {
   let client: jest.Mocked<any>;
   let clusterClient: ScopedClusterClientMock;
   let actions: ReindexActions;
+  const log = loggingSystemMock.createLogger();
 
   const unimplemented = (name: string) => () =>
     Promise.reject(`Mock function ${name} was not implemented!`);
@@ -45,7 +50,7 @@ describe('ReindexActions', () => {
       ) as any,
     };
     clusterClient = elasticsearchServiceMock.createScopedClusterClient();
-    actions = reindexActionsFactory(client, clusterClient.asCurrentUser);
+    actions = reindexActionsFactory(client, clusterClient.asCurrentUser, log);
   });
 
   describe('createReindexOp', () => {

x-pack/platform/plugins/private/upgrade_assistant/server/lib/reindexing/reindex_actions.ts

@@ -11,6 +11,7 @@ import {
   SavedObjectsFindResponse,
   SavedObjectsClientContract,
   ElasticsearchClient,
+  Logger,
 } from '@kbn/core/server';
 import {
   REINDEX_OP_TYPE,
@@ -22,6 +23,7 @@ import {
 } from '../../../common/types';
 import { generateNewIndexName } from './index_settings';
 import { FlatSettings } from './types';
+import { getRollupJobByIndexName } from '../rollup_job';
 
 // TODO: base on elasticsearch.requestTimeout?
 export const LOCK_WINDOW = moment.duration(90, 'seconds');
@@ -84,7 +86,8 @@ export interface ReindexActions {
 
 export const reindexActionsFactory = (
   client: SavedObjectsClientContract,
-  esClient: ElasticsearchClient
+  esClient: ElasticsearchClient,
+  log: Logger
 ): ReindexActions => {
   // ----- Internal functions
   const isLocked = (reindexOp: ReindexSavedObject) => {
@@ -125,6 +128,9 @@ export const reindexActionsFactory = (
   // ----- Public interface
   return {
     async createReindexOp(indexName: string, opts?: ReindexOptions) {
+      // gets rollup job if it exists and needs stopping, otherwise returns undefined
+      const rollupJob = await getRollupJobByIndexName(esClient, log, indexName);
+
       return client.create<ReindexOperation>(REINDEX_OP_TYPE, {
         indexName,
         newIndexName: generateNewIndexName(indexName),
@@ -136,6 +142,7 @@ export const reindexActionsFactory = (
         errorMessage: null,
         runningReindexCount: null,
         reindexOptions: opts,
+        rollupJob,
       });
     },
 

x-pack/platform/plugins/private/upgrade_assistant/server/lib/reindexing/reindex_service.ts

@@ -169,7 +169,12 @@ export const reindexServiceFactory = (
    * @param reindexOp
    */
   const setReadonly = async (reindexOp: ReindexSavedObject) => {
-    const { indexName } = reindexOp.attributes;
+    const { indexName, rollupJob } = reindexOp.attributes;
+
+    if (rollupJob) {
+      await esClient.rollup.stopJob({ id: rollupJob, wait_for_completion: true });
+    }
+
     const putReadonly = await esClient.indices.putSettings({
       index: indexName,
       body: { blocks: { write: true } },
@@ -458,6 +463,11 @@ export const reindexServiceFactory = (
       await esClient.indices.close({ index: indexName });
     }
 
+    if (reindexOp.attributes.rollupJob) {
+      // start the rollup job. rollupJob is undefined if the rollup job is stopped
+      await esClient.rollup.startJob({ id: reindexOp.attributes.rollupJob });
+    }
+
     return actions.updateReindexOp(reindexOp, {
       lastCompletedStep: ReindexStep.aliasCreated,
     });

x-pack/platform/plugins/private/upgrade_assistant/server/lib/reindexing/worker.ts

@@ -99,7 +99,7 @@ export class ReindexWorker {
 
     this.reindexService = reindexServiceFactory(
       callAsInternalUser,
-      reindexActionsFactory(this.client, callAsInternalUser),
+      reindexActionsFactory(this.client, callAsInternalUser, this.log),
       log,
       this.licensing
     );
@@ -195,7 +195,7 @@ export class ReindexWorker {
     const fakeRequest: FakeRequest = { headers: credential };
     const scopedClusterClient = this.clusterClient.asScoped(fakeRequest);
     const callAsCurrentUser = scopedClusterClient.asCurrentUser;
-    const actions = reindexActionsFactory(this.client, callAsCurrentUser);
+    const actions = reindexActionsFactory(this.client, callAsCurrentUser, this.log);
     return reindexServiceFactory(callAsCurrentUser, actions, this.log, this.licensing);
   };
 

x-pack/platform/plugins/private/upgrade_assistant/server/lib/rollup_job.ts

@@ -0,0 +1,65 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { ElasticsearchClient, Logger } from '@kbn/core/server';
+import {
+  RollupGetRollupIndexCapsResponse,
+  RollupGetJobsResponse,
+} from '@elastic/elasticsearch/lib/api/types';
+
+export async function getRollupJobByIndexName(
+  esClient: ElasticsearchClient,
+  log: Logger,
+  index: string
+) {
+  let rollupCaps: RollupGetRollupIndexCapsResponse;
+
+  try {
+    rollupCaps = await esClient.rollup.getRollupIndexCaps({ index }, { ignore: [404] });
+    // may catch if not found in some circumstances, such as a closed index, etc
+    // would be nice to handle the error better but little info is provided
+  } catch (e) {
+    log.warn(`Get rollup index capabilities failed: ${e}`);
+    return;
+  }
+
+  const rollupIndices = Object.keys(rollupCaps);
+  let rollupJob: string | undefined;
+
+  // there should only be one job
+  if (rollupIndices.length === 1) {
+    rollupJob = rollupCaps[rollupIndices[0]].rollup_jobs[0].job_id;
+    let jobs: RollupGetJobsResponse;
+
+    try {
+      jobs = await esClient.rollup.getJobs({ id: rollupJob }, { ignore: [404] });
+      // may catch if not found in some circumstances, such as a closed index, etc
+      // would be nice to handle the error better but little info is provided
+    } catch (e) {
+      log.warn(`Get rollup job failed: ${e}`);
+      return;
+    }
+
+    // there can only be one job. If its stopped then we don't need rollup handling
+    if (
+      // zero jobs shouldn't happen but we can handle it gracefully
+      jobs.jobs.length === 0 ||
+      // rollup job is stopped so we can treat it like a regular index
+      (jobs.jobs.length === 1 && jobs.jobs[0].status.job_state === 'stopped')
+    ) {
+      rollupJob = undefined;
+      // this shouldn't be possible but just in case
+    } else if (jobs.jobs.length > 1) {
+      throw new Error(`Multiple jobs returned for a single rollup job id: + ${rollupJob}`);
+    }
+    // this shouldn't be possible but just in case
+  } else if (rollupIndices.length > 1) {
+    throw new Error(`Multiple indices returned for a single index name: + ${index}`);
+  }
+
+  return rollupJob;
+}

x-pack/platform/plugins/private/upgrade_assistant/server/lib/update_index/index.test.ts

@@ -25,6 +25,7 @@ describe('updateIndex', () => {
   const mockGetReindexWarnings = getReindexWarnings as jest.Mock;
   const mockLogger = loggingSystemMock.create().get();
   const mockClient = elasticsearchServiceMock.createScopedClusterClient().asCurrentUser;
+  mockClient.rollup.getRollupIndexCaps.mockResponse({});
 
   beforeEach(() => {
     jest.clearAllMocks();

x-pack/platform/plugins/private/upgrade_assistant/server/lib/update_index/index.ts

@@ -5,16 +5,16 @@
  * 2.0.
  */
 
-import type { ElasticsearchClient } from '@kbn/core/server';
-import type { Logger } from '@kbn/core/server';
+import type { ElasticsearchClient, Logger } from '@kbn/core/server';
 import type { UpdateIndexOperation } from '../../../common/update_index';
+import { getRollupJobByIndexName } from '../rollup_job';
 import { getReindexWarnings } from '../reindexing/index_settings';
 
 export interface UpdateIndexParams {
   esClient: ElasticsearchClient;
   index: string;
   operations: UpdateIndexOperation[];
-  log?: Logger;
+  log: Logger;
 }
 
 /**
@@ -30,6 +30,12 @@ export async function updateIndex({ esClient, index, operations, log }: UpdateIn
 
     switch (operation) {
       case 'blockWrite': {
+        // stop related rollup job if it exists
+        const rollupJob = await getRollupJobByIndexName(esClient, log, index);
+        if (rollupJob) {
+          await esClient.rollup.stopJob({ id: rollupJob, wait_for_completion: true });
+        }
+
         res = await esClient.indices.addBlock({ index, block: 'write' });
 
         await removeDeprecatedSettings(esClient, index, log);

x-pack/platform/plugins/private/upgrade_assistant/server/routes/es_deprecations.ts

@@ -41,7 +41,7 @@ export function registerESDeprecationRoutes({
           dataSourceExclusions,
         });
         const asCurrentUser = client.asCurrentUser;
-        const reindexActions = reindexActionsFactory(savedObjectsClient, asCurrentUser);
+        const reindexActions = reindexActionsFactory(savedObjectsClient, asCurrentUser, log);
         const reindexService = reindexServiceFactory(asCurrentUser, reindexActions, log, licensing);
         const indexNames = [...status.migrationsDeprecations, ...status.enrichedHealthIndicators]
           .filter(({ index }) => typeof index !== 'undefined')

x-pack/platform/plugins/private/upgrade_assistant/server/routes/reindex_indices/batch_reindex_indices.ts

@@ -57,7 +57,8 @@ export function registerBatchReindexIndicesRoutes(
       const callAsCurrentUser = esClient.asCurrentUser;
       const reindexActions = reindexActionsFactory(
         getClient({ includedHiddenTypes: [REINDEX_OP_TYPE] }),
-        callAsCurrentUser
+        callAsCurrentUser,
+        log
       );
       try {
         const inProgressOps = await reindexActions.findAllByStatus(ReindexStatus.inProgress);