Add tests for scan monitoring metrics

This commit adds unit tests for new metrics and conductorScanId generation.
Updates functional tests to handle conductorScanId in messages.
Normalizes conductorScanId in BackbeatTestConsumer for test
comparisons (similar to reqId handling).

Issue: BB-740

Author: benzekrimaha

tests/functional/lifecycle/LifecycleConductor.spec.js

@@ -41,15 +41,15 @@ const expected2Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: { reqId: 'test-request-id', conductorScanId: 'test-scan-id' },
             target: { bucket: 'bucket1', owner: 'owner1', taskVersion: version },
             details: {},
         },
     },
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: { reqId: 'test-request-id', conductorScanId: 'test-scan-id' },
             target: { bucket: 'bucket1-2', owner: 'owner1', taskVersion: version },
             details: {},
         },
@@ -60,31 +60,31 @@ const expected4Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: { reqId: 'test-request-id', conductorScanId: 'test-scan-id' },
             target: { bucket: 'bucket1', owner: 'owner1', taskVersion: version },
             details: {},
         },
     },
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: { reqId: 'test-request-id', conductorScanId: 'test-scan-id' },
             target: { bucket: 'bucket1-2', owner: 'owner1', taskVersion: version },
             details: {},
         },
     },
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: { reqId: 'test-request-id', conductorScanId: 'test-scan-id' },
             target: { bucket: 'bucket3', owner: 'owner3', taskVersion: version },
             details: {},
         },
     },
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: { reqId: 'test-request-id', conductorScanId: 'test-scan-id' },
             target: { bucket: 'bucket4', owner: 'owner4', taskVersion: version },
             details: {},
         },

tests/unit/lifecycle/LifecycleConductor.spec.js

@@ -7,6 +7,7 @@ const { splitter } = require('arsenal').constants;
 
 const LifecycleConductor = require(
     '../../../extensions/lifecycle/conductor/LifecycleConductor');
