Support 2.17 BWC with latest backend integrations (#612)

Signed-off-by: Tyler Ohlsen <ohltyler@amazon.com>
opensearch-project · Feb 12, 2025 · 474a8a7 · 474a8a7
1 parent c77a2be
commit 474a8a7
Show file tree

Hide file tree

Showing 13 changed files with 241 additions and 60 deletions.
diff --git a/public/pages/workflow_detail/tools/query/query.tsx b/public/pages/workflow_detail/tools/query/query.tsx
@@ -36,6 +36,7 @@ import {
   containsEmptyValues,
   containsSameValues,
   getDataSourceId,
+  getEffectiveVersion,
   getPlaceholdersFromQuery,
   getSearchPipelineErrors,
   injectParameters,
@@ -64,6 +65,17 @@ const SEARCH_OPTIONS = [
 export function Query(props: QueryProps) {
   const dispatch = useAppDispatch();
   const dataSourceId = getDataSourceId();
+  const [dataSourceVersion, setDataSourceVersion] = useState<
+    string | undefined
+  >(undefined);
+  useEffect(() => {
+    async function getVersion() {
+      if (dataSourceId !== undefined) {
+        setDataSourceVersion(await getEffectiveVersion(dataSourceId));
+      }
+    }
+    getVersion();
+  }, [dataSourceId]);
 
   const { loading } = useSelector((state: AppState) => state.opensearch);
 
@@ -204,6 +216,7 @@ export function Query(props: QueryProps) {
                                 : '_none',
                             },
                             dataSourceId,
+                            dataSourceVersion,
                             verbose: includePipeline,
                           })
                         )

diff --git a/public/pages/workflow_detail/workflow_inputs/processors_list.tsx b/public/pages/workflow_detail/workflow_inputs/processors_list.tsx
@@ -29,8 +29,11 @@ import {
   WorkflowConfig,
   WorkflowFormValues,
 } from '../../../../common';
-import { formikToUiConfig, getDataSourceFromURL } from '../../../utils';
-import { getEffectiveVersion } from '../../../pages/workflows/new_workflow/new_workflow';
+import {
+  formikToUiConfig,
+  getDataSourceFromURL,
+  getEffectiveVersion,
+} from '../../../utils';
 import {
   CollapseProcessor,
   CopyIngestProcessor,

diff --git a/public/pages/workflow_detail/workflow_inputs/search_inputs/search_inputs.tsx b/public/pages/workflow_detail/workflow_inputs/search_inputs/search_inputs.tsx
@@ -15,8 +15,7 @@ import {
   WorkflowConfig,
 } from '../../../../../common';
 import { catIndices, useAppDispatch } from '../../../../store';
-import { getDataSourceId } from '../../../../utils';
-import { getEffectiveVersion } from '../../../workflows/new_workflow/new_workflow';
+import { getDataSourceId, getEffectiveVersion } from '../../../../utils';
 
 interface SearchInputsProps {
   uiConfig: WorkflowConfig;
@@ -42,8 +41,10 @@ export function SearchInputs(props: SearchInputsProps) {
   useEffect(() => {
     const checkVersion = async () => {
       try {
-        const version = await getEffectiveVersion(dataSourceId);
-        setShowTransformQuery(semver.gte(version, '2.19.0'));
+        if (dataSourceId !== undefined) {
+          const version = await getEffectiveVersion(dataSourceId);
+          setShowTransformQuery(semver.gte(version, '2.19.0'));
+        }
       } catch (error) {
         console.error('Error checking version:', error);
         setShowTransformQuery(true);

diff --git a/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx b/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx
@@ -6,6 +6,7 @@
 import React, { useEffect, useState } from 'react';
 import { getIn, useFormikContext } from 'formik';
 import { isEmpty, isEqual } from 'lodash';
+import semver from 'semver';
 import {
   EuiSmallButton,
   EuiSmallButtonEmpty,
@@ -23,6 +24,7 @@ import {
 import {
   CONFIG_STEP,
   CachedFormikState,
+  MINIMUM_FULL_SUPPORTED_VERSION,
   SimulateIngestPipelineResponseVerbose,
   TemplateNode,
   WORKFLOW_STEP_TYPE,
@@ -56,6 +58,8 @@ import {
   getDataSourceId,
   prepareDocsForSimulate,
   getIngestPipelineErrors,
+  getEffectiveVersion,
+  sleep,
 } from '../../../utils';
 import { BooleanField } from './input_fields';
 import '../workspace/workspace-styles.scss';
@@ -97,6 +101,21 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
   } = useFormikContext<WorkflowFormValues>();
   const dispatch = useAppDispatch();
   const dataSourceId = getDataSourceId();
+  const [dataSourceVersion, setDataSourceVersion] = useState<
+    string | undefined
+  >(undefined);
+  useEffect(() => {
+    async function getVersion() {
+      if (dataSourceId !== undefined) {
+        setDataSourceVersion(await getEffectiveVersion(dataSourceId));
+      }
+    }
+    getVersion();
+  }, [dataSourceId]);
+  const isPreV219 =
+    dataSourceVersion !== undefined
+      ? semver.lt(dataSourceVersion, MINIMUM_FULL_SUPPORTED_VERSION)
+      : false;
 
   // transient running states
   const [isUpdatingSearchPipeline, setIsUpdatingSearchPipeline] = useState<
@@ -390,10 +409,16 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
             reprovision: true,
           },
           dataSourceId,
+          dataSourceVersion,
         })
       )
         .unwrap()
         .then(async (result) => {
+          // if the datasource < 2.19, only async provisioning/reprovisioning is supported.
+          // so, we manually wait some time before trying to fetch the updated workflow
+          if (isPreV219) {
+            await sleep(1000);
+          }
           props.setUnsavedIngestProcessors(false);
           props.setUnsavedSearchProcessors(false);
           success = true;
@@ -426,6 +451,7 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
       )
         .unwrap()
         .then(async (result) => {
+          await sleep(100);
           await dispatch(
             updateWorkflow({
               apiBody: {
@@ -438,16 +464,24 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
           )
             .unwrap()
             .then(async (result) => {
+              await sleep(100);
               props.setUnsavedIngestProcessors(false);
               props.setUnsavedSearchProcessors(false);
               await dispatch(
                 provisionWorkflow({
                   workflowId: updatedWorkflow.id as string,
                   dataSourceId,
+                  dataSourceVersion,
                 })
               )
                 .unwrap()
                 .then(async (result) => {
+                  await sleep(100);
+                  // if the datasource < 2.19, only async provisioning/reprovisioning is supported.
+                  // so, we manually wait some time before trying to fetch the updated workflow
+                  if (isPreV219) {
+                    await sleep(1000);
+                  }
                   await dispatch(
                     getWorkflow({
                       workflowId: updatedWorkflow.id as string,

diff --git a/public/pages/workflows/new_workflow/new_workflow.tsx b/public/pages/workflows/new_workflow/new_workflow.tsx
@@ -27,11 +27,13 @@ import {
   searchConnectors,
 } from '../../../store';
 import { enrichPresetWorkflowWithUiMetadata } from './utils';
-import { getDataSourceId, isDataSourceReady } from '../../../utils';
+import {
+  getDataSourceId,
+  isDataSourceReady,
+  getEffectiveVersion,
+} from '../../../utils';
 import { getDataSourceEnabled } from '../../../services';
 import semver from 'semver';
-import { DataSourceAttributes } from '../../../../../../src/plugins/data_source/common/data_sources';
-import { getSavedObjectsClient } from '../../../../public/services';
 import {
   WORKFLOW_TYPE,
   MIN_SUPPORTED_VERSION,
@@ -40,27 +42,6 @@ import {
 
 interface NewWorkflowProps {}
 
-export const getEffectiveVersion = async (
-  dataSourceId: string | undefined
-): Promise<string> => {
-  try {
-    if (dataSourceId === undefined) {
-      throw new Error('Data source is required');
-    }
-
-    const dataSource = await getSavedObjectsClient().get<DataSourceAttributes>(
-      'data-source',
-      dataSourceId
-    );
-    const version =
-      dataSource?.attributes?.dataSourceVersion || MIN_SUPPORTED_VERSION;
-    return version;
-  } catch (error) {
-    console.error('Error getting version:', error);
-    return MIN_SUPPORTED_VERSION;
-  }
-};
-
 const filterPresetsByVersion = async (
   workflows: WorkflowTemplate[],
   dataSourceId: string | undefined

diff --git a/public/route_service.ts b/public/route_service.ts
@@ -60,11 +60,13 @@ export interface RouteService {
     workflowTemplate: WorkflowTemplate,
     updateFields: boolean,
     reprovision: boolean,
-    dataSourceId?: string
+    dataSourceId?: string,
+    dataSourceVersion?: string
   ) => Promise<any | HttpFetchError>;
   provisionWorkflow: (
     workflowId: string,
-    dataSourceId?: string
+    dataSourceId?: string,
+    dataSourceVersion?: string
   ) => Promise<any | HttpFetchError>;
   deprovisionWorkflow: ({
     workflowId,
@@ -96,12 +98,14 @@ export interface RouteService {
     index,
     body,
     dataSourceId,
+    dataSourceVersion,
     searchPipeline,
     verbose,
   }: {
     index: string;
     body: {};
     dataSourceId?: string;
+    dataSourceVersion?: string;
     searchPipeline?: string;
     verbose?: boolean;
   }) => Promise<any | HttpFetchError>;
@@ -205,7 +209,8 @@ export function configureRoutes(core: CoreStart): RouteService {
       workflowTemplate: WorkflowTemplate,
       updateFields: boolean,
       reprovision: boolean,
-      dataSourceId?: string
+      dataSourceId?: string,
+      dataSourceVersion?: string
     ) => {
       try {
         const url = dataSourceId
@@ -215,20 +220,32 @@ export function configureRoutes(core: CoreStart): RouteService {
           `${url}/${workflowId}/${updateFields}/${reprovision}`,
           {
             body: JSON.stringify(workflowTemplate),
+            query: {
+              data_source_version: dataSourceVersion,
+            },
           }
         );
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    provisionWorkflow: async (workflowId: string, dataSourceId?: string) => {
+    provisionWorkflow: async (
+      workflowId: string,
+      dataSourceId?: string,
+      dataSourceVersion?: string
+    ) => {
       try {
         const url = dataSourceId
           ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow/provision`
           : PROVISION_WORKFLOW_NODE_API_PATH;
         const response = await core.http.post<{ respString: string }>(
-          `${url}/${workflowId}`
+          `${url}/${workflowId}`,
+          {
+            query: {
+              data_source_version: dataSourceVersion,
+            },
+          }
         );
         return response;
       } catch (e: any) {
@@ -323,12 +340,14 @@ export function configureRoutes(core: CoreStart): RouteService {
       index,
       body,
       dataSourceId,
+      dataSourceVersion,
       searchPipeline,
       verbose,
     }: {
       index: string;
       body: {};
       dataSourceId?: string;
+      dataSourceVersion?: string;
       searchPipeline?: string;
       verbose?: boolean;
     }) => {
@@ -344,6 +363,7 @@ export function configureRoutes(core: CoreStart): RouteService {
           body: JSON.stringify(body),
           query: {
             verbose: verbose ?? false,
+            data_source_version: dataSourceVersion,
           },
         });
         return response;

diff --git a/public/store/reducers/opensearch_reducer.ts b/public/store/reducers/opensearch_reducer.ts
@@ -110,10 +110,12 @@ export const searchIndex = createAsyncThunk(
     {
       apiBody,
       dataSourceId,
+      dataSourceVersion,
       verbose,
     }: {
       apiBody: { index: string; body: {}; searchPipeline?: string };
       dataSourceId?: string;
+      dataSourceVersion?: string;
       verbose?: boolean;
     },
     { rejectWithValue }
@@ -123,6 +125,7 @@ export const searchIndex = createAsyncThunk(
       index,
       body,
       dataSourceId,
+      dataSourceVersion,
       searchPipeline,
       verbose,
     });

diff --git a/public/store/reducers/workflows_reducer.ts b/public/store/reducers/workflows_reducer.ts
@@ -116,6 +116,7 @@ export const updateWorkflow = createAsyncThunk(
     {
       apiBody,
       dataSourceId,
+      dataSourceVersion,
     }: {
       apiBody: {
         workflowId: string;
@@ -124,6 +125,7 @@ export const updateWorkflow = createAsyncThunk(
         reprovision?: boolean;
       };
       dataSourceId?: string;
+      dataSourceVersion?: string;
     },
     { rejectWithValue }
   ) => {
@@ -135,7 +137,8 @@ export const updateWorkflow = createAsyncThunk(
       workflowTemplate,
       updateFields || false,
       reprovision || false,
-      dataSourceId
+      dataSourceId,
+      dataSourceVersion
     );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
@@ -150,14 +153,23 @@ export const updateWorkflow = createAsyncThunk(
 export const provisionWorkflow = createAsyncThunk(
   PROVISION_WORKFLOW_ACTION,
   async (
-    { workflowId, dataSourceId }: { workflowId: string; dataSourceId?: string },
+    {
+      workflowId,
+      dataSourceId,
+      dataSourceVersion,
+    }: {
+      workflowId: string;
+      dataSourceId?: string;
+      dataSourceVersion?: string;
+    },
     { rejectWithValue }
   ) => {
     const response:
       | any
       | HttpFetchError = await getRouteService().provisionWorkflow(
       workflowId,
-      dataSourceId
+      dataSourceId,
+      dataSourceVersion
     );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(