bump version

* initial changes to use udp and dns in K8s for rtpengine ng

* github actions

* logging

* remove port from service name before dns lookup

* query rtpengine dns immediately on startup

* bugfix prev checkin

* use dns.lookup instead of resolve4 (k8s does not appear to consult search patterns in resolve.conf with the latter)

* return all rtpengine endpoints in dns lookup

* bugfix: polling rtpengine endpoints

* fix prev commit

* catch error on rtpengine failure

* update deps

* require node 14 in gh action build

* specify node version in gh actions

.github/workflows/npm-ci.yml

@@ -11,8 +11,8 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions/setup-node@v1
         with:
-          node-version: 12
-      - run: npm install
+          node-version: 14.x
+      - run: npm ci
       - run: npm run jslint
       - run: npm test
 

.gitignore

@@ -1,7 +1,7 @@
 # Logs
 logs
 *.log
-
+.vscode/
 # Runtime data
 pids
 *.pid

Dockerfile

@@ -1,4 +1,4 @@
-FROM node:17-slim
+FROM node:17.4-slim
 WORKDIR /opt/app/
 COPY package.json ./
 RUN npm install

app.js

@@ -26,6 +26,7 @@ const {
 });
 const StatsCollector = require('@jambonz/stats-collector');
 const stats = new StatsCollector(logger);
+const {equalsIgnoreOrder} = require('./lib/utils');
 const {LifeCycleEvents} = require('./lib/constants');
 const setNameRtp = `${(process.env.JAMBONES_CLUSTER_ID || 'default')}:active-rtp`;
 const rtpServers = [];
