From 474115e5d4262b4663a2f9dadb4357debb68a1fa Mon Sep 17 00:00:00 2001 From: Andrey Sobolev Date: Sun, 4 Aug 2024 22:06:19 +0700 Subject: [PATCH] UBERF-7796: Rework index creation logic Signed-off-by: Andrey Sobolev --- .vscode/launch.json | 1 + common/scripts/version.txt | 2 +- models/core/src/core.ts | 4 +- models/core/src/index.ts | 21 +- server/core/src/adapter.ts | 25 ++- server/core/src/indexer/indexer.ts | 39 ++-- server/core/src/mem.ts | 14 +- server/core/src/server/domainHelper.ts | 29 +-- server/core/src/server/index.ts | 6 +- server/core/src/server/storage.ts | 122 +++++++----- server/core/src/types.ts | 3 + server/elastic/src/backup.ts | 21 +- server/mongo/src/storage.ts | 239 ++++++++++++----------- server/mongo/src/utils.ts | 10 +- server/server-pipeline/src/indexing.ts | 13 +- server/server-storage/src/blobStorage.ts | 11 +- server/tool/src/index.ts | 12 +- tests/restore-local.sh | 2 +- tests/restore-workspace.sh | 2 +- 19 files changed, 327 insertions(+), 249 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index 7c1bf864052..829d0722dae 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -41,6 +41,7 @@ "METRICS_FILE": "${workspaceRoot}/metrics.txt", // Show metrics in console evert 30 seconds., "STORAGE_CONFIG": "minio|localhost?accessKey=minioadmin&secretKey=minioadmin", "SERVER_SECRET": "secret", + "ENABLE_CONSOLE": "true", "COLLABORATOR_URL": "ws://localhost:3078", "COLLABORATOR_API_URL": "http://localhost:3078", "REKONI_URL": "http://localhost:4004", diff --git a/common/scripts/version.txt b/common/scripts/version.txt index 29749785392..17e55ee553d 100644 --- a/common/scripts/version.txt +++ b/common/scripts/version.txt @@ -1 +1 @@ -"0.6.271" \ No newline at end of file +"0.6.274" \ No newline at end of file diff --git a/models/core/src/core.ts b/models/core/src/core.ts index 783c86ab83e..2f3bc0a7777 100644 --- a/models/core/src/core.ts +++ b/models/core/src/core.ts @@ -329,13 +329,13 @@ export class TDocIndexState extends TDoc implements DocIndexState { attributes!: Record @Prop(TypeBoolean(), getEmbeddedLabel('Removed')) - @Index(IndexKind.Indexed) + // @Index(IndexKind.Indexed) @Hidden() removed!: boolean // States for different stages @Prop(TypeRecord(), getEmbeddedLabel('Stages')) - @Index(IndexKind.Indexed) + // @Index(IndexKind.Indexed) @Hidden() stages!: Record diff --git a/models/core/src/index.ts b/models/core/src/index.ts index 088e98a8d47..f4a97a58181 100644 --- a/models/core/src/index.ts +++ b/models/core/src/index.ts @@ -27,7 +27,6 @@ import { type AttachedDoc, type Class, type Doc, - type DocIndexState, type IndexingConfiguration, type TxCollectionCUD } from '@hcengineering/core' @@ -284,8 +283,10 @@ export function createModel (builder: Builder): void { builder.createDoc(core.class.DomainIndexConfiguration, core.space.Model, { domain: DOMAIN_DOC_INDEX_STATE, + indexes: [{ keys: { removed: 1 }, filter: { removed: true } }], disabled: [ { attachedToClass: 1 }, + { objectClass: 1 }, { stages: 1 }, { generationId: 1 }, { space: 1 }, @@ -298,24 +299,6 @@ export function createModel (builder: Builder): void { skip: ['stages.'] }) - builder.mixin, IndexingConfiguration>>( - core.class.DocIndexState, - core.class.Class, - core.mixin.IndexConfiguration, - { - indexes: [ - { - keys: { - _class: 1, - stages: 1, - _id: 1, - modifiedOn: 1 - } - } - ] - } - ) - builder.mixin(core.class.Space, core.class.Class, core.mixin.FullTextSearchContext, { childProcessingAllowed: false }) diff --git a/server/core/src/adapter.ts b/server/core/src/adapter.ts index b3ea0602140..578cecd513d 100644 --- a/server/core/src/adapter.ts +++ b/server/core/src/adapter.ts @@ -23,7 +23,6 @@ import { type FindOptions, type FindResult, type Hierarchy, - type IndexingConfiguration, type MeasureContext, type ModelDb, type Ref, @@ -38,19 +37,23 @@ import type { ServerFindOptions } from './types' export interface DomainHelperOperations { create: (domain: Domain) => Promise exists: (domain: Domain) => boolean + + listDomains: () => Promise> createIndex: (domain: Domain, value: string | FieldIndexConfig, options?: { name: string }) => Promise dropIndex: (domain: Domain, name: string) => Promise listIndexes: (domain: Domain) => Promise<{ name: string }[]> - hasDocuments: (domain: Domain, count: number) => Promise + + // Could return 0 even if it has documents + estimatedCount: (domain: Domain) => Promise } export interface DomainHelper { checkDomain: ( ctx: MeasureContext, domain: Domain, - forceCreate: boolean, + documents: number, operations: DomainHelperOperations - ) => Promise + ) => Promise } export interface RawDBAdapterStream { @@ -87,15 +90,20 @@ export interface RawDBAdapter { close: () => Promise } +export type DbAdapterHandler = ( + domain: Domain, + event: 'add' | 'update' | 'delete' | 'read', + count: number, + time: number, + helper: DomainHelperOperations +) => void /** * @public */ export interface DbAdapter { init?: () => Promise - helper?: () => DomainHelperOperations - createIndexes: (domain: Domain, config: Pick, 'indexes'>) => Promise - removeOldIndex: (domain: Domain, deletePattern: RegExp[], keepPattern: RegExp[]) => Promise + helper: () => DomainHelperOperations close: () => Promise findAll: ( @@ -116,6 +124,9 @@ export interface DbAdapter { // Bulk update operations update: (ctx: MeasureContext, domain: Domain, operations: Map, DocumentUpdate>) => Promise + + // Allow to register a handler to listen for domain operations + on?: (handler: DbAdapterHandler) => void } /** diff --git a/server/core/src/indexer/indexer.ts b/server/core/src/indexer/indexer.ts index cca74d4277d..40889eb95f6 100644 --- a/server/core/src/indexer/indexer.ts +++ b/server/core/src/indexer/indexer.ts @@ -349,23 +349,36 @@ export class FullTextIndexPipeline implements FullTextPipeline { keepPattern.push(new RegExp(st.stageId)) } } + const helper = this.storage.helper() if (deletePattern.length > 0) { - await this.storage.removeOldIndex(DOMAIN_DOC_INDEX_STATE, deletePattern, keepPattern) + try { + const existingIndexes = await helper.listIndexes(DOMAIN_DOC_INDEX_STATE) + for (const existingIndex of existingIndexes) { + if (existingIndex.name !== undefined) { + const name: string = existingIndex.name + if (deletePattern.some((it) => it.test(name)) && !keepPattern.some((it) => it.test(name))) { + await helper.dropIndex(DOMAIN_DOC_INDEX_STATE, name) + } + } + } + } catch (err: any) { + console.error(err) + } } for (const st of this.stages) { if (this.cancelling) { return } - await this.storage.createIndexes(DOMAIN_DOC_INDEX_STATE, { - indexes: [ - { - keys: { - ['stages.' + st.stageId]: 1 - } + await this.storage.helper().createIndex( + DOMAIN_DOC_INDEX_STATE, + { + keys: { + ['stages.' + st.stageId]: 1 } - ] - }) + }, + { name: 'stages.' + st.stageId + '_1' } + ) } } @@ -481,7 +494,9 @@ export class FullTextIndexPipeline implements FullTextPipeline { async (ctx) => await this.storage.findAll(ctx, core.class.DocIndexState, q, { sort: { modifiedOn: SortingOrder.Descending }, - limit: globalIndexer.processingSize + limit: globalIndexer.processingSize, + skipClass: true, + skipSpace: true }) ) const toRemove: DocIndexState[] = [] @@ -594,7 +609,9 @@ export class FullTextIndexPipeline implements FullTextPipeline { _id: 1, stages: 1, objectClass: 1 - } + }, + skipSpace: true, + skipClass: true } ) diff --git a/server/core/src/mem.ts b/server/core/src/mem.ts index f3c0dcd43b3..06dcdcdfa51 100644 --- a/server/core/src/mem.ts +++ b/server/core/src/mem.ts @@ -32,7 +32,7 @@ import core, { type TxResult, type WorkspaceId } from '@hcengineering/core' -import { type DbAdapter } from './adapter' +import { type DbAdapter, type DomainHelperOperations } from './adapter' /** * @public @@ -49,6 +49,18 @@ export class DummyDbAdapter implements DbAdapter { async init (): Promise {} + helper (): DomainHelperOperations { + return { + create: async () => {}, + exists: () => true, + listDomains: async () => new Set(), + createIndex: async () => {}, + dropIndex: async () => {}, + listIndexes: async () => [], + estimatedCount: async () => 0 + } + } + async createIndexes (domain: Domain, config: Pick, 'indexes'>): Promise {} async removeOldIndex (domain: Domain, deletePattern: RegExp[], keepPattern: RegExp[]): Promise {} diff --git a/server/core/src/server/domainHelper.ts b/server/core/src/server/domainHelper.ts index 15955291c38..b0347f9319a 100644 --- a/server/core/src/server/domainHelper.ts +++ b/server/core/src/server/domainHelper.ts @@ -74,41 +74,24 @@ export class DomainIndexHelperImpl implements DomainHelper { } /** - * return false if and only if domain underline structures are not required. + * Check if some indexes need to be created for domain. */ async checkDomain ( ctx: MeasureContext, domain: Domain, - forceCreate: boolean, + documents: number, operations: DomainHelperOperations - ): Promise { + ): Promise { const domainInfo = this.domains.get(domain) const cfg = this.domainConfigurations.find((it) => it.domain === domain) - let exists = operations.exists(domain) - const hasDocuments = exists && (await operations.hasDocuments(domain, 1)) - // Drop collection if it exists and should not exists or doesn't have documents. - if (exists && (cfg?.disableCollection === true || (!hasDocuments && !forceCreate))) { - // We do not need this collection - return false - } - - if (forceCreate && !exists) { - await operations.create(domain) - ctx.info('collection will be created', domain) - exists = true - } - if (!exists) { - // Do not need to create, since not force and no documents. - return false - } const bb: (string | FieldIndexConfig)[] = [] const added = new Set() try { - const has50Documents = await operations.hasDocuments(domain, 50) + const has50Documents = documents > 50 const allIndexes = (await operations.listIndexes(domain)).filter((it) => it.name !== '_id_') - ctx.info('check indexes', { domain, has50Documents }) + ctx.info('check indexes', { domain, has50Documents, documents }) if (has50Documents) { for (const vv of [...(domainInfo?.values() ?? []), ...(cfg?.indexes ?? [])]) { try { @@ -188,7 +171,5 @@ export class DomainIndexHelperImpl implements DomainHelper { if (bb.length > 0) { ctx.info('created indexes', { domain, bb }) } - - return true } } diff --git a/server/core/src/server/index.ts b/server/core/src/server/index.ts index c44d93b6b77..8065f4b20fc 100644 --- a/server/core/src/server/index.ts +++ b/server/core/src/server/index.ts @@ -167,7 +167,7 @@ export async function createServerStorage ( const domainHelper = new DomainIndexHelperImpl(metrics, hierarchy, modelDb, conf.workspace) - return new TServerStorage( + const serverStorage = new TServerStorage( conf.domains, conf.defaultAdapter, adapters, @@ -184,6 +184,10 @@ export async function createServerStorage ( model, domainHelper ) + await ctx.with('init-domain-info', {}, async () => { + await serverStorage.initDomainInfo() + }) + return serverStorage } /** diff --git a/server/core/src/server/storage.ts b/server/core/src/server/storage.ts index 71a05827f23..9e320e291c3 100644 --- a/server/core/src/server/storage.ts +++ b/server/core/src/server/storage.ts @@ -79,6 +79,11 @@ import type { } from '../types' import { SessionContextImpl, createBroadcastEvent } from '../utils' +interface DomainInfo { + exists: boolean + documents: number +} + export class TServerStorage implements ServerStorage { private readonly fulltext: FullTextIndex hierarchy: Hierarchy @@ -92,14 +97,8 @@ export class TServerStorage implements ServerStorage { liveQuery: LQ branding: Branding | null - domainInfo = new Map< - Domain, - { - exists: boolean - checkPromise: Promise | undefined - lastCheck: number - } - >() + domainInfo = new Map() + statsCtx: MeasureContext emptyAdapter = new DummyDbAdapter() @@ -126,6 +125,71 @@ export class TServerStorage implements ServerStorage { this.branding = options.branding this.setModel(model) + this.statsCtx = metrics.newChild('stats-' + this.workspaceId.name, {}) + } + + async initDomainInfo (): Promise { + const adapterDomains = new Map>() + for (const d of this.hierarchy.domains()) { + // We need to init domain info + const info = this.getDomainInfo(d) + const adapter = this.adapters.get(d) ?? this.adapters.get(this.defaultAdapter) + if (adapter !== undefined) { + const h = adapter.helper?.() + if (h !== undefined) { + const dbDomains = adapterDomains.get(adapter) ?? (await h.listDomains()) + adapterDomains.set(adapter, dbDomains) + const dbIdIndex = dbDomains.has(d) + info.exists = dbIdIndex !== undefined + if (info.exists) { + info.documents = await h.estimatedCount(d) + } + } else { + info.exists = true + } + } else { + info.exists = false + } + } + for (const adapter of this.adapters.values()) { + adapter.on?.((domain, event, count, time, helper) => { + const info = this.getDomainInfo(domain) + const oldDocuments = info.documents + switch (event) { + case 'add': + info.documents += count + break + case 'update': + break + case 'delete': + info.documents -= count + break + case 'read': + break + } + + if (oldDocuments < 50 && info.documents > 50) { + // We have more 50 documents, we need to check for indexes + void this.domainHelper.checkDomain(this.metrics, domain, info.documents, helper) + } + if (oldDocuments > 50 && info.documents < 50) { + // We have more 50 documents, we need to check for indexes + void this.domainHelper.checkDomain(this.metrics, domain, info.documents, helper) + } + }) + } + } + + private getDomainInfo (domain: Domain): DomainInfo { + let info = this.domainInfo.get(domain) + if (info === undefined) { + info = { + documents: -1, + exists: false + } + this.domainInfo.set(domain, info) + } + return info } private newCastClient (hierarchy: Hierarchy, modelDb: ModelDb, metrics: MeasureContext): Client { @@ -172,13 +236,6 @@ export class TServerStorage implements ServerStorage { async close (): Promise { await this.fulltext.close() - for (const [domain, info] of this.domainInfo.entries()) { - if (info.checkPromise !== undefined) { - this.metrics.info('wait for check domain', { domain }) - // We need to be sure we wait for check to be complete - await info.checkPromise - } - } for (const o of this.adapters.values()) { await o.close() } @@ -193,36 +250,13 @@ export class TServerStorage implements ServerStorage { throw new Error('adapter not provided: ' + name) } - const helper = adapter.helper?.() - if (helper !== undefined) { - let info = this.domainInfo.get(domain) - if (info == null) { - // For first time, lets assume all is fine - info = { - exists: true, - lastCheck: Date.now(), - checkPromise: undefined - } - this.domainInfo.set(domain, info) - return adapter - } - if (Date.now() - info.lastCheck > 5 * 60 * 1000) { - // Re-check every 5 minutes - const exists = helper.exists(domain) - // We will create necessary indexes if required, and not touch collection if not required. - info = { - exists, - lastCheck: Date.now(), - checkPromise: this.domainHelper.checkDomain(this.metrics, domain, requireExists, helper) - } - this.domainInfo.set(domain, info) - } - if (!info.exists && !requireExists) { - return this.emptyAdapter - } - // If we require it exists, it will be exists - info.exists = true + const info = this.getDomainInfo(domain) + + if (!info.exists && !requireExists) { + return this.emptyAdapter } + // If we require it exists, it will be exists + info.exists = true return adapter } diff --git a/server/core/src/types.ts b/server/core/src/types.ts index 99f4ebc2d63..a2fa30fcbb5 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -51,6 +51,9 @@ import { type StorageAdapter } from './storage' export interface ServerFindOptions extends FindOptions { domain?: Domain // Allow to find for Doc's in specified domain only. prefix?: string + + skipClass?: boolean + skipSpace?: boolean } /** * @public diff --git a/server/elastic/src/backup.ts b/server/elastic/src/backup.ts index 53755ab1a50..2ac433ab841 100644 --- a/server/elastic/src/backup.ts +++ b/server/elastic/src/backup.ts @@ -33,7 +33,7 @@ import { WorkspaceId } from '@hcengineering/core' import { getMetadata } from '@hcengineering/platform' -import serverCore, { DbAdapter } from '@hcengineering/server-core' +import serverCore, { DbAdapter, type DomainHelperOperations } from '@hcengineering/server-core' function getIndexName (): string { return getMetadata(serverCore.metadata.ElasticIndexName) ?? 'storage_index' @@ -61,7 +61,24 @@ class ElasticDataAdapter implements DbAdapter { this.getDocId = (fulltext) => fulltext.slice(0, -1 * (this.workspaceString.length + 1)) as Ref } - async groupBy(ctx: MeasureContext, domain: Domain, field: string): Promise> { + helper (): DomainHelperOperations { + return { + create: async () => {}, + exists: () => true, + listDomains: async () => new Set(), + createIndex: async () => {}, + dropIndex: async () => {}, + listIndexes: async () => [], + estimatedCount: async () => 0 + } + } + + async groupBy( + ctx: MeasureContext, + domain: Domain, + field: string, + query?: DocumentQuery + ): Promise> { return new Set() } diff --git a/server/mongo/src/storage.ts b/server/mongo/src/storage.ts index c1ee4ec32ca..a011c93566f 100644 --- a/server/mongo/src/storage.ts +++ b/server/mongo/src/storage.ts @@ -37,7 +37,6 @@ import core, { type FindResult, type FullParamsType, type Hierarchy, - type IndexingConfiguration, type Lookup, type MeasureContext, type Mixin, @@ -65,6 +64,7 @@ import { estimateDocSize, updateHashForDoc, type DbAdapter, + type DbAdapterHandler, type DomainHelperOperations, type ServerFindOptions, type StorageAdapter, @@ -76,7 +76,6 @@ import { type AbstractCursor, type AnyBulkWriteOperation, type Collection, - type CreateIndexesOptions, type Db, type Document, type Filter, @@ -131,6 +130,18 @@ abstract class MongoAdapterBase implements DbAdapter { findRateLimit = new RateLimiter(parseInt(process.env.FIND_RLIMIT ?? '1000')) rateLimit = new RateLimiter(parseInt(process.env.TX_RLIMIT ?? '5')) + handlers: DbAdapterHandler[] = [] + + on (handler: DbAdapterHandler): void { + this.handlers.push(handler) + } + + handleEvent (domain: Domain, event: 'add' | 'update' | 'delete' | 'read', count: number, time: number): void { + for (const handler of this.handlers) { + handler(domain, event, count, time, this._db) + } + } + constructor ( protected readonly db: Db, protected readonly hierarchy: Hierarchy, @@ -151,45 +162,6 @@ abstract class MongoAdapterBase implements DbAdapter { return this._db } - async createIndexes (domain: Domain, config: Pick, 'indexes'>): Promise { - for (const value of config.indexes) { - try { - if (typeof value === 'string') { - await this.collection(domain).createIndex(value) - } else { - const opt: CreateIndexesOptions = {} - if (value.filter !== undefined) { - opt.partialFilterExpression = value.filter - } else if (value.sparse === true) { - opt.sparse = true - } - await this.collection(domain).createIndex(value.keys, opt) - } - } catch (err: any) { - console.error('failed to create index', domain, value, err) - } - } - } - - async removeOldIndex (domain: Domain, deletePattern: RegExp[], keepPattern: RegExp[]): Promise { - try { - const existingIndexes = await this.collection(domain).indexes() - for (const existingIndex of existingIndexes) { - if (existingIndex.name !== undefined) { - const name: string = existingIndex.name - if ( - deletePattern.some((it) => it.test(name)) && - (existingIndex.sparse === true || !keepPattern.some((it) => it.test(name))) - ) { - await this.collection(domain).dropIndex(name) - } - } - } - } catch (err: any) { - console.error(err) - } - } - async tx (ctx: MeasureContext, ...tx: Tx[]): Promise { return [] } @@ -198,7 +170,11 @@ abstract class MongoAdapterBase implements DbAdapter { this.client.close() } - private translateQuery(clazz: Ref>, query: DocumentQuery): Filter { + private translateQuery( + clazz: Ref>, + query: DocumentQuery, + options?: ServerFindOptions + ): Filter { const translated: any = {} for (const key in query) { const value = (query as any)[key] @@ -213,6 +189,13 @@ abstract class MongoAdapterBase implements DbAdapter { } translated[tkey] = value } + if (options?.skipSpace === true) { + delete translated.space + } + if (options?.skipClass === true) { + delete translated._class + return translated + } const baseClass = this.hierarchy.getBaseClass(clazz) if (baseClass !== core.class.Doc) { const classes = this.hierarchy.getDescendants(baseClass).filter((it) => !this.hierarchy.isMixin(it)) @@ -473,12 +456,15 @@ abstract class MongoAdapterBase implements DbAdapter { private async findWithPipeline( ctx: MeasureContext, + domain: Domain, clazz: Ref>, query: DocumentQuery, - options?: ServerFindOptions + options: ServerFindOptions, + stTime: number ): Promise> { + const st = Date.now() const pipeline: any[] = [] - const match = { $match: this.translateQuery(clazz, query) } + const match = { $match: this.translateQuery(clazz, query, options) } const slowPipeline = isLookupQuery(query) || isLookupSort(options?.sort) const steps = await ctx.with('get-lookups', {}, async () => await this.getLookups(clazz, options?.lookup)) if (slowPipeline) { @@ -506,9 +492,6 @@ abstract class MongoAdapterBase implements DbAdapter { pipeline.push({ $project: projection }) } - // const domain = this.hierarchy.getDomain(clazz) - const domain = options?.domain ?? this.hierarchy.getDomain(clazz) - const cursor = this.collection(domain).aggregate>(pipeline) let result: WithLookup[] = [] let total = options?.total === true ? 0 : -1 @@ -558,6 +541,17 @@ abstract class MongoAdapterBase implements DbAdapter { ) total = arr?.[0]?.total ?? 0 } + const edTime = Date.now() + if (edTime - stTime > 1000 || st - stTime > 1000) { + ctx.error('aggregate', { + time: edTime - stTime, + clazz, + query: cutObjectArray(query), + options, + queueTime: st - stTime + }) + } + this.handleEvent(domain, 'read', result.length, edTime - st) return toFindResult(this.stripHash(result) as T[], total) } @@ -643,7 +637,12 @@ abstract class MongoAdapterBase implements DbAdapter { } @withContext('groupBy') - async groupBy(ctx: MeasureContext, domain: Domain, field: string): Promise> { + async groupBy( + ctx: MeasureContext, + domain: Domain, + field: string, + query?: DocumentQuery + ): Promise> { const result = await ctx.with( 'groupBy', { domain }, @@ -651,6 +650,7 @@ abstract class MongoAdapterBase implements DbAdapter { const coll = this.collection(domain) const grResult = await coll .aggregate([ + ...(query !== undefined ? [{ $match: query }] : []), { $group: { _id: '$' + field @@ -716,20 +716,20 @@ abstract class MongoAdapterBase implements DbAdapter { const stTime = Date.now() return await this.findRateLimit.exec(async () => { const st = Date.now() + const domain = options?.domain ?? this.hierarchy.getDomain(_class) const result = await this.collectOps( ctx, - this.hierarchy.findDomain(_class), + domain, 'find', async (ctx) => { - const domain = options?.domain ?? this.hierarchy.getDomain(_class) if ( options != null && (options?.lookup != null || this.isEnumSort(_class, options) || this.isRulesSort(options)) ) { - return await this.findWithPipeline(ctx, _class, query, options) + return await this.findWithPipeline(ctx, domain, _class, query, options, stTime) } const coll = this.collection(domain) - const mongoQuery = this.translateQuery(_class, query) + const mongoQuery = this.translateQuery(_class, query, options) if (options?.limit === 1) { // Skip sort/projection/etc. @@ -825,6 +825,7 @@ abstract class MongoAdapterBase implements DbAdapter { queueTime: st - stTime }) } + this.handleEvent(domain, 'read', result.length, edTime - st) return result }) } @@ -1122,7 +1123,6 @@ class MongoAdapter extends MongoAdapterBase { }) await this.rateLimit.exec(async () => { - const domains: Promise[] = [] for (const [domain, txs] of byDomain) { if (domain === undefined) { continue @@ -1146,75 +1146,80 @@ class MongoAdapter extends MongoAdapterBase { ) { continue } - domains.push( - this.collectOps( - ctx, - domain, - 'tx', - async (ctx) => { - const coll = this.db.collection(domain) - - // Minir optimizations - // Add Remove optimization - - if (domainBulk.add.length > 0) { - await ctx.with('insertMany', {}, async () => { - await coll.insertMany(domainBulk.add, { ordered: false }) - }) - } - if (domainBulk.update.size > 0) { - // Extract similar update to update many if possible - // TODO: - await ctx.with('updateMany-bulk', {}, async () => { - await coll.bulkWrite( - Array.from(domainBulk.update.entries()).map((it) => ({ - updateOne: { - filter: { _id: it[0] }, - update: { - $set: it[1] - } + await this.collectOps( + ctx, + domain, + 'tx', + async (ctx) => { + const coll = this.db.collection(domain) + + // Minir optimizations + // Add Remove optimization + + if (domainBulk.add.length > 0) { + await ctx.with('insertMany', {}, async () => { + const st = Date.now() + const result = await coll.insertMany(domainBulk.add, { ordered: false }) + this.handleEvent(domain, 'add', result.insertedCount, Date.now() - st) + }) + } + if (domainBulk.update.size > 0) { + // Extract similar update to update many if possible + // TODO: + await ctx.with('updateMany-bulk', {}, async () => { + const st = Date.now() + const result = await coll.bulkWrite( + Array.from(domainBulk.update.entries()).map((it) => ({ + updateOne: { + filter: { _id: it[0] }, + update: { + $set: it[1] } - })), - { - ordered: false } - ) - }) - } - if (domainBulk.bulkOperations.length > 0) { - await ctx.with('bulkWrite', {}, async () => { - await coll.bulkWrite(domainBulk.bulkOperations, { + })), + { ordered: false - }) - }) - } - if (domainBulk.findUpdate.size > 0) { - await ctx.with('find-result', {}, async () => { - const docs = await coll.find({ _id: { $in: Array.from(domainBulk.findUpdate) } }).toArray() - result.push(...docs) - }) - } - - if (domainBulk.raw.length > 0) { - await ctx.with('raw', {}, async () => { - for (const r of domainBulk.raw) { - result.push({ object: await r() }) } + ) + this.handleEvent(domain, 'update', result.modifiedCount, Date.now() - st) + }) + } + if (domainBulk.bulkOperations.length > 0) { + await ctx.with('bulkWrite', {}, async () => { + const st = Date.now() + const result = await coll.bulkWrite(domainBulk.bulkOperations, { + ordered: false }) - } - }, - { - domain, - add: domainBulk.add.length, - update: domainBulk.update.size, - bulk: domainBulk.bulkOperations.length, - find: domainBulk.findUpdate.size, - raw: domainBulk.raw.length + this.handleEvent(domain, 'update', result.modifiedCount, Date.now() - st) + }) } - ) + if (domainBulk.findUpdate.size > 0) { + await ctx.with('find-result', {}, async () => { + const st = Date.now() + const docs = await coll.find({ _id: { $in: Array.from(domainBulk.findUpdate) } }).toArray() + result.push(...docs) + this.handleEvent(domain, 'read', docs.length, Date.now() - st) + }) + } + + if (domainBulk.raw.length > 0) { + await ctx.with('raw', {}, async () => { + for (const r of domainBulk.raw) { + result.push({ object: await r() }) + } + }) + } + }, + { + domain, + add: domainBulk.add.length, + update: domainBulk.update.size, + bulk: domainBulk.bulkOperations.length, + find: domainBulk.findUpdate.size, + raw: domainBulk.raw.length + } ) } - await Promise.all(domains) }) return result } @@ -1395,6 +1400,7 @@ class MongoAdapter extends MongoAdapterBase { if (tx.retrieve === true) { bulk.raw.push(async () => { + const st = Date.now() const res = await this.collection(domain).findOneAndUpdate( { _id: tx.objectId }, { @@ -1407,6 +1413,9 @@ class MongoAdapter extends MongoAdapterBase { } as unknown as UpdateFilter, { returnDocument: 'after', includeResultMetadata: true } ) + const dnow = Date.now() - st + this.handleEvent(domain, 'read', 1, dnow) + this.handleEvent(domain, 'update', 1, dnow) return res.value as TxResult }) } else { @@ -1459,6 +1468,7 @@ class MongoTxAdapter extends MongoAdapterBase implements TxAdapter { if (tx.length === 0) { return [] } + const st = Date.now() await this.collectOps( ctx, DOMAIN_TX, @@ -1468,6 +1478,7 @@ class MongoTxAdapter extends MongoAdapterBase implements TxAdapter { }, { tx: tx.length } ) + this.handleEvent(DOMAIN_TX, 'add', tx.length, Date.now() - st) return [] } diff --git a/server/mongo/src/utils.ts b/server/mongo/src/utils.ts index 4410a07f452..9e32881cf04 100644 --- a/server/mongo/src/utils.ts +++ b/server/mongo/src/utils.ts @@ -158,6 +158,11 @@ export class DBCollectionHelper implements DomainHelperOperations { collections = new Map>() constructor (readonly db: Db) {} + async listDomains (): Promise> { + const collections = await this.db.listCollections({}, { nameOnly: true }).toArray() + return new Set(collections.map((it) => it.name as unknown as Domain)) + } + async init (domain?: Domain): Promise { if (domain === undefined) { // Init existing collecfions @@ -224,7 +229,8 @@ export class DBCollectionHelper implements DomainHelperOperations { return await this.collection(domain).listIndexes().toArray() } - async hasDocuments (domain: Domain, count: number): Promise { - return (await this.collection(domain).countDocuments({}, { limit: count })) >= count + async estimatedCount (domain: Domain): Promise { + const c = this.collection(domain) + return await c.estimatedDocumentCount() } } diff --git a/server/server-pipeline/src/indexing.ts b/server/server-pipeline/src/indexing.ts index e79988b1d05..26a84b26df1 100644 --- a/server/server-pipeline/src/indexing.ts +++ b/server/server-pipeline/src/indexing.ts @@ -1,6 +1,5 @@ /* eslint-disable @typescript-eslint/unbound-method */ import { type Branding, type MeasureContext, type WorkspaceId } from '@hcengineering/core' -import { OpenAIEmbeddingsStage } from '@hcengineering/openai' import { CollaborativeContentRetrievalStage } from '@hcengineering/server-collaboration' import { ContentRetrievalStage, @@ -65,14 +64,14 @@ export function createIndexStages ( const pushStage = new FullTextPushStage(storage, adapter, workspace, branding) stages.push(pushStage) - // OpenAI prepare stage - const openAIStage = new OpenAIEmbeddingsStage(adapter, workspace) - // We depend on all available stages. - openAIStage.require = stages.map((it) => it.stageId) + // // OpenAI prepare stage + // const openAIStage = new OpenAIEmbeddingsStage(adapter, workspace) + // // We depend on all available stages. + // openAIStage.require = stages.map((it) => it.stageId) - openAIStage.updateSummary(summaryStage) + // openAIStage.updateSummary(summaryStage) - stages.push(openAIStage) + // stages.push(openAIStage) return stages } diff --git a/server/server-storage/src/blobStorage.ts b/server/server-storage/src/blobStorage.ts index 818ded03571..ef46bd51fe1 100644 --- a/server/server-storage/src/blobStorage.ts +++ b/server/server-storage/src/blobStorage.ts @@ -34,7 +34,12 @@ import core, { } from '@hcengineering/core' import { createMongoAdapter } from '@hcengineering/mongo' import { PlatformError, unknownError } from '@hcengineering/platform' -import { DbAdapter, StorageAdapter, type StorageAdapterEx } from '@hcengineering/server-core' +import { + DbAdapter, + StorageAdapter, + type DomainHelperOperations, + type StorageAdapterEx +} from '@hcengineering/server-core' class StorageBlobAdapter implements DbAdapter { constructor ( @@ -53,6 +58,10 @@ class StorageBlobAdapter implements DbAdapter { return await this.blobAdapter.findAll(ctx, _class, query, options) } + helper (): DomainHelperOperations { + return this.blobAdapter.helper() + } + async groupBy(ctx: MeasureContext, domain: Domain, field: string): Promise> { return await this.blobAdapter.groupBy(ctx, domain, field) } diff --git a/server/tool/src/index.ts b/server/tool/src/index.ts index 0f1f4346600..36d54b0d795 100644 --- a/server/tool/src/index.ts +++ b/server/tool/src/index.ts @@ -516,17 +516,7 @@ async function createUpdateIndexes ( if (domain === DOMAIN_MODEL || domain === DOMAIN_TRANSIENT || domain === DOMAIN_BENCHMARK) { continue } - const result = await domainHelper.checkDomain(ctx, domain, false, dbHelper) - if (!result && dbHelper.exists(domain)) { - try { - logger.log('dropping domain', { domain }) - if ((await db.collection(domain).countDocuments({})) === 0) { - await db.dropCollection(domain) - } - } catch (err) { - logger.error('error: failed to delete collection', { domain, err }) - } - } + await domainHelper.checkDomain(ctx, domain, await dbHelper.estimatedCount(domain), dbHelper) completed++ await progress((100 / allDomains.length) * completed) } diff --git a/tests/restore-local.sh b/tests/restore-local.sh index 10c39937011..df4b077ec8c 100755 --- a/tests/restore-local.sh +++ b/tests/restore-local.sh @@ -9,7 +9,7 @@ export SERVER_SECRET=secret # Restore workspace contents in mongo/elastic ./tool-local.sh backup-restore ./sanity-ws sanity-ws -./tool-local.sh upgrade-workspace sanity-ws +./tool-local.sh upgrade-workspace sanity-ws --indexes # Re-assign user to workspace. ./tool-local.sh assign-workspace user1 sanity-ws diff --git a/tests/restore-workspace.sh b/tests/restore-workspace.sh index 4eec408b3cf..0fb8ac485b5 100755 --- a/tests/restore-workspace.sh +++ b/tests/restore-workspace.sh @@ -3,7 +3,7 @@ # Restore workspace contents in mongo/elastic ./tool.sh backup-restore ./sanity-ws sanity-ws -./tool.sh upgrade-workspace sanity-ws +./tool.sh upgrade-workspace sanity-ws --indexes # Re-assign user to workspace. ./tool.sh assign-workspace user1 sanity-ws