fix: log more stats in datalake (#9624)

Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>

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 }

services/datalake/pod-datalake/src/datalake/db.ts

@@ -471,47 +471,58 @@ export class LoggedDB implements BlobDB {
   ) {}
 
   async getData (ctx: MeasureContext, dataId: BlobDataId): Promise<BlobDataRecord | null> {
-    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<BlobWithDataRecord | null> {
-    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<ListBlobResult> {
-    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<void> {
-    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<BlobRecord, 'filename'>): Promise<void> {
-    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<void> {
-    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<void> {
-    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<void> {
-    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<BlobMeta | null> {
-    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<void> {
-    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<void> {
-    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<StatsResult> {
@@ -519,7 +530,8 @@ export class LoggedDB implements BlobDB {
   }
 
   async getWorkspaceStats (ctx: MeasureContext, workspace: string): Promise<WorkspaceStatsResult> {
-    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)
   }
 }
 

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()
   }
 }

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)

services/datalake/pod-datalake/src/s3/bucket.ts

@@ -63,8 +63,11 @@ class S3BucketImpl implements S3Bucket {
 
   async head (ctx: MeasureContext, key: string): Promise<S3Object | null> {
     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<void> {
@@ -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
+    })
   }
 }