Add waitedOn metric

.github/workflows/test.yml

@@ -31,7 +31,7 @@ jobs:
           - '--release 2.7.3'
           - '--release 2.8.2'
           - '--release 2.9.0'
-          # Latest version
+          # Left empty to use latest version
           -
     steps:
       - name: Checkout code

.gitignore

@@ -6,3 +6,4 @@ versions.json
 .idea
 .eslintcache
 .envrc
+/packages/*

lib/hijack/async.js

@@ -24,6 +24,7 @@ export function wrapFibers () {
   wrapped = true;
 
   let originalYield = Fibers.yield;
+
   Fibers.yield = function () {
     let kadiraInfo = Kadira._getInfo();
     if (kadiraInfo) {

lib/hijack/wrap_session.js

@@ -5,6 +5,7 @@ import { TimeoutManager } from './timeout_manager';
 
 const MAX_PARAMS_LENGTH = 4000;
 
+
 export function wrapSession (sessionProto) {
   let originalProcessMessage = sessionProto.processMessage;
   sessionProto.processMessage = function (msg) {
@@ -101,6 +102,7 @@ export function wrapSession (sessionProto) {
       response = Kadira.env.kadiraInfo.withValue(kadiraInfo, function () {
         return orginalSubHandler.call(self, msg, unblock);
       });
+
       unblock();
     } else {
       response = orginalSubHandler.call(self, msg, unblock);

lib/models/methods.js

@@ -1,8 +1,10 @@
 import { _ } from 'meteor/underscore';
-const { DDSketch } = require('monti-apm-sketches-js');
 import { KadiraModel } from './0model';
 import { TracerStore } from '../tracer/tracer_store';
 import { Ntp } from '../ntp';
+import { calculateWaitedOnTime } from '../utils';
+
+const { DDSketch } = require('monti-apm-sketches-js');
 
 const METHOD_METRICS_FIELDS = ['wait', 'db', 'http', 'email', 'async', 'compute', 'total'];
 
@@ -55,6 +57,7 @@
       errors: 0,
       fetchedDocSize: 0,
       sentMsgSize: 0,
+      waitedOn: 0,
       histogram: new DDSketch({
         alpha: 0.02
       })
@@ -117,6 +120,14 @@
   methodMetrics.sentMsgSize += size;
 };
 
+MethodsModel.prototype.trackWaitedOn = function (method, messageQueue) {
+  const timestamp = Ntp._now();
+  const dateId = this._getDateId(timestamp);
+
+  let methodMetrics = this._getMetrics(dateId, method);
+  methodMetrics.waitedOn += calculateWaitedOnTime(messageQueue);
+};
+
 /*
   There are two types of data
 
@@ -134,13 +145,13 @@
   this.methodMetricsByMinute = Object.create(null);
 
   // create final payload for methodMetrics
   for (let key in methodMetricsByMinute) {
     const methodMetrics = methodMetricsByMinute[key];
     // converting startTime into the actual serverTime
     let startTime = methodMetrics.startTime;
     methodMetrics.startTime = Kadira.syncedDate.syncTime(startTime);
 
     for (let methodName in methodMetrics.methods) {
       METHOD_METRICS_FIELDS.forEach(function (field) {
         methodMetrics.methods[methodName][field] /=
           methodMetrics.methods[methodName].count;

lib/models/pubsub.js

@@ -1,11 +1,12 @@
 import { Meteor } from 'meteor/meteor';
 import { _ } from 'meteor/underscore';
-const logger = Npm.require('debug')('kadira:pubsub');
-const { DDSketch } = require('monti-apm-sketches-js');
 import { KadiraModel } from './0model';
 import { TracerStore } from '../tracer/tracer_store';
 import { Ntp } from '../ntp';
-import {countKeys, getProperty, iterate} from '../utils';
+import { calculateWaitedOnTime, countKeys, getProperty, iterate } from '../utils';
+
+const logger = Npm.require('debug')('kadira:pubsub');
+const { DDSketch } = require('monti-apm-sketches-js');
 
 export function PubsubModel () {
   this.metricsByMinute = Object.create(null);
@@ -153,6 +154,7 @@
       liveFetchedDocSize: 0,
       initiallySentMsgSize: 0,
       liveSentMsgSize: 0,
+      waitedOn: 0,
       histogram: new DDSketch({
         alpha: 0.02
       })
@@ -231,12 +233,12 @@
   let avgObserverReuse = subscriptionData.avgObserverReuse;
 
   // to the averaging
   for (let dateId in metricsByMinute) {
     let dateMetrics = metricsByMinute[dateId];
     // We need to convert startTime into actual serverTime
     dateMetrics.startTime = Kadira.syncedDate.syncTime(dateMetrics.startTime);
 
     for (let publication in metricsByMinute[dateId].pubs) {
       let singlePubMetrics = metricsByMinute[dateId].pubs[publication];
       // We only calculate resTime for new subscriptions
       singlePubMetrics.resTime /= singlePubMetrics.subs;
@@ -384,3 +386,11 @@
     throw new Error('Kadira: Unknown docs fetched type');
   }
 };
+
+PubsubModel.prototype.trackWaitedOn = function (name, messageQueue) {
+  let timestamp = Ntp._now();
+  let publicationName = this._getPublicationName(name);
+  let metrics = this._getMetrics(timestamp, publicationName);
+
+  metrics.waitedOn += calculateWaitedOnTime(messageQueue);
+};

lib/utils.js

@@ -1,4 +1,5 @@
 import { getClientArchVersion } from './common/utils';
+import { Ntp } from './ntp';
 
 const { DDSketch } = require('monti-apm-sketches-js');
 
@@ -113,3 +114,17 @@ export function pick (obj, keys) {
     return result;
   }, {});
 }
+
+export function calculateWaitedOnTime (messageQueue) {
+  let waitedOnTime = 0;
+
+  const now = Ntp._now();
+
+  messageQueue.toArray().forEach(function (msg) {
+    if (msg._waitEventId) {
+      waitedOnTime += now - msg._waitEventId.at;
+    }
+  });
+
+  return waitedOnTime;
+}

lib/wait_time_builder.js

@@ -100,6 +100,13 @@ export class WaitTimeBuilder {
         }
         delete self._currentProcessingMessages[session.id];
         unblocked = true;
+
+        if (msg.msg === 'method') {
+          Kadira.models.methods.trackWaitedOn(msg.method, session.inQueue);
+        } else if (msg.msg === 'sub') {
+          Kadira.models.pubsub.trackWaitedOn(msg.name, session.inQueue);
+        }
+
         unblock();
 
         TimeoutManager.clearTimeout();

package.js

@@ -123,7 +123,7 @@ function configurePackage (api, isTesting) {
   api.use('meteorhacks:zones@1.2.1', { weak: true });
   api.use('simple:json-routes@2.1.0', { weak: true });
   api.use('zodern:meteor-package-versions@0.2.0');
-  api.use('zodern:types@1.0.9');
+  api.use('zodern:types@1.0.11');
 
   api.use([
     'minimongo', 'mongo', 'ddp', 'ejson', 'ddp-common',

tests/_helpers/helpers.js

@@ -1,12 +1,27 @@
 import { Meteor } from 'meteor/meteor';
 import { Random } from 'meteor/random';
 import { DDP } from 'meteor/ddp';
-const Future = Npm.require('fibers/future');
 import { MethodStore, TestData } from './globals';
 
+const Future = Npm.require('fibers/future');
+
 export const GetMeteorClient = function (_url) {
   const url = _url || Meteor.absoluteUrl();
-  return DDP.connect(url, {retry: false});
+  return DDP.connect(url, {retry: false, });
+};
+
+export const waitForConnection = function (client) {
+  let timeout = Date.now() + 1000;
+  while (Date.now() < timeout) {
+    let status = client.status();
+    if (status.connected) {
+      return;
+    }
+
+    Meteor._sleepForMs(50);
+  }
+
+  throw new Error('timed out waiting for connection');
 };
 
 export const RegisterMethod = function (F) {
@@ -31,7 +46,7 @@ export const EnableTrackingMethods = function () {
   // };
 };
 
-export const GetLastMethodEvents = function (_indices) {
+export const GetLastMethodEvents = function (_indices, ignore = []) {
   if (MethodStore.length < 1) {
     return [];
   }
@@ -43,7 +58,7 @@ export const GetLastMethodEvents = function (_indices) {
   return events;
 
   function isNotCompute (event) {
-    return event[0] !== 'compute';
+    return event[0] !== 'compute' && !ignore.includes(event[0]);
   }
 
   function filterFields (event) {

tests/_helpers/init.js

@@ -64,6 +64,18 @@ Meteor.publish('tinytest-data-random', function () {
   return TestData.find({aa: {$ne: Random.id()}});
 });
 
+Meteor.publish('tinytest-waited-on', function () {
+  Meteor._sleepForMs(25);
+  return TestData.find();
+});
+
+Meteor.publish('tinytest-waited-on2', function () {
+  Meteor._sleepForMs(10);
+  if (this.unblock) this.unblock();
+  Meteor._sleepForMs(40);
+  return TestData.find();
+});
+
 Meteor.publish('tinytest-data-cursor-fetch', function () {
   TestData.find({}).fetch();
   this.ready();

tests/check_for_oplog.js

@@ -22,7 +22,8 @@ Tinytest.addAsync('CheckForOplog - Kadira.checkWhyNoOplog - reactive publish', f
 
       const info = Kadira._getInfo();
 
-      const event = _.last(info.trace.events);
+      const event = _.last(info.trace.events.filter(e => e.type !== 'async'));
+
       observeChangesEvent = _.first(event.nested);
 
       return TestData.find({});

tests/hijack/db.js

@@ -138,7 +138,7 @@ Tinytest.add(
     });
     let client = GetMeteorClient();
     let result = client.call(methodId);
-    let events = GetLastMethodEvents([0, 2]);
+    let events = GetLastMethodEvents([0, 2], ['async']);
     let expected = [
       ['start',undefined,{userId: null, params: '[]'}],
       ['wait',undefined,{waitOn: []}],
@@ -174,7 +174,7 @@ Tinytest.add(
     });
     let client = GetMeteorClient();
     let result = client.call(methodId);
-    let events = GetLastMethodEvents([0, 2]);
+    let events = GetLastMethodEvents([0, 2], ['async']);
     let expected = [
       ['start',undefined,{userId: null, params: '[]'}],
       ['wait',undefined,{waitOn: []}],

tests/hijack/mongo_driver_events.js

@@ -53,7 +53,7 @@ Tinytest.addAsync(
     checkRange(stats.checkoutTime, 0, 100, 40000);
     checkRange(stats.maxCheckoutTime, 0, 10, 300);
     checkRange(stats.pending, 0, 0, 200);
-    checkRange(stats.checkedOut, 0, 0, 15);
+    checkRange(stats.checkedOut, 0, 0, 25);
     checkRange(stats.created, 0, 1, 100);
     done();
   }

tests/models/methods.js

@@ -2,6 +2,8 @@ import { EJSON } from 'meteor/ejson';
 import { MethodsModel } from '../../lib/models/methods';
 import { TestData } from '../_helpers/globals';
 import { CleanTestData, GetMeteorClient, RegisterMethod, Wait, WithDocCacheGetSize } from '../_helpers/helpers';
+import { Meteor } from 'meteor/meteor';
+import { Ntp } from '../../lib/ntp';
 
 Tinytest.add(
   'Models - Method - buildPayload simple',
@@ -21,6 +23,7 @@ Tinytest.add(
               count: 2,
               errors: 0,
               wait: 0,
+              waitedOn: 0,
               db: 0,
               http: 0,
               email: 0,
@@ -68,6 +71,7 @@ Tinytest.add(
           count: 2,
           errors: 1,
           wait: 0,
+          waitedOn: 0,
           db: 0,
           http: 0,
           email: 0,
@@ -186,6 +190,69 @@ Tinytest.add(
   }
 );
 
+Tinytest.addAsync('Models - Method - Waited On - track wait time of queued messages', async (test, done) => {
+  let methodId = RegisterMethod( function (id) {
+    Meteor._sleepForMs(25);
+    return id;
+  });
+
+  let client = GetMeteorClient();
+
+  for (let i = 0; i < 10; i++) {
+    client.call(methodId, i, () => {});
+  }
+
+  Meteor._sleepForMs(1000);
+
+  const metrics = Kadira.models.methods._getMetrics(Ntp._now(), methodId);
+
+  test.isTrue(metrics.waitedOn > 25, `${metrics.waitedOn} should be greater than 25`);
+  test.isTrue(metrics.waitedOn <= 6000, `${metrics.waitedOn} should be less than 6k`);
+  console.log(metrics.waitedOn);
+  done();
+});
+
+Tinytest.addAsync('Models - Method - Waited On - check unblock time', async (test, done) => {
+  let methodId = RegisterMethod( function (id) {
+    this.unblock();
+    Meteor._sleepForMs(25);
+    return id;
+  });
+
+  let client = GetMeteorClient();
+
+  for (let i = 0; i < 10; i++) {
+    client.call(methodId, i, () => {});
+  }
+
+  Meteor._sleepForMs(1000);
+
+  const metrics = Kadira.models.methods._getMetrics(Ntp._now(), methodId);
+
+  test.isTrue(metrics.waitedOn <= 1, 'waitedOn should be less or equal than 1');
+
+  done();
+});
+
+Tinytest.addAsync('Models - Method - Waited On - track wait time of next message', async (test, done) => {
+  let slowMethod = RegisterMethod( function () {
+    Meteor._sleepForMs(25);
+  });
+  let fastMethod = RegisterMethod( function () {});
+
+  let client = GetMeteorClient();
+
+  client.call(slowMethod, () => {});
+  client.call(fastMethod, () => {});
+
+  Meteor._sleepForMs(200);
+
+  const metrics = Kadira.models.methods._getMetrics(Ntp._now(), slowMethod);
+  test.isTrue(metrics.waitedOn >= 20, `${metrics.waitedOn} should be greater than 20`);
+
+  done();
+});
+
 export const model = new MethodsModel();
 
 function CreateMethodCompleted (sessionName, methodName, methodId, startTime, methodDelay) {