From 5f465112bfa402e4f5fbbaee773b15c98a27ee4b Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 25 May 2023 13:56:30 +0800 Subject: [PATCH] feat: update matter importer metrics --- packages/import-handler/src/csv.ts | 14 ++-- packages/import-handler/src/index.ts | 6 -- packages/import-handler/src/matterHistory.ts | 67 +++++++++++++++++--- packages/import-handler/src/metrics.ts | 5 +- 4 files changed, 68 insertions(+), 24 deletions(-) diff --git a/packages/import-handler/src/csv.ts b/packages/import-handler/src/csv.ts index 2ff059735..70bb8b955 100644 --- a/packages/import-handler/src/csv.ts +++ b/packages/import-handler/src/csv.ts @@ -6,9 +6,12 @@ import { parse } from '@fast-csv/parse' import { Stream } from 'stream' import { ImportContext } from '.' -import { ImportStatus, updateMetrics } from './metrics' +import { createMetrics, ImportStatus, updateMetrics } from './metrics' export const importCsv = async (ctx: ImportContext, stream: Stream) => { + // create metrics in redis + await createMetrics(ctx.redisClient, ctx.userId, ctx.taskId, 'csv-importer') + const parser = parse() stream.pipe(parser) for await (const row of parser) { @@ -30,8 +33,7 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => { ctx.redisClient, ctx.userId, ctx.taskId, - ImportStatus.TOTAL, - ctx.source + ImportStatus.TOTAL ) await ctx.urlHandler(ctx, url, state, labels) @@ -42,8 +44,7 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => { ctx.redisClient, ctx.userId, ctx.taskId, - ImportStatus.STARTED, - ctx.source + ImportStatus.STARTED ) } catch (error) { console.log('invalid url', row, error) @@ -54,8 +55,7 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => { ctx.redisClient, ctx.userId, ctx.taskId, - ImportStatus.INVALID, - ctx.source + ImportStatus.INVALID ) } } diff --git a/packages/import-handler/src/index.ts b/packages/import-handler/src/index.ts index 53d02e926..b20c51e8d 100644 --- a/packages/import-handler/src/index.ts +++ b/packages/import-handler/src/index.ts @@ -10,7 +10,6 @@ import { promisify } from 'util' import { v4 as uuid } from 'uuid' import { importCsv } from './csv' import { importMatterArchive } from './matterHistory' -import { createMetrics } from './metrics' import { createRedisClient } from './redis' import { CONTENT_FETCH_URL, createCloudTask, emailUserUrl } from './task' @@ -59,7 +58,6 @@ export type ImportContext = { contentHandler: ContentHandler redisClient: RedisClient taskId: string - source: string } type importHandlerFunc = (ctx: ImportContext, stream: Stream) => Promise @@ -277,12 +275,8 @@ const handleEvent = async (data: StorageEvent, redisClient: RedisClient) => { contentHandler, redisClient, taskId: data.name, - source: 'csv-importer', } - // create metrics in redis - await createMetrics(redisClient, ctx.userId, ctx.taskId, ctx.source) - await handler(ctx, stream) if (ctx.countImported > 0) { diff --git a/packages/import-handler/src/matterHistory.ts b/packages/import-handler/src/matterHistory.ts index 6b626d9bc..378f9d51e 100644 --- a/packages/import-handler/src/matterHistory.ts +++ b/packages/import-handler/src/matterHistory.ts @@ -4,20 +4,19 @@ /* eslint-disable @typescript-eslint/no-unsafe-argument */ import { parse } from '@fast-csv/parse' -import { Stream } from 'stream' -import unzip from 'unzip-stream' +import { Readability } from '@omnivore/readability' +import crypto from 'crypto' +import createDOMPurify, { SanitizeElementHookEvent } from 'dompurify' import fs from 'fs' -import path from 'path' import * as fsExtra from 'fs-extra' import glob from 'glob' - import { parseHTML } from 'linkedom' -import { Readability } from '@omnivore/readability' -import createDOMPurify, { SanitizeElementHookEvent } from 'dompurify' - +import path from 'path' +import { Stream } from 'stream' +import unzip from 'unzip-stream' import { encode } from 'urlsafe-base64' -import crypto from 'crypto' import { ImportContext } from '.' +import { createMetrics, ImportStatus, updateMetrics } from './metrics' export type UrlHandler = (url: URL) => Promise @@ -36,8 +35,22 @@ export const importMatterHistoryCsv = async ( for await (const row of parser) { try { const url = new URL(row['URL']) + // update total counter + await updateMetrics( + ctx.redisClient, + ctx.userId, + ctx.taskId, + ImportStatus.TOTAL + ) await ctx.urlHandler(ctx, url) ctx.countImported += 1 + // update started counter + await updateMetrics( + ctx.redisClient, + ctx.userId, + ctx.taskId, + ImportStatus.STARTED + ) } catch (error) { console.log('invalid url', row, error) ctx.countFailed += 1 @@ -204,6 +217,13 @@ const handleMatterHistoryRow = async ( if (!url) { ctx.countFailed += 1 + // update failed counter + await updateMetrics( + ctx.redisClient, + ctx.userId, + ctx.taskId, + ImportStatus.FAILED + ) return } @@ -232,6 +252,14 @@ export const importMatterArchive = async ( const archiveDir = await unarchive(stream) try { + // create metrics in redis + await createMetrics( + ctx.redisClient, + ctx.userId, + ctx.taskId, + 'matter-importer' + ) + const historyFile = path.join(archiveDir, '_matter_history.csv') const parser = parse({ @@ -243,11 +271,34 @@ export const importMatterArchive = async ( for await (const row of parser) { try { + // update total metrics + await updateMetrics( + ctx.redisClient, + ctx.userId, + ctx.taskId, + ImportStatus.TOTAL + ) + await handleMatterHistoryRow(ctx, archiveDir, row) + ctx.countImported += 1 + // update started metrics + await updateMetrics( + ctx.redisClient, + ctx.userId, + ctx.taskId, + ImportStatus.STARTED + ) } catch (error) { console.log('invalid url', row, error) ctx.countFailed += 1 + // update failed metrics + await updateMetrics( + ctx.redisClient, + ctx.userId, + ctx.taskId, + ImportStatus.FAILED + ) } } } catch (err) { diff --git a/packages/import-handler/src/metrics.ts b/packages/import-handler/src/metrics.ts index 75086639d..1fef1f7f8 100644 --- a/packages/import-handler/src/metrics.ts +++ b/packages/import-handler/src/metrics.ts @@ -53,8 +53,7 @@ export const updateMetrics = async ( redisClient: RedisClient, userId: string, taskId: string, - status: ImportStatus, - source: string + status: ImportStatus ) => { const key = `import:${userId}:${taskId}` @@ -62,7 +61,7 @@ export const updateMetrics = async ( // use lua script to increment hash field const state = await redisClient.evalSha(lua.sha, { keys: [key], - arguments: [status, Date.now().toString(), source], + arguments: [status, Date.now().toString()], }) // if the task is finished, send email