add monitoring

Author: kuruk-mm

.env.default

@@ -8,4 +8,8 @@ PROCESS_METHOD=
 AWS_REGION=
 TASK_QUEUE=
 BUCKET=
-AWS_ENDPOINT=
+AWS_ENDPOINT=
+
+# Pipeline monitoring (optional)
+MONITORING_URL=
+MONITORING_SECRET=

src/adapters/monitoring-reporter.ts

@@ -0,0 +1,147 @@
+import type { ILoggerComponent, IConfigComponent, IBaseComponent } from '@well-known-components/interfaces'
+import type { IFetchComponent } from '@well-known-components/http-server'
+import crypto from 'crypto'
+
+export interface IMonitoringReporter extends IBaseComponent {
+  reportHeartbeat(data: HeartbeatData): void
+  reportJobComplete(data: JobCompleteData): void
+  getConsumerId(): string
+}
+
+export interface HeartbeatData {
+  status: 'idle' | 'processing'
+  currentSceneId?: string
+  currentStep?: string
+  progressPercent?: number
+  startedAt?: string
+}
+
+export interface JobCompleteData {
+  sceneId: string
+  status: 'success' | 'failed'
+  startedAt: string
+  completedAt: string
+  durationMs: number
+  errorMessage?: string
+}
+
+interface MonitoringReporterComponents {
+  logs: ILoggerComponent
+  config: IConfigComponent
+  fetch: IFetchComponent
+}
+
+export function createMonitoringReporter(
+  components: MonitoringReporterComponents,
+  processMethod: string
+): IMonitoringReporter {
+  const { logs, config, fetch } = components
+  const logger = logs.getLogger('monitoring-reporter')
+
+  const consumerId = crypto.randomUUID()
+  let monitoringUrl: string | undefined
+  let monitoringSecret: string | undefined
+  let heartbeatInterval: NodeJS.Timeout | undefined
+  let currentHeartbeatData: HeartbeatData = { status: 'idle' }
+  let isRunning = false
+
+  async function initConfig() {
+    monitoringUrl = await config.getString('MONITORING_URL')
+    monitoringSecret = await config.getString('MONITORING_SECRET')
+
+    if (!monitoringUrl || !monitoringSecret) {
+      logger.info('Monitoring not configured (MONITORING_URL or MONITORING_SECRET missing)')
+    } else {
+      logger.info('Monitoring configured', { consumerId, monitoringUrl })
+    }
+  }
+
+  async function report(endpoint: string, data: object): Promise<void> {
+    if (!monitoringUrl || !monitoringSecret) {
+      return
+    }
+
+    try {
+      const url = `${monitoringUrl}${endpoint}`
+      const controller = new AbortController()
+      const timeoutId = setTimeout(() => controller.abort(), 5000)
+
+      await fetch.fetch(url, {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ ...data, secret: monitoringSecret }),
+        signal: controller.signal
+      })
+
+      clearTimeout(timeoutId)
+    } catch (error) {
+      // Silently ignore - monitoring should never block pipeline
+      logger.debug('Monitoring report failed (non-blocking)', {
+        error: error instanceof Error ? error.message : 'Unknown error'
+      })
+    }
+  }
+
+  function sendHeartbeat() {
+    report('/api/monitoring/heartbeat', {
+      consumerId,
+      processMethod,
+      ...currentHeartbeatData
+    })
+  }
+
+  function startHeartbeat() {
+    if (heartbeatInterval) {
+      return
+    }
+
+    // Send initial heartbeat
+    sendHeartbeat()
+
+    // Set up interval (every 10 seconds)
+    heartbeatInterval = setInterval(sendHeartbeat, 10000)
+  }
+
+  function stopHeartbeat() {
+    if (heartbeatInterval) {
+      clearInterval(heartbeatInterval)
+      heartbeatInterval = undefined
+    }
+  }
+
+  return {
+    async start() {
+      await initConfig()
+      isRunning = true
+      startHeartbeat()
+    },
+
+    async stop() {
+      isRunning = false
+      stopHeartbeat()
+    },
+
+    getConsumerId() {
+      return consumerId
+    },
+
+    reportHeartbeat(data: HeartbeatData) {
+      currentHeartbeatData = data
+      // Heartbeat will be sent on next interval, but also send immediately for status changes
+      if (isRunning) {
+        sendHeartbeat()
+      }
+    },
+
+    reportJobComplete(data: JobCompleteData) {
+      report('/api/monitoring/job-complete', {
+        consumerId,
+        processMethod,
+        ...data
+      })
+
+      // Reset heartbeat data to idle
+      currentHeartbeatData = { status: 'idle' }
+    }
+  }
+}

