fix: enhance datalake performance logging (#8197)

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

services/datalake/pod-datalake/package.json

@@ -56,6 +56,8 @@
     "@types/sharp": "~0.32.0"
   },
   "dependencies": {
+    "@hcengineering/analytics": "^0.6.0",
+    "@hcengineering/analytics-service": "^0.6.0",
     "@hcengineering/server-token": "^0.6.11",
     "@hcengineering/server-core": "^0.6.1",
     "@hcengineering/server-client": "^0.6.0",

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

@@ -24,7 +24,6 @@ import { type S3Bucket } from '../s3'
 
 export class DatalakeImpl implements Datalake {
   constructor (
-    private readonly ctx: MeasureContext,
     private readonly db: BlobDB,
     private readonly buckets: Partial<Record<Location, S3Bucket>>,
     private readonly options: {
@@ -32,8 +31,8 @@ export class DatalakeImpl implements Datalake {
     }
   ) {}
 
-  async list (workspace: string, cursor?: string, limit?: number): Promise<BlobList> {
-    const blobs = await this.db.listBlobs(workspace, cursor, limit)
+  async list (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number): Promise<BlobList> {
+    const blobs = await this.db.listBlobs(ctx, workspace, cursor, limit)
 
     return {
       cursor: blobs.cursor,
@@ -44,15 +43,15 @@ export class DatalakeImpl implements Datalake {
     }
   }
 
-  async head (workspace: string, name: string): Promise<BlobHead | null> {
-    const { bucket } = await this.selectStorage(workspace)
+  async head (ctx: MeasureContext, workspace: string, name: string): Promise<BlobHead | null> {
+    const { bucket } = await this.selectStorage(ctx, workspace)
 
-    const blob = await this.db.getBlob({ workspace, name })
+    const blob = await this.db.getBlob(ctx, { workspace, name })
     if (blob === null) {
       return null
     }
 
-    const head = await bucket.head(blob.filename)
+    const head = await bucket.head(ctx, blob.filename)
     if (head == null) {
       return null
     }
@@ -67,16 +66,21 @@ export class DatalakeImpl implements Datalake {
     }
   }
 
-  async get (workspace: string, name: string, options: { range?: string }): Promise<BlobBody | null> {
-    const { bucket } = await this.selectStorage(workspace)
+  async get (
+    ctx: MeasureContext,
+    workspace: string,
+    name: string,
+    options: { range?: string }
+  ): Promise<BlobBody | null> {
+    const { bucket } = await this.selectStorage(ctx, workspace)
 
-    const blob = await this.db.getBlob({ workspace, name })
+    const blob = await this.db.getBlob(ctx, { workspace, name })
     if (blob === null) {
       return null
     }
 
     const range = options.range
-    const object = await bucket.get(blob.filename, { range })
+    const object = await bucket.get(ctx, blob.filename, { range })
     if (object == null) {
       return null
     }
@@ -95,15 +99,16 @@ export class DatalakeImpl implements Datalake {
     }
   }
 
-  async delete (workspace: string, name: string | string[]): Promise<void> {
+  async delete (ctx: MeasureContext, workspace: string, name: string | string[]): Promise<void> {
     if (Array.isArray(name)) {
-      await this.db.deleteBlobList({ workspace, names: name })
+      await this.db.deleteBlobList(ctx, { workspace, names: name })
     } else {
-      await this.db.deleteBlob({ workspace, name })
+      await this.db.deleteBlob(ctx, { workspace, name })
     }
   }
 
   async put (
+    ctx: MeasureContext,
     workspace: string,
     name: string,
     sha256: string,
@@ -113,9 +118,9 @@ export class DatalakeImpl implements Datalake {
     const cacheControl = options.cacheControl ?? this.options.cacheControl
 
     const { size, contentType, lastModified } = options
-    const { location, bucket } = await this.selectStorage(workspace)
+    const { location, bucket } = await this.selectStorage(ctx, workspace)
 
-    const blob = await this.db.getBlob({ workspace, name })
+    const blob = await this.db.getBlob(ctx, { workspace, name })
 
     const hash = digestToUUID(sha256)
     const filename = hash
@@ -125,11 +130,11 @@ export class DatalakeImpl implements Datalake {
       return { name, size, contentType, lastModified, etag: hash }
     }
 
-    const data = await this.db.getData({ hash, location })
+    const data = await this.db.getData(ctx, { hash, location })
 
     if (data !== null) {
       // Lucky boy, nothing to upload, use existing blob
-      await this.db.createBlob({ workspace, name, hash, location })
+      await this.db.createBlob(ctx, { workspace, name, hash, location })
       return { name, size, contentType, lastModified, etag: hash }
     } else {
       const putOptions = {
@@ -138,16 +143,16 @@ export class DatalakeImpl implements Datalake {
         cacheControl,
         lastModified
       }
-      await bucket.put(filename, body, putOptions)
-      await this.db.createBlobData({ workspace, name, hash, location, filename, size, type: contentType })
+      await bucket.put(ctx, filename, body, putOptions)
+      await this.db.createBlobData(ctx, { workspace, name, hash, location, filename, size, type: contentType })
       return { name, size, contentType, lastModified, etag: hash }
     }
   }
 
-  async create (workspace: string, name: string, filename: string): Promise<BlobHead | null> {
-    const { location, bucket } = await this.selectStorage(workspace)
+  async create (ctx: MeasureContext, workspace: string, name: string, filename: string): Promise<BlobHead | null> {
+    const { location, bucket } = await this.selectStorage(ctx, workspace)
 
-    const head = await bucket.head(filename)
+    const head = await bucket.head(ctx, filename)
     if (head == null) {
       return null
     }
@@ -157,21 +162,21 @@ export class DatalakeImpl implements Datalake {
     const contentType = head.contentType ?? 'application/octet-stream'
     const lastModified = head.lastModified
 
-    const data = await this.db.getData({ hash, location })
+    const data = await this.db.getData(ctx, { hash, location })
     if (data !== null) {
-      await Promise.all([bucket.delete(filename), this.db.createBlob({ workspace, name, hash, location })])
+      await Promise.all([bucket.delete(ctx, filename), this.db.createBlob(ctx, { workspace, name, hash, location })])
     } else {
-      await this.db.createBlobData({ workspace, name, hash, location, filename, size, type: contentType })
+      await this.db.createBlobData(ctx, { workspace, name, hash, location, filename, size, type: contentType })
     }
 
     return { name, size, contentType, lastModified, etag: hash }
   }
 
-  async setParent (workspace: string, name: string, parent: string | null): Promise<void> {
-    await this.db.setParent({ workspace, name }, parent !== null ? { workspace, name: parent } : null)
+  async setParent (ctx: MeasureContext, workspace: string, name: string, parent: string | null): Promise<void> {
+    await this.db.setParent(ctx, { workspace, name }, parent !== null ? { workspace, name: parent } : null)
   }
 
-  async selectStorage (workspace: string): Promise<BlobStorage> {
+  async selectStorage (ctx: MeasureContext, workspace: string): Promise<BlobStorage> {
     const location = this.selectLocation(workspace)
     const bucket = this.buckets[location]
     if (bucket == null) {

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

@@ -87,21 +87,21 @@ export function createDb (ctx: MeasureContext, connectionString: string): BlobDB
   })
 
   const dbAdapter = new PostgresDBAdapter(sql)
-  return new LoggedDB(new PostgresDB(new RetryDBAdapter(dbAdapter, { retries: 5 })), ctx)
+  return new LoggedDB(ctx, new PostgresDB(new RetryDBAdapter(dbAdapter, { retries: 5 })))
 }
 
 export interface BlobDB {
-  getData: (dataId: BlobDataId) => Promise<BlobDataRecord | null>
-  getBlob: (blobId: BlobId) => Promise<BlobWithDataRecord | null>
-  listBlobs: (workspace: string, cursor?: string, limit?: number) => Promise<ListBlobResult>
-  createData: (data: BlobDataRecord) => Promise<void>
-  createBlob: (blob: Omit<BlobRecord, 'filename'>) => Promise<void>
-  createBlobData: (blob: BlobWithDataRecord) => Promise<void>
-  deleteBlob: (blob: BlobId) => Promise<void>
-  setParent: (blob: BlobId, parent: BlobId | null) => Promise<void>
-  deleteBlobList: (list: BlobIds) => Promise<void>
-  getStats: () => Promise<StatsResult>
-  getWorkspaceStats: (workspace: string) => Promise<WorkspaceStatsResult>
+  getData: (ctx: MeasureContext, dataId: BlobDataId) => Promise<BlobDataRecord | null>
+  getBlob: (ctx: MeasureContext, blobId: BlobId) => Promise<BlobWithDataRecord | null>
+  listBlobs: (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number) => Promise<ListBlobResult>
+  createData: (ctx: MeasureContext, data: BlobDataRecord) => Promise<void>
+  createBlob: (ctx: MeasureContext, blob: Omit<BlobRecord, 'filename'>) => Promise<void>
+  createBlobData: (ctx: MeasureContext, blob: BlobWithDataRecord) => Promise<void>
+  deleteBlob: (ctx: MeasureContext, blob: BlobId) => Promise<void>
+  setParent: (ctx: MeasureContext, blob: BlobId, parent: BlobId | null) => Promise<void>
+  deleteBlobList: (ctx: MeasureContext, list: BlobIds) => Promise<void>
+  getStats: (ctx: MeasureContext) => Promise<StatsResult>
+  getWorkspaceStats: (ctx: MeasureContext, workspace: string) => Promise<WorkspaceStatsResult>
 }
 
 interface DBAdapter {
@@ -131,7 +131,7 @@ class PostgresDBAdapter implements DBAdapter {
 export class PostgresDB implements BlobDB {
   constructor (private readonly db: DBAdapter) {}
 
-  async getData (dataId: BlobDataId): Promise<BlobDataRecord | null> {
+  async getData (ctx: MeasureContext, dataId: BlobDataId): Promise<BlobDataRecord | null> {
     const { hash, location } = dataId
 
     const rows = await this.db.execute<BlobDataRecord[]>(
@@ -146,7 +146,7 @@ export class PostgresDB implements BlobDB {
     return rows.length > 0 ? rows[0] : null
   }
 
-  async deleteBlobList (blobList: BlobIds): Promise<void> {
+  async deleteBlobList (ctx: MeasureContext, blobList: BlobIds): Promise<void> {
     const { workspace, names } = blobList
 
     await this.db.execute(
@@ -159,7 +159,7 @@ export class PostgresDB implements BlobDB {
     )
   }
 
-  async getBlob (blobId: BlobId): Promise<BlobWithDataRecord | null> {
+  async getBlob (ctx: MeasureContext, blobId: BlobId): Promise<BlobWithDataRecord | null> {
     const { workspace, name } = blobId
 
     const rows = await this.db.execute<BlobWithDataRecord[]>(
@@ -179,7 +179,7 @@ export class PostgresDB implements BlobDB {
     return null
   }
 
-  async listBlobs (workspace: string, cursor?: string, limit?: number): Promise<ListBlobResult> {
+  async listBlobs (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number): Promise<ListBlobResult> {
     cursor = cursor ?? ''
     limit = Math.min(limit ?? 100, 1000)
 
@@ -201,7 +201,7 @@ export class PostgresDB implements BlobDB {
     }
   }
 
-  async createBlob (blob: Omit<BlobRecord, 'filename'>): Promise<void> {
+  async createBlob (ctx: MeasureContext, blob: Omit<BlobRecord, 'filename'>): Promise<void> {
     const { workspace, name, hash, location, parent } = blob
 
     await this.db.execute(
@@ -213,7 +213,7 @@ export class PostgresDB implements BlobDB {
     )
   }
 
-  async createData (data: BlobDataRecord): Promise<void> {
+  async createData (ctx: MeasureContext, data: BlobDataRecord): Promise<void> {
     const { hash, location, filename, size, type } = data
 
     await this.db.execute(
@@ -225,7 +225,7 @@ export class PostgresDB implements BlobDB {
     )
   }
 
-  async createBlobData (data: BlobWithDataRecord): Promise<void> {
+  async createBlobData (ctx: MeasureContext, data: BlobWithDataRecord): Promise<void> {
     const { workspace, name, hash, location, parent, filename, size, type } = data
 
     await this.db.execute(
@@ -245,7 +245,7 @@ export class PostgresDB implements BlobDB {
     )
   }
 
-  async deleteBlob (blob: BlobId): Promise<void> {
+  async deleteBlob (ctx: MeasureContext, blob: BlobId): Promise<void> {
     const { workspace, name } = blob
 
     const blobs = new Set<BlobId['name']>()
@@ -284,7 +284,7 @@ export class PostgresDB implements BlobDB {
     )
   }
 
-  async setParent (blob: BlobId, parent: BlobId | null): Promise<void> {
+  async setParent (ctx: MeasureContext, blob: BlobId, parent: BlobId | null): Promise<void> {
     const { workspace, name } = blob
 
     await this.db.execute(
@@ -297,7 +297,7 @@ export class PostgresDB implements BlobDB {
     )
   }
 
-  async getStats (): Promise<StatsResult> {
+  async getStats (ctx: MeasureContext): Promise<StatsResult> {
     const blobStatsRows = await this.db.execute(`
       SELECT count(distinct b.workspace) as workspaces, count(1) as count, sum(d.size) as size
       FROM blob.blob b
@@ -326,7 +326,7 @@ export class PostgresDB implements BlobDB {
     }
   }
 
-  async getWorkspaceStats (workspace: string): Promise<WorkspaceStatsResult> {
+  async getWorkspaceStats (ctx: MeasureContext, workspace: string): Promise<WorkspaceStatsResult> {
     const rows = await this.db.execute(
       `
       SELECT count(1) as count, sum(d.size) as size
@@ -348,52 +348,52 @@ export class PostgresDB implements BlobDB {
 
 export class LoggedDB implements BlobDB {
   constructor (
-    private readonly db: BlobDB,
-    private readonly ctx: MeasureContext
+    private readonly ctx: MeasureContext,
+    private readonly db: BlobDB
   ) {}
 
-  async getData (dataId: BlobDataId): Promise<BlobDataRecord | null> {
-    return await this.ctx.with('db.getData', {}, () => this.db.getData(dataId))
+  async getData (ctx: MeasureContext, dataId: BlobDataId): Promise<BlobDataRecord | null> {
+    return await ctx.with('db.getData', {}, () => this.db.getData(this.ctx, dataId))
   }
 
-  async getBlob (blobId: BlobId): Promise<BlobWithDataRecord | null> {
-    return await this.ctx.with('db.getBlob', {}, () => this.db.getBlob(blobId))
+  async getBlob (ctx: MeasureContext, blobId: BlobId): Promise<BlobWithDataRecord | null> {
+    return await ctx.with('db.getBlob', {}, () => this.db.getBlob(this.ctx, blobId))
   }
 
-  async listBlobs (workspace: string, cursor?: string, limit?: number): Promise<ListBlobResult> {
-    return await this.ctx.with('db.listBlobs', {}, () => this.db.listBlobs(workspace, cursor, limit))
+  async listBlobs (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number): Promise<ListBlobResult> {
+    return await ctx.with('db.listBlobs', {}, () => this.db.listBlobs(this.ctx, workspace, cursor, limit))
   }
 
-  async createData (data: BlobDataRecord): Promise<void> {
-    await this.ctx.with('db.createData', {}, () => this.db.createData(data))
+  async createData (ctx: MeasureContext, data: BlobDataRecord): Promise<void> {
+    await ctx.with('db.createData', {}, () => this.db.createData(this.ctx, data))
   }
 
-  async createBlob (blob: Omit<BlobRecord, 'filename'>): Promise<void> {
-    await this.ctx.with('db.createBlob', {}, () => this.db.createBlob(blob))
+  async createBlob (ctx: MeasureContext, blob: Omit<BlobRecord, 'filename'>): Promise<void> {
+    await ctx.with('db.createBlob', {}, () => this.db.createBlob(this.ctx, blob))
   }
 
-  async createBlobData (data: BlobWithDataRecord): Promise<void> {
-    await this.ctx.with('db.createBlobData', {}, () => this.db.createBlobData(data))
+  async createBlobData (ctx: MeasureContext, data: BlobWithDataRecord): Promise<void> {
+    await ctx.with('db.createBlobData', {}, () => this.db.createBlobData(this.ctx, data))
   }
 
-  async deleteBlobList (blobs: BlobIds): Promise<void> {
-    await this.ctx.with('db.deleteBlobList', {}, () => this.db.deleteBlobList(blobs))
+  async deleteBlobList (ctx: MeasureContext, blobs: BlobIds): Promise<void> {
+    await ctx.with('db.deleteBlobList', {}, () => this.db.deleteBlobList(this.ctx, blobs))
   }
 
-  async deleteBlob (blob: BlobId): Promise<void> {
-    await this.ctx.with('db.deleteBlob', {}, () => this.db.deleteBlob(blob))
+  async deleteBlob (ctx: MeasureContext, blob: BlobId): Promise<void> {
+    await ctx.with('db.deleteBlob', {}, () => this.db.deleteBlob(this.ctx, blob))
   }
 
-  async setParent (blob: BlobId, parent: BlobId | null): Promise<void> {
-    await this.ctx.with('db.setParent', {}, () => this.db.setParent(blob, parent))
+  async setParent (ctx: MeasureContext, blob: BlobId, parent: BlobId | null): Promise<void> {
+    await ctx.with('db.setParent', {}, () => this.db.setParent(this.ctx, blob, parent))
   }
 
-  async getStats (): Promise<StatsResult> {
-    return await this.ctx.with('db.getStats', {}, () => this.db.getStats())
+  async getStats (ctx: MeasureContext): Promise<StatsResult> {
+    return await ctx.with('db.getStats', {}, () => this.db.getStats(this.ctx))
   }
 
-  async getWorkspaceStats (workspace: string): Promise<WorkspaceStatsResult> {
-    return await this.ctx.with('db.getWorkspaceStats', {}, () => this.db.getWorkspaceStats(workspace))
+  async getWorkspaceStats (ctx: MeasureContext, workspace: string): Promise<WorkspaceStatsResult> {
+    return await ctx.with('db.getWorkspaceStats', {}, () => this.db.getWorkspaceStats(this.ctx, workspace))
   }
 }
 

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 //
 
+import { MeasureContext } from '@hcengineering/core'
 import { type Readable } from 'stream'
 import { S3Bucket } from '../s3'
 
@@ -47,19 +48,20 @@ export interface BlobStorage {
 }
 
 export interface Datalake {
-  list: (workspace: string, cursor?: string, limit?: number) => Promise<BlobList>
-  head: (workspace: string, name: string) => Promise<BlobHead | null>
-  get: (workspace: string, name: string, options: { range?: string }) => Promise<BlobBody | null>
-  delete: (workspace: string, name: string | string[]) => Promise<void>
+  list: (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number) => Promise<BlobList>
+  head: (ctx: MeasureContext, workspace: string, name: string) => Promise<BlobHead | null>
+  get: (ctx: MeasureContext, workspace: string, name: string, options: { range?: string }) => Promise<BlobBody | null>
+  delete: (ctx: MeasureContext, workspace: string, name: string | string[]) => Promise<void>
   put: (
+    ctx: MeasureContext,
     workspace: string,
     name: string,
     sha256: string,
     body: Buffer | Readable,
     options: Omit<BlobHead, 'name' | 'etag'>
   ) => Promise<BlobHead>
-  create: (workspace: string, name: string, filename: string) => Promise<BlobHead | null>
+  create: (ctx: MeasureContext, workspace: string, name: string, filename: string) => Promise<BlobHead | null>
 
-  setParent: (workspace: string, name: string, parent: string | null) => Promise<void>
-  selectStorage: (workspace: string) => Promise<BlobStorage>
+  setParent: (ctx: MeasureContext, workspace: string, name: string, parent: string | null) => Promise<void>
+  selectStorage: (ctx: MeasureContext, workspace: string) => Promise<BlobStorage>
 }

services/datalake/pod-datalake/src/handlers/blob.ts

@@ -40,7 +40,7 @@ export async function handleBlobList (
   const cursor = req.query.cursor as string
   const limit = extractIntParam(req.query.limit as string)
 
-  const blobs = await datalake.list(workspace, cursor, limit)
+  const blobs = await datalake.list(ctx, workspace, cursor, limit)
   res.status(200).json(blobs)
 }
 
@@ -54,7 +54,7 @@ export async function handleBlobGet (
 
   const range = req.headers.range
 
-  const blob = await datalake.get(workspace, name, { range })
+  const blob = await datalake.get(ctx, workspace, name, { range })
   if (blob == null) {
     res.status(404).send()
     return
@@ -107,7 +107,7 @@ export async function handleBlobHead (
 ): Promise<void> {
   const { workspace, name, filename } = req.params
 
-  const head = await datalake.head(workspace, name)
+  const head = await datalake.head(ctx, workspace, name)
   if (head == null) {
     res.status(404).send()
     return
@@ -134,7 +134,7 @@ export async function handleBlobDelete (
   const { workspace, name } = req.params
 
   try {
-    await datalake.delete(workspace, name)
+    await datalake.delete(ctx, workspace, name)
     res.status(204).send()
   } catch (err: any) {
     const message = err instanceof Error ? err.message : String(err)
@@ -153,7 +153,7 @@ export async function handleBlobDeleteList (
   const body = req.body.names as DeleteBlobsRequest
 
   try {
-    await datalake.delete(workspace, body.names)
+    await datalake.delete(ctx, workspace, body.names)
     res.status(204).send()
   } catch (err: any) {
     const message = err instanceof Error ? err.message : String(err)
@@ -172,7 +172,7 @@ export async function handleBlobSetParent (
   const { parent } = (await req.body) as BlobParentRequest
 
   try {
-    await datalake.setParent(workspace, name, parent)
+    await datalake.setParent(ctx, workspace, name, parent)
     res.status(204).send()
   } catch (err: any) {
     const message = err instanceof Error ? err.message : String(err)
@@ -220,7 +220,7 @@ export async function handleUploadFormData (
       const data = file.tempFilePath !== undefined ? fs.createReadStream(file.tempFilePath) : file.data
 
       try {
-        const metadata = await datalake.put(workspace, name, sha256, data, {
+        const metadata = await datalake.put(ctx, workspace, name, sha256, data, {
           size,
           contentType,
           lastModified: Date.now()

services/datalake/pod-datalake/src/handlers/image.ts

@@ -82,7 +82,7 @@ export async function handleImageGet (
   const accept = req.headers.accept ?? 'image/*'
   const image = parseImageTransform(accept, transform)
 
-  const blob = await datalake.get(workspace, name, {})
+  const blob = await datalake.get(ctx, workspace, name, {})
   if (blob == null) {
     res.status(404).send()
     return
@@ -151,7 +151,7 @@ export async function handleImageGet (
     res.setHeader('Content-Type', contentType)
     res.setHeader('Cache-Control', cacheControl)
 
-    await ctx.with('resize', {}, () => pipeline.toFile(outFile))
+    await ctx.with('sharp', {}, () => pipeline.toFile(outFile))
     pipeline.destroy()
 
     createReadStream(outFile).pipe(res)

services/datalake/pod-datalake/src/handlers/multipart.ts

@@ -40,14 +40,14 @@ export async function handleMultipartUploadStart (
 ): Promise<void> {
   const { workspace, name } = req.params
 
-  const { bucket } = await datalake.selectStorage(workspace)
+  const { bucket } = await datalake.selectStorage(ctx, workspace)
 
   const contentType = req.headers['content-type'] ?? 'application/octet-stream'
   const lastModifiedHeader = req.headers['last-modified']
   const lastModified =
     lastModifiedHeader != null ? new Date(lastModifiedHeader).toISOString() : new Date().toISOString()
 
-  const { uploadId } = await bucket.createMultipartUpload(name, { contentType, cacheControl, lastModified })
+  const { uploadId } = await bucket.createMultipartUpload(ctx, name, { contentType, cacheControl, lastModified })
   res.status(200).json({ key: name, uploadId })
 }
 
@@ -72,9 +72,9 @@ export async function handleMultipartUploadPart (
     return
   }
 
-  const { bucket } = await datalake.selectStorage(workspace)
+  const { bucket } = await datalake.selectStorage(ctx, workspace)
 
-  const part = await bucket.uploadMultipartPart(key, { uploadId }, req.body, {
+  const part = await bucket.uploadMultipartPart(ctx, key, { uploadId }, req.body, {
     partNumber: Number.parseInt(partNumber)
   })
 
@@ -88,7 +88,7 @@ export async function handleMultipartUploadComplete (
   datalake: Datalake
 ): Promise<void> {
   const { workspace, name } = req.params
-  const { bucket } = await datalake.selectStorage(workspace)
+  const { bucket } = await datalake.selectStorage(ctx, workspace)
 
   const uploadId = req.query.uploadId as string
   if (typeof uploadId !== 'string') {
@@ -97,8 +97,8 @@ export async function handleMultipartUploadComplete (
   }
 
   const { parts } = req.body as MultipartUploadCompleteRequest
-  await bucket.completeMultipartUpload(name, { uploadId }, parts)
-  const metadata = await datalake.create(workspace, name, name)
+  await bucket.completeMultipartUpload(ctx, name, { uploadId }, parts)
+  const metadata = await datalake.create(ctx, workspace, name, name)
 
   res.status(200).json(metadata)
 }
@@ -117,8 +117,8 @@ export async function handleMultipartUploadAbort (
     return
   }
 
-  const { bucket } = await datalake.selectStorage(workspace)
-  await bucket.abortMultipartUpload(name, { uploadId })
+  const { bucket } = await datalake.selectStorage(ctx, workspace)
+  await bucket.abortMultipartUpload(ctx, name, { uploadId })
 
   res.status(204).send()
 }

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

@@ -25,7 +25,7 @@ export async function handleS3CreateBlobParams (
   datalake: Datalake
 ): Promise<void> {
   const { workspace } = req.params
-  const { location, bucket } = await datalake.selectStorage(workspace)
+  const { location, bucket } = await datalake.selectStorage(ctx, workspace)
   res.status(200).json({ location, bucket: bucket.bucket })
 }
 
@@ -43,7 +43,7 @@ export async function handleS3CreateBlob (
   }
 
   try {
-    await datalake.create(workspace, name, filename)
+    await datalake.create(ctx, workspace, name, filename)
     res.status(200).send()
   } catch (err: any) {
     const error = err instanceof Error ? err.message : String(err)

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

@@ -13,8 +13,13 @@
 // limitations under the License.
 //
 
+import { Analytics } from '@hcengineering/analytics'
+import { configureAnalytics, SplitLogger } from '@hcengineering/analytics-service'
+import { MeasureMetricsContext, newMetrics } from '@hcengineering/core'
 import { setMetadata } from '@hcengineering/platform'
+import { initStatisticsContext } from '@hcengineering/server-core'
 import serverToken from '@hcengineering/server-token'
+import { join } from 'path'
 
 import config from './config'
 import { createServer, listen } from './server'
@@ -26,7 +31,24 @@ const setupMetadata = (): void => {
 export const main = async (): Promise<void> => {
   setupMetadata()
 
-  const { app, close } = createServer(config)
+  configureAnalytics(process.env.SENTRY_DSN, {})
+  Analytics.setTag('application', 'datalake')
+
+  const metricsContext = initStatisticsContext('datalake', {
+    factory: () =>
+      new MeasureMetricsContext(
+        'datalake',
+        {},
+        {},
+        newMetrics(),
+        new SplitLogger('datalake', {
+          root: join(process.cwd(), 'logs'),
+          enableConsole: (process.env.ENABLE_CONSOLE ?? 'true') === 'true'
+        })
+      )
+  })
+
+  const { app, close } = createServer(metricsContext, config)
   const server = listen(app, config.Port)
 
   const shutdown = (): void => {

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 //
 
+import { MeasureContext } from '@hcengineering/core'
 import { S3 } from '@aws-sdk/client-s3'
 import { Upload } from '@aws-sdk/lib-storage'
 import { Readable } from 'stream'
@@ -39,8 +40,9 @@ class S3BucketImpl implements S3Bucket {
     readonly bucket: string
   ) {}
 
-  async head (key: string): Promise<S3Object | null> {
-    const result = await this.client.headObject({ Bucket: this.bucket, Key: key })
+  async head (ctx: MeasureContext, key: string): Promise<S3Object | null> {
+    const result = await ctx.with('s3.headObject', {}, () => this.client.headObject({ Bucket: this.bucket, Key: key }))
+
     return {
       key,
       etag: result.ETag ?? '',
@@ -51,9 +53,11 @@ class S3BucketImpl implements S3Bucket {
     }
   }
 
-  async get (key: string, options?: S3GetOptions): Promise<S3ObjectBody | null> {
+  async get (ctx: MeasureContext, key: string, options?: S3GetOptions): Promise<S3ObjectBody | null> {
     const command = { Bucket: this.bucket, Key: key, Range: options?.range }
-    const result = await this.client.getObject(command)
+
+    const result = await ctx.with('s3.getObject', {}, () => this.client.getObject(command))
+
     if (result.Body === undefined) {
       return null
     }
@@ -80,7 +84,12 @@ class S3BucketImpl implements S3Bucket {
     }
   }
 
-  async put (key: string, body: Readable | Buffer | string, options: S3PutOptions): Promise<S3Object> {
+  async put (
+    ctx: MeasureContext,
+    key: string,
+    body: Readable | Buffer | string,
+    options: S3PutOptions
+  ): Promise<S3Object> {
     const command = {
       Bucket: this.bucket,
       Key: key,
@@ -94,7 +103,7 @@ class S3BucketImpl implements S3Bucket {
     }
 
     if (options.contentLength < 5 * 1024 * 1024) {
-      const result = await this.client.putObject(command)
+      const result = await ctx.with('s3.putObject', {}, () => this.client.putObject(command))
 
       return {
         key,
@@ -112,7 +121,8 @@ class S3BucketImpl implements S3Bucket {
         leavePartsOnError: false
       })
 
-      const result = await upload.done()
+      const result = await ctx.with('s3.upload', {}, () => upload.done())
+
       return {
         key,
         etag: result.ETag ?? '',
@@ -124,11 +134,15 @@ class S3BucketImpl implements S3Bucket {
     }
   }
 
-  async delete (key: string): Promise<void> {
+  async delete (ctx: MeasureContext, key: string): Promise<void> {
     await this.client.deleteObject({ Bucket: this.bucket, Key: key })
   }
 
-  async createMultipartUpload (key: string, options: S3CreateMultipartUploadOptions): Promise<S3MultipartUpload> {
+  async createMultipartUpload (
+    ctx: MeasureContext,
+    key: string,
+    options: S3CreateMultipartUploadOptions
+  ): Promise<S3MultipartUpload> {
     const command = {
       Bucket: this.bucket,
       Key: key,
@@ -139,7 +153,7 @@ class S3BucketImpl implements S3Bucket {
       }
     }
 
-    const result = await this.client.createMultipartUpload(command)
+    const result = await ctx.with('s3.createMultipartUpload', {}, () => this.client.createMultipartUpload(command))
     if (result.UploadId === undefined) {
       throw new Error('failed to create multipart upload')
     }
@@ -150,6 +164,7 @@ class S3BucketImpl implements S3Bucket {
   }
 
   async uploadMultipartPart (
+    ctx: MeasureContext,
     key: string,
     multipart: S3MultipartUpload,
     body: Readable | Buffer | string,
@@ -162,7 +177,7 @@ class S3BucketImpl implements S3Bucket {
       UploadId: multipart.uploadId,
       PartNumber: options.partNumber
     }
-    const result = await this.client.uploadPart(command)
+    const result = await ctx.with('s3.uploadPart', {}, () => this.client.uploadPart(command))
     return {
       etag: result.ETag ?? '',
       partNumber: options.partNumber
@@ -170,6 +185,7 @@ class S3BucketImpl implements S3Bucket {
   }
 
   async completeMultipartUpload (
+    ctx: MeasureContext,
     key: string,
     multipart: S3MultipartUpload,
     parts: S3MultipartUploadPart[]
@@ -187,15 +203,15 @@ class S3BucketImpl implements S3Bucket {
         })
       }
     }
-    await this.client.completeMultipartUpload(command)
+    await ctx.with('s3.completeMultipartUpload', {}, () => this.client.completeMultipartUpload(command))
   }
 
-  async abortMultipartUpload (key: string, multipart: S3MultipartUpload): Promise<void> {
+  async abortMultipartUpload (ctx: MeasureContext, key: string, multipart: S3MultipartUpload): Promise<void> {
     const command = {
       Bucket: this.bucket,
       Key: key,
       UploadId: multipart.uploadId
     }
-    await this.client.abortMultipartUpload(command)
+    await ctx.with('s3.abortMultipartUpload', {}, () => this.client.abortMultipartUpload(command))
   }
 }

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 //
 
+import { MeasureContext } from '@hcengineering/core'
 import { Readable } from 'stream'
 
 export interface S3Object {
@@ -62,19 +63,29 @@ export interface S3UploadPartOptions {
 export interface S3Bucket {
   bucket: string
 
-  head: (key: string) => Promise<S3Object | null>
-  get: (key: string, options?: S3GetOptions) => Promise<S3ObjectBody | null>
-  put: (key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise<S3Object>
-  delete: (key: string) => Promise<void>
+  head: (ctx: MeasureContext, key: string) => Promise<S3Object | null>
+  get: (ctx: MeasureContext, key: string, options?: S3GetOptions) => Promise<S3ObjectBody | null>
+  put: (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise<S3Object>
+  delete: (ctx: MeasureContext, key: string) => Promise<void>
 
   // multipart
-  createMultipartUpload: (key: string, options: S3CreateMultipartUploadOptions) => Promise<S3MultipartUpload>
+  createMultipartUpload: (
+    ctx: MeasureContext,
+    key: string,
+    options: S3CreateMultipartUploadOptions
+  ) => Promise<S3MultipartUpload>
   uploadMultipartPart: (
+    ctx: MeasureContext,
     key: string,
     multipart: S3MultipartUpload,
     body: Readable | Buffer | string,
     options: S3UploadPartOptions
   ) => Promise<S3MultipartUploadPart>
-  completeMultipartUpload: (key: string, multipart: S3MultipartUpload, parts: S3MultipartUploadPart[]) => Promise<void>
-  abortMultipartUpload: (key: string, multipart: S3MultipartUpload) => Promise<void>
+  completeMultipartUpload: (
+    ctx: MeasureContext,
+    key: string,
+    multipart: S3MultipartUpload,
+    parts: S3MultipartUploadPart[]
+  ) => Promise<void>
+  abortMultipartUpload: (ctx: MeasureContext, key: string, multipart: S3MultipartUpload) => Promise<void>
 }

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

@@ -13,8 +13,9 @@
 // limitations under the License.
 //
 
-import { MeasureContext } from '@hcengineering/core'
-import { initStatisticsContext } from '@hcengineering/server-core'
+import { Analytics } from '@hcengineering/analytics'
+import { MeasureContext, metricsAggregate } from '@hcengineering/core'
+import { decodeToken, TokenError } from '@hcengineering/server-token'
 
 import cors from 'cors'
 import express, { type Express, type NextFunction, type Request, type Response } from 'express'
@@ -22,6 +23,7 @@ import fileUpload from 'express-fileupload'
 import { mkdtempSync } from 'fs'
 import { type Server } from 'http'
 import { tmpdir } from 'os'
+import { join } from 'path'
 
 import { cacheControl } from './const'
 import { createDb } from './datalake/db'
@@ -43,12 +45,14 @@ import { DatalakeImpl } from './datalake/datalake'
 import { Config } from './config'
 import { createBucket, S3Bucket } from './s3'
 import { createClient } from './s3/client'
-import { join } from 'path'
+
+const cacheControlNoCache = 'public, no-store, no-cache, must-revalidate, max-age=0'
 
 type AsyncRequestHandler = (ctx: MeasureContext, req: Request, res: Response, datalake: Datalake) => Promise<void>
 
 const handleRequest = async (
   ctx: MeasureContext,
+  name: string,
   datalake: Datalake,
   fn: AsyncRequestHandler,
   req: Request,
@@ -56,22 +60,20 @@ const handleRequest = async (
   next: NextFunction
 ): Promise<void> => {
   try {
-    await fn(ctx, req, res, datalake)
+    await ctx.with(name, {}, (ctx) => fn(ctx, req, res, datalake))
   } catch (err: unknown) {
     next(err)
   }
 }
 
 const wrapRequest =
-  (ctx: MeasureContext, datalake: Datalake, fn: AsyncRequestHandler) =>
+  (ctx: MeasureContext, name: string, datalake: Datalake, fn: AsyncRequestHandler) =>
     (req: Request, res: Response, next: NextFunction) => {
     // eslint-disable-next-line @typescript-eslint/no-floating-promises
-      handleRequest(ctx, datalake, fn, req, res, next)
+      handleRequest(ctx, name, datalake, fn, req, res, next)
     }
 
-export function createServer (config: Config): { app: Express, close: () => void } {
-  const ctx = initStatisticsContext('datalake', {})
-
+export function createServer (ctx: MeasureContext, config: Config): { app: Express, close: () => void } {
   const buckets: Partial<Record<Location, S3Bucket>> = {}
   for (const bucket of config.Buckets) {
     const location = bucket.location as Location
@@ -87,7 +89,7 @@ export function createServer (config: Config): { app: Express, close: () => void
   }
 
   const db = createDb(ctx, config.DbUrl)
-  const datalake = new DatalakeImpl(ctx, db, buckets, { cacheControl })
+  const datalake = new DatalakeImpl(db, buckets, { cacheControl })
 
   const tempFileDir = mkdtempSync(join(tmpdir(), 'datalake-'))
 
@@ -101,26 +103,36 @@ export function createServer (config: Config): { app: Express, close: () => void
     })
   )
 
-  app.get('/blob/:workspace', withAuthorization, withWorkspace, wrapRequest(ctx, datalake, handleBlobList))
+  app.get('/blob/:workspace', withAuthorization, withWorkspace, wrapRequest(ctx, 'listBlobs', datalake, handleBlobList))
 
-  app.head('/blob/:workspace/:name', withBlob, wrapRequest(ctx, datalake, handleBlobHead))
+  app.head('/blob/:workspace/:name', withBlob, wrapRequest(ctx, 'headBlob', datalake, handleBlobHead))
 
-  app.head('/blob/:workspace/:name/:filename', withBlob, wrapRequest(ctx, datalake, handleBlobHead))
+  app.head('/blob/:workspace/:name/:filename', withBlob, wrapRequest(ctx, 'headBlob', datalake, handleBlobHead))
 
-  app.get('/blob/:workspace/:name', withBlob, wrapRequest(ctx, datalake, handleBlobGet))
+  app.get('/blob/:workspace/:name', withBlob, wrapRequest(ctx, 'getBlob', datalake, handleBlobGet))
 
-  app.get('/blob/:workspace/:name/:filename', withBlob, wrapRequest(ctx, datalake, handleBlobGet))
+  app.get('/blob/:workspace/:name/:filename', withBlob, wrapRequest(ctx, 'getBlob', datalake, handleBlobGet))
 
-  app.delete('/blob/:workspace/:name', withAuthorization, withBlob, wrapRequest(ctx, datalake, handleBlobDelete))
+  app.delete(
+    '/blob/:workspace/:name',
+    withAuthorization,
+    withBlob,
+    wrapRequest(ctx, 'deleteBlob', datalake, handleBlobDelete)
+  )
 
   app.delete(
     '/blob/:workspace/:name/:filename',
     withAuthorization,
     withBlob,
-    wrapRequest(ctx, datalake, handleBlobDelete)
+    wrapRequest(ctx, 'deleteBlob', datalake, handleBlobDelete)
   )
 
-  app.delete('/blob/:workspace', withAuthorization, withWorkspace, wrapRequest(ctx, datalake, handleBlobDeleteList))
+  app.delete(
+    '/blob/:workspace',
+    withAuthorization,
+    withWorkspace,
+    wrapRequest(ctx, 'deleteBlob', datalake, handleBlobDeleteList)
+  )
 
   // Form Data upload
 
@@ -128,7 +140,7 @@ export function createServer (config: Config): { app: Express, close: () => void
     '/upload/form-data/:workspace',
     withAuthorization,
     withWorkspace,
-    wrapRequest(ctx, datalake, handleUploadFormData)
+    wrapRequest(ctx, 'uploadFormData', datalake, handleUploadFormData)
   )
 
   // S3 upload
@@ -137,10 +149,15 @@ export function createServer (config: Config): { app: Express, close: () => void
     '/upload/s3/:workspace',
     withAuthorization,
     withWorkspace,
-    wrapRequest(ctx, datalake, handleS3CreateBlobParams)
+    wrapRequest(ctx, 's3UploadParams', datalake, handleS3CreateBlobParams)
   )
 
-  app.post('/upload/s3/:workspace/:name', withAuthorization, withBlob, wrapRequest(ctx, datalake, handleS3CreateBlob))
+  app.post(
+    '/upload/s3/:workspace/:name',
+    withAuthorization,
+    withBlob,
+    wrapRequest(ctx, 's3Upload', datalake, handleS3CreateBlob)
+  )
 
   // // Multipart upload
   // app.post('/upload/multipart/:workspace/:name',
@@ -161,7 +178,9 @@ export function createServer (config: Config): { app: Express, close: () => void
   //   wrapRequest(ctx, handleMultipartUploadComplete)
   // )
 
-  app.get('/image/:transform/:workspace/:name', withBlob, wrapRequest(ctx, datalake, handleImageGet)) // no auth
+  // Image
+
+  app.get('/image/:transform/:workspace/:name', withBlob, wrapRequest(ctx, 'transformImage', datalake, handleImageGet)) // no auth
 
   app.use((err: any, _req: any, res: any, _next: any) => {
     console.log(err)
@@ -173,6 +192,35 @@ export function createServer (config: Config): { app: Express, close: () => void
     res.status(500).json({ message: err.message?.length > 0 ? err.message : 'Internal Server Error' })
   })
 
+  app.get('/api/v1/statistics', (req, res) => {
+    try {
+      const token = req.query.token as string
+      const payload = decodeToken(token)
+      const admin = payload.extra?.admin === 'true'
+      res.setHeader('Content-Type', 'application/json')
+      res.setHeader('Connection', 'keep-alive')
+      res.setHeader('Keep-Alive', 'timeout=5')
+      res.setHeader('Cache-Control', cacheControlNoCache)
+
+      const json = JSON.stringify({
+        metrics: metricsAggregate((ctx as any).metrics),
+        statistics: {
+          activeSessions: {}
+        },
+        admin
+      })
+      res.status(200).send(json)
+    } catch (err: any) {
+      if (err instanceof TokenError) {
+        res.status(401).send()
+        return
+      }
+      ctx.error('statistics error', { err })
+      Analytics.handleError(err)
+      res.status(404).send()
+    }
+  })
+
   app.get('/', (_req, res) => {
     res.send(`
       Huly® Datalake™ <a href="https://huly.io">https://huly.io</a>