LiveQueriesCollector bug fix | P95 sorting | MongoPoolCollector listener cleanup | shared utils | container memory fix

Author: mvogttech

.versions

@@ -46,7 +46,7 @@ reactive-var@1.0.13
 reload@1.3.2
 retry@1.1.1
 routepolicy@1.1.2
-skysignal:agent@1.0.17
+skysignal:agent@1.0.18
 socket-stream-client@0.6.1
 tracker@1.3.4
 typescript@5.9.3

README.md

@@ -854,6 +854,16 @@ Main agent singleton instance.
 
 ## Changelog
 
+### v1.0.19 (Bug Fixes & Code Quality)
+
+- **Fix container memory usage reporting >100%** - `SystemMetricsCollector` previously calculated container memory as `processMemory.rss / constrainedMemory * 100`, which could exceed 100% because RSS (Resident Set Size) includes shared library pages, memory-mapped files, and kernel page cache that don't count against the container's cgroup memory limit. Now uses `process.availableMemory()` (Node 19+), which reads directly from the cgroup memory controller and accounts for reclaimable buffers, to compute usage as `(constrainedMemory - availableMemory) / constrainedMemory * 100`. Falls back to `heapUsed / constrainedMemory * 100` on older Node versions. This aligns reported memory with what container orchestrators (e.g., Meteor Galaxy) actually report.
+- **Fix observer stop logging crash** - `LiveQueriesCollector._wrapHandle()` used `this._log()` inside a regular `function()` callback where `this` referred to the handle object, not the collector instance. Changed to `self._log()` to use the captured closure variable. Previously, calling `handle.stop()` would throw `TypeError: this._log is not a function`, silently preventing observer lifecycle metrics from being recorded.
+- **Fix P95 percentile off-by-one** - `MongoPoolCollector`, `DnsTimingCollector`, and `DiagnosticsChannelCollector` all used `Math.floor(count * 0.95)` to index into a sorted array, which overshoots the true 95th percentile by one position (e.g., for 100 items, returns the 96th element instead of the 95th). Changed to `Math.ceil(count * p) - 1` across all three collectors. Extracted to shared `percentile()` utility in `lib/utils/percentile.js`.
+- **Fix MongoPoolCollector.stop() killing other event listeners** - `stop()` called `client.removeAllListeners(eventName)` for each pool event, which removed ALL listeners for that event — including those registered by the application or other collectors. Now stores individual handler references in `start()` and calls `client.removeListener(eventName, handler)` in `stop()` to remove only the collector's own handlers.
+- **Fix circular buffer read after wrap-around** - `MongoPoolCollector._calculateCheckoutMetrics()` used `checkoutSamples.slice(0, count)` to extract samples, which returns incorrect data after the circular buffer wraps (old data mixed with new). Now correctly reads from the current write index forward using modular arithmetic to reconstruct the proper time-ordered sequence.
+- **Shared percentile utility** - Extracted percentile calculation to `lib/utils/percentile.js` with `percentile(sorted, p)` and `percentiles(values)` functions, replacing duplicated math in `MongoPoolCollector`, `DnsTimingCollector`, and `DiagnosticsChannelCollector`.
+- **Shared buffer eviction utility** - Extracted array trimming to `lib/utils/buffer.js` with `trimToMaxSize(array, maxSize)`, replacing duplicated `splice(0, length - max)` patterns in `DnsTimingCollector`, `DiagnosticsChannelCollector`, and `MongoPoolCollector._recordPoolWaitTime`.
+
 ### v1.0.18 (Container-Aware Metrics)
 
 - **Container-aware memory usage** - `SystemMetricsCollector` now uses `process.constrainedMemory()` (Node 19+) to detect cgroup memory limits in containerized deployments. When a cgroup limit is present, memory usage is calculated as `processMemory.rss / constrainedMemory * 100` instead of `(os.totalmem() - os.freemem()) / os.totalmem() * 100`. The OS-level calculation counts kernel buffer/cache as "used", which dramatically overstates actual memory pressure in containers (e.g., reporting 89% when real RSS usage is 27%).

lib/collectors/DiagnosticsChannelCollector.js

@@ -12,6 +12,9 @@
  * used by OpenTelemetry, Undici, and Node's native fetch.
  */
 
