Refactor: pass scanId as parameter instead of instance field

In this commit we refactor conductorScanId to be passed as a parameter through the call chain
instead of storing it as an instance field (_scanId), as scanId is only relevant within the context
of a single scan execution passing it as a parameter makes dependencies explicit and Reduces instance state and improves code clarity

Issue: BB-740

Author: benzekrimaha

extensions/lifecycle/conductor/LifecycleConductor.js

@@ -106,7 +106,6 @@ class LifecycleConductor {
         this._vaultClientCache = null;
         this._initialized = false;
         this._batchInProgress = false;
-        this._scanId = null;
 
         // this cache only needs to be the size of one listing.
         // worst case scenario is 1 account per bucket:
@@ -310,13 +309,13 @@ class LifecycleConductor {
         });
     }
 
-    _taskToMessage(task, taskVersion, log) {
+    _taskToMessage(task, taskVersion, scanId, log) {
         return {
             message: JSON.stringify({
                 action: 'processObjects',
                 contextInfo: {
                     reqId: log.getSerializedUids(),
-                    conductorScanId: this._scanId,
+                    conductorScanId: scanId,
                 },
                 target: {
                     bucket: task.bucketName,
@@ -357,28 +356,28 @@ class LifecycleConductor {
             cb);
     }
 
-    _createBucketTaskMessages(tasks, log, cb) {
+    _createBucketTaskMessages(tasks, scanId, log, cb) {
         if (this.lcConfig.forceLegacyListing) {
-            return process.nextTick(cb, null, tasks.map(t => this._taskToMessage(t, lifecycleTaskVersions.v1, log)));
+            return process.nextTick(cb, null, tasks.map(t => this._taskToMessage(t, lifecycleTaskVersions.v1, scanId, log)));
         }
 
         return async.mapLimit(tasks, 10, (t, taskDone) =>
             this._indexesGetOrCreate(t, log, (err, taskVersion) => {
                 if (err) {
                     // should not happen as indexes methods would
                     // ignore the errors and fallback to v1 listing
-                    return taskDone(null, this._taskToMessage(t, lifecycleTaskVersions.v1, log));
+                    return taskDone(null, this._taskToMessage(t, lifecycleTaskVersions.v1, scanId, log));
                 }
-                return taskDone(null, this._taskToMessage(t, taskVersion, log));
+                return taskDone(null, this._taskToMessage(t, taskVersion, scanId, log));
             }), cb);
     }
 
     processBuckets(cb) {
         const log = this.logger.newRequestLogger();
         const start = Date.now();
-        this._scanId = uuid();
+        const scanId = uuid();
         if (typeof log.addDefaultFields === 'function') {
-            log.addDefaultFields({ conductorScanId: this._scanId });
+            log.addDefaultFields({ conductorScanId: scanId });
         }
         LifecycleMetrics.onProcessBuckets(log);
         let nBucketsQueued = 0;
@@ -410,7 +409,7 @@ class LifecycleConductor {
                         }
                         return true;
                     }))),
-                (tasksWithAccountId, next) => this._createBucketTaskMessages(tasksWithAccountId, log, next),
+                (tasksWithAccountId, next) => this._createBucketTaskMessages(tasksWithAccountId, scanId, log, next),
             ],
             (err, messages) => {
                 nBucketsQueued += tasks.length;
@@ -419,7 +418,7 @@ class LifecycleConductor {
                 ).length;
 
                 log.info('bucket push progress', {
-                    conductorScanId: this._scanId,
+                    conductorScanId: scanId,
                     nBucketsQueued,
                     bucketsInCargo: tasks.length,
                     kafkaBucketMessagesDeliveryReports: messageDeliveryReports,
@@ -445,9 +444,9 @@ class LifecycleConductor {
                 this._batchInProgress = true;
                 log.info('starting new lifecycle batch', {
                     bucketSource: this._bucketSource,
-                    conductorScanId: this._scanId,
+                    conductorScanId: scanId,
                 });
-                this.listBuckets(messageSendQueue, log, (err, listedBucketsCount) => {
+                this.listBuckets(messageSendQueue, scanId, log, (err, listedBucketsCount) => {
                     LifecycleMetrics.onBucketListing(log, err);
                     nBucketsListed = listedBucketsCount;
                     return next(err, listedBucketsCount);
@@ -471,7 +470,7 @@ class LifecycleConductor {
             if (err && err.Throttling) {
                 log.info('not starting new lifecycle batch', {
                     reason: err,
-                    conductorScanId: this._scanId,
+                    conductorScanId: scanId,
                     fullScanElapsedMs: elapsedMs,
                 });
                 if (cb) {
@@ -489,7 +488,7 @@ class LifecycleConductor {
             log.error('lifecycle batch failed', {
                 error: err,
                 unknownCanonicalIds,
-                conductorScanId: this._scanId,
+                conductorScanId: scanId,
                 fullScanElapsedMs: elapsedMs,
                 nBucketsListed,
                 workflowsQueued: messageDeliveryReports,
@@ -504,7 +503,7 @@ class LifecycleConductor {
                 nBucketsQueued,
                 nLifecycledBuckets,
                 unknownCanonicalIds,
-                conductorScanId: this._scanId,
+                conductorScanId: scanId,
                 fullScanElapsedMs: elapsedMs,
                 nBucketsListed,
                 workflowsQueued: messageDeliveryReports,
@@ -515,25 +514,25 @@ class LifecycleConductor {
         });
     }
 
-    listBuckets(queue, log, cb) {
+    listBuckets(queue, scanId, log, cb) {
         if (this._bucketSource === 'zookeeper') {
-            return this.listZookeeperBuckets(queue, log, cb);
+            return this.listZookeeperBuckets(queue, scanId, log, cb);
         }
 
         if (this._bucketSource === 'mongodb') {
-            return this.listMongodbBuckets(queue, log, cb);
+            return this.listMongodbBuckets(queue, scanId, log, cb);
         }
 
         return this.restoreBucketCheckpoint((err, marker) => {
             if (err) {
                 return cb(err);
             }
 
-            return this.listBucketdBuckets(queue, marker || null, log, cb);
+            return this.listBucketdBuckets(queue, marker || null, scanId, log, cb);
         });
     }
 
-    listZookeeperBuckets(queue, log, cb) {
+    listZookeeperBuckets(queue, scanId, log, cb) {
         const zkBucketsPath = this.getBucketsZkPath();
         this._zkClient.getChildren(
             zkBucketsPath,
@@ -616,7 +615,7 @@ class LifecycleConductor {
         });
     }
 
-    listBucketdBuckets(queue, initMarker, log, cb) {
+    listBucketdBuckets(queue, initMarker, scanId, log, cb) {
         let isTruncated = true;
         let marker = initMarker;
         let nEnqueued = 0;
@@ -635,7 +634,7 @@ class LifecycleConductor {
                     maxInFlight: this._maxInFlightBatchSize,
                     bucketListingPushRateHz: Math.round(nEnqueued * 1000 / (new Date() - start)),
                     breakerState,
-                    conductorScanId: this._scanId,
+                    conductorScanId: scanId,
                 };
 
                 if (queue.length() > this._maxInFlightBatchSize ||
@@ -717,7 +716,7 @@ class LifecycleConductor {
         );
     }
 
-    listMongodbBuckets(queue, log, cb) {
+    listMongodbBuckets(queue, scanId, log, cb) {
         let nEnqueued = 0;
         const start = new Date();
 
@@ -791,7 +790,7 @@ class LifecycleConductor {
                             maxInFlight: this._maxInFlightBatchSize,
                             enqueueRateHz: Math.round(nEnqueued * 1000 / (new Date() - start)),
                             breakerState,
-                            conductorScanId: this._scanId,
+                            conductorScanId: scanId,
                         };
 
                         if (queue.length() > this._maxInFlightBatchSize ||