fix: adds more info on stream monitoring (#575)

Author: fenos

src/internal/streams/monitor.ts

@@ -10,19 +10,28 @@ import { Readable } from 'node:stream'
 export function monitorStream(dataStream: Readable) {
   const speedMonitor = monitorStreamSpeed(dataStream)
   const byteCounter = createByteCounterStream()
+  const span = trace.getActiveSpan()
 
-  let measures: number[] = []
+  let measures: any[] = []
 
   // Handle the 'speed' event to collect speed measurements
   speedMonitor.on('speed', (bps) => {
-    measures.push(bps)
-    const span = trace.getActiveSpan()
-    span?.setAttributes({ 'stream.speed': measures, bytesRead: byteCounter.bytes })
+    measures.push({
+      'stream.speed': bps,
+      'stream.bytesRead': byteCounter.bytes,
+      'stream.status': {
+        paused: dataStream.isPaused(),
+        closed: dataStream.closed,
+      },
+    })
+
+    span?.setAttributes({
+      stream: JSON.stringify(measures),
+    })
   })
 
   speedMonitor.on('close', () => {
     measures = []
-    const span = trace.getActiveSpan()
     span?.setAttributes({ uploadRead: byteCounter.bytes })
   })
 

src/internal/streams/stream-speed.ts

@@ -1,35 +1,32 @@
 import { Readable } from 'stream'
 import { PassThrough } from 'node:stream'
 
-/**
- * Keep track of a stream's speed
- * @param stream
- * @param frequency
- */
 /**
  * Keep track of a stream's speed
  * @param stream
  * @param frequency
  */
 export function monitorStreamSpeed(stream: Readable, frequency = 1000) {
-  let totalBytes = 0
-  const startTime = Date.now()
+  let lastIntervalBytes = 0
 
   const passThrough = new PassThrough()
 
-  const interval = setInterval(() => {
-    const currentTime = Date.now()
-    const elapsedTime = (currentTime - startTime) / 1000
-    const currentSpeedBytesPerSecond = totalBytes / elapsedTime
-
+  const emitSpeed = () => {
+    const currentSpeedBytesPerSecond = lastIntervalBytes / (frequency / 1000)
     passThrough.emit('speed', currentSpeedBytesPerSecond)
+    lastIntervalBytes = 0 // Reset for the next interval
+  }
+
+  const interval = setInterval(() => {
+    emitSpeed()
   }, frequency)
 
   passThrough.on('data', (chunk) => {
-    totalBytes += chunk.length
+    lastIntervalBytes += chunk.length // Increment bytes for the current interval
   })
 
   const cleanup = () => {
+    emitSpeed()
     clearInterval(interval)
     passThrough.removeAllListeners('speed')
   }