Implement multi tenancy in Flow Framework

.github/workflows/CI.yml

@@ -98,3 +98,22 @@ jobs:
       - name: Build and Run Tests
         run: |
           ./gradlew integTest -PnumNodes=3
+  integTenantAwareTest:
+    needs: [spotless, javadoc]
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, macos-latest, windows-latest]
+        java: [21]
+    name: Tenant Aware Integ Test JDK${{ matrix.java }}, ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up JDK ${{ matrix.java }}
+        uses: actions/setup-java@v4
+        with:
+          java-version: ${{ matrix.java }}
+          distribution: temurin
+      - name: Build and Run Tests
+        run: |
+          ./gradlew integTest "-Dtests.rest.tenantaware=true"

.github/workflows/test_security.yml

@@ -34,7 +34,6 @@ jobs:
     steps:
       - name: Run start commands
         run: ${{ needs.Get-CI-Image-Tag.outputs.ci-image-start-command }}
-
       - name: Checkout Flow Framework
         uses: actions/checkout@v4
       - name: Setup Java ${{ matrix.java }}

CHANGELOG.md

@@ -18,6 +18,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 
 ## [Unreleased 2.x](https://github.com/opensearch-project/flow-framework/compare/2.18...2.x)
 ### Features
+- Implemented multitenant remote metadata client ([#980](https://github.com/opensearch-project/flow-framework/pull/980))
 - Add synchronous execution option to workflow provisioning ([#990](https://github.com/opensearch-project/flow-framework/pull/990))
 
 ### Enhancements

build.gradle

@@ -37,6 +37,9 @@ buildscript {
         isSameMajorVersion = opensearch_version.split("\\.")[0] == bwcVersionShort.split("\\.")[0]
         swaggerVersion = "2.1.24"
         swaggerCoreVersion = "2.2.28"
+        apacheHttpVersion = "5.3.1"
+        apacheHttpClientVersion = "5.4.1"
+        log4jVersion = "2.24.2"
     }
 
     repositories {
@@ -177,7 +180,7 @@ dependencies {
     implementation "org.dafny:DafnyRuntime:4.9.1"
     implementation "software.amazon.smithy.dafny:conversion:0.1.1"
     implementation 'org.bouncycastle:bcprov-jdk18on:1.80'
-    api "org.apache.httpcomponents.core5:httpcore5:5.3.2"
+    api "org.apache.httpcomponents.core5:httpcore5:${apacheHttpVersion}"
     implementation "jakarta.json.bind:jakarta.json.bind-api:3.0.1"
     implementation "org.glassfish:jakarta.json:2.0.1"
     implementation "org.eclipse:yasson:3.0.4"
@@ -192,6 +195,8 @@ dependencies {
     testImplementation "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}"
     testImplementation "com.fasterxml.jackson.core:jackson-databind:${versions.jackson_databind}"
     testImplementation "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:${versions.jackson_databind}"
+    // Multi-tenant SDK Client
+    implementation "org.opensearch:opensearch-remote-metadata-sdk:${opensearch_build}"
 
     // ZipArchive dependencies used for integration tests
     zipArchive group: 'org.opensearch.plugin', name:'opensearch-ml-plugin', version: "${opensearch_build}"
@@ -203,6 +208,12 @@ dependencies {
         resolutionStrategy {
             force("com.google.guava:guava:33.4.0-jre") // CVE for 31.1, keep to force transitive dependencies
             force("org.apache.httpcomponents.core5:httpcore5:5.3.2") // Dependency Jar Hell
+            // Force to prevent Jar Hell
+            force("org.apache.httpcomponents.core5:httpcore5:${apacheHttpVersion}")
+            force "org.apache.httpcomponents.core5:httpcore5-h2:${apacheHttpVersion}"
+            force("org.apache.httpcomponents.client5:httpclient5:${apacheHttpClientVersion}")
+            force "org.apache.logging.log4j:log4j-api:${log4jVersion}"
+            force "org.apache.logging.log4j:log4j-core:${log4jVersion}"
         }
     }
 }
@@ -262,10 +273,19 @@ integTest {
     systemProperty('user', user)
     systemProperty('password', password)
 
+    // Only tenant aware test if set
+    if (System.getProperty("tests.rest.tenantaware") == "true") {
+        filter {
+            includeTestsMatching "org.opensearch.flowframework.*TenantAwareIT"
+        }
+        systemProperty "plugins.flow_framework.multi_tenancy_enabled", "true"
+    }
+
     // Only rest case can run with remote cluster
-    if (System.getProperty("tests.rest.cluster") != null) {
+    if (System.getProperty("tests.rest.cluster") != null && System.getProperty("tests.rest.tenantaware") == null) {
         filter {
             includeTestsMatching "org.opensearch.flowframework.rest.*IT"
+            excludeTestsMatching "org.opensearch.ml.rest.*TenantAwareIT"
         }
     }
 
@@ -293,6 +313,28 @@ integTest {
 
     // doFirst delays this block until execution time
     doFirst {
+        if (System.getProperty("tests.rest.tenantaware") == "true") {
+            def ymlFile = file("$buildDir/testclusters/integTest-0/config/opensearch.yml")
+            if (ymlFile.exists()) {
+                ymlFile.withWriterAppend {
+                    writer ->
+                    writer.write("\n# Set multitenancy\n")
+                    writer.write("plugins.flow_framework.multi_tenancy_enabled: true\n")
+                }
+                // TODO this properly uses the remote client factory but needs a remote cluster set up
+                // TODO get the endpoint from a system property
+                if (System.getProperty("tests.rest.cluster") != null) {
+                    ymlFile.withWriterAppend { writer ->
+                        writer.write("\n# Use a remote cluster\n")
+                        writer.write("plugins.flow_framework.remote_metadata_type: RemoteOpenSearch\n")
+                        writer.write("plugins.flow_framework.remote_metadata_endpoint: https://127.0.0.1:9200\n")
+                    }
+                }
+            } else {
+                throw new GradleException("opensearch.yml not found at: $ymlFile")
+            }
+        }
+
         // Tell the test JVM if the cluster JVM is running under a debugger so that tests can
         // use longer timeouts for requests.
         def isDebuggingCluster = getDebug() || System.getProperty("test.debug") != null

src/main/java/org/opensearch/flowframework/FlowFrameworkPlugin.java

@@ -65,6 +65,8 @@
 import org.opensearch.plugins.ActionPlugin;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.plugins.SystemIndexPlugin;
+import org.opensearch.remote.metadata.client.SdkClient;
+import org.opensearch.remote.metadata.client.impl.SdkClientFactory;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.rest.RestController;
 import org.opensearch.rest.RestHandler;
@@ -75,22 +77,36 @@
 import org.opensearch.watcher.ResourceWatcherService;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Supplier;
 
 import static org.opensearch.flowframework.common.CommonValue.CONFIG_INDEX;
 import static org.opensearch.flowframework.common.CommonValue.DEPROVISION_WORKFLOW_THREAD_POOL;
 import static org.opensearch.flowframework.common.CommonValue.FLOW_FRAMEWORK_THREAD_POOL_PREFIX;
 import static org.opensearch.flowframework.common.CommonValue.GLOBAL_CONTEXT_INDEX;
 import static org.opensearch.flowframework.common.CommonValue.PROVISION_WORKFLOW_THREAD_POOL;
+import static org.opensearch.flowframework.common.CommonValue.TENANT_ID_FIELD;
 import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_STATE_INDEX;
 import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_THREAD_POOL;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.FILTER_BY_BACKEND_ROLES;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.FLOW_FRAMEWORK_ENABLED;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.FLOW_FRAMEWORK_MULTI_TENANCY_ENABLED;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOWS;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOW_STEPS;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.REMOTE_METADATA_ENDPOINT;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.REMOTE_METADATA_REGION;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.REMOTE_METADATA_SERVICE_NAME;
+import static org.opensearch.flowframework.common.FlowFrameworkSettings.REMOTE_METADATA_TYPE;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.TASK_REQUEST_RETRY_DURATION;
 import static org.opensearch.flowframework.common.FlowFrameworkSettings.WORKFLOW_REQUEST_TIMEOUT;
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_ENDPOINT_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_REGION_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_SERVICE_NAME_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_TYPE_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.TENANT_AWARE_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.TENANT_ID_FIELD_KEY;
 
 /**
  * An OpenSearch plugin that enables builders to innovate AI apps on OpenSearch.
@@ -121,9 +137,28 @@
         Settings settings = environment.settings();
         flowFrameworkSettings = new FlowFrameworkSettings(clusterService, settings);
         MachineLearningNodeClient mlClient = new MachineLearningNodeClient(client);
-        EncryptorUtils encryptorUtils = new EncryptorUtils(clusterService, client, xContentRegistry);
+        SdkClient sdkClient = SdkClientFactory.createSdkClient(
+            client,
+            xContentRegistry,
+            // Here we assume remote metadata client is only used with tenant awareness.
+            // This may change in the future allowing more options for this map
+            FLOW_FRAMEWORK_MULTI_TENANCY_ENABLED.get(settings)
+                ? Map.ofEntries(
+                    Map.entry(REMOTE_METADATA_TYPE_KEY, REMOTE_METADATA_TYPE.get(settings)),
+                    Map.entry(REMOTE_METADATA_ENDPOINT_KEY, REMOTE_METADATA_ENDPOINT.get(settings)),
+                    Map.entry(REMOTE_METADATA_REGION_KEY, REMOTE_METADATA_REGION.get(settings)),
+                    Map.entry(REMOTE_METADATA_SERVICE_NAME_KEY, REMOTE_METADATA_SERVICE_NAME.get(settings)),
+                    Map.entry(TENANT_AWARE_KEY, "true"),
+                    Map.entry(TENANT_ID_FIELD_KEY, TENANT_ID_FIELD)
+                )
+                : Collections.emptyMap(),
+            // TODO: Find a better thread pool or make one
+            client.threadPool().executor(ThreadPool.Names.GENERIC)
+        );
+        EncryptorUtils encryptorUtils = new EncryptorUtils(clusterService, client, sdkClient, xContentRegistry);
         FlowFrameworkIndicesHandler flowFrameworkIndicesHandler = new FlowFrameworkIndicesHandler(
             client,
+            sdkClient,
             clusterService,
             encryptorUtils,
             xContentRegistry
@@ -137,15 +172,22 @@
         );
         WorkflowProcessSorter workflowProcessSorter = new WorkflowProcessSorter(workflowStepFactory, threadPool, flowFrameworkSettings);
 
-        SearchHandler searchHandler = new SearchHandler(settings, clusterService, client, FlowFrameworkSettings.FILTER_BY_BACKEND_ROLES);
+        SearchHandler searchHandler = new SearchHandler(
+            settings,
+            clusterService,
+            client,
+            sdkClient,
+            FlowFrameworkSettings.FILTER_BY_BACKEND_ROLES
+        );
 
         return List.of(
             workflowStepFactory,
             workflowProcessSorter,
             encryptorUtils,
             flowFrameworkIndicesHandler,
             searchHandler,
-            flowFrameworkSettings
+            flowFrameworkSettings,
+            sdkClient
         );
     }
 
@@ -196,7 +238,12 @@
             MAX_WORKFLOW_STEPS,
             WORKFLOW_REQUEST_TIMEOUT,
             TASK_REQUEST_RETRY_DURATION,
-            FILTER_BY_BACKEND_ROLES
+            FILTER_BY_BACKEND_ROLES,
+            FLOW_FRAMEWORK_MULTI_TENANCY_ENABLED,
+            REMOTE_METADATA_TYPE,
+            REMOTE_METADATA_ENDPOINT,
+            REMOTE_METADATA_REGION,
+            REMOTE_METADATA_SERVICE_NAME
         );
     }
 
@@ -206,21 +253,21 @@
             new ScalingExecutorBuilder(
                 WORKFLOW_THREAD_POOL,
                 1,
-                Math.max(2, OpenSearchExecutors.allocatedProcessors(settings) - 1),
+                Math.max(4, OpenSearchExecutors.allocatedProcessors(settings) - 1),
                 TimeValue.timeValueMinutes(1),
                 FLOW_FRAMEWORK_THREAD_POOL_PREFIX + WORKFLOW_THREAD_POOL
             ),
             new ScalingExecutorBuilder(
                 PROVISION_WORKFLOW_THREAD_POOL,
                 1,
-                Math.max(4, OpenSearchExecutors.allocatedProcessors(settings) - 1),
+                Math.max(8, OpenSearchExecutors.allocatedProcessors(settings) - 1),
                 TimeValue.timeValueMinutes(5),
                 FLOW_FRAMEWORK_THREAD_POOL_PREFIX + PROVISION_WORKFLOW_THREAD_POOL
             ),
             new ScalingExecutorBuilder(
                 DEPROVISION_WORKFLOW_THREAD_POOL,
                 1,
-                Math.max(2, OpenSearchExecutors.allocatedProcessors(settings) - 1),
+                Math.max(4, OpenSearchExecutors.allocatedProcessors(settings) - 1),
                 TimeValue.timeValueMinutes(1),
                 FLOW_FRAMEWORK_THREAD_POOL_PREFIX + DEPROVISION_WORKFLOW_THREAD_POOL
             )

src/main/java/org/opensearch/flowframework/common/CommonValue.java

@@ -8,6 +8,8 @@
  */
 package org.opensearch.flowframework.common;
 
+import org.opensearch.Version;
+
 /**
  * Representation of common values that are used across project
  */
@@ -82,6 +84,10 @@ private CommonValue() {}
     public static final String USE_CASE = "use_case";
     /** The param name for reprovisioning, used by the create workflow API */
     public static final String REPROVISION_WORKFLOW = "reprovision";
+    /** The Rest header containing the tenant id */
+    public static final String TENANT_ID_HEADER = "x-tenant-id";
+    /** The field name containing the tenant id */
+    public static final String TENANT_ID_FIELD = "tenant_id";
 
     /*
      * Constants associated with plugin configuration
@@ -244,4 +250,9 @@ private CommonValue() {}
     public static final String ML_COMMONS_API_SPEC_YAML_URI =
         "https://raw.githubusercontent.com/opensearch-project/opensearch-api-specification/refs/heads/main/spec/namespaces/ml.yaml";
 
+    /*
+     * Constants associated with non-BWC features
+     */
+    /** Version 2.19.0 */
+    public static final Version VERSION_2_19_0 = Version.fromString("2.19.0");
 }

src/main/java/org/opensearch/flowframework/common/FlowFrameworkSettings.java

@@ -10,9 +10,15 @@
 
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Setting.Property;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_ENDPOINT_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_REGION_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_SERVICE_NAME_KEY;
+import static org.opensearch.remote.metadata.common.CommonValue.REMOTE_METADATA_TYPE_KEY;
+
 /** The common settings of flow framework  */
 public class FlowFrameworkSettings {
 
@@ -25,6 +31,8 @@ public class FlowFrameworkSettings {
     protected volatile Integer maxWorkflows;
     /** Timeout for internal requests*/
     protected volatile TimeValue requestTimeout;
+    /** Whether multitenancy is enabled */
+    private final Boolean isMultiTenancyEnabled;
 
     /** The upper limit of max workflows that can be created  */
     public static final int MAX_WORKFLOWS_LIMIT = 10000;
@@ -83,6 +91,53 @@ public class FlowFrameworkSettings {
         Setting.Property.Dynamic
     );
 
+    // Whether multi-tenancy is enabled in Flow Framework.
+    // This is a static setting which must be set before starting OpenSearch by (in priority order):
+    // 1. As a command-line argument using the -E flag (overrides other options):
+    // ./bin/opensearch -Eplugins.flow_framework.multi_tenancy_enabled=true
+    // 2. As a system property using OPENSEARCH_JAVA_OPTS (overrides opensearch.yml):
+    // export OPENSEARCH_JAVA_OPTS="-Dplugins.flow_framework.multi_tenancy_enabled=true"
+    // ./bin/opensearch
+    // Or inline when starting OpenSearch:
+    // OPENSEARCH_JAVA_OPTS="-Dplugins.flow_framework.multi_tenancy_enabled=true" ./bin/opensearch
+    // 3. In the opensearch.yml configuration file:
+    // plugins.flow_framework.multi_tenancy_enabled: true
+    // After setting it, a full cluster restart is required for the changes to take effect.
+    /** This setting determines if multitenancy is enabled */
+    public static final Setting<Boolean> FLOW_FRAMEWORK_MULTI_TENANCY_ENABLED = Setting.boolSetting(
+        "plugins.flow_framework.multi_tenancy_enabled",
+        false,
+        Setting.Property.NodeScope
+    );
+
+    /** This setting sets the remote metadata type */
+    public static final Setting<String> REMOTE_METADATA_TYPE = Setting.simpleString(
+        "plugins.flow_framework." + REMOTE_METADATA_TYPE_KEY,
+        Property.NodeScope,
+        Property.Final
+    );
+
+    /** This setting sets the remote metadata endpoint */
+    public static final Setting<String> REMOTE_METADATA_ENDPOINT = Setting.simpleString(
+        "plugins.flow_framework." + REMOTE_METADATA_ENDPOINT_KEY,
+        Property.NodeScope,
+        Property.Final
+    );
+
+    /** This setting sets the remote metadata region */
+    public static final Setting<String> REMOTE_METADATA_REGION = Setting.simpleString(
+        "plugins.flow_framework." + REMOTE_METADATA_REGION_KEY,
+        Property.NodeScope,
+        Property.Final
+    );
+
+    /** This setting sets the remote metadata service name */
+    public static final Setting<String> REMOTE_METADATA_SERVICE_NAME = Setting.simpleString(
+        "plugins.flow_framework." + REMOTE_METADATA_SERVICE_NAME_KEY,
+        Property.NodeScope,
+        Property.Final
+    );
+
     /**
      * Instantiate this class.
      *
@@ -97,6 +152,7 @@ public FlowFrameworkSettings(ClusterService clusterService, Settings settings) {
         this.maxWorkflowSteps = MAX_WORKFLOW_STEPS.get(settings);
         this.maxWorkflows = MAX_WORKFLOWS.get(settings);
         this.requestTimeout = WORKFLOW_REQUEST_TIMEOUT.get(settings);
+        this.isMultiTenancyEnabled = FLOW_FRAMEWORK_MULTI_TENANCY_ENABLED.get(settings);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(FLOW_FRAMEWORK_ENABLED, it -> isFlowFrameworkEnabled = it);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(TASK_REQUEST_RETRY_DURATION, it -> retryDuration = it);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(MAX_WORKFLOW_STEPS, it -> maxWorkflowSteps = it);
@@ -143,4 +199,12 @@ public Integer getMaxWorkflows() {
     public TimeValue getRequestTimeout() {
         return requestTimeout;
     }
+
+    /**
+     * Whether multitenancy is enabled.
+     * @return whether Flow Framework multitenancy is enabled.
+     */
+    public boolean isMultiTenancyEnabled() {
+        return isMultiTenancyEnabled;
+    }
 }