Add unit tests for scan monitoring metrics

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,