src/components.ts

@@ -13,6 +13,8 @@ import { AppComponents, GlobalContext } from './types'
 import path from 'path'
 import { createMockSnsAdapterComponent, createSnsAdapterComponent } from './adapters/sns'
 import { AwsCredentialIdentity } from '@smithy/types'
+import { createMonitoringReporter } from './adapters/monitoring-reporter'
+import { getProcessMethod } from './service'
 
 // Helper function to handle entityId logic
 async function handleEntityId(
@@ -152,6 +154,10 @@ export async function initComponents(): Promise<AppComponents> {
     ? createSnsAdapterComponent({ logs }, { snsArn, snsEndpoint: snsEndpoint })
     : createMockSnsAdapterComponent({ logs })
 
+  // Create monitoring reporter
+  const processMethod = await getProcessMethod(config)
+  const monitoringReporter = createMonitoringReporter({ logs, config, fetch }, processMethod)
+
   const entityIdIndex = process.argv.findIndex((p) => p === '--entityId')
   if (entityIdIndex !== -1) {
     const entityId = process.argv[entityIdIndex + 1]
@@ -173,6 +179,7 @@ export async function initComponents(): Promise<AppComponents> {
     runner,
     deploymentsByPointer: mitt(),
     storage,
-    snsAdapter
+    snsAdapter,
+    monitoringReporter
   }
 }

src/service.ts

@@ -49,6 +49,19 @@ export async function main(program: Lifecycle.EntryPointParameters<AppComponents
   components.runner.runTask(async (opt) => {
     while (opt.isRunning) {
       await components.taskQueue.consumeAndProcessJob(async (job, message) => {
+        const sceneId = job.entity.entityId
+        const startedAt = new Date().toISOString()
+        const startTime = Date.now()
+
+        // Report job start
+        components.monitoringReporter.reportHeartbeat({
+          status: 'processing',
+          currentSceneId: sceneId,
+          currentStep: 'Starting',
+          progressPercent: 0,
+          startedAt
+        })
+
         try {
           switch (processMethod) {
             case 'godot_optimizer':
@@ -67,9 +80,28 @@ export async function main(program: Lifecycle.EntryPointParameters<AppComponents
               logger.info('Consume and Process: ', { job: JSON.stringify(job), message: JSON.stringify(message) })
               break
           }
+
+          // Report job success
+          components.monitoringReporter.reportJobComplete({
+            sceneId,
+            status: 'success',
+            startedAt,
+            completedAt: new Date().toISOString(),
+            durationMs: Date.now() - startTime
+          })
         } catch (error) {
           logger.error(`Error processing job ${job.entity.entityId}`)
           logger.error(error as any)
+
+          // Report job failure
+          components.monitoringReporter.reportJobComplete({
+            sceneId,
+            status: 'failed',
+            startedAt,
+            completedAt: new Date().toISOString(),
+            durationMs: Date.now() - startTime,
+            errorMessage: error instanceof Error ? error.message : 'Unknown error'
+          })
         }
       })
     }

src/types.ts

@@ -13,6 +13,7 @@ import { IRunnerComponent } from './adapters/runner'
 import { Emitter } from 'mitt'
 import { IStorageComponent } from './adapters/storage'
 import { ISNSAdapterComponent } from './adapters/sns'
+import { IMonitoringReporter } from './adapters/monitoring-reporter'
 
 export type GlobalContext = {
   components: BaseComponents
@@ -30,6 +31,7 @@ export type BaseComponents = {
   deploymentsByPointer: Emitter<Record<string /* pointer */, DeploymentToSqs>>
   storage: IStorageComponent
   snsAdapter: ISNSAdapterComponent
+  monitoringReporter: IMonitoringReporter
 }
 
 // components used in runtime