feat: implement structured GCP Cloud Logging with explicit severity across all apps

apps/bigquery-export/Dockerfile

@@ -19,4 +19,5 @@ USER node
 COPY --chown=node:node --from=pruner /app/out/full/ .
 COPY --chown=node:node --from=installer /app/node_modules ./node_modules
 WORKDIR /app/apps/bigquery-export
+ENV NODE_NO_WARNINGS=1
 CMD ["node", "index.js"]

apps/bigquery-export/firestore.js

@@ -1,5 +1,5 @@
 import { Firestore, FieldPath } from '@google-cloud/firestore'
-import { BigQueryExport } from '@httparchive/shared'
+import { BigQueryExport, logger } from '@httparchive/shared'
 
 export class FirestoreBatch {
   constructor() {
@@ -38,12 +38,12 @@ export class FirestoreBatch {
       external: Math.round(used.external / 1024 / 1024 * 100) / 100
     }
 
-    console.log(`Memory usage ${operation}: RSS ${memoryInfo.rss}MB, Heap Used ${memoryInfo.heapUsed}MB, Heap Total ${memoryInfo.heapTotal}MB, External ${memoryInfo.external}MB`)
+    logger.info(`Memory usage ${operation}: RSS ${memoryInfo.rss}MB, Heap Used ${memoryInfo.heapUsed}MB, Heap Total ${memoryInfo.heapTotal}MB, External ${memoryInfo.external}MB`, { memoryInfo, operation })
 
     // Configurable memory warning threshold from environment
     const warningThreshold = parseInt(process.env.MEMORY_WARNING_THRESHOLD_MB || '1500')
     if (memoryInfo.heapUsed > warningThreshold) {
-      console.warn(`⚠️ High memory usage detected: ${memoryInfo.heapUsed}MB heap used (threshold: ${warningThreshold}MB)`)
+      logger.warn(`High memory usage detected: ${memoryInfo.heapUsed}MB heap used (threshold: ${warningThreshold}MB)`, { heapUsed: memoryInfo.heapUsed, warningThreshold })
     }
 
     return memoryInfo
@@ -54,13 +54,14 @@ export class FirestoreBatch {
     this.processedDocs = 0
     this.totalDocs = 0
     this.pendingCount = 0
+    this.failedWrites = 0
 
     // Clean up existing BulkWriter if it exists
     if (this.bulkWriter) {
       try {
         this.bulkWriter.close()
       } catch (error) {
-        console.warn('Error closing existing BulkWriter:', error.message)
+        logger.warn('Error closing existing BulkWriter', error)
       }
     }
     this.bulkWriter = null
@@ -83,13 +84,22 @@ export class FirestoreBatch {
     // Configure error handling with progress info
     bulkWriter.onWriteError((error) => {
       const progressInfo = this.totalDocs > 0 ? ` (${this.processedDocs}/${this.totalDocs})` : ''
-      console.warn(`${operation} operation failed${progressInfo}:`, error.message)
+      // Log only identifying fields; a stack per retried write adds nothing
+      const writeError = {
+        error: error.message,
+        code: error.code,
+        operationType: error.operationType,
+        documentPath: error.documentRef?.path,
+        failedAttempts: error.failedAttempts
+      }
+      logger.warn(`${operation} operation failed${progressInfo}: ${error.message}`, writeError)
 
       // Limit retry attempts to prevent infinite retry loops on persistent transient errors
       const MAX_RETRIES = 5
       if (error.failedAttempts >= MAX_RETRIES) {
-        console.error(`Operation failed after ${error.failedAttempts} attempts. Skipping/failing.`)
+        logger.error(`Operation failed after ${error.failedAttempts} attempts. Skipping/failing.`, writeError)
         this.pendingCount--
+        this.failedWrites++
         return false
       }
 
@@ -112,11 +122,12 @@ export class FirestoreBatch {
         (typeof error.code === 'string' && retryableErrorStrings.includes(error.code.toLowerCase()))
 
       if (isRetryable) {
-        console.log(`Retrying failed operation (attempt ${error.failedAttempts + 1}/${MAX_RETRIES})...`)
+        logger.info(`Retrying failed operation (attempt ${error.failedAttempts + 1}/${MAX_RETRIES})...`, { attempt: error.failedAttempts + 1 })
         return true
       }
 
       this.pendingCount--
+      this.failedWrites++
       return false
     })
 
@@ -128,7 +139,7 @@ export class FirestoreBatch {
       // Report progress periodically
       if (this.processedDocs % this.config.progressReportInterval === 0) {
         const progressInfo = this.totalDocs > 0 ? ` (${this.processedDocs}/${this.totalDocs})` : ` (${this.processedDocs} processed)`
-        console.log(`Progress${progressInfo} - ${operation}ing documents in ${this.collectionName}`)
+        logger.info(`Progress${progressInfo} - ${operation}ing documents in ${this.collectionName}`, { processedDocs: this.processedDocs, totalDocs: this.totalDocs })
 
         // Force garbage collection periodically
         if (this.processedDocs % this.config.gcInterval === 0 && global.gc) {
@@ -140,15 +151,21 @@ export class FirestoreBatch {
     return bulkWriter
   }
 
+  assertNoFailedWrites(operation) {
+    if (this.failedWrites > 0) {
+      throw new Error(`${this.failedWrites} ${operation} operations failed in ${this.collectionName}`)
+    }
+  }
+
   async waitIfNeeded() {
     const limit = 100000
     const target = 50000
     if (this.pendingCount > limit) {
-      console.log(`Pipeline full (${this.pendingCount} pending). Waiting for queue to drain below ${target}...`)
+      logger.info(`Pipeline full (${this.pendingCount} pending). Waiting for queue to drain below ${target}...`, { pendingCount: this.pendingCount })
       while (this.pendingCount > target) {
         await new Promise(resolve => setTimeout(resolve, 50))
       }
-      console.log(`Pipeline drained (${this.pendingCount} pending). Resuming...`)
+      logger.info(`Pipeline drained (${this.pendingCount} pending). Resuming...`, { pendingCount: this.pendingCount })
 
       // Force garbage collection after queue drains
       if (global.gc) {
@@ -160,11 +177,11 @@ export class FirestoreBatch {
   buildQuery(collectionRef) {
     const queryMap = {
       report: () => {
-        console.info(`Deleting documents from ${this.collectionName} for date ${this.date}`)
+        logger.info(`Deleting documents from ${this.collectionName} for date ${this.date}`, { collection: this.collectionName, date: this.date })
         return collectionRef.where('date', '==', this.date)
       },
       dict: () => {
-        console.info(`Deleting documents from ${this.collectionName}`)
+        logger.info(`Deleting documents from ${this.collectionName}`, { collection: this.collectionName })
         return collectionRef
       }
     }
@@ -182,13 +199,13 @@ export class FirestoreBatch {
       const countSnapshot = await query.count().get()
       return countSnapshot.data().count
     } catch (error) {
-      console.warn('Could not get document count for progress tracking:', error.message)
+      logger.warn('Could not get document count for progress tracking', error)
       return 0
     }
   }
 
   async batchDelete() {
-    console.info('Starting batch deletion...')
+    logger.info('Starting batch deletion...')
     const startTime = Date.now()
     this.reset()
 
@@ -198,7 +215,7 @@ export class FirestoreBatch {
     // Get total count for progress tracking
     this.totalDocs = await this.getDocumentCount(collectionQuery)
     if (this.totalDocs > 0) {
-      console.info(`Total documents to delete: ${this.totalDocs}`)
+      logger.info(`Total documents to delete: ${this.totalDocs}`, { totalDocs: this.totalDocs })
     }
 
     // Create BulkWriter for delete operations
@@ -216,7 +233,7 @@ export class FirestoreBatch {
         { start: 'v', end: '\uf8ff' }
       ]
 
-      console.info(`Split deletion into ${partitions.length} manual parallel partitions`)
+      logger.info(`Split deletion into ${partitions.length} manual parallel partitions`, { partitionsCount: partitions.length })
 
       await Promise.all(partitions.map(async (partition, index) => {
         let lastDocId = null
@@ -256,32 +273,34 @@ export class FirestoreBatch {
           }
 
           for (const doc of docs) {
-            this.bulkWriter.delete(doc.ref)
+            // Failures are counted in onWriteError and checked after flush
+            this.bulkWriter.delete(doc.ref).catch(() => {})
             this.pendingCount++
             partitionDeletedCount++
           }
 
           // Wait if the pending operations queue is too full
           await this.waitIfNeeded()
         }
-        console.log(`Partition ${index + 1}/${partitions.length} complete. Deleted ${partitionDeletedCount} documents.`)
+        logger.info(`Partition ${index + 1}/${partitions.length} complete. Deleted ${partitionDeletedCount} documents.`, { partition: index + 1, partitionDeletedCount })
       }))
     } catch (error) {
-      console.error('Error during batch deletion pagination:', error)
+      logger.error('Error during batch deletion pagination', error)
       throw error
     }
 
     // Final flush and close
-    console.log('Finalizing deletion operations...')
+    logger.info('Finalizing deletion operations...')
     await this.bulkWriter.flush()
     await this.bulkWriter.close()
+    this.assertNoFailedWrites('delete')
 
     const duration = (Date.now() - startTime) / 1000
-    console.info(`Deletion complete. Total docs deleted: ${this.processedDocs}. Time: ${duration} seconds`)
+    logger.info(`Deletion complete. Total docs deleted: ${this.processedDocs}. Time: ${duration} seconds`, { processedDocs: this.processedDocs, durationSeconds: duration })
   }
 
   async streamFromBigQuery(rowStream) {
-    console.info('Starting BigQuery to Firestore transfer...')
+    logger.info('Starting BigQuery to Firestore transfer...')
     const startTime = Date.now()
     this.reset()
 
@@ -295,7 +314,8 @@ export class FirestoreBatch {
       for await (const row of rowStream) {
         // Add document to BulkWriter
         const docRef = collectionRef.doc()
-        this.bulkWriter.set(docRef, row)
+        // Failures are counted in onWriteError and checked after flush
+        this.bulkWriter.set(docRef, row).catch(() => {})
         this.pendingCount++
         rowCount++
         this.totalDocs = rowCount // Update totalDocs for progress tracking
@@ -304,26 +324,27 @@ export class FirestoreBatch {
         await this.waitIfNeeded()
       }
     } catch (error) {
-      console.error('Error during BigQuery streaming:', error)
+      logger.error('Error during BigQuery streaming', error)
       throw error
     }
 
     // Final flush and close
-    console.log('Finalizing write operations...')
+    logger.info('Finalizing write operations...')
     await this.bulkWriter.flush()
     await this.bulkWriter.close()
+    this.assertNoFailedWrites('write')
 
     // Final garbage collection
     if (global.gc) {
       global.gc()
     }
 
     const duration = (Date.now() - startTime) / 1000
-    console.info(`Transfer to ${this.collectionName} complete. Total rows processed: ${this.processedDocs}. Time: ${duration} seconds`)
+    logger.info(`Transfer to ${this.collectionName} complete. Total rows processed: ${this.processedDocs}. Time: ${duration} seconds`, { processedDocs: this.processedDocs, durationSeconds: duration })
   }
 
   async export(query, exportConfig) {
-    console.log(`Starting export to ${exportConfig.collection}...`)
+    logger.info(`Starting export to ${exportConfig.collection}...`, { collection: exportConfig.collection })
     this.logMemoryUsage('at start')
 
     // Configure Firestore settings
@@ -347,23 +368,23 @@ export class FirestoreBatch {
       await this.streamFromBigQuery(rowStream)
 
       this.logMemoryUsage('at completion')
-      console.log(`✅ Export to ${exportConfig.collection} completed successfully`)
+      logger.info(`Export to ${exportConfig.collection} completed successfully`, { collection: exportConfig.collection })
     } catch (error) {
       this.logMemoryUsage('on error')
 
       // Avoid dumping the massive Firestore client instance (contained in documentRef)
       if (error && error.documentRef) {
         const cleanError = {
-          message: error.message,
+          error: error.message,
           code: error.code,
           documentPath: error.documentRef.path,
           failedAttempts: error.failedAttempts
         }
-        console.error(`❌ Export to ${exportConfig.collection} failed:`, cleanError)
-        throw new Error(`Export failed at document ${cleanError.documentPath}: ${cleanError.message} (code: ${cleanError.code})`, { cause: error })
+        logger.error(`Export to ${exportConfig.collection} failed`, cleanError)
+        throw new Error(`Export failed at document ${cleanError.documentPath}: ${cleanError.error} (code: ${cleanError.code})`, { cause: error })
       }
 
-      console.error(`❌ Export to ${exportConfig.collection} failed:`, error)
+      logger.error(`Export to ${exportConfig.collection} failed`, error)
       throw error
     }
   }

apps/bigquery-export/index.js

@@ -1,20 +1,23 @@
+import { logger, registerProcessLogging } from '@httparchive/shared'
 import { FirestoreBatch } from './firestore.js'
 
+registerProcessLogging({ exitOnUnhandledRejection: true })
+
 async function main () {
   const { query, destination, config } = process.env.EXPORT_CONFIG && JSON.parse(process.env.EXPORT_CONFIG)
 
   if (destination === 'firestore') {
-    console.log(query, config)
+    logger.info('Starting Firestore export', { query, config })
 
     const firestore = new FirestoreBatch()
     await firestore.export(query, config)
   }
 
-  console.info('Export finished successfully')
+  logger.info('Export finished successfully')
   return 'OK'
 }
 
 await main().catch((error) => {
-  console.error(error)
+  logger.error('Export failed', error)
   process.exit(1)
 })

apps/dataform-service/Dockerfile

@@ -19,5 +19,6 @@ USER node
 COPY --chown=node:node --from=pruner /app/out/full/ .
 COPY --chown=node:node --from=installer /app/node_modules ./node_modules
 WORKDIR /app/apps/dataform-service
+ENV NODE_NO_WARNINGS=1
 ENV PORT=8080
 CMD ["/app/node_modules/.bin/functions-framework", "--target=dataform-service", "--signature-type=http"]

apps/dataform-service/cloud_run.js

@@ -1,4 +1,5 @@
 import run from '@google-cloud/run'
+import { logger } from '@httparchive/shared'
 
 // Export functionality
 export async function callRunJob (name, payload = {}) {
@@ -20,5 +21,5 @@ export async function callRunJob (name, payload = {}) {
 
   const [operation] = await client.runJob(request)
 
-  console.info(`Job initialized: ${operation.name}`)
+  logger.info(`Job initialized: ${operation.name}`, { job: operation.name })
 }

apps/dataform-service/dataform.js

@@ -1,4 +1,5 @@
 import { DataformClient } from '@google-cloud/dataform'
+import { logger } from '@httparchive/shared'
 
 const dataformClient = new DataformClient()
 
@@ -16,9 +17,9 @@ export async function getCompilationResults (repoURI) {
     }
   }
 
-  console.info(`Creating Dataform compilation result: ${JSON.stringify(request, null, 2)}`)
+  logger.info('Creating Dataform compilation result', { request })
   const [response] = await dataformClient.createCompilationResult(request)
-  console.info(`Compilation result created: ${response.name}`)
+  logger.info(`Compilation result created: ${response.name}`, { compilationResult: response.name })
   return response.name
 }
 
@@ -44,7 +45,7 @@ export async function runWorkflow (repoURI, compilationResult, tags) {
     }
   }
 
-  console.info(`Invoking Dataform workflow: ${JSON.stringify(request, null, 2)}`)
+  logger.info('Invoking Dataform workflow', { request })
   const [response] = await dataformClient.createWorkflowInvocation(request)
-  console.info(`Workflow invoked: ${response.name}`)
+  logger.info(`Workflow invoked: ${response.name}`, { workflowInvocation: response.name })
 }