[ML] Support to set destination ingest pipeline.

x-pack/plugins/transform/common/api_schemas/type_guards.ts

@@ -8,6 +8,7 @@
 import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 
 import type { EsIndex } from '../types/es_index';
+import type { EsIngestPipeline } from '../types/es_ingest_pipeline';
 import { isPopulatedObject } from '../shared_imports';
 
 // To be able to use the type guards on the client side, we need to make sure we don't import
@@ -60,6 +61,10 @@ export const isEsIndices = (arg: unknown): arg is EsIndex[] => {
   return Array.isArray(arg);
 };
 
+export const isEsIngestPipelines = (arg: unknown): arg is EsIngestPipeline[] => {
+  return Array.isArray(arg) && arg.every((d) => isPopulatedObject(d, ['name']));
+};
+
 export const isEsSearchResponse = (arg: unknown): arg is estypes.SearchResponse => {
   return isPopulatedObject(arg, ['hits']);
 };

x-pack/plugins/transform/common/types/es_ingest_pipeline.ts

@@ -0,0 +1,13 @@
+/*
+ * 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.
+ */
+
+// This interface doesn't cover a full ingest pipeline spec,
+// just what's necessary to make it work in the transform creation wizard.
+// The full interface can be found in x-pack/plugins/ingest_pipelines/common/types.ts
+export interface EsIngestPipeline {
+  name: string;
+}

x-pack/plugins/transform/public/app/common/request.ts

@@ -219,6 +219,10 @@ export const getCreateTransformRequestBody = (
     : {}),
   dest: {
     index: transformDetailsState.destinationIndex,
+    // conditionally add optional ingest pipeline
+    ...(transformDetailsState.destinationIngestPipeline !== ''
+      ? { pipeline: transformDetailsState.destinationIngestPipeline }
+      : {}),
   },
   // conditionally add continuous mode config
   ...(transformDetailsState.isContinuousModeEnabled

x-pack/plugins/transform/public/app/hooks/use_api.ts

@@ -46,6 +46,7 @@ import type { GetTransformsStatsResponseSchema } from '../../../common/api_schem
 import { TransformId } from '../../../common/types/transform';
 import { API_BASE_PATH } from '../../../common/constants';
 import { EsIndex } from '../../../common/types/es_index';
+import { EsIngestPipeline } from '../../../common/types/es_ingest_pipeline';
 
 import { useAppDependencies } from '../app_dependencies';
 
@@ -202,6 +203,13 @@ export const useApi = () => {
           return e;
         }
       },