+import { percentile } from "../utils/percentile";
+import { trimToMaxSize } from "../utils/buffer";
+
 let diagnostics_channel;
 try {
   diagnostics_channel = require("diagnostics_channel");
@@ -223,10 +226,8 @@ export default class DiagnosticsChannelCollector {
       timestamp: Date.now(),
     });
 
-    // Drop oldest entries in-place instead of allocating a new array
-    if (this._requests.length > this._maxRequests) {
-      this._requests.splice(0, this._requests.length - this._maxRequests);
-    }
+    // Keep buffer bounded
+    trimToMaxSize(this._requests, this._maxRequests);
   }
 
   _collect() {
@@ -265,9 +266,9 @@ export default class DiagnosticsChannelCollector {
 
     // Percentiles
     times.sort((a, b) => a - b);
-    const p50 = times[Math.floor(times.length * 0.5)] || 0;
-    const p95 = times[Math.floor(times.length * 0.95)] || 0;
-    const p99 = times[Math.floor(times.length * 0.99)] || 0;
+    const p50 = percentile(times, 0.5);
+    const p95 = percentile(times, 0.95);
+    const p99 = percentile(times, 0.99);
 
     // Build top external hosts
     const topHosts = Object.entries(byHost)

lib/collectors/DnsTimingCollector.js

@@ -1,5 +1,7 @@
 import dns from "dns";
 import { performance } from "perf_hooks";
+import { percentile } from "../utils/percentile";
+import { trimToMaxSize } from "../utils/buffer";
 
 /**
  * DnsTimingCollector
@@ -129,10 +131,8 @@ export default class DnsTimingCollector {
       timestamp: Date.now(),
     });
 
-    // Keep buffer bounded — drop oldest entries in-place instead of allocating a new array
-    if (this._samples.length > this._maxSamples) {
-      this._samples.splice(0, this._samples.length - this._maxSamples);
-    }
+    // Keep buffer bounded
+    trimToMaxSize(this._samples, this._maxSamples);
   }
 
   _collect() {
@@ -164,9 +164,9 @@ export default class DnsTimingCollector {
 
     // Calculate percentiles
     durations.sort((a, b) => a - b);
-    const p50 = durations[Math.floor(durations.length * 0.5)] || 0;
-    const p95 = durations[Math.floor(durations.length * 0.95)] || 0;
-    const p99 = durations[Math.floor(durations.length * 0.99)] || 0;
+    const p50 = percentile(durations, 0.5);
+    const p95 = percentile(durations, 0.95);
+    const p99 = percentile(durations, 0.99);
 
     // Build top hostnames by count
     const topHostnames = Object.entries(byHostname)

lib/collectors/LiveQueriesCollector.js

@@ -340,7 +340,7 @@ export default class LiveQueriesCollector {
 					observer.status = "stopped";
 					observer.stoppedAt = Date.now();
 					observer.observerLifespan = Math.round((observer.stoppedAt - observer.createdAt) / 1000);
-					this._log(`Observer stopped: ${observer.collectionName} - ${observerId}`);
+					self._log(`Observer stopped: ${observer.collectionName} - ${observerId}`);
 				}
 			}
 			return originalStop.call(this);

lib/collectors/MongoPoolCollector.js

@@ -1,5 +1,7 @@
 import { Meteor } from 'meteor/meteor';
 import { MongoInternals } from 'meteor/mongo';
+import { percentile } from '../utils/percentile';
+import { trimToMaxSize } from '../utils/buffer';
 
 /**
  * MongoPoolCollector
@@ -311,39 +313,21 @@ export default class MongoPoolCollector {
       // (connectionPoolCreated event already fired before we started)
       this._capturePoolConfigFromClient();
 
-      // Track pool creation and configuration (for reconnections)
-      this.client.on('connectionPoolCreated', (event) => {
-        this._onPoolCreated(event);
-      });
-
-      // Track connection lifecycle
-      this.client.on('connectionCreated', (event) => {
-        this._onConnectionCreated(event);
-      });
-
-      this.client.on('connectionClosed', (event) => {
-        this._onConnectionClosed(event);
-      });
-
-      // Track when connection is checked in (returned to pool)
-      this.client.on('connectionCheckedIn', (event) => {
-        this._onConnectionCheckedIn(event);
-      });
-
-      // Track when connection checkout starts (wait begins)
-      this.client.on('connectionCheckOutStarted', (event) => {
-        this._onCheckoutStarted(event);
-      });
-
-      // Track when connection is successfully checked out (wait ends)
-      this.client.on('connectionCheckedOut', (event) => {
-        this._onConnectionCheckedOut(event);
-      });
+      // Store bound handler references so stop() can remove them individually
+      // (removeAllListeners would nuke handlers from other collectors/app code)
+      this._handlers = {
+        connectionPoolCreated: (event) => this._onPoolCreated(event),
+        connectionCreated: (event) => this._onConnectionCreated(event),
+        connectionClosed: (event) => this._onConnectionClosed(event),
+        connectionCheckedIn: (event) => this._onConnectionCheckedIn(event),
+        connectionCheckOutStarted: (event) => this._onCheckoutStarted(event),
+        connectionCheckedOut: (event) => this._onConnectionCheckedOut(event),
+        connectionCheckOutFailed: (event) => this._onCheckoutFailed(event),
+      };
 
-      // Track checkout failures (timeouts, errors)
-      this.client.on('connectionCheckOutFailed', (event) => {
-        this._onCheckoutFailed(event);
-      });
+      for (const [eventName, handler] of Object.entries(this._handlers)) {
+        this.client.on(eventName, handler);
+      }
 
       // Start periodic snapshot collection
       this.snapshotTimer = setInterval(() => {
@@ -374,15 +358,12 @@ export default class MongoPoolCollector {
       this.snapshotTimer = null;
     }
 
-    // Remove all event listeners
-    if (this.client) {
-      this.client.removeAllListeners('connectionPoolCreated');
-      this.client.removeAllListeners('connectionCreated');
-      this.client.removeAllListeners('connectionClosed');
-      this.client.removeAllListeners('connectionCheckedIn');
-      this.client.removeAllListeners('connectionCheckOutStarted');
-      this.client.removeAllListeners('connectionCheckedOut');
-      this.client.removeAllListeners('connectionCheckOutFailed');
+    // Remove only our own event listeners (not all listeners for the event)
+    if (this.client && this._handlers) {
+      for (const [eventName, handler] of Object.entries(this._handlers)) {
+        this.client.removeListener(eventName, handler);
+      }
+      this._handlers = null;
     }
 
     // Clear state
@@ -631,10 +612,8 @@ export default class MongoPoolCollector {
         ...metadata
       });
 
-      // Keep only recent samples — batch eviction instead of per-item shift() which is O(n)
-      if (global._skySignalPoolWaitTimes.length > 1100) {
-        global._skySignalPoolWaitTimes.splice(0, global._skySignalPoolWaitTimes.length - 1000);
-      }
+      // Keep only recent samples
+      trimToMaxSize(global._skySignalPoolWaitTimes, 1000);
 
     } catch (error) {
       this._warn(' Error recording pool wait time:', error);
@@ -918,8 +897,17 @@ export default class MongoPoolCollector {
       };
     }
 
-    // Extract valid samples from circular buffer
-    const samples = this.poolState.checkoutSamples.slice(0, count);
+    // Extract valid samples from circular buffer in correct order
+    const buffer = this.poolState.checkoutSamples;
+    const idx = this.poolState.checkoutSampleIndex;
+    let samples;
+    if (count < 1000) {
+      // Buffer hasn't wrapped yet — first `count` elements are valid
+      samples = buffer.slice(0, count);
+    } else {
+      // Buffer has wrapped — read from current index forward (oldest first)
+      samples = buffer.slice(idx).concat(buffer.slice(0, idx));
+    }
 
     // Calculate average
     let sum = 0;
@@ -933,8 +921,7 @@ export default class MongoPoolCollector {
 
     // Calculate P95 (95th percentile) - only sort when needed
     const sorted = samples.slice().sort((a, b) => a - b);
-    const p95Index = Math.floor(count * 0.95);
-    const p95CheckoutTime = sorted[p95Index] || 0;
+    const p95CheckoutTime = percentile(sorted, 0.95);
 
     return {
       avgCheckoutTime,

lib/collectors/SystemMetricsCollector.js

@@ -10,7 +10,7 @@ import v8 from "v8";
 const execAsync = promisify(exec);
 
 // Agent version - must be updated alongside package.js on each release
-const AGENT_VERSION = '1.0.18';
+const AGENT_VERSION = '1.0.19';
 
 /**
  * SystemMetricsCollector
@@ -181,12 +181,25 @@ export default class SystemMetricsCollector {
       // Constrained memory (cgroup limit, Node 19+)
       const constrainedMemory = this._getConstrainedMemory();
 
-      // Container-aware memory: os.freemem() excludes reclaimable buffer/cache,
-      // overstating memory pressure. When a cgroup limit is available, use
-      // RSS / cgroup-limit which matches what container orchestrators report.
+      // Container-aware memory: when a cgroup limit is available, compute
+      // actual usage via process.availableMemory() (Node 19+) which accounts
+      // for reclaimable cache/buffers. Falls back to heapUsed / limit.
+      // NOTE: RSS is NOT suitable here — it includes shared library pages and
+      // memory-mapped files, often exceeding the cgroup limit (producing >100%).
       let effectiveMemoryUsage = memoryUsage;
       if (constrainedMemory && constrainedMemory > 0) {
-        effectiveMemoryUsage = (processMemory.rss / constrainedMemory) * 100;
+        let availableMemory = null;
+        try {
+          if (typeof process.availableMemory === "function") {
+            availableMemory = process.availableMemory();
+          }
+        } catch (e) { /* ignore */ }
+
+        if (availableMemory != null && availableMemory >= 0) {
+          effectiveMemoryUsage = ((constrainedMemory - availableMemory) / constrainedMemory) * 100;
+        } else {
+          effectiveMemoryUsage = (processMemory.heapUsed / constrainedMemory) * 100;
+        }
       }
 
       // Process-level CPU: measures actual Node.js CPU consumption rather than

