Merge branch 'nrk/feat/pgw-health-endpoints' into nrk/release52

# Conflicts:
#	packages/playout-gateway/package.json
#	packages/yarn.lock

Author: nytamin

packages/playout-gateway/package.json

@@ -55,11 +55,17 @@
 		"rundown",
 		"production"
 	],
+	"devDependencies": {
+		"@types/koa": "^3.0.0",
+		"@types/koa__router": "^12.0.4"
+	},
 	"dependencies": {
+		"@koa/router": "^14.0.0",
 		"@sofie-automation/server-core-integration": "1.52.9-nrk",
 		"@sofie-automation/shared-lib": "1.52.9-nrk",
 		"debug": "^4.4.0",
 		"influx": "^5.9.7",
+		"koa": "^3.0.1",
 		"timeline-state-resolver": "9.3.2-nightly-test-release52-nrk-20250910-071718-b75f9a20c.0",
 		"tslib": "^2.8.1",
 		"underscore": "^1.13.7",

packages/playout-gateway/src/config.ts

@@ -20,6 +20,8 @@ let influxUser: string | undefined = process.env.INFLUX_USER || 'sofie'
 let influxPassword: string | undefined = process.env.INFLUX_PASSWORD || undefined
 let influxDatabase: string | undefined = process.env.INFLUX_DB || 'sofie'
 
+let healthPort: number | undefined = parseInt(process.env.HEALTH_PORT + '') || undefined
+
 let prevProcessArg = ''
 process.argv.forEach((val) => {
 	val = val + ''
@@ -50,14 +52,18 @@ process.argv.forEach((val) => {
 		influxPassword = val
 	} else if (prevProcessArg.match(/-influxDatabase/i)) {
 		influxDatabase = val
+	} else if (prevProcessArg.match(/-healthPort/i)) {
+		healthPort = parseInt(val)
 
 		// arguments with no options:
 	} else if (val.match(/-disableWatchdog/i)) {
 		disableWatchdog = true
 	} else if (val.match(/-disableAtemUpload/i)) {
 		disableAtemUpload = true
 	} else if (val.match(/-unsafeSSL/i)) {
-		// Will cause the Node applocation to blindly accept all certificates. Not recommenced unless in local, controlled networks.
+		// Will cause the Node application to blindly accept all certificates.
+		// Not recommenced unless in local, controlled networks.
+		// Instead use "-certificates cert1 cert2"
 		unsafeSSL = true
 	}
 	prevProcessArg = nextPrevProcessArg + ''
@@ -85,6 +91,9 @@ const config: Config = {
 		password: influxPassword,
 		database: influxDatabase,
 	},
+	health: {
+		port: healthPort,
+	},
 }
 
 export { config, logPath, logLevel, disableWatchdog, disableAtemUpload }

packages/playout-gateway/src/connector.ts

@@ -6,21 +6,27 @@ import {
 	CertificatesConfig,
 	PeripheralDeviceId,
 	loadCertificatesFromDisk,
+	stringifyError,
 } from '@sofie-automation/server-core-integration'
+import { HealthConfig, HealthEndpoints } from './health'
 
 export interface Config {
 	certificates: CertificatesConfig
 	device: DeviceConfig
 	core: CoreConfig
 	tsr: TSRConfig
 	influx: InfluxConfig
+	health: HealthConfig
 }
 
 export interface DeviceConfig {
 	deviceId: PeripheralDeviceId
 	deviceToken: string
 }
 export class Connector {
+	public initialized = false
+	public initializedError: string | undefined = undefined
+
 	private tsrHandler: TSRHandler | undefined
 	private coreHandler: CoreHandler | undefined
 	private _logger: Logger
@@ -38,6 +44,8 @@ export class Connector {
 
 			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, this._certificates)
 			this._logger.info('Core initialized')
 
@@ -47,12 +55,15 @@ export class Connector {
 			this._logger.info('TSR initialized')
 
 			this._logger.info('Initialization done')
+			this.initialized = true
 			return
 		} catch (e: any) {
 			this._logger.error('Error during initialization:')
 			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/playout-gateway/src/coreHandler.ts

@@ -59,6 +59,8 @@ export class CoreHandler {
 	private _statusInitialized = false
 	private _statusDestroyed = false
 
+	public connectedToCore = false
+
 	constructor(logger: Logger, deviceOptions: DeviceConfig) {
 		this.logger = logger
 		this._deviceOptions = deviceOptions
@@ -73,11 +75,13 @@ export class CoreHandler {
 
 		this.core.onConnected(() => {
 			this.logger.info('Core Connected!')
+			this.connectedToCore = true
 
 			if (this._onConnected) this._onConnected()
 		})
 		this.core.onDisconnected(() => {
 			this.logger.warn('Core Disconnected!')
+			this.connectedToCore = false
 		})
 		this.core.onError((err: any) => {
 			this.logger.error('Core Error: ' + (typeof err === 'string' ? err : err.message || err.toString() || err))
@@ -389,9 +393,12 @@ export class CoreHandler {
 
 		return Object.fromEntries(this._tsrHandler.getDebugStates().entries())
 	}
-	async updateCoreStatus(): Promise<any> {
+	getCoreStatus(): {
+		statusCode: StatusCode
+		messages: string[]
+	} {
 		let statusCode = StatusCode.GOOD
-		const messages: Array<string> = []
+		const messages: string[] = []
 
 		if (!this._statusInitialized) {
 			statusCode = StatusCode.BAD
@@ -401,11 +408,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())
 	}
 }
 

packages/playout-gateway/src/health.ts

@@ -0,0 +1,67 @@
+import * as Koa from 'koa'
+import * as Router from '@koa/router'
+import { StatusCode } from 'timeline-state-resolver'
+import { assertNever } from '@sofie-automation/server-core-integration'
+import { CoreHandler } from './coreHandler'
+import { Connector } from './connector'
+
+export interface HealthConfig {
+	/** If set, exposes health HTTP endpoints on the given port */
+	port?: number
+}
+
+/**
+ * Exposes health endpoints for Kubernetes or other orchestrators to monitor
+ * see https://kubernetes.io/docs/concepts/configuration/liveness-readiness-startup-probes
+ */
+export class HealthEndpoints {
+	private app = new Koa()
+	constructor(private connector: Connector, private coreHandler: CoreHandler, private config: HealthConfig) {
+		if (!config.port) return // disabled
+
+		const router = new Router()
+
+		router.get('/healthz', async (ctx) => {
+			if (this.connector.initializedError !== undefined) {
+				ctx.status = 503
+				ctx.body = `Error during initialization: ${this.connector.initializedError}`
+				return
+			}
+			if (!this.connector.initialized) {
+				ctx.status = 503
+				ctx.body = 'Not initialized'
+				return
+			}
+
+			const coreStatus = this.coreHandler.getCoreStatus()
+
+			if (coreStatus.statusCode === StatusCode.UNKNOWN) ctx.status = 503
+			else if (coreStatus.statusCode === StatusCode.FATAL) ctx.status = 503
+			else if (coreStatus.statusCode === StatusCode.BAD) ctx.status = 503
+			else if (coreStatus.statusCode === StatusCode.WARNING_MAJOR) ctx.status = 200
+			else if (coreStatus.statusCode === StatusCode.WARNING_MINOR) ctx.status = 200
+			else if (coreStatus.statusCode === StatusCode.GOOD) ctx.status = 200
+			else assertNever(coreStatus.statusCode)
+
+			if (ctx.status !== 200) {
+				ctx.body = `Status: ${StatusCode[coreStatus.statusCode]}, messages: ${coreStatus.messages.join(', ')}`
+			} else {
+				ctx.body = 'OK'
+			}
+		})
+
+		router.get('/readyz', async (ctx) => {
+			if (!this.coreHandler.connectedToCore) {
+				ctx.status = 503
+				ctx.body = 'Not connected to Core'
+				return
+			}
+			// else
+			ctx.status = 200
+			ctx.body = 'READY'
+		})
+
+		this.app.use(router.routes()).use(router.allowedMethods())
+		this.app.listen(this.config.port)
+	}
+}