SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Add support for starting new trace contexts in workflows

Browse source 
Introduced `StartNewTrace` input to workflows and updated OpenTelemetry middleware to support initiating new trace contexts. This helps improve trace isolation and linking when desired, enhancing observability during workflow execution.
This commit is contained in:
Sipke Schoorstra 2025-03-15 21:14:10 +01:00
parent 5f74a83a77
commit 83253b2fae
No known key found for this signature in database
GPG key ID: 5C10502B28A4268F
4 changed files with 49 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
src/modules/Elsa.OpenTelemetry/Middleware/OpenTelemetryTracingWorkflowExecutionMiddleware.cs
View file

@ -27,8 +27,17 @@ public class OpenTelemetryTracingWorkflowExecutionMiddleware(WorkflowMiddlewareD
{
var workflowInstanceId = context.Id;
var workflow = context.Workflow;
using var span = ElsaOpenTelemetry.ActivitySource.StartActivity($"execute workflow {workflow.WorkflowMetadata.Name}", ActivityKind.Server, Activity.Current?.Context ?? default);
var now = systemClock.UtcNow;
var startNewTrace = context.Properties.TryGetValue("StartNewTrace", out var startNewTraceValue) && (bool)startNewTraceValue;
var parentTraceContext = startNewTrace ? default : Activity.Current?.Context ?? default;
var linkedTraceContext = startNewTrace ? Activity.Current : null;
if(startNewTrace)
{
Activity.Current?.Stop();
Activity.Current = null;
}
using var span = ElsaOpenTelemetry.ActivitySource.StartActivity($"execute workflow {workflow.WorkflowMetadata.Name}", ActivityKind.Server, parentTraceContext);
if (span == null) // No listener is registered.
{
@ -36,6 +45,12 @@ public class OpenTelemetryTracingWorkflowExecutionMiddleware(WorkflowMiddlewareD
return;
}
if(startNewTrace)
{
if (linkedTraceContext != null)
span.AddLink(new(linkedTraceContext.Context));
}
span.SetTag("operation.name", "elsa.workflow.execution");
span.SetTag("span.type", "workflow");
span.SetTag("workflow.definition.id", workflow.Identity.DefinitionId);

15
src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs
View file

@ -78,6 +78,12 @@ public class BulkDispatchWorkflows : Activity
Description = "Wait for the dispatched workflows to complete before completing this activity.",
DefaultValue = true)]
public Input<bool> WaitForCompletion { get; set; } = new(true);
/// <summary>
/// Indicates whether a new trace context should be started for the workflow execution.
/// </summary>
[Input(Description = "Start a new trace context when using Open Telemetry.", Category = "Open Telemetry")]
public Input<bool> StartNewTrace { get; set; }
/// <summary>
/// The channel to dispatch the workflow to.
@ -104,12 +110,13 @@ public class BulkDispatchWorkflows : Activity
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var waitForCompletion = WaitForCompletion.GetOrDefault(context);
var startNewTrace = StartNewTrace.GetOrDefault(context);
var items = await context.GetItemSource<object>(Items).ToListAsync(context.CancellationToken);
var count = items.Count;
// Dispatch the child workflows.
foreach (var item in items)
await DispatchChildWorkflowAsync(context, item, waitForCompletion);
await DispatchChildWorkflowAsync(context, item, waitForCompletion, startNewTrace);
// Store the number of dispatched instances for tracking.
context.SetProperty(DispatchedInstancesCountKey, count);
@ -139,7 +146,7 @@ public class BulkDispatchWorkflows : Activity
}
}
private async ValueTask<string> DispatchChildWorkflowAsync(ActivityExecutionContext context, object item, bool waitForCompletion)
private async ValueTask<string> DispatchChildWorkflowAsync(ActivityExecutionContext context, object item, bool waitForCompletion, bool startNewTrace)
{
var workflowDefinitionId = WorkflowDefinitionId.Get(context);
var workflowDefinitionService = context.GetRequiredService<IWorkflowDefinitionService>();
@ -157,8 +164,8 @@ public class BulkDispatchWorkflows : Activity
["ParentInstanceId"] = parentInstanceId
};
if (waitForCompletion)
properties["WaitForCompletion"] = true;
if (waitForCompletion) properties["WaitForCompletion"] = true;
if (startNewTrace) properties["StartNewTrace"] = true;
var itemDictionary = new Dictionary<string, object>
{

27
src/modules/Elsa.Workflows.Runtime/Activities/DispatchWorkflow.cs
View file

@ -20,7 +20,7 @@ namespace Elsa.Workflows.Runtime.Activities;
public class DispatchWorkflow : Activity<object>
{
/// <inheritdoc />
public DispatchWorkflow([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
public DispatchWorkflow([CallerFilePath] string? source = null, [CallerLineNumber] int? line = null) : base(source, line)
{
}
@ -32,7 +32,7 @@ public class DispatchWorkflow : Activity<object>
Description = "The definition ID of the workflow to dispatch.",
UIHint = InputUIHints.WorkflowDefinitionPicker
)]
public Input<string> WorkflowDefinitionId { get; set; } = default!;
public Input<string> WorkflowDefinitionId { get; set; } = null!;
/// <summary>
/// The correlation ID to associate the workflow with.
@ -41,19 +41,25 @@ public class DispatchWorkflow : Activity<object>
DisplayName = "Correlation ID",
Description = "The correlation ID to associate the workflow with."
)]
public Input<string?> CorrelationId { get; set; } = default!;
public Input<string?> CorrelationId { get; set; } = null!;
/// <summary>
/// The input to send to the workflow.
/// </summary>
[Input(Description = "The input to send to the workflow.")]
public Input<IDictionary<string, object>?> Input { get; set; } = default!;
public Input<IDictionary<string, object>?> Input { get; set; } = null!;
/// <summary>
/// True to wait for the child workflow to complete before completing this activity, false to "fire and forget".
/// </summary>
[Input(Description = "Wait for the child workflow to complete before completing this activity.")]
public Input<bool> WaitForCompletion { get; set; } = default!;
public Input<bool> WaitForCompletion { get; set; } = null!;
/// <summary>
/// Indicates whether a new trace context should be started for the workflow execution.
/// </summary>
[Input(Description = "Start a new trace context when using Open Telemetry.", Category = "Open Telemetry")]
public Input<bool> StartNewTrace { get; set; }
/// <summary>
/// The channel to dispatch the workflow to.
@ -64,7 +70,7 @@ public class DispatchWorkflow : Activity<object>
UIHint = InputUIHints.DropDown,
UIHandler = typeof(DispatcherChannelOptionsProvider)
)]
public Input<string?> ChannelName { get; set; } = default!;
public Input<string?> ChannelName { get; set; } = null!;
/// <inheritdoc />
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
@ -103,16 +109,17 @@ public class DispatchWorkflow : Activity<object>
var input = Input.GetOrDefault(context) ?? new Dictionary<string, object>();
var channelName = ChannelName.GetOrDefault(context);
var startNewTrace = StartNewTrace.GetOrDefault(context);
var parentInstanceId = context.WorkflowExecutionContext.Id;
var properties = new Dictionary<string, object>
{
["ParentInstanceId"] = parentInstanceId
["ParentInstanceId"] = parentInstanceId,
};
// If we need to wait for the child workflow to complete, set the property. This will be used by the ResumeDispatchWorkflowActivity handler.
if (waitForCompletion)
properties["WaitForCompletion"] = true;
if (waitForCompletion) properties["WaitForCompletion"] = true;
if (startNewTrace) properties["StartNewTrace"] = true;
input["ParentInstanceId"] = parentInstanceId;
var correlationId = CorrelationId.GetOrDefault(context);

