From 927d5a62bbb7621b0925633379c9a48069ed4db2 Mon Sep 17 00:00:00 2001 From: Anton Shepilov Date: Thu, 4 Jan 2024 12:59:05 +0300 Subject: [PATCH] =?UTF-8?q?=F0=9F=8C=9F=20Support=20of=20OpenSearch=20as?= =?UTF-8?q?=20a=20search=20database=20(#304)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Added common search adapter for ES and OpenSearch - Open Search integration tests add to the build workflow - Stop docker-compose environment before running the next stop --- .github/workflows/backend.yml | 6 + ...ndex.ts => elastic-open-search-adapter.ts} | 50 +-- .../search/adapters/opensearch/opensearch.ts | 328 ------------------ .../src/core/platform/services/search/api.ts | 19 +- .../core/platform/services/search/index.ts | 25 +- .../storage/connectors/local/service.ts | 2 +- .../docker-compose.dev.tests.opensearch.yml | 23 +- tdrive/docker/tdrive-node/Dockerfile | 3 +- 8 files changed, 79 insertions(+), 377 deletions(-) rename tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/{index.ts => elastic-open-search-adapter.ts} (83%) delete mode 100644 tdrive/backend/node/src/core/platform/services/search/adapters/opensearch/opensearch.ts diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml index 8d57b60f..7d4783de 100644 --- a/.github/workflows/backend.yml +++ b/.github/workflows/backend.yml @@ -21,6 +21,12 @@ jobs: - uses: actions/checkout@v2 - name: e2e-mongo-test run: cd tdrive && docker-compose -f docker-compose.tests.yml run -e NODE_OPTIONS=--unhandled-rejections=warn -e SEARCH_DRIVER=mongodb -e DB_DRIVER=mongodb -e PUBSUB_TYPE=local node npm run test:all + - name: e2e-mongo-test-clean + run: cd tdrive && docker-compose -f docker-compose.tests.yml stop + - name: e2e-opensearch-test + run: cd tdrive && docker-compose -f docker-compose.dev.tests.opensearch.yml run -e NODE_OPTIONS=--unhandled-rejections=warn -e SEARCH_DRIVER=opensearch -e DB_DRIVER=mongodb -e PUBSUB_TYPE=local node npm run test:all + - name: e2e-opensearch-test-clean + run: cd tdrive && docker-compose -f docker-compose.dev.tests.opensearch.yml stop - name: e2e-cassandra-test run: cd tdrive && docker-compose -f docker-compose.tests.yml up -d scylladb elasticsearch rabbitmq && sleep 60 && docker-compose -f docker-compose.tests.yml run -e NODE_OPTIONS=--unhandled-rejections=warn -e SEARCH_DRIVER=elasticsearch -e DB_DRIVER=cassandra node npm run test:all - name: coverage diff --git a/tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/index.ts b/tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/elastic-open-search-adapter.ts similarity index 83% rename from tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/index.ts rename to tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/elastic-open-search-adapter.ts index d6458e73..2102bc77 100644 --- a/tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/index.ts +++ b/tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/elastic-open-search-adapter.ts @@ -1,23 +1,23 @@ -import { Client } from "@elastic/elasticsearch"; import { logger } from "../../../../framework"; import _ from "lodash"; import { ColumnDefinition, EntityDefinition, EntityTarget, + ESSearchConfiguration, FindFilter, FindOptions, IndexedEntity, SearchAdapterInterface, - SearchConfiguration, } from "../../api"; import { SearchAdapter } from "../abstract"; import { DatabaseServiceAPI } from "../../../database/api"; import { getEntityDefinition, unwrapPrimarykey } from "../../api"; import { ListResult, Paginable, Pagination } from "../../../../framework/api/crud-service"; import { asciiFold, parsePrimaryKey, stringifyPrimaryKey } from "../utils"; -import { ApiResponse } from "@elastic/elasticsearch/lib/Transport"; import { buildSearchQuery } from "./search"; +import { Client as OpenClient } from "@opensearch-project/opensearch"; +import { Client as ESClient } from "@elastic/elasticsearch"; type Operation = { index?: { _index: string; _id: string }; @@ -25,15 +25,16 @@ type Operation = { [key: string]: any; }; -export default class ElasticSearch extends SearchAdapter implements SearchAdapterInterface { - private client: Client; +export default class ESAndOpenSearch extends SearchAdapter implements SearchAdapterInterface { private bulkReaders = 0; private buffer: Operation[] = []; - private name = "ElasticSearch"; + private client: OpenClient | ESClient; constructor( readonly database: DatabaseServiceAPI, - readonly configuration: SearchConfiguration["elasticsearch"], + readonly configuration: ESSearchConfiguration, + readonly newClient: (arg: any) => OpenClient | ESClient, + readonly name, ) { super(); } @@ -55,10 +56,10 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte }; } - this.client = new Client(clientOptions); + this.client = this.newClient(clientOptions); } catch (e) { logger.error( - `Unable to connect to ElasticSearch for options: ${JSON.stringify({ + `Unable to connect to ${this.name} for options: ${JSON.stringify({ node: this.configuration.endpoint, auth: { useAuth: this.configuration.useAuth, @@ -86,6 +87,8 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte const mapping = entity.options?.search?.esMapping; try { + // eslint-disable-next-line @typescript-eslint/ban-ts-comment + // @ts-ignore await this.client.indices.get({ index: name, }); @@ -109,7 +112,8 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte mappings: { ...mapping, _source: { enabled: false } }, }, }; - + // eslint-disable-next-line @typescript-eslint/ban-ts-comment + // @ts-ignore const rep = await this.client.indices.create(indice, { ignore: [400] }); if (rep.statusCode !== 200) { @@ -165,7 +169,7 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte ...body, }; - logger.info(`Add operation upsert to elasticsearch for doc ${record.id}`); + logger.info(`Add operation upsert to ${this.name} for doc ${record.id}`); this.buffer.push(record); } @@ -192,7 +196,7 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte }, }; - logger.info(`Add operation remove from elasticsearch for doc ${record.id}`); + logger.info(`Add operation remove from ${this.name} for doc ${record.id}`); this.buffer.push(record); } @@ -205,7 +209,7 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte return; } - logger.info("Start new Elasticsearch bulk reader."); + logger.info(`Start new ${this.name} bulk reader.`); this.bulkReaders += 1; let buffer; @@ -224,7 +228,7 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte onDocument: (doc: Operation) => { if (doc.delete) { logger.info( - `Operation ${"DELETE"} pushed to elasticsearch index ${doc.delete._index} (doc.id: ${ + `Operation ${"DELETE"} pushed to ${this.name} index ${doc.delete._index} (doc.id: ${ doc.delete._id })`, ); @@ -234,7 +238,7 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte } if (doc.index) { logger.info( - `Operation ${"INDEX"} pushed to elasticsearch index ${doc.index._index} (doc.id: ${ + `Operation ${"INDEX"} pushed to ${this.name} index ${doc.index._index} (doc.id: ${ doc.index._id })`, ); @@ -248,11 +252,9 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte onDrop: res => { const doc = res.document; logger.error( - `Operation ${ - doc.action - } was droped while pushing to elasticsearch index ${JSON.stringify( - doc.index, - )} (doc.id: ${doc.id})`, + `Operation ${doc.action} was droped while pushing to ${ + this.name + } index ${JSON.stringify(doc.index)} (doc.id: ${doc.id})`, ); logger.error(res.error); }, @@ -262,7 +264,7 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte logger.error(err); } - logger.info("Elasticsearch bulk flushed."); + logger.info(`${this.name} bulk flushed.`); this.bulkReaders += -1; this.startBulkReader(); @@ -284,9 +286,11 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte scroll: "1m", }; - let esResponse: ApiResponse; + let esResponse: any; if (options.pagination.page_token) { + // eslint-disable-next-line @typescript-eslint/ban-ts-comment + // @ts-ignore esResponse = await this.client.scroll( { scroll_id: options.pagination.page_token, @@ -294,6 +298,8 @@ export default class ElasticSearch extends SearchAdapter implements SearchAdapte esOptions, ); } else { + // eslint-disable-next-line @typescript-eslint/ban-ts-comment + // @ts-ignore esResponse = await this.client.search(esParamsWithScroll, esOptions); } diff --git a/tdrive/backend/node/src/core/platform/services/search/adapters/opensearch/opensearch.ts b/tdrive/backend/node/src/core/platform/services/search/adapters/opensearch/opensearch.ts deleted file mode 100644 index 2237b2a4..00000000 --- a/tdrive/backend/node/src/core/platform/services/search/adapters/opensearch/opensearch.ts +++ /dev/null @@ -1,328 +0,0 @@ -import { logger } from "../../../../framework"; -import _ from "lodash"; -import { - ColumnDefinition, - EntityDefinition, - EntityTarget, - FindFilter, - FindOptions, - IndexedEntity, - SearchAdapterInterface, - SearchConfiguration, -} from "../../api"; -import { SearchAdapter } from "../abstract"; -import { DatabaseServiceAPI } from "../../../database/api"; -import { getEntityDefinition, unwrapPrimarykey } from "../../api"; -import { ListResult, Paginable, Pagination } from "../../../../framework/api/crud-service"; -import { asciiFold, parsePrimaryKey, stringifyPrimaryKey } from "../utils"; -import { buildSearchQuery } from "../elasticsearch/search"; -import { Client } from "@opensearch-project/opensearch"; - -type Operation = { - index?: { _index: string; _id: string }; - delete?: { _index: string; _id: string }; - [key: string]: any; -}; - -export default class OpenSearch extends SearchAdapter implements SearchAdapterInterface { - private client: Client; - private bulkReaders = 0; - private buffer: Operation[] = []; - private name = "ElasticSearch"; - - constructor( - readonly database: DatabaseServiceAPI, - readonly configuration: SearchConfiguration["elasticsearch"], - ) { - super(); - } - - public async connect() { - try { - const clientOptions: any = { - node: this.configuration.endpoint, - ssl: { - rejectUnauthorized: false, - }, - }; - - if (this.configuration.useAuth) { - logger.info("Using auth for ES client"); - clientOptions.auth = { - username: this.configuration.username, - password: this.configuration.password, - }; - } - - this.client = new Client(clientOptions); - } catch (e) { - logger.error( - `Unable to connect to ElasticSearch for options: ${JSON.stringify({ - node: this.configuration.endpoint, - auth: { - useAuth: this.configuration.useAuth, - username: this.configuration.username, - password: this.configuration.password, - }, - ssl: { - rejectUnauthorized: false, - }, - })} at: ${this.configuration.endpoint}`, - ); - } - this.startBulkReader(); - } - - private async createIndex( - entity: EntityDefinition, - _columns: { [name: string]: ColumnDefinition }, - ) { - if (!entity.options?.search) { - return; - } - - const name = entity.options?.search?.index || entity.name; - const mapping = entity.options?.search?.esMapping; - - try { - await this.client.indices.get({ - index: name, - }); - logger.info(`Index "${name}" already created`); - } catch (e) { - logger.info(`Create index ${name} with mapping %o`, mapping); - - const indice = { - index: name, - body: { - settings: { - analysis: { - analyzer: { - folding: { - tokenizer: "standard", - filter: ["lowercase", "asciifolding"], - }, - }, - }, - }, - mappings: { ...mapping, _source: { enabled: false } }, - }, - }; - - const rep = await this.client.indices.create(indice, { ignore: [400] }); - - if (rep.statusCode !== 200) { - logger.error(`${this.name} - ${JSON.stringify(rep.body)}`); - } - } - } - - public async upsert(entities: any[]) { - for (const entity of entities) { - const { entityDefinition, columnsDefinition } = getEntityDefinition(entity); - const pkColumns = unwrapPrimarykey(entityDefinition); - - await this.ensureIndex(entityDefinition, columnsDefinition, this.createIndex.bind(this)); - - if (!entityDefinition.options?.search) { - return; - } - - if ( - entityDefinition.options.search.shouldUpdate && - !entityDefinition.options.search.shouldUpdate(entity) - ) { - return; - } - - if (!entityDefinition.options?.search?.source) { - logger.info(`Unable to do operation upsert to elasticsearch for doc ${entity}`); - return; - } - - const body = { - ..._.pick(entity, ...pkColumns), - ...entityDefinition.options.search.source(entity), - }; - - Object.keys(entityDefinition.options?.search.esMapping?.properties || []).forEach( - (key: string) => { - const mapping: any = entityDefinition.options?.search?.esMapping?.properties[key]; - if (mapping.type === "text") { - body[key] = asciiFold(body[key]).toLocaleLowerCase(); - } - }, - ); - - const index = entityDefinition.options?.search?.index || entityDefinition.name; - - const record: Operation = { - index: { - _index: index, - _id: stringifyPrimaryKey(entity), - }, - ...body, - }; - - logger.info(`Add operation upsert to elasticsearch for doc ${record.id}`); - - this.buffer.push(record); - } - - this.startBulkReader(); - } - - public async remove(entities: any[]) { - for (const entity of entities) { - const { entityDefinition, columnsDefinition } = getEntityDefinition(entity); - - await this.ensureIndex(entityDefinition, columnsDefinition, this.createIndex.bind(this)); - - if (!entityDefinition.options?.search) { - return; - } - - const index = entityDefinition.options?.search?.index || entityDefinition.name; - - const record: Operation = { - delete: { - _index: index, - _id: stringifyPrimaryKey(entity), - }, - }; - - logger.info(`Add operation remove from elasticsearch for doc ${record.id}`); - - this.buffer.push(record); - } - - this.startBulkReader(); - } - - private async startBulkReader() { - if (this.bulkReaders > 0) { - return; - } - - logger.info("Start new Elasticsearch bulk reader."); - this.bulkReaders += 1; - - let buffer; - do { - await new Promise(r => - setTimeout(r, parseInt(`${this.configuration.flushInterval}`) || 3000), - ); - buffer = this.buffer; - } while (buffer.length === 0); - this.buffer = []; - - try { - await this.client.helpers.bulk({ - flushInterval: 1, - datasource: buffer, - onDocument: (doc: Operation) => { - if (doc.delete) { - logger.info( - `Operation ${"DELETE"} pushed to elasticsearch index ${doc.delete._index} (doc.id: ${ - doc.delete._id - })`, - ); - return { - delete: doc.delete, - }; - } - if (doc.index) { - logger.info( - `Operation ${"INDEX"} pushed to elasticsearch index ${doc.index._index} (doc.id: ${ - doc.index._id - })`, - ); - return { - index: doc.index, - ...doc.index, - }; - } - return null; - }, - onDrop: res => { - const doc = res.document; - logger.error( - `Operation ${ - doc.action - } was droped while pushing to elasticsearch index ${JSON.stringify( - doc.index, - )} (doc.id: ${doc.id})`, - ); - logger.error(res.error); - }, - }); - } catch (err) { - logger.error(`${this.name} - An error occured with the bulk reader`); - logger.error(err); - } - - logger.info("Elasticsearch bulk flushed."); - this.bulkReaders += -1; - - this.startBulkReader(); - } - - public async search( - _table: string, - entityType: EntityTarget, - filters: FindFilter, - options: FindOptions = {}, - ) { - const instance = new (entityType as any)(); - const { entityDefinition } = getEntityDefinition(instance); - - const { esParams, esOptions } = buildSearchQuery(entityType, filters, options); - const esParamsWithScroll = { - ...esParams, - size: parseInt(options.pagination.limitStr || "100"), - scroll: "1m", - }; - - let esResponse: any; - - if (options.pagination.page_token) { - esResponse = await this.client.scroll( - { - scroll_id: options.pagination.page_token, - }, - esOptions, - ); - } else { - esResponse = await this.client.search(esParamsWithScroll, esOptions); - } - - if (esResponse.statusCode !== 200) { - logger.error(`${this.name} - ${JSON.stringify(esResponse.body)}`); - } - - const nextToken = esResponse.body?._scroll_id || ""; - const hits = esResponse.body?.hits?.hits || []; - - logger.debug(`${this.name} got response: ${JSON.stringify(esResponse)}`); - - const entities: IndexedEntity[] = []; - for await (const hit of hits) { - try { - entities.push({ - primaryKey: parsePrimaryKey(entityDefinition, hit._id), - score: hit._score, - }); - } catch (err) { - logger.error( - `${this.name} failed to get entity from search result: ${JSON.stringify( - hit._id, - )}, ${JSON.stringify(err)}`, - ); - } - } - - const nextPage: Paginable = new Pagination(nextToken, options.pagination.limitStr || "100"); - - return new ListResult(entityDefinition.type, entities, nextPage); - } -} diff --git a/tdrive/backend/node/src/core/platform/services/search/api.ts b/tdrive/backend/node/src/core/platform/services/search/api.ts index 164a3e3c..927d4e8b 100644 --- a/tdrive/backend/node/src/core/platform/services/search/api.ts +++ b/tdrive/backend/node/src/core/platform/services/search/api.ts @@ -65,12 +65,15 @@ export interface SearchServiceAPI extends TdriveServiceProvider { } export type SearchConfiguration = { - type?: false | "elasticsearch" | "mongodb"; - elasticsearch?: { - useAuth: boolean; - username?: string; - password?: string; - endpoint: string; - flushInterval: number; //In milliseconds - }; + type?: false | "elasticsearch" | "mongodb" | "opensearch"; + elasticsearch?: ESSearchConfiguration; + opensearch?: ESSearchConfiguration; +}; + +export type ESSearchConfiguration = { + useAuth: boolean; + username?: string; + password?: string; + endpoint: string; + flushInterval: number; //In milliseconds }; diff --git a/tdrive/backend/node/src/core/platform/services/search/index.ts b/tdrive/backend/node/src/core/platform/services/search/index.ts index f9a90d3e..6bd246b7 100644 --- a/tdrive/backend/node/src/core/platform/services/search/index.ts +++ b/tdrive/backend/node/src/core/platform/services/search/index.ts @@ -3,15 +3,18 @@ import { DatabaseEntitiesRemovedEvent, DatabaseEntitiesSavedEvent, EntityTarget, + ESSearchConfiguration, SearchAdapterInterface, SearchConfiguration, SearchServiceAPI, } from "./api"; -import ElasticsearchService from "./adapters/elasticsearch"; +import ESAndOpenSearch from "./adapters/elasticsearch/elastic-open-search-adapter"; import MongosearchService from "./adapters/mongosearch"; import { localEventBus } from "../../framework/event-bus"; import { DatabaseServiceAPI } from "../database/api"; import SearchRepository from "./repository"; +import { Client as OpenClient } from "@opensearch-project/opensearch"; +import { Client as ESClient } from "@elastic/elasticsearch"; @ServiceName("search") @Consumes(["database"]) @@ -29,9 +32,19 @@ export default class Search extends TdriveService { if (type === "elasticsearch") { logger.info("Loaded Elasticsearch adapter for search."); - this.service = new ElasticsearchService( + this.service = new ESAndOpenSearch( this.database, - this.configuration.get("elasticsearch") as SearchConfiguration["elasticsearch"], + this.configuration.get("elasticsearch") as ESSearchConfiguration, + config => new ESClient(config), + "ElasticSearch", + ); + } else if (type === "opensearch") { + logger.info("Loaded OpenSearch adapter for search."); + this.service = new ESAndOpenSearch( + this.database, + this.configuration.get("opensearch") as ESSearchConfiguration, + config => new OpenClient(config), + "OpenSearch", ); } else if (type === "mongodb") { logger.info("Loaded Mongo adapter for search."); @@ -41,7 +54,7 @@ export default class Search extends TdriveService { this.service = null; } - if (this.service) this.service.connect(); + if (this.service) await this.service.connect(); return this; } @@ -60,11 +73,11 @@ export default class Search extends TdriveService { return this; } - public async upsert(entities: any[]) { + public async upsert(entities: never[]) { return this.service.upsert(entities); } - public async remove(entities: any[]) { + public async remove(entities: never[]) { return this.service.remove(entities); } diff --git a/tdrive/backend/node/src/core/platform/services/storage/connectors/local/service.ts b/tdrive/backend/node/src/core/platform/services/storage/connectors/local/service.ts index 249ed14a..85b03ecd 100644 --- a/tdrive/backend/node/src/core/platform/services/storage/connectors/local/service.ts +++ b/tdrive/backend/node/src/core/platform/services/storage/connectors/local/service.ts @@ -45,7 +45,7 @@ export default class LocalConnectorService implements StorageConnectorAPI { async read(path: string): Promise { const fullPath = this.getFullPath(path); if (!fs.existsSync(fullPath)) { - throw new Error("File doesn't exists"); + throw new Error(`File doesn't exists ${fullPath}`); } return createReadStream(fullPath); } diff --git a/tdrive/docker-compose.dev.tests.opensearch.yml b/tdrive/docker-compose.dev.tests.opensearch.yml index c08e4496..e349b7c4 100644 --- a/tdrive/docker-compose.dev.tests.opensearch.yml +++ b/tdrive/docker-compose.dev.tests.opensearch.yml @@ -25,8 +25,8 @@ services: # - opensearch-data1:/usr/share/opensearch/data # Creates volume called opensearch-data1 and mounts it to the container ports: - 9200:9200 # REST API - - 9600:9600 # Performance Analyzer - # networks: +# - 9600:9600 # Performance Analyzer +# networks: # - opensearch-net # All of the containers will join the same Docker bridge network mongo: @@ -39,22 +39,25 @@ services: node: image: tdrive/tdrive-node:test - container_name: node-test build: context: . dockerfile: docker/tdrive-node/Dockerfile - target: test + target: development + volumes: + - ./coverage/:/usr/src/app/coverage/ environment: + - LOG_LEVEL=error - NODE_ENV=test - DB_DRIVER - #- PUBSUB_URLS=amqp://guest:guest@rabbitmq:5672 - - SEARCH_ES_ENDPOINT=http://elasticsearch:9200 - - SEARCH_ES_FLUSHINTERVAL=1 - volumes: - - ./backend/node:/usr/src/app - - ./docker-data/documents/:/storage/ + - PUBSUB_URLS=amqp://guest:guest@rabbitmq:5672 + - SEARCH_OS_FLUSHINTERVAL=1 + - SEARCH_OS_ENDPOINT=https://opensearch-node1:9200 + - SEARCH_OS_PASSWORD=admin + - SEARCH_OS_USE_AUTH=true + - SEARCH_OS_USERNAME=admin depends_on: - mongo - opensearch-node1 links: - mongo + - opensearch-node1 diff --git a/tdrive/docker/tdrive-node/Dockerfile b/tdrive/docker/tdrive-node/Dockerfile index 45202e04..5b571a96 100755 --- a/tdrive/docker/tdrive-node/Dockerfile +++ b/tdrive/docker/tdrive-node/Dockerfile @@ -11,9 +11,7 @@ RUN apt-get update && \ # RUN apt-get update && apt-get install -y libc6 # RUN ln -s /lib/libc.musl-x86_64.so.1 /lib/ld-linux-x86-64.so.2 - ### Install TDrive - WORKDIR /usr/src/app COPY backend/node/package*.json ./ @@ -30,6 +28,7 @@ COPY backend/node/ . #Install dev dependancies for build RUN export NODE_ENV=development RUN npm install + #Build in production mode RUN export NODE_ENV=production RUN npm run build