eduvia/workavia-drive
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

🌟 Support of OpenSearch as a search database (#304)

Browse Source 
- 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
This commit is contained in:
Anton Shepilov
2024-01-04 12:59:05 +03:00
committed by GitHub
parent 6f7f714fe8
commit 927d5a62bb
8 changed files with 79 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/backend.yml
+6
View File
@@ -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
tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/index.ts → tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/elastic-open-search-adapter.ts
+28 -22
View File
@@ -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);
}
tdrive/backend/node/src/core/platform/services/search/adapters/opensearch/opensearch.ts
-328
View File
@@ -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<EntityType>(
_table: string,
entityType: EntityTarget<EntityType>,
filters: FindFilter,
options: FindOptions = {},
) {
const instance = new (entityType as any)();
const { entityDefinition } = getEntityDefinition(instance);
const { esParams, esOptions } = buildSearchQuery<EntityType>(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);
}
}
tdrive/backend/node/src/core/platform/services/search/api.ts
+11 -8
View File
@@ -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
};
tdrive/backend/node/src/core/platform/services/search/index.ts
+19 -6
View File
@@ -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<SearchServiceAPI> {
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<SearchServiceAPI> {
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<SearchServiceAPI> {
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);
}
tdrive/backend/node/src/core/platform/services/storage/connectors/local/service.ts
+1 -1
View File
@@ -45,7 +45,7 @@ export default class LocalConnectorService implements StorageConnectorAPI {
async read(path: string): Promise<Readable> {
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);
}
tdrive/docker-compose.dev.tests.opensearch.yml
+13 -10
View File
@@ -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
tdrive/docker/tdrive-node/Dockerfile
+1 -2
View File
@@ -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