K8s changes (#24)

* support multiple CIDRs

* use outbound in k8s

* k8s: user service for rtpengine

* bugfix: drachtio connection was dropped after successful connect

* Dockerfile

* k8s pre-stop hook

* make hjook executable

* update deps

Dockerfile

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

app.js

@@ -11,7 +11,6 @@ assert.ok(process.env.JAMBONES_NETWORK_CIDR, 'missing JAMBONES_NETWORK_CIDR env
 const Srf = require('drachtio-srf');
 const srf = new Srf('sbc-outbound');
 const CIDRMatcher = require('cidr-matcher');
-const matcher = new CIDRMatcher([process.env.JAMBONES_NETWORK_CIDR]);
 const opts = Object.assign({
   timestamp: () => {return `, "time": "${new Date().toISOString()}"`;}
 }, {level: process.env.JAMBONES_LOGLEVEL || 'info'});
@@ -51,6 +50,12 @@ const {createHash, retrieveHash, incrKey, decrKey, retrieveSet} = require('@jamb
   port: process.env.JAMBONES_REDIS_PORT || 6379
 }, logger);
 
+const cidrs = process.env.JAMBONES_NETWORK_CIDR
+  .split(',')
+  .map((s) => s.trim());
+logger.info({cidrs}, 'internal network CIDRs');
+const matcher = new CIDRMatcher(cidrs);
+
 const activeCallIds = new Map();
 
 srf.locals = {...srf.locals,
@@ -86,7 +91,7 @@ const {getRtpEngine, setRtpEngines} = require('@jambonz/rtpengine-utils')([], lo
 });
 srf.locals.getRtpEngine = getRtpEngine;
 
-if (process.env.DRACHTIO_HOST) {
+if (process.env.DRACHTIO_HOST && !process.env.K8S) {
   srf.connect({host: process.env.DRACHTIO_HOST, port: process.env.DRACHTIO_PORT, secret: process.env.DRACHTIO_SECRET });
   srf.on('connect', (err, hp) => {
     logger.info(`connected to drachtio listening on ${hp}`);
@@ -107,6 +112,7 @@ if (process.env.DRACHTIO_HOST) {
   });
 }
 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});
 }
 if (process.env.NODE_ENV === 'test') {
@@ -121,6 +127,11 @@ srf.invite((req, res) => {
   session.connect();
 });
 
+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});
+
 /* update call stats periodically */
 setInterval(() => {
   stats.gauge('sbc.sip.calls.count', activeCallIds.size, ['direction:outbound']);
@@ -137,11 +148,13 @@ const arrayCompare = (a, b) => {
   return true;
 };
 
-/* update rtpengines periodically */
-if (process.env.JAMBONES_RTPENGINES) {
-  setRtpEngines([process.env.JAMBONES_RTPENGINES]);
+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]);
 }
 else {
+  /* update rtpengines periodically */
   const getActiveRtpServers = async() => {
     try {
       const set = await retrieveSet(setNameRtp);

bin/k8s-pre-stop-hook.js

@@ -0,0 +1,29 @@
+#!/usr/bin/env node
+const bent = require('bent');
+const getJSON = bent('json');
+const PORT = process.env.HTTP_PORT || 3000;
+
+const sleep = (ms) => {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+};
+
+(async function() {
+
+  try {
+    do {
+      const obj = await getJSON(`http://127.0.0.1:${PORT}/`);
+      const {calls} = obj;
+      if (calls === 0) {
+        console.log('no calls on the system, we can exit');
+        process.exit(0);
+      }
+      else {
+        console.log(`waiting for ${calls} to exit..`);
+      }
+      await sleep(10000);
+    } while (1);
+  } catch (err) {
+    console.error(err, 'Error querying health endpoint');
+    process.exit(-1);
+  }
+})();

lib/call-session.js

@@ -208,7 +208,8 @@ class CallSession extends Emitter {
         } catch (err) {
           debug(err);
           this.logger.error(err, 'Error performing lcr');
-          return this.res.send(488);
+          this.res.send(488);
+          return this.srf.endSession(this.req);
         }
         debug(`sending call to PSTN ${uris}`);
       }
@@ -375,8 +376,9 @@ class CallSession extends Emitter {
         }
       }
     } catch (err) {
-      if ('abandonded' !== err.message) this.logger.error(err, `Error setting up outbonund call to: ${uris}`);
+      if ('abandonded' !== err.message) this.logger.error(err, `Error setting up outbound call to: ${uris}`);
       this.emit('failed');
+      this.srf.endSession(this.req);
       this.rtpEngineResource.destroy();
     }
   }