8
src/modules/Elsa.Workflows.Runtime/Services/BackgroundWorkflowDispatcher.cs
View file

@ -22,7 +22,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowDefinitionRequest request, DispatchWorkflowOptions? options = null, CancellationToken cancellationToken = default)
{
var command = new DispatchWorkflowDefinitionCommand(request.DefinitionVersionId)
{
@ -38,7 +38,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchWorkflowInstanceRequest request, DispatchWorkflowOptions? options = null, CancellationToken cancellationToken = default)
{
var command = new DispatchWorkflowInstanceCommand(request.InstanceId){
BookmarkId = request.BookmarkId,
@ -52,7 +52,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchTriggerWorkflowsRequest request, DispatchWorkflowOptions? options = null, CancellationToken cancellationToken = default)
{
var command = new DispatchTriggerWorkflowsCommand(request.ActivityTypeName, request.BookmarkPayload)
{
@ -67,7 +67,7 @@ public class BackgroundWorkflowDispatcher : IWorkflowDispatcher
}
/// <inheritdoc />
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = default, CancellationToken cancellationToken = default)
public async Task<DispatchWorkflowResponse> DispatchAsync(DispatchResumeWorkflowsRequest request, DispatchWorkflowOptions? options = null, CancellationToken cancellationToken = default)
{
var command = new DispatchResumeWorkflowsCommand(request.ActivityTypeName, request.BookmarkPayload)
{