fixup! Refactor: pass scanId as parameter instead of instance field and fix metric timing

fix(tests): update tests to reflect messageId removal and scanId parameter changes

Author: benzekrimaha

extensions/lifecycle/LifecycleMetrics.js

@@ -5,6 +5,7 @@ const LIFECYCLE_LABEL_OP = 'op';
 const LIFECYCLE_LABEL_STATUS = 'status';
 const LIFECYCLE_LABEL_LOCATION = 'location';
 const LIFECYCLE_LABEL_TYPE = 'type';
+const LIFECYCLE_LABEL_BUCKET_SOURCE = 'bucket_source';
 
 const LIFECYCLE_MARKER_METRICS_LOCATION = '-delete-marker-';
 
@@ -59,7 +60,7 @@ const conductorFullScanElapsedSeconds = ZenkoMetrics.createGauge({
 const conductorScanCount = ZenkoMetrics.createGauge({
     name: 's3_lifecycle_conductor_scan_count',
     help: 'Number of buckets and queued workflows for the latest full lifecycle conductor scan',
-    labelNames: [LIFECYCLE_LABEL_ORIGIN, LIFECYCLE_LABEL_TYPE],
+    labelNames: [LIFECYCLE_LABEL_ORIGIN, LIFECYCLE_LABEL_TYPE, LIFECYCLE_LABEL_BUCKET_SOURCE],
 });
 
 const bucketProcessorScanStartTime = ZenkoMetrics.createGauge({
@@ -207,7 +208,7 @@ class LifecycleMetrics {
      * @param {number} lifecycleBucketCount - buckets that have lifecycle enabled
      */
     static onConductorFullScan(
-        log, elapsedMs, bucketCount, workflowCount, lifecycleBucketCount
+        log, elapsedMs, bucketCount, workflowCount, lifecycleBucketCount, bucketSource
     ) {
         try {
             conductorFullScanElapsedSeconds.set(
@@ -217,20 +218,35 @@ class LifecycleMetrics {
             conductorScanCount.set({
                 [LIFECYCLE_LABEL_ORIGIN]: 'conductor',
                 [LIFECYCLE_LABEL_TYPE]: 'bucket',
+                [LIFECYCLE_LABEL_BUCKET_SOURCE]: bucketSource || 'unknown',
             }, bucketCount);
-            conductorScanCount.set({
-                [LIFECYCLE_LABEL_ORIGIN]: 'conductor',
-                [LIFECYCLE_LABEL_TYPE]: 'lifecycle_bucket',
-            }, lifecycleBucketCount);
+            // For bucketd, we can't know lifecycle buckets at conductor level,
+            // so we don't set the metric here. The bucket processor will
+            // increment it when it discovers buckets with lifecycle.
+            if (bucketSource !== 'bucketd') {
+                conductorScanCount.set({
+                    [LIFECYCLE_LABEL_ORIGIN]: 'conductor',
+                    [LIFECYCLE_LABEL_TYPE]: 'lifecycle_bucket',
+                    [LIFECYCLE_LABEL_BUCKET_SOURCE]: bucketSource || 'unknown',
+                }, lifecycleBucketCount);
+            } else {
+                // Reset to 0 for bucketd at start of scan, bucket processor will increment
+                conductorScanCount.set({
+                    [LIFECYCLE_LABEL_ORIGIN]: 'conductor',
+                    [LIFECYCLE_LABEL_TYPE]: 'lifecycle_bucket',
+                    [LIFECYCLE_LABEL_BUCKET_SOURCE]: 'bucketd',
+                }, 0);
+            }
             conductorScanCount.set({
                 [LIFECYCLE_LABEL_ORIGIN]: 'conductor',
                 [LIFECYCLE_LABEL_TYPE]: 'workflow',
+                [LIFECYCLE_LABEL_BUCKET_SOURCE]: bucketSource || 'unknown',
             }, workflowCount);
         } catch (err) {
             LifecycleMetrics.handleError(
                 log, err, 'LifecycleMetrics.onConductorFullScan', {
                     elapsedMs, bucketCount, workflowCount,
-                    lifecycleBucketCount,
+                    lifecycleBucketCount, bucketSource,
                 }
             );
         }
@@ -266,17 +282,31 @@ class LifecycleMetrics {
      * Increment the count of lifecycle-enabled buckets successfully processed
      * during the current scan. This is called after the bucket task completes
      * successfully, not when it is dispatched to the scheduler.
+     * Also updates the conductor metric for bucketd source since we can't know
+     * at conductor level which buckets have lifecycle.
      * @param {Object} log - logger
+     * @param {string} bucketSource - source of bucket (mongodb, zookeeper, bucketd)
+     * @param {string} conductorScanId - scan ID from conductor
      */
-    static onBucketProcessorBucketDone(log) {
+    static onBucketProcessorBucketDone(log, bucketSource, conductorScanId) {
         try {
             bucketProcessorBucketsCount.inc(
                 { [LIFECYCLE_LABEL_ORIGIN]: 'bucket_processor' }
             );
+            // For bucketd source, update conductor metric since we can't know
+            // at conductor level which buckets have lifecycle
+            if (bucketSource === 'bucketd' && conductorScanId) {
+                conductorScanCount.inc({
+                    [LIFECYCLE_LABEL_ORIGIN]: 'conductor',
+                    [LIFECYCLE_LABEL_TYPE]: 'lifecycle_bucket',
+                    [LIFECYCLE_LABEL_BUCKET_SOURCE]: 'bucketd',
+                });
+            }
         } catch (err) {
             LifecycleMetrics.handleError(
                 log, err,
-                'LifecycleMetrics.onBucketProcessorBucketDone'
+                'LifecycleMetrics.onBucketProcessorBucketDone',
+                { bucketSource, conductorScanId }
             );
         }
     }

extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js

@@ -388,7 +388,10 @@ class LifecycleBucketProcessor {
                 // This ensures we count actual processing completion, not just dispatch
                 // Only count successful completions, not failures
                 if (!err) {
-                    LifecycleMetrics.onBucketProcessorBucketDone(this._log);
+                    const bucketSource = result.contextInfo && result.contextInfo.bucketSource;
+                    LifecycleMetrics.onBucketProcessorBucketDone(
+                        this._log, bucketSource, conductorScanId
+                    );
                 }
                 return cb(err);
             });

extensions/lifecycle/conductor/LifecycleConductor.js

@@ -316,6 +316,7 @@ class LifecycleConductor {
                 contextInfo: {
                     reqId: log.getSerializedUids(),
                     conductorScanId: scanId,
+                    bucketSource: this._bucketSource,
                 },
                 target: {
                     bucket: task.bucketName,
@@ -415,7 +416,7 @@ class LifecycleConductor {
             (err, messages) => {
                 nBucketsQueued += tasks.length;
                 nLifecycledBuckets += tasks.filter(
-                    t => t.isLifecycled !== false
+                    t => t.isLifecycled === true
                 ).length;
 
                 log.info('bucket push progress', {
@@ -498,7 +499,8 @@ class LifecycleConductor {
                 elapsedMs,
                 nBucketsListed,
                 messageDeliveryReports,
-                nLifecycledBuckets
+                nLifecycledBuckets,
+                this._bucketSource
             );
 
             log.info('finished pushing lifecycle batch', {
@@ -673,7 +675,7 @@ class LifecycleConductor {
                                     const [canonicalId, bucketName] = marker.split(constants.splitter);
                                     if (!this._isBlacklisted(canonicalId, bucketName)) {
                                         nEnqueued += 1;
-                                        queue.push({ canonicalId, bucketName, isLifecycled: true });
+                                        queue.push({ canonicalId, bucketName });
                                         this.lastSentId = o.key;
                                     // Optimization:
                                     // If we only blacklist by accounts, and the last bucket is blacklisted