Add metrics

Author: cedric-cordenier

pkg/workflows/wasm/host/execution.go

@@ -40,6 +40,9 @@ type execution[T any] struct {
 	donLogCount          uint32
 	nodeLogCount         uint32
 	awaiting             []int32
+	// peakMemoryBytes is the largest linear memory observed across (re)starts.
+	// It is populated by callWasm and read by Execute to emit the memory metric.
+	peakMemoryBytes int64
 	// suspendOnAwait gates the suspend/resume behaviour. When false, the
 	// execution behaves as it did before suspension was introduced:
 	// awaitCapabilities blocks until each response is available and callCapAsync

pkg/workflows/wasm/host/module.go

@@ -132,6 +132,8 @@ type module struct {
 
 	cfg *ModuleConfig
 
+	metrics *moduleMetrics
+
 	wg     sync.WaitGroup
 	stopCh chan struct{}
 
@@ -394,11 +396,17 @@ func newModule(modCfg *ModuleConfig, binary []byte) (*module, error) {
 
 	modCfg.SdkLabeler(v2ImportName)
 
+	metrics, err := newModuleMetrics()
+	if err != nil {
+		return nil, fmt.Errorf("error creating module metrics: %w", err)
+	}
+
 	m := &module{
 		engine:       engine,
 		module:       mod,
 		wconfig:      cfg,
 		cfg:          modCfg,
+		metrics:      metrics,
 		stopCh:       make(chan struct{}),
 		v2ImportName: v2ImportName,
 	}
@@ -626,6 +634,33 @@ func (m *module) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, executo
 	defer cancel()
 	overallStart := time.Now()
 
+	m.metrics.IncActiveExecutions(ctx, req.SuspendOnAwait)
+	defer m.metrics.DecActiveExecutions(ctx, req.SuspendOnAwait)
+
+	// Accumulated across (re)starts and emitted once the execution completes,
+	// regardless of outcome. wasmDuration is the wall-clock time spent executing
+	// guest code (which is also the CPU-time signal), waitDuration the time spent
+	// suspended waiting for capability responses, and suspensions the number of
+	// times the execution suspended.
+	var (
+		wasmDuration time.Duration
+		waitDuration time.Duration
+		suspensions  int64
+	)
+	defer func() {
+		// The remaining wall-clock time - store setup/teardown, (de)serialisation
+		// and scheduling - is attributed to the "parked" phase.
+		parked := time.Since(overallStart) - wasmDuration - waitDuration
+		if parked < 0 {
+			parked = 0
+		}
+		m.metrics.RecordExecutionPhase(ctx, req.SuspendOnAwait, phaseWasm, wasmDuration)
+		m.metrics.RecordExecutionPhase(ctx, req.SuspendOnAwait, phaseWaiting, waitDuration)
+		m.metrics.RecordExecutionPhase(ctx, req.SuspendOnAwait, phaseParked, parked)
+		m.metrics.RecordSuspensions(ctx, req.SuspendOnAwait, suspensions)
+		m.metrics.RecordMemory(ctx, req.SuspendOnAwait, exec.peakMemoryBytes)
+	}()
+
 	for {
 		// Each (re)start is bounded by the time left in the overall timeout, so a
 		// resumed run - and the wait for its capability responses - cannot extend
@@ -636,6 +671,7 @@ func (m *module) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, executo
 		}
 
 		executionDuration, err := m.callWasm(remaining, req, linkNoDAG, exec)
+		wasmDuration += executionDuration
 
 		switch {
 		case containsCode(err, wasm.CodeSuccess):
@@ -645,17 +681,30 @@ func (m *module) Execute(ctx context.Context, req *sdkpb.ExecuteRequest, executo
 			return exec.response, nil
 		case isSuspendTrap(err):
 			m.cfg.Logger.Debugw("received suspension, awaiting responses", "executionID", executionId)
+			suspensions++
+			// The execution is suspended for as long as we wait for its pending
+			// capability responses; track it as such for the duration of the wait.
+			m.metrics.IncSuspendedExecutions(ctx, req.SuspendOnAwait)
+			waitStart := time.Now()
 			// Wait for the pending capability responses, then resume on the next
 			// loop iteration with a fresh store.
-			for _, id := range exec.awaiting {
-				ar, ok := exec.capabilityResponses[id]
-				if !ok {
-					return nil, fmt.Errorf("missing capability response for awaited callback %d", id)
-				}
-
-				if _, werr := ar.wait(ctxWithTimeout); werr != nil {
-					return nil, werr
+			werr := func() error {
+				for _, id := range exec.awaiting {
+					ar, ok := exec.capabilityResponses[id]
+					if !ok {
+						return fmt.Errorf("missing capability response for awaited callback %d", id)
+					}
+
+					if _, werr := ar.wait(ctxWithTimeout); werr != nil {
+						return werr
+					}
 				}
+				return nil
+			}()
+			waitDuration += time.Since(waitStart)
+			m.metrics.DecSuspendedExecutions(ctx, req.SuspendOnAwait)
+			if werr != nil {
+				return nil, werr
 			}
 
 			continue
@@ -813,6 +862,17 @@ func callWasm[I, O proto.Message](
 	startTime := time.Now()
 	_, err = start.Call(store)
 	executionDuration := time.Since(startTime)
+
+	// Capture the linear memory the guest grew to before the store is closed, so
+	// Execute can emit it as the peak memory metric.
+	if mem := instance.GetExport(store, "memory"); mem != nil {
+		if memory := mem.Memory(); memory != nil {
+			if used := int64(memory.DataSize(store)); used > exec.peakMemoryBytes {
+				exec.peakMemoryBytes = used
+			}
+		}
+	}
+
 	return executionDuration, err
 }