feat: send flamegraph from an unhealthy worker (#119)

Author: ivan-tymoshenko

plugins/alerts.js

@@ -40,6 +40,7 @@ async function alerts (app, _opts) {
       }
 
       const timestamp = Date.now()
+      const workerId = healthInfo.id
       const serviceId = healthInfo.application
       const healthWithTimestamp = { ...healthInfo, timestamp, service: serviceId }
       delete healthWithTimestamp.healthConfig // we don't need to store this
@@ -112,7 +113,7 @@ async function alerts (app, _opts) {
         const alert = await body.json()
 
         app.sendFlamegraphs({
-          serviceIds: [serviceId],
+          workerIds: [workerId],
           alertId: alert.id
         }).catch(err => {
           app.log.error({ err }, 'Failed to send a flamegraph')

plugins/flamegraphs.js

@@ -128,88 +128,98 @@ async function flamegraphs (app, _opts) {
     }
   }
 
-  async function getServiceFlamegraph (serviceId, profileType, attempt = 1) {
+  async function getServiceFlamegraph (workerId, profileType, attempt = 1) {
     const runtime = app.watt.runtime
 
     try {
-      const profile = await runtime.sendCommandToApplication(serviceId, 'getLastProfile', { type: profileType })
+      const profile = await runtime.sendCommandToApplication(workerId, 'getLastProfile', { type: profileType })
       return profile
     } catch (err) {
       if (err.code === 'PLT_PPROF_NO_PROFILE_AVAILABLE') {
         app.log.info(
-          { serviceId, attempt, maxAttempts, attemptTimeout },
+          { workerId, attempt, maxAttempts, attemptTimeout },
           'No profile available for the service. Waiting for profiling to complete.'
         )
         if (attempt <= maxAttempts) {
           await sleep(attemptTimeout)
-          return getServiceFlamegraph(serviceId, profileType, attempt + 1)
+          return getServiceFlamegraph(workerId, profileType, attempt + 1)
         }
       } else if (err.code === 'PLT_PPROF_NOT_ENOUGH_ELU') {
-        app.log.info({ serviceId }, 'ELU low, CPU profiling not active')
+        app.log.info({ workerId }, 'ELU low, CPU profiling not active')
       } else {
-        app.log.warn({ err, serviceId }, 'Failed to get profile from service')
+        app.log.warn({ err, workerId }, 'Failed to get profile from a worker')
+
+        const [serviceId, workerIndex] = workerId.split(':')
+        if (workerIndex) {
+          app.log.warn('Worker not available, trying to get profile from another worker')
+          return getServiceFlamegraph(serviceId, profileType)
+        }
       }
     }
   }
 
+  async function sendServiceFlamegraph (scalerUrl, serviceId, profile, profileType, alertId) {
+    const podId = app.instanceId
+    const url = `${scalerUrl}/pods/${podId}/services/${serviceId}/flamegraph`
+    app.log.info({ serviceId, podId, profileType }, 'Sending flamegraph')
+
+    const query = { profileType }
+    if (alertId) {
+      query.alertId = alertId
+    }
+
+    try {
+      const authHeaders = await app.getAuthorizationHeader()
+      const { statusCode, body } = await request(url, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/octet-stream',
+          ...authHeaders
+        },
+        query,
+        body: profile
+      })
+
+      if (statusCode !== 200) {
+        const error = await body.text()
+        app.log.error({ error }, 'Failed to send flamegraph')
+        throw new Error(`Failed to send flamegraph: ${error}`)
+      }
+    } catch (err) {
+      app.log.warn({ err, serviceId, podId }, 'Failed to send flamegraph from service')
+    }
+  }
+
   app.sendFlamegraphs = async (options = {}) => {
     if (isFlamegraphsDisabled) {
       app.log.info('PLT_DISABLE_FLAMEGRAPHS is set, flamegraphs are disabled')
       return
     }
 
-    let { serviceIds, alertId, profileType = 'cpu' } = options
+    let { workerIds, alertId, profileType = 'cpu' } = options
 
     const scalerUrl = app.instanceConfig?.iccServices?.scaler?.url
     if (!scalerUrl) {
       app.log.error('No scaler URL found in ICC services, cannot send flamegraph')
       throw new Error('No scaler URL found in ICC services, cannot send flamegraph')
     }
 
-    const podId = app.instanceId
     const runtime = app.watt.runtime
 
-    if (!serviceIds) {
+    if (!workerIds) {
       const { applications } = await runtime.getApplications()
-      serviceIds = applications.map(app => app.id)
+      workerIds = applications.map(app => app.id)
     }
 
-    const authHeaders = await app.getAuthorizationHeader()
-
-    const uploadPromises = serviceIds.map(async (serviceId) => {
-      const profile = await getServiceFlamegraph(serviceId, profileType)
+    const uploadPromises = workerIds.map(async (workerId) => {
+      const profile = await getServiceFlamegraph(workerId, profileType)
       if (!profile || !(profile instanceof Uint8Array)) {
-        app.log.error({ serviceId }, 'Failed to get profile from service')
+        app.log.error({ workerId }, 'Failed to get profile from service')
         return
       }
 
-      const url = `${scalerUrl}/pods/${podId}/services/${serviceId}/flamegraph`
-      app.log.info({ serviceId, podId, profileType }, 'Sending flamegraph')
-
-      const query = { profileType }
-      if (alertId) {
-        query.alertId = alertId
-      }
-
-      try {
-        const { statusCode, body } = await request(url, {
-          method: 'POST',
-          headers: {
-            'Content-Type': 'application/octet-stream',
-            ...authHeaders
-          },
-          query,
-          body: profile
-        })
-
-        if (statusCode !== 200) {
-          const error = await body.text()
-          app.log.error({ error }, 'Failed to send flamegraph')
-          throw new Error(`Failed to send flamegraph: ${error}`)
-        }
-      } catch (err) {
-        app.log.warn({ err, serviceId, podId }, 'Failed to send flamegraph from service')
-      }
+      const serviceId = workerId.split(':')[0]
+      await sendServiceFlamegraph(scalerUrl, serviceId, profile, profileType, alertId)
     })
 
     await Promise.all(uploadPromises)

plugins/health-signals.js

@@ -80,6 +80,7 @@ async function healthSignals (app, _opts) {
       }
 
       const {
+        id: workerId,
         application: serviceId,
         currentHealth,
         healthSignals
@@ -125,13 +126,13 @@ async function healthSignals (app, _opts) {
       }
 
       if (healthSignals.length > 0) {
-        await sendHealthSignalsWithTimeout(serviceId, healthSignals)
+        await sendHealthSignalsWithTimeout(serviceId, workerId, healthSignals)
       }
     })
   }
   app.setupHealthSignals = setupHealthSignals
 
-  async function sendHealthSignalsWithTimeout (serviceId, signals) {
+  async function sendHealthSignalsWithTimeout (serviceId, workerId, signals) {
     signalsCaches[serviceId] ??= new HealthSignalsCache()
     servicesSendingStatuses[serviceId] ??= false
 
@@ -148,15 +149,15 @@ async function healthSignals (app, _opts) {
 
         try {
           const signals = signalsCache.getAll()
-          await sendHealthSignals(serviceId, signals, metrics)
+          await sendHealthSignals(serviceId, workerId, signals, metrics)
         } catch (err) {
           app.log.error({ err }, 'Failed to send health signals to scaler')
         }
       }, 5000).unref()
     }
   }
 
-  async function sendHealthSignals (serviceId, signals, metrics) {
+  async function sendHealthSignals (serviceId, workerId, signals, metrics) {
     const scalerUrl = app.instanceConfig?.iccServices?.scaler?.url
     const applicationId = app.instanceConfig?.applicationId
     const authHeaders = await app.getAuthorizationHeader()
@@ -186,6 +187,7 @@ async function healthSignals (app, _opts) {
 
     app.sendFlamegraphs({
       serviceIds: [serviceId],
+      workerIds: [workerId],
       alertId: alert.id
     }).catch(err => {
       app.log.error({ err }, 'Failed to send a flamegraph')