chore: temporal alerting (#3426)

Signed-off-by: Uroš Marolt <[email protected]>

Author: themarolt

services/archetypes/worker/src/activities/activityInterceptor.ts

@@ -19,6 +19,22 @@ export class ActivityMonitoringInterceptor implements ActivityInboundCallsInterc
       attempts: this.ctx.info.attempt,
     }
 
+    const log = getServiceChildLogger('activity-interceptor', {
+      activityType: this.ctx.info.activityType,
+      activityId: this.ctx.info.activityId,
+      workflowType: this.ctx.info.workflowType,
+      taskQueue: this.ctx.info.taskQueue,
+      workflowId: this.ctx.info.workflowExecution.workflowId,
+      runId: this.ctx.info.workflowExecution.runId,
+    })
+
+    if (this.ctx.info.attempt > 10) {
+      log.warn(
+        { slackNotify: true },
+        `Activity ${this.ctx.info.activityType} with id ${this.ctx.info.activityId} was retried ${this.ctx.info.attempt} times!`,
+      )
+    }
+
     telemetry.increment('temporal.activity_execution', 1, tags)
 
     const start = new Date()
@@ -27,15 +43,6 @@ export class ActivityMonitoringInterceptor implements ActivityInboundCallsInterc
       const res = await next(input)
       return res
     } catch (err) {
-      const log = getServiceChildLogger('activity-interceptor', {
-        activityType: this.ctx.info.activityType,
-        activityId: this.ctx.info.activityId,
-        workflowType: this.ctx.info.workflowType,
-        taskQueue: this.ctx.info.taskQueue,
-        workflowId: this.ctx.info.workflowExecution.workflowId,
-        runId: this.ctx.info.workflowExecution.runId,
-      })
-
       log.error(err, 'Error while processing an activity!')
       throw err
     } finally {

services/archetypes/worker/src/activities/index.ts

@@ -1,5 +1,8 @@
+import { getServiceChildLogger } from '@crowd/logging'
 import telemetry from '@crowd/telemetry'
 
+const log = getServiceChildLogger('activity-interceptor')
+
 async function telemetryDistribution(
   name: string,
   value: number,
@@ -16,4 +19,8 @@ async function telemetryIncrement(
   telemetry.increment(name, value, tags)
 }
 
-export { telemetryDistribution, telemetryIncrement }
+async function slackNotify(message: string) {
+  log.warn({ slackNotify: true }, message)
+}
+
+export { telemetryDistribution, telemetryIncrement, slackNotify }

services/archetypes/worker/src/interceptors.ts

@@ -34,6 +34,9 @@ export class WorkflowMonitoringInterceptor implements WorkflowInboundCallsInterc
       return result
     } catch (err) {
       await activity.telemetryIncrement('temporal.workflow_execution_error', 1, tags)
+      await activity.slackNotify(
+        `Workflow ${info.workflowType} with id ${info.workflowId} failed with error: ${err.message}!`,
+      )
       throw err
     } finally {
       const end = new Date()