fix: address code review comments on scan metrics

- Clear _scanEndTimeout in close() to prevent callback firing
  on a partially torn-down bucket processor instance
- Propagate conductor's start timestamp (Date.now() at scan
  start) via contextInfo.scanStartTimestamp instead of using
  Date.now() in the bucket processor, so the scan progress
  dashboard accurately reflects the full scan duration
- Document that bucketProcessorBucketsCount.reset() clears all
  label combinations (acceptable: single bucket source per
  deployment)
- Update tests and BackbeatTestConsumer normalizer for the new
  scanStartTimestamp field

Author: benzekrimaha

extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js

@@ -310,6 +310,8 @@ class LifecycleBucketProcessor {
         }
         const { bucket, owner, accountId, taskVersion } = result.target;
         const conductorScanId = result.contextInfo && result.contextInfo.conductorScanId;
+        const scanStartTimestamp = (result.contextInfo
+            && result.contextInfo.scanStartTimestamp) || Date.now();
         if (conductorScanId && conductorScanId !== this._metricsScanId) {
             // New scan: cancel any pending debounce, full metric reset
             if (this._scanEndTimeout) {
@@ -318,7 +320,7 @@ class LifecycleBucketProcessor {
             }
             this._metricsScanId = conductorScanId;
             LifecycleMetrics.onBucketProcessorScanStart(
-                this._log, Date.now()
+                this._log, scanStartTimestamp
             );
             this._processorScanMetricsActive = true;
             this._log.info('new conductor scan detected', {
@@ -584,6 +586,10 @@ class LifecycleBucketProcessor {
      * @return {undefined}
      */
     close(cb) {
+        if (this._scanEndTimeout) {
+            clearTimeout(this._scanEndTimeout);
+            this._scanEndTimeout = null;
+        }
         if (this._deleteInactiveCredentialsInterval) {
             clearInterval(this._deleteInactiveCredentialsInterval);
         }

extensions/lifecycle/conductor/LifecycleConductor.js

@@ -309,13 +309,14 @@ class LifecycleConductor {
         });
     }
 
-    _taskToMessage(task, taskVersion, scanId, log) {
+    _taskToMessage(task, taskVersion, scanId, scanStartTimestamp, log) {
         return {
             message: JSON.stringify({
                 action: 'processObjects',
                 contextInfo: {
                     reqId: log.getSerializedUids(),
                     conductorScanId: scanId,
+                    scanStartTimestamp,
                     bucketSource: this._bucketSource,
                 },
                 target: {
@@ -357,20 +358,22 @@ class LifecycleConductor {
             cb);
     }
 
-    _createBucketTaskMessages(tasks, scanId, log, cb) {
+    _createBucketTaskMessages(tasks, scanId, scanStartTimestamp, log, cb) {
         if (this.lcConfig.forceLegacyListing) {
             return process.nextTick(cb, null, tasks.map(t =>
-                this._taskToMessage(t, lifecycleTaskVersions.v1, scanId, log)));
+                this._taskToMessage(t, lifecycleTaskVersions.v1, scanId, scanStartTimestamp, 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, scanId, log));
+                    return taskDone(null, this._taskToMessage(t, lifecycleTaskVersions.v1, scanId,
+                        scanStartTimestamp, log));
                 }
-                return taskDone(null, this._taskToMessage(t, taskVersion, scanId, log));
+                return taskDone(null, this._taskToMessage(t, taskVersion, scanId,
+                    scanStartTimestamp, log));
             }), cb);
     }
 
@@ -411,7 +414,8 @@ class LifecycleConductor {
                         }
                         return true;
                     }))),
-                (tasksWithAccountId, next) => this._createBucketTaskMessages(tasksWithAccountId, scanId, log, next),
+                (tasksWithAccountId, next) => this._createBucketTaskMessages(
+                    tasksWithAccountId, scanId, start, log, next),
             ],
             (err, messages) => {
                 nBucketsQueued += tasks.length;

tests/unit/lifecycle/LifecycleConductor.spec.js

@@ -300,7 +300,7 @@ describe('Lifecycle Conductor', () => {
                     setTimeout(() => cb(null, []), 1000);
                 });
             sinon.stub(conductor, '_createBucketTaskMessages')
-                .callsFake((_a, _b, _c, cb) => {
+                .callsFake((_a, _b, _c, _d, cb) => {
                     order.push(2);
                     assert(conductor.activeIndexingJobsRetrieved);
                     setTimeout(() => cb(null, []), 1000);
@@ -326,9 +326,12 @@ describe('Lifecycle Conductor', () => {
 
     describe('_indexesGetOrCreate', () => {
         it('should include conductor scan id in task context', () => {
-            const taskMessage = conductor._taskToMessage(getTask(true), lifecycleTaskVersions.v2, 'scan-id-test', log);
+            const taskMessage = conductor._taskToMessage(
+                getTask(true), lifecycleTaskVersions.v2,
+                'scan-id-test', 1706000000000, log);
             const parsed = JSON.parse(taskMessage.message);
             assert.strictEqual(parsed.contextInfo.conductorScanId, 'scan-id-test');
+            assert.strictEqual(parsed.contextInfo.scanStartTimestamp, 1706000000000);
         });
 
         it('should return v2 for bucketd bucket sources', done => {

tests/utils/BackbeatTestConsumer.js

@@ -45,6 +45,12 @@ class BackbeatTestConsumer extends BackbeatConsumer {
                         !expectedMsg.value.contextInfo?.bucketSource) {
                         delete parsedMsg.contextInfo.bucketSource;
                     }
+                    // scanStartTimestamp is a dynamic conductor timestamp:
+                    // normalize for comparison
+                    if (parsedMsg.contextInfo?.scanStartTimestamp &&
+                        !expectedMsg.value.contextInfo?.scanStartTimestamp) {
+                        delete parsedMsg.contextInfo.scanStartTimestamp;
+                    }
                 }
                 assert.deepStrictEqual(
                     parsedMsg, expectedMsg.value,