lib/utils/buffer.js

@@ -0,0 +1,13 @@
+/**
+ * Trim an array to maxSize by dropping the oldest (first) entries in-place.
+ * Only triggers when the array exceeds maxSize, and drops the excess plus a
+ * small batch margin (10%) to avoid triggering on every single push.
+ *
+ * @param {Array} array - The array to trim (mutated in place)
+ * @param {number} maxSize - Maximum allowed length
+ */
+export function trimToMaxSize(array, maxSize) {
+  if (array.length > maxSize) {
+    array.splice(0, array.length - maxSize);
+  }
+}

lib/utils/percentile.js

@@ -0,0 +1,32 @@
+/**
+ * Calculate a percentile value from a sorted array.
+ * Uses the ceiling method: index = ceil(count * p) - 1
+ *
+ * @param {number[]} sorted - Pre-sorted array of numeric values (ascending)
+ * @param {number} p - Percentile as a fraction (e.g., 0.95 for P95)
+ * @returns {number} The percentile value, or 0 if array is empty
+ */
+export function percentile(sorted, p) {
+  if (!sorted || sorted.length === 0) return 0;
+  const index = Math.ceil(sorted.length * p) - 1;
+  return sorted[Math.max(0, index)] || 0;
+}
+
+/**
+ * Calculate common percentiles (P50, P95, P99) from an unsorted array.
+ * Sorts a copy of the input array.
+ *
+ * @param {number[]} values - Array of numeric values (will not be mutated)
+ * @returns {{ p50: number, p95: number, p99: number }}
+ */
+export function percentiles(values) {
+  if (!values || values.length === 0) {
+    return { p50: 0, p95: 0, p99: 0 };
+  }
+  const sorted = values.slice().sort((a, b) => a - b);
+  return {
+    p50: percentile(sorted, 0.5),
+    p95: percentile(sorted, 0.95),
+    p99: percentile(sorted, 0.99),
+  };
+}

package.js

@@ -1,6 +1,6 @@
 Package.describe({
   name: "skysignal:agent",
-  version: "1.0.18",
+  version: "1.0.19",
   summary:
     "SkySignal APM agent for Meteor applications - monitors performance, errors, and system metrics",
   git: "https://github.com/skysignalapm/agent.git",