Feature/opentelemetry (#89)

* initial adds for otel tracing

* initial basic testing

* basic tracing for incoming calls

* linting

* add traceId to the webhook params

* trace webhook calls

* tracing: add new commands as tags when receiving async commands over websocket

* tracing new commands

* add summary for config verb

* trace async commands

* bugfix: undefined ref

* tracing: give time for final webhooks before closing root span

* tracing bugfix: span for background gather was not ended

* tracing - minor tag changes

* tracing - add span atttribute for reason call ended

* trace call status webhooks, add app version to trace output

* config: add support for automatically re-enabling

* env var to customize service name in tracing UI

* config: change to use 'sticky' attribute to re-enable bargein automatically

* fix warnings

* when adulting create a new root span

* when background gather triggers bargein via vad clear queue of tasks

* additional trace attributes for dial and refer

* fix dial tracing

* add better summary for dial

* fix prev commit

* add exponential backoff to WsRequestor reconnection logic

* add calling number to log metadata, as this will be frequently the key data given for troubleshooting

* add accountSid to log metadata

* make handshake timeout for ws connections configurable with default 1.5 secs

* rename env var

* fix bug prev checkin

* logging fixes

* consistent env naming

lib/utils/call-tracer.js

@@ -0,0 +1,63 @@
+const {context, trace} = require('@opentelemetry/api');
+const {Dialog} = require('drachtio-srf');
+class RootSpan {
+  constructor(callType, req) {
+    let tracer, callSid, linkedSpanId;
+
+    if (req instanceof Dialog) {
+      const dlg = req;
+      tracer = dlg.srf.locals.otel.tracer;
+      callSid = dlg.callSid;
+      linkedSpanId = dlg.linkedSpanId;
+    }
+    else {
+      tracer = req.srf.locals.otel.tracer;
+      callSid = req.locals.callSid;
+    }
+    this._span = tracer.startSpan(callType || 'incoming-call');
+    if (req instanceof Dialog) {
+      const dlg = req;
+      this._span.setAttributes({
+        linkedSpanId,
+        callId: dlg.sip.callId
+      });
+    }
+    else {
+      this._span.setAttributes({
+        callSid,
+        accountSid: req.get('X-Account-Sid'),
+        applicationSid: req.locals.application_sid,
+        callId: req.get('Call-ID'),
+        externalCallId: req.get('X-CID')
+      });
+    }
+
+    this._ctx = trace.setSpan(context.active(), this._span);
+    this.tracer = tracer;
+  }
+
+  get context() {
+    return this._ctx;
+  }
+
+  get traceId() {
+    return this._span.spanContext().traceId;
+  }
+
+  setAttributes(attrs) {
+    this._span.setAttributes(attrs);
+  }
+
+  end() {
+    this._span.end();
+  }
+
+  startChildSpan(name, attributes) {
+    const span = this.tracer.startSpan(name, attributes, this._ctx);
+    const ctx = trace.setSpan(context.active(), span);
+    return {span, ctx};
+  }
+}
+
+module.exports = RootSpan;
+

lib/utils/db-utils.js

@@ -48,6 +48,7 @@ module.exports = (logger, srf) => {
   const pp = pool.promise();
 
   const lookupAccountDetails = async(account_sid) => {
+
     const [r] = await pp.query({sql: sqlAccountDetails, nestTables: true}, account_sid);
     if (0 === r.length) throw new Error(`invalid accountSid: ${account_sid}`);
     const [r2] = await pp.query(sqlSpeechCredentials, account_sid);

lib/utils/install-srf-locals.js

@@ -32,6 +32,7 @@ function initMS(logger, wrapper, ms) {
 function installSrfLocals(srf, logger) {
   logger.debug('installing srf locals');
   assert(!srf.locals.dbHelpers);
+  const {tracer} = srf.locals.otel;
   const {getSBC, lifecycleEmitter} = require('./sbc-pinger')(logger);
   const StatsCollector = require('@jambonz/stats-collector');
   const stats = srf.locals.stats = new StatsCollector(logger);
@@ -127,7 +128,7 @@ function installSrfLocals(srf, logger) {
     password: process.env.JAMBONES_MYSQL_PASSWORD,
     database: process.env.JAMBONES_MYSQL_DATABASE,
     connectionLimit: process.env.JAMBONES_MYSQL_CONNECTION_LIMIT || 10
-  }, logger);
+  }, logger, tracer);
   const {
     client,
     updateCallStatus,
@@ -152,7 +153,7 @@ function installSrfLocals(srf, logger) {
   } = require('@jambonz/realtimedb-helpers')({
     host: process.env.JAMBONES_REDIS_HOST,
     port: process.env.JAMBONES_REDIS_PORT || 6379
-  }, logger);
+  }, logger, tracer);
   const {
     writeAlerts,
     AlertType

lib/utils/place-outdial.js

@@ -9,10 +9,11 @@ const AdultingCallSession = require('../session/adulting-call-session');
 const deepcopy = require('deepcopy');
 const moment = require('moment');
 const stripCodecs = require('./strip-ancillary-codecs');
+const RootSpan = require('./call-tracer');
 const { v4: uuidv4 } = require('uuid');
 
 class SingleDialer extends Emitter {
-  constructor({logger, sbcAddress, target, opts, application, callInfo, accountInfo}) {
+  constructor({logger, sbcAddress, target, opts, application, callInfo, accountInfo, rootSpan, startSpan}) {
     super();
     assert(target.type);
 
@@ -22,6 +23,8 @@ class SingleDialer extends Emitter {
     this.opts = opts;
     this.application = application;
     this.confirmHook = target.confirmHook;
+    this.rootSpan = rootSpan;
+    this.startSpan = startSpan;
 
     this.bindings = logger.bindings();
 
@@ -71,7 +74,7 @@ class SingleDialer extends Emitter {
       };
     }
     this.ms = ms;
-    let uri, to;
+    let uri, to, inviteSpan;
     try {
       switch (this.target.type) {
         case 'phone':
@@ -137,13 +140,24 @@ class SingleDialer extends Emitter {
         localSdp: this.ep.local.sdp
       });
       if (this.target.auth) opts.auth = this.target.auth;
+      inviteSpan = this.startSpan('invite', {
+        'invite.uri': uri,
+        'invite.dest_type': this.target.type
+      });
+
       this.dlg = await srf.createUAC(uri, {...opts,  followRedirects: true, keepUriOnRedirect: true}, {
         cbRequest: (err, req) => {
           if (err) {
             this.logger.error(err, 'SingleDialer:exec Error creating call');
             this.emit('callCreateFail', err);
+            inviteSpan.setAttributes({
+              'invite.status_code': 500,
+              'invite.err': err.message
+            });
+            inviteSpan.end();
             return;
           }
+          inviteSpan.setAttributes({'invite.call_id': req.get('Call-ID')});
 
           /**
            * INVITE has been sent out
@@ -156,7 +170,8 @@ class SingleDialer extends Emitter {
             parentCallInfo: this.parentCallInfo,
             req,
             to,
-            callSid: this.callSid
+            callSid: this.callSid,
+            traceId: this.rootSpan.traceId
           });
           this.logger = srf.locals.parentLogger.child({
             callSid: this.callSid,
@@ -193,6 +208,9 @@ class SingleDialer extends Emitter {
       });
       this.logger.debug(`SingleDialer:exec call connected: ${this.callSid}`);
       const connectTime = this.dlg.connectTime = moment();
+      inviteSpan.setAttributes({'invite.status_code': 200});
+      inviteSpan.end();
+
 
       /* race condition: we were killed just as call was answered */
       if (this.killed) {
@@ -246,10 +264,17 @@ class SingleDialer extends Emitter {
         if (err.status === 487) status.callStatus = CallStatus.NoAnswer;
         else if ([486, 600].includes(err.status)) status.callStatus = CallStatus.Busy;
         this.logger.info(`SingleDialer:exec outdial failure ${err.status}`);
+        inviteSpan.setAttributes({'invite.status_code': err.status});
+        inviteSpan.end();
       }
       else {
         this.logger.error(err, 'SingleDialer:exec');
         status.sipStatus = 500;
+        inviteSpan.setAttributes({
+          'invite.status_code': 500,
+          'invite.err': err.message
+        });
+        inviteSpan.end();
       }
       this.emit('callStatusChange', status);
       if (this.ep) this.ep.destroy();
@@ -305,7 +330,8 @@ class SingleDialer extends Emitter {
         dlg: this.dlg,
         ep: this.ep,
         callInfo: this.callInfo,
-        tasks
+        tasks,
+        rootSpan: this.rootSpan
       });
       await cs.exec();
 
@@ -330,13 +356,18 @@ class SingleDialer extends Emitter {
     else {
       await this.reAnchorMedia();
     }
+
+    this.dlg.callSid = this.callSid;
+    this.dlg.linkedSpanId = this.rootSpan.traceId;
+    const rootSpan = new RootSpan('outbound-call', this.dlg);
     const cs = new AdultingCallSession({
       logger: this.logger,
       singleDialer: this,
       application,
       callInfo: this.callInfo,
       accountInfo: this.accountInfo,
-      tasks
+      tasks,
+      rootSpan
     });
     cs.exec();
     return cs;
@@ -387,9 +418,13 @@ class SingleDialer extends Emitter {
   }
 }
 
-function placeOutdial({logger, srf, ms, sbcAddress, target, opts, application, callInfo, accountInfo}) {
+function placeOutdial({
+  logger, srf, ms, sbcAddress, target, opts, application, callInfo, accountInfo, rootSpan, startSpan
+}) {
   const myOpts = deepcopy(opts);
-  const sd = new SingleDialer({logger, sbcAddress, target, myOpts, application, callInfo, accountInfo});
+  const sd = new SingleDialer({
+    logger, sbcAddress, target, myOpts, application, callInfo, accountInfo, rootSpan, startSpan
+  });
   sd.exec(srf, ms, myOpts);
   return sd;
 }

lib/utils/ws-requestor.js

@@ -15,6 +15,7 @@ class WsRequestor extends BaseRequestor {
     this.messagesInFlight = new Map();
     this.maliciousClient = false;
     this.closedByUs = false;
+    this.backoffMs = 500;
 
     assert(this._isAbsoluteUrl(this.url));
 
@@ -123,10 +124,10 @@ class WsRequestor extends BaseRequestor {
   }
 
   close() {
-    this.logger.info('WsRequestor:close closing socket');
     this.closedByUs = true;
     try {
       if (this.ws) {
+        this.logger.info('WsRequestor:close closing socket');
         this.ws.close();
         this.ws.removeAllListeners();
       }
@@ -144,10 +145,13 @@ class WsRequestor extends BaseRequestor {
   _connect() {
     assert(!this.ws);
     return new Promise((resolve, reject) => {
+      const handshakeTimeout = process.env.JAMBONES_WS_HANDSHAKE_TIMEOUT_MS ?
+        parseInt(process.env.JAMBONES_WS_HANDSHAKE_TIMEOUT_MS) :
+        1500;
       let opts = {
         followRedirects: true,
         maxRedirects: 2,
-        handshakeTimeout: 1000,
+        handshakeTimeout,
         maxPayload: 8096,
       };
       if (this.username && this.password) opts = {...opts, auth: `${this.username}:${this.password}`};
@@ -156,7 +160,7 @@ class WsRequestor extends BaseRequestor {
         .once('ready', (ws) => {
           this.ws = ws;
           this.removeAllListeners('not-ready');
-          if (this.connections++ > 0) this.request('session:reconnect', this.url);
+          if (this.connections > 0) this.request('session:reconnect', this.url);
           resolve();
         })
         .once('not-ready', (err) => {
@@ -213,8 +217,10 @@ class WsRequestor extends BaseRequestor {
 
   _onSocketClosed() {
     this.ws = null;
-    if (this.connections > 0 && this.connections < MAX_RECONNECTS && !this.closedByUs) {
-      setTimeout(this._connect.bind(this), 500);
+    this.emit('connection-dropped');
+    if (this.connections++ > 0 && this.connections < MAX_RECONNECTS && !this.closedByUs) {
+      setTimeout(this._connect.bind(this), this.backoffMs);
+      this.backoffMs = this.backoffMs < 2000 ? this.backoffMs * 2 : (this.backoffMs + 2000);
     }
   }