+      async getEsIngestPipelines(): Promise<EsIngestPipeline[] | HttpFetchError> {
+        try {
+          return await http.get('/api/ingest_pipelines');
+        } catch (e) {
+          return e;
+        }
+      },
       async getHistogramsForFields(
         indexPatternTitle: string,
         fields: FieldHistogramRequestConfig[],

x-pack/plugins/transform/public/app/sections/create_transform/components/step_details/common.ts

@@ -8,13 +8,15 @@
 import type { TransformConfigUnion, TransformId } from '../../../../../../common/types/transform';
 
 export type EsIndexName = string;
+export type EsIngestPipelineName = string;
 export type IndexPatternTitle = string;
 
 export interface StepDetailsExposedState {
   continuousModeDateField: string;
   continuousModeDelay: string;
   createIndexPattern: boolean;
   destinationIndex: EsIndexName;
+  destinationIngestPipeline: EsIngestPipelineName;
   isContinuousModeEnabled: boolean;
   isRetentionPolicyEnabled: boolean;
   retentionPolicyDateField: string;
@@ -48,6 +50,7 @@ export function getDefaultStepDetailsState(): StepDetailsExposedState {
     transformFrequency: defaultTransformFrequency,
     transformSettingsMaxPageSearchSize: defaultTransformSettingsMaxPageSearchSize,
     destinationIndex: '',
+    destinationIngestPipeline: '',
     touched: false,
     valid: false,
     indexPatternTimeField: undefined,
@@ -73,6 +76,11 @@ export function applyTransformConfigToDetailsState(
       state.transformDescription = transformConfig.description;
     }
 
+    // Ingest Pipeline
+    if (transformConfig.dest.pipeline !== undefined) {
+      state.destinationIngestPipeline = transformConfig.dest.pipeline;
+    }
+
     // Frequency
     if (transformConfig.frequency !== undefined) {
       state.transformFrequency = transformConfig.frequency;

x-pack/plugins/transform/public/app/sections/create_transform/components/step_details/step_details_form.tsx

@@ -28,6 +28,7 @@ import { toMountPoint } from '../../../../../../../../../src/plugins/kibana_reac
 
 import {
   isEsIndices,
+  isEsIngestPipelines,
   isPostTransformsPreviewResponseSchema,
 } from '../../../../../../common/api_schemas/type_guards';
 import { TransformId } from '../../../../../../common/types/transform';
@@ -82,8 +83,12 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
     const [destinationIndex, setDestinationIndex] = useState<EsIndexName>(
       defaults.destinationIndex
     );
+    const [destinationIngestPipeline, setDestinationIngestPipeline] = useState<string>(
+      defaults.destinationIngestPipeline
+    );
     const [transformIds, setTransformIds] = useState<TransformId[]>([]);
     const [indexNames, setIndexNames] = useState<EsIndexName[]>([]);
+    const [ingestPipelineNames, setIngestPipelineNames] = useState<string[]>([]);
 
     const canCreateDataView = useMemo(
       () =>
@@ -180,7 +185,10 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
           setTransformIds(resp.transforms.map((transform) => transform.id));
         }
 
-        const indices = await api.getEsIndices();
+        const [indices, ingestPipelines] = await Promise.all([
+          api.getEsIndices(),
+          api.getEsIngestPipelines(),
+        ]);
 
         if (isEsIndices(indices)) {
           setIndexNames(indices.map((index) => index.name));
@@ -200,6 +208,24 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
           });
         }
 
+        if (isEsIngestPipelines(ingestPipelines)) {
+          setIngestPipelineNames(ingestPipelines.map(({ name }) => name));
+        } else {
+          toastNotifications.addDanger({
+            title: i18n.translate('xpack.transform.stepDetailsForm.errorGettingIngestPipelines', {
+              defaultMessage: 'An error occurred getting the existing ingest pipeline names:',
+            }),
+            text: toMountPoint(
+              <ToastNotificationText
+                overlays={overlays}
+                theme={theme}
+                text={getErrorMessage(ingestPipelines)}
+              />,
+              { theme$: theme.theme$ }
+            ),
+          });
+        }
+
         try {
           setIndexPatternTitles(await deps.data.indexPatterns.getTitles());
         } catch (e) {
@@ -311,6 +337,7 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
         transformSettingsMaxPageSearchSize,
         transformSettingsDocsPerSecond,
         destinationIndex,
+        destinationIngestPipeline,
         touched: true,
         valid,
         indexPatternTimeField,
@@ -443,6 +470,25 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
             />
           </EuiFormRow>
 
+          {ingestPipelineNames.length > 0 && (
+            <EuiFormRow
+              label={i18n.translate(
+                'xpack.transform.stepDetailsForm.destinationIngestPipelineLabel',
+                {
+                  defaultMessage: 'Destination ingest pipeline',
+                }
+              )}
+            >
+              <EuiSelect
+                options={ingestPipelineNames.map((text: string) => ({ text, value: text }))}
+                value={destinationIngestPipeline}
+                onChange={(e) => setDestinationIngestPipeline(e.target.value)}
+                hasNoInitialSelection={defaults.destinationIngestPipeline === ''}
+                data-test-subj="transformDestinationPipelineSelect"
+              />
+            </EuiFormRow>
+          )}
+
           {stepDefineState.transformFunction === TRANSFORM_FUNCTION.LATEST ? (
             <>
               <EuiSpacer size={'m'} />