From 5129ed0f561bf0cc7303769fefaaf2169d2fcf4d Mon Sep 17 00:00:00 2001 From: Alexander Onnikov Date: Thu, 31 Jul 2025 21:26:46 +0700 Subject: [PATCH] fix: log more stats in datalake (#9624) Signed-off-by: Alexander Onnikov --- .../pod-datalake/src/datalake/datalake.ts | 8 ++--- .../datalake/pod-datalake/src/datalake/db.ts | 36 ++++++++++++------- .../pod-datalake/src/handlers/blob.ts | 6 ++-- .../pod-datalake/src/handlers/image.ts | 11 ++++-- .../datalake/pod-datalake/src/s3/bucket.ts | 31 ++++++++++------ 5 files changed, 59 insertions(+), 33 deletions(-) diff --git a/services/datalake/pod-datalake/src/datalake/datalake.ts b/services/datalake/pod-datalake/src/datalake/datalake.ts index c1e4cf599f..5b3898fd88 100644 --- a/services/datalake/pod-datalake/src/datalake/datalake.ts +++ b/services/datalake/pod-datalake/src/datalake/datalake.ts @@ -115,7 +115,7 @@ export class DatalakeImpl implements Datalake { const events = Array.isArray(name) ? name.map((n) => blobEvents.deleted(n)) : [blobEvents.deleted(name)] await this.producer.send(workspace, events) } catch (err) { - ctx.error('failed to send blob deleted event', { err }) + ctx.error('failed to send blob deleted event', { workspace, name, err }) } } @@ -155,7 +155,7 @@ export class DatalakeImpl implements Datalake { : blobEvents.created(name, { contentType, lastModified, size, etag: hash }) await this.producer.send(workspace, [event]) } catch (err) { - ctx.error('failed to send blob created event', { err }) + ctx.error('failed to send blob created event', { workspace, name, err }) } return { name, size, contentType, lastModified, etag: hash } @@ -176,7 +176,7 @@ export class DatalakeImpl implements Datalake { : blobEvents.created(name, { contentType, lastModified, size, etag: hash }) await this.producer.send(workspace, [event]) } catch (err) { - ctx.error('failed to send blob created event', { err }) + ctx.error('failed to send blob created event', { workspace, name, err }) } return { name, size, contentType, lastModified, etag: hash } @@ -215,7 +215,7 @@ export class DatalakeImpl implements Datalake { : blobEvents.created(name, { contentType, lastModified, size, etag: hash }) await this.producer.send(workspace, [event]) } catch (err) { - ctx.error('failed to send blob created event', { err }) + ctx.error('failed to send blob created event', { workspace, name, err }) } return { name, size, contentType, lastModified, etag: hash } diff --git a/services/datalake/pod-datalake/src/datalake/db.ts b/services/datalake/pod-datalake/src/datalake/db.ts index 7d45011c28..2ad1c8758b 100644 --- a/services/datalake/pod-datalake/src/datalake/db.ts +++ b/services/datalake/pod-datalake/src/datalake/db.ts @@ -471,47 +471,58 @@ export class LoggedDB implements BlobDB { ) {} async getData (ctx: MeasureContext, dataId: BlobDataId): Promise { - return await ctx.with('db.getData', {}, () => this.db.getData(this.ctx, dataId)) + const params = { location: dataId.location } + return await ctx.with('db.getData', {}, () => this.db.getData(this.ctx, dataId), params) } async getBlob (ctx: MeasureContext, blobId: BlobId): Promise { - return await ctx.with('db.getBlob', {}, () => this.db.getBlob(this.ctx, blobId)) + const params = { workspace: blobId.workspace } + return await ctx.with('db.getBlob', {}, () => this.db.getBlob(this.ctx, blobId), params) } async listBlobs (ctx: MeasureContext, workspace: string, options: ListBlobOptions): Promise { - return await ctx.with('db.listBlobs', {}, () => this.db.listBlobs(this.ctx, workspace, options)) + const params = { workspace } + return await ctx.with('db.listBlobs', {}, () => this.db.listBlobs(this.ctx, workspace, options), params) } async createData (ctx: MeasureContext, data: BlobDataRecord): Promise { - await ctx.with('db.createData', {}, () => this.db.createData(this.ctx, data)) + const params = { type: data.type } + await ctx.with('db.createData', {}, () => this.db.createData(this.ctx, data), params) } async createBlob (ctx: MeasureContext, blob: Omit): Promise { - await ctx.with('db.createBlob', {}, () => this.db.createBlob(this.ctx, blob)) + const params = { workspace: blob.workspace, location: blob.location } + await ctx.with('db.createBlob', {}, () => this.db.createBlob(this.ctx, blob), params) } async createBlobData (ctx: MeasureContext, data: BlobWithDataRecord): Promise { - await ctx.with('db.createBlobData', {}, () => this.db.createBlobData(this.ctx, data)) + const params = { workspace: data.workspace, location: data.location, type: data.type } + await ctx.with('db.createBlobData', {}, () => this.db.createBlobData(this.ctx, data), params) } async deleteBlobList (ctx: MeasureContext, blobs: BlobIds): Promise { - await ctx.with('db.deleteBlobList', {}, () => this.db.deleteBlobList(this.ctx, blobs)) + const params = { workspace: blobs.workspace } + await ctx.with('db.deleteBlobList', {}, () => this.db.deleteBlobList(this.ctx, blobs), params) } async deleteBlob (ctx: MeasureContext, blob: BlobId): Promise { - await ctx.with('db.deleteBlob', {}, () => this.db.deleteBlob(this.ctx, blob)) + const params = { workspace: blob.workspace } + await ctx.with('db.deleteBlob', {}, () => this.db.deleteBlob(this.ctx, blob), params) } async getMeta (ctx: MeasureContext, blobId: BlobId): Promise { - return await this.ctx.with('db.getMeta', {}, () => this.db.getMeta(ctx, blobId)) + const params = { workspace: blobId.workspace } + return await this.ctx.with('db.getMeta', {}, () => this.db.getMeta(ctx, blobId), params) } async setMeta (ctx: MeasureContext, blobId: BlobId, meta: BlobMeta): Promise { - await this.ctx.with('db.setMeta', {}, () => this.db.setMeta(ctx, blobId, meta)) + const params = { workspace: blobId.workspace } + await this.ctx.with('db.setMeta', {}, () => this.db.setMeta(ctx, blobId, meta), params) } async setParent (ctx: MeasureContext, blob: BlobId, parent: BlobId | null): Promise { - await ctx.with('db.setParent', {}, () => this.db.setParent(this.ctx, blob, parent)) + const params = { workspace: blob.workspace } + await ctx.with('db.setParent', {}, () => this.db.setParent(this.ctx, blob, parent), params) } async getStats (ctx: MeasureContext): Promise { @@ -519,7 +530,8 @@ export class LoggedDB implements BlobDB { } async getWorkspaceStats (ctx: MeasureContext, workspace: string): Promise { - return await ctx.with('db.getWorkspaceStats', {}, () => this.db.getWorkspaceStats(this.ctx, workspace)) + const params = { workspace } + return await ctx.with('db.getWorkspaceStats', {}, () => this.db.getWorkspaceStats(this.ctx, workspace), params) } } diff --git a/services/datalake/pod-datalake/src/handlers/blob.ts b/services/datalake/pod-datalake/src/handlers/blob.ts index 7da529c341..0061591cb6 100644 --- a/services/datalake/pod-datalake/src/handlers/blob.ts +++ b/services/datalake/pod-datalake/src/handlers/blob.ts @@ -161,7 +161,7 @@ export async function handleBlobDelete ( res.status(204).send() } catch (error: any) { Analytics.handleError(error) - ctx.error('failed to delete blob', { error }) + ctx.error('failed to delete blob', { workspace, name, error }) res.status(500).send() } } @@ -182,7 +182,7 @@ export async function handleBlobDeleteList ( res.status(204).send() } catch (error: any) { Analytics.handleError(error) - ctx.error('failed to delete blobs', { error }) + ctx.error('failed to delete blobs', { workspace, names: body.names, error }) res.status(500).send() } } @@ -224,7 +224,7 @@ export async function handleBlobSetParent ( res.status(204).send() } catch (error: any) { Analytics.handleError(error) - ctx.error('failed to delete blob', { error }) + ctx.error('failed to delete blob', { workspace, name, error }) res.status(500).send() } } diff --git a/services/datalake/pod-datalake/src/handlers/image.ts b/services/datalake/pod-datalake/src/handlers/image.ts index 5e56c60fa7..876fb70443 100644 --- a/services/datalake/pod-datalake/src/handlers/image.ts +++ b/services/datalake/pod-datalake/src/handlers/image.ts @@ -132,9 +132,14 @@ export async function handleImageGet ( await writeTempFile(tmpFile, blob.body) try { - const { contentType } = await ctx.with('sharp', {}, () => { - return runPipeline(tmpFile, outFile, { format, width, height, fit }) - }) + const { contentType } = await ctx.with( + 'sharp', + { format }, + () => { + return runPipeline(tmpFile, outFile, { format, width, height, fit }) + }, + { fit, width, height, size: blob.size } + ) res.setHeader('Content-Type', contentType) res.setHeader('Cache-Control', cacheControl) diff --git a/services/datalake/pod-datalake/src/s3/bucket.ts b/services/datalake/pod-datalake/src/s3/bucket.ts index 0ff6a07482..00af00130b 100644 --- a/services/datalake/pod-datalake/src/s3/bucket.ts +++ b/services/datalake/pod-datalake/src/s3/bucket.ts @@ -63,8 +63,11 @@ class S3BucketImpl implements S3Bucket { async head (ctx: MeasureContext, key: string): Promise { try { - const result = await ctx.with('s3.headObject', {}, () => - this.client.headObject({ Bucket: this.bucket, Key: key }) + const result = await ctx.with( + 's3.headObject', + {}, + () => this.client.headObject({ Bucket: this.bucket, Key: key }), + { bucket: this.bucket } ) return { @@ -77,7 +80,7 @@ class S3BucketImpl implements S3Bucket { } } catch (err: any) { if (err?.$metadata?.httpStatusCode !== 404) { - ctx.warn('no object found', { error: err, key }) + ctx.warn('no object found', { error: err, bucket: this.bucket, key }) } return null } @@ -87,7 +90,7 @@ class S3BucketImpl implements S3Bucket { try { const command = { Bucket: this.bucket, Key: key, Range: options?.range } - const result = await ctx.with('s3.getObject', {}, () => this.client.getObject(command)) + const result = await ctx.with('s3.getObject', {}, () => this.client.getObject(command), { bucket: this.bucket }) if (result.Body === undefined) { return null @@ -115,7 +118,7 @@ class S3BucketImpl implements S3Bucket { } } catch (err: any) { if (err?.$metadata?.httpStatusCode !== 404) { - ctx.warn('no object found', { error: err, key }) + ctx.warn('no object found', { error: err, bucket: this.bucket, key }) } return null } @@ -140,7 +143,7 @@ class S3BucketImpl implements S3Bucket { } if (Buffer.isBuffer(body)) { - const result = await ctx.with('s3.putObject', {}, () => this.client.putObject(command)) + const result = await ctx.with('s3.putObject', {}, () => this.client.putObject(command), { bucket: this.bucket }) return { key, @@ -158,7 +161,7 @@ class S3BucketImpl implements S3Bucket { leavePartsOnError: false }) - const result = await ctx.with('s3.upload', {}, () => upload.done()) + const result = await ctx.with('s3.upload', {}, () => upload.done(), { bucket: this.bucket }) return { key, @@ -190,7 +193,9 @@ class S3BucketImpl implements S3Bucket { } } - const result = await ctx.with('s3.createMultipartUpload', {}, () => this.client.createMultipartUpload(command)) + const result = await ctx.with('s3.createMultipartUpload', {}, () => this.client.createMultipartUpload(command), { + bucket: this.bucket + }) if (result.UploadId === undefined) { throw new Error('failed to create multipart upload') } @@ -215,7 +220,7 @@ class S3BucketImpl implements S3Bucket { UploadId: multipart.uploadId, PartNumber: options.partNumber } - const result = await ctx.with('s3.uploadPart', {}, () => this.client.uploadPart(command)) + const result = await ctx.with('s3.uploadPart', {}, () => this.client.uploadPart(command), { bucket: this.bucket }) return { etag: result.ETag ?? '', partNumber: options.partNumber @@ -241,7 +246,9 @@ class S3BucketImpl implements S3Bucket { }) } } - await ctx.with('s3.completeMultipartUpload', {}, () => this.client.completeMultipartUpload(command)) + await ctx.with('s3.completeMultipartUpload', {}, () => this.client.completeMultipartUpload(command), { + bucket: this.bucket + }) } async abortMultipartUpload (ctx: MeasureContext, key: string, multipart: S3MultipartUpload): Promise { @@ -250,6 +257,8 @@ class S3BucketImpl implements S3Bucket { Key: key, UploadId: multipart.uploadId } - await ctx.with('s3.abortMultipartUpload', {}, () => this.client.abortMultipartUpload(command)) + await ctx.with('s3.abortMultipartUpload', {}, () => this.client.abortMultipartUpload(command), { + bucket: this.bucket + }) } }