matching: add more NATS metrics

- nats_nr_of_msgs_{received,sent}_total
- nats_latest_msg_{received,sent}_timestamp_seconds

Author: derhuerst

lib/match.js

@@ -91,11 +91,56 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 	}
 	ok(Number.isInteger(natsAckWait), 'opt.natsAckWait must be an integer')
 
+	// NATS-related metrics
+	// Note: We mirror OpenDataVBB/gtfs-rt-feed's metrics here.
+	const natsNrOfMessagesReceivedTotal = new Counter({
+		name: 'nats_nr_of_msgs_received_total',
+		help: 'number of messages received from NATS',
+		registers: [register],
+		labelNames: [
+			'stream', // name of the JetStream stream
+			'consumer', // name of the JetStream consumer
+			'topic_root', // first "segment" of the topic, e.g. `AUS` with `aus.istfahrt.foo.bar`
+			'redelivered', // 1/0
+		],
+	})
+	const natsLatestMessageReceivedTimestampSeconds = new Gauge({
+		name: 'nats_latest_msg_received_timestamp_seconds',
+		help: 'when the latest message has been received from NATS',
+		registers: [register],
+		labelNames: [
+			'stream', // name of the JetStream stream
+			'consumer', // name of the JetStream consumer
+			'topic_root', // first "segment" of the topic, e.g. `AUS` with `aus.istfahrt.foo.bar`
+			'redelivered', // 1/0
+		],
+	})
+	// todo: track redeliveries as `Summary` using `msg.info.redeliveryCount`
+	const natsNrOfMessagesSentTotal = new Counter({
+		name: 'nats_nr_of_msgs_sent_total',
+		help: 'number of messages sent to NATS',
+		registers: [register],
+		labelNames: [
+			'topic_root', // first "segment" of the topic, e.g. `AUS` with `aus.istfahrt.foo.bar`
+		],
+	})
+	const natsLatestMessageSentTimestampSeconds = new Gauge({
+		name: 'nats_latest_msg_sent_timestamp_seconds',
+		help: 'when the latest message has been sent to NATS',
+		registers: [register],
+		labelNames: [
+			'topic_root', // first "segment" of the topic, e.g. `AUS` with `aus.istfahrt.foo.bar`
+		],
+	})
+	// NATS gives separate sequence numbers to both a) messages in a stream and b) messages as (re-)received by a consumer.
+	// We currently use `msg.seq`. – todo: what is that?
+	// todo [breaking]: change ot use the stream sequence or remove this metric, as it's misleading!
 	const natsMsgSeq = new Gauge({
 		name: 'nats_msg_seq',
 		help: 'sequence number of the latest NATS message being processed',
 		registers: [register],
 	})
+
 	const successesTotal = new Counter({
 		name: 'matching_successes_total',
 		help: 'number of successfully matched movements/trips',
@@ -143,6 +188,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 			} = await matchVdvAusIstFahrtWithGtfs(vdvAusIstFahrt)
 
 			const topic = getNatsTopicFromGtfsRtTripUpdate(gtfsRtTripUpdate)
+			const tPublished = Date.now()
 
 			logger.trace({
 				topic,
@@ -156,6 +202,18 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 			}, 'publishing GTFS-RT TripUpdate')
 			natsClient.publish(topic, natsJson.encode(gtfsRtTripUpdate))
 
+			// update NATS metrics
+			{
+				// We slice() to keep the cardinality low in case of a bug.
+				const topic_root = (topic.split('.')[0] || '').slice(0, 7)
+				natsNrOfMessagesSentTotal.inc({
+					topic_root,
+				})
+				natsLatestMessageSentTimestampSeconds.set({
+					topic_root,
+				}, tPublished / 1000)
+			}
+
 			if (isMatched) {
 				successesTotal.inc({
 					cached: isCached ? '1' : '0',
@@ -181,6 +239,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 	}
 
 	const processAusIstFahrtMsg = async (msg) => {
+		const tReceived = Date.now()
 		const {
 			subject,
 			seq,
@@ -193,7 +252,30 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 			redelivered,
 			dataSlice: data.slice(0, 100).toString('utf8'),
 		}, 'processing AUS IstFahrt msg')
-		natsMsgSeq.set(seq) // todo: is `seq` an integer?
+
+		// update NATS metrics
+		{
+			const {
+				stream,
+				consumer,
+			} = msg.info
+			// We slice() to keep the cardinality low in case of a bug.
+			const topic_root = (subject.split('.')[0] || '').slice(0, 7)
+			const redelivered = msg.info.redelivered ? '1' : '0'
+			natsNrOfMessagesReceivedTotal.inc({
+				stream, // name of the JetStream stream
+				consumer, // name of the JetStream consumer
+				topic_root,
+				redelivered,
+			})
+			natsLatestMessageReceivedTimestampSeconds.set({
+				stream, // name of the JetStream stream
+				consumer, // name of the JetStream consumer
+				topic_root,
+				redelivered,
+			}, tReceived / 1000)
+			natsMsgSeq.set(seq)
+		}
 
 		let ausIstFahrt = null
 		try {