using System.Diagnostics; using System.Text.Json; using Elsa.Common.Contracts; using Elsa.Expressions.Services; using Elsa.Extensions; using Elsa.OpenTelemetry.Helpers; using Elsa.Workflows; using Elsa.Workflows.Contracts; using Elsa.Workflows.Pipelines.WorkflowExecution; using Elsa.Workflows.Serialization.Converters; using JetBrains.Annotations; using Activity = System.Diagnostics.Activity; using ActivityKind = System.Diagnostics.ActivityKind; namespace Elsa.OpenTelemetry.Middleware; /// /// Middleware that traces workflow execution using OpenTelemetry. /// [UsedImplicitly] public class OpenTelemetryTracingWorkflowExecutionMiddleware(WorkflowMiddlewareDelegate next, ISystemClock systemClock) : WorkflowExecutionMiddleware(next) { private readonly JsonSerializerOptions? _incidentSerializerOptions = new JsonSerializerOptions().WithConverters(new TypeJsonConverter(WellKnownTypeRegistry.CreateDefault())); /// public override async ValueTask InvokeAsync(WorkflowExecutionContext context) { var workflowInstanceId = context.Id; var workflow = context.Workflow; using var span = ElsaOpenTelemetry.ActivitySource.StartActivity("WorkflowExecution", ActivityKind.Internal, Activity.Current?.Context ?? default); if (span == null) // No listener is registered. { await Next(context); return; } span.SetTag("workflowInstance.id", workflowInstanceId); span.SetTag("workflowDefinition.definitionId", workflow.Identity.DefinitionId); span.SetTag("workflowDefinition.version", workflow.Identity.Version); span.SetTag("workflowDefinition.name", workflow.WorkflowMetadata.Name); span.SetTag("workflowExecution.startTimeUtc", span.StartTimeUtc); if(context.TriggerActivityId != null) { var activity = context.FindActivityById(context.TriggerActivityId) ?? throw new Exception($"Trigger activity with ID {context.TriggerActivityId} not found. This should not happen."); span.SetTag("workflowExecution.trigger.activityId", activity.Id); span.SetTag("workflowExecution.trigger.activityName", activity.Name); span.SetTag("workflowExecution.trigger.activityType", activity.Type); } span.AddEvent(new ActivityEvent("Executing", tags: CreateStatusTags(context))); await Next(context); if (context.SubStatus == WorkflowSubStatus.Faulted) { span.AddEvent(new ActivityEvent("Faulted", tags: CreateStatusTags(context))); span.SetStatus(ActivityStatusCode.Error); span.SetTag("error", true); } else { span.AddEvent(new ActivityEvent("Executed", tags: CreateStatusTags(context))); span.SetStatus(ActivityStatusCode.Ok); } if(context.Incidents.Any()) { span.SetStatus(ActivityStatusCode.Error); span.SetTag("workflowInstance.hasIncidents", true); span.SetTag("error", true); if (context.Incidents.Count > 0) span.SetTag("error.message", JsonSerializer.Serialize(context.Incidents, _incidentSerializerOptions)); } if (!string.IsNullOrWhiteSpace(context.CorrelationId)) span.SetTag("workflowInstance.correlationId", context.CorrelationId); var now = systemClock.UtcNow; span.SetTag("workflowExecution.endTimeUtc", now); span.SetTag("workflowExecution.durationMs", (now - span.StartTimeUtc).TotalMilliseconds); } private ActivityTagsCollection CreateStatusTags(WorkflowExecutionContext context) { return new ActivityTagsCollection(new Dictionary { ["workflowInstance.status"] = context.Status.ToString(), ["workflowInstance.subStatus"] = context.SubStatus.ToString() }); } } /// /// Contains extension methods for . /// [UsedImplicitly] public static class OpenTelemetryWorkflowExecutionMiddlewareExtensions { /// Installs the component in the workflow execution pipeline. public static IWorkflowExecutionPipelineBuilder UseWorkflowExecutionTracing(this IWorkflowExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.Insert(0); }