feat: record RUM workflow execution vitals

Author: huang47

src/platform/telemetry/providers/cloud/DatadogRumTelemetryProvider.test.ts

@@ -0,0 +1,48 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+import { DatadogRumTelemetryProvider } from './DatadogRumTelemetryProvider'
+
+const addDurationVital = vi.fn()
+
+function installDatadogRum(): void {
+  Object.defineProperty(window, 'DD_RUM', {
+    configurable: true,
+    value: { addDurationVital }
+  })
+}
+
+afterEach(() => {
+  vi.restoreAllMocks()
+  vi.clearAllMocks()
+  Reflect.deleteProperty(window, 'DD_RUM')
+})
+
+describe('DatadogRumTelemetryProvider', () => {
+  it('records a workflow vital with its duration and outcome', () => {
+    installDatadogRum()
+    vi.spyOn(performance, 'now').mockReturnValue(142)
+
+    new DatadogRumTelemetryProvider().trackExecutionOutcome({
+      startTime: 42,
+      outcome: 'success'
+    })
+
+    expect(addDurationVital).toHaveBeenCalledWith('workflow_execution', {
+      startTime: performance.timeOrigin + 42,
+      duration: 100,
+      context: {
+        outcome: 'success',
+        product: 'cloud_generation'
+      }
+    })
+  })
+
+  it('does nothing when Datadog RUM is unavailable', () => {
+    expect(() =>
+      new DatadogRumTelemetryProvider().trackExecutionOutcome({
+        startTime: 42,
+        outcome: 'failure'
+      })
+    ).not.toThrow()
+  })
+})

src/platform/telemetry/providers/cloud/DatadogRumTelemetryProvider.ts

@@ -1,3 +1,35 @@
-import type { TelemetryProvider } from '../../types'
+import type { ExecutionOutcomeMetadata, TelemetryProvider } from '../../types'
 
-export class DatadogRumTelemetryProvider implements TelemetryProvider {}
+interface DatadogRumDurationVitalOptions {
+  startTime: number
+  duration: number
+  context?: Record<string, unknown>
+}
+
+interface DatadogRumClient {
+  addDurationVital(name: string, options: DatadogRumDurationVitalOptions): void
+}
+
+interface WindowWithDatadogRum extends Window {
+  DD_RUM?: DatadogRumClient
+}
+
+function getDatadogRum(): DatadogRumClient | undefined {
+  return (window as WindowWithDatadogRum).DD_RUM
+}
+
+export class DatadogRumTelemetryProvider implements TelemetryProvider {
+  trackExecutionOutcome({
+    startTime,
+    outcome
+  }: ExecutionOutcomeMetadata): void {
+    getDatadogRum()?.addDurationVital('workflow_execution', {
+      startTime: performance.timeOrigin + startTime,
+      duration: performance.now() - startTime,
+      context: {
+        outcome,
+        product: 'cloud_generation'
+      }
+    })
+  }
+}

src/scripts/app.ts

@@ -1662,6 +1662,7 @@ export class ComfyApp {
           // user switches tabs while the request is in flight.
           const queuedWorkflow = useWorkspaceStore().workflow
             .activeWorkflow as ComfyWorkflow
+          const startTime = performance.now()
           const p = await this.graphToPrompt(this.rootGraph)
           const queuedNodes = collectAllNodes(this.rootGraph)
           try {
@@ -1685,6 +1686,7 @@ export class ComfyApp {
                   id: res.prompt_id,
                   nodes: Object.keys(p.output),
                   promptOutput: p.output,
+                  startTime,
                   workflow: queuedWorkflow
                 })
               }

src/stores/executionStore.test.ts

