From e424be155daa6cdffb4234065b3f216922e79605 Mon Sep 17 00:00:00 2001
From: "opensearch-trigger-bot[bot]"
 <98922864+opensearch-trigger-bot[bot]@users.noreply.github.com>
Date: Mon, 12 Aug 2024 12:36:04 -0700
Subject: [PATCH] Support MDS on workflows and workflow_detail pages  (#274)
 (#276)

* Added MDS support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Added MDS support, Fixed failing tests

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

* Fixing MDS Support

Signed-off-by: saimedhi <saimedhi@amazon.com>

---------

Signed-off-by: saimedhi <saimedhi@amazon.com>
Signed-off-by: Sai Medhini Reddy Maryada <117196660+saimedhi@users.noreply.github.com>
(cherry picked from commit c8bc6822b1afb08ce9fbffcb8b1860c1beb46f22)

Co-authored-by: Sai Medhini Reddy Maryada <117196660+saimedhi@users.noreply.github.com>
---
 common/interfaces.ts                          |   3 +
 common/utils.ts                               |  15 +
 opensearch_dashboards.json                    |   2 +-
 public/app.tsx                                |  41 ++-
 .../pages/workflow_detail/workflow_detail.tsx |  63 +++-
 .../input_transform_modal.tsx                 |  22 +-
 .../output_transform_modal.tsx                |  22 +-
 .../configure_search_request.tsx              |  13 +-
 .../workflow_inputs/workflow_inputs.tsx       |  93 ++++--
 .../import_workflow/import_workflow_modal.tsx |  16 +-
 .../pages/workflows/new_workflow/use_case.tsx |  13 +-
 .../pages/workflows/workflow_list/columns.tsx |  78 +++--
 .../workflows/workflow_list/workflow_list.tsx |   9 +-
 public/pages/workflows/workflows.test.tsx     |   4 +-
 public/pages/workflows/workflows.tsx          | 127 +++++++-
 public/plugin.ts                              |  26 +-
 public/render_app.tsx                         |  22 +-
 public/route_service.ts                       | 254 +++++++++++-----
 public/services.ts                            |  27 +-
 public/store/reducers/models_reducer.ts       |   8 +-
 public/store/reducers/opensearch_reducer.ts   |  74 +++--
 public/store/reducers/workflows_reducer.ts    | 101 +++++--
 public/types.ts                               |   8 +-
 public/utils/constants.ts                     |   7 +-
 public/utils/utils.ts                         |  38 ++-
 server/plugin.ts                              |  29 +-
 .../routes/flow_framework_routes_service.ts   | 286 +++++++++++++++---
 server/routes/ml_routes_service.ts            |  34 ++-
 server/routes/opensearch_routes_service.ts    | 194 ++++++++++--
 server/utils/helpers.ts                       |  26 ++
 test/mocks/mock_core_services.ts              |   5 +
 31 files changed, 1327 insertions(+), 333 deletions(-)
 create mode 100644 server/utils/helpers.ts

diff --git a/common/interfaces.ts b/common/interfaces.ts
index ba1f1b99..9f705c8f 100644
--- a/common/interfaces.ts
+++ b/common/interfaces.ts
@@ -16,6 +16,9 @@ export type Index = {
 /**
  ********** WORKFLOW TYPES/INTERFACES **********
  */
+export type MDSQueryParams = {
+  dataSourceId?: string;
+};
 
 export type ConfigFieldType =
   | 'string'
diff --git a/common/utils.ts b/common/utils.ts
index 6a5b9c0f..b6892191 100644
--- a/common/utils.ts
+++ b/common/utils.ts
@@ -5,7 +5,22 @@
 
 import moment from 'moment';
 import { DATE_FORMAT_PATTERN } from './';
+import { isEmpty } from 'lodash';
 
 export function toFormattedDate(timestampMillis: number): String {
   return moment(new Date(timestampMillis)).format(DATE_FORMAT_PATTERN);
 }
+
+const PERMISSIONS_ERROR_PATTERN = /no permissions for \[(.+)\] and User \[name=(.+), backend_roles/;
+
+export const prettifyErrorMessage = (rawErrorMessage: string) => {
+  if (isEmpty(rawErrorMessage) || rawErrorMessage === 'undefined') {
+    return 'Unknown error is returned.';
+  }
+  const match = rawErrorMessage.match(PERMISSIONS_ERROR_PATTERN);
+  if (isEmpty(match)) {
+    return rawErrorMessage;
+  } else {
+    return `User ${match[2]} has no permissions to [${match[1]}].`;
+  }
+};
diff --git a/opensearch_dashboards.json b/opensearch_dashboards.json
index a5836447..3bad0752 100644
--- a/opensearch_dashboards.json
+++ b/opensearch_dashboards.json
@@ -9,5 +9,5 @@
     "navigation",
     "opensearchDashboardsUtils"
   ],
-  "optionalPlugins": []
+  "optionalPlugins": ["dataSource","dataSourceManagement"]
 }
\ No newline at end of file
diff --git a/public/app.tsx b/public/app.tsx
index f2a352c8..78e29ef1 100644
--- a/public/app.tsx
+++ b/public/app.tsx
@@ -4,7 +4,12 @@
  */
 
 import React from 'react';
-import { Route, RouteComponentProps, Switch } from 'react-router-dom';
+import {
+  Route,
+  RouteComponentProps,
+  Switch,
+  useLocation,
+} from 'react-router-dom';
 import {
   EuiPageSideBar,
   EuiSideNav,
@@ -18,13 +23,24 @@ import {
   WorkflowDetailRouterProps,
   WorkflowsRouterProps,
 } from './pages';
+import { MountPoint } from '../../../src/core/public';
+import {
+  constructHrefWithDataSourceId,
+  getDataSourceFromURL,
+} from './utils/utils';
 
 // styling
 import './global-styles.scss';
 
-interface Props extends RouteComponentProps {}
+interface Props extends RouteComponentProps {
+  setHeaderActionMenu: (menuMount?: MountPoint) => void;
+}
 
 export const FlowFrameworkDashboardsApp = (props: Props) => {
+  const { setHeaderActionMenu } = props;
+  const location = useLocation();
+  const queryParams = getDataSourceFromURL(location);
+  const dataSourceId = queryParams.dataSourceId;
   const sidebar = (
     <EuiPageSideBar style={{ minWidth: 190 }} hidden={false} paddingSize="l">
       <EuiSideNav
@@ -37,7 +53,10 @@ export const FlowFrameworkDashboardsApp = (props: Props) => {
               {
                 name: Navigation.Workflows,
                 id: 1,
-                href: `#${APP_PATH.WORKFLOWS}`,
+                href: constructHrefWithDataSourceId(
+                  APP_PATH.WORKFLOWS,
+                  dataSourceId
+                ),
                 isSelected: props.location.pathname === APP_PATH.WORKFLOWS,
               },
             ],
@@ -61,12 +80,17 @@ export const FlowFrameworkDashboardsApp = (props: Props) => {
             path={APP_PATH.WORKFLOW_DETAIL}
             render={(
               routeProps: RouteComponentProps<WorkflowDetailRouterProps>
-            ) => <WorkflowDetail {...routeProps} />}
+            ) => (
+              <WorkflowDetail
+                setActionMenu={setHeaderActionMenu}
+                {...routeProps}
+              />
+            )}
           />
           <Route
             path={APP_PATH.WORKFLOWS}
             render={(routeProps: RouteComponentProps<WorkflowsRouterProps>) => (
-              <Workflows {...routeProps} />
+              <Workflows setActionMenu={setHeaderActionMenu} {...routeProps} />
             )}
           />
           {/*
@@ -82,7 +106,12 @@ export const FlowFrameworkDashboardsApp = (props: Props) => {
                   pathname: APP_PATH.WORKFLOWS,
                 });
               }
-              return <Workflows {...routeProps} />;
+              return (
+                <Workflows
+                  setActionMenu={setHeaderActionMenu}
+                  {...routeProps}
+                />
+              );
             }}
           />
         </Switch>
diff --git a/public/pages/workflow_detail/workflow_detail.tsx b/public/pages/workflow_detail/workflow_detail.tsx
index 63fbc6b7..d7e3a7a7 100644
--- a/public/pages/workflow_detail/workflow_detail.tsx
+++ b/public/pages/workflow_detail/workflow_detail.tsx
@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import React, { useEffect } from 'react';
+import React, { useEffect, ReactElement } from 'react';
 import { RouteComponentProps } from 'react-router-dom';
 import { useSelector } from 'react-redux';
 import { ReactFlowProvider } from 'reactflow';
@@ -22,17 +22,30 @@ import {
   DEFAULT_NEW_WORKFLOW_NAME,
   FETCH_ALL_QUERY_BODY,
 } from '../../../common';
+import { MountPoint } from '../../../../../src/core/public';
 
 // styling
 import './workflow-detail-styles.scss';
 import '../../global-styles.scss';
 
+import { getDataSourceId } from '../../utils/utils';
+
+import {
+  getDataSourceManagementPlugin,
+  getDataSourceEnabled,
+  getNotifications,
+  getSavedObjectsClient,
+} from '../../services';
+import { DataSourceViewConfig } from '../../../../../src/plugins/data_source_management/public';
+
 export interface WorkflowDetailRouterProps {
   workflowId: string;
 }
 
 interface WorkflowDetailProps
-  extends RouteComponentProps<WorkflowDetailRouterProps> {}
+  extends RouteComponentProps<WorkflowDetailRouterProps> {
+  setActionMenu: (menuMount?: MountPoint) => void;
+}
 
 /**
  * The workflow details page. This is where users will configure, create, and
@@ -42,6 +55,8 @@ interface WorkflowDetailProps
 
 export function WorkflowDetail(props: WorkflowDetailProps) {
   const dispatch = useAppDispatch();
+  const dataSourceEnabled = getDataSourceEnabled().enabled;
+  const dataSourceId = getDataSourceId();
   const { workflows } = useSelector((state: AppState) => state.workflows);
 
   // selected workflow state
@@ -50,23 +65,51 @@ export function WorkflowDetail(props: WorkflowDetailProps) {
   const workflowName = workflow ? workflow.name : DEFAULT_NEW_WORKFLOW_NAME;
 
   useEffect(() => {
-    getCore().chrome.setBreadcrumbs([
-      BREADCRUMBS.FLOW_FRAMEWORK,
-      BREADCRUMBS.WORKFLOWS,
-      { text: workflowName },
-    ]);
-  });
+    if (dataSourceEnabled) {
+      getCore().chrome.setBreadcrumbs([
+        BREADCRUMBS.FLOW_FRAMEWORK,
+        BREADCRUMBS.WORKFLOWS(dataSourceId),
+        { text: workflowName },
+      ]);
+    } else {
+      getCore().chrome.setBreadcrumbs([
+        BREADCRUMBS.FLOW_FRAMEWORK,
+        BREADCRUMBS.WORKFLOWS(),
+        { text: workflowName },
+      ]);
+    }
+  }, []);
 
   // On initial load:
   // - fetch workflow
   // - fetch available models as their IDs may be used when building flows
   useEffect(() => {
-    dispatch(getWorkflow(workflowId));
-    dispatch(searchModels(FETCH_ALL_QUERY_BODY));
+    dispatch(getWorkflow({ workflowId, dataSourceId }));
+    dispatch(searchModels({ apiBody: FETCH_ALL_QUERY_BODY, dataSourceId }));
   }, []);
 
+  let renderDataSourceComponent: ReactElement | null = null;
+  if (dataSourceEnabled && getDataSourceManagementPlugin()) {
+    const DataSourceMenu = getDataSourceManagementPlugin().ui.getDataSourceMenu<
+      DataSourceViewConfig
+    >();
+    renderDataSourceComponent = (
+      <DataSourceMenu
+        setMenuMountPoint={props.setActionMenu}
+        componentType={'DataSourceView'}
+        componentConfig={{
+          activeOption: [{ id: dataSourceId }],
+          fullWidth: false,
+          savedObjects: getSavedObjectsClient(),
+          notifications: getNotifications(),
+        }}
+      />
+    );
+  }
+
   return (
     <ReactFlowProvider>
+      {dataSourceEnabled && renderDataSourceComponent}
       <EuiPage>
         <EuiPageBody className="workflow-detail stretch-relative">
           <WorkflowDetailHeader workflow={workflow} />
diff --git a/public/pages/workflow_detail/workflow_inputs/processor_inputs/input_transform_modal.tsx b/public/pages/workflow_detail/workflow_inputs/processor_inputs/input_transform_modal.tsx
index 068656a4..f49420c7 100644
--- a/public/pages/workflow_detail/workflow_inputs/processor_inputs/input_transform_modal.tsx
+++ b/public/pages/workflow_detail/workflow_inputs/processor_inputs/input_transform_modal.tsx
@@ -46,6 +46,7 @@ import {
   useAppDispatch,
 } from '../../../../store';
 import { getCore } from '../../../../services';
+import { getDataSourceId } from '../../../../utils/utils';
 import { MapArrayField } from '../input_fields';
 
 interface InputTransformModalProps {
@@ -68,6 +69,7 @@ interface InputTransformModalProps {
  */
 export function InputTransformModal(props: InputTransformModalProps) {
   const dispatch = useAppDispatch();
+  const dataSourceId = getDataSourceId();
   const { values } = useFormikContext<WorkflowFormValues>();
 
   // source input / transformed output state
@@ -120,8 +122,11 @@ export function InputTransformModal(props: InputTransformModalProps) {
                         );
                         await dispatch(
                           simulatePipeline({
-                            pipeline: curIngestPipeline as IngestPipelineConfig,
-                            docs: curDocs,
+                            apiBody: {
+                              pipeline: curIngestPipeline as IngestPipelineConfig,
+                              docs: curDocs,
+                            },
+                            dataSourceId,
                           })
                         )
                           .unwrap()
@@ -169,11 +174,14 @@ export function InputTransformModal(props: InputTransformModalProps) {
                       // the partial search pipeline (inline) to get the latest transformed version of the response.
                       dispatch(
                         searchIndex({
-                          index: values.ingest.index.name,
-                          body: JSON.stringify({
-                            ...JSON.parse(values.search.request as string),
-                            search_pipeline: curSearchPipeline,
-                          }),
+                          apiBody: {
+                            index: values.ingest.index.name,
+                            body: JSON.stringify({
+                              ...JSON.parse(values.search.request as string),
+                              search_pipeline: curSearchPipeline,
+                            }),
+                          },
+                          dataSourceId,
                         })
                       )
                         .unwrap()
diff --git a/public/pages/workflow_detail/workflow_inputs/processor_inputs/output_transform_modal.tsx b/public/pages/workflow_detail/workflow_inputs/processor_inputs/output_transform_modal.tsx
index 246a0887..6cd6ff0c 100644
--- a/public/pages/workflow_detail/workflow_inputs/processor_inputs/output_transform_modal.tsx
+++ b/public/pages/workflow_detail/workflow_inputs/processor_inputs/output_transform_modal.tsx
@@ -48,6 +48,7 @@ import {
 } from '../../../../store';
 import { getCore } from '../../../../services';
 import { MapArrayField } from '../input_fields';
+import { getDataSourceId } from '../../../../utils/utils';
 
 interface OutputTransformModalProps {
   uiConfig: WorkflowConfig;
@@ -65,6 +66,7 @@ interface OutputTransformModalProps {
  */
 export function OutputTransformModal(props: OutputTransformModalProps) {
   const dispatch = useAppDispatch();
+  const dataSourceId = getDataSourceId();
   const { values } = useFormikContext<WorkflowFormValues>();
 
   // source input / transformed output state
@@ -124,8 +126,11 @@ export function OutputTransformModal(props: OutputTransformModalProps) {
                       );
                       await dispatch(
                         simulatePipeline({
-                          pipeline: curIngestPipeline,
-                          docs: curDocs,
+                          apiBody: {
+                            pipeline: curIngestPipeline,
+                            docs: curDocs,
+                          },
+                          dataSourceId,
                         })
                       )
                         .unwrap()
@@ -162,11 +167,14 @@ export function OutputTransformModal(props: OutputTransformModalProps) {
                       // version of the request.
                       dispatch(
                         searchIndex({
-                          index: values.ingest.index.name,
-                          body: JSON.stringify({
-                            ...JSON.parse(values.search.request as string),
-                            search_pipeline: curSearchPipeline,
-                          }),
+                          apiBody: {
+                            index: values.ingest.index.name,
+                            body: JSON.stringify({
+                              ...JSON.parse(values.search.request as string),
+                              search_pipeline: curSearchPipeline,
+                            }),
+                          },
+                          dataSourceId,
                         })
                       )
                         .unwrap()
diff --git a/public/pages/workflow_detail/workflow_inputs/search_inputs/configure_search_request.tsx b/public/pages/workflow_detail/workflow_inputs/search_inputs/configure_search_request.tsx
index a6dac344..998385de 100644
--- a/public/pages/workflow_detail/workflow_inputs/search_inputs/configure_search_request.tsx
+++ b/public/pages/workflow_detail/workflow_inputs/search_inputs/configure_search_request.tsx
@@ -30,6 +30,7 @@ import {
   searchIndex,
   useAppDispatch,
 } from '../../../../store';
+import { getDataSourceId } from '../../../../utils/utils';
 
 interface ConfigureSearchRequestProps {
   setQuery: (query: string) => void;
@@ -42,6 +43,7 @@ interface ConfigureSearchRequestProps {
  */
 export function ConfigureSearchRequest(props: ConfigureSearchRequestProps) {
   const dispatch = useAppDispatch();
+  const dataSourceId = getDataSourceId();
 
   // Form state
   const { values, setFieldValue, setFieldTouched } = useFormikContext<
@@ -80,7 +82,7 @@ export function ConfigureSearchRequest(props: ConfigureSearchRequestProps) {
   useEffect(() => {
     if (!ingestEnabled) {
       // Fetch all indices besides system indices
-      dispatch(catIndices('*,-.*'));
+      dispatch(catIndices({ pattern: '*,-.*', dataSourceId }));
     }
   }, []);
 
@@ -177,9 +179,12 @@ export function ConfigureSearchRequest(props: ConfigureSearchRequestProps) {
               // see https://opensearch.org/docs/latest/search-plugins/search-pipelines/using-search-pipeline/#disabling-the-default-pipeline-for-a-request
               dispatch(
                 searchIndex({
-                  index: values.search.index.name,
-                  body: values.search.request,
-                  searchPipeline: '_none',
+                  apiBody: {
+                    index: values.search.index.name,
+                    body: values.search.request,
+                    searchPipeline: '_none',
+                  },
+                  dataSourceId,
                 })
               )
                 .unwrap()
diff --git a/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx b/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx
index 8e10c11b..d559a375 100644
--- a/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx
+++ b/public/pages/workflow_detail/workflow_inputs/workflow_inputs.tsx
@@ -60,6 +60,7 @@ import {
 } from '../../../utils';
 import { BooleanField } from './input_fields';
 import { ExportOptions } from './export_options';
+import { getDataSourceId } from '../../../utils/utils';
 
 // styling
 import '../workspace/workspace-styles.scss';
@@ -102,6 +103,7 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
     touched,
   } = useFormikContext<WorkflowFormValues>();
   const dispatch = useAppDispatch();
+  const dataSourceId = getDataSourceId();
 
   // Overall workspace state
   const { isDirty } = useSelector((state: AppState) => state.form);
@@ -160,17 +162,25 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
         } as WorkflowTemplate;
         await dispatch(
           updateWorkflow({
-            workflowId: props.workflow?.id as string,
-            workflowTemplate: updatedTemplate,
-            updateFields: true,
-            reprovision: false,
+            apiBody: {
+              workflowId: props.workflow?.id as string,
+              workflowTemplate: updatedTemplate,
+              updateFields: true,
+              reprovision: false,
+            },
+            dataSourceId,
           })
         )
           .unwrap()
           .then(async (result) => {
             // get any updates after autosave
             new Promise((f) => setTimeout(f, 1000)).then(async () => {
-              dispatch(getWorkflow(props.workflow?.id as string));
+              dispatch(
+                getWorkflow({
+                  workflowId: props.workflow?.id as string,
+                  dataSourceId,
+                })
+              );
             });
           })
           .catch((error: any) => {
@@ -204,9 +214,12 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
     if (reprovision) {
       await dispatch(
         updateWorkflow({
-          workflowId: updatedWorkflow.id as string,
-          workflowTemplate: reduceToTemplate(updatedWorkflow),
-          reprovision: true,
+          apiBody: {
+            workflowId: updatedWorkflow.id as string,
+            workflowTemplate: reduceToTemplate(updatedWorkflow),
+            reprovision: true,
+          },
+          dataSourceId,
         })
       )
         .unwrap()
@@ -219,7 +232,12 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
           // auto-fetching to cover the majority of provisioning updates which
           // are inexpensive and will finish within milliseconds.
           setTimeout(async () => {
-            dispatch(getWorkflow(updatedWorkflow.id as string));
+            dispatch(
+              getWorkflow({
+                workflowId: updatedWorkflow.id as string,
+                dataSourceId,
+              })
+            );
           }, 1000);
         })
         .catch((error: any) => {
@@ -228,23 +246,34 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
     } else {
       await dispatch(
         deprovisionWorkflow({
-          workflowId: updatedWorkflow.id as string,
-          resourceIds: getResourcesToBeForceDeleted(props.workflow),
+          apiBody: {
+            workflowId: updatedWorkflow.id as string,
+            resourceIds: getResourcesToBeForceDeleted(props.workflow),
+          },
+          dataSourceId,
         })
       )
         .unwrap()
         .then(async (result) => {
           await dispatch(
             updateWorkflow({
-              workflowId: updatedWorkflow.id as string,
-              workflowTemplate: reduceToTemplate(updatedWorkflow),
-              reprovision: false,
+              apiBody: {
+                workflowId: updatedWorkflow.id as string,
+                workflowTemplate: reduceToTemplate(updatedWorkflow),
+                reprovision: false,
+              },
+              dataSourceId,
             })
           )
             .unwrap()
             .then(async (result) => {
               await sleep(1000);
-              await dispatch(provisionWorkflow(updatedWorkflow.id as string))
+              await dispatch(
+                provisionWorkflow({
+                  workflowId: updatedWorkflow.id as string,
+                  dataSourceId,
+                })
+              )
                 .unwrap()
                 .then(async (result) => {
                   await sleep(1000);
@@ -255,7 +284,12 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
                   // auto-fetching to cover the majority of provisioning updates which
                   // are inexpensive and will finish within milliseconds.
                   setTimeout(async () => {
-                    dispatch(getWorkflow(updatedWorkflow.id as string));
+                    dispatch(
+                      getWorkflow({
+                        workflowId: updatedWorkflow.id as string,
+                        dataSourceId,
+                      })
+                    );
                   }, 1000);
                 })
                 .catch((error: any) => {
@@ -342,7 +376,7 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
             values.ingest.index.name,
             ingestDocsObjs
           );
-          dispatch(bulk({ body: bulkBody }))
+          dispatch(bulk({ apiBody: { body: bulkBody }, dataSourceId }))
             .unwrap()
             .then(async (resp) => {
               props.setIngestResponse(JSON.stringify(resp, undefined, 2));
@@ -387,7 +421,12 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
 
         if (success) {
           const indexName = values.search.index.name;
-          dispatch(searchIndex({ index: indexName, body: props.query }))
+          dispatch(
+            searchIndex({
+              apiBody: { index: indexName, body: props.query },
+              dataSourceId,
+            })
+          )
             .unwrap()
             .then(async (resp) => {
               props.setQueryResponse(
@@ -470,17 +509,25 @@ export function WorkflowInputs(props: WorkflowInputsProps) {
                     onClick={async () => {
                       await dispatch(
                         deprovisionWorkflow({
-                          workflowId: props.workflow?.id as string,
-                          resourceIds: getResourcesToBeForceDeleted(
-                            props.workflow
-                          ),
+                          apiBody: {
+                            workflowId: props.workflow?.id as string,
+                            resourceIds: getResourcesToBeForceDeleted(
+                              props.workflow
+                            ),
+                          },
+                          dataSourceId,
                         })
                       )
                         .unwrap()
                         .then(async (result) => {
                           setFieldValue('ingest.enabled', false);
                           // @ts-ignore
-                          await dispatch(getWorkflow(props.workflow.id));
+                          await dispatch(
+                            getWorkflow({
+                              workflowId: props.workflow.id,
+                              dataSourceId,
+                            })
+                          );
                         })
                         .catch((error: any) => {})
                         .finally(() => {
diff --git a/public/pages/workflows/import_workflow/import_workflow_modal.tsx b/public/pages/workflows/import_workflow/import_workflow_modal.tsx
index a223d4dc..f713a7c6 100644
--- a/public/pages/workflows/import_workflow/import_workflow_modal.tsx
+++ b/public/pages/workflows/import_workflow/import_workflow_modal.tsx
@@ -32,6 +32,7 @@ import {
 } from '../../../store';
 import { FETCH_ALL_QUERY_BODY, Workflow } from '../../../../common';
 import { WORKFLOWS_TAB } from '../workflows';
+import {  getDataSourceId } from '../../../utils/utils';
 
 interface ImportWorkflowModalProps {
   isImportModalOpen: boolean;
@@ -48,6 +49,7 @@ interface ImportWorkflowModalProps {
  */
 export function ImportWorkflowModal(props: ImportWorkflowModalProps) {
   const dispatch = useAppDispatch();
+  const dataSourceId = getDataSourceId();
 
   // file contents & file obj state
   const [fileContents, setFileContents] = useState<string | undefined>(
@@ -133,11 +135,21 @@ export function ImportWorkflowModal(props: ImportWorkflowModalProps) {
         <EuiButton
           disabled={!isValidWorkflow(fileObj)}
           onClick={() => {
-            dispatch(createWorkflow(fileObj as Workflow))
+            dispatch(
+              createWorkflow({
+                apiBody: fileObj as Workflow,
+                dataSourceId,
+              })
+            )
               .unwrap()
               .then((result) => {
                 const { workflow } = result;
-                dispatch(searchWorkflows(FETCH_ALL_QUERY_BODY));
+                dispatch(
+                  searchWorkflows({
+                    apiBody: FETCH_ALL_QUERY_BODY,
+                    dataSourceId,
+                  })
+                );
                 props.setSelectedTabId(WORKFLOWS_TAB.MANAGE);
                 getCore().notifications.toasts.addSuccess(
                   `Successfully imported ${workflow.name}`
diff --git a/public/pages/workflows/new_workflow/use_case.tsx b/public/pages/workflows/new_workflow/use_case.tsx
index 0262836a..64d16030 100644
--- a/public/pages/workflows/new_workflow/use_case.tsx
+++ b/public/pages/workflows/new_workflow/use_case.tsx
@@ -26,6 +26,7 @@ import { Workflow } from '../../../../common';
 import { APP_PATH } from '../../../utils';
 import { processWorkflowName } from './utils';
 import { createWorkflow, useAppDispatch } from '../../../store';
+import { getDataSourceId } from '../../../utils/utils';
 
 interface UseCaseProps {
   workflow: Workflow;
@@ -33,6 +34,7 @@ interface UseCaseProps {
 
 export function UseCase(props: UseCaseProps) {
   const dispatch = useAppDispatch();
+  const dataSourceId = getDataSourceId();
   const history = useHistory();
 
   // name modal state
@@ -78,11 +80,18 @@ export function UseCase(props: UseCaseProps) {
                   ...props.workflow,
                   name: workflowName,
                 };
-                dispatch(createWorkflow(workflowToCreate))
+                dispatch(
+                  createWorkflow({
+                    apiBody: workflowToCreate,
+                    dataSourceId,
+                  })
+                )
                   .unwrap()
                   .then((result) => {
                     const { workflow } = result;
-                    history.replace(`${APP_PATH.WORKFLOWS}/${workflow.id}`);
+                    history.replace(
+                      `${APP_PATH.WORKFLOWS}/${workflow.id}?dataSourceId=${dataSourceId}`
+                    );
                     history.go(0);
                   })
                   .catch((err: any) => {
diff --git a/public/pages/workflows/workflow_list/columns.tsx b/public/pages/workflows/workflow_list/columns.tsx
index 1637dff1..46956ba0 100644
--- a/public/pages/workflows/workflow_list/columns.tsx
+++ b/public/pages/workflows/workflow_list/columns.tsx
@@ -7,39 +7,53 @@ import React from 'react';
 import { EuiLink } from '@elastic/eui';
 import {
   EMPTY_FIELD_STRING,
-  PLUGIN_ID,
   Workflow,
   toFormattedDate,
 } from '../../../../common';
+import {
+  constructHrefWithDataSourceId,
+  getDataSourceId,
+} from '../../../utils/utils';
+
+export const columns = (actions: any[]) => {
+  const dataSourceId = getDataSourceId();
 
-export const columns = (actions: any[]) => [
-  {
-    field: 'name',
-    name: 'Name',
-    width: '33%',
-    sortable: true,
-    render: (name: string, workflow: Workflow) => (
-      <EuiLink href={`${PLUGIN_ID}#/workflows/${workflow.id}`}>{name}</EuiLink>
-    ),
-  },
-  {
-    field: 'ui_metadata.type',
-    name: 'Type',
-    width: '33%',
-    sortable: true,
-  },
-  {
-    field: 'lastUpdated',
-    name: 'Last saved',
-    width: '33%',
-    sortable: true,
-    render: (lastUpdated: number) =>
-      lastUpdated !== undefined
-        ? toFormattedDate(lastUpdated)
-        : EMPTY_FIELD_STRING,
-  },
-  {
-    name: 'Actions',
-    actions,
-  },
-];
+  return [
+    {
+      field: 'name',
+      name: 'Name',
+      width: '33%',
+      sortable: true,
+      render: (name: string, workflow: Workflow) => (
+        <EuiLink
+          href={constructHrefWithDataSourceId(
+            `/workflows/${workflow.id}`,
+            dataSourceId
+          )}
+        >
+          {name}
+        </EuiLink>
+      ),
+    },
+    {
+      field: 'ui_metadata.type',
+      name: 'Type',
+      width: '33%',
+      sortable: true,
+    },
+    {
+      field: 'lastUpdated',
+      name: 'Last saved',
+      width: '33%',
+      sortable: true,
+      render: (lastUpdated: number) =>
+        lastUpdated !== undefined
+          ? toFormattedDate(lastUpdated)
+          : EMPTY_FIELD_STRING,
+    },
+    {
+      name: 'Actions',
+      actions,
+    },
+  ];
+};
diff --git a/public/pages/workflows/workflow_list/workflow_list.tsx b/public/pages/workflows/workflow_list/workflow_list.tsx
index 70972512..ee944ac1 100644
--- a/public/pages/workflows/workflow_list/workflow_list.tsx
+++ b/public/pages/workflows/workflow_list/workflow_list.tsx
@@ -30,6 +30,7 @@ import {
 } from '../../../general_components';
 import { WORKFLOWS_TAB } from '../workflows';
 import { getCore } from '../../../services';
+import { getDataSourceId } from '../../../utils/utils';
 
 interface WorkflowListProps {
   setSelectedTabId: (tabId: WORKFLOWS_TAB) => void;
@@ -65,6 +66,7 @@ const filterOptions = [
  */
 export function WorkflowList(props: WorkflowListProps) {
   const dispatch = useAppDispatch();
+  const dataSourceId = getDataSourceId();
   const { workflows, loading } = useSelector(
     (state: AppState) => state.workflows
   );
@@ -144,7 +146,12 @@ export function WorkflowList(props: WorkflowListProps) {
           }}
           onConfirm={async () => {
             clearDeleteState();
-            await dispatch(deleteWorkflow(selectedWorkflow.id as string))
+            await dispatch(
+              deleteWorkflow({
+                workflowId: selectedWorkflow.id as string,
+                dataSourceId,
+              })
+            )
               .unwrap()
               .then((result) => {
                 getCore().notifications.toasts.addSuccess(
diff --git a/public/pages/workflows/workflows.test.tsx b/public/pages/workflows/workflows.test.tsx
index cdf14c57..5d82ce9e 100644
--- a/public/pages/workflows/workflows.test.tsx
+++ b/public/pages/workflows/workflows.test.tsx
@@ -30,7 +30,9 @@ const renderWithRouter = () => ({
       <Router>
         <Switch>
           <Route
-            render={(props: RouteComponentProps) => <Workflows {...props} />}
+            render={(props: RouteComponentProps) => (
+              <Workflows setActionMenu={jest.fn()} {...props} />
+            )}
           />
         </Switch>
       </Router>
diff --git a/public/pages/workflows/workflows.tsx b/public/pages/workflows/workflows.tsx
index dda03517..1a5ca628 100644
--- a/public/pages/workflows/workflows.tsx
+++ b/public/pages/workflows/workflows.tsx
@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import React, { useEffect, useState } from 'react';
+import React, { useEffect, useState, useMemo } from 'react';
 import { RouteComponentProps, useLocation } from 'react-router-dom';
 import {
   EuiPageHeader,
@@ -26,10 +26,25 @@ import { AppState, searchWorkflows, useAppDispatch } from '../../store';
 import { EmptyListMessage } from './empty_list_message';
 import { FETCH_ALL_QUERY_BODY } from '../../../common';
 import { ImportWorkflowModal } from './import_workflow';
+import { MountPoint } from '../../../../../src/core/public';
+import { DataSourceSelectableConfig } from '../../../../../src/plugins/data_source_management/public';
+
+import { getDataSourceFromURL } from '../../utils/utils';
+
+import {
+  getDataSourceManagementPlugin,
+  getDataSourceEnabled,
+  getNotifications,
+  getSavedObjectsClient,
+} from '../../services';
+
+import { prettifyErrorMessage } from '../../../common/utils';
 
 export interface WorkflowsRouterProps {}
 
-interface WorkflowsProps extends RouteComponentProps<WorkflowsRouterProps> {}
+interface WorkflowsProps extends RouteComponentProps<WorkflowsRouterProps> {
+  setActionMenu: (menuMount: MountPoint | undefined) => void;
+}
 
 export enum WORKFLOWS_TAB {
   MANAGE = 'manage',
@@ -38,11 +53,16 @@ export enum WORKFLOWS_TAB {
 
 const ACTIVE_TAB_PARAM = 'tab';
 
-function replaceActiveTab(activeTab: string, props: WorkflowsProps) {
+function replaceActiveTab(
+  activeTab: string,
+  props: WorkflowsProps,
+  dataSourceId?: string
+) {
   props.history.replace({
     ...history,
     search: queryString.stringify({
       [ACTIVE_TAB_PARAM]: activeTab,
+      dataSourceId,
     }),
   });
 }
@@ -54,6 +74,12 @@ function replaceActiveTab(activeTab: string, props: WorkflowsProps) {
  */
 export function Workflows(props: WorkflowsProps) {
   const dispatch = useAppDispatch();
+  const location = useLocation();
+  const queryParams = getDataSourceFromURL(location);
+  const dataSourceEnabled = getDataSourceEnabled().enabled;
+  const [dataSourceId, setDataSourceId] = useState<string>(
+    queryParams.dataSourceId
+  );
   const { workflows, loading } = useSelector(
     (state: AppState) => state.workflows
   );
@@ -74,29 +100,103 @@ export function Workflows(props: WorkflowsProps) {
       !Object.values(WORKFLOWS_TAB).includes(selectedTabId)
     ) {
       setSelectedTabId(WORKFLOWS_TAB.MANAGE);
-      replaceActiveTab(WORKFLOWS_TAB.MANAGE, props);
+      replaceActiveTab(WORKFLOWS_TAB.MANAGE, props, dataSourceId);
     }
   }, [selectedTabId, workflows]);
 
   // If the user navigates back to the manage tab, re-fetch workflows
   useEffect(() => {
     if (selectedTabId === WORKFLOWS_TAB.MANAGE) {
-      dispatch(searchWorkflows(FETCH_ALL_QUERY_BODY));
+      dispatch(
+        searchWorkflows({
+          apiBody: FETCH_ALL_QUERY_BODY,
+          dataSourceId: dataSourceId,
+        })
+      );
     }
   }, [selectedTabId]);
 
   useEffect(() => {
-    getCore().chrome.setBreadcrumbs([
-      BREADCRUMBS.FLOW_FRAMEWORK,
-      BREADCRUMBS.WORKFLOWS,
-    ]);
+    if (dataSourceEnabled) {
+      getCore().chrome.setBreadcrumbs([
+        BREADCRUMBS.FLOW_FRAMEWORK,
+        BREADCRUMBS.WORKFLOWS(dataSourceId),
+      ]);
+    } else {
+      getCore().chrome.setBreadcrumbs([
+        BREADCRUMBS.FLOW_FRAMEWORK,
+        BREADCRUMBS.WORKFLOWS(),
+      ]);
+    }
   });
 
   // On initial render: fetch all workflows
   useEffect(() => {
-    dispatch(searchWorkflows(FETCH_ALL_QUERY_BODY));
+    dispatch(
+      searchWorkflows({
+        apiBody: FETCH_ALL_QUERY_BODY,
+        dataSourceId: dataSourceId,
+      })
+    );
   }, []);
 
+  useEffect(() => {
+    const { history, location } = props;
+    if (dataSourceEnabled) {
+      const updatedParams = {
+        dataSourceId: dataSourceId,
+      };
+
+      history.replace({
+        ...location,
+        search: queryString.stringify(updatedParams),
+      });
+    }
+    dispatch(
+      searchWorkflows({
+        apiBody: FETCH_ALL_QUERY_BODY,
+        dataSourceId: dataSourceId,
+      })
+    );
+  }, [dataSourceId, setDataSourceId]);
+
+  const handleDataSourceChange = ([event]) => {
+    const dataSourceEventId = event?.id;
+    if (dataSourceEnabled) {
+      if (dataSourceEventId === undefined) {
+        getNotifications().toasts.addDanger(
+          prettifyErrorMessage('Unable to set data source.')
+        );
+      } else {
+        setDataSourceId(dataSourceEventId);
+      }
+    }
+  };
+
+  let renderDataSourceComponent = null;
+  if (dataSourceEnabled && getDataSourceManagementPlugin()) {
+    const DataSourceMenu = getDataSourceManagementPlugin().ui.getDataSourceMenu<
+      DataSourceSelectableConfig
+    >();
+    renderDataSourceComponent = useMemo(() => {
+      return (
+        <DataSourceMenu
+          setMenuMountPoint={props.setActionMenu}
+          componentType={'DataSourceSelectable'}
+          componentConfig={{
+            fullWidth: false,
+            activeOption:
+              dataSourceId === undefined ? undefined : [{ id: dataSourceId }],
+            savedObjects: getSavedObjectsClient(),
+            notifications: getNotifications(),
+            onSelectedDataSources: (dataSources) =>
+              handleDataSourceChange(dataSources),
+          }}
+        />
+      );
+    }, [getSavedObjectsClient, getNotifications(), props.setActionMenu]);
+  }
+
   return (
     <>
       {isImportModalOpen && (
@@ -106,6 +206,7 @@ export function Workflows(props: WorkflowsProps) {
           setSelectedTabId={setSelectedTabId}
         />
       )}
+      {dataSourceEnabled && renderDataSourceComponent}
       <EuiPage>
         <EuiPageBody>
           <EuiPageHeader
@@ -127,7 +228,7 @@ export function Workflows(props: WorkflowsProps) {
                 isSelected: selectedTabId === WORKFLOWS_TAB.MANAGE,
                 onClick: () => {
                   setSelectedTabId(WORKFLOWS_TAB.MANAGE);
-                  replaceActiveTab(WORKFLOWS_TAB.MANAGE, props);
+                  replaceActiveTab(WORKFLOWS_TAB.MANAGE, props, dataSourceId);
                 },
               },
               {
@@ -136,7 +237,7 @@ export function Workflows(props: WorkflowsProps) {
                 isSelected: selectedTabId === WORKFLOWS_TAB.CREATE,
                 onClick: () => {
                   setSelectedTabId(WORKFLOWS_TAB.CREATE);
-                  replaceActiveTab(WORKFLOWS_TAB.CREATE, props);
+                  replaceActiveTab(WORKFLOWS_TAB.CREATE, props, dataSourceId);
                 },
               },
             ]}
@@ -203,7 +304,7 @@ export function Workflows(props: WorkflowsProps) {
                 <EmptyListMessage
                   onClickNewWorkflow={() => {
                     setSelectedTabId(WORKFLOWS_TAB.CREATE);
-                    replaceActiveTab(WORKFLOWS_TAB.CREATE, props);
+                    replaceActiveTab(WORKFLOWS_TAB.CREATE, props, dataSourceId);
                   }}
                 />
               )}
diff --git a/public/plugin.ts b/public/plugin.ts
index 940b8565..f8c406ff 100644
--- a/public/plugin.ts
+++ b/public/plugin.ts
@@ -10,11 +10,18 @@ import {
   Plugin,
 } from '../../../src/core/public';
 import {
-  FlowFrameworkDashboardsPluginSetup,
   FlowFrameworkDashboardsPluginStart,
+  FlowFrameworkDashboardsPluginSetup,
 } from './types';
 import { PLUGIN_ID } from '../common';
-import { setCore, setRouteService } from './services';
+import {
+  setCore,
+  setRouteService,
+  setSavedObjectsClient,
+  setDataSourceManagementPlugin,
+  setDataSourceEnabled,
+  setNotifications,
+} from './services';
 import { configureRoutes } from './route_service';
 
 export class FlowFrameworkDashboardsPlugin
@@ -23,7 +30,10 @@ export class FlowFrameworkDashboardsPlugin
       FlowFrameworkDashboardsPluginSetup,
       FlowFrameworkDashboardsPluginStart
     > {
-  public setup(core: CoreSetup): FlowFrameworkDashboardsPluginSetup {
+  public setup(
+    core: CoreSetup,
+    plugins: any
+  ): FlowFrameworkDashboardsPluginSetup {
     // Register the plugin in the side navigation
     core.application.register({
       id: PLUGIN_ID,
@@ -45,10 +55,18 @@ export class FlowFrameworkDashboardsPlugin
         return renderApp(coreStart, params);
       },
     });
-    return {};
+    setDataSourceManagementPlugin(plugins.dataSourceManagement);
+    const enabled = !!plugins.dataSource;
+    setDataSourceEnabled({ enabled });
+    return {
+      dataSourceManagement: plugins.dataSourceManagement,
+      dataSource: plugins.dataSource,
+    };
   }
 
   public start(core: CoreStart): FlowFrameworkDashboardsPluginStart {
+    setNotifications(core.notifications);
+    setSavedObjectsClient(core.savedObjects.client);
     return {};
   }
 
diff --git a/public/render_app.tsx b/public/render_app.tsx
index 07bf09e8..b1c03d88 100644
--- a/public/render_app.tsx
+++ b/public/render_app.tsx
@@ -14,20 +14,24 @@ import { store } from './store';
 // styling
 import './global-styles.scss';
 
-export const renderApp = (
-  coreStart: CoreStart,
-  { appBasePath, element }: AppMountParameters
-) => {
+export const renderApp = (coreStart: CoreStart, params: AppMountParameters) => {
   // This is so our base element stretches to fit the entire webpage
-  element.className = 'stretch-absolute';
+  params.element.className = 'stretch-absolute';
   ReactDOM.render(
     <Provider store={store}>
-      <Router basename={appBasePath + '#/'}>
-        <Route render={(props) => <FlowFrameworkDashboardsApp {...props} />} />
+      <Router basename={params.appBasePath + '#/'}>
+        <Route
+          render={(props) => (
+            <FlowFrameworkDashboardsApp
+              setHeaderActionMenu={params.setHeaderActionMenu}
+              {...props}
+            />
+          )}
+        />
       </Router>
     </Provider>,
-    element
+    params.element
   );
 
-  return () => ReactDOM.unmountComponentAtNode(element);
+  return () => ReactDOM.unmountComponentAtNode(params.element);
 };
diff --git a/public/route_service.ts b/public/route_service.ts
index 14435646..1ba67046 100644
--- a/public/route_service.ts
+++ b/public/route_service.ts
@@ -23,6 +23,7 @@ import {
   IngestPipelineConfig,
   SimulateIngestPipelineDoc,
   BULK_NODE_API_PATH,
+  BASE_NODE_API_PATH,
 } from '../common';
 
 /**
@@ -33,81 +34,138 @@ import {
  * Used in redux by wrapping them in async thunk functions which mutate redux state when executed.
  */
 export interface RouteService {
-  getWorkflow: (workflowId: string) => Promise<any | HttpFetchError>;
-  searchWorkflows: (body: {}) => Promise<any | HttpFetchError>;
-  getWorkflowState: (workflowId: string) => Promise<any | HttpFetchError>;
-  createWorkflow: (body: {}) => Promise<any | HttpFetchError>;
+  getWorkflow: (
+    workflowId: string,
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
+  searchWorkflows: (
+    body: {},
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
+  getWorkflowState: (
+    workflowId: string,
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
+  createWorkflow: (
+    body: {},
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
   updateWorkflow: (
     workflowId: string,
     workflowTemplate: WorkflowTemplate,
     updateFields: boolean,
-    reprovision: boolean
+    reprovision: boolean,
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
+  provisionWorkflow: (
+    workflowId: string,
+    dataSourceId?: string
   ) => Promise<any | HttpFetchError>;
-  provisionWorkflow: (workflowId: string) => Promise<any | HttpFetchError>;
-  deprovisionWorkflow: (
+  deprovisionWorkflow: ({
+    workflowId,
+    dataSourceId,
+    resourceIds,
+  }: {
+    workflowId: string;
+    dataSourceId?: string;
+    resourceIds?: string;
+  }) => Promise<any | HttpFetchError>;
+  deleteWorkflow: (
     workflowId: string,
-    resourceIds?: string
+    dataSourceId?: string
   ) => Promise<any | HttpFetchError>;
-  deleteWorkflow: (workflowId: string) => Promise<any | HttpFetchError>;
   getWorkflowPresets: () => Promise<any | HttpFetchError>;
-  catIndices: (pattern: string) => Promise<any | HttpFetchError>;
-  searchIndex: (
+  catIndices: (
+    pattern: string,
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
+  searchIndex: ({
+    index,
+    body,
+    dataSourceId,
+    searchPipeline,
+  }: {
+    index: string;
+    body: {};
+    dataSourceId?: string;
+    searchPipeline?: string;
+  }) => Promise<any | HttpFetchError>;
+  ingest: (
     index: string,
-    body: {},
-    searchPipeline?: string
+    doc: {},
+    dataSourceId?: string
   ) => Promise<any | HttpFetchError>;
-  ingest: (index: string, doc: {}) => Promise<any | HttpFetchError>;
-  bulk: (body: {}, ingestPipeline?: string) => Promise<any | HttpFetchError>;
-  searchModels: (body: {}) => Promise<any | HttpFetchError>;
-  simulatePipeline: (body: {
-    pipeline: IngestPipelineConfig;
-    docs: SimulateIngestPipelineDoc[];
+  bulk: ({
+    body,
+    dataSourceId,
+    ingestPipeline,
+  }: {
+    body: {};
+    dataSourceId?: string;
+    ingestPipeline?: string;
   }) => Promise<any | HttpFetchError>;
+  searchModels: (
+    body: {},
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
+  simulatePipeline: (
+    body: {
+      pipeline: IngestPipelineConfig;
+      docs: SimulateIngestPipelineDoc[];
+    },
+    dataSourceId?: string
+  ) => Promise<any | HttpFetchError>;
 }
 
 export function configureRoutes(core: CoreStart): RouteService {
   return {
-    getWorkflow: async (workflowId: string) => {
+    getWorkflow: async (workflowId: string, dataSourceId?: string) => {
       try {
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow`
+          : GET_WORKFLOW_NODE_API_PATH;
         const response = await core.http.get<{ respString: string }>(
-          `${GET_WORKFLOW_NODE_API_PATH}/${workflowId}`
+          `${url}/${workflowId}`
         );
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    searchWorkflows: async (body: {}) => {
+    searchWorkflows: async (body: {}, dataSourceId?: string) => {
       try {
-        const response = await core.http.post<{ respString: string }>(
-          SEARCH_WORKFLOWS_NODE_API_PATH,
-          {
-            body: JSON.stringify(body),
-          }
-        );
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow/search`
+          : SEARCH_WORKFLOWS_NODE_API_PATH;
+        const response = await core.http.post<{ respString: string }>(url, {
+          body: JSON.stringify(body),
+        });
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    getWorkflowState: async (workflowId: string) => {
+    getWorkflowState: async (workflowId: string, dataSourceId?: string) => {
       try {
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow/state`
+          : GET_WORKFLOW_STATE_NODE_API_PATH;
         const response = await core.http.get<{ respString: string }>(
-          `${GET_WORKFLOW_STATE_NODE_API_PATH}/${workflowId}`
+          `${url}/${workflowId}`
         );
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    createWorkflow: async (body: {}) => {
+    createWorkflow: async (body: {}, dataSourceId?: string) => {
       try {
-        const response = await core.http.post<{ respString: string }>(
-          CREATE_WORKFLOW_NODE_API_PATH,
-          {
-            body: JSON.stringify(body),
-          }
-        );
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow/create`
+          : CREATE_WORKFLOW_NODE_API_PATH;
+        const response = await core.http.post<{ respString: string }>(url, {
+          body: JSON.stringify(body),
+        });
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
@@ -117,11 +175,15 @@ export function configureRoutes(core: CoreStart): RouteService {
       workflowId: string,
       workflowTemplate: WorkflowTemplate,
       updateFields: boolean,
-      reprovision: boolean
+      reprovision: boolean,
+      dataSourceId?: string
     ) => {
       try {
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow/update`
+          : UPDATE_WORKFLOW_NODE_API_PATH;
         const response = await core.http.put<{ respString: string }>(
-          `${UPDATE_WORKFLOW_NODE_API_PATH}/${workflowId}/${updateFields}/${reprovision}`,
+          `${url}/${workflowId}/${updateFields}/${reprovision}`,
           {
             body: JSON.stringify(workflowTemplate),
           }
@@ -131,31 +193,48 @@ export function configureRoutes(core: CoreStart): RouteService {
         return e as HttpFetchError;
       }
     },
-    provisionWorkflow: async (workflowId: string) => {
+    provisionWorkflow: async (workflowId: string, dataSourceId?: 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 }>(
-          `${PROVISION_WORKFLOW_NODE_API_PATH}/${workflowId}`
+          `${url}/${workflowId}`
         );
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    deprovisionWorkflow: async (workflowId: string, resourceIds?: string) => {
+    deprovisionWorkflow: async ({
+      workflowId,
+      dataSourceId,
+      resourceIds,
+    }: {
+      workflowId: string;
+      dataSourceId?: string;
+      resourceIds?: string;
+    }) => {
       try {
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow/deprovision`
+          : DEPROVISION_WORKFLOW_NODE_API_PATH;
         const path = resourceIds
-          ? `${DEPROVISION_WORKFLOW_NODE_API_PATH}/${workflowId}/${resourceIds}`
-          : `${DEPROVISION_WORKFLOW_NODE_API_PATH}/${workflowId}`;
+          ? `${url}/${workflowId}/${resourceIds}`
+          : `${url}/${workflowId}`;
         const response = await core.http.post<{ respString: string }>(path);
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    deleteWorkflow: async (workflowId: string) => {
+    deleteWorkflow: async (workflowId: string, dataSourceId?: string) => {
       try {
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/workflow/delete`
+          : DELETE_WORKFLOW_NODE_API_PATH;
         const response = await core.http.delete<{ respString: string }>(
-          `${DELETE_WORKFLOW_NODE_API_PATH}/${workflowId}`
+          `${url}/${workflowId}`
         );
         return response;
       } catch (e: any) {
@@ -172,19 +251,35 @@ export function configureRoutes(core: CoreStart): RouteService {
         return e as HttpFetchError;
       }
     },
-    catIndices: async (pattern: string) => {
+    catIndices: async (pattern: string, dataSourceId?: string) => {
       try {
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/opensearch/catIndices`
+          : CAT_INDICES_NODE_API_PATH;
         const response = await core.http.get<{ respString: string }>(
-          `${CAT_INDICES_NODE_API_PATH}/${pattern}`
+          `${url}/${pattern}`
         );
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    searchIndex: async (index: string, body: {}, searchPipeline?: string) => {
+    searchIndex: async ({
+      index,
+      body,
+      dataSourceId,
+      searchPipeline,
+    }: {
+      index: string;
+      body: {};
+      dataSourceId?: string;
+      searchPipeline?: string;
+    }) => {
       try {
-        const basePath = `${SEARCH_INDEX_NODE_API_PATH}/${index}`;
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/opensearch/search`
+          : SEARCH_INDEX_NODE_API_PATH;
+        const basePath = `${url}/${index}`;
         const path = searchPipeline
           ? `${basePath}/${searchPipeline}`
           : basePath;
@@ -196,10 +291,13 @@ export function configureRoutes(core: CoreStart): RouteService {
         return e as HttpFetchError;
       }
     },
-    ingest: async (index: string, doc: {}) => {
+    ingest: async (index: string, doc: {}, dataSourceId?: string) => {
       try {
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/opensearch/ingest`
+          : INGEST_NODE_API_PATH;
         const response = await core.http.put<{ respString: string }>(
-          `${INGEST_NODE_API_PATH}/${index}`,
+          `${url}/${index}`,
           {
             body: JSON.stringify(doc),
           }
@@ -209,11 +307,20 @@ export function configureRoutes(core: CoreStart): RouteService {
         return e as HttpFetchError;
       }
     },
-    bulk: async (body: {}, ingestPipeline?: string) => {
+    bulk: async ({
+      body,
+      dataSourceId,
+      ingestPipeline,
+    }: {
+      body: {};
+      dataSourceId?: string;
+      ingestPipeline?: string;
+    }) => {
       try {
-        const path = ingestPipeline
-          ? `${BULK_NODE_API_PATH}/${ingestPipeline}`
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/opensearch/bulk`
           : BULK_NODE_API_PATH;
+        const path = ingestPipeline ? `${url}/${ingestPipeline}` : url;
         const response = await core.http.post<{ respString: string }>(path, {
           body: JSON.stringify(body),
         });
@@ -222,30 +329,33 @@ export function configureRoutes(core: CoreStart): RouteService {
         return e as HttpFetchError;
       }
     },
-    searchModels: async (body: {}) => {
+    searchModels: async (body: {}, dataSourceId?: string) => {
       try {
-        const response = await core.http.post<{ respString: string }>(
-          SEARCH_MODELS_NODE_API_PATH,
-          {
-            body: JSON.stringify(body),
-          }
-        );
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/model/search`
+          : SEARCH_MODELS_NODE_API_PATH;
+        const response = await core.http.post<{ respString: string }>(url, {
+          body: JSON.stringify(body),
+        });
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
       }
     },
-    simulatePipeline: async (body: {
-      pipeline: IngestPipelineConfig;
-      docs: SimulateIngestPipelineDoc[];
-    }) => {
+    simulatePipeline: async (
+      body: {
+        pipeline: IngestPipelineConfig;
+        docs: SimulateIngestPipelineDoc[];
+      },
+      dataSourceId?: string
+    ) => {
       try {
-        const response = await core.http.post<{ respString: string }>(
-          SIMULATE_PIPELINE_NODE_API_PATH,
-          {
-            body: JSON.stringify(body),
-          }
-        );
+        const url = dataSourceId
+          ? `${BASE_NODE_API_PATH}/${dataSourceId}/opensearch/simulatePipeline`
+          : SIMULATE_PIPELINE_NODE_API_PATH;
+        const response = await core.http.post<{ respString: string }>(url, {
+          body: JSON.stringify(body),
+        });
         return response;
       } catch (e: any) {
         return e as HttpFetchError;
diff --git a/public/services.ts b/public/services.ts
index bd4d85a2..35eeb56b 100644
--- a/public/services.ts
+++ b/public/services.ts
@@ -4,11 +4,36 @@
  */
 
 import { createGetterSetter } from '../../../src/plugins/opensearch_dashboards_utils/public';
-import { CoreStart } from '../../../src/core/public';
+import { CoreStart, NotificationsStart } from '../../../src/core/public';
 import { RouteService } from './route_service';
+import { DataSourceManagementPluginSetup } from '../../../src/plugins/data_source_management/public';
+
+export interface DataSourceEnabled {
+  enabled: boolean;
+}
 
 export const [getCore, setCore] = createGetterSetter<CoreStart>('Core');
 
 export const [getRouteService, setRouteService] = createGetterSetter<
   RouteService
 >('');
+
+export const [
+  getSavedObjectsClient,
+  setSavedObjectsClient,
+] = createGetterSetter<CoreStart['savedObjects']['client']>(
+  'SavedObjectsClient'
+);
+
+export const [
+  getDataSourceManagementPlugin,
+  setDataSourceManagementPlugin,
+] = createGetterSetter<DataSourceManagementPluginSetup>('DataSourceManagement');
+
+export const [getDataSourceEnabled, setDataSourceEnabled] = createGetterSetter<
+  DataSourceEnabled
+>('DataSourceEnabled');
+
+export const [getNotifications, setNotifications] = createGetterSetter<
+  NotificationsStart
+>('Notifications');
diff --git a/public/store/reducers/models_reducer.ts b/public/store/reducers/models_reducer.ts
index a43ccb79..00941610 100644
--- a/public/store/reducers/models_reducer.ts
+++ b/public/store/reducers/models_reducer.ts
@@ -19,9 +19,13 @@ const SEARCH_MODELS_ACTION = `${MODELS_ACTION_PREFIX}/searchModels`;
 
 export const searchModels = createAsyncThunk(
   SEARCH_MODELS_ACTION,
-  async (body: {}, { rejectWithValue }) => {
+  async (
+    { apiBody, dataSourceId }: { apiBody: {}, dataSourceId: string },
+    { rejectWithValue }
+  ) => {
     const response: any | HttpFetchError = await getRouteService().searchModels(
-      body
+      apiBody,
+      dataSourceId
     );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
diff --git a/public/store/reducers/opensearch_reducer.ts b/public/store/reducers/opensearch_reducer.ts
index c977a05b..ddffae86 100644
--- a/public/store/reducers/opensearch_reducer.ts
+++ b/public/store/reducers/opensearch_reducer.ts
@@ -27,11 +27,15 @@ const SIMULATE_PIPELINE_ACTION = `${OPENSEARCH_PREFIX}/simulatePipeline`;
 
 export const catIndices = createAsyncThunk(
   CAT_INDICES_ACTION,
-  async (pattern: string, { rejectWithValue }) => {
+  async (
+    { pattern, dataSourceId }: { pattern: string; dataSourceId?: string },
+    { rejectWithValue }
+  ) => {
     // defaulting to fetch everything except system indices (starting with '.')
     const patternString = pattern || '*,-.*';
     const response: any | HttpFetchError = await getRouteService().catIndices(
-      patternString
+      patternString,
+      dataSourceId
     );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
@@ -46,15 +50,22 @@ export const catIndices = createAsyncThunk(
 export const searchIndex = createAsyncThunk(
   SEARCH_INDEX_ACTION,
   async (
-    searchIndexInfo: { index: string; body: {}; searchPipeline?: string },
+    {
+      apiBody,
+      dataSourceId,
+    }: {
+      apiBody: { index: string; body: {}; searchPipeline?: string };
+      dataSourceId?: string;
+    },
     { rejectWithValue }
   ) => {
-    const { index, body, searchPipeline } = searchIndexInfo;
-    const response: any | HttpFetchError = await getRouteService().searchIndex(
+    const { index, body, searchPipeline } = apiBody;
+    const response: any | HttpFetchError = await getRouteService().searchIndex({
       index,
       body,
-      searchPipeline
-    );
+      dataSourceId,
+      searchPipeline,
+    });
     if (response instanceof HttpFetchError) {
       return rejectWithValue('Error searching index: ' + response.body.message);
     } else {
@@ -65,11 +76,21 @@ export const searchIndex = createAsyncThunk(
 
 export const ingest = createAsyncThunk(
   INGEST_ACTION,
-  async (ingestInfo: { index: string; doc: {} }, { rejectWithValue }) => {
-    const { index, doc } = ingestInfo;
+  async (
+    {
+      apiBody,
+      dataSourceId,
+    }: {
+      apiBody: { index: string; doc: {} };
+      dataSourceId?: string;
+    },
+    { rejectWithValue }
+  ) => {
+    const { index, doc } = apiBody;
     const response: any | HttpFetchError = await getRouteService().ingest(
       index,
-      doc
+      doc,
+      dataSourceId
     );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
@@ -84,14 +105,21 @@ export const ingest = createAsyncThunk(
 export const bulk = createAsyncThunk(
   BULK_ACTION,
   async (
-    bulkInfo: { body: {}; ingestPipeline?: string },
+    {
+      apiBody,
+      dataSourceId,
+    }: {
+      apiBody: { body: {}; ingestPipeline?: string };
+      dataSourceId?: string;
+    },
     { rejectWithValue }
   ) => {
-    const { body, ingestPipeline } = bulkInfo;
-    const response: any | HttpFetchError = await getRouteService().bulk(
+    const { body, ingestPipeline } = apiBody;
+    const response: any | HttpFetchError = await getRouteService().bulk({
       body,
-      ingestPipeline
-    );
+      dataSourceId,
+      ingestPipeline,
+    });
     if (response instanceof HttpFetchError) {
       return rejectWithValue('Error performing bulk: ' + response.body.message);
     } else {
@@ -103,12 +131,24 @@ export const bulk = createAsyncThunk(
 export const simulatePipeline = createAsyncThunk(
   SIMULATE_PIPELINE_ACTION,
   async (
-    body: { pipeline: IngestPipelineConfig; docs: SimulateIngestPipelineDoc[] },
+    {
+      apiBody,
+      dataSourceId,
+    }: {
+      apiBody: {
+        pipeline: IngestPipelineConfig;
+        docs: SimulateIngestPipelineDoc[];
+      };
+      dataSourceId?: string;
+    },
     { rejectWithValue }
   ) => {
     const response:
       | any
-      | HttpFetchError = await getRouteService().simulatePipeline(body);
+      | HttpFetchError = await getRouteService().simulatePipeline(
+      apiBody,
+      dataSourceId
+    );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
         'Error simulating ingest pipeline: ' + response.body.message
diff --git a/public/store/reducers/workflows_reducer.ts b/public/store/reducers/workflows_reducer.ts
index a87e3545..53a8e588 100644
--- a/public/store/reducers/workflows_reducer.ts
+++ b/public/store/reducers/workflows_reducer.ts
@@ -26,9 +26,13 @@ const DELETE_WORKFLOW_ACTION = `${WORKFLOWS_ACTION_PREFIX}/delete`;
 
 export const getWorkflow = createAsyncThunk(
   GET_WORKFLOW_ACTION,
-  async (workflowId: string, { rejectWithValue }) => {
+  async (
+    { workflowId, dataSourceId }: { workflowId: string; dataSourceId?: string },
+    { rejectWithValue }
+  ) => {
     const response: any | HttpFetchError = await getRouteService().getWorkflow(
-      workflowId
+      workflowId,
+      dataSourceId
     );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
@@ -42,10 +46,16 @@ export const getWorkflow = createAsyncThunk(
 
 export const searchWorkflows = createAsyncThunk(
   SEARCH_WORKFLOWS_ACTION,
-  async (body: {}, { rejectWithValue }) => {
+  async (
+    { apiBody, dataSourceId }: { apiBody: {}; dataSourceId?: string },
+    { rejectWithValue }
+  ) => {
     const response:
       | any
-      | HttpFetchError = await getRouteService().searchWorkflows(body);
+      | HttpFetchError = await getRouteService().searchWorkflows(
+      apiBody,
+      dataSourceId
+    );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
         'Error searching workflows: ' + response.body.message
@@ -58,10 +68,16 @@ export const searchWorkflows = createAsyncThunk(
 
 export const getWorkflowState = createAsyncThunk(
   GET_WORKFLOW_STATE_ACTION,
-  async (workflowId: string, { rejectWithValue }) => {
+  async (
+    { workflowId, dataSourceId }: { workflowId: string; dataSourceId?: string },
+    { rejectWithValue }
+  ) => {
     const response:
       | any
-      | HttpFetchError = await getRouteService().getWorkflowState(workflowId);
+      | HttpFetchError = await getRouteService().getWorkflowState(
+      workflowId,
+      dataSourceId
+    );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
         'Error getting workflow state: ' + response.body.message
@@ -74,10 +90,16 @@ export const getWorkflowState = createAsyncThunk(
 
 export const createWorkflow = createAsyncThunk(
   CREATE_WORKFLOW_ACTION,
-  async (workflowBody: {}, { rejectWithValue }) => {
+  async (
+    { apiBody, dataSourceId }: { apiBody: {}; dataSourceId?: string },
+    { rejectWithValue }
+  ) => {
     const response:
       | any
-      | HttpFetchError = await getRouteService().createWorkflow(workflowBody);
+      | HttpFetchError = await getRouteService().createWorkflow(
+      apiBody,
+      dataSourceId
+    );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
         'Error creating workflow: ' + response.body.message
@@ -91,27 +113,29 @@ export const createWorkflow = createAsyncThunk(
 export const updateWorkflow = createAsyncThunk(
   UPDATE_WORKFLOW_ACTION,
   async (
-    workflowInfo: {
-      workflowId: string;
-      workflowTemplate: WorkflowTemplate;
-      updateFields?: boolean;
-      reprovision?: boolean;
+    {
+      apiBody,
+      dataSourceId,
+    }: {
+      apiBody: {
+        workflowId: string;
+        workflowTemplate: WorkflowTemplate;
+        updateFields?: boolean;
+        reprovision?: boolean;
+      };
+      dataSourceId?: string;
     },
     { rejectWithValue }
   ) => {
-    const {
-      workflowId,
-      workflowTemplate,
-      updateFields,
-      reprovision,
-    } = workflowInfo;
+    const { workflowId, workflowTemplate, updateFields, reprovision } = apiBody;
     const response:
       | any
       | HttpFetchError = await getRouteService().updateWorkflow(
       workflowId,
       workflowTemplate,
       updateFields || false,
-      reprovision || false
+      reprovision || false,
+      dataSourceId
     );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
@@ -125,10 +149,16 @@ export const updateWorkflow = createAsyncThunk(
 
 export const provisionWorkflow = createAsyncThunk(
   PROVISION_WORKFLOW_ACTION,
-  async (workflowId: string, { rejectWithValue }) => {
+  async (
+    { workflowId, dataSourceId }: { workflowId: string; dataSourceId?: string },
+    { rejectWithValue }
+  ) => {
     const response:
       | any
-      | HttpFetchError = await getRouteService().provisionWorkflow(workflowId);
+      | HttpFetchError = await getRouteService().provisionWorkflow(
+      workflowId,
+      dataSourceId
+    );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
         'Error provisioning workflow: ' + response.body.message
@@ -142,16 +172,23 @@ export const provisionWorkflow = createAsyncThunk(
 export const deprovisionWorkflow = createAsyncThunk(
   DEPROVISION_WORKFLOW_ACTION,
   async (
-    deprovisionInfo: { workflowId: string; resourceIds?: string },
+    {
+      apiBody,
+      dataSourceId,
+    }: {
+      apiBody: { workflowId: string; resourceIds?: string };
+      dataSourceId?: string;
+    },
     { rejectWithValue }
   ) => {
-    const { workflowId, resourceIds } = deprovisionInfo;
+    const { workflowId, resourceIds } = apiBody;
     const response:
       | any
-      | HttpFetchError = await getRouteService().deprovisionWorkflow(
+      | HttpFetchError = await getRouteService().deprovisionWorkflow({
       workflowId,
-      resourceIds
-    );
+      resourceIds,
+      dataSourceId,
+    });
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
         'Error deprovisioning workflow: ' + response.body.message
@@ -164,10 +201,16 @@ export const deprovisionWorkflow = createAsyncThunk(
 
 export const deleteWorkflow = createAsyncThunk(
   DELETE_WORKFLOW_ACTION,
-  async (workflowId: string, { rejectWithValue }) => {
+  async (
+    { workflowId, dataSourceId }: { workflowId: string; dataSourceId?: string },
+    { rejectWithValue }
+  ) => {
     const response:
       | any
-      | HttpFetchError = await getRouteService().deleteWorkflow(workflowId);
+      | HttpFetchError = await getRouteService().deleteWorkflow(
+      workflowId,
+      dataSourceId
+    );
     if (response instanceof HttpFetchError) {
       return rejectWithValue(
         'Error deleting workflow: ' + response.body.message
diff --git a/public/types.ts b/public/types.ts
index 41726cd7..d46b8319 100644
--- a/public/types.ts
+++ b/public/types.ts
@@ -4,8 +4,14 @@
  */
 
 import { NavigationPublicPluginStart } from '../../../src/plugins/navigation/public';
+import { DataSourceManagementPluginSetup } from '../../../src/plugins/data_source_management/public';
+import { DataSourcePluginSetup } from '../../../src/plugins/data_source/public';
+
+export interface FlowFrameworkDashboardsPluginSetup {
+  dataSourceManagement: DataSourceManagementPluginSetup;
+  dataSource: DataSourcePluginSetup;
+}
 
-export interface FlowFrameworkDashboardsPluginSetup {}
 export interface FlowFrameworkDashboardsPluginStart {}
 
 export interface AppPluginStartDependencies {
diff --git a/public/utils/constants.ts b/public/utils/constants.ts
index 45647716..151ebb63 100644
--- a/public/utils/constants.ts
+++ b/public/utils/constants.ts
@@ -3,6 +3,8 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+import { constructHrefWithDataSourceId } from './utils';
+
 export enum Navigation {
   FlowFramework = 'Flow Framework',
   Workflows = 'Workflows',
@@ -16,5 +18,8 @@ export enum APP_PATH {
 
 export const BREADCRUMBS = Object.freeze({
   FLOW_FRAMEWORK: { text: 'Flow Framework' },
-  WORKFLOWS: { text: 'Workflows', href: `#${APP_PATH.WORKFLOWS}` },
+  WORKFLOWS: (dataSourceId?: string) => ({
+    text: 'Workflows',
+    href: constructHrefWithDataSourceId(APP_PATH.WORKFLOWS, dataSourceId),
+  }),
 });
diff --git a/public/utils/utils.ts b/public/utils/utils.ts
index 828052c0..5b0e5882 100644
--- a/public/utils/utils.ts
+++ b/public/utils/utils.ts
@@ -20,7 +20,10 @@ import {
   WORKFLOW_STEP_TYPE,
   Workflow,
 } from '../../common';
-import { getCore } from '../services';
+import { getCore, getDataSourceEnabled } from '../services';
+import { MDSQueryParams } from '../../common/interfaces';
+import queryString from 'query-string';
+import { useLocation } from 'react-router-dom';
 
 // Append 16 random characters
 export function generateId(prefix?: string): string {
@@ -89,7 +92,7 @@ export function getResourcesToBeForceDeleted(
 }
 
 export function getObjFromJsonOrYamlString(
-  fileContents: string | undefined
+  fileContents?: string
 ): object | undefined {
   try {
     const jsonObj = JSON.parse(fileContents);
@@ -229,6 +232,37 @@ export function parseModelOutputs(
   );
 }
 
+export const getDataSourceFromURL = (location: {
+  search: string;
+}): MDSQueryParams => {
+  const queryParams = queryString.parse(location.search);
+  const dataSourceId = queryParams.dataSourceId;
+  return {
+    dataSourceId: typeof dataSourceId === 'string' ? dataSourceId : undefined,
+  };
+};
+
+export const constructHrefWithDataSourceId = (
+  basePath: string,
+  dataSourceId: string = ''
+): string => {
+  const dataSourceEnabled = getDataSourceEnabled().enabled;
+  const url = new URLSearchParams();
+  if (dataSourceEnabled && dataSourceId !== undefined) {
+    url.set('dataSourceId', dataSourceId);
+  }
+  // we share this helper function to construct the href with dataSourceId
+  // some places we need to return the url with hash, some places we don't need to
+  // so adding this flag to indicate if we want to return the url with hash
+  return `#${basePath}?${url.toString()}`;
+};
+
+export const getDataSourceId = () => {
+  const location = useLocation();
+  const mdsQueryParams = getDataSourceFromURL(location);
+  return mdsQueryParams.dataSourceId;
+};
+
 // converts camelCase to a space-delimited string with the first word capitalized.
 // useful for converting config IDs (in snake_case) to a formatted form title
 export function camelCaseToTitleString(snakeCaseString: string): string {
diff --git a/server/plugin.ts b/server/plugin.ts
index 62c8ae27..98ae9f76 100644
--- a/server/plugin.ts
+++ b/server/plugin.ts
@@ -24,9 +24,16 @@ import {
   registerMLRoutes,
   MLRoutesService,
 } from './routes';
+import { DataSourcePluginSetup } from '../../../src/plugins/data_source/server/types';
+import { DataSourceManagementPlugin } from '../../../src/plugins/data_source_management/public';
 
 import { ILegacyClusterClient } from '../../../src/core/server/';
 
+export interface FlowFrameworkPluginSetupDependencies {
+  dataSourceManagement?: ReturnType<DataSourceManagementPlugin['setup']>;
+  dataSource?: DataSourcePluginSetup;
+}
+
 export class FlowFrameworkDashboardsPlugin
   implements
     Plugin<
@@ -41,7 +48,10 @@ export class FlowFrameworkDashboardsPlugin
     this.globalConfig$ = initializerContext.config.legacy.globalConfig$;
   }
 
-  public async setup(core: CoreSetup) {
+  public async setup(
+    core: CoreSetup,
+    { dataSource }: FlowFrameworkPluginSetupDependencies
+  ) {
     this.logger.debug('flow-framework-dashboards: Setup');
     const router = core.http.createRouter();
 
@@ -57,9 +67,20 @@ export class FlowFrameworkDashboardsPlugin
       }
     );
 
-    const opensearchRoutesService = new OpenSearchRoutesService(client);
-    const flowFrameworkRoutesService = new FlowFrameworkRoutesService(client);
-    const mlRoutesService = new MLRoutesService(client);
+    const dataSourceEnabled = !!dataSource;
+    if (dataSourceEnabled) {
+      dataSource.registerCustomApiSchema(flowFrameworkPlugin);
+      dataSource.registerCustomApiSchema(mlPlugin);
+    }
+    const opensearchRoutesService = new OpenSearchRoutesService(
+      client,
+      dataSourceEnabled
+    );
+    const flowFrameworkRoutesService = new FlowFrameworkRoutesService(
+      client,
+      dataSourceEnabled
+    );
+    const mlRoutesService = new MLRoutesService(client, dataSourceEnabled);
 
     // Register server side APIs with the corresponding service functions
     registerOpenSearchRoutes(router, opensearchRoutesService);
diff --git a/server/routes/flow_framework_routes_service.ts b/server/routes/flow_framework_routes_service.ts
index 861e0752..5efa27d0 100644
--- a/server/routes/flow_framework_routes_service.ts
+++ b/server/routes/flow_framework_routes_service.ts
@@ -14,6 +14,7 @@ import {
   OpenSearchDashboardsResponseFactory,
 } from '../../../../src/core/server';
 import {
+  BASE_NODE_API_PATH,
   CREATE_WORKFLOW_NODE_API_PATH,
   DELETE_WORKFLOW_NODE_API_PATH,
   DEPROVISION_WORKFLOW_NODE_API_PATH,
@@ -30,6 +31,7 @@ import {
   WorkflowResource,
   WorkflowTemplate,
 } from '../../common';
+import { getClientBasedOnDataSource } from '../utils/helpers';
 import {
   generateCustomError,
   getResourcesCreatedFromResponse,
@@ -59,6 +61,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.getWorkflow
   );
 
+  router.get(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/{workflow_id}`,
+      validate: {
+        params: schema.object({
+          workflow_id: schema.string(),
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    flowFrameworkRoutesService.getWorkflow
+  );
+
   router.post(
     {
       path: SEARCH_WORKFLOWS_NODE_API_PATH,
@@ -69,6 +84,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.searchWorkflows
   );
 
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/search`,
+      validate: {
+        params: schema.object({
+          data_source_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    flowFrameworkRoutesService.searchWorkflows
+  );
+
   router.get(
     {
       path: `${GET_WORKFLOW_STATE_NODE_API_PATH}/{workflow_id}`,
@@ -81,6 +109,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.getWorkflowState
   );
 
+  router.get(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/state/{workflow_id}`,
+      validate: {
+        params: schema.object({
+          workflow_id: schema.string(),
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    flowFrameworkRoutesService.getWorkflowState
+  );
+
   router.post(
     {
       path: CREATE_WORKFLOW_NODE_API_PATH,
@@ -91,6 +132,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.createWorkflow
   );
 
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/create`,
+      validate: {
+        params: schema.object({
+          data_source_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    flowFrameworkRoutesService.createWorkflow
+  );
+
   router.put(
     {
       path: `${UPDATE_WORKFLOW_NODE_API_PATH}/{workflow_id}/{update_fields}/{reprovision}`,
@@ -106,6 +160,22 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.updateWorkflow
   );
 
+  router.put(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/update/{workflow_id}/{update_fields}/{reprovision}`,
+      validate: {
+        params: schema.object({
+          data_source_id: schema.string(),
+          workflow_id: schema.string(),
+          update_fields: schema.boolean(),
+          reprovision: schema.boolean(),
+        }),
+        body: schema.any(),
+      },
+    },
+    flowFrameworkRoutesService.updateWorkflow
+  );
+
   router.post(
     {
       path: `${PROVISION_WORKFLOW_NODE_API_PATH}/{workflow_id}`,
@@ -118,6 +188,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.provisionWorkflow
   );
 
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/provision/{workflow_id}`,
+      validate: {
+        params: schema.object({
+          workflow_id: schema.string(),
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    flowFrameworkRoutesService.provisionWorkflow
+  );
+
   router.post(
     {
       path: `${DEPROVISION_WORKFLOW_NODE_API_PATH}/{workflow_id}`,
@@ -130,6 +213,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.deprovisionWorkflow
   );
 
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/deprovision/{workflow_id}`,
+      validate: {
+        params: schema.object({
+          workflow_id: schema.string(),
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    flowFrameworkRoutesService.deprovisionWorkflow
+  );
+
   router.post(
     {
       path: `${DEPROVISION_WORKFLOW_NODE_API_PATH}/{workflow_id}/{resource_ids}`,
@@ -143,6 +239,20 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.deprovisionWorkflow
   );
 
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/deprovision/{workflow_id}/{resource_ids}`,
+      validate: {
+        params: schema.object({
+          workflow_id: schema.string(),
+          resource_ids: schema.string(),
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    flowFrameworkRoutesService.deprovisionWorkflow
+  );
+
   router.delete(
     {
       path: `${DELETE_WORKFLOW_NODE_API_PATH}/{workflow_id}`,
@@ -155,6 +265,19 @@ export function registerFlowFrameworkRoutes(
     flowFrameworkRoutesService.deleteWorkflow
   );
 
+  router.delete(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/workflow/delete/{workflow_id}`,
+      validate: {
+        params: schema.object({
+          workflow_id: schema.string(),
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    flowFrameworkRoutesService.deleteWorkflow
+  );
+
   router.get(
     {
       path: GET_PRESET_WORKFLOWS_NODE_API_PATH,
@@ -165,10 +288,12 @@ export function registerFlowFrameworkRoutes(
 }
 
 export class FlowFrameworkRoutesService {
+  dataSourceEnabled: boolean;
   private client: any;
 
-  constructor(client: any) {
+  constructor(client: any, dataSourceEnabled: boolean) {
     this.client = client;
+    this.dataSourceEnabled = dataSourceEnabled;
   }
 
   // TODO: can remove or simplify if we can fetch all data from a single API call. Tracking issue:
@@ -181,14 +306,25 @@ export class FlowFrameworkRoutesService {
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const { workflow_id } = req.params as { workflow_id: string };
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.getWorkflow', { workflow_id });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+      const response = await callWithRequest('flowFramework.getWorkflow', {
+        workflow_id,
+      });
       const workflow = toWorkflowObj(response, workflow_id);
 
-      const stateResponse = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.getWorkflowState', { workflow_id });
+      const stateResponse = await callWithRequest(
+        'flowFramework.getWorkflowState',
+        {
+          workflow_id,
+        }
+      );
       const state = getWorkflowStateFromResponse(
         stateResponse.state as typeof WORKFLOW_STATE
       );
@@ -217,14 +353,29 @@ export class FlowFrameworkRoutesService {
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const body = req.body;
     try {
-      const workflowsResponse = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.searchWorkflows', { body });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+
+      const workflowsResponse = await callWithRequest(
+        'flowFramework.searchWorkflows',
+        {
+          body,
+        }
+      );
       const workflowHits = workflowsResponse.hits.hits as SearchHit[];
 
-      const workflowStatesResponse = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.searchWorkflowState', { body });
+      const workflowStatesResponse = await callWithRequest(
+        'flowFramework.searchWorkflowState',
+        {
+          body,
+        }
+      );
       const workflowStateHits = workflowStatesResponse.hits.hits as SearchHit[];
 
       const workflowDict = getWorkflowsFromResponses(
@@ -247,11 +398,17 @@ export class FlowFrameworkRoutesService {
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const { workflow_id } = req.params as { workflow_id: string };
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.getWorkflowState', {
-          workflow_id,
-        });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+      const response = await callWithRequest('flowFramework.getWorkflowState', {
+        workflow_id,
+      });
       const state = getWorkflowStateFromResponse(
         response.state as typeof WORKFLOW_STATE | undefined
       );
@@ -277,9 +434,17 @@ export class FlowFrameworkRoutesService {
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const body = req.body as Workflow;
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.createWorkflow', { body });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+      const response = await callWithRequest('flowFramework.createWorkflow', {
+        body,
+      });
       const workflowWithId = {
         ...body,
         id: response.workflow_id,
@@ -302,15 +467,21 @@ export class FlowFrameworkRoutesService {
     };
     const workflowTemplate = req.body as WorkflowTemplate;
     try {
-      await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.updateWorkflow', {
-          workflow_id,
-          // default update_fields to false if not explicitly set otherwise
-          update_fields: update_fields,
-          reprovision: reprovision,
-          body: workflowTemplate,
-        });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+      await callWithRequest('flowFramework.updateWorkflow', {
+        workflow_id,
+        // default update_fields to false if not explicitly set otherwise
+        update_fields: update_fields,
+        reprovision: reprovision,
+        body: workflowTemplate,
+      });
 
       return res.ok({ body: { workflowId: workflow_id, workflowTemplate } });
     } catch (err: any) {
@@ -325,9 +496,17 @@ export class FlowFrameworkRoutesService {
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const { workflow_id } = req.params as { workflow_id: string };
     try {
-      await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.provisionWorkflow', { workflow_id });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+      await callWithRequest('flowFramework.provisionWorkflow', {
+        workflow_id,
+      });
       return res.ok();
     } catch (err: any) {
       return generateCustomError(res, err);
@@ -344,19 +523,23 @@ export class FlowFrameworkRoutesService {
       resource_ids?: string;
     };
     try {
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
       if (resource_ids !== undefined) {
-        await this.client
-          .asScoped(req)
-          .callAsCurrentUser('flowFramework.forceDeprovisionWorkflow', {
-            workflow_id,
-            resource_ids,
-          });
+        await callWithRequest('flowFramework.forceDeprovisionWorkflow', {
+          workflow_id,
+          resource_ids,
+        });
       } else {
-        await this.client
-          .asScoped(req)
-          .callAsCurrentUser('flowFramework.deprovisionWorkflow', {
-            workflow_id,
-          });
+        await callWithRequest('flowFramework.deprovisionWorkflow', {
+          workflow_id,
+        });
       }
       return res.ok();
     } catch (err: any) {
@@ -371,9 +554,18 @@ export class FlowFrameworkRoutesService {
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const { workflow_id } = req.params as { workflow_id: string };
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('flowFramework.deleteWorkflow', { workflow_id });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+
+      const response = await callWithRequest('flowFramework.deleteWorkflow', {
+        workflow_id,
+      });
       return res.ok({ body: { id: response._id } });
     } catch (err: any) {
       return generateCustomError(res, err);
diff --git a/server/routes/ml_routes_service.ts b/server/routes/ml_routes_service.ts
index 31ef1e63..9f6e6639 100644
--- a/server/routes/ml_routes_service.ts
+++ b/server/routes/ml_routes_service.ts
@@ -11,8 +11,9 @@ import {
   OpenSearchDashboardsRequest,
   OpenSearchDashboardsResponseFactory,
 } from '../../../../src/core/server';
-import { SEARCH_MODELS_NODE_API_PATH, SearchHit } from '../../common';
+import { SEARCH_MODELS_NODE_API_PATH, BASE_NODE_API_PATH, SearchHit } from '../../common';
 import { generateCustomError, getModelsFromResponses } from './helpers';
+import { getClientBasedOnDataSource } from '../utils/helpers';
 
 /**
  * Server-side routes to process ml-plugin-related node API calls and execute the
@@ -31,13 +32,27 @@ export function registerMLRoutes(
     },
     mlRoutesService.searchModels
   );
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/model/search`,
+      validate: {
+        body: schema.any(),
+        params: schema.object({
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    mlRoutesService.searchModels
+  );
 }
 
 export class MLRoutesService {
   private client: any;
+  dataSourceEnabled: boolean;
 
-  constructor(client: any) {
+  constructor(client: any, dataSourceEnabled: boolean) {
     this.client = client;
+    this.dataSourceEnabled = dataSourceEnabled;
   }
 
   searchModels = async (
@@ -47,9 +62,18 @@ export class MLRoutesService {
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const body = req.body;
     try {
-      const modelsResponse = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('mlClient.searchModels', { body });
+      const { data_source_id = '' } = req.params as { data_source_id?: string };
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+      const modelsResponse = await callWithRequest('mlClient.searchModels', {
+        body,
+      });
+
       const modelHits = modelsResponse.hits.hits as SearchHit[];
       const modelDict = getModelsFromResponses(modelHits);
 
diff --git a/server/routes/opensearch_routes_service.ts b/server/routes/opensearch_routes_service.ts
index 084e2a3b..dfcd61af 100644
--- a/server/routes/opensearch_routes_service.ts
+++ b/server/routes/opensearch_routes_service.ts
@@ -12,6 +12,7 @@ import {
   OpenSearchDashboardsResponseFactory,
 } from '../../../../src/core/server';
 import {
+  BASE_NODE_API_PATH,
   BULK_NODE_API_PATH,
   CAT_INDICES_NODE_API_PATH,
   INGEST_NODE_API_PATH,
@@ -23,6 +24,7 @@ import {
   SimulateIngestPipelineResponse,
 } from '../../common';
 import { generateCustomError } from './helpers';
+import { getClientBasedOnDataSource } from '../utils/helpers';
 
 /**
  * Server-side routes to process OpenSearch-related node API calls and execute the
@@ -43,6 +45,18 @@ export function registerOpenSearchRoutes(
     },
     opensearchRoutesService.catIndices
   );
+  router.get(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/opensearch/catIndices/{pattern}`,
+      validate: {
+        params: schema.object({
+          pattern: schema.string(),
+          data_source_id: schema.string(),
+        }),
+      },
+    },
+    opensearchRoutesService.catIndices
+  );
   router.post(
     {
       path: `${SEARCH_INDEX_NODE_API_PATH}/{index}`,
@@ -55,6 +69,19 @@ export function registerOpenSearchRoutes(
     },
     opensearchRoutesService.searchIndex
   );
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/opensearch/search/{index}`,
+      validate: {
+        params: schema.object({
+          index: schema.string(),
+          data_source_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    opensearchRoutesService.searchIndex
+  );
   router.post(
     {
       path: `${SEARCH_INDEX_NODE_API_PATH}/{index}/{search_pipeline}`,
@@ -68,6 +95,20 @@ export function registerOpenSearchRoutes(
     },
     opensearchRoutesService.searchIndex
   );
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/opensearch/search/{index}/{search_pipeline}`,
+      validate: {
+        params: schema.object({
+          index: schema.string(),
+          search_pipeline: schema.string(),
+          data_source_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    opensearchRoutesService.searchIndex
+  );
   router.put(
     {
       path: `${INGEST_NODE_API_PATH}/{index}`,
@@ -80,6 +121,19 @@ export function registerOpenSearchRoutes(
     },
     opensearchRoutesService.ingest
   );
+  router.put(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/opensearch/ingest/{index}`,
+      validate: {
+        params: schema.object({
+          index: schema.string(),
+          data_source_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    opensearchRoutesService.ingest
+  );
   router.post(
     {
       path: `${BULK_NODE_API_PATH}/{pipeline}`,
@@ -92,6 +146,19 @@ export function registerOpenSearchRoutes(
     },
     opensearchRoutesService.bulk
   );
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/opensearch/bulk/{pipeline}`,
+      validate: {
+        params: schema.object({
+          pipeline: schema.string(),
+          data_source_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    opensearchRoutesService.bulk
+  );
   router.post(
     {
       path: BULK_NODE_API_PATH,
@@ -101,6 +168,18 @@ export function registerOpenSearchRoutes(
     },
     opensearchRoutesService.bulk
   );
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/opensearch/bulk`,
+      validate: {
+        params: schema.object({
+          data_source_id: schema.string(),
+        }),
+        body: schema.any(),
+      },
+    },
+    opensearchRoutesService.bulk
+  );
   router.post(
     {
       path: SIMULATE_PIPELINE_NODE_API_PATH,
@@ -113,13 +192,30 @@ export function registerOpenSearchRoutes(
     },
     opensearchRoutesService.simulatePipeline
   );
+  router.post(
+    {
+      path: `${BASE_NODE_API_PATH}/{data_source_id}/opensearch/simulatePipeline`,
+      validate: {
+        params: schema.object({
+          data_source_id: schema.string(),
+        }),
+        body: schema.object({
+          pipeline: schema.any(),
+          docs: schema.any(),
+        }),
+      },
+    },
+    opensearchRoutesService.simulatePipeline
+  );
 }
 
 export class OpenSearchRoutesService {
   private client: any;
+  dataSourceEnabled: boolean;
 
-  constructor(client: any) {
+  constructor(client: any, dataSourceEnabled: boolean) {
     this.client = client;
+    this.dataSourceEnabled = dataSourceEnabled;
   }
 
   catIndices = async (
@@ -128,14 +224,21 @@ export class OpenSearchRoutesService {
     res: OpenSearchDashboardsResponseFactory
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
     const { pattern } = req.params as { pattern: string };
+    const { data_source_id = '' } = req.params as { data_source_id?: string };
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('cat.indices', {
-          index: pattern,
-          format: 'json',
-          h: 'health,index',
-        });
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+
+      const response = await callWithRequest('cat.indices', {
+        index: pattern,
+        format: 'json',
+        h: 'health,index',
+      });
 
       // re-formatting the index results to match Index
       const cleanedIndices = response.map((index: any) => ({
@@ -158,15 +261,22 @@ export class OpenSearchRoutesService {
       index: string;
       search_pipeline: string | undefined;
     };
+    const { data_source_id = '' } = req.params as { data_source_id?: string };
     const body = req.body;
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('search', {
-          index,
-          body,
-          search_pipeline,
-        });
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+
+      const response = await callWithRequest('search', {
+        index,
+        body,
+        search_pipeline,
+      });
 
       return res.ok({ body: response });
     } catch (err: any) {
@@ -179,15 +289,22 @@ export class OpenSearchRoutesService {
     req: OpenSearchDashboardsRequest,
     res: OpenSearchDashboardsResponseFactory
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
+    const { data_source_id = '' } = req.params as { data_source_id?: string };
     const { index } = req.params as { index: string };
     const doc = req.body;
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('index', {
-          index,
-          body: doc,
-        });
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+
+      const response = await callWithRequest('index', {
+        index,
+        body: doc,
+      });
 
       return res.ok({ body: response });
     } catch (err: any) {
@@ -200,18 +317,25 @@ export class OpenSearchRoutesService {
     req: OpenSearchDashboardsRequest,
     res: OpenSearchDashboardsResponseFactory
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
+    const { data_source_id = '' } = req.params as { data_source_id?: string };
     const { pipeline } = req.params as {
       pipeline: string | undefined;
     };
     const body = req.body;
 
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('bulk', {
-          body,
-          pipeline,
-        });
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+
+      const response = await callWithRequest('bulk', {
+        body,
+        pipeline,
+      });
 
       return res.ok({ body: response });
     } catch (err: any) {
@@ -224,14 +348,24 @@ export class OpenSearchRoutesService {
     req: OpenSearchDashboardsRequest,
     res: OpenSearchDashboardsResponseFactory
   ): Promise<IOpenSearchDashboardsResponse<any>> => {
+    const { data_source_id = '' } = req.params as { data_source_id?: string };
     const { pipeline, docs } = req.body as {
       pipeline: IngestPipelineConfig;
       docs: SimulateIngestPipelineDoc[];
     };
     try {
-      const response = await this.client
-        .asScoped(req)
-        .callAsCurrentUser('ingest.simulate', { body: { pipeline, docs } });
+      const callWithRequest = getClientBasedOnDataSource(
+        context,
+        this.dataSourceEnabled,
+        req,
+        data_source_id,
+        this.client
+      );
+
+      const response = await callWithRequest('ingest.simulate', {
+        body: { pipeline, docs },
+      });
+
       return res.ok({
         body: { docs: response.docs } as SimulateIngestPipelineResponse,
       });
diff --git a/server/utils/helpers.ts b/server/utils/helpers.ts
new file mode 100644
index 00000000..b69ce180
--- /dev/null
+++ b/server/utils/helpers.ts
@@ -0,0 +1,26 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import {
+  ILegacyClusterClient,
+  OpenSearchDashboardsRequest,
+  RequestHandlerContext,
+} from '../../../../src/core/server';
+
+export function getClientBasedOnDataSource(
+  context: RequestHandlerContext,
+  dataSourceEnabled: boolean,
+  request: OpenSearchDashboardsRequest,
+  dataSourceId: string,
+  client: ILegacyClusterClient
+): (endpoint: string, clientParams?: Record<string, any>) => any {
+  if (dataSourceEnabled && dataSourceId && dataSourceId.trim().length != 0) {
+    // client for remote cluster
+    return context.dataSource.opensearch.legacy.getClient(dataSourceId).callAPI;
+  } else {
+    // fall back to default local cluster
+    return client.asScoped(request).callAsCurrentUser;
+  }
+}
diff --git a/test/mocks/mock_core_services.ts b/test/mocks/mock_core_services.ts
index 40b86e08..5b246d63 100644
--- a/test/mocks/mock_core_services.ts
+++ b/test/mocks/mock_core_services.ts
@@ -19,4 +19,9 @@ export const mockCoreServices = {
       },
     };
   },
+  getDataSourceEnabled: () => {
+    return {
+      enabled: false,
+    };
+  },
 };