matching: prepare for REF-AUS: extend metrics and logging with labels 💥

Author: derhuerst

lib/match.js

@@ -141,13 +141,17 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		name: 'nats_msg_seq',
 		help: 'sequence number of the latest NATS message being processed',
 		registers: [register],
+		labelNames: [
+			'topic_root', // first "segment" of the topic, e.g. `AUS` with `aus.istfahrt.foo.bar`
+		],
 	})
 
 	const successesTotal = new Counter({
 		name: 'matching_successes_total',
 		help: 'number of successfully matched movements/trips',
 		registers: [register],
 		labelNames: [
+			'kind', // sollfahrt, istfahrt
 			'cached',
 		],
 	})
@@ -156,20 +160,24 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		help: 'number of matching failures',
 		registers: [register],
 		labelNames: [
+			'kind', // sollfahrt, istfahrt
 			'cached',
 		],
 	})
 	const errorsTotal = new Counter({
 		name: 'matching_errors_total',
 		help: 'number of errors that have occured while matching',
 		registers: [register],
-		labelNames: [],
+		labelNames: [
+			'kind', // sollfahrt, istfahrt
+		],
 	})
 	const matchingTimeSeconds = new Summary({
 		name: 'matching_time_seconds',
 		help: 'seconds trips need to be matched',
 		registers: [register],
 		labelNames: [
+			'kind', // sollfahrt, istfahrt
 			'matched',
 			'cached',
 		],
@@ -250,7 +258,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 				topic_root,
 				redelivered,
 			}, tReceived / 1000)
-			natsMsgSeq.set(seq)
+			natsMsgSeq.set(seq, {topic_root})
 		}
 
 		let vdvItem = null
@@ -275,14 +283,17 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 
 			if (isMatched) {
 				successesTotal.inc({
+					kind,
 					cached: isCached ? '1' : '0',
 				})
 			} else {
 				failuresTotal.inc({
+					kind,
 					cached: isCached ? '1' : '0',
 				})
 			}
 			matchingTimeSeconds.observe({
+				kind,
 				matched: isMatched ? '1' : '0',
 				cached: isCached ? '1' : '0',
 			}, matchingTime / 1000)
@@ -305,7 +316,9 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 					[kind]: vdvItem,
 					natsMsgSeq: msg.seq,
 				}, `failed to match trip: ${err.message || (err + '')}`)
-				errorsTotal.inc()
+				errorsTotal.inc({
+					kind,
+				})
 			}
 
 			// We catch all non-programmer errors in order not to abort the message processing (see below).
@@ -371,8 +384,9 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 			// query details of the (externally created) NATS JetStream consumer
 			const consumerInfo = await istFahrtsConsumer.info()
 			serviceLogger.debug({
+				stream: NATS_JETSTREAM_AUS_ISTFAHRT_STREAM_NAME,
 				consumerInfo,
-			}, 'using NATS JetStream consumer')
+			}, 'using NATS JetStream consumer for AUS IstFahrts')
 		}
 
 		const istFahrtsSub = await istFahrtsConsumer.consume()