@@ -56,7 +57,7 @@ const {createSet, retrieveSet, addToSet, removeFromSet, incrKey, decrKey} = requ
 const {getRtpEngine, setRtpEngines} = require('@jambonz/rtpengine-utils')([], logger, {
   emitter: stats,
   dtmfListenPort: process.env.DTMF_LISTEN_PORT || 22224,
-  protocol: process.env.RTPENGINE_NG_PROTOCOL || (process.env.K8S ? 'ws' : 'udp')
+  protocol: 'udp'
 });
 srf.locals = {...srf.locals,
   stats,
@@ -170,40 +171,58 @@ srf.use((req, res, next, err) => {
   res.send(500);
 });
 
-const PORT = process.env.HTTP_PORT || 3000;
-const getCount = () => activeCallIds.size;
-const healthCheck = require('@jambonz/http-health-check');
-healthCheck({port: PORT, logger, path: '/', fn: getCount});
+if (process.env.K8S) {
+  const PORT = process.env.HTTP_PORT || 3000;
+  const getCount = () => activeCallIds.size;
+  const healthCheck = require('@jambonz/http-health-check');
+  healthCheck({port: PORT, logger, path: '/', fn: getCount});
+}
+if ('test' !== process.env.NODE_ENV) {
+  /* update call stats periodically */
+  setInterval(() => {
+    stats.gauge('sbc.sip.calls.count', activeCallIds.size, ['direction:inbound']);
+  }, 20000);
+}
 
-/* update call stats periodically */
-setInterval(() => {
-  stats.gauge('sbc.sip.calls.count', activeCallIds.size, ['direction:inbound']);
-}, 20000);
-
-const arrayCompare = (a, b) => {
-  if (a.length !== b.length) return false;
-  const uniqueValues = new Set([...a, ...b]);
-  for (const v of uniqueValues) {
-    const aCount = a.filter((e) => e === v).length;
-    const bCount = b.filter((e) => e === v).length;
-    if (aCount !== bCount) return false;
-  }
-  return true;
+const lookupRtpServiceEndpoints = (lookup, serviceName) => {
+  logger.debug(`dns lookup for ${serviceName}..`);
+  lookup(serviceName, {family: 4, all: true}, (err, addresses) => {
+    if (err) {
+      logger.error({err}, `Error looking up ${serviceName}`);
+      return;
+    }
+    logger.debug({addresses, rtpServers}, `dns lookup for ${serviceName} returned`);
+    const addrs = addresses.map((a) => a.address);
+    if (!equalsIgnoreOrder(addrs, rtpServers)) {
+      rtpServers.length = 0;
+      Array.prototype.push.apply(rtpServers, addrs);
+      logger.info({rtpServers}, 'rtpserver endpoints have been updated');
+      setRtpEngines(rtpServers.map((a) => `${a}:${process.env.RTPENGINE_PORT || 22222}`));
+    }
+  });
 };
 
-const serviceName = process.env.JAMBONES_RTPENGINES || process.env.K8S_RTPENGINE_SERVICE_NAME;
-if (serviceName) {
-  logger.info(`rtpengine(s) will be found at: ${serviceName}`);
-  setRtpEngines([serviceName]);
+if (process.env.K8S_RTPENGINE_SERVICE_NAME) {
+  /* poll dns for endpoints every so often */
+  const arr = /^(.*):(\d+)$/.exec(process.env.K8S_RTPENGINE_SERVICE_NAME);
+  const svc = arr[1];
+  logger.info(`rtpengine(s) will be found at dns name: ${svc}`);
+  const {lookup} = require('dns');
+  lookupRtpServiceEndpoints(lookup, svc);
+  setInterval(lookupRtpServiceEndpoints.bind(null, lookup, svc), process.env.RTPENGINE_DNS_POLL_INTERVAL || 10000);
+}
+else if (process.env.JAMBONES_RTPENGINES) {
+  /* static list of rtpengines */
+  setRtpEngines([process.env.JAMBONES_RTPENGINES]);
 }
 else {
-  /* update rtpengines periodically */
+  /* poll redis periodically for rtpengines that have registered via OPTIONS ping */
   const getActiveRtpServers = async() => {
     try {
       const set = await retrieveSet(setNameRtp);
       const newArray = Array.from(set);
       logger.debug({newArray, rtpServers}, 'getActiveRtpServers');
-      if (!arrayCompare(newArray, rtpServers)) {
+      if (!equalsIgnoreOrder(newArray, rtpServers)) {
         logger.info({newArray}, 'resetting active rtpengines');
         setRtpEngines(newArray.map((a) => `${a}:${process.env.RTPENGINE_PORT || 22222}`));
         rtpServers.length = 0;

lib/call-session.js

@@ -197,7 +197,7 @@ class CallSession extends Emitter {
       this._setHandlers({uas, uac});
       return;
     } catch (err) {
-      this.rtpEngineResource.destroy();
+      this.rtpEngineResource.destroy().catch((err) => this.logger.info({err}, 'Error destroying rtpe after failure'));
       this.activeCallIds.delete(this.req.get('Call-ID'));
       this.stats.gauge('sbc.sip.calls.count', this.activeCallIds.size);
       if (err instanceof SipError) {

lib/utils.js

@@ -45,11 +45,23 @@ const normalizeDID = (tel) => {
   return arr ? arr[1] : tel;
 };
 
+const equalsIgnoreOrder = (a, b) => {
+  if (a.length !== b.length) return false;
+  const uniqueValues = new Set([...a, ...b]);
+  for (const v of uniqueValues) {
+    const aCount = a.filter((e) => e === v).length;
+    const bCount = b.filter((e) => e === v).length;
+    if (aCount !== bCount) return false;
+  }
+  return true;
+};
+
 module.exports = {
   isWSS,
   SdpWantsSrtp,
   getAppserver,
   makeRtpEngineOpts,
   makeCallCountKey,
-  normalizeDID
+  normalizeDID,
+  equalsIgnoreOrder
 };

package.json

@@ -1,9 +1,9 @@
 {
   "name": "sbc-inbound",
-  "version": "v0.7.1",
+  "version": "v0.7.3",
   "main": "app.js",
   "engines": {
-    "node": ">= 10.16.0"
+    "node": ">= 12.0.0"
   },
   "keywords": [
     "sip",
@@ -28,8 +28,8 @@
     "@jambonz/db-helpers": "^0.6.16",
     "@jambonz/http-authenticator": "^0.2.0",
     "@jambonz/http-health-check": "^0.0.1",
-    "@jambonz/realtimedb-helpers": "^0.4.9",
-    "@jambonz/rtpengine-utils": "^0.2.2",
+    "@jambonz/realtimedb-helpers": "^0.4.24",
+    "@jambonz/rtpengine-utils": "^0.3.1",
     "@jambonz/stats-collector": "^0.1.6",
     "@jambonz/time-series": "^0.1.6",
     "aws-sdk": "^2.1036.0",
@@ -46,8 +46,8 @@
   },
   "devDependencies": {
     "clear-module": "^4.1.1",
-    "eslint": "^7.15.0",
-    "eslint-plugin-promise": "^4.2.1",
+    "eslint": "^7.32.0",
+    "eslint-plugin-promise": "^4.3.1",
     "nyc": "^15.1.0",
     "tape": "^4.13.3"
   }