CR-14519-heartbeat

service.yaml

@@ -1,2 +1,2 @@
 name: codefresh-report-image
-version: 0.0.155
+version: 0.0.156

src/logger.ts

@@ -1,7 +1,6 @@
 import winston from 'winston'
 import chalk from 'chalk'
 
-const format = winston.format
 const paint = (kind, message) => {
     const painter = {
         error: chalk.red,
@@ -17,14 +16,14 @@ const paint = (kind, message) => {
 
 
 export const logger = winston.createLogger({
-    format: format.printf((info) => {
+    format: winston.format.printf((info) => {
         return paint(info.level, `${info.message}`)
     }),
-    transports: [ new winston.transports.Console() ],
+    transports: [ new winston.transports.Console({ level: process.env.DEBUG === '1' ?  'debug' : 'info' }) ],
 })
 
 export const workflowLogger = winston.createLogger({
-    format: format.printf((info) => {
+    format: winston.format.printf((info) => {
         let logMessage = `${info.message}`
         if (info.pod) {
             logMessage = `[${info.pod}] ${logMessage}`

src/main.ts

@@ -7,6 +7,7 @@ import { logger, workflowLogger } from './logger'
 
 const { EventSourceError } = errors
 
+const INITIAL_HEARTBEAT_TIMEOUT_IN_SEC = Number(process.env.INITIAL_HEARTBEAT_TIMEOUT_IN_SECONDS) || 5
 
 /**
  * Take (CF_ prefixed) Env variables and perform http/s request (SSE) to app-proxy for image-report with CF_ENRICHERS
@@ -26,12 +27,21 @@ async function main(argv, env): Promise<void> {
         logger.info(`CI provider: ${payload['CF_CI_TYPE']}, job URL: ${payload['CF_WORKFLOW_URL']}`)
     }
     const eventSource = new EventSource(url, { headers })
-    eventSource.reconnectInterval = 1000*10000 // prevent retry. client should not issue a reconnect
+    eventSource.reconnectInterval = 1000 * 100000 // prevent retry. client should not issue a reconnect
+
+    let heartbeatTimer: Utils.Timer
+
     const waitFor = new Promise<void>((resolve, reject) => {
-
-        eventSource.addEventListener('report', function (event) {
-            logger.info(JSON.stringify(JSON.parse(event.data), null, 2))
+
+        eventSource.addEventListener('open', function () {
+            logger.debug('event-source connected')
+
+            heartbeatTimer = Utils.createHeartbeatTimer(() => {
+                logger.debug(`missing heartbeat after ${heartbeatTimer.timeoutTime / 1000} seconds`)
+                heartbeatTimer.restart()
+            }, INITIAL_HEARTBEAT_TIMEOUT_IN_SEC * 1000)
         })
+
         eventSource.addEventListener('info', function (event) {
             logger.info(event.data)
         })
@@ -46,7 +56,22 @@ async function main(argv, env): Promise<void> {
                 workflowLogger.info(event.data)
             }
         })
+        eventSource.addEventListener('heartbeat', function (event) {
+            logger.debug(`heartbeat ${JSON.stringify(event)}`)
+
+            const heartBeatInterval = Number(event.data)
+            if (heartBeatInterval) {
+                const extraGap = 1 * 1000
+                heartbeatTimer.restart(heartBeatInterval + extraGap)
+            } else {
+                heartbeatTimer.restart()
+            }
+        })
         eventSource.addEventListener('error', (errorEvent) => {
+            // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+            if (heartbeatTimer) {
+                heartbeatTimer.stop()
+            }
             eventSource.close()
 
             const error = Utils.tryParseJson(errorEvent.data)
@@ -67,7 +92,12 @@ async function main(argv, env): Promise<void> {
             reject(new EventSourceError(message, name))
         })
         eventSource.addEventListener('end', (event) => {
+            // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+            if (heartbeatTimer) {
+                heartbeatTimer.stop()
+            }
             eventSource.close()
+
             logger.info(event.data)
             resolve()
         })

src/utils.ts

@@ -57,5 +57,26 @@ export namespace Utils {
             return str
         }
     }
+
+    export type Timer = {
+        timeoutTime: number,
+        restart: (timeoutMs?: number) => void,
+        stop: () => void,
+    }
+
+    export function createHeartbeatTimer(cb: () => void, timeoutTime: number): Timer {
+        let timeout: NodeJS.Timeout = setTimeout(cb, timeoutTime)
+
+        return {
+            timeoutTime,
+            restart(_timeoutMs?: number) {
+                this.stop()
+                timeout = setTimeout(cb, _timeoutMs || this.timeoutTime)
+            },
+            stop() {
+                clearTimeout(timeout)
+            }
+        }
+    }
 }