diff --git a/services/datalake/pod-datalake/package.json b/services/datalake/pod-datalake/package.json index 67ad3d036d..47e66d652d 100644 --- a/services/datalake/pod-datalake/package.json +++ b/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", diff --git a/services/datalake/pod-datalake/src/datalake/datalake.ts b/services/datalake/pod-datalake/src/datalake/datalake.ts index ac0c2f6792..fe70b151d5 100644 --- a/services/datalake/pod-datalake/src/datalake/datalake.ts +++ b/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>, private readonly options: { @@ -32,8 +31,8 @@ export class DatalakeImpl implements Datalake { } ) {} - async list (workspace: string, cursor?: string, limit?: number): Promise { - const blobs = await this.db.listBlobs(workspace, cursor, limit) + async list (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number): Promise { + 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 { - const { bucket } = await this.selectStorage(workspace) + async head (ctx: MeasureContext, workspace: string, name: string): Promise { + 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 { - const { bucket } = await this.selectStorage(workspace) + async get ( + ctx: MeasureContext, + workspace: string, + name: string, + options: { range?: string } + ): Promise { + 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 { + async delete (ctx: MeasureContext, workspace: string, name: string | string[]): Promise { 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 { - const { location, bucket } = await this.selectStorage(workspace) + async create (ctx: MeasureContext, workspace: string, name: string, filename: string): Promise { + 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 { - await this.db.setParent({ workspace, name }, parent !== null ? { workspace, name: parent } : null) + async setParent (ctx: MeasureContext, workspace: string, name: string, parent: string | null): Promise { + await this.db.setParent(ctx, { workspace, name }, parent !== null ? { workspace, name: parent } : null) } - async selectStorage (workspace: string): Promise { + async selectStorage (ctx: MeasureContext, workspace: string): Promise { const location = this.selectLocation(workspace) const bucket = this.buckets[location] if (bucket == null) { diff --git a/services/datalake/pod-datalake/src/datalake/db.ts b/services/datalake/pod-datalake/src/datalake/db.ts index 74ad9aa879..26ba22a6cf 100644 --- a/services/datalake/pod-datalake/src/datalake/db.ts +++ b/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 - getBlob: (blobId: BlobId) => Promise - listBlobs: (workspace: string, cursor?: string, limit?: number) => Promise - createData: (data: BlobDataRecord) => Promise - createBlob: (blob: Omit) => Promise - createBlobData: (blob: BlobWithDataRecord) => Promise - deleteBlob: (blob: BlobId) => Promise - setParent: (blob: BlobId, parent: BlobId | null) => Promise - deleteBlobList: (list: BlobIds) => Promise - getStats: () => Promise - getWorkspaceStats: (workspace: string) => Promise + getData: (ctx: MeasureContext, dataId: BlobDataId) => Promise + getBlob: (ctx: MeasureContext, blobId: BlobId) => Promise + listBlobs: (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number) => Promise + createData: (ctx: MeasureContext, data: BlobDataRecord) => Promise + createBlob: (ctx: MeasureContext, blob: Omit) => Promise + createBlobData: (ctx: MeasureContext, blob: BlobWithDataRecord) => Promise + deleteBlob: (ctx: MeasureContext, blob: BlobId) => Promise + setParent: (ctx: MeasureContext, blob: BlobId, parent: BlobId | null) => Promise + deleteBlobList: (ctx: MeasureContext, list: BlobIds) => Promise + getStats: (ctx: MeasureContext) => Promise + getWorkspaceStats: (ctx: MeasureContext, workspace: string) => Promise } 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 { + async getData (ctx: MeasureContext, dataId: BlobDataId): Promise { const { hash, location } = dataId const rows = await this.db.execute( @@ -146,7 +146,7 @@ export class PostgresDB implements BlobDB { return rows.length > 0 ? rows[0] : null } - async deleteBlobList (blobList: BlobIds): Promise { + async deleteBlobList (ctx: MeasureContext, blobList: BlobIds): Promise { const { workspace, names } = blobList await this.db.execute( @@ -159,7 +159,7 @@ export class PostgresDB implements BlobDB { ) } - async getBlob (blobId: BlobId): Promise { + async getBlob (ctx: MeasureContext, blobId: BlobId): Promise { const { workspace, name } = blobId const rows = await this.db.execute( @@ -179,7 +179,7 @@ export class PostgresDB implements BlobDB { return null } - async listBlobs (workspace: string, cursor?: string, limit?: number): Promise { + async listBlobs (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number): Promise { cursor = cursor ?? '' limit = Math.min(limit ?? 100, 1000) @@ -201,7 +201,7 @@ export class PostgresDB implements BlobDB { } } - async createBlob (blob: Omit): Promise { + async createBlob (ctx: MeasureContext, blob: Omit): Promise { 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 { + async createData (ctx: MeasureContext, data: BlobDataRecord): Promise { 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 { + async createBlobData (ctx: MeasureContext, data: BlobWithDataRecord): Promise { 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 { + async deleteBlob (ctx: MeasureContext, blob: BlobId): Promise { const { workspace, name } = blob const blobs = new Set() @@ -284,7 +284,7 @@ export class PostgresDB implements BlobDB { ) } - async setParent (blob: BlobId, parent: BlobId | null): Promise { + async setParent (ctx: MeasureContext, blob: BlobId, parent: BlobId | null): Promise { const { workspace, name } = blob await this.db.execute( @@ -297,7 +297,7 @@ export class PostgresDB implements BlobDB { ) } - async getStats (): Promise { + async getStats (ctx: MeasureContext): Promise { 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 { + async getWorkspaceStats (ctx: MeasureContext, workspace: string): Promise { 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 { - return await this.ctx.with('db.getData', {}, () => this.db.getData(dataId)) + async getData (ctx: MeasureContext, dataId: BlobDataId): Promise { + return await ctx.with('db.getData', {}, () => this.db.getData(this.ctx, dataId)) } - async getBlob (blobId: BlobId): Promise { - return await this.ctx.with('db.getBlob', {}, () => this.db.getBlob(blobId)) + async getBlob (ctx: MeasureContext, blobId: BlobId): Promise { + return await ctx.with('db.getBlob', {}, () => this.db.getBlob(this.ctx, blobId)) } - async listBlobs (workspace: string, cursor?: string, limit?: number): Promise { - return await this.ctx.with('db.listBlobs', {}, () => this.db.listBlobs(workspace, cursor, limit)) + async listBlobs (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number): Promise { + return await ctx.with('db.listBlobs', {}, () => this.db.listBlobs(this.ctx, workspace, cursor, limit)) } - async createData (data: BlobDataRecord): Promise { - await this.ctx.with('db.createData', {}, () => this.db.createData(data)) + async createData (ctx: MeasureContext, data: BlobDataRecord): Promise { + await ctx.with('db.createData', {}, () => this.db.createData(this.ctx, data)) } - async createBlob (blob: Omit): Promise { - await this.ctx.with('db.createBlob', {}, () => this.db.createBlob(blob)) + async createBlob (ctx: MeasureContext, blob: Omit): Promise { + await ctx.with('db.createBlob', {}, () => this.db.createBlob(this.ctx, blob)) } - async createBlobData (data: BlobWithDataRecord): Promise { - await this.ctx.with('db.createBlobData', {}, () => this.db.createBlobData(data)) + async createBlobData (ctx: MeasureContext, data: BlobWithDataRecord): Promise { + await ctx.with('db.createBlobData', {}, () => this.db.createBlobData(this.ctx, data)) } - async deleteBlobList (blobs: BlobIds): Promise { - await this.ctx.with('db.deleteBlobList', {}, () => this.db.deleteBlobList(blobs)) + async deleteBlobList (ctx: MeasureContext, blobs: BlobIds): Promise { + await ctx.with('db.deleteBlobList', {}, () => this.db.deleteBlobList(this.ctx, blobs)) } - async deleteBlob (blob: BlobId): Promise { - await this.ctx.with('db.deleteBlob', {}, () => this.db.deleteBlob(blob)) + async deleteBlob (ctx: MeasureContext, blob: BlobId): Promise { + await ctx.with('db.deleteBlob', {}, () => this.db.deleteBlob(this.ctx, blob)) } - async setParent (blob: BlobId, parent: BlobId | null): Promise { - await this.ctx.with('db.setParent', {}, () => this.db.setParent(blob, parent)) + async setParent (ctx: MeasureContext, blob: BlobId, parent: BlobId | null): Promise { + await ctx.with('db.setParent', {}, () => this.db.setParent(this.ctx, blob, parent)) } - async getStats (): Promise { - return await this.ctx.with('db.getStats', {}, () => this.db.getStats()) + async getStats (ctx: MeasureContext): Promise { + return await ctx.with('db.getStats', {}, () => this.db.getStats(this.ctx)) } - async getWorkspaceStats (workspace: string): Promise { - return await this.ctx.with('db.getWorkspaceStats', {}, () => this.db.getWorkspaceStats(workspace)) + async getWorkspaceStats (ctx: MeasureContext, workspace: string): Promise { + return await ctx.with('db.getWorkspaceStats', {}, () => this.db.getWorkspaceStats(this.ctx, workspace)) } } diff --git a/services/datalake/pod-datalake/src/datalake/types.ts b/services/datalake/pod-datalake/src/datalake/types.ts index 46c9874af6..05aa949204 100644 --- a/services/datalake/pod-datalake/src/datalake/types.ts +++ b/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 - head: (workspace: string, name: string) => Promise - get: (workspace: string, name: string, options: { range?: string }) => Promise - delete: (workspace: string, name: string | string[]) => Promise + list: (ctx: MeasureContext, workspace: string, cursor?: string, limit?: number) => Promise + head: (ctx: MeasureContext, workspace: string, name: string) => Promise + get: (ctx: MeasureContext, workspace: string, name: string, options: { range?: string }) => Promise + delete: (ctx: MeasureContext, workspace: string, name: string | string[]) => Promise put: ( + ctx: MeasureContext, workspace: string, name: string, sha256: string, body: Buffer | Readable, options: Omit ) => Promise - create: (workspace: string, name: string, filename: string) => Promise + create: (ctx: MeasureContext, workspace: string, name: string, filename: string) => Promise - setParent: (workspace: string, name: string, parent: string | null) => Promise - selectStorage: (workspace: string) => Promise + setParent: (ctx: MeasureContext, workspace: string, name: string, parent: string | null) => Promise + selectStorage: (ctx: MeasureContext, workspace: string) => Promise } diff --git a/services/datalake/pod-datalake/src/handlers/blob.ts b/services/datalake/pod-datalake/src/handlers/blob.ts index 738175bfc2..73e12bc023 100644 --- a/services/datalake/pod-datalake/src/handlers/blob.ts +++ b/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 { 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() diff --git a/services/datalake/pod-datalake/src/handlers/image.ts b/services/datalake/pod-datalake/src/handlers/image.ts index a12d8c5ead..d69d7c0430 100644 --- a/services/datalake/pod-datalake/src/handlers/image.ts +++ b/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) diff --git a/services/datalake/pod-datalake/src/handlers/multipart.ts b/services/datalake/pod-datalake/src/handlers/multipart.ts index 0a1b164380..cd54029d85 100644 --- a/services/datalake/pod-datalake/src/handlers/multipart.ts +++ b/services/datalake/pod-datalake/src/handlers/multipart.ts @@ -40,14 +40,14 @@ export async function handleMultipartUploadStart ( ): Promise { 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 { 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() } diff --git a/services/datalake/pod-datalake/src/handlers/s3.ts b/services/datalake/pod-datalake/src/handlers/s3.ts index 9f356cb9c6..db53e5a65b 100644 --- a/services/datalake/pod-datalake/src/handlers/s3.ts +++ b/services/datalake/pod-datalake/src/handlers/s3.ts @@ -25,7 +25,7 @@ export async function handleS3CreateBlobParams ( datalake: Datalake ): Promise { 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) diff --git a/services/datalake/pod-datalake/src/main.ts b/services/datalake/pod-datalake/src/main.ts index 699535d965..f0106092c0 100644 --- a/services/datalake/pod-datalake/src/main.ts +++ b/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 => { 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 => { diff --git a/services/datalake/pod-datalake/src/s3/bucket.ts b/services/datalake/pod-datalake/src/s3/bucket.ts index 49efe8a9a1..b628a65ddb 100644 --- a/services/datalake/pod-datalake/src/s3/bucket.ts +++ b/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 { - const result = await this.client.headObject({ Bucket: this.bucket, Key: key }) + async head (ctx: MeasureContext, key: string): Promise { + 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 { + async get (ctx: MeasureContext, key: string, options?: S3GetOptions): Promise { 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 { + async put ( + ctx: MeasureContext, + key: string, + body: Readable | Buffer | string, + options: S3PutOptions + ): Promise { 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 { + async delete (ctx: MeasureContext, key: string): Promise { await this.client.deleteObject({ Bucket: this.bucket, Key: key }) } - async createMultipartUpload (key: string, options: S3CreateMultipartUploadOptions): Promise { + async createMultipartUpload ( + ctx: MeasureContext, + key: string, + options: S3CreateMultipartUploadOptions + ): Promise { 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 { + async abortMultipartUpload (ctx: MeasureContext, key: string, multipart: S3MultipartUpload): Promise { const command = { Bucket: this.bucket, Key: key, UploadId: multipart.uploadId } - await this.client.abortMultipartUpload(command) + await ctx.with('s3.abortMultipartUpload', {}, () => this.client.abortMultipartUpload(command)) } } diff --git a/services/datalake/pod-datalake/src/s3/types.ts b/services/datalake/pod-datalake/src/s3/types.ts index 5e5f58a0c6..a69211929a 100644 --- a/services/datalake/pod-datalake/src/s3/types.ts +++ b/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 - get: (key: string, options?: S3GetOptions) => Promise - put: (key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise - delete: (key: string) => Promise + head: (ctx: MeasureContext, key: string) => Promise + get: (ctx: MeasureContext, key: string, options?: S3GetOptions) => Promise + put: (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise + delete: (ctx: MeasureContext, key: string) => Promise // multipart - createMultipartUpload: (key: string, options: S3CreateMultipartUploadOptions) => Promise + createMultipartUpload: ( + ctx: MeasureContext, + key: string, + options: S3CreateMultipartUploadOptions + ) => Promise uploadMultipartPart: ( + ctx: MeasureContext, key: string, multipart: S3MultipartUpload, body: Readable | Buffer | string, options: S3UploadPartOptions ) => Promise - completeMultipartUpload: (key: string, multipart: S3MultipartUpload, parts: S3MultipartUploadPart[]) => Promise - abortMultipartUpload: (key: string, multipart: S3MultipartUpload) => Promise + completeMultipartUpload: ( + ctx: MeasureContext, + key: string, + multipart: S3MultipartUpload, + parts: S3MultipartUploadPart[] + ) => Promise + abortMultipartUpload: (ctx: MeasureContext, key: string, multipart: S3MultipartUpload) => Promise } diff --git a/services/datalake/pod-datalake/src/server.ts b/services/datalake/pod-datalake/src/server.ts index 238e6b6262..0b7a617ef5 100644 --- a/services/datalake/pod-datalake/src/server.ts +++ b/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 const handleRequest = async ( ctx: MeasureContext, + name: string, datalake: Datalake, fn: AsyncRequestHandler, req: Request, @@ -56,22 +60,20 @@ const handleRequest = async ( next: NextFunction ): Promise => { 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> = {} 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™ https://huly.io