feat: add /metrics endpoint to gateways SOFIE-456 (Sofie-Automation#1723)

Author: Julusian

packages/live-status-gateway/src/config.ts

@@ -11,6 +11,7 @@ let deviceToken: string = process.env.DEVICE_TOKEN || ''
 let disableWatchdog: boolean = process.env.DISABLE_WATCHDOG === '1' || false
 let unsafeSSL: boolean = process.env.UNSAFE_SSL === '1' || false
 const certs: string[] = (process.env.CERTIFICATES || '').split(';') || []
+let healthPort: number | undefined = parseInt(process.env.HEALTH_PORT + '') || undefined
 
 let prevProcessArg = ''
 process.argv.forEach((val) => {
@@ -37,12 +38,14 @@ process.argv.forEach((val) => {
 	} else if (val.match(/-unsafeSSL/i)) {
 		// Will cause the Node applocation to blindly accept all certificates. Not recommenced unless in local, controlled networks.
 		unsafeSSL = true
+	} else if (prevProcessArg.match(/-healthPort/i)) {
+		healthPort = parseInt(val)
 	}
 	prevProcessArg = nextPrevProcessArg + ''
 })
 
 const config: Config = {
-	process: {
+	certificates: {
 		unsafeSSL: unsafeSSL,
 		certificates: certs.filter((c) => c !== undefined && c !== null && c.length !== 0),
 	},
@@ -55,6 +58,9 @@ const config: Config = {
 		port: port,
 		watchdog: !disableWatchdog,
 	},
+	health: {
+		port: healthPort,
+	},
 }
 
 export { config, logPath, logLevel, disableWatchdog }

packages/live-status-gateway/src/connector.ts

@@ -1,25 +1,31 @@
 import { CoreHandler, CoreConfig } from './coreHandler.js'
 import { Logger } from 'winston'
-import { loadDDPTLSOptions } from '@sofie-automation/server-core-integration'
 import { PeripheralDeviceId } from '@sofie-automation/shared-lib/dist/core/model/Ids'
 import { LiveStatusServer } from './liveStatusServer.js'
+import {
+	CertificatesConfig,
+	HealthConfig,
+	HealthEndpoints,
+	IConnector,
+	loadDDPTLSOptions,
+	stringifyError,
+} from '@sofie-automation/server-core-integration'
 
 export interface Config {
-	process: ProcessConfig
+	certificates: CertificatesConfig
 	device: DeviceConfig
 	core: CoreConfig
+	health: HealthConfig
 }
-export interface ProcessConfig {
-	/** Will cause the Node applocation to blindly accept all certificates. Not recommenced unless in local, controlled networks. */
-	unsafeSSL: boolean
-	/** Paths to certificates to load, for SSL-connections */
-	certificates: string[]
-}
+
 export interface DeviceConfig {
 	deviceId: PeripheralDeviceId
 	deviceToken: string
 }
-export class Connector {
+export class Connector implements IConnector {
+	public initialized = false
+	public initializedError: string | undefined = undefined
+
 	private coreHandler: CoreHandler | undefined
 	private _logger: Logger
 	private _liveStatusServer: LiveStatusServer | undefined
@@ -31,11 +37,13 @@ export class Connector {
 	public async init(config: Config): Promise<void> {
 		try {
 			this._logger.info('Initializing Process...')
-			const tlsOptions = loadDDPTLSOptions(this._logger, config.process)
+			const tlsOptions = loadDDPTLSOptions(this._logger, config.certificates)
 			this._logger.info('Process initialized')
 
 			this._logger.info('Initializing Core...')
 			this.coreHandler = new CoreHandler(this._logger, config.device)
+			new HealthEndpoints(this, this.coreHandler, config.health)
+
 			await this.coreHandler.init(config.core, tlsOptions)
 			this._logger.info('Core initialized')
 
@@ -51,6 +59,8 @@ export class Connector {
 			this._logger.error(e)
 			this._logger.error(e.stack)
 
+			this.initializedError = stringifyError(e)
+
 			try {
 				if (this.coreHandler) {
 					this.coreHandler.destroy().catch(this._logger.error)

packages/live-status-gateway/src/coreHandler.ts

@@ -4,6 +4,7 @@ import {
 	CoreOptions,
 	DDPConnectorOptions,
 	DDPTLSOptions,
+	ICoreHandler,
 	Observer,
 	PeripheralDevicePubSub,
 	PeripheralDevicePubSubCollections,
@@ -36,7 +37,7 @@ export interface CoreConfig {
 /**
  * Represents a connection between the Gateway and Core
  */
-export class CoreHandler {
+export class CoreHandler implements ICoreHandler {
 	core!: CoreConnection<
 		CorelibPubSubTypes & PeripheralDevicePubSubTypes,
 		CorelibPubSubCollections & PeripheralDevicePubSubCollections
@@ -45,12 +46,7 @@ export class CoreHandler {
 	public _observers: Array<any> = []
 	public deviceSettings: LiveStatusGatewayConfig = {}
 
-	public errorReporting = false
-	public multithreading = false
-	public reportAllCommands = false
-
 	private _deviceOptions: DeviceConfig
-	private _onConnected?: () => any
 	private _executedFunctions = new Set<PeripheralDeviceCommandId>()
 	private _coreConfig?: CoreConfig
 
@@ -59,6 +55,10 @@ export class CoreHandler {
 	private _statusInitialized = false
 	private _statusDestroyed = false
 
+	public get connectedToCore(): boolean {
+		return this.core && this.core.connected
+	}
+
 	constructor(logger: Logger, deviceOptions: DeviceConfig) {
 		this.logger = logger
 		this._deviceOptions = deviceOptions
@@ -77,7 +77,6 @@ export class CoreHandler {
 			this.setupObserversAndSubscriptions().catch((e) => {
 				this.logger.error('Core Error during setupObserversAndSubscriptions:', e)
 			})
-			if (this._onConnected) this._onConnected()
 		})
 		this.core.onDisconnected(() => {
 			this.logger.warn('Core Disconnected!')
@@ -97,7 +96,6 @@ export class CoreHandler {
 
 		this.logger.info('Core id: ' + this.core.deviceId)
 		await this.setupObserversAndSubscriptions()
-		if (this._onConnected) this._onConnected()
 
 		this._statusInitialized = true
 		await this.updateCoreStatus()
@@ -180,9 +178,6 @@ export class CoreHandler {
 
 		return options
 	}
-	onConnected(fcn: () => any): void {
-		this._onConnected = fcn
-	}
 
 	onDeviceChanged(): void {
 		const col = this.core.getCollection(PeripheralDevicePubSubCollectionsNames.peripheralDeviceForDevice)
@@ -319,7 +314,10 @@ export class CoreHandler {
 		this.logger.info('getDevicesInfo')
 		return []
 	}
-	async updateCoreStatus(): Promise<any> {
+	getCoreStatus(): {
+		statusCode: StatusCode
+		messages: string[]
+	} {
 		let statusCode = StatusCode.GOOD
 		const messages: Array<string> = []
 
@@ -331,11 +329,13 @@ export class CoreHandler {
 			statusCode = StatusCode.BAD
 			messages.push('Shut down')
 		}
-
-		return this.core.setStatus({
-			statusCode: statusCode,
-			messages: messages,
-		})
+		return {
+			statusCode,
+			messages,
+		}
+	}
+	async updateCoreStatus(): Promise<any> {
+		return this.core.setStatus(this.getCoreStatus())
 	}
 	private _getVersions() {
 		const versions: { [packageName: string]: string } = {}

packages/live-status-gateway/src/liveStatusServer.ts

@@ -34,6 +34,7 @@ import { NotificationsHandler } from './collections/notifications/notificationsH
 import { NotificationsTopic } from './topics/notificationsTopic.js'
 import { PlaylistNotificationsHandler } from './collections/notifications/playlistNotificationsHandler.js'
 import { RundownNotificationsHandler } from './collections/notifications/rundownNotificationsHandler.js'
+import { wsConnectionsGauge } from './wsMetrics.js'
 
 export interface CollectionHandlers {
 	studioHandler: StudioHandler
@@ -153,8 +154,10 @@ export class LiveStatusServer {
 				this._logger.info(`Closing websocket`)
 				rootChannel.removeSubscriber(ws)
 				this._clients.delete(ws)
+				wsConnectionsGauge.set(this._clients.size)
 			})
 			this._clients.add(ws)
+			wsConnectionsGauge.set(this._clients.size)
 
 			if (typeof request.url === 'string' && request.url === '/') {
 				rootChannel.addSubscriber(ws)

packages/live-status-gateway/src/topics/root.ts

@@ -10,6 +10,7 @@ import {
 	SubscriptionStatus,
 	SubscriptionName,
 } from '@sofie-automation/live-status-gateway-api'
+import { activeSubscriptionsGauge, subscriptionSubscribersGauge } from '../wsMetrics.js'
 
 enum PublishMsg {
 	ping = 'ping',
@@ -41,6 +42,7 @@ export class RootChannel extends WebSocketTopicBase implements WebSocketTopic {
 	removeSubscriber(ws: WebSocket): void {
 		super.removeSubscriber(ws)
 		this._topics.forEach((h) => h.removeSubscriber(ws))
+		this._updateSubscriptionMetrics()
 	}
 
 	processMessage(ws: WebSocket, msg: object): void {
@@ -74,6 +76,16 @@ export class RootChannel extends WebSocketTopicBase implements WebSocketTopic {
 		if (Object.values<string>(SubscriptionName).includes(channel)) this._topics.set(channel, topic)
 	}
 
+	private _updateSubscriptionMetrics(): void {
+		let total = 0
+		for (const [name, topic] of this._topics) {
+			const count = topic.subscriberCount
+			subscriptionSubscribersGauge.set({ subscription: name }, count)
+			total += count
+		}
+		activeSubscriptionsGauge.set(total)
+	}
+
 	subscribe(ws: WebSocket, name: SubscriptionName, reqid: number): void {
 		const topic = this._topics.get(name)
 		const curUnsubscribed =
@@ -91,6 +103,7 @@ export class RootChannel extends WebSocketTopicBase implements WebSocketTopic {
 				})
 			)
 			topic.addSubscriber(ws)
+			this._updateSubscriptionMetrics()
 		} else {
 			this.sendMessage(
 				ws,
@@ -112,6 +125,7 @@ export class RootChannel extends WebSocketTopicBase implements WebSocketTopic {
 		const curSubscribed = topic && topic.hasSubscriber(ws) && Object.values<string>(SubscriptionName).includes(name)
 		if (curSubscribed) {
 			topic.removeSubscriber(ws)
+			this._updateSubscriptionMetrics()
 			this.sendMessage(
 				ws,
 				literal<SubscriptionStatusSuccess>({

packages/live-status-gateway/src/wsHandler.ts

@@ -23,6 +23,10 @@ export abstract class WebSocketTopicBase {
 				: this.sendStatusToAll
 	}
 
+	get subscriberCount(): number {
+		return this._subscribers.size
+	}
+
 	addSubscriber(ws: WebSocket): void {
 		this._logger.info(`${this._name} adding a websocket subscriber`)
 		this._subscribers.add(ws)
@@ -77,6 +81,7 @@ export abstract class WebSocketTopicBase {
 }
 
 export interface WebSocketTopic {
+	subscriberCount: number
 	addSubscriber(ws: WebSocket): void
 	hasSubscriber(ws: WebSocket): boolean
 	removeSubscriber(ws: WebSocket): void

packages/live-status-gateway/src/wsMetrics.ts

@@ -0,0 +1,17 @@
+import { MetricsGauge } from '@sofie-automation/server-core-integration/dist/lib/prometheus'
+
+export const wsConnectionsGauge = new MetricsGauge({
+	name: 'sofie_lsg_websocket_connections',
+	help: 'Number of open WebSocket connections',
+})
+
+export const activeSubscriptionsGauge = new MetricsGauge({
+	name: 'sofie_lsg_active_subscriptions_total',
+	help: 'Total number of active subscriptions across all topics',
+})
+
+export const subscriptionSubscribersGauge = new MetricsGauge({
+	name: 'sofie_lsg_subscription_subscribers',
+	help: 'Number of subscribers per subscription',
+	labelNames: ['subscription'] as const,
+})

packages/mos-gateway/src/CoreMosDeviceHandler.ts

@@ -35,6 +35,12 @@ import { PartialDeep } from 'type-fest'
 import type { CoreHandler } from './coreHandler.js'
 import { CoreConnectionChild } from '@sofie-automation/server-core-integration/dist/lib/CoreConnectionChild'
 import { Queue } from '@sofie-automation/server-core-integration/dist/lib/queue'
+import {
+	mosDeviceConnectedGauge,
+	mosMessagesFailedCounter,
+	mosMessagesReceivedCounter,
+	mosQueueDepthGauge,
+} from './mosMetrics.js'
 
 function deepMatch(object: any, attrs: any, deep: boolean): boolean {
 	const keys = Object.keys(attrs)
@@ -197,6 +203,18 @@ export class CoreMosDeviceHandler {
 				messages: messages,
 			})
 			.catch((e) => this._coreParentHandler.logger.warn('Error when setting status:' + e))
+
+		const deviceId = this._mosDevice.idPrimary
+		mosDeviceConnectedGauge.set(
+			{ device_id: deviceId, connection: 'primary' },
+			connectionStatus.PrimaryConnected ? 1 : 0
+		)
+		if (this._mosDevice.idSecondary) {
+			mosDeviceConnectedGauge.set(
+				{ device_id: deviceId, connection: 'secondary' },
+				connectionStatus.SecondaryConnected ? 1 : 0
+			)
+		}
 	}
 	async getMachineInfo(): Promise<IMOSListMachInfo> {
 		const info: IMOSListMachInfo = {
@@ -456,8 +474,15 @@ export class CoreMosDeviceHandler {
 			return this.fixMosData(attr)
 		}) as any
 
+		const deviceId = this._mosDevice.idPrimary
+		const commandName = methodName as string
+		mosMessagesReceivedCounter.inc({ device_id: deviceId, command: commandName })
+		mosQueueDepthGauge.inc({ device_id: deviceId })
+
 		// Make the commands be sent sequantially:
 		return this._messageQueue.putOnQueue(async () => {
+			mosQueueDepthGauge.dec({ device_id: deviceId })
+
 			// Log info about the sent command:
 			let msg = 'Command: ' + methodName
 			const attr0 = attrs[0] as any | undefined
@@ -476,6 +501,7 @@ export class CoreMosDeviceHandler {
 			const res = (this.core.coreMethods[methodName] as any)(...attrs)
 			return res.catch((e: any) => {
 				this._coreParentHandler.logger.info('MOS command rejected: ' + ((e && JSON.stringify(e)) || e))
+				mosMessagesFailedCounter.inc({ device_id: deviceId, command: commandName })
 				throw e
 			})
 		})

packages/mos-gateway/src/coreHandler.ts

@@ -35,7 +35,6 @@ export class CoreHandler implements ICoreHandler {
 	core: CoreConnection | undefined
 	logger: Winston.Logger
 	public _observers: Array<Observer<any>> = []
-	public connectedToCore = false
 	private _deviceOptions: DeviceConfig
 	private _coreMosHandlers: Array<CoreMosDeviceHandler> = []
 	private _onConnected?: () => any
@@ -44,6 +43,10 @@ export class CoreHandler implements ICoreHandler {
 	private _executedFunctions = new Set<PeripheralDeviceCommandId>()
 	private _coreConfig?: CoreConfig
 
+	public get connectedToCore(): boolean {
+		return !!this.core && this.core.connected
+	}
+
 	public static async create(
 		logger: Winston.Logger,
 		config: CoreConfig,
@@ -67,12 +70,10 @@ export class CoreHandler implements ICoreHandler {
 
 		this.core.onConnected(() => {
 			this.logger.info('Core Connected!')
-			this.connectedToCore = true
 			if (this._isInitialized) this.onConnectionRestored()
 		})
 		this.core.onDisconnected(() => {
 			this.logger.info('Core Disconnected!')
-			this.connectedToCore = false
 		})
 		this.core.onError((err) => {
 			this.logger.error('Core Error: ' + (typeof err === 'string' ? err : err.message || err.toString()))