Otel improvements (#6890)
* Otel improvements * Update src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.cs Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * Update src/modules/Elsa.OpenTelemetry/Middleware/OpenTelemetryTracingActivityExecutionMiddleware.cs Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * Fixed comments --------- Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>
This commit is contained in:
MariusVuscanNx
GitHub
parent
48be5a1569
commit
17bb1d08c9
|
|
@ -1,4 +1,5 @@
|
|||
using System.Diagnostics;
|
||||
using System.Text;
|
||||
using Elsa.Common;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.OpenTelemetry.Contracts;
|
||||
|
|
@ -6,6 +7,7 @@ using Elsa.OpenTelemetry.Helpers;
|
|||
using Elsa.OpenTelemetry.Models;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Pipelines.ActivityExecution;
|
||||
using Humanizer;
|
||||
using JetBrains.Annotations;
|
||||
using Activity = System.Diagnostics.Activity;
|
||||
using ActivityKind = System.Diagnostics.ActivityKind;
|
||||
|
|
@ -36,9 +38,9 @@ public class OpenTelemetryTracingActivityExecutionMiddleware(ActivityMiddlewareD
|
|||
span.SetTag("activity.version", activity.Version);
|
||||
span.SetTag("activity.instance.id", context.Id);
|
||||
span.SetTag("activity.tenant.id", context.WorkflowExecutionContext.Workflow.Identity.TenantId);
|
||||
|
||||
|
||||
var activityKind = context.ActivityDescriptor.Kind;
|
||||
if (activityKind == Elsa.Workflows.ActivityKind.Job || (activityKind == Workflows.ActivityKind.Task && activity.GetRunAsynchronously()))
|
||||
if (activityKind == Elsa.Workflows.ActivityKind.Job || (activityKind == Workflows.ActivityKind.Task && activity.GetRunAsynchronously()))
|
||||
span.SetTag("span.type", "job");
|
||||
|
||||
span.AddEvent(new("executing"));
|
||||
|
|
@ -54,7 +56,7 @@ public class OpenTelemetryTracingActivityExecutionMiddleware(ActivityMiddlewareD
|
|||
var errorSpanHandler = context.GetServices<IActivityErrorSpanHandler>()
|
||||
.OrderBy(x => x.Order)
|
||||
.FirstOrDefault(x => x.CanHandle(errorSpanHandlerContext));
|
||||
|
||||
|
||||
errorSpanHandler?.Handle(errorSpanHandlerContext);
|
||||
}
|
||||
else if (context.Status == ActivityStatus.Canceled)
|
||||
|
|
@ -72,6 +74,44 @@ public class OpenTelemetryTracingActivityExecutionMiddleware(ActivityMiddlewareD
|
|||
|
||||
if (!string.IsNullOrWhiteSpace(context.WorkflowExecutionContext.CorrelationId))
|
||||
span.SetTag("workflow.correlation_id", context.WorkflowExecutionContext.CorrelationId);
|
||||
|
||||
span.SetTag("workflow.instance.id", context.WorkflowExecutionContext.Id);
|
||||
span.SetTag("workflow.definition.id", context.WorkflowExecutionContext.Workflow.Identity.DefinitionId);
|
||||
span.SetTag("workflow.definition.version", context.WorkflowExecutionContext.Workflow.Identity.Version);
|
||||
|
||||
SetExtensions(context, span);
|
||||
}
|
||||
|
||||
private void SetExtensions(ActivityExecutionContext context, Activity span)
|
||||
{
|
||||
var extensions = context.GetExtensionsMetadata();
|
||||
if (extensions is not null)
|
||||
{
|
||||
foreach (var key in extensions.Keys)
|
||||
{
|
||||
span.SetTag($"activity.extensions.{ToOtelFormat(key)}", extensions[key]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public string ToOtelFormat(string input)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(input))
|
||||
return string.Empty;
|
||||
|
||||
var humanized = input.Humanize();
|
||||
|
||||
var sb = new StringBuilder(humanized.Length);
|
||||
|
||||
foreach (var c in humanized)
|
||||
{
|
||||
if (char.IsWhiteSpace(c))
|
||||
sb.Append('_');
|
||||
else
|
||||
sb.Append(char.ToLowerInvariant(c));
|
||||
}
|
||||
|
||||
return sb.ToString();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -96,10 +96,16 @@ public class OpenTelemetryTracingWorkflowExecutionMiddleware(WorkflowMiddlewareD
|
|||
|
||||
if (context.Incidents.Any())
|
||||
{
|
||||
span.SetTag("workflow.incidents.count", context.Incidents.Count);
|
||||
|
||||
var incidentTagsList = new List<ActivityTagsCollection>();
|
||||
foreach (var incident in context.Incidents)
|
||||
span.AddEvent(new("incident", incident.Timestamp, CreateIncidentTags(incident)));
|
||||
{
|
||||
var incidentTags = CreateIncidentTags(incident);
|
||||
incidentTagsList.Add(incidentTags);
|
||||
span.AddEvent(new("incident", incident.Timestamp, incidentTags));
|
||||
}
|
||||
|
||||
span.SetTag("workflow.incidents.items", incidentTagsList);
|
||||
span.SetTag("workflow.incidents.count", context.Incidents.Count);
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(context.CorrelationId))
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
using System.Text.Json;
|
||||
using Elsa.Expressions.Helpers;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Resilience.Entities;
|
||||
using Elsa.Resilience.Extensions;
|
||||
using Elsa.Resilience.Models;
|
||||
|
|
@ -12,55 +13,56 @@ using Polly.Telemetry;
|
|||
namespace Elsa.Resilience;
|
||||
|
||||
public class ResilientActivityInvoker(
|
||||
IResilienceStrategyConfigEvaluator resilienceStrategyConfigEvaluator,
|
||||
IRetryAttemptRecorder retryAttemptRecorder,
|
||||
IIdentityGenerator identityGenerator,
|
||||
IResilienceStrategyConfigEvaluator resilienceStrategyConfigEvaluator,
|
||||
IRetryAttemptRecorder retryAttemptRecorder,
|
||||
IIdentityGenerator identityGenerator,
|
||||
ResilienceStrategySerializer resilienceStrategySerializer) : IResilientActivityInvoker
|
||||
{
|
||||
private const string ResilienceStrategyIdPropKey = "resilienceStrategy";
|
||||
private const string RetryAttemptsCountKey = "RetryAttemptsCount";
|
||||
|
||||
public async Task<T> InvokeAsync<T>(IResilientActivity activity, ActivityExecutionContext context, Func<Task<T>> action, CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Get the resilience strategy.
|
||||
var strategyConfig = GetStrategyConfig(activity);
|
||||
var resilienceStrategy = await resilienceStrategyConfigEvaluator.EvaluateAsync(strategyConfig, context.ExpressionExecutionContext, cancellationToken);
|
||||
|
||||
|
||||
// If no resilience strategy is configured, execute the action as-is.
|
||||
if (resilienceStrategy == null)
|
||||
return await action();
|
||||
|
||||
|
||||
// Record the applied strategy as part of the activity execution context for diagnostics.
|
||||
var resilienceStrategyModel = JsonSerializer.SerializeToNode(resilienceStrategy, resilienceStrategySerializer.SerializerOptions)!;
|
||||
context.SetResilienceStrategy(resilienceStrategyModel);
|
||||
|
||||
|
||||
// Create a resilience pipeline builder.
|
||||
var builder = CreateResiliencePipelineBuilder<T>();
|
||||
var retries = new List<RetryAttempt>();
|
||||
context.TransientProperties[RetryAttempt.RetriesKey] = retries;
|
||||
|
||||
|
||||
// Create a resilience context.
|
||||
var resilienceContext = ResilienceContextPool.Shared.Get(cancellationToken);
|
||||
resilienceContext.Properties.Set(new(nameof(ActivityExecutionContext)), context);
|
||||
|
||||
|
||||
try
|
||||
{
|
||||
// Configure the resilience pipeline.
|
||||
await resilienceStrategy.ConfigurePipeline(builder, resilienceContext);
|
||||
var pipeline = builder.Build();
|
||||
|
||||
|
||||
// Execute the action within the resilience pipeline.
|
||||
var result = await pipeline.ExecuteAsync<T>(async _ => await action(), resilienceContext);
|
||||
|
||||
|
||||
// Record the retry attempts.
|
||||
await RecordRetryAttempts(activity, context, retries, cancellationToken);
|
||||
|
||||
|
||||
return result;
|
||||
}
|
||||
finally
|
||||
{
|
||||
ResilienceContextPool.Shared.Return(resilienceContext);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
private async Task RecordRetryAttempts(IResilientActivity activity, ActivityExecutionContext context, ICollection<RetryAttempt> attempts, CancellationToken cancellationToken = default)
|
||||
|
|
@ -70,9 +72,11 @@ public class ResilientActivityInvoker(
|
|||
var records = Map(context, activity, attempts);
|
||||
var recordContext = new RecordRetryAttemptsContext(context, records, cancellationToken);
|
||||
await retryAttemptRecorder.RecordAsync(recordContext);
|
||||
|
||||
|
||||
// Propagate a flag that retries have occurred. This information can then be used to show the retry attempts in the workflow designer.
|
||||
context.SetRetriesAttemptedFlag();
|
||||
|
||||
context.SetExtensionsMetadata(RetryAttemptsCountKey, attempts.Count);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -89,7 +93,7 @@ public class ResilientActivityInvoker(
|
|||
? null
|
||||
: value.ConvertTo<ResilienceStrategyConfig>();
|
||||
}
|
||||
|
||||
|
||||
private ICollection<RetryAttemptRecord> Map(ActivityExecutionContext activityExecutionContext, IResilientActivity resilientActivity, ICollection<RetryAttempt> attempts)
|
||||
{
|
||||
return attempts.Select(x => Map(activityExecutionContext, resilientActivity, x)).ToList();
|
||||
|
|
|
|||
|
|
@ -24,6 +24,8 @@ namespace Elsa.Extensions;
|
|||
[PublicAPI]
|
||||
public static partial class ActivityExecutionContextExtensions
|
||||
{
|
||||
private const string ExtensionsMetadataKey = "Extensions";
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to get a value from the input provided via <see cref="WorkflowExecutionContext"/>. If a value was found, an attempt is made to convert it into the specified type <code>T</code>.
|
||||
/// </summary>
|
||||
|
|
@ -436,6 +438,29 @@ public static partial class ActivityExecutionContextExtensions
|
|||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sets extension data in the metadata. Represents specific data that is exposed generically for an activity.
|
||||
/// </summary>
|
||||
public static void SetExtensionsMetadata(this ActivityExecutionContext context, string key, object? value)
|
||||
{
|
||||
var extensionsDictionary = context.GetExtensionsMetadata();
|
||||
|
||||
if(extensionsDictionary == null) extensionsDictionary = new();
|
||||
|
||||
extensionsDictionary[key] = value;
|
||||
|
||||
context.Metadata[ExtensionsMetadataKey] = extensionsDictionary;
|
||||
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Retrives the extensin data from the metdata. Represents specific data that is exposed generically for an activity.
|
||||
/// </summary>
|
||||
public static Dictionary<string, object?>? GetExtensionsMetadata(this ActivityExecutionContext context)
|
||||
{
|
||||
return context.Metadata[ExtensionsMetadataKey] as Dictionary<string, object?>;
|
||||
}
|
||||
|
||||
internal static bool GetHasEvaluatedProperties(this ActivityExecutionContext context) => context.TransientProperties.TryGetValue<bool>("HasEvaluatedProperties", out var value) && value;
|
||||
internal static void SetHasEvaluatedProperties(this ActivityExecutionContext context) => context.TransientProperties["HasEvaluatedProperties"] = true;
|
||||
}
|
||||
Loading…
Reference in a new issue