From b88d12cdd820a929c8ddbfa470f37b8341a44935 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Thu, 17 Apr 2025 18:03:06 +0200 Subject: [PATCH] Fixed Log Persistence Mode Evaluation For Activity Execution (#6595) * Refactor log persistence mode evaluation for activity execution Updated functionality to evaluate and filter log persistence modes during activity execution. Added middleware, interfaces, and supporting methods to manage log configuration, ensuring only persistable inputs and outputs are retained in execution mapping. Updated pipeline to include the new middleware. * Refactor log persistence mapping and evaluation logic Updated log persistence evaluation to use separate input/output maps, ensuring a more modular and maintainable structure. Adjusted property key handling and introduced a dedicated helper method for evaluating persistence properties. Added middleware for log persistence mode evaluation to the workflow execution pipeline. * Refactor log persistence logic and restructure namespaces. Centralizes log persistence logic under a dedicated `LogPersistence` namespace. Simplifies interfaces and methods to streamline functionality, ensuring clarity and consistency in log persistence evaluations. * Refactor log persistence methods for encapsulation. Converted multiple public methods to private to enhance encapsulation and adherence to the principle of least privilege. Introduced a helper method `ResolveFinalLogPersistenceMode` to improve code clarity and maintainability. * Refactor log persistence with improved type safety and structure Updated methods and properties related to log persistence to remove nullable types, enhance clarity, and ensure type safety. Simplified configurations and refactored logic for evaluating persistence modes, reducing redundancy and improving maintainability. * Add XML documentation to IActivityPropertyLogPersistenceEvaluator This commit introduces XML comments to provide clarity on the purpose and functionality of the interface and its methods, aiding developers in understanding their usage and behavior during workflow execution. * Refine log persistence logic in activity execution mapping. Ensure that log persistence modes are correctly handled by adding explicit checks for `LogPersistenceMode.Inherit`. This prevents potential ambiguities and ensures accurate property mapping during workflow execution. * Refactor state handling and log persistence evaluation. Updated method signatures for stricter type consistency and improved readability. Introduced additional resolution step in log persistence to handle legacy configurations more effectively. This enhances code maintainability and alignment with expected behavior. * Update src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- ...cutionContextExtensions.InputEvaluation.cs | 8 +- ...yExecutionContextExtensions.InputOutput.cs | 38 +++ .../Extensions/ActivityExtensions.cs | 4 +- .../Contracts/IActivityExecutonMapper.cs | 11 +- .../Elsa.Workflows.Runtime.csproj.DotSettings | 6 + .../PipelineWorkflowsFeatureExtensions.cs | 1 + .../Extensions/WorkflowsFeatureExtensions.cs | 1 + .../Features/WorkflowRuntimeFeature.cs | 1 + .../TriggerWorkflowInstruction.cs | 1 - ...ActivityPropertyLogPersistenceEvaluator.cs | 19 ++ .../ActivityExecutionContextExtensions.cs | 18 ++ ...ivityExecutionPipelineBuilderExtensions.cs | 6 + .../EvaluateLogPersistenceModesMiddleware.cs | 13 + .../Models/ActivityLogPersistenceModeMap.cs | 9 + .../Models/LogPersistenceConfiguration.cs | 0 .../Models}/LogPersistenceEvaluationMode.cs | 0 ...ActivityPropertyLogPersistenceEvaluator.cs | 230 +++++++++++++++ .../PersistActivityExecutionLogMiddleware.cs | 3 - .../PersistWorkflowExecutionLogMiddleware.cs | 3 - .../Services/BackgroundActivityInvoker.cs | 4 +- .../DefaultActivityExecutionMapper.cs | 265 ++---------------- 21 files changed, 373 insertions(+), 268 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputOutput.cs delete mode 100644 src/modules/Elsa.Workflows.Runtime/Instructions/TriggerWorkflowInstruction.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/LogPersistence/Contracts/IActivityPropertyLogPersistenceEvaluator.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionContextExtensions.cs rename src/modules/Elsa.Workflows.Runtime/{ => LogPersistence}/Extensions/ActivityExecutionPipelineBuilderExtensions.cs (61%) create mode 100644 src/modules/Elsa.Workflows.Runtime/LogPersistence/Middleware/EvaluateLogPersistenceModesMiddleware.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/ActivityLogPersistenceModeMap.cs rename src/modules/Elsa.Workflows.Runtime/{ => LogPersistence}/Models/LogPersistenceConfiguration.cs (100%) rename src/modules/Elsa.Workflows.Runtime/{Enums => LogPersistence/Models}/LogPersistenceEvaluationMode.cs (100%) create mode 100644 src/modules/Elsa.Workflows.Runtime/LogPersistence/Services/ActivityPropertyLogPersistenceEvaluator.cs diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs index 00b332f3d..f47aa314d 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs @@ -101,7 +101,7 @@ public static partial class ActivityExecutionContextExtensions // When input is created from an activity provider, there may be no memory block reference. if (memoryReference?.Id != null!) { - // Declare the input memory block on the current context. + // Declare the input memory block in the current context. context.ExpressionExecutionContext.Set(memoryReference, value!); } } @@ -110,12 +110,12 @@ public static partial class ActivityExecutionContextExtensions value = input; } - await StoreInputValueAsync(context, inputDescriptor, value); + await StoreInputValueAsync(context, inputDescriptor, value!); return value; } - private static async Task StoreInputValueAsync(ActivityExecutionContext context, InputDescriptor inputDescriptor, object? value) + private static Task StoreInputValueAsync(ActivityExecutionContext context, InputDescriptor inputDescriptor, object value) { // Store the serialized input value in the activity state. // Serializing the value ensures we store a copy of the value and not a reference to the input, which may change over time. @@ -128,5 +128,7 @@ public static partial class ActivityExecutionContextExtensions // var filterResult = await manager.RunFiltersAsync(filterContext); context.ActivityState[inputDescriptor.Name] = value; } + + return Task.CompletedTask; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputOutput.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputOutput.cs new file mode 100644 index 000000000..00c6c626e --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputOutput.cs @@ -0,0 +1,38 @@ +using Elsa.Extensions; +using Elsa.Workflows.Models; + +namespace Elsa.Workflows; + +public static class ActivityExecutionContextExtensions +{ + public static IDictionary GetInputs(this ActivityExecutionContext context) + { + return context.ActivityState!; + } + + public static IDictionary GetOutputs(this ActivityExecutionContext context) + { + var activity = context.Activity; + var expressionExecutionContext = context.ExpressionExecutionContext; + var activityDescriptor = context.ActivityDescriptor; + var outputDescriptors = activityDescriptor.Outputs; + + var outputs = outputDescriptors.ToDictionary(x => x.Name, x => + { + if (x.IsSerializable == false) + return "(not serializable)"; + + var cachedValue = activity.GetOutput(expressionExecutionContext, x.Name); + + if (cachedValue != null) + return cachedValue; + + if (x.ValueGetter(activity) is Output output && context.TryGet(output.MemoryBlockReference(), out var outputValue)) + return outputValue!; + + return null!; + }); + + return outputs; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs index a99cd50b3..b6f033699 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs @@ -64,7 +64,7 @@ public static class ActivityExtensions /// The activity execution context. /// Name of the output. /// The output value. - public static object? GetOutput(this IActivity activity, ActivityExecutionContext context, string? outputName = default) + public static object? GetOutput(this IActivity activity, ActivityExecutionContext context, string? outputName = null) { var workflowExecutionContext = context.WorkflowExecutionContext; var outputRegister = workflowExecutionContext.GetActivityOutputRegister(); @@ -84,7 +84,7 @@ public static class ActivityExtensions /// The expression execution context. /// Name of the output. /// The output value. - public static object? GetOutput(this IActivity activity, ExpressionExecutionContext context, string? outputName = default) + public static object? GetOutput(this IActivity activity, ExpressionExecutionContext context, string? outputName = null) { var activityExecutionContext = context.GetActivityExecutionContext(); diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs index b73e3758e..05b943522 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs @@ -10,12 +10,11 @@ public interface IActivityExecutionMapper /// /// Maps an activity execution context to an activity execution record. /// - Task MapAsync(ActivityExecutionContext source); - + ActivityExecutionRecord Map(ActivityExecutionContext source); + /// - /// Retrieves a dictionary containing the persistable output of an activity execution context. + /// Maps an activity execution context to an activity execution record. /// - /// The activity execution context to extract persistable output from. - /// A dictionary containing the persistable output. - Task> GetPersistableOutputAsync(ActivityExecutionContext context); + [Obsolete( "Use Map instead.", error: false)] + Task MapAsync(ActivityExecutionContext source); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings b/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings index 41b655e3e..fb25136b5 100644 --- a/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings +++ b/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings @@ -4,6 +4,12 @@ True True True + True + True + True + True + True + True True True True diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs index 86f648e4a..642e449eb 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs @@ -33,6 +33,7 @@ public static class PipelineWorkflowsFeatureExtensions .UseExceptionHandling() .UseExecutionLogging() .UseNotifications() + .UseLogPersistenceModeEvaluation() .UseBackgroundActivityInvoker(); configurePipeline?.Invoke(pipeline); diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs index 7031ca4a1..233e35343 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs @@ -28,6 +28,7 @@ public static class WorkflowsFeatureExtensions workflowsFeature.WithActivityExecutionPipeline(pipeline => pipeline .UseExceptionHandling() + .UseLogPersistenceModeEvaluation() .UseExecutionLogging() .UseNotifications() .UseBackgroundActivityInvoker()); diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index b9a15ffa4..caa7aba26 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -284,6 +284,7 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module) .AddScoped() .AddScoped() .AddScoped, WorkflowExecutionLogRecordExtractor>() + .AddScoped() .AddScoped() .AddScoped() .AddScoped() diff --git a/src/modules/Elsa.Workflows.Runtime/Instructions/TriggerWorkflowInstruction.cs b/src/modules/Elsa.Workflows.Runtime/Instructions/TriggerWorkflowInstruction.cs deleted file mode 100644 index 4770bcbd7..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Instructions/TriggerWorkflowInstruction.cs +++ /dev/null @@ -1 +0,0 @@ -namespace Elsa.Workflows.Runtime.Instructions; diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Contracts/IActivityPropertyLogPersistenceEvaluator.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Contracts/IActivityPropertyLogPersistenceEvaluator.cs new file mode 100644 index 000000000..c587d107f --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Contracts/IActivityPropertyLogPersistenceEvaluator.cs @@ -0,0 +1,19 @@ +namespace Elsa.Workflows.Runtime; + +/// +/// Provides functionality for evaluating log persistence settings for activity properties +/// during the execution of a workflow. +/// +public interface IActivityPropertyLogPersistenceEvaluator +{ + /// + /// Evaluates the log persistence settings for activity properties within the context of a workflow's execution. + /// + Task EvaluateLogPersistenceModesAsync(ActivityExecutionContext context); + + /// + /// Retrieves a dictionary of persistable output values generated during the execution of an activity. + /// + /// A dictionary where the keys represent output property names and the values represent their persistable data. + Task> GetPersistableOutputAsync(ActivityExecutionContext context); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionContextExtensions.cs new file mode 100644 index 000000000..1d6ce8577 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionContextExtensions.cs @@ -0,0 +1,18 @@ +using Elsa.Extensions; + +namespace Elsa.Workflows.Runtime; + +public static class ActivityExecutionContextExtensions +{ + private static object LogPersistenceMapKey { get; } = new(); + + public static ActivityLogPersistenceModeMap GetLogPersistenceModeMap(this ActivityExecutionContext context) + { + return context.TransientProperties.GetValueOrDefault(LogPersistenceMapKey, () => new ActivityLogPersistenceModeMap())!; + } + + public static void SetLogPersistenceModeMap(this ActivityExecutionContext context, ActivityLogPersistenceModeMap map) + { + context.TransientProperties[LogPersistenceMapKey] = map; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionPipelineBuilderExtensions.cs similarity index 61% rename from src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs rename to src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionPipelineBuilderExtensions.cs index 30408d920..988b6d87b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionPipelineBuilderExtensions.cs @@ -1,5 +1,6 @@ using Elsa.Workflows; using Elsa.Workflows.Pipelines.ActivityExecution; +using Elsa.Workflows.Runtime.Middleware; using Elsa.Workflows.Runtime.Middleware.Activities; // ReSharper disable once CheckNamespace @@ -14,4 +15,9 @@ public static class ActivityExecutionPipelineBuilderExtensions /// Installs the . /// public static IActivityExecutionPipelineBuilder UseBackgroundActivityInvoker(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware(); + + /// + /// Installs the which evaluates log persistence modes during activity execution. + /// + public static IActivityExecutionPipelineBuilder UseLogPersistenceModeEvaluation(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware(); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Middleware/EvaluateLogPersistenceModesMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Middleware/EvaluateLogPersistenceModesMiddleware.cs new file mode 100644 index 000000000..838f22e7d --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Middleware/EvaluateLogPersistenceModesMiddleware.cs @@ -0,0 +1,13 @@ +using Elsa.Workflows.Pipelines.ActivityExecution; + +namespace Elsa.Workflows.Runtime.Middleware; + +public class EvaluateLogPersistenceModesMiddleware(ActivityMiddlewareDelegate next, IActivityPropertyLogPersistenceEvaluator persistenceEvaluator) : IActivityExecutionMiddleware +{ + public async ValueTask InvokeAsync(ActivityExecutionContext context) + { + await next(context); + var persistenceLogMap = await persistenceEvaluator.EvaluateLogPersistenceModesAsync(context); + context.SetLogPersistenceModeMap(persistenceLogMap); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/ActivityLogPersistenceModeMap.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/ActivityLogPersistenceModeMap.cs new file mode 100644 index 000000000..a8472268c --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/ActivityLogPersistenceModeMap.cs @@ -0,0 +1,9 @@ +using Elsa.Workflows.LogPersistence; + +namespace Elsa.Workflows.Runtime; + +public class ActivityLogPersistenceModeMap +{ + public IDictionary Inputs { get; set; } = new Dictionary(); + public IDictionary Outputs { get; set; } = new Dictionary(); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Models/LogPersistenceConfiguration.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceConfiguration.cs similarity index 100% rename from src/modules/Elsa.Workflows.Runtime/Models/LogPersistenceConfiguration.cs rename to src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceConfiguration.cs diff --git a/src/modules/Elsa.Workflows.Runtime/Enums/LogPersistenceEvaluationMode.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceEvaluationMode.cs similarity index 100% rename from src/modules/Elsa.Workflows.Runtime/Enums/LogPersistenceEvaluationMode.cs rename to src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceEvaluationMode.cs diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Services/ActivityPropertyLogPersistenceEvaluator.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Services/ActivityPropertyLogPersistenceEvaluator.cs new file mode 100644 index 000000000..1f5e2c2e9 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Services/ActivityPropertyLogPersistenceEvaluator.cs @@ -0,0 +1,230 @@ +using System.Text.Json; +using System.Text.Json.Serialization; +using Elsa.Expressions.Contracts; +using Elsa.Expressions.Models; +using Elsa.Extensions; +using Elsa.Workflows.Activities; +using Elsa.Workflows.LogPersistence; +using Elsa.Workflows.LogPersistence.Strategies; +using Elsa.Workflows.Management.Options; +using Elsa.Workflows.Models; +using Elsa.Workflows.Serialization.Converters; +using Humanizer; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace Elsa.Workflows.Runtime; + +/* The following legacy JSON structure is expected to be found in the custom properties of the workflow and activity: + * { + * "logPersistenceMode": { + * "default": "default", + * "inputs": { k : v }, + * "outputs": { k: v } + * } + * } + */ + +/* The following JSON structure is expected to be found in the custom properties of the workflow and activity: + * { + * "logPersistenceConfig": { + * "default": { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." }, + * "inputs": { "input1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } }, + * "outputs": { "output1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } } + * } + * } + */ +public class ActivityPropertyLogPersistenceEvaluator : IActivityPropertyLogPersistenceEvaluator +{ + private readonly IExpressionEvaluator _expressionEvaluator; + private readonly JsonSerializerOptions _jsonOptions; + private readonly IDictionary _strategies; + private readonly IOptions _options; + private readonly ILogger _logger; + + private const string LegacyKey = "logPersistenceMode"; + const string ConfigKey = "logPersistenceConfig"; + + public ActivityPropertyLogPersistenceEvaluator( + ILogPersistenceStrategyService strategyService, + IExpressionDescriptorRegistry expressionDescriptorRegistry, + IExpressionEvaluator expressionEvaluator, + IOptions options, + ILogger logger) + { + _expressionEvaluator = expressionEvaluator; + _options = options; + _logger = logger; + _strategies = strategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x); + _jsonOptions = new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true + }.WithConverters( + new ExpressionJsonConverterFactory(expressionDescriptorRegistry), + new JsonStringEnumConverter(), + new ExpandoObjectConverterFactory()); + } + + public async Task EvaluateLogPersistenceModesAsync(ActivityExecutionContext context) + { + var cancellationToken = context.CancellationToken; + var (legacyProps, configProps, defaultMode) = await GetPersistenceDefaultsAsync(context, cancellationToken); + var map = new ActivityLogPersistenceModeMap(); + + await EvaluatePropertiesAsync(context, "inputs", context.ActivityDescriptor.Inputs, legacyProps, configProps, defaultMode, map.Inputs, cancellationToken); + await EvaluatePropertiesAsync(context, "outputs", context.ActivityDescriptor.Outputs, legacyProps, configProps, defaultMode, map.Outputs, cancellationToken); + + return map; + } + + public async Task> GetPersistableOutputAsync(ActivityExecutionContext context) + { + var cancellationToken = context.WorkflowExecutionContext.CancellationToken; + var (legacyProps, configProps, defaultMode) = await GetPersistenceDefaultsAsync(context, cancellationToken); + var outputs = context.GetOutputs(); + return await GetPersistablePropertiesAsync(context, outputs, "outputs", legacyProps, configProps, defaultMode, cancellationToken); + } + + private async Task<(IDictionary legacyProps, IDictionary configProps, LogPersistenceMode defaultMode)> + GetPersistenceDefaultsAsync(ActivityExecutionContext context, CancellationToken cancellationToken) + { + var legacyProps = context.Activity.CustomProperties.GetValueOrDefault>(LegacyKey, () => new Dictionary())!; + var rootContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null); + var workflow = (Workflow?)context.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? context.WorkflowExecutionContext.Workflow; + var workflowDefault = await GetDefaultPersistenceModeAsync(rootContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken); + var activityDefault = await GetDefaultPersistenceModeAsync(context.ExpressionExecutionContext, context.Activity.CustomProperties, () => workflowDefault, cancellationToken); + var configProps = context.Activity.CustomProperties.GetValueOrDefault>(ConfigKey, () => new Dictionary())!; + return (legacyProps, configProps, activityDefault); + } + + private async Task EvaluatePropertiesAsync( + ActivityExecutionContext context, + string key, + IEnumerable descriptors, + IDictionary legacyConfig, + IDictionary currentConfig, + LogPersistenceMode defaultMode, + IDictionary resultMap, + CancellationToken cancellationToken) + { + var legacySection = legacyConfig.GetValueOrDefault(key, () => new Dictionary())!; + var currentSection = currentConfig.GetValueOrDefault(key, () => new Dictionary())!; + + foreach (var descriptor in descriptors) + { + resultMap[descriptor.Name] = await EvaluatePropertyModeAsync(context.ExpressionExecutionContext, descriptor, legacySection, currentSection, defaultMode, cancellationToken); + } + } + + private async Task EvaluatePropertyModeAsync( + ExpressionExecutionContext executionContext, + PropertyDescriptor descriptor, + IDictionary legacySection, + IDictionary currentSection, + LogPersistenceMode defaultMode, + CancellationToken cancellationToken) + { + var key = descriptor.Name.Camelize(); + var configObject = currentSection.GetValueOrDefault(key, () => null); + var config = ConvertToConfig(configObject); + if (config != null) + return await EvaluateConfigAsync(config, executionContext, () => defaultMode, cancellationToken); + + var mode = legacySection.GetValueOrDefault(key, () => defaultMode); + return ResolveMode(mode, () => defaultMode); + } + + private async Task> GetPersistablePropertiesAsync( + ActivityExecutionContext context, + IDictionary state, + string key, + IDictionary legacyConfig, + IDictionary currentConfig, + LogPersistenceMode defaultMode, + CancellationToken cancellationToken) + { + var result = new Dictionary(); + var legacySection = legacyConfig.GetValueOrDefault(key, () => new Dictionary()); + var currentSection = currentConfig.GetValueOrDefault(key, () => new Dictionary()); + + foreach (var item in state) + { + var propKey = item.Key.Camelize(); + var configObject = currentSection!.GetValueOrDefault(propKey, () => null); + var config = ConvertToConfig(configObject); + var mode = config != null + ? await EvaluateConfigAsync(config, context.ExpressionExecutionContext, () => defaultMode, cancellationToken) + : legacySection!.GetValueOrDefault(propKey, () => defaultMode); + + if (mode == LogPersistenceMode.Include || (mode == LogPersistenceMode.Inherit && (defaultMode == LogPersistenceMode.Include || defaultMode == LogPersistenceMode.Inherit))) + result.Add(item.Key, item.Value); + } + + return result; + } + + private async Task GetDefaultPersistenceModeAsync( + ExpressionExecutionContext executionContext, + IDictionary properties, + Func defaultFactory, + CancellationToken cancellationToken) + { + var legacyProps = properties.GetValueOrDefault>(LegacyKey, () => new Dictionary()); + var configProps = properties.GetValueOrDefault>(ConfigKey, () => new Dictionary()); + var defaultObj = configProps!.TryGetValue("default", out var val) ? val : null; + if (defaultObj == null) + { + var legacyDefault = legacyProps!.GetValueOrDefault("default", defaultFactory); + return legacyDefault == LogPersistenceMode.Inherit ? defaultFactory() : legacyDefault; + } + + var config = ConvertToConfig(defaultObj); + return await EvaluateConfigAsync(config, executionContext, defaultFactory, cancellationToken); + } + + private async Task EvaluateConfigAsync( + LogPersistenceConfiguration? config, + ExpressionExecutionContext executionContext, + Func defaultFactory, + CancellationToken cancellationToken) + { + if (config?.EvaluationMode == LogPersistenceEvaluationMode.Strategy) + { + var strategyType = config.StrategyType ?? typeof(Inherit).GetSimpleAssemblyQualifiedName(); + if (!_strategies.TryGetValue(strategyType, out var strategy)) + return defaultFactory(); + + var strategyContext = new LogPersistenceStrategyContext(cancellationToken); + var mode = await strategy.GetPersistenceModeAsync(strategyContext); + return ResolveMode(mode, defaultFactory); + } + + if (config?.Expression == null) + return defaultFactory(); + + try + { + var mode = await _expressionEvaluator.EvaluateAsync(config.Expression, executionContext); + return ResolveMode(mode, defaultFactory); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Error evaluating log persistence expression"); + return defaultFactory(); + } + } + + private LogPersistenceMode ResolveMode(LogPersistenceMode mode, Func defaultFactory) + { + var m = mode == LogPersistenceMode.Inherit ? defaultFactory() : mode; + return m == LogPersistenceMode.Inherit ? LogPersistenceMode.Include : m; + } + + private LogPersistenceConfiguration? ConvertToConfig(object? value) + { + if (value == null) return null; + if (value is LogPersistenceConfiguration config) return config; + var json = JsonSerializer.Serialize(value); + return JsonSerializer.Deserialize(json, _jsonOptions); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs index b659c8635..ceee49fd3 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs @@ -12,8 +12,5 @@ public class PersistActivityExecutionLogMiddleware(WorkflowMiddlewareDelegate ne public override async ValueTask InvokeAsync(WorkflowExecutionContext context) { await Next(context); - - // Not used anymore. - //await sink.PersistExecutionLogsAsync(context); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs index 4c2d39534..d7f03fa20 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs @@ -13,8 +13,5 @@ public class PersistWorkflowExecutionLogMiddleware(WorkflowMiddlewareDelegate ne { // Invoke next middleware. await Next(context); - - // Not used anymore. - //await sink.PersistExecutionLogsAsync(context); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs index 7c1a71048..13ff7f8a0 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs @@ -15,7 +15,7 @@ public class BackgroundActivityInvoker( IWorkflowDefinitionService workflowDefinitionService, IVariablePersistenceManager variablePersistenceManager, IActivityInvoker activityInvoker, - IActivityExecutionMapper activityExecutionMapper, + IActivityPropertyLogPersistenceEvaluator activityPropertyLogPersistenceEvaluator, WorkflowHeartbeatGeneratorFactory workflowHeartbeatGeneratorFactory, IServiceProvider serviceProvider, ILogger logger) @@ -66,7 +66,7 @@ public class BackgroundActivityInvoker( var completed = activityExecutionContext.GetBackgroundCompleted(); var scheduledActivities = activityExecutionContext.GetBackgroundScheduledActivities().ToList(); var workflowInstanceId = scheduledBackgroundActivity.WorkflowInstanceId; - var outputValues = await activityExecutionMapper.GetPersistableOutputAsync(activityExecutionContext); + var outputValues = await activityPropertyLogPersistenceEvaluator.GetPersistableOutputAsync(activityExecutionContext); var properties = new Dictionary { [scheduledActivitiesKey] = JsonSerializer.Serialize(scheduledActivities), diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs index 4457225f6..8a3b04c69 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs @@ -1,88 +1,20 @@ -using System.Text.Json; -using System.Text.Json.Serialization; -using Elsa.Expressions.Contracts; -using Elsa.Expressions.Models; -using Elsa.Extensions; -using Elsa.Workflows.Activities; using Elsa.Workflows.LogPersistence; -using Elsa.Workflows.LogPersistence.Strategies; -using Elsa.Workflows.Management.Options; -using Elsa.Workflows.Models; using Elsa.Workflows.Runtime.Entities; -using Elsa.Workflows.Serialization.Converters; using Elsa.Workflows.State; -using Humanizer; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; namespace Elsa.Workflows.Runtime; /// public class DefaultActivityExecutionMapper : IActivityExecutionMapper { - private readonly JsonSerializerOptions _logPersistenceConfigSerializerOptions; - private readonly IOptions _options; - private readonly IExpressionEvaluator _expressionEvaluator; - private readonly ILogger _logger; - private readonly IDictionary _logPersistenceStrategies; - - public DefaultActivityExecutionMapper( - IOptions options, - ILogPersistenceStrategyService logPersistenceStrategyService, - IExpressionEvaluator expressionEvaluator, - IExpressionDescriptorRegistry expressionDescriptorRegistry, - ILogger logger) + public ActivityExecutionRecord Map(ActivityExecutionContext source) { - _options = options; - _expressionEvaluator = expressionEvaluator; - _logger = logger; - _logPersistenceStrategies = logPersistenceStrategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x); - - _logPersistenceConfigSerializerOptions = new JsonSerializerOptions - { - PropertyNameCaseInsensitive = true - }.WithConverters( - new ExpressionJsonConverterFactory(expressionDescriptorRegistry), - new JsonStringEnumConverter(), - new ExpandoObjectConverterFactory()); - } - - private const string LegacyLogPersistenceModeKey = "logPersistenceMode"; - private const string LogPersistenceConfigKey = "logPersistenceConfig"; - - /// - public async Task MapAsync(ActivityExecutionContext source) - { - /* The following legacy JSON structure is expected to be found in the custom properties of the workflow and activity: - * { - * "logPersistenceMode": { - * "default": "default", - * "inputs": { k : v }, - * "outputs": { k: v } - * } - * } - */ - - /* The following JSON structure is expected to be found in the custom properties of the workflow and activity: - * { - * "logPersistenceConfig": { - * "default": { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." }, - * "inputs": { "input1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } }, - * "outputs": { "output1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } } - * } - * } - */ - - var cancellationToken = source.CancellationToken; - var legacyActivityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary())!; - var rootActivityExecutionContext = source.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null); - var workflow = (Workflow?)source.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? source.WorkflowExecutionContext.Workflow; - var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(rootActivityExecutionContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken); - var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(source.ExpressionExecutionContext, source.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken); - var activityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault>(LogPersistenceConfigKey, () => new Dictionary())!; var payload = GetPayload(source); - var outputs = await GetPersistableOutputAsync(source, legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken); - var inputs = await GetPersistableInputAsync(source, legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken); + var outputs = source.GetOutputs(); + var inputs = source.GetInputs(); + var persistenceMap = source.GetLogPersistenceModeMap(); + var persistableInputs = GetPersistableProperties(inputs, persistenceMap.Inputs); + var persistableOutputs = GetPersistableProperties(outputs, persistenceMap.Outputs); return new() { @@ -92,8 +24,8 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper WorkflowInstanceId = source.WorkflowExecutionContext.Id, ActivityType = source.Activity.Type, ActivityName = source.Activity.Name, - ActivityState = inputs, - Outputs = outputs, + ActivityState = persistableInputs, + Outputs = persistableOutputs, Properties = source.Properties, Payload = payload, Exception = ExceptionState.FromException(source.Exception), @@ -105,162 +37,25 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper }; } - public async Task> GetPersistableOutputAsync(ActivityExecutionContext context) + /// + public Task MapAsync(ActivityExecutionContext source) { - var cancellationToken = context.WorkflowExecutionContext.CancellationToken; - var legacyActivityPersistenceProperties = context.Activity.CustomProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary()); - var rootActivityExecutionContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null); - var workflow = (Workflow?)context.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? context.WorkflowExecutionContext.Workflow; - var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(rootActivityExecutionContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken); - var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(context.ExpressionExecutionContext, context.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken); - var activityPersistenceProperties = context.Activity.CustomProperties.GetValueOrDefault>(LogPersistenceConfigKey, () => new Dictionary()); - - return await GetPersistableOutputAsync( - context, - legacyActivityPersistenceProperties, - activityPersistenceProperties, - activityPersistencePropertyDefault, - cancellationToken); + return Task.FromResult(Map(source)); } - private async Task> GetPersistableOutputAsync( - ActivityExecutionContext context, - IDictionary legacyActivityPersistenceProperties, - IDictionary activityPersistenceProperties, - LogPersistenceMode activityPersistencePropertyDefault, - CancellationToken cancellationToken) - { - var outputs = GetOutputs(context); - return await GetPersistablePropertiesAsync(context, outputs, "outputs", legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken); - } - - private async Task> GetPersistableInputAsync(ActivityExecutionContext context, - IDictionary legacyActivityPersistenceProperties, - IDictionary activityPersistenceProperties, - LogPersistenceMode activityPersistencePropertyDefault, - CancellationToken cancellationToken) - { - return await GetPersistablePropertiesAsync( - context, - context.ActivityState!, - "inputs", - legacyActivityPersistenceProperties, - activityPersistenceProperties, - activityPersistencePropertyDefault, - cancellationToken); - } - - private async Task> GetPersistablePropertiesAsync( - ActivityExecutionContext context, - IDictionary state, - string key, - IDictionary legacyActivityPersistenceProperties, - IDictionary activityPersistenceProperties, - LogPersistenceMode activityPersistencePropertyDefault, - CancellationToken cancellationToken) - { - return await FilterPropertiesUsingPersistenceMode( - context.ExpressionExecutionContext, - state, - legacyActivityPersistenceProperties!.GetValueOrDefault(key, () => new Dictionary())!, - activityPersistenceProperties!.GetValueOrDefault(key, () => new Dictionary())!, - activityPersistencePropertyDefault, - cancellationToken); - } - - private async Task GetDefaultPersistenceModeAsync(ExpressionExecutionContext expressionExecutionContext, IDictionary customProperties, Func defaultFactory, CancellationToken cancellationToken) - { - var legacyProperties = customProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary()); - var properties = customProperties.GetValueOrDefault>(LogPersistenceConfigKey, () => new Dictionary()); - var defaultPersistenceConfigObject = properties?.TryGetValue("default", out var defaultPersistenceConfigObjectValue) == true ? defaultPersistenceConfigObjectValue : null; - - if (defaultPersistenceConfigObject == null) - { - var legacyPersistencePropertyDefault = legacyProperties!.GetValueOrDefault("default", defaultFactory); - - if (legacyPersistencePropertyDefault == LogPersistenceMode.Inherit) - return defaultFactory(); - return legacyPersistencePropertyDefault; - } - - var defaultPersistenceConfig = Convert(defaultPersistenceConfigObject); - return await EvaluateLogPersistenceConfigAsync(defaultPersistenceConfig, expressionExecutionContext, defaultFactory, cancellationToken); - } - - private async Task> FilterPropertiesUsingPersistenceMode( - ExpressionExecutionContext expressionExecutionContext, - IDictionary state, - IDictionary obsoletePersistenceModeConfiguration, - IDictionary persistenceStrategyConfiguration, - LogPersistenceMode defaultLogPersistenceMode, - CancellationToken cancellationToken) + private IDictionary GetPersistableProperties(IDictionary state, IDictionary map) { var result = new Dictionary(); - - foreach (var value in state) + foreach (var stateEntry in state) { - var propKey = value.Key.Camelize(); - var logPersistenceConfigObject = persistenceStrategyConfiguration.GetValueOrDefault(propKey, () => null); - var logPersistenceConfig = Convert(logPersistenceConfigObject); - var mode = logPersistenceConfig != null - ? await EvaluateLogPersistenceConfigAsync(logPersistenceConfig, expressionExecutionContext, () => defaultLogPersistenceMode, cancellationToken) - : obsoletePersistenceModeConfiguration.GetValueOrDefault(propKey, () => defaultLogPersistenceMode); - - if (mode == LogPersistenceMode.Include || mode == LogPersistenceMode.Inherit && defaultLogPersistenceMode is LogPersistenceMode.Include or LogPersistenceMode.Inherit) - result.Add(value.Key, value.Value); + var mode = map.TryGetValue(stateEntry.Key, out var value) ? value : LogPersistenceMode.Include; + if (mode == LogPersistenceMode.Include) + result.Add(stateEntry.Key, stateEntry.Value); } return result; } - - private LogPersistenceConfiguration? Convert(object? value) - { - if (value == null) - return null; - - if (value is LogPersistenceConfiguration c) - return c; - - var json = JsonSerializer.Serialize(value); - var config = JsonSerializer.Deserialize(json, _logPersistenceConfigSerializerOptions); - - return config; - } - - private async Task EvaluateLogPersistenceConfigAsync(LogPersistenceConfiguration? config, ExpressionExecutionContext executionContext, Func defaultMode, CancellationToken cancellationToken) - { - if (config == null) - return defaultMode(); - - if (config.EvaluationMode == LogPersistenceEvaluationMode.Strategy) - { - var strategyTypeName = config.StrategyType ?? typeof(Inherit).GetSimpleAssemblyQualifiedName(); - var strategy = _logPersistenceStrategies.TryGetValue(strategyTypeName, out var v) ? v : null; - - if (strategy == null) - return defaultMode(); - - var strategyContext = new LogPersistenceStrategyContext(cancellationToken); - var logMode = await strategy.GetPersistenceModeAsync(strategyContext); - return logMode == LogPersistenceMode.Inherit ? defaultMode() : logMode; - } - - if (config.Expression == null) - return defaultMode(); - - var expression = config.Expression; - - try - { - return await _expressionEvaluator.EvaluateAsync(expression, executionContext); - } - catch (Exception e) - { - _logger.LogWarning(e, "Error evaluating log persistence expression"); - return defaultMode(); - } - } - + private static IDictionary GetPayload(ActivityExecutionContext source) { var outcomes = source.JournalData.TryGetValue("Outcomes", out var resultValue) ? resultValue as string[] : null; @@ -271,30 +66,4 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper return payload; } - - private static IDictionary GetOutputs(ActivityExecutionContext source) - { - var activity = source.Activity; - var expressionExecutionContext = source.ExpressionExecutionContext; - var activityDescriptor = source.ActivityDescriptor; - var outputDescriptors = activityDescriptor.Outputs; - - var outputs = outputDescriptors.ToDictionary(x => x.Name, x => - { - if (x.IsSerializable == false) - return "(not serializable)"; - - var cachedValue = activity.GetOutput(expressionExecutionContext, x.Name); - - if (cachedValue != null) - return cachedValue; - - if (x.ValueGetter(activity) is Output output && source.TryGet(output.MemoryBlockReference(), out var outputValue)) - return outputValue; - - return null; - }); - - return outputs; - } } \ No newline at end of file