[8.x] [ML] Fix file upload with no ingest pipeline (#193744) (#194019)

# Backport This will backport the following commits from `main` to `8.x`: - [[ML] Fix file upload with no ingest pipeline (#193744)](#193744)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: James Gowdy <[email protected]>
elastic · Sep 25, 2024 · 603f51f · 603f51f
1 parent 44e1ca2
commit 603f51f
Show file tree

Hide file tree

Showing 9 changed files with 128 additions and 52 deletions.
diff --git a/.../data_visualizer/public/application/file_data_visualizer/components/import_view/import.ts b/.../data_visualizer/public/application/file_data_visualizer/components/import_view/import.ts
@@ -29,6 +29,7 @@ interface Config {
   mappingsString: string;
   pipelineString: string;
   pipelineId: string | null;
+  createPipeline: boolean;
 }
 
 export async function importData(props: Props, config: Config, setState: (state: unknown) => void) {
@@ -41,6 +42,7 @@ export async function importData(props: Props, config: Config, setState: (state:
     mappingsString,
     pipelineString,
     pipelineId,
+    createPipeline,
   } = config;
   const { format } = results;
 
@@ -86,7 +88,7 @@ export async function importData(props: Props, config: Config, setState: (state:
 
   let settings = {};
   let mappings = {};
-  let pipeline = {};
+  let pipeline;
 
   try {
     settings = JSON.parse(indexSettingsString);
@@ -109,7 +111,9 @@ export async function importData(props: Props, config: Config, setState: (state:
   }
 
   try {
-    pipeline = JSON.parse(pipelineString);
+    if (createPipeline) {
+      pipeline = JSON.parse(pipelineString) as IngestPipeline;
+    }
   } catch (error) {
     success = false;
     const parseError = i18n.translate('xpack.dataVisualizer.file.importView.parsePipelineError', {
@@ -143,12 +147,7 @@ export async function importData(props: Props, config: Config, setState: (state:
     return;
   }
 
-  const initializeImportResp = await importer.initializeImport(
-    index,
-    settings,
-    mappings,
-    pipeline as IngestPipeline
-  );
+  const initializeImportResp = await importer.initializeImport(index, settings, mappings, pipeline);
 
   const timeFieldName = importer.getTimeField();
   setState({ timeFieldName });
@@ -158,14 +157,20 @@ export async function importData(props: Props, config: Config, setState: (state:
     indexCreatedStatus: getSuccess(indexCreated),
   });
 
-  const pipelineCreated = initializeImportResp.pipelineId !== undefined;
-  if (indexCreated) {
-    setState({
-      ingestPipelineCreatedStatus: pipelineCreated ? IMPORT_STATUS.COMPLETE : IMPORT_STATUS.FAILED,
-      pipelineId: pipelineCreated ? initializeImportResp.pipelineId : '',
-    });
+  if (createPipeline) {
+    const pipelineCreated = initializeImportResp.pipelineId !== undefined;
+    if (indexCreated) {
+      setState({
+        ingestPipelineCreatedStatus: pipelineCreated
+          ? IMPORT_STATUS.COMPLETE
+          : IMPORT_STATUS.FAILED,
+        pipelineId: pipelineCreated ? initializeImportResp.pipelineId : '',
+      });
+    }
+    success = indexCreated && pipelineCreated;
+  } else {
+    success = indexCreated;
   }
-  success = indexCreated && pipelineCreated;
 
   if (success === false) {
     errors.push(initializeImportResp.error);

diff --git a/..._visualizer/public/application/file_data_visualizer/components/import_view/import_view.js b/..._visualizer/public/application/file_data_visualizer/components/import_view/import_view.js
@@ -109,6 +109,11 @@ export class ImportView extends Component {
       pipelineId,
     } = this.state;
 
+    const createPipeline = pipelineString !== '';
+    this.setState({
+      createPipeline,
+    });
+
     importData(
       { data, results, dataViewsContract, fileUpload },
       {
@@ -119,6 +124,7 @@ export class ImportView extends Component {
         mappingsString,
         pipelineString,
         pipelineId,
+        createPipeline,
       },
       (state) => this.setState(state)
     );

diff --git a/x-pack/plugins/file_upload/public/importer/geo/abstract_geo_file_importer.tsx b/x-pack/plugins/file_upload/public/importer/geo/abstract_geo_file_importer.tsx
@@ -196,9 +196,12 @@ export class AbstractGeoFileImporter extends Importer implements GeoFileImporter
             data: chunks[i],
             settings: {},
             mappings: {},
-            ingestPipeline: {
-              id: pipelineId,
-            },
+            ingestPipeline:
+              pipelineId !== undefined
+                ? {
+                    id: pipelineId,
+                  }
+                : undefined,
           });
 
           if (!this._isActive) {

diff --git a/x-pack/plugins/file_upload/public/importer/importer.ts b/x-pack/plugins/file_upload/public/importer/importer.ts
@@ -15,7 +15,13 @@ import { i18n } from '@kbn/i18n';
 import { isPopulatedObject } from '@kbn/ml-is-populated-object';
 import { getHttp } from '../kibana_services';
 import { MB } from '../../common/constants';
-import type { ImportDoc, ImportFailure, ImportResponse, IngestPipeline } from '../../common/types';
+import type {
+  ImportDoc,
+  ImportFailure,
+  ImportResponse,
+  IngestPipeline,
+  IngestPipelineWrapper,
+} from '../../common/types';
 import { CreateDocsResponse, IImporter, ImportResults } from './types';
 
 const CHUNK_SIZE = 5000;
@@ -79,26 +85,25 @@ export abstract class Importer implements IImporter {
     index: string,
     settings: IndicesIndexSettings,
     mappings: MappingTypeMapping,
-    pipeline: IngestPipeline
+    pipeline: IngestPipeline | undefined
   ) {
-    updatePipelineTimezone(pipeline);
-
-    if (pipelineContainsSpecialProcessors(pipeline)) {
-      // pipeline contains processors which we know are slow
-      // so reduce the chunk size significantly to avoid timeouts
-      this._chunkSize = REDUCED_CHUNK_SIZE;
+    let ingestPipeline: IngestPipelineWrapper | undefined;
+    if (pipeline !== undefined) {
+      updatePipelineTimezone(pipeline);
+
+      if (pipelineContainsSpecialProcessors(pipeline)) {
+        // pipeline contains processors which we know are slow
+        // so reduce the chunk size significantly to avoid timeouts
+        this._chunkSize = REDUCED_CHUNK_SIZE;
+      }
+      // if no pipeline has been supplied,
+      // send an empty object
+      ingestPipeline = {
+        id: `${index}-pipeline`,
+        pipeline,
+      };
     }
 
-    // if no pipeline has been supplied,
-    // send an empty object
-    const ingestPipeline =
-      pipeline !== undefined
-        ? {
-            id: `${index}-pipeline`,
-            pipeline,
-          }
-        : {};
-
     this._index = index;
     this._pipeline = pipeline;
 
@@ -139,9 +144,11 @@ export abstract class Importer implements IImporter {
 
     const chunks = createDocumentChunks(this._docArray, this._chunkSize);
 
-    const ingestPipeline = {
-      id: pipelineId,
-    };
+    const ingestPipeline: IngestPipelineWrapper | undefined = pipelineId
+      ? {
+          id: pipelineId,
+        }
+      : undefined;
 
     let success = true;
     const failures: ImportFailure[] = [];
@@ -345,10 +352,7 @@ export function callImportRoute({
   data: ImportDoc[];
   settings: IndicesIndexSettings;
   mappings: MappingTypeMapping;
-  ingestPipeline: {
-    id?: string;
-    pipeline?: IngestPipeline;
-  };
+  ingestPipeline: IngestPipelineWrapper | undefined;
 }) {
   const query = id !== undefined ? { id } : {};
   const body = JSON.stringify({

diff --git a/x-pack/plugins/file_upload/public/importer/types.ts b/x-pack/plugins/file_upload/public/importer/types.ts
@@ -43,7 +43,7 @@ export interface IImporter {
     index: string,
     settings: IndicesIndexSettings,
     mappings: MappingTypeMapping,
-    pipeline: IngestPipeline
+    pipeline: IngestPipeline | undefined
   ): Promise<ImportResponse>;
   import(
     id: string,

diff --git a/x-pack/plugins/file_upload/server/import_data.ts b/x-pack/plugins/file_upload/server/import_data.ts
@@ -21,15 +21,16 @@ export function importDataProvider({ asCurrentUser }: IScopedClusterClient) {
     index: string,
     settings: IndicesIndexSettings,
     mappings: MappingTypeMapping,
-    ingestPipeline: IngestPipelineWrapper,
+    ingestPipeline: IngestPipelineWrapper | undefined,
     data: InputData
   ): Promise<ImportResponse> {
     let createdIndex;
     let createdPipelineId;
     const docCount = data.length;
 
     try {
-      const { id: pipelineId, pipeline } = ingestPipeline;
+      const pipelineId = ingestPipeline?.id;
+      const pipeline = ingestPipeline?.pipeline;
 
       if (id === undefined) {
         // first chunk of data, create the index and id to return
@@ -48,7 +49,6 @@ export function importDataProvider({ asCurrentUser }: IScopedClusterClient) {
         createdPipelineId = pipelineId;
       } else {
         createdIndex = index;
-        createdPipelineId = pipelineId;
       }
 
       let failures: ImportFailure[] = [];
@@ -109,7 +109,7 @@ export function importDataProvider({ asCurrentUser }: IScopedClusterClient) {
     await asCurrentUser.indices.create({ index, body }, { maxRetries: 0 });
   }
 
-  async function indexData(index: string, pipelineId: string, data: InputData) {
+  async function indexData(index: string, pipelineId: string | undefined, data: InputData) {
     try {
       const body = [];
       for (let i = 0; i < data.length; i++) {

diff --git a/x-pack/plugins/file_upload/server/schemas.ts b/x-pack/plugins/file_upload/server/schemas.ts
@@ -37,10 +37,12 @@ export const importFileBodySchema = schema.object({
   /** Mappings */
   mappings: schema.any(),
   /** Ingest pipeline definition */
-  ingestPipeline: schema.object({
-    id: schema.maybe(schema.string()),
-    pipeline: schema.maybe(schema.any()),
-  }),
+  ingestPipeline: schema.maybe(
+    schema.object({
+      id: schema.maybe(schema.string()),
+      pipeline: schema.maybe(schema.any()),
+    })
+  ),
 });
 
 export const runtimeMappingsSchema = schema.object(

diff --git a/x-pack/test/functional/apps/ml/data_visualizer/file_data_visualizer.ts b/x-pack/test/functional/apps/ml/data_visualizer/file_data_visualizer.ts
@@ -180,6 +180,42 @@ export default function ({ getService }: FtrProviderContext) {
         ingestedDocCount: 3,
       },
     },
+    {
+      suiteSuffix: 'with a file which does not generate a ingest pipeline',
+      filePath: require.resolve('./files_to_import/flights_small.json'),
+      indexName: 'user-import_4',
+      createIndexPattern: false,
+      fieldTypeFilters: [ML_JOB_FIELD_TYPES.KEYWORD],
+      fieldNameFilters: ['timestamp'],
+      expected: {
+        results: {
+          title: 'flights_small.json',
+          highlightedText: false,
+        },
+        metricFields: [],
+        nonMetricFields: [
+          {
+            fieldName: 'Carrier',
+            type: ML_JOB_FIELD_TYPES.KEYWORD,
+            docCountFormatted: '20 (100%)',
+            exampleCount: 4,
+          },
+          {
+            fieldName: 'timestamp',
+            type: ML_JOB_FIELD_TYPES.KEYWORD,
+            docCountFormatted: '20 (100%)',
+            exampleCount: 11,
+          },
+        ],
+        visibleMetricFieldsCount: 0,
+        totalMetricFieldsCount: 0,
+        populatedFieldsCount: 3,
+        totalFieldsCount: 25,
+        fieldTypeFiltersResultCount: 16,
+        fieldNameFiltersResultCount: 1,
+        ingestedDocCount: 20,
+      },
+    },
   ];
 
   const testDataListNegative = [