feat: add metrics for observability

Author: mbani01

backend/src/database/migrations/U1771605060__addMetricsToSnowflakeExport.sql

@@ -0,0 +1,6 @@
+ALTER TABLE integration."snowflakeExportJobs"
+  ADD COLUMN metrics JSONB;
+
+ALTER TABLE integration."snowflakeExportJobs"
+  DROP COLUMN "totalRows",
+  DROP COLUMN "totalBytes";

backend/src/database/migrations/V1771605060__addMetricsToSnowflakeExport.sql

@@ -54,6 +54,5 @@ export async function executeExport(platform: PlatformType): Promise<void> {
     throw err
   } finally {
     await exporter.destroy().catch((err) => log.warn({ err }, 'Failed to close Snowflake connection'))
-    await db.$pool.end().catch((err) => log.warn({ err }, 'Failed to close DB connection'))
   }
 }

services/apps/snowflake_connectors/src/activities/exportActivity.ts

@@ -71,6 +71,8 @@ export class TransformerConsumer {
   private async processJob(job: SnowflakeExportJob): Promise<void> {
     log.info({ jobId: job.id, platform: job.platform, s3Path: job.s3Path }, 'Processing job')
 
+    const startTime = Date.now()
+
     try {
       const platformDef = getPlatform(job.platform as PlatformType)
 
@@ -99,18 +101,21 @@ export class TransformerConsumer {
         transformedCount++
       }
 
-      await this.metadataStore.markCompleted(job.id)
+      const processingMetrics = {
+        transformedCount,
+        skippedCount,
+        processingDurationMs: Date.now() - startTime,
+      }
+
+      await this.metadataStore.markCompleted(job.id, processingMetrics)
 
-      log.info(
-        { jobId: job.id, totalRows: rows.length, transformedCount, skippedCount },
-        'Job completed',
-      )
+      log.info({ jobId: job.id, ...processingMetrics }, 'Job completed')
     } catch (err) {
       const errorMessage = err instanceof Error ? err.message : String(err)
       log.error({ jobId: job.id, err }, 'Job failed')
 
       try {
-        await this.metadataStore.markFailed(job.id, errorMessage)
+        await this.metadataStore.markFailed(job.id, errorMessage, { processingDurationMs: Date.now() - startTime })
       } catch (updateErr) {
         log.error({ jobId: job.id, updateErr }, 'Failed to mark job as failed')
       }

services/apps/snowflake_connectors/src/consumer/transformerConsumer.ts

@@ -7,19 +7,26 @@
 
 import type { DbConnection } from '@crowd/database'
 
+export interface JobMetrics {
+  exportedRows?: number
+  exportedBytes?: number
+  transformedCount?: number
+  skippedCount?: number
+  processingDurationMs?: number
+}
+
 export interface SnowflakeExportJob {
   id: number
   platform: string
   s3Path: string
-  totalRows: number
-  totalBytes: number
   exportStartedAt: Date | null
   createdAt: Date
   updatedAt: Date
   processingStartedAt: Date | null
   completedAt: Date | null
   cleanedAt: Date | null
   error: string | null
+  metrics: JobMetrics | null
 }
 
 export class MetadataStore {
@@ -32,19 +39,19 @@ export class MetadataStore {
     totalBytes: number,
     exportStartedAt: Date,
   ): Promise<void> {
+    const metrics: JobMetrics = { exportedRows: totalRows, exportedBytes: totalBytes }
     await this.db.none(
-      `INSERT INTO integration."snowflakeExportJobs" (platform, s3_path, "totalRows", "totalBytes", "exportStartedAt")
-       VALUES ($1, $2, $3, $4, $5)
+      `INSERT INTO integration."snowflakeExportJobs" (platform, s3_path, "exportStartedAt", metrics)
+       VALUES ($1, $2, $3, $4)
        ON CONFLICT (s3_path) DO UPDATE SET
-         "totalRows" = EXCLUDED."totalRows",
-         "totalBytes" = EXCLUDED."totalBytes",
          "exportStartedAt" = EXCLUDED."exportStartedAt",
          "processingStartedAt" = NULL,
          "completedAt" = NULL,
          "cleanedAt" = NULL,
          error = NULL,
+         metrics = EXCLUDED.metrics,
          "updatedAt" = NOW()`,
-      [platform, s3Path, totalRows, totalBytes, exportStartedAt],
+      [platform, s3Path, exportStartedAt, JSON.stringify(metrics)],
     )
   }
 
@@ -57,15 +64,14 @@ export class MetadataStore {
       id: number
       platform: string
       s3_path: string
-      totalRows: string
-      totalBytes: string
       exportStartedAt: Date | null
       createdAt: Date
       updatedAt: Date
       processingStartedAt: Date | null
       completedAt: Date | null
       cleanedAt: Date | null
       error: string | null
+      metrics: JobMetrics | null
     }>(
       `UPDATE integration."snowflakeExportJobs"
        SET "processingStartedAt" = NOW(), "updatedAt" = NOW()
@@ -76,27 +82,35 @@ export class MetadataStore {
          LIMIT 1
          FOR UPDATE SKIP LOCKED
        )
-       RETURNING id, platform, s3_path, "totalRows", "totalBytes", "exportStartedAt",
-                 "createdAt", "updatedAt", "processingStartedAt", "completedAt", "cleanedAt", error`,
+       RETURNING id, platform, s3_path, "exportStartedAt",
+                 "createdAt", "updatedAt", "processingStartedAt", "completedAt", "cleanedAt", error, metrics`,
     )
     return row ? mapRowToJob(row) : null
   }
 
-  async markCompleted(jobId: number): Promise<void> {
+  
+  // TODO: Add a cleanup workflow that deletes S3 files for completed/failed jobs
+  // and sets "cleanedAt" to reclaim storage.
+
+  async markCompleted(jobId: number, metrics?: Partial<JobMetrics>): Promise<void> {
     await this.db.none(
       `UPDATE integration."snowflakeExportJobs"
-       SET "completedAt" = NOW(), "updatedAt" = NOW()
+       SET "completedAt" = NOW(),
+           metrics = COALESCE(metrics, '{}'::jsonb) || COALESCE($2::jsonb, '{}'::jsonb),
+           "updatedAt" = NOW()
        WHERE id = $1`,
-      [jobId],
+      [jobId, metrics ? JSON.stringify(metrics) : null],
     )
   }
 
-  async markFailed(jobId: number, error: string): Promise<void> {
+  async markFailed(jobId: number, error: string, metrics?: Partial<JobMetrics>): Promise<void> {
     await this.db.none(
       `UPDATE integration."snowflakeExportJobs"
-       SET error = $2, "completedAt" = NOW(), "updatedAt" = NOW()
+       SET error = $2, "completedAt" = NOW(),
+           metrics = COALESCE(metrics, '{}'::jsonb) || COALESCE($3::jsonb, '{}'::jsonb),
+           "updatedAt" = NOW()
        WHERE id = $1`,
-      [jobId, error],
+      [jobId, error, metrics ? JSON.stringify(metrics) : null],
     )
   }
 
@@ -118,28 +132,26 @@ function mapRowToJob(row: {
   id: number
   platform: string
   s3_path: string
-  totalRows: string
-  totalBytes: string
   exportStartedAt: Date | null
   createdAt: Date
   updatedAt: Date
   processingStartedAt: Date | null
   completedAt: Date | null
   cleanedAt: Date | null
   error: string | null
+  metrics: JobMetrics | null
 }): SnowflakeExportJob {
   return {
     id: row.id,
     platform: row.platform,
     s3Path: row.s3_path,
-    totalRows: Number(row.totalRows),
-    totalBytes: Number(row.totalBytes),
     exportStartedAt: row.exportStartedAt,
     createdAt: row.createdAt,
     updatedAt: row.updatedAt,
     processingStartedAt: row.processingStartedAt,
     completedAt: row.completedAt,
     cleanedAt: row.cleanedAt,
     error: row.error,
+    metrics: row.metrics,
   }
 }

services/apps/snowflake_connectors/src/core/metadataStore.ts

@@ -46,6 +46,8 @@ export class SnowflakeExporter {
       const filename = `batch_${batch}.parquet`
       const s3Path = `${s3FilenamePrefix}/${filename}`
 
+      // TODO: LIMIT/OFFSET over mutable data can skip or duplicate rows if the source changes between batches.
+      // Consider materializing a temp table from the source query first, then paginating over it.
       const copyQuery = `
         COPY INTO '${s3Path}'
         FROM (${sourceQuery} LIMIT ${limit} OFFSET ${offset})