Skip intermediate stream ListRecordStream

scality · Dec 30, 2024 · a1eeff9 · a1eeff9
1 parent df54ae7
commit a1eeff9
Showing 1 changed file with 14 additions and 1 deletion.
diff --git a/lib/storage/metadata/bucketclient/LogConsumer.js b/lib/storage/metadata/bucketclient/LogConsumer.js
@@ -208,7 +208,12 @@ class LogConsumer {
      * @return {undefined}
      */
     readRecords(params, cb) {
-        const recordStream = new ListRecordStream(this.logger, this.metrics);
+        let recordStream;
+        if (process.env.LOG_CONSUMER_SKIP_RECORD_STREAM) {
+            recordStream = new stream.PassThrough();
+        } else {
+            recordStream = new ListRecordStream(this.logger, this.metrics);
+        }
         const _params = params || {};
         const cbOnce = jsutil.once(cb);
 
@@ -267,6 +272,14 @@ class LogConsumer {
                     this.metrics.json.readOn.data.inc();
                     this.metrics.json.writeBufferLength.observe(jsonResponse.buffer.length);
                     this.metrics.json.readFlowing[+!jsonResponse.paused].inc();
+
+                    if (process.env.LOG_CONSUMER_SKIP_RECORD_STREAM) {
+                        this.metrics.record.readBufferLength.observe(recordStream.readableLength);
+                        this.metrics.record.readFlowing[+recordStream.readableFlowing || 0].inc();
+                        this.metrics.record.writeBufferLength.observe(recordStream.writableLength);
+                        this.metrics.record.writeDrain[+recordStream.writableNeedDrain || 0].inc();
+                        this.metrics.record.writeCorked.observe(recordStream.writableCorked);
+                    }
                 });
                 jsonResponse.on('resume', () => {
                     this.metrics.json.readOn.resume.inc();