[MD] Support SigV4 as a new auth type of datasource (#3058)

* [Multiple DataSource] Add support for SigV4 authentication

Signed-off-by: Su <szhongna@amazon.com>

CHANGELOG.md

@@ -60,6 +60,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - [Optimizer] Increase timeout waiting for the exiting of an optimizer worker ([#3193](https://github.com/opensearch-project/OpenSearch-Dashboards/pull/3193))
 - [Data] Update `createAggConfig` so that newly created configs can be added to beginning of `aggConfig` array ([#3160](https://github.com/opensearch-project/OpenSearch-Dashboards/pull/3160))
 - Add disablePrototypePoisoningProtection configuration to prevent JS client from erroring when cluster utilizes JS reserved words ([#2992](https://github.com/opensearch-project/OpenSearch-Dashboards/pull/2992))
+- [Multiple DataSource] Add support for SigV4 authentication ([#3058](https://github.com/opensearch-project/OpenSearch-Dashboards/issues/3058))
 
 ### 🐛 Bug Fixes
 

package.json

@@ -133,7 +133,7 @@
     "@hapi/podium": "^4.1.3",
     "@hapi/vision": "^6.1.0",
     "@hapi/wreck": "^17.1.0",
-    "@opensearch-project/opensearch": "^1.1.0",
+    "@opensearch-project/opensearch": "^2.1.0",
     "@osd/ace": "1.0.0",
     "@osd/analytics": "1.0.0",
     "@osd/apm-config-loader": "1.0.0",
@@ -166,6 +166,7 @@
     "dns-sync": "^0.2.1",
     "elastic-apm-node": "^3.7.0",
     "elasticsearch": "^16.7.0",
+    "http-aws-es": "6.0.0",
     "execa": "^4.0.2",
     "expiry-js": "0.1.7",
     "fast-deep-equal": "^3.1.1",
@@ -334,6 +335,7 @@
     "@types/zen-observable": "^0.8.0",
     "@typescript-eslint/eslint-plugin": "^3.10.0",
     "@typescript-eslint/parser": "^3.10.0",
+    "@types/http-aws-es": "6.0.2",
     "angular-aria": "^1.8.0",
     "angular-mocks": "^1.8.2",
     "angular-recursion": "^1.0.5",

packages/osd-opensearch/package.json

@@ -12,7 +12,7 @@
     "osd:watch": "node scripts/build --watch"
   },
   "dependencies": {
-    "@opensearch-project/opensearch": "^1.1.0",
+    "@opensearch-project/opensearch": "^2.1.0",
     "@osd/dev-utils": "1.0.0",
     "abort-controller": "^3.0.0",
     "chalk": "^4.1.0",

src/dev/jest/config.js

@@ -52,6 +52,8 @@ export default {
   moduleNameMapper: {
     '@elastic/eui$': '<rootDir>/node_modules/@elastic/eui/test-env',
     '@elastic/eui/lib/(.*)?': '<rootDir>/node_modules/@elastic/eui/test-env/$1',
+    '@opensearch-project/opensearch/aws':
+      '<rootDir>/node_modules/@opensearch-project/opensearch/lib/aws',
     '^src/plugins/(.*)': '<rootDir>/src/plugins/$1',
     '^test_utils/(.*)': '<rootDir>/src/test_utils/public/$1',
     '^fixtures/(.*)': '<rootDir>/src/fixtures/$1',

src/plugins/data_source/common/data_sources/types.ts

@@ -11,8 +11,20 @@ export interface DataSourceAttributes extends SavedObjectAttributes {
   endpoint: string;
   auth: {
     type: AuthType;
-    credentials: UsernamePasswordTypedContent | undefined;
+    credentials: UsernamePasswordTypedContent | SigV4Content | undefined;
   };
+  lastUpdatedTime?: string;
+}
+
+/**
+ * Multiple datasource supports authenticating as IAM user, it doesn't support IAM role.
+ * Because IAM role session requires temporary security credentials through assuming role,
+ * which makes no sense to store the credentials.
+ */
+export interface SigV4Content extends SavedObjectAttributes {
+  accessKey: string;
+  secretKey: string;
+  region: string;
 }
 
 export interface UsernamePasswordTypedContent extends SavedObjectAttributes {
@@ -23,4 +35,5 @@ export interface UsernamePasswordTypedContent extends SavedObjectAttributes {
 export enum AuthType {
   NoAuth = 'no_auth',
   UsernamePasswordType = 'username_password',
+  SigV4 = 'sigv4',
 }

src/plugins/data_source/server/client/client_pool.ts

@@ -7,11 +7,12 @@ import { Client } from '@opensearch-project/opensearch';
 import { Client as LegacyClient } from 'elasticsearch';
 import LRUCache from 'lru-cache';
 import { Logger } from 'src/core/server';
+import { AuthType } from '../../common/data_sources';
 import { DataSourcePluginConfigType } from '../../config';
 
 export interface OpenSearchClientPoolSetup {
-  getClientFromPool: (id: string) => Client | LegacyClient | undefined;
-  addClientToPool: (endpoint: string, client: Client | LegacyClient) => void;
+  getClientFromPool: (endpoint: string, authType: AuthType) => Client | LegacyClient | undefined;
+  addClientToPool: (endpoint: string, authType: AuthType, client: Client | LegacyClient) => void;
 }
 
 /**
@@ -21,23 +22,28 @@ export interface OpenSearchClientPoolSetup {
  * It reuse TPC connections for each OpenSearch endpoint.
  */
 export class OpenSearchClientPool {
-  // LRU cache
+  // LRU cache of client
   //   key: data source endpoint
-  //   value: OpenSearch client object | Legacy client object
-  private cache?: LRUCache<string, Client | LegacyClient>;
+  //   value: OpenSearch client | Legacy client
+  private clientCache?: LRUCache<string, Client | LegacyClient>;
+  // LRU cache of aws clients
+  //   key: endpoint + dataSourceId + lastUpdatedTime together to support update case.
+  //   value: OpenSearch client | Legacy client
+  private awsClientCache?: LRUCache<string, Client | LegacyClient>;
   private isClosed = false;
 
   constructor(private logger: Logger) {}
 
   public setup(config: DataSourcePluginConfigType): OpenSearchClientPoolSetup {
     const logger = this.logger;
     const { size } = config.clientPool;
+    const MAX_AGE = 15 * 60 * 1000; // by default, TCP connection times out in 15 minutes
 
-    this.cache = new LRUCache({
+    this.clientCache = new LRUCache({
       max: size,
-      maxAge: 15 * 60 * 1000, // by default, TCP connection times out in 15 minutes
+      maxAge: MAX_AGE,
 
-      async dispose(endpoint, client) {
+      async dispose(key, client) {
         try {
           await client.close();
         } catch (error: any) {
@@ -50,12 +56,34 @@ export class OpenSearchClientPool {
     });
     this.logger.info(`Created data source client pool of size ${size}`);
 
-    const getClientFromPool = (endpoint: string) => {
-      return this.cache!.get(endpoint);
+    // aws client specific pool
+    this.awsClientCache = new LRUCache({
+      max: size,
+      maxAge: MAX_AGE,
+
+      async dispose(key, client) {
+        try {
+          await client.close();
+        } catch (error: any) {
+          logger.warn(
+            `Error closing OpenSearch client when removing from aws client pool: ${error.message}`
+          );
+        }
+      },
+    });
+    this.logger.info(`Created data source aws client pool of size ${size}`);
+
+    const getClientFromPool = (key: string, authType: AuthType) => {
+      const selectedCache = authType === AuthType.SigV4 ? this.awsClientCache : this.clientCache;
+
+      return selectedCache!.get(key);
     };
 
-    const addClientToPool = (endpoint: string, client: Client | LegacyClient) => {
-      this.cache!.set(endpoint, client);
+    const addClientToPool = (key: string, authType: string, client: Client | LegacyClient) => {
+      const selectedCache = authType === AuthType.SigV4 ? this.awsClientCache : this.clientCache;
+      if (!selectedCache?.has(key)) {
+        return selectedCache!.set(key, client);
+      }
     };
 
     return {
@@ -71,7 +99,15 @@ export class OpenSearchClientPool {
     if (this.isClosed) {
       return;
     }
-    await Promise.all(this.cache!.values().map((client) => client.close()));
-    this.isClosed = true;
+
+    try {
+      await Promise.all([
+        ...this.clientCache!.values().map((client) => client.close()),
+        ...this.awsClientCache!.values().map((client) => client.close()),
+      ]);
+      this.isClosed = true;
+    } catch (error) {
+      this.logger.error(`Error closing clients in pool. ${error}`);
+    }
   }
 }

src/plugins/data_source/server/client/configure_client.test.ts

@@ -137,7 +137,7 @@ describe('configureClient', () => {
       configureClient(dataSourceClientParams, clientPoolSetup, config, logger)
     ).rejects.toThrowError();
 
-    expect(ClientMock).toHaveBeenCalledTimes(1);
+    expect(ClientMock).not.toHaveBeenCalled();
     expect(savedObjectsMock.get).toHaveBeenCalledTimes(1);
     expect(decodeAndDecryptSpy).toHaveBeenCalledTimes(1);
   });
@@ -152,7 +152,7 @@ describe('configureClient', () => {
       configureClient(dataSourceClientParams, clientPoolSetup, config, logger)
     ).rejects.toThrowError();
 
-    expect(ClientMock).toHaveBeenCalledTimes(1);
+    expect(ClientMock).not.toHaveBeenCalled();
     expect(savedObjectsMock.get).toHaveBeenCalledTimes(1);
     expect(decodeAndDecryptSpy).toHaveBeenCalledTimes(1);
   });