Add more info-level logging for sync processes

Author: TimCsaky

app/src/components/queueManager.js

@@ -57,7 +57,10 @@ class QueueManager {
   checkQueue() {
     if (!this.isBusy && !this.toClose) {
       objectQueueService.queueSize().then(size => {
-        if (size > 0) this.processNextJob();
+        if (size > 0) {
+          log.verbose(`There are ${size} jobs in the queue to process`, { function: 'checkQueue' });
+          this.processNextJob();
+        }
       }).catch((err) => {
         log.error(`Error encountered while checking queue: ${err.message}`, { function: 'checkQueue', error: err });
       });
@@ -93,7 +96,7 @@ class QueueManager {
         this.isBusy = true;
         job = response[0];
 
-        log.verbose(`Started processing job id ${job.id}`, { function: 'processNextJob', job: job });
+        log.info(`Started processing job id ${job.id}`, { function: 'processNextJob', job: job });
 
         const objectId = await syncService.syncJob(job.path, job.bucketId, job.full, job.createdBy);
 

app/src/controllers/sync.js

@@ -59,8 +59,14 @@ const controller = {
       // parent + child bucket records already in COMS db
       const dbChildBuckets = await bucketService.searchChildBuckets(parentBucket, false, userId);
       let dbBuckets = [parentBucket].concat(dbChildBuckets);
+      log.info(`Found ${dbBuckets.length} bucket records in COMS db for parent bucketId ${bucketId}`,
+        { function: 'syncBucketRecursive' });
+
       // 'folders' that exist below (and including) the parent 'folder' in S3
       const s3Response = await storageService.listAllObjectVersions({ bucketId: bucketId, precisePath: false });
+      log.info(`Found ${s3Response.Versions.length} object versions and ${s3Response.DeleteMarkers.length} 
+        delete markers in S3 for bucketId ${bucketId}`, { function: 'syncBucketRecursive' });
+
       const s3Keys = [...new Set([
         ...s3Response.DeleteMarkers.map(object => formatS3KeyForCompare(object.Key)),
         ...s3Response.Versions.map(object => formatS3KeyForCompare(object.Key)),
@@ -78,6 +84,7 @@ const controller = {
           userId,
           trx
         );
+        log.info(`${syncedBuckets.length} buckets records synced`, { function: 'syncBucketRecursive' });
 
         /**
          * Queue objects in all the folders for syncing
@@ -107,6 +114,8 @@ const controller = {
       const userId = await userService.getCurrentUserId(getCurrentIdentity(req.currentUser, SYSTEM_USER), SYSTEM_USER);
 
       const s3Objects = await storageService.listAllObjectVersions({ bucketId: bucketId, filterLatest: true });
+      log.info(`Found ${s3Objects.Versions.length} object versions and ${s3Objects.DeleteMarkers.length} 
+        delete markers in S3 for bucketId ${bucketId}`, { function: 'syncBucketSingle' });
 
       const response = await utils.trxWrapper(async (trx) => {
         return this.queueObjectRecords([bucket], s3Objects, userId, trx);
@@ -139,6 +148,8 @@ const controller = {
         oldDbBuckets.map(dbBucket =>
           bucketService.delete(dbBucket.bucketId, trx)
             .then(() => {
+              log.info(`Deleted bucketId ${dbBucket.bucketId} from COMS db as key ${dbBucket.key} not found in S3`,
+                { function: 'syncBucketRecords' });
               dbBuckets = dbBuckets.filter(b => b.bucketId !== dbBucket.bucketId);
             })
         )
@@ -161,6 +172,7 @@ const controller = {
       const newS3Keys = s3Keys.filter(k => !dbBuckets.map(b => b.key).includes(k));
       await Promise.all(
         newS3Keys.map(s3Key => {
+          log.info(`Creating new bucket record in COMS db for S3 key ${s3Key}`, { function: 'syncBucketRecords' });
           const data = {
             bucketName: s3Key.substring(s3Key.lastIndexOf('/') + 1),
             accessKeyId: parentBucket.accessKeyId,
@@ -204,6 +216,9 @@ const controller = {
         bucketId: dbBuckets.map(b => b.bucketId)
       }, trx);
 
+      log.info(`Found ${dbObjects.data.length} object records in COMS db for ${dbBuckets.length} buckets`,
+        { function: 'queueObjectRecords' });
+
       /**
        * merge arrays of objects from COMS db and S3 to form an array of jobs with format:
        *  [ { path: '/images/img3.jpg', bucketId: '123' }, { path: '/images/album1/img1.jpg', bucketId: '456' } ]
@@ -239,6 +254,7 @@ const controller = {
 
       // merge and remove duplicates
       const jobs = [...new Map(objects.map(o => [o.path, o])).values()];
+      log.info(`Prepared ${jobs.length} jobs to enqueue to object queue`, { function: 'queueObjectRecords' });
 
       // create jobs in COMS db object_queue for each object
       // update 'lastSyncRequestedDate' value in COMS db for each bucket

app/src/services/objectQueue.js

@@ -1,6 +1,8 @@
 const { NIL: SYSTEM_USER } = require('uuid');
 
 const { ObjectQueue } = require('../db/models');
+const log = require('../components/log')(module.filename);
+
 
 /**
  * Max number of parameters in a prepared statement (this is a Postgres hard-coded limit).
@@ -53,6 +55,8 @@ const service = {
     let trx;
     try {
       trx = etrx ? etrx : await ObjectQueue.startTransaction();
+      log.info(`Enqueuing ${jobs.length} jobs to object queue (full: ${full}, 
+        retries: ${retries}, createdBy: ${createdBy})`, { function: 'enqueue' });
 
       const jobsArray = jobs.map(job => ({
         bucketId: job.bucketId,
@@ -78,6 +82,8 @@ const service = {
         const response = await ObjectQueue.query(trx).insert(batch).onConflict().ignore();
 
         totalInserted += response.reduce((acc, job) => job?.id ? acc + 1 : acc, 0);
+        log.verbose(`Inserted ${response.length} jobs to object queue (batch starting at index ${i})`,
+          { function: 'enqueue' });
       }
 
       if (!etrx) await trx.commit();

app/src/services/sync.js

@@ -98,11 +98,14 @@ const service = {
         // 1. Sync Object
         const object = await service.syncObject(path, bucketId, userId, trx)
           .then(obj => obj.object);
+        log.info(`Synchronized object at path ${path} in bucket ${bucketId}`,
+          { function: 'syncJob', objectId: object?.id });
 
         // 2. Sync Object Versions
         let versions = [];
         if (object) {
           versions = await service.syncVersions(object, userId, trx);
+          log.info(`Synchronized ${versions.length} versions for object id ${object.id}`, { function: 'syncJob' });
         }
 
         // 3. Sync Version Metadata & Tags
@@ -407,8 +410,10 @@ const service = {
 
       // COMS Tags
       const comsTags = comsTagsForVersion[0]?.tagset ?? [];
+      // log.info(`Found ${comsTags.length} tags in COMS for version id ${comsVersion.id}`, { function: 'syncTags' });
       // S3 Tags
       const s3Tags = toLowerKeys(s3TagsForVersion?.TagSet ?? []);
+      // log.info(`Found ${s3Tags.length} tags in S3 for version id ${comsVersion.id}`, { function: 'syncTags' });
       /**
        * Add coms-id tag to latest version in S3 if not already present
        * NOTE: For a sync job the _deriveObjectId() function will have already added
@@ -501,8 +506,12 @@ const service = {
 
       // COMS Metadata
       const comsMetadata = comsMetadataForVersion[0]?.metadata ?? [];
+      // log.info(`Found ${comsMetadata.length} metadata entries in COMS for version id ${comsVersion.id}`, 
+      // { function: 'syncMetadata' });
       // S3 Metadata
       const s3Metadata = getKeyValue(s3ObjectHead?.Metadata ?? {});
+      // log.info(`Found ${s3Metadata.length} metadata entries in S3 for version id ${comsVersion.id}`, 
+      // { function: 'syncMetadata' });
 
       // Dissociate Metadata not in S3
       const oldMetadata = [];