From 95c19740623fdc87367ea2f458696d8aecdf01cb Mon Sep 17 00:00:00 2001 From: Andrey Sobolev Date: Wed, 2 Jul 2025 15:48:41 +0700 Subject: [PATCH] Improve backup logging (#9434) Signed-off-by: Andrey Sobolev --- server/backup/src/backup.ts | 33 ++++++--- server/backup/src/service.ts | 35 ++++++---- server/backup/src/utils.ts | 70 +++++++++++-------- ws-tests/api-tests/src/__tests__/rest.test.ts | 2 +- 4 files changed, 86 insertions(+), 54 deletions(-) diff --git a/server/backup/src/backup.ts b/server/backup/src/backup.ts index db09888f61..9576a07485 100644 --- a/server/backup/src/backup.ts +++ b/server/backup/src/backup.ts @@ -85,6 +85,7 @@ export async function backup ( token?: string fullVerify?: boolean keepSnapshots: number + msg?: Record } = { force: false, timeout: 0, @@ -128,7 +129,7 @@ export async function backup ( const tmpRoot = mkdtempSync('huly') - const forcedFullCheck = '2' + const forcedFullCheck = '3' try { let backupInfo: BackupInfo = { @@ -202,7 +203,8 @@ export async function backup ( ctx.warn('Compacting backup') await compactBackup(ctx, storage, true, { blobLimit: options.blobDownloadLimit, - skipContentTypes: options.skipBlobContentTypes + skipContentTypes: options.skipBlobContentTypes, + msg: { workspaceId, url: wsIds.url } }) backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) @@ -253,7 +255,7 @@ export async function backup ( ) ] - ctx.info('domains for dump', { domains: domains.length }) + ctx.info('domains for dump', { domains: domains.length, workspace: workspaceId, url: wsIds.url }) backupInfo.lastTxId = '' // Clear until full backup will be complete @@ -289,7 +291,9 @@ export async function backup ( ctx.info('downloaded', { msg, written: newDownloadedMb, - pending + pending, + workspace: workspaceId, + url: wsIds.url }) } } @@ -357,7 +361,8 @@ export async function backup ( processed, digest: digest.size, time: Date.now() - st, - workspace: workspaceId + workspace: workspaceId, + url: wsIds.url }) st = Date.now() } @@ -408,7 +413,8 @@ export async function backup ( processed, digest: digest.size, time: Date.now() - st, - workspace: workspaceId + workspace: workspaceId, + url: wsIds.url }) await ctx.with('closeChunk', {}, async () => { await connection.closeChunk(ctx, idx as number) @@ -461,11 +467,13 @@ export async function backup ( const dHash = await connection.getDomainHash(ctx, domain) if (backupInfo.domainHashes[domain] === dHash && !fullCheck) { - ctx.info('no changes in domain', { domain }) + ctx.info('no changes in domain', { domain, workspaceId, url: wsIds.url }) return } // Cumulative digest - const digest = await ctx.with('load-digest', {}, (ctx) => loadDigest(ctx, storage, backupInfo.snapshots, domain)) + const digest = await ctx.with('load-digest', {}, (ctx) => + loadDigest(ctx, storage, backupInfo.snapshots, domain, undefined, options.msg) + ) const same = new Map, string>() let _pack: Pack | undefined @@ -578,6 +586,7 @@ export async function backup ( needRetrieve: needRetrieveChunks.reduce((v, docs) => v + docs.size, 0), toLoad: needRetrieve.size, workspace: workspaceId, + url: wsIds.url, lastSize: Math.round((lastSize * 100) / (1024 * 1024)) / 100 }) let docs: Doc[] = [] @@ -695,7 +704,7 @@ export async function backup ( }) // We need to upload file to storage - ctx.info('>>>> upload pack', { storageFile, size: sz, workspace: wsIds.url }) + ctx.info('>>>> upload pack', { storageFile, size: sz, url: wsIds.url, workspace: workspaceId }) await storage.writeFile(storageFile, createReadStream(tmpFile)) await rm(tmpFile) @@ -734,7 +743,9 @@ export async function backup ( blob: blob._id, provider: blob.provider, size: Math.round(blob.size / (1024 * 1024)), - limit: options.blobDownloadLimit + limit: options.blobDownloadLimit, + workspace: workspaceId, + url: wsIds.url }) processChanges(d, true) if (progress !== undefined) { @@ -868,7 +879,7 @@ export async function backup ( current: Math.round(process.memoryUsage().heapUsed / (1024 * 1024)) } if (mm.old > mm.current + mm.current / 10) { - ctx.info('memory-stats', mm) + ctx.info('memory-stats', { ...mm, workspace: workspaceId }) } await ctx.with('process-domain', { domain }, async (ctx) => { await processDomain( diff --git a/server/backup/src/service.ts b/server/backup/src/service.ts index d8d6b3053d..481ae1a210 100644 --- a/server/backup/src/service.ts +++ b/server/backup/src/service.ts @@ -224,6 +224,21 @@ class BackupWorker { } this.workspacesToBackup.delete(ws.uuid) this.activeWorkspaces.add(ws.uuid) + const handleFailedBackup = (ws: WorkspaceInfoWithStatus): void => { + const f = this.failedWorkspaces.get(ws.uuid) + if (f === undefined) { + this.failedWorkspaces.set(ws.uuid, { + info: ws, + counter: 1 + }) + } else { + f.counter++ + } + if ((f?.counter ?? 1) < 5) { + this.workspacesToBackup.set(ws.uuid, ws) + } + } + await this.rateLimiter.add( async () => { try { @@ -236,21 +251,13 @@ class BackupWorker { const totalTime = Date.now() - st this.allBackupTime += totalTime this.processed++ + } else { + handleFailedBackup(ws) + ctx.error('Backup failed, put back to queue', { workspace: ws.uuid, url: ws.url }) } } catch (err: any) { ctx.error('Backup failed', { err }) - const f = this.failedWorkspaces.get(ws.uuid) - if (f === undefined) { - this.failedWorkspaces.set(ws.uuid, { - info: ws, - counter: 1 - }) - } else { - f.counter++ - } - if ((f?.counter ?? 1) < 5) { - this.workspacesToBackup.set(ws.uuid, ws) - } + handleFailedBackup(ws) } finally { this.activeWorkspaces.delete(ws.uuid) } @@ -324,6 +331,10 @@ class BackupWorker { fullVerify: this.fullCheck, progress: (progress) => { return notify?.(progress) ?? Promise.resolve() + }, + msg: { + workspaceUrl: ws.url, + workspaceUuid: ws.uuid } }), { workspace: ws.uuid, url: ws.url } diff --git a/server/backup/src/utils.ts b/server/backup/src/utils.ts index 3cc3fb4c8e..7348f35bab 100644 --- a/server/backup/src/utils.ts +++ b/server/backup/src/utils.ts @@ -335,6 +335,7 @@ export async function compactBackup ( opt?: { blobLimit?: number skipContentTypes?: string[] + msg?: Record }, recalculateDigest: boolean = false ): Promise { @@ -352,16 +353,16 @@ export async function compactBackup ( if (await storage.exists(infoFile)) { backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) } else { - console.log('No backup found') + ctx.info('No backup found', { ...(opt?.msg ?? {}) }) return } if (backupInfo.version !== '0.6.2') { - console.log('Invalid backup version') + ctx.info('Invalid backup version', { ...(opt?.msg ?? {}) }) return } if (backupInfo.snapshots.length < 1 && !force) { - console.log('No need to compact') + ctx.info('No need to compact', { ...(opt?.msg ?? {}) }) return } @@ -406,7 +407,7 @@ export async function compactBackup ( } for (const domain of domains) { - console.log('compacting domain...', domain) + ctx.info('compacting domain...', { domain, ...(opt?.msg ?? {}) }) const processedChanges: Snapshot = { added: new Map(), @@ -427,10 +428,10 @@ export async function compactBackup ( // Cumulative digest // Documents modified in compacting snapshots in case of restart - const untouchedDigest = await loadDigest(ctx, storage, [snapshot], domain) + const untouchedDigest = await loadDigest(ctx, storage, [snapshot], domain, undefined, opt?.msg) // We need to load snapshots from removed ones and - const digest = await loadDigest(ctx, storage, snapshotsToClean, domain) + const digest = await loadDigest(ctx, storage, snapshotsToClean, domain, undefined, opt?.msg) // We remove all items we have in last part Array.from(untouchedDigest.keys()).forEach((it) => digest.delete(it)) @@ -493,7 +494,7 @@ export async function compactBackup ( backupIndex, `${domain}-data-${snapshot.date}-${extendZero(snapshot.stIndex)}.tar.gz` ) - console.log('storing from domain', domain, storageFile) + ctx.info('storing from domain', { domain, storageFile, ...(opt?.msg ?? {}) }) domainInfo.storage = [...(domainInfo.storage ?? []), storageFile] const tmpFile = join(tmpRoot, basename(storageFile) + '.tmp') @@ -518,7 +519,7 @@ export async function compactBackup ( storageZip.pipe(sizePass) _packClose = async () => { - ctx.info('finalize pack', { storageFile, size: sz }) + ctx.info('finalize pack(compact)', { storageFile, size: sz, ...(opt?.msg ?? {}) }) await new Promise((resolve) => { tempFile.on('close', () => { resolve() @@ -526,7 +527,7 @@ export async function compactBackup ( _pack?.finalize() }) // We need to upload file to storage - ctx.info('>>>> upload pack', { storageFile, size: sz }) + ctx.info('>>>> upload pack(compact)', { storageFile, size: sz, ...(opt?.msg ?? {}) }) await storage.writeFile(storageFile, createReadStream(tmpFile)) await rm(tmpFile) @@ -574,7 +575,11 @@ export async function compactBackup ( const newSkipPrint = Math.round(skipSize / (1024 * 1024 * 100)) if (newSkipPrint !== lastSkipPrint) { lastSkipPrint = newSkipPrint - console.log('skipping blobs', skipBlobs, Math.round(skipSize / (1024 * 1024))) + ctx.info('skipping blobs', { + skipBlobs, + size: Math.round(skipSize / (1024 * 1024)), + ...(opt?.msg ?? {}) + }) } const bsize = blob.size == null || Number.isNaN(blob.size) || !Number.isInteger(blob.size) ? 0 : blob.size @@ -606,18 +611,18 @@ export async function compactBackup ( const d = s.domains[domain] if (d !== undefined && digest.size > 0) { - ctx.info('checking-domain', { domain, name: s.date }) + ctx.info('checking-domain', { domain, name: s.date, ...(opt?.msg ?? {}) }) const sDigest = await loadDigest(ctx, storage, [s], domain) const requiredDocs = new Map(Array.from(sDigest.entries()).filter(([it]) => digest.has(it))) if (requiredDocs.size > 0) { - console.log('updating', domain, requiredDocs.size) + ctx.info('updating', { domain, requiredDocs: requiredDocs.size, ...(opt?.msg ?? {}) }) // We have required documents here. for (const sf of d.storage ?? []) { if (digest.size === 0) { break } try { - console.log('processing', sf, processed) + ctx.info('processing', { sf, processed, ...(opt?.msg ?? {}) }) const readStream = await storage.load(sf) const ex = extract() @@ -718,11 +723,11 @@ export async function compactBackup ( resolve(null) }) readStream.on('error', (err) => { - ctx.error('error during processing', { snapshot, err }) + ctx.error('error during processing', { snapshot, err, ...(opt?.msg ?? {}) }) reject(err) }) unzip.on('error', (err) => { - ctx.error('error during processing', { snapshot, err }) + ctx.error('error during processing', { snapshot, err, ...(opt?.msg ?? {}) }) reject(err) }) }) @@ -739,17 +744,17 @@ export async function compactBackup ( } } } else { - console.log('domain had no changes', domain) + ctx.info('domain had no changes', { domain, ...(opt?.msg ?? {}) }) } } if (d !== undefined) { for (const sf of d.storage ?? []) { - console.log('removing', sf) + ctx.info('removing', { sf, ...(opt?.msg ?? {}) }) dirsToClean.add(dirname(sf)) filesToClean.add(sf) } for (const sf of d.snapshots ?? []) { - console.log('removing', sf) + ctx.info('removing', { sf, ...(opt?.msg ?? {}) }) dirsToClean.add(dirname(sf)) filesToClean.add(sf) } @@ -807,10 +812,10 @@ export async function compactBackup ( await rebuildSizeInfo(storage, [], ctx, result, backupInfo, infoFile) } } catch (err: any) { - console.error(err) + ctx.error(err, { ...(opt?.msg ?? {}) }) } finally { await rm(tmpRoot, { recursive: true }) - console.log('end compacting') + ctx.info('end compacting', { ...(opt?.msg ?? {}) }) } } @@ -833,8 +838,12 @@ export function migradeBlobData (blob: Blob, etag: string): string { * Will check backup integrity, and in case of some missing resources, will update digest files, so next backup will backup all missing parts. * @public */ -export async function checkBackupIntegrity (ctx: MeasureContext, storage: BackupStorage): Promise { - console.log('check backup integrity') +export async function checkBackupIntegrity ( + ctx: MeasureContext, + storage: BackupStorage, + msg?: Record +): Promise { + ctx.info('check backup integrity', { ...(msg ?? {}) }) try { let backupInfo: BackupInfo @@ -845,11 +854,11 @@ export async function checkBackupIntegrity (ctx: MeasureContext, storage: Backup if (await storage.exists(infoFile)) { backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) } else { - console.log('No backup found') + ctx.info('No backup found', { ...(msg ?? {}) }) return } if (backupInfo.version !== '0.6.2') { - console.log('Invalid backup version') + ctx.info('Invalid backup version', { ...(msg ?? {}) }) return } @@ -866,7 +875,7 @@ export async function checkBackupIntegrity (ctx: MeasureContext, storage: Backup let modified = false for (const domain of domains) { - console.log('checking domain...', domain) + ctx.info('checking domain...', { domain, ...(msg ?? {}) }) const { modified: mm, modifiedFiles } = await verifyDigest(ctx, storage, backupInfo.snapshots, domain) if (mm) { recheckSizes.push(...modifiedFiles) @@ -892,9 +901,9 @@ export async function checkBackupIntegrity (ctx: MeasureContext, storage: Backup } await rebuildSizeInfo(storage, recheckSizes, ctx, bresult, backupInfo, infoFile) } catch (err: any) { - console.error(err) + ctx.error(err, { ...(msg ?? {}) }) } finally { - console.log('end compacting') + ctx.info('end checking integrity', { ...(msg ?? {}) }) } } @@ -913,7 +922,8 @@ export async function loadDigest ( storage: BackupStorage, snapshots: BackupSnapshot[], domain: Domain, - date?: number + date?: number, + msg?: Record ): Promise, string>> { const result = new Map, string>() for (const s of snapshots) { @@ -935,7 +945,7 @@ export async function loadDigest ( result.delete(d) } } catch (err: any) { - ctx.warn('failed to load digest', { snapshot: d.snapshot }) + ctx.warn('failed to load digest', { snapshot: d.snapshot, ...(msg ?? {}) }) } } for (const snapshot of d?.snapshots ?? []) { @@ -963,7 +973,7 @@ export async function loadDigest ( result.delete(k as Ref) } } catch (err: any) { - ctx.warn('digest is broken', { domain, err: err.message, snapshot }) + ctx.warn('digest is broken', { domain, err: err.message, snapshot, ...(msg ?? {}) }) } } // Stop if stop date is matched and provided diff --git a/ws-tests/api-tests/src/__tests__/rest.test.ts b/ws-tests/api-tests/src/__tests__/rest.test.ts index a0698bf884..f7cd7ce555 100644 --- a/ws-tests/api-tests/src/__tests__/rest.test.ts +++ b/ws-tests/api-tests/src/__tests__/rest.test.ts @@ -280,7 +280,7 @@ describe('rest-api-server', () => { async function checkFindPerformance (conn: RestClient): Promise { let ops = 0 let total = 0 - const attempts = 1000 + const attempts = 500 for (let i = 0; i < attempts; i++) { const st = performance.now() const spaces = await conn.findAll(core.class.Space, {})