[ML] Transforms: 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/constants.ts

@@ -43,7 +43,7 @@ export const API_BASE_PATH = '/api/transform/';
 // In the UI additional privileges are required:
 // - kibana_admin (builtin)
 // - dest index: monitor (applied to df-*)
-// - cluster: monitor
+// - cluster: monitor, read_pipeline
 //
 // Note that users with kibana_admin can see all Kibana data views and saved searches
 // in the source selection modal when creating a transform, but the wizard will trigger
@@ -69,7 +69,7 @@ export const APP_GET_TRANSFORM_CLUSTER_PRIVILEGES = [
   'cluster.cluster:monitor/transform/stats/get',
 ];
 
-// Equivalent of capabilities.canGetTransform
+// Equivalent of capabilities.canCreateTransform
 export const APP_CREATE_TRANSFORM_CLUSTER_PRIVILEGES = [
   'cluster.cluster:monitor/transform/get',
   'cluster.cluster:monitor/transform/stats/get',

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.test.ts

@@ -237,6 +237,7 @@ describe('Transform: Common', () => {
       transformSettingsMaxPageSearchSize: 100,
       transformSettingsDocsPerSecond: 400,
       destinationIndex: 'the-destination-index',
+      destinationIngestPipeline: 'the-destination-ingest-pipeline',
       touched: true,
       valid: true,
     };
@@ -249,7 +250,7 @@ describe('Transform: Common', () => {
 
     expect(request).toEqual({
       description: 'the-transform-description',
-      dest: { index: 'the-destination-index' },
+      dest: { index: 'the-destination-index', pipeline: 'the-destination-ingest-pipeline' },
       frequency: '1m',
       pivot: {
         aggregations: { 'the-agg-agg-name': { avg: { field: 'the-agg-field' } } },
@@ -315,6 +316,7 @@ describe('Transform: Common', () => {
       transformSettingsMaxPageSearchSize: 100,
       transformSettingsDocsPerSecond: 400,
       destinationIndex: 'the-destination-index',
+      destinationIngestPipeline: 'the-destination-ingest-pipeline',
       touched: true,
       valid: true,
     };
@@ -327,7 +329,7 @@ describe('Transform: Common', () => {
 
     expect(request).toEqual({
       description: 'the-transform-description',
-      dest: { index: 'the-destination-index' },
+      dest: { index: 'the-destination-index', pipeline: 'the-destination-ingest-pipeline' },
       frequency: '1m',
       pivot: {
         aggregations: { 'the-agg-agg-name': { avg: { field: 'the-agg-field' } } },

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

@@ -9,7 +9,7 @@ import { useMemo } from 'react';
 
 import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 
-import { HttpFetchError } from 'kibana/public';
+import type { HttpFetchError } from 'kibana/public';
 
 import { KBN_FIELD_TYPES } from '../../../../../../src/plugins/data/public';
 
@@ -43,9 +43,10 @@ import type {
   PostTransformsUpdateResponseSchema,
 } from '../../../common/api_schemas/update_transforms';
 import type { GetTransformsStatsResponseSchema } from '../../../common/api_schemas/transforms_stats';
-import { TransformId } from '../../../common/types/transform';
+import type { TransformId } from '../../../common/types/transform';
 import { API_BASE_PATH } from '../../../common/constants';
-import { EsIndex } from '../../../common/types/es_index';
+import type { EsIndex } from '../../../common/types/es_index';
+import type { 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

@@ -12,6 +12,7 @@ import { FormattedMessage } from '@kbn/i18n-react';
 
 import {
   EuiAccordion,
+  EuiComboBox,
   EuiLink,
   EuiSwitch,
   EuiFieldText,
@@ -28,6 +29,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 +84,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 +186,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 +209,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 +338,7 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
         transformSettingsMaxPageSearchSize,
         transformSettingsDocsPerSecond,
         destinationIndex,
+        destinationIngestPipeline,
         touched: true,
         valid,
         indexPatternTimeField,
@@ -331,6 +359,7 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
       transformFrequency,
       transformSettingsMaxPageSearchSize,
       destinationIndex,
+      destinationIngestPipeline,
       valid,
       indexPatternTimeField,
       /* eslint-enable react-hooks/exhaustive-deps */
@@ -443,6 +472,37 @@ export const StepDetailsForm: FC<StepDetailsFormProps> = React.memo(
             />
           </EuiFormRow>
 
+          {ingestPipelineNames.length > 0 && (
+            <EuiFormRow
+              label={i18n.translate(
+                'xpack.transform.stepDetailsForm.destinationIngestPipelineLabel',
+                {
+                  defaultMessage: 'Destination ingest pipeline',
+                }
+              )}
+            >
+              <EuiComboBox
+                data-test-subj="transformDestinationPipelineSelect"
+                aria-label={i18n.translate(
+                  'xpack.transform.stepDetailsForm.destinationIngestPipelineAriaLabel',
+                  {
+                    defaultMessage: 'Select an ingest pipeline',
+                  }
+                )}
+                placeholder={i18n.translate(
+                  'xpack.transform.stepDetailsForm.destinationIngestPipelineComboBoxPlaceholder',
+                  {
+                    defaultMessage: 'Select an ingest pipeline',
+                  }
+                )}
+                singleSelection={{ asPlainText: true }}
+                options={ingestPipelineNames.map((label: string) => ({ label }))}
+                selectedOptions={[{ label: destinationIngestPipeline }]}
+                onChange={(options) => setDestinationIngestPipeline(options[0]?.label ?? '')}
+              />
+            </EuiFormRow>
+          )}
+
           {stepDefineState.transformFunction === TRANSFORM_FUNCTION.LATEST ? (
             <>
               <EuiSpacer size={'m'} />