From 6c8506b27156daceaae9c4b887d35bbce3a32d01 Mon Sep 17 00:00:00 2001 From: Niels Vandekeybus Date: Tue, 10 Oct 2023 10:29:20 +0200 Subject: [PATCH 1/4] use mu errorHandler --- app.js | 2 ++ 1 file changed, 2 insertions(+) diff --git a/app.js b/app.js index 2ba5c84..2cdc861 100644 --- a/app.js +++ b/app.js @@ -139,3 +139,5 @@ async function getServiceIp(entry) { } ); } ); }; + +app.use(errorHandler); From 67d02cdcbbfc824ab4ee0e190667e3ec388b7c76 Mon Sep 17 00:00:00 2001 From: Niels Vandekeybus Date: Tue, 10 Feb 2026 16:56:47 +0100 Subject: [PATCH 2/4] upgrade mu-javascript-template to 1.9.1 Brings Node 20, arm64 support, ESM module support, and beforeExit for graceful shutdown. --- Dockerfile | 2 +- app.js | 9 +++++++-- package.json | 1 + 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/Dockerfile b/Dockerfile index b3ebbbc..46b9105 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1 +1 @@ -FROM semtech/mu-javascript-template:1.8.0 +FROM semtech/mu-javascript-template:1.9.1 diff --git a/app.js b/app.js index 2cdc861..95f3955 100644 --- a/app.js +++ b/app.js @@ -1,5 +1,4 @@ -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'; @@ -7,6 +6,12 @@ import { foldChangeSets } from './folding'; import { sendRequest } from './send-request'; import { sendBundledRequest } from './bundle-requests'; +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"] ) console.log(JSON.stringify( services )); diff --git a/package.json b/package.json index fc4843e..af7ca09 100644 --- a/package.json +++ b/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" }, From 9326772442007812e674ff98175d647910957080 Mon Sep 17 00:00:00 2001 From: Niels Vandekeybus Date: Wed, 11 Feb 2026 14:49:16 +0100 Subject: [PATCH 3/4] initial metrics endpoint --- app.js | 5 +++ bundle-requests.js | 3 ++ metrics.js | 93 ++++++++++++++++++++++++++++++++++++++++++++++ package.json | 5 ++- send-request.js | 10 +++++ 5 files changed, 115 insertions(+), 1 deletion(-) create mode 100644 metrics.js diff --git a/app.js b/app.js index 95f3955..0b12622 100644 --- a/app.js +++ b/app.js @@ -5,6 +5,7 @@ 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'; @@ -21,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"); @@ -29,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] ); diff --git a/bundle-requests.js b/bundle-requests.js index 38b1226..0489842 100644 --- a/bundle-requests.js +++ b/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 diff --git a/metrics.js b/metrics.js new file mode 100644 index 0000000..b082c71 --- /dev/null +++ b/metrics.js @@ -0,0 +1,93 @@ +import client from 'prom-client'; +import services from './config/rules.js'; + +const register = new client.Registry(); + +// --- Info metrics --- + +const info = new client.Gauge({ + name: 'deltanotifier_info', + help: 'Delta-notifier service info', + labelNames: ['version'], + registers: [register] +}); + +const version = process.env.DELTA_NOTIFIER_VERSION || 'unknown'; +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); + } +} diff --git a/package.json b/package.json index af7ca09..7c20902 100644 --- a/package.json +++ b/package.json @@ -21,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" + } } diff --git a/send-request.js b/send-request.js index e19a990..61933b3 100644 --- a/send-request.js +++ b/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 { From 1754a93eb6df9e5ffb012b4a24477a8a06145188 Mon Sep 17 00:00:00 2001 From: Niels Vandekeybus Date: Thu, 12 Feb 2026 15:53:29 +0100 Subject: [PATCH 4/4] properly include version --- metrics.js | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/metrics.js b/metrics.js index b082c71..cdeda48 100644 --- a/metrics.js +++ b/metrics.js @@ -1,6 +1,10 @@ +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 --- @@ -11,8 +15,6 @@ const info = new client.Gauge({ labelNames: ['version'], registers: [register] }); - -const version = process.env.DELTA_NOTIFIER_VERSION || 'unknown'; info.set({ version }, 1); const configuredRulesTotal = new client.Gauge({