revert back to using _scroll, unskip the CSV export tests

Author: jloleysens

x-pack/plugins/reporting/server/export_types/csv_searchsource/generate_csv/generate_csv.ts

@@ -14,9 +14,10 @@ import { Datatable } from 'src/plugins/expressions/server';
 import { ReportingConfig } from '../../..';
 import {
   cellHasFormulas,
+  ES_SEARCH_STRATEGY,
+  IndexPattern,
   ISearchSource,
   ISearchStartSearchSource,
-  ES_SEARCH_STRATEGY,
   SearchFieldValue,
   SearchSourceFields,
   tabifyDocs,
@@ -79,38 +80,44 @@ export class CsvGenerator {
     private stream: Writable
   ) {}
 
-  private async scroll(
-    pitId: string,
-    settings: CsvExportSettings,
+  private async scan(
+    index: IndexPattern,
     searchSource: ISearchSource,
-    searchAfter?: estypes.SearchSortResults
+    settings: CsvExportSettings
   ) {
-    this.logger.debug(`executing scroll request`);
     const { scroll: scrollSettings, includeFrozen } = settings;
     const searchBody = searchSource.getSearchRequestBody();
-    const searchParams: estypes.SearchRequest = {
-      body: {
-        ...searchBody,
-        pit: {
-          id: pitId,
-          keep_alive: scrollSettings.duration,
-        },
-        search_after: searchAfter,
+    this.logger.debug(`executing search request`);
+    const searchParams = {
+      params: {
+        body: searchBody,
+        index: index.title,
+        scroll: scrollSettings.duration,
+        size: scrollSettings.size,
+        ignore_throttled: !includeFrozen,
       },
-      size: scrollSettings.size,
-      ignore_throttled: !includeFrozen,
-      ignore_unavailable: undefined,
     };
 
     const results = (
-      await this.clients.data
-        .search({ params: searchParams }, { strategy: ES_SEARCH_STRATEGY })
-        .toPromise()
+      await this.clients.data.search(searchParams, { strategy: ES_SEARCH_STRATEGY }).toPromise()
     ).rawResponse as estypes.SearchResponse<unknown>;
 
     return results;
   }
 
+  private async scroll(scrollId: string, scrollSettings: CsvExportSettings['scroll']) {
+    this.logger.debug(`executing scroll request`);
+    const results = (
+      await this.clients.es.asCurrentUser.scroll({
+        body: {
+          scroll: scrollSettings.duration,
+          scroll_id: scrollId,
+        },
+      })
+    ).body;
+    return results;
+  }
+
   /*
    * Load field formats for each field in the list
    */
@@ -287,15 +294,14 @@ export class CsvGenerator {
       throw new Error(`The search must have a reference to an index pattern!`);
     }
 
-    const { maxSizeBytes, bom, escapeFormulaValues } = settings;
+    const { maxSizeBytes, bom, escapeFormulaValues, scroll: scrollSettings } = settings;
 
     const builder = new MaxSizeStringBuilder(this.stream, byteSizeValueToNumber(maxSizeBytes), bom);
     const warnings: string[] = [];
     let first = true;
     let currentRecord = -1;
     let totalRecords = 0;
-    let pitId: string | undefined;
-    let searchAfter: undefined | estypes.SearchSortResults;
+    let scrollId: string | undefined;
 
     // apply timezone from the job to all date field formatters
     try {
@@ -320,34 +326,25 @@ export class CsvGenerator {
         if (this.cancellationToken.isCancelled()) {
           break;
         }
-        if (pitId == null) {
-          pitId = (
-            await this.clients.es.asCurrentUser.openPointInTime({
-              index: index.title,
-              keep_alive: settings.scroll.duration,
-            })
-          ).body.id;
-        }
-
-        const results = await this.scroll(pitId, settings, searchSource, searchAfter);
-        if (results.hits?.total != null && !totalRecords) {
-          totalRecords =
-            typeof results.hits.total === 'number' ? results.hits.total : results.hits.total.value;
-          this.logger.debug(`Total search results: ${totalRecords}`);
+        let results: estypes.SearchResponse<unknown> | undefined;
+        if (scrollId == null) {
+          // open a scroll cursor in Elasticsearch
+          results = await this.scan(index, searchSource, settings);
+          scrollId = results?._scroll_id;
+          if (results.hits?.total != null) {
+            totalRecords = results.hits.total as number;
+            this.logger.debug(`Total search results: ${totalRecords}`);
+          }
+        } else {
+          // use the scroll cursor in Elasticsearch
+          results = await this.scroll(scrollId, scrollSettings);
         }
 
-        pitId = results.pit_id ? results.pit_id : pitId;
-
         if (!results) {
           this.logger.warning(`Search results are undefined!`);
           break;
         }
 
-        const totalHits = results.hits.hits.length;
-        if (totalHits) {
-          searchAfter = results.hits.hits[totalHits - 1].sort;
-        }
-
         let table: Datatable | undefined;
         try {
           table = tabifyDocs(results, index, { shallow: true, meta: true });
@@ -393,16 +390,16 @@ export class CsvGenerator {
         throw JSON.stringify(err.errBody.error);
       }
     } finally {
-      // close PIT
-      if (pitId) {
-        this.logger.debug(`executing close point-in-time request`);
+      // clear scrollID
+      if (scrollId) {
+        this.logger.debug(`executing clearScroll request`);
         try {
-          await this.clients.es.asCurrentUser.closePointInTime({ body: { id: pitId } });
+          await this.clients.es.asCurrentUser.clearScroll({ body: { scroll_id: [scrollId] } });
         } catch (err) {
           this.logger.error(err);
         }
       } else {
-        this.logger.warn(`No point-in-time to close!`);
+        this.logger.warn(`No scrollId to clear!`);
       }
     }