K8s scale in (#73)

* changes for proper scale-in for K8S

* fix check for call count

.github/workflows/docker-publish.yml

@@ -20,7 +20,7 @@ jobs:
     if: github.event_name == 'push'
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
 
       - name: Build image
         run: docker build . --file Dockerfile --tag $IMAGE_NAME

.github/workflows/npm-ci.yml

@@ -1,15 +1,13 @@
 name: CI
 
-on:
-  push:
-  workflow_dispatch:
+on: [push, pull_request]
 
 jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
-      - uses: actions/setup-node@v1
+      - uses: actions/checkout@v3
+      - uses: actions/setup-node@v3
         with:
           node-version: 18.x
       - run: npm ci

app.js

@@ -28,6 +28,7 @@ const {
   commitInterval: 'test' === process.env.NODE_ENV ? 7 : 20
 });
 const StatsCollector = require('@jambonz/stats-collector');
+const CIDRMatcher = require('cidr-matcher');
 const stats = new StatsCollector(logger);
 const {equalsIgnoreOrder, createHealthCheckApp, systemHealth} = require('./lib/utils');
 const {LifeCycleEvents} = require('./lib/constants');
@@ -129,7 +130,6 @@ const {
 const CallSession = require('./lib/call-session');
 
 if (process.env.DRACHTIO_HOST && !process.env.K8S) {
-  const CIDRMatcher = require('cidr-matcher');
   const cidrs = process.env.JAMBONES_NETWORK_CIDR
     .split(',')
     .map((s) => s.trim());
@@ -164,6 +164,23 @@ else {
     logger.info(`listening in outbound mode on port ${process.env.DRACHTIO_PORT}`);
   });
   srf.listen({port: process.env.DRACHTIO_PORT, secret: process.env.DRACHTIO_SECRET});
+  srf.on('connect', (err, hp) => {
+    if (err) return this.logger.error({err}, 'Error connecting to drachtio server');
+    logger.info(`connected to drachtio listening on ${hp}`);
+
+    if (process.env.K8S_FEATURE_SERVER_TRANSPORT === 'tcp') {
+      const matcher = new CIDRMatcher(['192.168.0.0/24', '172.16.0.0/16', '10.0.0.0/8']);
+      const hostports = hp.split(',');
+      for (const hp of hostports) {
+        const arr = /^(.*)\/(.*):(\d+)$/.exec(hp);
+        if (arr && matcher.contains(arr[2])) {
+          const hostport = `${arr[2]}:${arr[3]}`;
+          logger.info(`using sbc private address when sending to feature-server: ${hostport}`);
+          srf.locals.privateSipAddress = hostport;
+        }
+      }
+    }
+  });
 }
 if (process.env.NODE_ENV === 'test') {
   srf.on('error', (err) => {
@@ -235,7 +252,6 @@ if ('test' !== process.env.NODE_ENV) {
 }
 
 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}`);
@@ -305,8 +321,18 @@ process.on('SIGUSR2', handle.bind(null, removeFromSet, setName));
 process.on('SIGTERM', handle.bind(null, removeFromSet, setName));
 
 function handle(removeFromSet, setName, signal) {
-  logger.info(`got signal ${signal}, removing ${srf.locals.privateSipAddress} from set ${setName}`);
-  removeFromSet(setName, srf.locals.privateSipAddress);
+  logger.info(`got signal ${signal}`);
+  if (srf.locals.privateSipAddress && setName) {
+    logger.info(`removing ${srf.locals.privateSipAddress} from set ${setName}`);
+    removeFromSet(setName, srf.locals.privateSipAddress);
+  }
+  if (process.env.K8S) {
+    lifecycleEmitter.operationalState = LifeCycleEvents.ScaleIn;
+    if (0 === activeCallIds.size) {
+      logger.info('exiting immediately since we have no calls in progress');
+      process.exit(0);
+    }
+  }
 }
 
 module.exports = {srf, logger};

data/srtp-transcoding.json

@@ -8,7 +8,7 @@
   },
   "teams": {
     "transport-protocol": "RTP/SAVP",
-    "ICE": "force",
+    "ICE": "default",
     "SDES": "off",
     "flags": ["generate mid", "SDES-no", "media handover", "port latching"],
     "rtcp-mux": ["accept"]

lib/autoscale-manager.js

@@ -57,6 +57,9 @@ module.exports = (logger) => {
       }
     })();
   }
+  else if (process.env.K8S) {
+    lifecycleEmitter.scaleIn = () => process.exit(0);
+  }
 
   return {lifecycleEmitter};
 };

lib/call-session.js

@@ -3,6 +3,7 @@ const SrsClient = require('@jambonz/siprec-client-utils');
 const {
   makeRtpEngineOpts,
   SdpWantsSrtp,
+  SdpWantsSDES,
   nudgeCallCounts,
   roundTripTime,
   parseConnectionIp
@@ -84,10 +85,6 @@ class CallSession extends Emitter {
     return this.req.has('X-Five9-StreamingPairId');
   }
 
-  get isPossibleWebRtcClient() {
-    return this.req.locals.isPossibleWebRtcClient;
-  }
-
   subscribeForDTMF(dlg) {
     if (!this._subscribedForDTMF) {
       this._subscribedForDTMF = true;
@@ -152,7 +149,8 @@ class CallSession extends Emitter {
     this.logger.debug(`using feature server ${featureServer}`);
 
     const wantsSrtp = this.req.locals.possibleWebRtcClient = SdpWantsSrtp(sdp);
-    this.rtpEngineOpts = makeRtpEngineOpts(this.req, wantsSrtp, false, this.isFromMSTeams);
+    const wantsSDES = SdpWantsSDES(sdp);
+    this.rtpEngineOpts = makeRtpEngineOpts(this.req, wantsSrtp, false, this.isFromMSTeams || wantsSDES);
     this.rtpEngineResource = {destroy: this.del.bind(null, this.rtpEngineOpts.common)};
     const obj = parseUri(this.req.uri);
     let proxy, host, uri;
@@ -262,6 +260,7 @@ class CallSession extends Emitter {
           };
           const startAt = process.hrtime();
           const response = await this.answer(opts);
+          this.logger.debug({response, opts}, 'response from rtpengine to answer');
           const rtt = roundTripTime(startAt);
           this.stats.histogram('app.rtpengine.response_time', rtt, [
             'direction:inbound', 'command:answer', `rtpengine:${this.rtpengineIp}`]);
@@ -309,7 +308,7 @@ class CallSession extends Emitter {
   _setDlgHandlers(dlg) {
     const {callId} = dlg.sip;
     this.activeCallIds.set(callId, this);
-    if (this.isPossibleWebRtcClient) this.subscribeForDTMF(dlg);
+    this.subscribeForDTMF(this.uac);
     dlg.on('destroy', () => {
       debug('call ended with normal termination');
       this.logger.info('call ended with normal termination');
@@ -332,11 +331,11 @@ class CallSession extends Emitter {
   _setHandlers({uas, uac}) {
     this.emit('connected');
     const callStart = Date.now();
+    const call_sid = uac.res?.get('X-Call-Sid');
+    const application_sid = this.application_sid || uac.res?.get('X-Application-Sid');
     const tags = ['accepted:yes', 'sipStatus:200', `originator:${this.req.locals.originator}`];
     this.stats.increment('sbc.terminations', tags);
     this.activeCallIds.set(this.req.get('Call-ID'), this);
-    const call_sid = uac.res?.get('X-Call-Sid');
-    const application_sid = this.application_sid || uac.res?.get('X-Application-Sid');
     if (this.req.locals.cdr) {
       this.req.locals.cdr = {
         ...this.req.locals.cdr,
@@ -414,7 +413,7 @@ class CallSession extends Emitter {
       });
     });
 
-    if (this.isPossibleWebRtcClient) this.subscribeForDTMF(uac);
+    this.subscribeForDTMF(uac);
 
     uas.on('modify', this._onReinvite.bind(this, uas));
     uac.on('modify', this._onReinvite.bind(this, uac));
@@ -541,7 +540,7 @@ Duration=${payload.duration} `
         req.body;
 
       const reason = req.get('X-Reason');
-      const isReleasingMedia = reason && dlg.type === 'uas' && ['release-media', 'anchor-media'].includes(reason);
+      const isReleasingMedia = reason && dlg.type === 'uac' && ['release-media', 'anchor-media'].includes(reason);
       const fromTag = dlg.type === 'uas' ? this.rtpEngineOpts.uas.tag : this.rtpEngineOpts.uac.tag;
       const toTag = dlg.type === 'uas' ? this.rtpEngineOpts.uac.tag : this.rtpEngineOpts.uas.tag;
       const offerMedia = dlg.type === 'uas' ? this.rtpEngineOpts.uac.mediaOpts : this.rtpEngineOpts.uas.mediaOpts;
@@ -774,30 +773,38 @@ Duration=${payload.duration} `
         const referredBy = req.getParsedHeader('Referred-By');
         if (!referredBy) return res.send(400);
         const u = parseUri(referredBy.uri);
-        const isDotDecimal = /^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$/.test(uri.host);
-        let selectedGateway = false;
-        let e164 = false;
-        if (gateway && isDotDecimal) {
-          /* host of Refer-to to an outbound gateway */
-          const gw = await this.srf.locals.getOutboundGatewayForRefer(gateway.voip_carrier_sid);
-          if (gw) {
-            selectedGateway = true;
-            e164 = gw.e164_leading_plus;
-            uri.host = gw.ipv4;
-            uri.port = gw.port;
+        const leaveReferToAlone = req.has('X-Refer-To-Leave-Untouched');
+        if (leaveReferToAlone) {
+          this.logger.debug({referTo}, 'passing Refer-To header through untouched');
+        }
+        else {
+          const isDotDecimal = /^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$/.test(uri.host);
+          let selectedGateway = false;
+          let e164 = false;
+          if (gateway && isDotDecimal) {
+            /* host of Refer-to to an outbound gateway */
+            const gw = await this.srf.locals.getOutboundGatewayForRefer(gateway.voip_carrier_sid);
+            if (gw) {
+              selectedGateway = true;
+              e164 = gw.e164_leading_plus;
+              uri.host = gw.ipv4;
+              uri.port = gw.port;
+            }
+          }
+          if (!selectedGateway && isDotDecimal) {
+            uri.host = this.req.source_address;
+            uri.port = this.req.source_port;
+          }
+          if (e164 && !uri.user.startsWith('+')) {
+            uri.user = `+${uri.user}`;
           }
-        }
-        if (!selectedGateway && isDotDecimal) {
-          uri.host = this.req.source_address;
-          uri.port = this.req.source_port;
-        }
-        if (e164 && !uri.user.startsWith('+')) {
-          uri.user = `+${uri.user}`;
         }
         // eslint-disable-next-line no-unused-vars
         const {via, from, to, 'call-id':callid, cseq, 'max-forwards':maxforwards,
           // eslint-disable-next-line no-unused-vars
           'content-length':contentlength, 'refer-to':_referto, 'referred-by':_referredby,
+          // eslint-disable-next-line no-unused-vars
+          'X-Refer-To-Leave-Untouched': _leave,
           ...customHeaders
         } = req.headers;
 

lib/db-utils.js

@@ -100,7 +100,7 @@ module.exports = (srf, logger) => {
   };
 
   const getApplicationForDidAndCarrier = async(req, voip_carrier_sid) => {
-    const did = normalizeDID(req.calledNumber);
+    const did = normalizeDID(req.calledNumber) || 'anonymous';
 
     try {
       /* straight DID match */
@@ -214,7 +214,7 @@ module.exports = (srf, logger) => {
       if (matches.length) {
         /* we have one or more matches.  Now check for one with a provisioned phone number matching the DID */
         const vc_sids = matches.map((m) => `'${m.voip_carrier_sid}'`).join(',');
-        const did = normalizeDID(req.calledNumber);
+        const did = normalizeDID(req.calledNumber) || 'anonymous';
         const sql =  `SELECT * FROM phone_numbers WHERE number = '${did}' AND voip_carrier_sid IN (${vc_sids})`;
         logger.debug({matches, sql, did, vc_sids}, 'looking up DID');
 
@@ -233,13 +233,15 @@ module.exports = (srf, logger) => {
             };
           }
           else if (accountLevelGateways.length === 1) {
+            const [accounts] = await pp.query('SELECT * from accounts where account_sid = ?',
+              accountLevelGateways[0].account_sid);
             return {
               fromCarrier: true,
               gateway: accountLevelGateways[0],
               service_provider_sid: accountLevelGateways[0].service_provider_sid,
               account_sid: accountLevelGateways[0].account_sid,
               application_sid: accountLevelGateways[0].application_sid,
-              account: accountLevelGateways[0]
+              account: accounts[0]
             };
           }
           else {

lib/fs-tracking.js

@@ -15,7 +15,9 @@ module.exports = (srf, logger) => {
   return async() => {
     try {
       if (process.env.K8S) {
-        return process.env.K8S_FEATURE_SERVER_SERVICE_NAME;
+        return process.env.K8S_FEATURE_SERVER_TRANSPORT ?
+          `${process.env.K8S_FEATURE_SERVER_SERVICE_NAME};transport=${process.env.K8S_FEATURE_SERVER_TRANSPORT}` :
+          process.env.K8S_FEATURE_SERVER_SERVICE_NAME;
       }
       else {
         const fs = await retrieveSet(setName);

lib/utils.js

@@ -26,7 +26,8 @@ function makeRtpEngineOpts(req, srcIsUsingSrtp, dstIsUsingSrtp, teams = false) {
   }
   const common = {
     'call-id': req.get('Call-ID'),
-    'replace': ['origin', 'session-connection']
+    'replace': ['origin', 'session-connection'],
+    'record call': process.env.JAMBONES_RECORD_ALL_CALLS ? 'yes' : 'no'
   };
   return {
     common,
@@ -41,6 +42,9 @@ function makeRtpEngineOpts(req, srcIsUsingSrtp, dstIsUsingSrtp, teams = false) {
   };
 }
 
+const SdpWantsSDES = (sdp) => {
+  return /m=audio.*\s+RTP\/SAVP/.test(sdp);
+};
 const SdpWantsSrtp = (sdp) => {
   return /m=audio.*SAVP/.test(sdp);
 };
@@ -178,6 +182,7 @@ const parseConnectionIp = (sdp) => {
 module.exports = {
   isWSS,
   SdpWantsSrtp,
+  SdpWantsSDES,
   getAppserver,
   makeRtpEngineOpts,
   makeAccountCallCountKey,

package.json

@@ -1,6 +1,6 @@
 {
   "name": "sbc-inbound",
-  "version": "v0.7.7",
+  "version": "v0.7.8",
   "main": "app.js",
   "engines": {
     "node": ">= 12.0.0"
@@ -25,12 +25,12 @@
     "jslint": "eslint app.js lib"
   },
   "dependencies": {
-    "@jambonz/db-helpers": "^0.7.3",
+    "@jambonz/db-helpers": "^0.7.4",
     "@jambonz/http-authenticator": "^0.2.2",
     "@jambonz/http-health-check": "^0.0.1",
-    "@jambonz/realtimedb-helpers": "^0.5.7",
-    "@jambonz/rtpengine-utils": "^0.3.11",
-    "@jambonz/siprec-client-utils": "^0.1.4",
+    "@jambonz/realtimedb-helpers": "^0.6.3",
+    "@jambonz/rtpengine-utils": "^0.4.3",
+    "@jambonz/siprec-client-utils": "^0.2.0",
     "@jambonz/stats-collector": "^0.1.6",
     "@jambonz/time-series": "^0.2.5",
     "aws-sdk": "^2.1261.0",
@@ -38,11 +38,9 @@
     "cidr-matcher": "^2.1.1",
     "debug": "^4.3.4",
     "drachtio-fn-b2b-sugar": "0.0.12",
-    "drachtio-srf": "^4.5.2",
+    "drachtio-srf": "^4.5.21",
     "express": "^4.18.1",
     "pino": "^7.11.0",
-    "sdp-transform": "^2.14.1",
-    "uuid": "^8.3.2",
     "verify-aws-sns-signature": "^0.1.0",
     "xml2js": "^0.4.23"
   },