add a metrics endpoint

Dockerfile

@@ -1 +1 @@
-FROM semtech/mu-javascript-template:1.8.0
+FROM semtech/mu-javascript-template:1.9.1

app.js

@@ -1,11 +1,17 @@
-import { app } from 'mu';
-import services from './config/rules';
+import { app, errorHandler, beforeExit, uuid } from 'mu';
 import normalizeQuad from './config/normalize-quad';
 import bodyParser from 'body-parser';
 import dns from 'dns';
 import { foldChangeSets } from './folding';
 import { sendRequest } from './send-request';
 import { sendBundledRequest } from './bundle-requests';
+import { metricsHandler, recordDeltaReceived } from './metrics.js';
+
+import services from './config/rules.js';
+
+beforeExit( async () => {
+  console.log('Shutting down delta-notifier gracefully...');
+});
 
 // Log server config if requested
 if( process.env["LOG_SERVER_CONFIGURATION"] )
@@ -16,6 +22,8 @@ app.get( '/', function( req, res ) {
   res.send("Hello, delta notification is running");
 } );
 
+app.get( '/metrics', metricsHandler );
+
 app.post( '/', bodyParser.json({limit: '500mb'}), function( req, res ) {
   if( process.env["LOG_REQUESTS"] ) {
     console.log("Logging request body");
@@ -24,6 +32,8 @@ app.post( '/', bodyParser.json({limit: '500mb'}), function( req, res ) {
 
   const changeSets = req.body.changeSets;
 
+  recordDeltaReceived(changeSets.length);
+
   const originalMuCallIdTrail = JSON.parse( req.get('mu-call-id-trail') || "[]" );
   const originalMuCallId = req.get('mu-call-id');
   const muCallIdTrail = JSON.stringify( [...originalMuCallIdTrail, originalMuCallId] );
@@ -139,3 +149,5 @@ async function getServiceIp(entry) {
     } );
   } );
 };
+
+app.use(errorHandler);

bundle-requests.js

@@ -1,5 +1,6 @@
 import { foldChangeSets } from './folding';
 import { sendRequest } from "./send-request.js";
+import { incPendingBundles, decPendingBundles } from "./metrics.js";
 
 // map from bundle key to bundle object
 const bundles = {};
@@ -18,6 +19,7 @@ const getBundleKey = (entry, muSessionId, changeSets) => {
 const executeBundledRequest = (bundleKey) => {
   const bundle = bundles[bundleKey];
   delete bundles[bundleKey];
+  decPendingBundles();
 
   if (!bundle) {
     console.error(
@@ -76,6 +78,7 @@ export const sendBundledRequest = (
       muSessionId,
       bundledCallIdTrails: [],
     };
+    incPendingBundles();
     setTimeout(
       () => executeBundledRequest(bundleKey),
       entry.options.gracePeriod

metrics.js

@@ -0,0 +1,95 @@
+import { createRequire } from 'module';
+import client from 'prom-client';
+import services from './config/rules.js';
+
+const require = createRequire(import.meta.url);
+const { version } = require('./package.json');
+
+const register = new client.Registry();
+
+// --- Info metrics ---
+
+const info = new client.Gauge({
+  name: 'deltanotifier_info',
+  help: 'Delta-notifier service info',
+  labelNames: ['version'],
+  registers: [register]
+});
+info.set({ version }, 1);
+
+const configuredRulesTotal = new client.Gauge({
+  name: 'deltanotifier_configured_rules_total',
+  help: 'Number of configured rules',
+  registers: [register]
+});
+configuredRulesTotal.set(services.length);
+
+// --- Deltas received ---
+
+const deltasReceivedTotal = new client.Counter({
+  name: 'deltanotifier_deltas_received_total',
+  help: 'Total POST / requests received',
+  registers: [register]
+});
+
+const changesetsReceivedTotal = new client.Counter({
+  name: 'deltanotifier_changesets_received_total',
+  help: 'Total changesets across all deltas',
+  registers: [register]
+});
+
+// --- Notifications ---
+
+const notificationsTotal = new client.Counter({
+  name: 'deltanotifier_notifications_total',
+  help: 'Notifications sent per target service',
+  labelNames: ['target', 'status'],
+  registers: [register]
+});
+
+const notificationDurationSeconds = new client.Histogram({
+  name: 'deltanotifier_notification_duration_seconds',
+  help: 'Notification send latency per target service',
+  labelNames: ['target'],
+  buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10],
+  registers: [register]
+});
+
+// --- Pending bundles ---
+
+const pendingBundles = new client.Gauge({
+  name: 'deltanotifier_pending_bundles',
+  help: 'Number of grace-period bundles waiting to be sent',
+  registers: [register]
+});
+
+// --- Recording functions ---
+
+export function recordDeltaReceived(changesetCount) {
+  deltasReceivedTotal.inc();
+  changesetsReceivedTotal.inc(changesetCount);
+}
+
+export function recordNotification(target, status, durationSeconds) {
+  notificationsTotal.inc({ target, status });
+  notificationDurationSeconds.observe({ target }, durationSeconds);
+}
+
+export function incPendingBundles() {
+  pendingBundles.inc();
+}
+
+export function decPendingBundles() {
+  pendingBundles.dec();
+}
+
+// --- Handler ---
+
+export async function metricsHandler(req, res) {
+  try {
+    res.set('Content-Type', register.contentType);
+    res.end(await register.metrics());
+  } catch (err) {
+    res.status(500).end(err.message);
+  }
+}

package.json

@@ -3,6 +3,7 @@
   "version": "0.4.0",
   "description": "Sends notifications about changes to interested mu.semte.ch microservices",
   "main": "app.js",
+  "type": "module",
   "scripts": {
     "test": "echo \"Error: no test specified\" && exit 1"
   },
@@ -20,5 +21,8 @@
   "bugs": {
     "url": "https://github.com/mu-semtech/delta-notifier/issues"
   },
-  "homepage": "https://github.com/mu-semtech/delta-notifier#readme"
+  "homepage": "https://github.com/mu-semtech/delta-notifier#readme",
+  "dependencies": {
+    "prom-client": "^15.0.0"
+  }
 }

send-request.js

@@ -1,5 +1,6 @@
 import http from "http";
 import { uuid } from "mu";
+import { recordNotification } from "./metrics.js";
 
 const DEFAULT_RETRY_TIMEOUT = 250;
 
@@ -117,6 +118,8 @@ export async function sendRequest(
     }
     if (process.env["DEBUG_DELTA_SEND"])
       console.log(`Executing send ${method} to ${url}`);
+    const target = (new URL(url)).hostname;
+    const startTime = Date.now();
     try {
       const keepAliveAgent = new http.Agent({
         keepAlive: true,
@@ -127,6 +130,12 @@ export async function sendRequest(
         body,
         agent: keepAliveAgent,
       });
+      const duration = (Date.now() - startTime) / 1000;
+      if (response.ok) {
+        recordNotification(target, 'success', duration);
+      } else {
+        recordNotification(target, 'error', duration);
+      }
       await handleResponse(
         response,
         entry,
@@ -137,6 +146,7 @@ export async function sendRequest(
         retriesLeft
       );
     } catch (error) {
+      recordNotification(target, 'error', (Date.now() - startTime) / 1000);
       console.log(error);
     }
   } else {