+const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics');
 const {
     lifecycleTaskVersions,
     indexesForFeature
@@ -190,6 +191,64 @@ describe('Lifecycle Conductor', () => {
             conductor.processBuckets(done);
         });
 
+        it('should publish full scan metrics at end of scan', done => {
+            conductor._mongodbClient = {
+                getIndexingJobs: (_, cb) => cb(null, ['job1']),
+                getCollection: () => ({
+                    find: () => ({
+                        project: () => ({
+                            hasNext: () => Promise.resolve(false),
+                        }),
+                    }),
+                }),
+            };
+            conductor._zkClient = {
+                getData: (_, cb) => cb(null, null, null),
+                setData: (path, data, version, cb) => cb(null, { version: 1 }),
+            };
+
+            sinon.stub(conductor, '_controlBacklog').callsFake(cb => cb(null));
+            const metricStub = sinon.stub(LifecycleMetrics, 'onConductorFullScan');
+
+            conductor.processBuckets(err => {
+                assert.ifError(err);
+                assert(metricStub.calledOnce);
+                const [, , bucketCount, workflowCount, lifecycleBucketCount] =
+                    metricStub.firstCall.args;
+                assert.strictEqual(bucketCount, 0);
+                assert.strictEqual(workflowCount, 0);
+                assert.strictEqual(lifecycleBucketCount, 0);
+                done();
+            });
+        });
+
+        it('should generate a conductorScanId', done => {
+            conductor._mongodbClient = {
+                getIndexingJobs: (_, cb) => cb(null, []),
+                getCollection: () => ({
+                    find: () => ({
+                        project: () => ({
+                            hasNext: () => Promise.resolve(false),
+                        }),
+                    }),
+                }),
+            };
+            conductor._zkClient = {
+                getData: (_, cb) => cb(null, null, null),
+                setData: (path, data, version, cb) => cb(null, { version: 1 }),
+            };
+
+            sinon.stub(conductor, '_controlBacklog').callsFake(cb => cb(null));
+
+            conductor.processBuckets(err => {
+                assert.ifError(err);
+                assert(conductor._scanId);
+                assert(typeof conductor._scanId === 'string');
+                assert(conductor._scanId.length > 0);
+                done();
+            });
+        });
+
         // tests that `activeIndexingJobRetrieved` is not reset until the e
         it('should not reset `activeIndexingJobsRetrieved` while async operations are in progress', done => {
             const order = [];
@@ -244,6 +303,14 @@ describe('Lifecycle Conductor', () => {
     });
 
     describe('_indexesGetOrCreate', () => {
+        it('should include conductor scan id in task context', () => {
+            conductor._scanId = 'scan-id-test';
+
+            const taskMessage = conductor._taskToMessage(getTask(true), lifecycleTaskVersions.v2, log);
+            const parsed = JSON.parse(taskMessage.message);
+            assert.strictEqual(parsed.contextInfo.conductorScanId, 'scan-id-test');
+        });
+
         it('should return v2 for bucketd bucket sources', done => {
             conductor._bucketSource = 'bucketd';
             conductor._indexesGetOrCreate(getTask(undefined), log, (err, taskVersion) => {

tests/unit/lifecycle/LifecycleMetrics.spec.js

@@ -79,6 +79,72 @@ describe('LifecycleMetrics', () => {
             }));
         });
 
+        it('should catch errors in onConductorFullScan', () => {
+            const metric = ZenkoMetrics.getMetric('s3_lifecycle_conductor_full_scan_elapsed_seconds');
+            sinon.stub(metric, 'set').throws(new Error('Metric error'));
+
+            LifecycleMetrics.onConductorFullScan(log, 5000, 10, 8, 5);
+
+            assert(log.error.calledOnce);
+            assert(log.error.calledWithMatch('failed to update prometheus metrics', {
+                method: 'LifecycleMetrics.onConductorFullScan',
+                elapsedMs: 5000,
+                bucketCount: 10,
+                workflowCount: 8,
+                lifecycleBucketCount: 5,
+            }));
+        });
+
+        it('should set and reset bucket processor scan gauges', () => {
+            const scanStartMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_start_time');
+            const bucketsMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_buckets_count');
+            const startSet = sinon.stub(scanStartMetric, 'set');
+            const bucketsSet = sinon.stub(bucketsMetric, 'set');
+            const bucketsInc = sinon.stub(bucketsMetric, 'inc');
+
+            // start a scan
+            LifecycleMetrics.onBucketProcessorScanStart(
+                log, 1706000000000);
+            assert(startSet.calledOnce);
+            assert(startSet.calledWithMatch(
+                { origin: 'bucket_processor' }, 1706000000000));
+            assert(bucketsSet.calledOnce);
+            assert(bucketsSet.calledWithMatch(
+                { origin: 'bucket_processor' }, 0));
+
+            // process a bucket
+            LifecycleMetrics.onBucketProcessorBucketDone(log);
+            assert(bucketsInc.calledOnce);
+
+            // end the scan
+            startSet.resetHistory();
+            LifecycleMetrics.onBucketProcessorScanEnd(log);
+            assert(startSet.calledOnce);
+            assert(startSet.calledWithMatch(
+                { origin: 'bucket_processor' }, 0));
+
+            assert(log.error.notCalled);
+        });
+
+        it('should catch errors in onBucketProcessorScanStart', () => {
+            const scanStartMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_start_time');
+            sinon.stub(scanStartMetric, 'set')
+                .throws(new Error('Metric error'));
+
+            LifecycleMetrics.onBucketProcessorScanStart(
+                log, 1706000000000);
+
+            assert(log.error.calledOnce);
+            assert(log.error.calledWithMatch(
+                'failed to update prometheus metrics', {
+                    method:
+                        'LifecycleMetrics.onBucketProcessorScanStart',
+                }));
+        });
+
         it('should catch errors in onLifecycleTriggered', () => {
             LifecycleMetrics.onLifecycleTriggered(log, 'conductor', 'expiration', 'us-east-1', NaN);
 
@@ -169,5 +235,28 @@ describe('LifecycleMetrics', () => {
                 process: 'conductor',
             }));
         });
+
+        it('should set full scan metrics including lifecycle bucket count', () => {
+            const elapsedMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_conductor_full_scan_elapsed_seconds');
+            const countMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_conductor_scan_count');
+            const elapsedSet = sinon.stub(elapsedMetric, 'set');
+            const countSet = sinon.stub(countMetric, 'set');
+
+            LifecycleMetrics.onConductorFullScan(log, 3000, 7, 6, 4);
+
+            assert(elapsedSet.calledOnce);
+            assert(elapsedSet.calledWithMatch(
+                { origin: 'conductor' }, 3));
+            assert.strictEqual(countSet.callCount, 3);
+            assert(countSet.getCall(0).calledWithMatch(
+                { origin: 'conductor', type: 'bucket' }, 7));
+            assert(countSet.getCall(1).calledWithMatch(
+                { origin: 'conductor', type: 'lifecycle_bucket' }, 4));
+            assert(countSet.getCall(2).calledWithMatch(
+                { origin: 'conductor', type: 'workflow' }, 6));
+            assert(log.error.notCalled);
+        });
     });
 });

tests/utils/BackbeatTestConsumer.js

@@ -32,6 +32,14 @@ class BackbeatTestConsumer extends BackbeatConsumer {
                     // present
                     assert(parsedMsg.contextInfo?.reqId, 'expected contextInfo.reqId field');
                     parsedMsg.contextInfo.reqId = expectedMsg.value.contextInfo?.reqId;
+                    // conductorScanId is also generated per scan: check it exists,
+                    // then normalize for comparison
+                    if (expectedMsg.value.contextInfo?.conductorScanId === 'test-scan-id') {
+                        assert(parsedMsg.contextInfo?.conductorScanId,
+                            'expected contextInfo.conductorScanId field');
+                        parsedMsg.contextInfo.conductorScanId =
+                            expectedMsg.value.contextInfo.conductorScanId;
+                    }
                 }
                 assert.deepStrictEqual(
                     parsedMsg, expectedMsg.value,