@@ -422,6 +424,7 @@ class CallSession extends Emitter {
             duration: Math.floor((now - callStart) / 1000)
           }).catch((err) => this.logger.error({err}, 'Error writing cdr for completed call'));
         }
+        this.srf.endSession(this.req);
       });
     });
 
@@ -598,6 +601,7 @@ Duration=${payload.duration} `
         this.rtpEngineResource.destroy();
         this.activeCallIds.delete(this.req.get('Call-ID'));
         this.uas.other.destroy();
+        this.srf.endSession(this.req);
       });
 
       // modify rtpengine to stream to new feature server

lib/middleware.js

@@ -26,11 +26,12 @@ module.exports = (srf, logger, opts) => {
 
     if (!req.locals.account_sid) {
       logger.info('missing X-Account-Sid on outbound call');
-      return res.send(403, {
+      res.send(403, {
         headers: {
           'X-Reason': 'missing X-Account-Sid'
         }
       });
+      return req.srf.endSession(req);
     }
 
     stats.increment('sbc.invites', ['direction:outbound']);
@@ -47,7 +48,8 @@ module.exports = (srf, logger, opts) => {
       req.locals.account = await lookupAccountBySid(req.locals.account_sid);
     } catch (err) {
       req.locals.logger.error({err}, `Error looking up account sid ${req.locals.account_sid}`);
-      return res.send(500);
+      res.send(500);
+      return req.srf.endSession(req);
     }
     next();
   };
@@ -106,6 +108,7 @@ module.exports = (srf, logger, opts) => {
         }).catch((err) => logger.info({err}, 'checkLimits: error writing alert'));
 
         res.send(503, 'Maximum Calls In Progress');
+        return req.srf.endSession(req);
       }
       next();
     } catch (err) {
@@ -127,7 +130,7 @@ module.exports = (srf, logger, opts) => {
           'X-Reason': 'invalid request-uri'
         }
       });
-      return;
+      return req.srf.endSession(req);
     }
     const aor = `${uri.user}@${uri.host}`;
     let reg;
@@ -166,7 +169,7 @@ module.exports = (srf, logger, opts) => {
         if (account) {
           logger.info({host: uri.host, account}, `returning 404 to unregistered user in valid domain: ${req.uri}`);
           res.send(404);
-          return;
+          return req.srf.endSession(req);
         }
       }
     }
@@ -181,7 +184,8 @@ module.exports = (srf, logger, opts) => {
     if (!/^\d+$/.test(req.calledNumber.slice(1))) {
       debug(`unable to route call to ${aor}; no registered user found`);
       logger.info(`unable to route call to ${aor}; no registered user found`);
-      return res.send(404);
+      res.send(404);
+      return req.srf.endSession(req);
     }
 
     debug('sending call to LCR');

package.json

@@ -27,7 +27,8 @@
     "jslint": "eslint app.js lib"
   },
   "dependencies": {
-    "@jambonz/db-helpers": "^0.6.14",
+    "@jambonz/db-helpers": "^0.6.16",
+    "@jambonz/http-health-check": "^0.0.1",
     "@jambonz/mw-registrar": "0.2.1",
     "@jambonz/realtimedb-helpers": "^0.4.9",
     "@jambonz/rtpengine-utils": "^0.1.21",