@@ -21,6 +21,7 @@ const {
   mockOpenWorkflows,
   mockShowTextPreview,
   mockTrackExecutionError,
+  mockTrackExecutionOutcome,
   mockTrackExecutionSuccess,
   mockTrackSharedWorkflowRun
 } = await vi.hoisted(async () => {
@@ -33,6 +34,7 @@ const {
     mockOpenWorkflows: shallowRef<{ path: string }[]>([]),
     mockShowTextPreview: vi.fn(),
     mockTrackExecutionError: vi.fn(),
+    mockTrackExecutionOutcome: vi.fn(),
     mockTrackExecutionSuccess: vi.fn(),
     mockTrackSharedWorkflowRun: vi.fn()
   }
@@ -92,6 +94,7 @@ vi.mock('@/platform/distribution/types', async () => ({
 vi.mock('@/platform/telemetry', () => ({
   useTelemetry: () => ({
     trackExecutionError: mockTrackExecutionError,
+    trackExecutionOutcome: mockTrackExecutionOutcome,
     trackExecutionSuccess: mockTrackExecutionSuccess,
     trackSharedWorkflowRun: mockTrackSharedWorkflowRun
   })
@@ -614,6 +617,7 @@ describe('useExecutionStore - workflowStatus', () => {
       nodes: ['1'],
       id: jobId,
       promptOutput: { '1': createPromptNode('Node', 'TestNode') },
+      startTime: 42,
       workflow
     })
   }
@@ -631,6 +635,7 @@ describe('useExecutionStore - workflowStatus', () => {
     callStoreJob('job-1', workflowA)
     fireExecutionStart('job-1')
 
+    expect(mockTrackExecutionOutcome).not.toHaveBeenCalled()
     expect(store.getWorkflowStatus(workflowA)).toBe('running')
   })
 
@@ -648,6 +653,11 @@ describe('useExecutionStore - workflowStatus', () => {
     fireExecutionSuccess('job-1')
 
     callStoreJob('job-1', workflowA)
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledOnce()
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledWith({
+      startTime: 42,
+      outcome: 'success'
+    })
     expect(store.getWorkflowStatus(workflowA)).toBe('completed')
   })
 
@@ -656,6 +666,10 @@ describe('useExecutionStore - workflowStatus', () => {
     fireExecutionError('job-1')
 
     callStoreJob('job-1', workflowA)
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledWith({
+      startTime: 42,
+      outcome: 'failure'
+    })
     expect(store.getWorkflowStatus(workflowA)).toBe('failed')
   })
 
@@ -672,6 +686,10 @@ describe('useExecutionStore - workflowStatus', () => {
     fireExecutionStart('job-1')
     fireExecutionSuccess('job-1')
 
+    expect(mockTrackExecutionOutcome).toHaveBeenCalledWith({
+      startTime: 42,
+      outcome: 'success'
+    })
     expect(store.getWorkflowStatus(workflowA)).toBe('completed')
   })
 

src/stores/executionStore.ts

@@ -51,6 +51,7 @@ interface QueuedJob {
    * value is a boolean indicating if the node has been executed.
    */
   nodes: Record<string, boolean>
+  startTime?: number
   /**
    * The workflow that is queued to be executed
    */
@@ -177,13 +178,27 @@ export const useExecutionStore = defineStore('execution', () => {
     mutateStatus((m) => m.set(workflow, status))
   }
 
+  function trackExecutionOutcome(
+    jobId: string,
+    status: WorkflowExecutionStatus
+  ) {
+    if (status === 'running') return
+    const startTime = queuedJobs.value[jobId]?.startTime
+    if (startTime === undefined) return
+    useTelemetry()?.trackExecutionOutcome({
+      startTime,
+      outcome: status === 'completed' ? 'success' : 'failure'
+    })
+  }
+
   function setWorkflowStatus(jobId: string, status: WorkflowExecutionStatus) {
     const workflow = jobIdToWorkflow.get(jobId)
     if (!workflow) {
       bufferPendingWorkflowStatus(jobId, status)
       return
     }
     applyWorkflowStatus(workflow, status)
+    trackExecutionOutcome(jobId, status)
   }
 
   function clearWorkflowStatus(workflow: ComfyWorkflow) {
@@ -718,11 +733,13 @@ export const useExecutionStore = defineStore('execution', () => {
     nodes,
     id,
     promptOutput,
+    startTime,
     workflow
   }: {
     nodes: string[]
     id: JobId
     promptOutput: ComfyApiWorkflow
+    startTime?: number
     workflow: ComfyWorkflow
   }) {
     queuedJobs.value[id] ??= { nodes: {} }
@@ -735,6 +752,7 @@ export const useExecutionStore = defineStore('execution', () => {
       ...queuedJob.nodes
     }
     queuedJob.nodeLookup = buildExecutionNodeLookup(promptOutput)
+    queuedJob.startTime = startTime
     queuedJob.workflow = workflow
     if (workflow) jobIdToWorkflow.set(String(id), workflow)
     queuedJob.shareId = workflow?.shareId
@@ -761,6 +779,7 @@ export const useExecutionStore = defineStore('execution', () => {
     // Don't let a stale 'running' overwrite a terminal status already set.
     if (pending === 'running' && workflowStatus.value.has(workflow)) return
     applyWorkflowStatus(workflow, pending)
+    trackExecutionOutcome(jobId, pending)
   }
 
   // ~0.65 MB at capacity (32 char GUID key + 50 char path value)